Audit PostgreSQL Slow Queries and Missing Indexes with Kestra

id: postgresql-slow-query-index-advisor
namespace: company.team
description: |
  Audit PostgreSQL performance from pg_stat_statements: list the slowest queries
  over a mean-time floor and the tables where sequential scans dominate, then
  post one Slack advisory. Runs weekly or on demand via webhook.

inputs:
  - id: min_mean_ms
    type: INT
    displayName: Minimum mean time (ms)
    description: Only report queries whose mean execution time is at least this many
      milliseconds.
    defaults: 100

  - id: min_calls
    type: INT
    displayName: Minimum calls
    description: Ignore queries that ran fewer times than this, so one-off
      statements do not alert.
    defaults: 5

  - id: top_n
    type: INT
    displayName: Rows per list
    description: Maximum number of slow queries and index candidates to include in
      the report.
    defaults: 10

  - id: min_table_rows
    type: INT
    displayName: Minimum table rows
    description: Only suggest indexes for tables with at least this many estimated rows.
    defaults: 10000

tasks:
  - id: fetch_slow_queries
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: The slowest statements over the thresholds, ordered by cumulative time.
    url: "{{ secret('POSTGRES_JDBC_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        left(regexp_replace(query, E'[\n\r]+', ' ', 'g'), 160) AS query,
        calls,
        round(mean_exec_time::numeric, 1) AS mean_ms,
        round(total_exec_time::numeric, 1) AS total_ms
      FROM pg_stat_statements
      WHERE mean_exec_time >= {{ inputs.min_mean_ms }}
        AND calls >= {{ inputs.min_calls }}
        AND query NOT ILIKE '%pg_stat_statements%'
        AND query NOT ILIKE 'SET %'
        AND query NOT ILIKE 'BEGIN%'
      ORDER BY total_exec_time DESC
      LIMIT {{ inputs.top_n }}

  - id: fetch_totals
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Aggregate stats for the same filters so the alert can report totals
      without parsing rows.
    url: "{{ secret('POSTGRES_JDBC_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        count(*) AS slow_count,
        coalesce(round(sum(mean_exec_time * calls) / 1000.0, 1), 0) AS est_total_s
      FROM pg_stat_statements
      WHERE mean_exec_time >= {{ inputs.min_mean_ms }}
        AND calls >= {{ inputs.min_calls }}
        AND query NOT ILIKE '%pg_stat_statements%'
        AND query NOT ILIKE 'SET %'
        AND query NOT ILIKE 'BEGIN%'

  - id: fetch_index_candidates
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Tables big enough to matter where sequential scans outnumber index scans.
    url: "{{ secret('POSTGRES_JDBC_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        relname AS table_name,
        seq_scan,
        coalesce(idx_scan, 0) AS idx_scan,
        n_live_tup AS est_rows
      FROM pg_stat_user_tables
      WHERE seq_scan > coalesce(idx_scan, 0)
        AND n_live_tup >= {{ inputs.min_table_rows }}
      ORDER BY seq_scan DESC
      LIMIT {{ inputs.top_n }}

  - id: audit_gate
    type: io.kestra.plugin.core.flow.If
    description: Send the advisory when either list found something, log a quiet
      all-clear otherwise.
    condition: "{{ outputs.fetch_slow_queries.size > 0 or
      outputs.fetch_index_candidates.size > 0 }}"
    then:
      - id: notify_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :snail: PostgreSQL slow-query audit - {{ flow.namespace }}.{{ flow.id }} (execution {{ execution.id }}).
          Slow queries (>= {{ inputs.min_mean_ms }} ms mean, >= {{ inputs.min_calls }} calls): {{ outputs.fetch_totals.row.slow_count }}, {{ outputs.fetch_totals.row.est_total_s }} s cumulative time measured.
          {% for row in outputs.fetch_slow_queries.rows %}- {{ row.query }}
          {{ row.calls }} calls | {{ row.mean_ms }} ms mean | {{ row.total_ms }} ms total
          {% endfor %}
          Seq-scan index candidates (>= {{ inputs.min_table_rows }} rows): {{ outputs.fetch_index_candidates.size }}
          {% for row in outputs.fetch_index_candidates.rows %}- {{ row.table_name }}: {{ row.seq_scan }} seq scans vs {{ row.idx_scan }} index scans over ~{{ row.est_rows }} rows - check the WHERE predicates for a missing index
          {% endfor %}
          Next step: EXPLAIN ANALYZE the top offenders and add indexes CONCURRENTLY during an off-peak window.
    else:
      - id: log_clean
        type: io.kestra.plugin.core.log.Log
        message: No slow queries over {{ inputs.min_mean_ms }} ms and no seq-scan-heavy
          tables over {{ inputs.min_table_rows }} rows. Nothing to advise.

outputs:
  - id: slow_query_count
    type: INT
    value: "{{ outputs.fetch_slow_queries.size }}"
  - id: index_candidate_count
    type: INT
    value: "{{ outputs.fetch_index_candidates.size }}"
  - id: slowest_query
    type: STRING
    value: "{{ outputs.fetch_slow_queries.rows | length > 0 ?
      outputs.fetch_slow_queries.rows[0].query : 'none' }}"
  - id: cumulative_time_seconds
    type: FLOAT
    value: "{{ outputs.fetch_totals.row.est_total_s }}"

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":rotating_light: Slow-query audit {{ flow.id }} FAILED (execution
      {{ execution.id }}). Check connectivity to PostgreSQL and that the
      pg_stat_statements extension is installed. Error: {{ error.message }}"

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly audit on Monday morning. Enable when ready.
    cron: "0 6 * * 1"
    disabled: true

  - id: on_demand
    type: io.kestra.plugin.core.trigger.Webhook
    description: Trigger the audit from monitoring or a runbook. Replace the key
      with your own value.
    key: replace-with-a-random-key