PostgreSQL Connection Pool Saturation and Deadlock Sentinel

id: postgres-connection-pool-and-deadlock-sentinel
namespace: company.team
description: |
  Automated PostgreSQL connection saturation and lock health monitor. Queries pg_stat_database and pg_stat_activity
  to detect pool exhaustion, aged idle-in-transaction connections, and deadlocks, alerting DBAs on Slack.

concurrency:
  limit: 1

inputs:
  - id: postgres_url
    type: STRING
    displayName: PostgreSQL JDBC URL
    description: JDBC connection string targeting the target PostgreSQL database.
    defaults: "jdbc:postgresql://localhost:5432/postgres"

  - id: max_connection_utilization_percent
    type: FLOAT
    displayName: Max Connection Utilization Percent
    description: Threshold percentage of max_connections beyond which an alert is triggered.
    defaults: 80.0

  - id: max_idle_transaction_age_seconds
    type: INT
    displayName: Max Idle in Transaction Age (Seconds)
    description: Duration in seconds an idle transaction can remain open before
      triggering an alert.
    defaults: 120

  - id: slack_channel
    type: STRING
    displayName: Alert Channel
    description: Slack channel destination for database performance alerts.
    defaults: "#dba-alerts"

triggers:
  - id: scheduled_health_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic PostgreSQL connection pool and lock health check executed
      every 5 minutes.
    cron: "*/5 * * * *"

tasks:
  - id: audit_postgres_health
    type: io.kestra.plugin.jdbc.postgres.Query
    description: Query pg_stat_activity and pg_stat_database to compute connection
      saturation and idle transaction metrics.
    url: "{{ inputs.postgres_url }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      WITH connection_stats AS (
        SELECT
          COUNT(*) AS total_connections,
          COUNT(*) FILTER (WHERE state = 'active') AS active_connections,
          COUNT(*) FILTER (WHERE state = 'idle') AS idle_connections,
          COUNT(*) FILTER (WHERE state = 'idle in transaction') AS idle_in_transaction_connections,
          COUNT(*) FILTER (
            WHERE state = 'idle in transaction'
              AND EXTRACT(EPOCH FROM (now() - state_change)) >= {{ inputs.max_idle_transaction_age_seconds }}
          ) AS aged_idle_in_transaction_connections,
          MAX(EXTRACT(EPOCH FROM (now() - state_change))) FILTER (WHERE state = 'idle in transaction') AS max_idle_in_trx_seconds
        FROM pg_stat_activity
        WHERE backend_type = 'client backend'
      ),
      database_deadlocks AS (
        SELECT COALESCE(SUM(deadlocks), 0) AS total_deadlocks
        FROM pg_stat_database
        WHERE datname = current_database()
      ),
      server_settings AS (
        SELECT setting::int AS max_connections
        FROM pg_settings
        WHERE name = 'max_connections'
      )
      SELECT
        c.total_connections,
        c.active_connections,
        c.idle_connections,
        c.idle_in_transaction_connections,
        c.aged_idle_in_transaction_connections,
        ROUND(COALESCE(c.max_idle_in_trx_seconds, 0)::numeric, 1) AS max_idle_in_trx_seconds,
        d.total_deadlocks,
        s.max_connections,
        ROUND((c.total_connections::numeric / s.max_connections::numeric) * 100, 1) AS connection_utilization_percent,
        CASE
          WHEN (c.total_connections::numeric / s.max_connections::numeric) * 100 >= {{ inputs.max_connection_utilization_percent }} THEN 'CONNECTION_POOL_SATURATION'
          WHEN c.aged_idle_in_transaction_connections > 0 THEN 'IDLE_IN_TRANSACTION_STARVATION'
          ELSE 'HEALTHY'
        END AS health_status
      FROM connection_stats c
      CROSS JOIN database_deadlocks d
      CROSS JOIN server_settings s;

  - id: evaluate_health_thresholds
    type: io.kestra.plugin.core.flow.If
    description: Branch execution when connection utilization or idle transaction
      duration breaches thresholds.
    condition: "{{ outputs.audit_postgres_health.rows[0].health_status != 'HEALTHY' }}"
    then:
      - id: notify_dba_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Broadcast critical connection pool warning with operational
          diagnostics to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          *PostgreSQL Connection Pool Alert*
          • *Status*: `{{ outputs.audit_postgres_health.rows[0].health_status }}`
          • *Connection Utilization*: `{{ outputs.audit_postgres_health.rows[0].total_connections }} / {{ outputs.audit_postgres_health.rows[0].max_connections }}` (`{{ outputs.audit_postgres_health.rows[0].connection_utilization_percent }}%`)
          • *Active Queries*: `{{ outputs.audit_postgres_health.rows[0].active_connections }}`
          • *Idle in Transaction*: `{{ outputs.audit_postgres_health.rows[0].idle_in_transaction_connections }}` (Aged >= {{ inputs.max_idle_transaction_age_seconds }}s: `{{ outputs.audit_postgres_health.rows[0].aged_idle_in_transaction_connections }}`)
          • *Max Idle Transaction Duration*: `{{ outputs.audit_postgres_health.rows[0].max_idle_in_trx_seconds }}s`
          • *Total Recorded Deadlocks*: `{{ outputs.audit_postgres_health.rows[0].total_deadlocks }}`
          • *Remediation*: Investigate unclosed client connections. Terminate stalled sessions with `SELECT pg_terminate_backend(pid);` or adjust PgBouncer pool sizing.

    else:
      - id: log_healthy_status
        type: io.kestra.plugin.core.log.Log
        description: Confirm that connection pool utilization is well within nominal
          operational limits.
        message: "PostgreSQL connection pool health is nominal. Current utilization: {{
          outputs.audit_postgres_health.rows[0].connection_utilization_percent
          }}% of max_connections."

  - id: export_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Record connection pool telemetry manifest for execution history and
      reliability records.
    format: |
      {
        "pipeline": "postgres-connection-pool-and-deadlock-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "health_status": "{{ outputs.audit_postgres_health.rows[0].health_status }}",
        "connection_utilization_percent": {{ outputs.audit_postgres_health.rows[0].connection_utilization_percent }},
        "total_connections": {{ outputs.audit_postgres_health.rows[0].total_connections }},
        "max_connections": {{ outputs.audit_postgres_health.rows[0].max_connections }},
        "aged_idle_in_trx": {{ outputs.audit_postgres_health.rows[0].aged_idle_in_transaction_connections }}
      }