ClickHouse Slow Query Threshold Gate with Kestra

id: clickhouse-slow-query-threshold-gate
namespace: company.team
description: |
  Query ClickHouse's own system.query_log for finished queries past a
  duration threshold and system.processes for currently active queries
  past an elapsed-time threshold, classify HEALTHY, AT_RISK, or
  UNREACHABLE (empty query log for the window), alert Slack, and
  optionally run KILL QUERY on the single longest-running active query.
  auto_remediate and dry_run are two independent gates: no table data is
  ever read, inserted, or deleted - only one running query can be killed.

triggers:
  - id: periodic_slow_query_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every 10 minutes - a stuck query or a burst of slow queries
      needs a tighter cadence than a schema or replication check. Shipped
      disabled; enable it once clickhouse_url/thresholds point at a real cluster
      and a manual run has been verified. Always fires in safe mode
      (auto_remediate false, dry_run true) regardless of this flow's own
      defaults.
    cron: "*/10 * * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: clickhouse_url
    type: STRING
    displayName: "ClickHouse JDBC URL"
    defaults: "jdbc:clickhouse://localhost:8124/default"
    description: JDBC connection URL for the target ClickHouse instance.

  - id: lookback_minutes
    type: INT
    displayName: "Lookback Window (minutes)"
    defaults: 60
    description: How far back in system.query_log.event_time to scan for finished
      queries. A window with zero finished queries at all is reported as
      UNREACHABLE.

  - id: slow_query_threshold_ms
    type: INT
    displayName: "Slow Query Threshold (ms)"
    defaults: 5000
    description: A finished query (system.query_log, type = 'QueryFinish') with
      query_duration_ms above this counts toward slow_query_count.

  - id: long_running_threshold_seconds
    type: INT
    displayName: "Active Query Elapsed Threshold (seconds)"
    defaults: 60
    description: A currently active query (system.processes.elapsed) above this
      counts as a long-running query, and is the only thing auto_remediate can
      act on.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and an active query is past
      long_running_threshold_seconds, run KILL QUERY on the single
      longest-running one (still gated by dry_run below). Has no effect when the
      AT_RISK state is caused only by historical slow queries in
      system.query_log, since those have already finished and there is nothing
      left to kill.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", remediation is only logged, KILL QUERY never runs. Must
      be explicitly "false", together with auto_remediate "true", for the query
      to actually be killed.

tasks:
  - id: check_historical_slow_queries
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: "system.query_log is ClickHouse's own system table (columns type,
      event_time, query_duration_ms, query_id verified from the official
      system-tables documentation). type = 'QueryFinish' selects only
      successfully completed queries, filtered to the lookback window."
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ secret('CLICKHOUSE_USERNAME') }}"
    password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        count(*) AS total_finished_queries,
        countIf(query_duration_ms > {{ inputs.slow_query_threshold_ms }}) AS slow_query_count,
        coalesce(max(query_duration_ms), 0) AS max_duration_ms
      FROM system.query_log
      WHERE type = 'QueryFinish'
        AND event_time >= now() - INTERVAL {{ inputs.lookback_minutes }} MINUTE

  - id: check_active_long_queries
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: "system.processes is ClickHouse's own system table of currently
      executing queries (columns query_id, elapsed, read_rows, memory_usage
      verified from the official system-tables documentation). Picks the single
      longest-elapsed active query's query_id through a scalar subquery so one
      statement returns both the count and the remediation target."
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ secret('CLICKHOUSE_USERNAME') }}"
    password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        count(*) AS active_long_count,
        coalesce(max(elapsed), 0) AS max_elapsed_seconds,
        (
          SELECT query_id
          FROM system.processes
          WHERE elapsed > {{ inputs.long_running_threshold_seconds }}
          ORDER BY elapsed DESC
          LIMIT 1
        ) AS oldest_active_query_id
      FROM system.processes
      WHERE elapsed > {{ inputs.long_running_threshold_seconds }}

  - id: evaluate_log_activity
    type: io.kestra.plugin.core.flow.If
    description: UNREACHABLE when system.query_log has zero QueryFinish rows for the
      entire lookback window - either the cluster has had no traffic, it was
      just restarted, or query logging is disabled, all of which make the
      slow-query count meaningless for this run.
    condition: "{{ outputs.check_historical_slow_queries.row.total_finished_queries
      > 0 }}"
    then:
      - id: evaluate_query_risk
        type: io.kestra.plugin.core.flow.If
        description: AT_RISK when any finished query exceeded slow_query_threshold_ms in
          the lookback window, or any currently active query has exceeded
          long_running_threshold_seconds.
        condition: "{{ outputs.check_historical_slow_queries.row.slow_query_count > 0 or
          outputs.check_active_long_queries.row.active_long_count > 0 }}"
        then:
          - id: route_remediation
            type: io.kestra.plugin.core.flow.Switch
            description: auto_remediate only decides whether remediation is considered at
              all; dry_run, checked separately below, decides whether KILL QUERY
              actually runs.
            value: "{{ inputs.auto_remediate }}"
            cases:
              "true":
                - id: check_has_active_query
                  type: io.kestra.plugin.core.flow.If
                  description: Only a currently active query can be remediated by this flow - a
                    finished slow query in system.query_log has already
                    completed and cannot be killed.
                  condition: "{{ outputs.check_active_long_queries.row.oldest_active_query_id is
                    defined }}"
                  then:
                    - id: check_dry_run
                      type: io.kestra.plugin.core.flow.If
                      condition: "{{ inputs.dry_run == \"false\" }}"
                      then:
                        - id: kill_long_running_query
                          type: io.kestra.plugin.jdbc.clickhouse.Query
                          description: "KILL QUERY WHERE query_id = '...' is ClickHouse's own statement
                            (verified from the official KILL statements
                            documentation). Targets only oldest_active_query_id
                            - never any other running query."
                          url: "{{ inputs.clickhouse_url }}"
                          username: "{{ secret('CLICKHOUSE_USERNAME') }}"
                          password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
                          fetchType: NONE
                          sql: "KILL QUERY WHERE query_id = '{{
                            outputs.check_active_long_queries.row.oldest_active\
                            _query_id }}'"
                        - id: log_remediation_applied
                          type: io.kestra.plugin.core.log.Log
                          message: "Killed query {{
                            outputs.check_active_long_queries.row.oldest_active_qu\
                            ery_id }} (elapsed {{
                            outputs.check_active_long_queries.row.max_elapsed_s\
                            econds }}s)."
                      else:
                        - id: log_dry_run_remediation
                          type: io.kestra.plugin.core.log.Log
                          message: "DRY RUN: would kill query {{
                            outputs.check_active_long_queries.row.oldest_active\
                            _query_id }} (elapsed {{
                            outputs.check_active_long_queries.row.max_elapsed_s\
                            econds }}s, dry_run is true). No query was killed."
                  else:
                    - id: log_no_active_query
                      type: io.kestra.plugin.core.log.Log
                      message: "AT_RISK with no active query to kill: the risk is historical slow
                        queries in system.query_log (slow_query_count={{
                        outputs.check_historical_slow_queries.row.slow_query_co\
                        unt }}, max_duration_ms={{
                        outputs.check_historical_slow_queries.row.max_duration_\
                        ms }}), which this flow cannot remediate retroactively."
              "false":
                - id: log_remediation_disabled
                  type: io.kestra.plugin.core.log.Log
                  message: "auto_remediate is false: slow_query_count={{
                    outputs.check_historical_slow_queries.row.slow_query_count
                    }}, active_long_count={{
                    outputs.check_active_long_queries.row.active_long_count }},
                    but no remediation was attempted."
          - id: alert_query_risk
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Always alert on AT_RISK, independent of whether remediation ran or
              was even possible.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "ClickHouse slow query gate: AT_RISK. Last {{ inputs.lookback_minutes }}m: slow_query_count={{ outputs.check_historical_slow_queries.row.slow_query_count }} (threshold {{ inputs.slow_query_threshold_ms }}ms, max {{ outputs.check_historical_slow_queries.row.max_duration_ms }}ms). Right now: active_long_count={{ outputs.check_active_long_queries.row.active_long_count }} (threshold {{ inputs.long_running_threshold_seconds }}s, max {{ outputs.check_active_long_queries.row.max_elapsed_seconds }}s). auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Remediated: {{ outputs.kill_long_running_query is defined }}. Execution {{ execution.id }}."
              }
        else:
          - id: log_query_health_healthy
            type: io.kestra.plugin.core.log.Log
            message: "HEALTHY: 0 of {{
              outputs.check_historical_slow_queries.row.total_finished_queries
              }} finished queries exceeded {{ inputs.slow_query_threshold_ms
              }}ms in the last {{ inputs.lookback_minutes }}m, and 0 active
              queries exceeded {{ inputs.long_running_threshold_seconds }}s."
    else:
      - id: alert_query_log_unreachable
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: UNREACHABLE means system.query_log returned zero QueryFinish rows
          for the entire lookback window - distinct from both HEALTHY and
          AT_RISK, and reported here rather than only through the errors block,
          since a non-matching WHERE clause does not throw.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "ClickHouse slow query gate: UNREACHABLE. system.query_log has 0 QueryFinish rows in the last {{ inputs.lookback_minutes }}m - the cluster may have no traffic, may have just restarted, or query logging (log_queries) may be disabled. Execution {{ execution.id }}."
          }

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Slow query audit complete. total_finished_queries={{
      outputs.check_historical_slow_queries.row.total_finished_queries }}.
      slow_query_count={{
      outputs.check_historical_slow_queries.row.slow_query_count }}.
      active_long_count={{
      outputs.check_active_long_queries.row.active_long_count }}.
      auto_remediate={{ inputs.auto_remediate }}. dry_run={{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable ClickHouse instance or invalid credentials.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "ClickHouse slow query gate FAILED in execution {{ execution.id }}. The query_log/processes audit did not complete; investigate before trusting the last known HEALTHY/AT_RISK/UNREACHABLE status."
      }

A query that takes ten seconds instead of one hundred milliseconds does not throw an error - it just quietly eats CPU, memory, and concurrency slots until enough of them pile up to degrade everything else on the cluster. This blueprint reads ClickHouse's own system.query_log for queries that have already finished slowly, and system.processes for queries that are still running past a time budget, classifies the cluster into a three-state risk level, alerts on anything but a clean bill of health, and can terminate the single longest-running active query with ClickHouse's own KILL QUERY statement, behind two independent gates.

How it works

periodic_slow_query_check (io.kestra.plugin.core.trigger.Schedule) runs every 10 minutes, always forcing auto_remediate: "false" and dry_run: "true" through its own inputs: override regardless of this flow's defaults. Shipped disabled so you can validate a manual run first.
check_historical_slow_queries (io.kestra.plugin.jdbc.clickhouse.Query, fetchType: FETCH_ONE) counts system.query_log rows with type = 'QueryFinish' in the lookback_minutes window, populating row.total_finished_queries, row.slow_query_count, row.max_duration_ms.
check_active_long_queries (io.kestra.plugin.jdbc.clickhouse.Query, fetchType: FETCH_ONE) counts system.processes rows with elapsed > long_running_threshold_seconds, and picks the longest-elapsed one's query_id through a scalar subquery in the same statement, populating row.active_long_count, row.max_elapsed_seconds, row.oldest_active_query_id.
evaluate_log_activity (io.kestra.plugin.core.flow.If) branches on total_finished_queries > 0.
- else (UNREACHABLE): alert_query_log_unreachable reports that the query log is empty for the whole window.
- then (log has data): nested evaluate_query_risk (io.kestra.plugin.core.flow.If) branches on slow_query_count > 0 or active_long_count > 0.
  - else (HEALTHY): log_query_health_healthy records a clean check.
  - then (AT_RISK): route_remediation (io.kestra.plugin.core.flow.Switch on auto_remediate) either considers remediation ("true", further gated by check_has_active_query - only a currently active query can be killed, not a finished one in the log - and then by dry_run through check_dry_run) or only logs that remediation is disabled ("false"). When every gate allows it, kill_long_running_query runs KILL QUERY WHERE query_id = 'oldest_active_query_id'. alert_query_risk always fires to Slack regardless of which remediation path ran.
log_audit_complete always runs last, printing the headline numbers regardless of which branch fired.
The errors block alerts Slack separately if the flow itself fails outright - an unreachable cluster or bad credentials must not read as "no risk".

What you get

A three-state answer (HEALTHY / AT_RISK / UNREACHABLE) sourced from ClickHouse's own query execution history and live process list, not from an external APM agent.
Both halves of the slow-query problem covered in one gate: queries that already finished slowly (a trend signal) and queries still running too long right now (an actionable signal).
Remediation scoped to exactly one KILL QUERY on the single longest-running active query, by query_id - it never kills an arbitrary or unrelated query, and never touches a query that has already finished.
Two independent gates (auto_remediate, dry_run) between "risk detected" and "query actually killed".
A final status log every run, risk or not, so the execution history doubles as a query-performance trend.

Who it's for

Platform and database reliability teams running ClickHouse who need an early signal when queries start running long, before concurrency limits or memory pressure turn it into an outage.
On-call engineers who want one Slack alert with the exact slow-query and active-query numbers instead of running SELECT * FROM system.processes by hand during an incident.
Anyone evaluating the ClickHouse plugin who wants a worked example of Query driving two different system tables (system.query_log, system.processes) plus a KILL administrative statement, distinct from this repo's own system.mutations/system.parts watchdog and system.replicas health gate.

Why orchestrate this with Kestra

system.query_log and system.processes run by hand answer one moment in time; they do not decide a cadence, classify the result into a three-state risk level, pick a single deterministic remediation target, gate a kill behind two independent confirmations, or notify anyone. Kestra supplies the schedule, the HEALTHY/AT_RISK/UNREACHABLE classification as a first-class branch, two independent authorization gates before any query is killed, and an execution history that shows exactly when query performance degraded and whether anything was done about it.

Prerequisites

A reachable ClickHouse instance with query logging enabled (log_queries, on by default per ClickHouse's own documentation).
A database user with SELECT on system.query_log and system.processes, and, if auto_remediate will ever be "true", permission to run KILL QUERY.
A Slack incoming webhook for alerts.

Local testing: docker run -d --name clickhouse-slow-query-gate -p 8124:8123 -p 9001:9000 clickhouse/clickhouse-server:latest docker network connect YOUR_KESTRA_NETWORK clickhouse-slow-query-gate (check existing networks first with docker network ls; only needed if the Kestra Worker runs in a separate Docker network than this container; set clickhouse_url to jdbc:clickhouse://clickhouse-slow-query-gate:8123/default from inside that network, or jdbc:clickhouse://localhost:8124/default from the host). Host port 8124 is used here specifically so it never collides with Kestra's own UI on 8080.

Secrets

CLICKHOUSE_USERNAME / CLICKHOUSE_PASSWORD: credentials used by every io.kestra.plugin.jdbc.clickhouse.Query task in this flow.
SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_query_risk, alert_query_log_unreachable, and the errors block.
In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_CLICKHOUSE_PASSWORD=$(echo -n '<your-local-password>' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

clickhouse_url (STRING, default jdbc:clickhouse://localhost:8124/default): JDBC URL.
lookback_minutes (INT, default 60): window scanned in system.query_log.
slow_query_threshold_ms (INT, default 5000): finished-query duration ceiling.
long_running_threshold_seconds (INT, default 60): active-query elapsed-time ceiling.
auto_remediate (SELECT: "false", "true"; default "false"): must be "true" for remediation to even be considered.
dry_run (SELECT: "true", "false"; default "true"): must be explicitly "false", together with auto_remediate: "true", for KILL QUERY to actually run.

Outputs

outputs.check_historical_slow_queries.row.total_finished_queries / .slow_query_count / .max_duration_ms: historical query metrics, every run.
outputs.check_active_long_queries.row.active_long_count / .max_elapsed_seconds / .oldest_active_query_id: live query metrics, every run; oldest_active_query_id is absent when active_long_count is 0.

Quick start

Start ClickHouse locally with the command above and run a deliberately slow query (for example a large cross join or SELECT sleep(10) repeated) to populate both system.query_log and, while it runs, system.processes.
Add the CLICKHOUSE_USERNAME, CLICKHOUSE_PASSWORD, and SLACK_WEBHOOK_URL secrets.
Run the flow manually with the defaults and confirm alert_query_risk fires once a query crosses either threshold.
Run once with auto_remediate: "true" and dry_run: "true" and confirm log_dry_run_remediation describes the kill without running it.
Run once with auto_remediate: "true" and dry_run: "false" while a long query is still active, and confirm kill_long_running_query runs, then verify with SELECT * FROM system.processes that it is gone.
Point lookback_minutes at a very short window on a quiet cluster and re-run to confirm alert_query_log_unreachable fires (UNREACHABLE) instead of either HEALTHY or AT_RISK.
Enable periodic_slow_query_check once you trust the check; it always runs in the safe auto_remediate: "false" / dry_run: "true" mode regardless of what you leave the flow's own defaults set to.

How to extend

Add a breakdown by user or current_database from system.query_log to the Slack alert, so the team that owns the slow query is named, not just the count.
Add exception_code/exception from system.query_log (type IN ('ExceptionBeforeStart', 'ExceptionWhileProcessing')) as a fourth signal alongside slow and long-running queries.
Loop over several clickhouse_url values with io.kestra.plugin.core.flow.Loop to run the same gate across every node in a cluster, since system.processes is local to the node it is queried on.
Route the UNREACHABLE branch to a different Slack channel than AT_RISK, since an empty query log after a known-busy period often signals a logging or connectivity problem, not an idle cluster.

Pitfalls

UNREACHABLE here means "the query log has nothing for this window", not "the cluster is down". A fully unreachable ClickHouse instance fails the Query task's own connection and is caught by the errors block, not by evaluate_log_activity - that branch only fires when the query itself succeeds but returns zero QueryFinish rows.
A finished slow query cannot be killed. check_has_active_query exists specifically so an AT_RISK state caused only by historical entries in system.query_log is reported as unremediable, rather than silently doing nothing while claiming a kill was considered.
system.processes is per-node, not cluster-wide. On a multi-node cluster, this flow's check_active_long_queries only sees queries currently running on the node named in clickhouse_url - see How to extend for looping across nodes.
There is no structured "query log" or "kill query" task in this plugin. Verified from this repo's own existing ClickHouse flows (system.mutations, system.parts, system.replicas, system.users): all system-table introspection and all SYSTEM/KILL administrative statements go through the same generic io.kestra.plugin.jdbc.clickhouse.Query task.
auto_remediate and dry_run are independent gates, not a single boolean. Both must be "true"/"false" respectively for KILL QUERY to run; setting only one leaves the other still blocking.
Switch case keys "true"/"false" are quoted strings. auto_remediate renders as the literal string "true" or "false"; unquoted true:/false: YAML map keys would parse as booleans instead and would not match.
This blueprint is UNTESTED against a live ClickHouse cluster. The system.query_log and system.processes column names are taken verbatim from ClickHouse's own system-tables documentation, and KILL QUERY syntax from ClickHouse's own SQL reference - but no Docker container or Kestra engine was run to execute this flow end to end.

ClickHouse Slow Query Threshold Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

ClickHouse Slow Query Threshold Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Build a Chat-With-Your-Data System Using Elasticsearch and OpenAI (RAG)

Chroma Vector Store Dangling Document and Quarantine Eviction Guard

Incremental Docs-to-Chroma RAG Sync with Deletions and a Retrieval Smoke Test

Citus Colocation Violation Guard

Chroma Collection Distance Metric Validator

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

More Infrastructure Blueprints

Kafka ACL Baseline Restore Guard

Kafka Connect Paused Connector Resume Guard

Kafka Consumer Lag Sentinel

Kafka Connect Sink Poison Pill Offset Rewind and Skip Gate

Self-heal Kafka Connect connectors with a restart budget and escalation

Kafka ACL Overly Broad Grant Audit

New to Kestra?