IBM Db2 CDC Stream and Change Table Backlog Sentinel

id: db2-cdc-stream-and-change-table-backlog-sentinel
namespace: company.dataops

inputs:
  - id: db2_hostname
    type: STRING
    defaults: "localhost"
    description: Hostname or IP address of the IBM Db2 LUW database server.

  - id: db2_port
    type: STRING
    defaults: "50000"
    description: Port number for the Db2 database instance (standard default 50000).

  - id: db2_database
    type: STRING
    defaults: "sample"
    description: Name of the Db2 database to inspect for active Change Data Capture
      (CDC) events.

  - id: max_backlog_threshold
    type: INT
    defaults: 250
    description: Threshold of unconsumed change records within a single audit probe
      before signaling an ingestion backlog surge.

  - id: probe_max_records
    type: INT
    defaults: 500
    description: Maximum number of CDC change records to drain during the diagnostic
      probe cycle.

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */3 * * *"
    disabled: true
    inputs:
      db2_hostname: "localhost"
      db2_port: "50000"
      db2_database: "sample"
      max_backlog_threshold: 250
      probe_max_records: 500

tasks:
  - id: probe_cdc_stream
    type: io.kestra.plugin.debezium.db2.Capture
    hostname: "{{ inputs.db2_hostname }}"
    port: "{{ inputs.db2_port }}"
    username: "{{ secret('DB2_USERNAME') }}"
    password: "{{ secret('DB2_PASSWORD') }}"
    database: "{{ inputs.db2_database }}"
    snapshotMode: INITIAL
    maxRecords: "{{ inputs.probe_max_records }}"
    allowFailure: true
    description: Connects to Db2 transaction logs via Debezium engine to sample
      active change records and verify capture pipeline throughput.

  - id: classify_cdc_posture
    type: io.kestra.plugin.core.debug.Return
    format: >-
      {{
        (outputs.probe_cdc_stream is not defined or outputs.probe_cdc_stream.size is not defined) ? 'DB2_CONNECTION_FAILED' :
        (outputs.probe_cdc_stream.size >= inputs.max_backlog_threshold) ? 'HIGH_BACKLOG_SURGE' :
        (outputs.probe_cdc_stream.size == 0) ? 'ZERO_EVENT_IDLE' :
        'HEALTHY'
      }}
    description: Evaluates captured event volume against operational thresholds
      using Pebble logic to determine CDC pipeline health.

  - id: route_posture_action
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify_cdc_posture.value }}"
    cases:
      HIGH_BACKLOG_SURGE:
        - id: alert_backlog_surge
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: "{{ {'text': '🚨 *CRITICAL: IBM Db2 CDC Event Backlog Surge Detected*

            • *Database:* `' ~ inputs.db2_database ~ '`

            • *Endpoint:* `' ~ inputs.db2_hostname ~ ':' ~ inputs.db2_port ~ '`

            • *Captured Records:* ' ~ outputs.probe_cdc_stream.size ~ '
            (Threshold: ' ~ inputs.max_backlog_threshold ~ ')

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

            *Action Required:* Downstream CDC consumer lag or uncommitted
            transaction spikes detected; inspect Db2 transaction log archive
            headroom.'} | toJson }}"

      ZERO_EVENT_IDLE:
        - id: log_cdc_idle
          type: io.kestra.plugin.core.log.Log
          message: "Db2 CDC stream on '{{ inputs.db2_database }}' returned 0 records.
            System is idle or current offset is fully caught up."

      DB2_CONNECTION_FAILED:
        - id: alert_connection_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: "{{ {'text': '🔥 *ALERT: IBM Db2 Instance Unreachable*

            • *Endpoint:* `' ~ inputs.db2_hostname ~ ':' ~ inputs.db2_port ~ '`

            • *Database:* `' ~ inputs.db2_database ~ '`

            • *Error:* Unable to establish connection to Db2 server or
            initialize Debezium capture engine.

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

      HEALTHY:
        - id: log_cdc_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Db2 CDC stream for '{{ inputs.db2_database }}' is healthy. Processed
            {{ outputs.probe_cdc_stream.size }} events within normal operating
            headroom."
    description: Routes Slack incident notifications or diagnostic logs depending on
      evaluated change data capture health.

  - id: audit_sentinel_run
    type: io.kestra.plugin.core.log.Log
    message: "Db2 CDC sentinel cycle complete for execution {{ execution.id }}.
      Verdict: {{ outputs.classify_cdc_posture.value }}, Total Events Captured:
      {{ outputs.probe_cdc_stream.size ?? 0 }}, Storage Key: {{
      outputs.probe_cdc_stream.stateOffsetKey ?? 'N/A' }}."
    description: Emits an immutable audit log entry capturing CDC record counts and
      cycle verdicts.

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

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