PostgreSQL Dead Tuple Bloat and Autovacuum Sentinel

id: postgres-dead-tuple-bloat-sentinel
namespace: company.team
description: |
  Monitor PostgreSQL table bloat and dead tuple accumulation via pg_stat_user_tables,
  export diagnostic metrics into execution outputs, and alert engineering and DBA
  teams via Slack when dead tuple ratios breach operational thresholds.

triggers:
  - id: scheduled_sentinel
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic off-peak bloat audit. Shipped disabled by default.
    cron: "0 3 * * *"
    disabled: true

  - id: manual_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook trigger for on-demand execution from CI/CD,
      maintenance scripts, or ChatOps.
    key: "{{ secret('WEBHOOK_KEY') }}"

inputs:
  - id: postgres_host
    type: STRING
    displayName: PostgreSQL Host
    description: Hostname or IP address of the target PostgreSQL database instance.
    defaults: postgres

  - id: postgres_port
    type: INT
    displayName: PostgreSQL Port
    description: Port number for the PostgreSQL connection.
    defaults: 5432

  - id: database_name
    type: STRING
    displayName: Database Name
    description: Name of the PostgreSQL database to inspect.
    defaults: postgres

  - id: dead_tuple_threshold
    type: INT
    displayName: Dead Tuple Minimum Threshold
    description: Minimum number of dead tuples required before a table is flagged
      for bloat inspection.
    defaults: 10000

  - id: dead_ratio_threshold_percent
    type: FLOAT
    displayName: Dead Ratio Threshold (%)
    description: Minimum percentage of dead tuples relative to total tuples required
      to trigger an alert.
    defaults: 20.0

tasks:
  - id: inspect_table_bloat
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Query pg_stat_user_tables to detect unvacuumed dead tuples and
      calculate bloat ratio percentages.
    url: "jdbc:postgresql://{{ inputs.postgres_host }}:{{ inputs.postgres_port }}/{{
      inputs.database_name }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        schemaname,
        relname AS table_name,
        n_live_tup AS live_tuples,
        n_dead_tup AS dead_tuples,
        ROUND(
          CASE
            WHEN (n_live_tup + n_dead_tup) > 0
            THEN (n_dead_tup::numeric / (n_live_tup + n_dead_tup)::numeric) * 100.0
            ELSE 0.0
          END, 2
        ) AS dead_ratio_percent,
        COALESCE(TO_CHAR(last_autovacuum, 'YYYY-MM-DD HH24:MI:SS'), 'Never') AS last_autovacuum,
        COALESCE(TO_CHAR(last_vacuum, 'YYYY-MM-DD HH24:MI:SS'), 'Never') AS last_manual_vacuum
      FROM pg_stat_user_tables
      WHERE n_dead_tup >= {{ inputs.dead_tuple_threshold }}
        AND CASE
          WHEN (n_live_tup + n_dead_tup) > 0
          THEN (n_dead_tup::numeric / (n_live_tup + n_dead_tup)::numeric) * 100.0
          ELSE 0.0
        END >= {{ inputs.dead_ratio_threshold_percent }}
      ORDER BY n_dead_tup DESC;

  - id: emit_audit_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Export computed table bloat metrics into execution outputs.
    values:
      bloated_table_count: "{{ outputs.inspect_table_bloat.size }}"
      database_scanned: "{{ inputs.database_name }}"
      host_scanned: "{{ inputs.postgres_host }}"
      scanned_at: "{{ now() }}"

  - id: evaluate_bloat_severity
    type: io.kestra.plugin.core.flow.If
    description: Branch based on whether any bloated tables breached operational thresholds.
    condition: "{{ outputs.inspect_table_bloat.size > 0 }}"
    then:
      - id: notify_dba_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alerts engineering and DBA team on Slack when dead tuple bloat
          exceeds operational thresholds.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :warning: *PostgreSQL Table Bloat Alert* — `{{ inputs.database_name }}` on `{{ inputs.postgres_host }}`
          *Execution:* `{{ execution.id }}`
          *Flagged Tables:* `{{ outputs.inspect_table_bloat.size }}` table(s) breached threshold (min: {{ inputs.dead_tuple_threshold }} dead tuples, {{ inputs.dead_ratio_threshold_percent }}% ratio).

          *Top Bloated Tables:*
          {% for row in outputs.inspect_table_bloat.rows %}- *`{{ row.schemaname }}.{{ row.table_name }}`*: {{ row.dead_tuples }} dead tuples ({{ row.dead_ratio_percent }}% bloat) | Live: {{ row.live_tuples }} | Last Autovacuum: {{ row.last_autovacuum }}
          {% endfor %}

          *Recommended Action:* Run `VACUUM ANALYZE <schema>.<table>` during off-peak hours or tune `autovacuum_vacuum_scale_factor`.
    else:
      - id: log_healthy_status
        type: io.kestra.plugin.core.log.Log
        description: Record healthy status when no tables exceed bloat limits.
        message: "PostgreSQL table health verified: 0 tables in '{{ inputs.database_name
          }}' exceeded dead tuple bloat thresholds."

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts on-call engineers if the database query or sentinel execution fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :rotating_light: *PostgreSQL Bloat Sentinel FAILED*
      *Workflow:* `{{ flow.id }}` in namespace `{{ flow.namespace }}`
      *Execution:* `{{ execution.id }}`
      *Database:* `{{ inputs.database_name }}` on `{{ inputs.postgres_host }}`
      Please check the Kestra execution logs for connection timeout or authentication errors.