MongoDB CDC Change Stream and Oplog Replication Lag Sentinel

id: mongodb-cdc-change-stream-and-oplog-lag-sentinel
namespace: company.data

inputs:
  - id: hostname
    type: STRING
    defaults: "localhost"
    description: Hostname or IP address of the target MongoDB replica set primary host.

  - id: port
    type: STRING
    defaults: "27017"
    description: Port number on which the MongoDB instance is listening.

  - id: connection_string
    type: STRING
    defaults: "mongodb://localhost:27017/?replicaSet=rs0"
    description: Connection string to the target MongoDB replica set or sharded
      cluster (must include replicaSet parameter).

  - id: included_collections
    type: STRING
    defaults: "production[.]*"
    description: Regular expression matching database and collection namespaces to
      monitor (e.g. production[.]*).

  - id: max_records
    type: INT
    defaults: 200
    description: Maximum number of change events to drain from the MongoDB oplog per
      audit cycle.

  - id: backlog_critical_threshold
    type: INT
    defaults: 50
    description: Number of accumulated change events in the polling window that
      indicates severe CDC ingestion lag.

  - id: max_wait_seconds
    type: INT
    defaults: 10
    description: Maximum duration in seconds to wait for incoming change events
      before closing the batch.

  - id: state_name
    type: STRING
    defaults: "mongodb-cdc-checkpoint"
    description: Kestra KV store key used to durably persist the Debezium oplog
      resume token and offset state.

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */2 * * *"
    disabled: true
    inputs:
      hostname: "localhost"
      port: "27017"
      connection_string: "mongodb://localhost:27017/?replicaSet=rs0"
      included_collections: "production[.]*"
      max_records: 200
      backlog_critical_threshold: 50
      max_wait_seconds: 10
      state_name: "mongodb-cdc-checkpoint"

tasks:
  - id: probe_cdc_backlog
    type: io.kestra.plugin.debezium.mongodb.Capture
    hostname: "{{ inputs.hostname }}"
    port: "{{ inputs.port }}"
    connectionString: "{{ inputs.connection_string }}"
    snapshotMode: NO_DATA
    includedCollections: "{{ inputs.included_collections }}"
    maxRecords: "{{ inputs.max_records }}"
    maxWait: "PT{{ inputs.max_wait_seconds }}S"
    stateName: "{{ inputs.state_name }}"
    format: INLINE
    allowFailure: true
    description: Drains pending change data capture records from MongoDB change
      streams into internal storage and commits the resume token.

  - id: classify_cdc_posture
    type: io.kestra.plugin.core.debug.Return
    format: >-
      {{
        (outputs.probe_cdc_backlog is not defined or outputs.probe_cdc_backlog.size is not defined) ? 'CDC_PROBE_FAILED' :
        (outputs.probe_cdc_backlog.size >= inputs.backlog_critical_threshold) ? 'CRITICAL_OPLOG_LAG' :
        (outputs.probe_cdc_backlog.size > 0) ? 'ACTIVE_STREAMING' :
        'QUIET_HEALTHY'
      }}
    description: Evaluates change event volume against critical lag thresholds to
      classify replication health.

  - id: route_posture_action
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify_cdc_posture.value }}"
    cases:
      CRITICAL_OPLOG_LAG:
        - id: alert_critical_oplog_lag
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: "{{ {'text': '*[CRITICAL]: MongoDB CDC Oplog Lag Breach*

            • *Endpoint:* `' ~ inputs.connection_string ~ '`

            • *Captured Events:* ' ~ outputs.probe_cdc_backlog.size ~ '
            (Threshold: ' ~ inputs.backlog_critical_threshold ~ ')

            • *State Checkpoint:* `' ~ (outputs.probe_cdc_backlog.stateOffsetKey
            ?? 'None') ~ '`

            • *Execution ID:* `' ~ execution.id ~ '`

            *Action Required:* Downstream consumers are falling behind oplog
            velocity; investigate consumer stalls before ChangeStreamHistoryLost
            occurs.'} | toJson }}"

      CDC_PROBE_FAILED:
        - id: alert_cdc_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: "{{ {'text': '*[ALERT]: MongoDB CDC Connection Failure*

            • *Endpoint:* `' ~ inputs.connection_string ~ '`

            • *Error:* Unable to establish Debezium Change Stream connection
            with MongoDB replica set.

            • *Execution ID:* `' ~ execution.id ~ '`'} | toJson }}"

      ACTIVE_STREAMING:
        - id: log_active_streaming
          type: io.kestra.plugin.core.log.Log
          message: "MongoDB CDC change stream active for {{ inputs.connection_string }}.
            Captured {{ outputs.probe_cdc_backlog.size }} change event(s) across
            monitored collections."

      QUIET_HEALTHY:
        - id: log_quiet_healthy
          type: io.kestra.plugin.core.log.Log
          message: "MongoDB CDC change stream healthy for {{ inputs.connection_string }}.
            Zero pending changes in oplog polling window."
    description: Dispatches high-priority Slack notifications or logs routine
      metrics based on classified status.

  - id: audit_sentinel_run
    type: io.kestra.plugin.core.log.Log
    message: "MongoDB CDC sentinel cycle finished for execution {{ execution.id }}.
      Verdict: {{ outputs.classify_cdc_posture.value }}, Captured records: {{
      outputs.probe_cdc_backlog.size ?? 0 }}, Checkpoint key: {{
      outputs.probe_cdc_backlog.stateOffsetKey ?? 'N/A' }}."
    description: Writes an immutable audit log entry documenting replication status
      and checkpoint keys.

errors:
  - id: alert_flow_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: "{{ {'text': '*[ERROR] MongoDB CDC Sentinel Execution Error*

      Flow failed during execution `' ~ execution.id ~ '` against `' ~
      inputs.connection_string ~ '`. Inspect Kestra logs for details.'} | toJson
      }}"