PostgreSQL Blocking Transaction and Lock Contention Monitor

id: postgres-blocking-transaction-monitor
namespace: company.team
description: |
  Inspect PostgreSQL active locks and session states via pg_locks and pg_stat_activity,
  detect root blocking transactions and cascading lock wait trees, export diagnostic
  metrics into execution outputs, and alert DBAs on Slack before connection pools exhaust.

triggers:
  - id: scheduled_monitor
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic lock contention check. Shipped disabled by default.
    cron: "*/10 * * * *"
    disabled: true

  - id: manual_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook trigger for on-demand invocation from
      incident triage tools or ChatOps.
    key: "{{ secret('WEBHOOK_KEY') }}"

inputs:
  - id: postgres_host
    type: STRING
    displayName: PostgreSQL Host
    description: Hostname or network address of the 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: min_wait_seconds
    type: INT
    displayName: Minimum Lock Wait (Seconds)
    description: Minimum wait duration in seconds before a blocked transaction
      triggers an operational alert.
    defaults: 30

tasks:
  - id: query_blocking_locks
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Inspect pg_locks and pg_stat_activity to detect granted locks that
      are blocking other sessions.
    url: "jdbc:postgresql://{{ inputs.postgres_host }}:{{ inputs.postgres_port }}/{{
      inputs.database_name }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        blocked_locks.pid AS blocked_pid,
        blocked_activity.usename AS blocked_user,
        blocking_locks.pid AS blocker_pid,
        blocking_activity.usename AS blocker_user,
        blocking_activity.client_addr::text AS blocker_client_ip,
        COALESCE(blocking_activity.application_name, 'unknown') AS blocker_app,
        blocking_activity.state AS blocker_state,
        ROUND(EXTRACT(EPOCH FROM (CLOCK_TIMESTAMP() - blocked_activity.query_start))::numeric, 1) AS wait_duration_seconds,
        SUBSTRING(REGEXP_REPLACE(blocking_activity.query, '\s+', ' ', 'g') FROM 1 FOR 150) AS blocker_query,
        SUBSTRING(REGEXP_REPLACE(blocked_activity.query, '\s+', ' ', 'g') FROM 1 FOR 150) AS blocked_query
      FROM pg_catalog.pg_locks blocked_locks
      JOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pid
      JOIN pg_catalog.pg_locks blocking_locks
        ON blocking_locks.locktype = blocked_locks.locktype
        AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database
        AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation
        AND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.page
        AND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tuple
        AND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxid
        AND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionid
        AND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classid
        AND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objid
        AND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubid
        AND blocking_locks.pid != blocked_locks.pid
      JOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pid
      WHERE NOT blocked_locks.granted
        AND blocking_locks.granted
        AND EXTRACT(EPOCH FROM (CLOCK_TIMESTAMP() - blocked_activity.query_start)) >= {{ inputs.min_wait_seconds }}
      ORDER BY wait_duration_seconds DESC;

  - id: emit_lock_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Export lock contention metrics into execution outputs for auditability.
    values:
      blocked_transactions_count: "{{ outputs.query_blocking_locks.size }}"
      database_scanned: "{{ inputs.database_name }}"
      host_scanned: "{{ inputs.postgres_host }}"
      audited_at: "{{ now() }}"

  - id: evaluate_lock_contention
    type: io.kestra.plugin.core.flow.If
    description: Branch based on whether any blocked sessions were detected.
    condition: "{{ outputs.query_blocking_locks.size > 0 }}"
    then:
      - id: notify_dba_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alerts DBA and engineering team on Slack with blocker details and
          termination commands.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :rotating_light: *PostgreSQL Lock Contention Alert* — `{{ inputs.database_name }}` on `{{ inputs.postgres_host }}`
          *Execution:* `{{ execution.id }}`
          *Blocked Sessions:* `{{ outputs.query_blocking_locks.size }}` query(ies) blocked for more than {{ inputs.min_wait_seconds }}s.

          *Active Lock Blockers:*
          {% for row in outputs.query_blocking_locks.rows %}- *Blocker PID `{{ row.blocker_pid }}`* (User: `{{ row.blocker_user }}`, App: `{{ row.blocker_app }}`, State: `{{ row.blocker_state }}`)
            • Blocking PID `{{ row.blocked_pid }}` (waiting {{ row.wait_duration_seconds }}s)
            • Blocker Query: `{{ row.blocker_query }}`
            • Remediation: `SELECT pg_cancel_backend({{ row.blocker_pid }});` or `SELECT pg_terminate_backend({{ row.blocker_pid }});`
          {% endfor %}
    else:
      - id: log_healthy_locks
        type: io.kestra.plugin.core.log.Log
        description: Record healthy status when no blocked transactions are detected.
        message: "PostgreSQL lock verification passed: 0 blocked transactions detected
          on '{{ inputs.database_name }}'."

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts on-call engineers if the database query or monitor execution fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :warning: *PostgreSQL Lock Monitor 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 permission issues.