PostgreSQL Autovacuum Wraparound Risk Sentinel

id: postgres-autovacuum-wraparound-risk-sentinel
namespace: company.team
description: |
  Automated PostgreSQL transaction ID wraparound sentinel. Monitors age(datfrozenxid) across all databases
  against the 2-billion freeze horizon and alerts DBAs before PostgreSQL enters emergency read-only shutdown.

concurrency:
  limit: 1

inputs:
  - id: postgres_url
    type: STRING
    displayName: PostgreSQL JDBC URL
    defaults: "jdbc:postgresql://postgres.internal:5432/production"
    description: "JDBC connection URL for the target PostgreSQL database cluster."

  - id: postgres_username
    type: STRING
    displayName: Database Username
    defaults: "db_monitor"
    description: "Database user with permissions to query pg_database system catalogs."

  - id: freeze_max_age_warning_threshold
    type: INT
    displayName: Freeze Age Warning Threshold
    defaults: 1500000000
    description: "Maximum allowable age(datfrozenxid) transactions before alerting
      (wraparound limit is 2.14 billion)."

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

triggers:
  - id: daily_wraparound_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Executes wraparound check daily at 03:00 UTC."
    cron: "0 3 * * *"

tasks:
  - id: query_database_freeze_horizon
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: "Queries pg_database catalog to compute transaction ages and
      remaining horizon capacity."
    url: "{{ inputs.postgres_url }}"
    username: "{{ inputs.postgres_username }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        datname,
        age(datfrozenxid) AS current_freeze_age,
        (2147483647 - age(datfrozenxid)) AS remaining_transactions,
        ROUND((age(datfrozenxid)::numeric / 2147483647.0) * 100, 2) AS capacity_percentage
      FROM pg_database
      WHERE datallowconn = true
      ORDER BY age(datfrozenxid) DESC;

  - id: analyze_wraparound_risk
    type: io.kestra.plugin.scripts.python.Script
    description: "Evaluates database transaction ages against warning thresholds and
      generates triage manifest."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    dependencies:
      - kestra
    inputFiles:
      rows.json: "{{ outputs.query_database_freeze_horizon.rows | default('[]') |
        toJson }}"
    outputFiles:
      - wraparound_report.json
    script: |
      import json
      from kestra import Kestra

      with open("rows.json", "r", encoding="utf-8") as f:
          databases = json.load(f)

      threshold = int("{{ inputs.freeze_max_age_warning_threshold }}")
      at_risk_dbs = []

      for db in databases:
          name = db.get("datname", "unknown")
          age_val = int(db.get("current_freeze_age", 0))
          remaining = int(db.get("remaining_transactions", 0))
          pct = float(db.get("capacity_percentage", 0.0))

          if age_val > threshold:
              at_risk_dbs.append({
                  "database": name,
                  "current_age": age_val,
                  "remaining_tx": remaining,
                  "capacity_pct": pct
              })

      has_risk = len(at_risk_dbs) > 0

      report = {
          "total_databases_scanned": len(databases),
          "warning_threshold": threshold,
          "at_risk_count": len(at_risk_dbs),
          "at_risk_databases": at_risk_dbs,
          "has_risk": has_risk
      }

      with open("wraparound_report.json", "w", encoding="utf-8") as f:
          json.dump(report, f, indent=2)

      Kestra.outputs({
          "wraparound_risk_detected": has_risk,
          "at_risk_count": len(at_risk_dbs),
          "highest_age_database": at_risk_dbs[0]["database"] if at_risk_dbs else "None",
          "highest_age_value": max((int(db.get("current_freeze_age", 0)) for db in databases), default=0),
          "remaining_tx": at_risk_dbs[0]["remaining_tx"] if at_risk_dbs else 0
      })

  - id: evaluate_risk_condition
    type: io.kestra.plugin.core.flow.If
    description: "Branches execution based on whether any database exceeded the
      freeze threshold."
    condition: "{{ outputs.analyze_wraparound_risk.vars.wraparound_risk_detected }}"
    then:
      - id: alert_dba_team_on_wraparound_risk
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches critical operational freeze alert to the DBA team via
          Slack."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*CRITICAL: PostgreSQL Autovacuum Wraparound Risk Alert*\n• *Cluster Endpoint*: `{{ inputs.postgres_url }}`\n• *Databases Breaching Threshold*: {{ outputs.analyze_wraparound_risk.vars.at_risk_count }}\n• *Highest Age Database*: `{{ outputs.analyze_wraparound_risk.vars.highest_age_database }}`\n• *Current Transaction Age*: {{ outputs.analyze_wraparound_risk.vars.highest_age_value }} transactions\n• *Remaining Capacity*: {{ outputs.analyze_wraparound_risk.vars.remaining_tx }} transactions before freeze horizon\n• *Operational Impact*: Approaching 2.0 billion transactions causes PostgreSQL to refuse all write operations and shut down into single-user emergency maintenance mode.\n• *Recommended Remediation*: Schedule an immediate manual `VACUUM FREEZE ANALYZE` on database `{{ outputs.analyze_wraparound_risk.vars.highest_age_database }}` during an emergency maintenance window."
          }

    else:
      - id: log_compliant_wraparound_status
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation of healthy transaction freeze horizon."
        message: "PostgreSQL transaction age audit compliant. All databases below
          warning threshold. Highest age: {{
          outputs.analyze_wraparound_risk.vars.highest_age_value }}."

  - id: export_wraparound_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON audit manifest for compliance records."
    format: |
      {
        "pipeline": "postgres-autovacuum-wraparound-risk-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "highest_age": {{ outputs.analyze_wraparound_risk.vars.highest_age_value }},
        "status": "{{ outputs.analyze_wraparound_risk.vars.wraparound_risk_detected ? 'CRITICAL_RISK' : 'COMPLIANT' }}"
      }