ClickHouse Asynchronous Insert Buffer Queue and Backpressure Sentinel

id: clickhouse-async-insert-buffer-sentinel
namespace: company.data

description: |
  Monitor ClickHouse asynchronous insert buffer memory and queue saturation using system.asynchronous_inserts.
  Detects delayed buffer flushes and ingestion backpressure, exports a downloadable CSV diagnostic report,
  gates operator-approved queue flushing via Pause, verifies buffer drain status, records telemetry snapshots
  in Kestra KV store, and dispatches automated Slack incident alerts to Data SREs.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: max_buffer_memory_mb
    type: FLOAT
    displayName: Maximum In-Memory Buffer Threshold (MB)
    description: Cumulative unwritten data threshold in megabytes held across
      asynchronous insert buffers. Exceeding this threshold signals disk flush
      bottlenecks.
    defaults: 250.0
    min: 10.0
    max: 100000.0

  - id: max_flush_delay_ms
    type: INT
    displayName: Maximum Flush Delay Threshold (ms)
    description: Maximum acceptable age in milliseconds of the oldest pending
      asynchronous batch waiting to flush to disk.
    defaults: 5000
    min: 500
    max: 600000

  - id: max_reported_tables
    type: INT
    displayName: Maximum Tables to Report
    description: Maximum number of congested tables to extract into the detailed
      diagnostic CSV summary.
    defaults: 25
    min: 1
    max: 100

  - id: clickhouse_database
    type: STRING
    displayName: ClickHouse Target Database
    description: Target database name to scope asynchronous insert buffer monitoring.
    defaults: default
    validator: ^[A-Za-z_][A-Za-z0-9_]{0,63}$

  - id: auto_flush_queue
    type: BOOL
    displayName: Enable Remediation Approval Gate
    description: When true, prompts an operator for manual approval before executing
      SYSTEM FLUSH ASYNC INSERT QUEUE to drain backpressured buffers to disk.
    defaults: true

tasks:
  - id: evaluate_buffer_saturation
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Query ClickHouse system.asynchronous_inserts to evaluate global
      buffer memory and pending batch latencies.
    url: "{{ secret('CLICKHOUSE_JDBC_URL') }}"
    username: "{{ secret('CLICKHOUSE_USERNAME') }}"
    password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
    timeZoneId: UTC
    fetchType: FETCH_ONE
    sql: |
      SELECT
        count(*) AS pending_batches_count,
        COALESCE(round(sum(bytes) / (1024 * 1024), 2), 0.0) AS total_buffered_mb,
        COALESCE(round(sum(bytes) / (1024 * 1024 * 1024), 3), 0.0) AS total_buffered_gb,
        COALESCE(sum(entries), 0) AS total_buffered_entries,
        COALESCE(max(dateDiff('millisecond', first_update, now())), 0) AS peak_flush_delay_ms,
        CASE
          WHEN (COALESCE(sum(bytes) / (1024 * 1024), 0.0) >= {{ inputs.max_buffer_memory_mb }})
            OR (COALESCE(max(dateDiff('millisecond', first_update, now())), 0) >= {{ inputs.max_flush_delay_ms }})
          THEN true
          ELSE false
        END AS is_saturated
      FROM system.asynchronous_inserts
      WHERE database = '{{ inputs.clickhouse_database }}';

  - id: fetch_saturated_tables_telemetry
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Stream detailed table-level buffer metrics and pending flush ages
      to internal storage.
    url: "{{ secret('CLICKHOUSE_JDBC_URL') }}"
    username: "{{ secret('CLICKHOUSE_USERNAME') }}"
    password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
    timeZoneId: UTC
    fetchType: STORE
    sql: |
      SELECT
        database,
        table,
        format,
        query_id,
        entries AS pending_rows,
        round(bytes / (1024 * 1024), 2) AS buffered_mb,
        first_update AS batch_opened_at,
        dateDiff('millisecond', first_update, now()) AS age_ms
      FROM system.asynchronous_inserts
      WHERE database = '{{ inputs.clickhouse_database }}'
      ORDER BY bytes DESC
      LIMIT {{ inputs.max_reported_tables }};

  - id: export_diagnostic_csv
    type: io.kestra.plugin.serdes.csv.IonToCsv
    description: Serialize the asynchronous insert telemetry into a downloadable CSV
      artifact.
    from: "{{ outputs.fetch_saturated_tables_telemetry.uri }}"

  - id: check_backpressure_and_alert
    type: io.kestra.plugin.core.flow.If
    description: Evaluate buffer health metrics, notify Slack on congestion, and
      gate operator-approved queue flush remediation.
    condition: "{{ outputs.evaluate_buffer_saturation.row.is_saturated }}"
    then:
      - id: log_backpressure_warning
        type: io.kestra.plugin.core.log.Log
        message: >-
          ClickHouse Ingestion Backpressure Warning on database '{{
          inputs.clickhouse_database }}': Total Buffered Memory: {{
          outputs.evaluate_buffer_saturation.row.total_buffered_mb }} MB
          (Threshold: {{ inputs.max_buffer_memory_mb }} MB). Pending Batches: {{
          outputs.evaluate_buffer_saturation.row.pending_batches_count }}
          batches ({{
          outputs.evaluate_buffer_saturation.row.total_buffered_entries }}
          rows). Peak Flush Delay: {{
          outputs.evaluate_buffer_saturation.row.peak_flush_delay_ms }} ms
          (Threshold: {{ inputs.max_flush_delay_ms }} ms). Diagnostic CSV Report
          URI: {{ outputs.export_diagnostic_csv.uri }}

      - id: notify_slack_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Send incident alert to Slack with buffer metrics and report artifact.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: >-
          [INGESTION BACKPRESSURE] *ClickHouse Asynchronous Insert Buffer
          Saturation Alert* *Database:* `{{ inputs.clickhouse_database }}`
          *Buffered In-Memory Data:* `{{
          outputs.evaluate_buffer_saturation.row.total_buffered_mb }} MB`
          (Threshold: `{{ inputs.max_buffer_memory_mb }} MB`) *Pending Batches:*
          `{{ outputs.evaluate_buffer_saturation.row.pending_batches_count }}`
          (`{{ outputs.evaluate_buffer_saturation.row.total_buffered_entries }}
          rows`) *Oldest Batch Flush Delay:* `{{
          outputs.evaluate_buffer_saturation.row.peak_flush_delay_ms }} ms`
          (SLA: <= `{{ inputs.max_flush_delay_ms }} ms`) *Diagnostic CSV
          Report:* `{{ outputs.export_diagnostic_csv.uri }}` *Execution ID:* `{{
          execution.id }}`

      - id: execute_queue_remediation
        type: io.kestra.plugin.core.flow.If
        description: Operator-approved queue flush remediation gate triggered when
          buffer backpressure is detected.
        condition: "{{ inputs.auto_flush_queue }}"
        then:
          - id: pause_for_operator_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause workflow execution for operator review before flushing the
              asynchronous insert queue. Resume in the Kestra UI to flush, or
              cancel the execution to skip.
            pauseDuration: PT10M
            behavior: CANCEL

          - id: flush_async_insert_queue
            type: io.kestra.plugin.jdbc.clickhouse.Query
            description: Force synchronous flush of all accumulated asynchronous insert
              buffers to disk across the cluster.
            url: "{{ secret('CLICKHOUSE_JDBC_URL') }}"
            username: "{{ secret('CLICKHOUSE_USERNAME') }}"
            password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
            timeZoneId: UTC
            sql: SYSTEM FLUSH ASYNC INSERT QUEUE;

          - id: verify_queue_drained
            type: io.kestra.plugin.jdbc.clickhouse.Query
            description: Query system.asynchronous_inserts post-flush to verify buffered
              entries and unwritten bytes drained.
            url: "{{ secret('CLICKHOUSE_JDBC_URL') }}"
            username: "{{ secret('CLICKHOUSE_USERNAME') }}"
            password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
            timeZoneId: UTC
            fetchType: FETCH_ONE
            sql: |
              SELECT
                count(*) AS remaining_batches_count,
                COALESCE(round(sum(bytes) / (1024 * 1024), 2), 0.0) AS remaining_buffered_mb,
                COALESCE(sum(entries), 0) AS remaining_buffered_entries
              FROM system.asynchronous_inserts
              WHERE database = '{{ inputs.clickhouse_database }}';

          - id: log_remediation_summary
            type: io.kestra.plugin.core.log.Log
            message: >-
              ClickHouse Async Insert Queue Remediation Complete on database '{{
              inputs.clickhouse_database }}': Remaining Batches: {{
              outputs.verify_queue_drained.row.remaining_batches_count }}.
              Remaining Buffered Memory: {{
              outputs.verify_queue_drained.row.remaining_buffered_mb }} MB ({{
              outputs.verify_queue_drained.row.remaining_buffered_entries }}
              rows).

          - id: notify_remediation_success
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Send queue flush resolution notification to Slack with
              post-remediation metrics.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: >-
              [REMEDIATION SUCCESS] *ClickHouse Async Insert Queue Flushed*
              *Database:* `{{ inputs.clickhouse_database }}` *Remaining
              Batches:* `{{
              outputs.verify_queue_drained.row.remaining_batches_count }}`
              *Remaining Buffered Memory:* `{{
              outputs.verify_queue_drained.row.remaining_buffered_mb }} MB` (`{{
              outputs.verify_queue_drained.row.remaining_buffered_entries }}
              rows`) *Status:* Asynchronous insert buffer drained to disk
              successfully.
    else:
      - id: log_healthy_state
        type: io.kestra.plugin.core.log.Log
        message: >-
          ClickHouse asynchronous insert buffers on database '{{
          inputs.clickhouse_database }}' are healthy. Total buffered memory is
          {{ outputs.evaluate_buffer_saturation.row.total_buffered_mb }} MB (<
          {{ inputs.max_buffer_memory_mb }} MB) and peak flush delay is {{
          outputs.evaluate_buffer_saturation.row.peak_flush_delay_ms }} ms (< {{
          inputs.max_flush_delay_ms }} ms).

  - id: update_telemetry_watermark
    type: io.kestra.plugin.core.kv.Set
    description: Record telemetry snapshot in Kestra KV store for historical
      capacity and flush trend tracking.
    key: "clickhouse_async_insert_audit_{{ inputs.clickhouse_database }}"
    value:
      last_audit_timestamp: "{{ now() }}"
      pending_batches_count: "{{ outputs.evaluate_buffer_saturation.row.pending_batches_count }}"
      total_buffered_mb: "{{ outputs.evaluate_buffer_saturation.row.total_buffered_mb }}"
      peak_flush_delay_ms: "{{ outputs.evaluate_buffer_saturation.row.peak_flush_delay_ms }}"
      is_saturated: "{{ outputs.evaluate_buffer_saturation.row.is_saturated }}"
      diagnostic_csv_uri: "{{ outputs.export_diagnostic_csv.uri }}"

errors:
  - id: alert_audit_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Notify on task failure if database connectivity or query execution fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: >-
      [ERROR] ClickHouse Asynchronous Insert Audit failed in execution `{{
      execution.id }}` on database `{{ inputs.clickhouse_database }}`. Failed
      task: `{{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown'
      }}`. Error message: `{{ (errorLogs() | length) > 0 ?
      errorLogs()[0].message : 'inspect execution logs' }}`.

triggers:
  - id: periodic_buffer_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"