Guard PostgreSQL Connection Saturation with Slack Alerts

id: postgres-connection-saturation-guard
namespace: company.team
description: |
  Watch PostgreSQL connection usage versus max_connections, break down states
  and top applications, and alert DBAs on Slack before pools saturate.

triggers:
  - id: every_5_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check connection saturation every 5 minutes
    cron: "*/5 * * * *"

  - id: on_demand_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: On-demand saturation check from incident response
    key: "{{ secret('WEBHOOK_KEY') }}"

inputs:
  - id: postgres_host
    type: STRING
    displayName: PostgreSQL Host
    description: Hostname of the PostgreSQL instance
    defaults: postgres
  - id: postgres_port
    type: INT
    displayName: PostgreSQL Port
    description: Port of the PostgreSQL instance
    defaults: 5432
  - id: database_name
    type: STRING
    displayName: Database Name
    description: Database to connect to for pg_stat_activity inspection
    defaults: postgres
  - id: warn_pct
    type: INT
    displayName: Warning threshold (%)
    description: Slack warning when usage meets or exceeds this percent
    defaults: 70
  - id: critical_pct
    type: INT
    displayName: Critical threshold (%)
    description: Escalated alert when usage meets or exceeds this percent
    defaults: 90

tasks:
  - id: query_connection_stats
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Count connections by state and read max_connections setting
    url: "jdbc:postgresql://{{ inputs.postgres_host }}:{{ inputs.postgres_port }}/{{
      inputs.database_name }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        (SELECT count(*) FROM pg_stat_activity) AS total_connections,
        (SELECT setting::int FROM pg_settings WHERE name = 'max_connections') AS max_connections,
        COUNT(*) FILTER (WHERE state = 'active') AS active,
        COUNT(*) FILTER (WHERE state = 'idle in transaction') AS idle_in_transaction,
        COUNT(*) FILTER (WHERE state = 'idle') AS idle,
        COUNT(*) FILTER (WHERE wait_event_type = 'Lock') AS waiting_on_lock
      FROM pg_stat_activity;

  - id: query_top_apps
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Break down connection holders by application for triage
    url: "jdbc:postgresql://{{ inputs.postgres_host }}:{{ inputs.postgres_port }}/{{
      inputs.database_name }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT COALESCE(application_name, 'unknown') AS app, count(*) AS connections
      FROM pg_stat_activity GROUP BY 1 ORDER BY 2 DESC LIMIT 10;

  - id: evaluate_saturation
    type: io.kestra.plugin.scripts.python.Script
    description: Compute usage percent and severity against warn/critical thresholds
    containerImage: python:3.11
    dependencies:
      - kestra
    script: |
      from kestra import Kestra
      total = {{ outputs.query_connection_stats.rows[0].total_connections }}
      maximum = {{ outputs.query_connection_stats.rows[0].max_connections }}
      warn = {{ inputs.warn_pct }}
      crit = {{ inputs.critical_pct }}
      pct = round(total / maximum * 100, 1) if maximum else 0
      severity = "critical" if pct >= crit else ("warn" if pct >= warn else "ok")
      Kestra.outputs({"total": total, "maximum": maximum, "pct": pct, "severity": severity, "breached": severity != "ok"})

  - id: route_by_severity
    type: io.kestra.plugin.core.flow.If
    description: Alert on warn/critical, log quiet all-clear otherwise
    condition: "{{ outputs.evaluate_saturation.vars.breached }}"
    then:
      - id: notify_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post saturation alert with state breakdown and top apps
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :rotating_light: *Postgres Connection Saturation ({{ outputs.evaluate_saturation.vars.severity | upper }})*
          Database: `{{ inputs.database_name }}` on `{{ inputs.postgres_host }}`
          Usage: `{{ outputs.evaluate_saturation.vars.total }}` / `{{ outputs.evaluate_saturation.vars.maximum }}` ({{ outputs.evaluate_saturation.vars.pct }}%)
          States: active={{ outputs.query_connection_stats.rows[0].active }} idle_in_txn={{ outputs.query_connection_stats.rows[0].idle_in_transaction }} idle={{ outputs.query_connection_stats.rows[0].idle }} lock_wait={{ outputs.query_connection_stats.rows[0].waiting_on_lock }}
          Top apps: {% for row in outputs.query_top_apps.rows %}`{{ row.app }}`={{ row.connections }} {% endfor %}
          Execution: `{{ execution.id }}`
          Triage: check pgbouncer pools, idle-in-transaction holders, then `SELECT pg_terminate_backend(pid)` for strays.
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        description: Record healthy saturation level
        message: "Postgres connections healthy: {{
          outputs.evaluate_saturation.vars.total }}/{{
          outputs.evaluate_saturation.vars.maximum }} ({{
          outputs.evaluate_saturation.vars.pct }}%)."

  - id: emit_saturation_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Export saturation metrics for history
    values:
      usage_pct: "{{ outputs.evaluate_saturation.vars.pct }}"
      severity: "{{ outputs.evaluate_saturation.vars.severity }}"
      checked_at: "{{ now() }}"

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert on-call if the saturation query fails
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :warning: *Postgres Saturation Guard FAILED*
      Flow: `{{ flow.id }}` execution `{{ execution.id }}`
      Check DB reachability, credentials, and pg_stat_activity grants.