Autonomous Database Incident Investigation Agent with DuckDB and AI

id: ai-database-incident-investigation-agent
namespace: company.sre

description: |
  Autonomous Tier-1 database incident investigator that ingests active connection alerts,
  traces lock dependency trees using in-memory DuckDB diagnostics, uses an AI agent to
  identify root-cause blocker queries, and synthesizes copy-paste remediation SQL to Slack.

inputs:
  - id: database_engine
    type: STRING
    defaults: POSTGRESQL
    description: Database engine type under operational investigation.

  - id: alert_metric
    type: STRING
    defaults: CONNECTION_POOL_EXHAUSTED
    description: Primary metric symptom reported by infrastructure monitoring.

  - id: incident_context
    type: STRING
    defaults: |
      Connection pool utilization breached 98% threshold (490 of 500 connections active).
      Application p99 checkout response times degraded from 45ms to 3200ms across all worker pods.
    description: Metric alert context and business impact summary.

  - id: model_name
    type: STRING
    defaults: gemini-2.5-flash
    description: Foundational model used by the autonomous incident investigation agent.

  - id: seed_demo_data
    type: BOOL
    defaults: true
    description: Populate realistic session lock tree telemetry for zero-setup demo
      verification.

tasks:
  - id: collect_diagnostic_telemetry
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Ingests session process state and executes lock dependency analysis
      using in-memory DuckDB.
    fetchType: FETCH
    sql: |
      CREATE TABLE active_sessions (
          pid INT,
          username VARCHAR,
          client_addr VARCHAR,
          state VARCHAR,
          wait_event_type VARCHAR,
          wait_event VARCHAR,
          duration_sec INT,
          blocked_by INT,
          query_text VARCHAR
      );

      INSERT INTO active_sessions VALUES
      (14201, 'app_checkout', '10.0.4.12', 'active', 'Lock', 'relation', 1240, NULL, 'ALTER TABLE orders ADD COLUMN loyalty_points INT DEFAULT 0;'),
      (14205, 'app_checkout', '10.0.4.15', 'active', 'Lock', 'relation', 1190, 14201, 'INSERT INTO orders (customer_id, amount) VALUES (9821, 149.99);'),
      (14209, 'app_checkout', '10.0.4.18', 'active', 'Lock', 'relation', 1150, 14201, 'UPDATE orders SET status = completed WHERE order_id = 45892;'),
      (14214, 'app_checkout', '10.0.4.22', 'active', 'Lock', 'relation', 1110, 14201, 'SELECT * FROM orders WHERE status = pending FOR UPDATE;'),
      (14220, 'read_replica', '10.0.5.11', 'idle', NULL, NULL, 5, NULL, 'SELECT 1;');

      SELECT 
          COUNT(*) AS total_active_sessions,
          COUNT(blocked_by) AS total_blocked_sessions,
          MAX(duration_sec) AS max_duration_sec,
          (SELECT pid FROM active_sessions WHERE blocked_by IS NULL AND wait_event_type = 'Lock' ORDER BY duration_sec DESC LIMIT 1) AS head_blocker_pid,
          (SELECT query_text FROM active_sessions WHERE blocked_by IS NULL AND wait_event_type = 'Lock' ORDER BY duration_sec DESC LIMIT 1) AS head_blocker_query
      FROM active_sessions;

  - id: investigate_incident
    type: io.kestra.plugin.ai.agent.AIAgent
    description: Investigates lock trees and query symptoms using Google Gemini with
      structured JSON Schema output.
    provider:
      type: io.kestra.plugin.ai.provider.GoogleGemini
      apiKey: "{{ secret('GEMINI_API_KEY') }}"
      modelName: "{{ inputs.model_name }}"
    configuration:
      temperature: 0.1
      maxToken: 2048
      responseFormat:
        type: JSON
        jsonSchema:
          type: object
          required:
            - incident_verdict
            - severity
            - culprit_pid
            - culprit_query
            - root_cause_analysis
            - immediate_remediation_sql
            - preventive_recommendations
          properties:
            incident_verdict:
              type: string
              enum:
                - CRITICAL_LOCK_CONTENTION
                - LONG_RUNNING_IDLE_TRANSACTION
                - HIGH_VOLUME_TRAFFIC_SPIKE
                - NOMINAL_FALSE_ALARM
            severity:
              type: string
              enum:
                - CRITICAL
                - WARNING
                - INFO
            culprit_pid:
              type: integer
            culprit_query:
              type: string
            root_cause_analysis:
              type: string
            immediate_remediation_sql:
              type: string
            preventive_recommendations:
              type: string
    prompt: |
      You are an autonomous Tier-1 Site Reliability Engineering (SRE) database investigator.
      Analyze the following live database incident telemetry and lock diagnostic summary:

      Incident Context:
      - Engine: {{ inputs.database_engine }}
      - Trigger Symptom: {{ inputs.alert_metric }}
      - Observed Degradation:
      """
      {{ inputs.incident_context }}
      """

      Session Telemetry Snapshot:
      {{ outputs.collect_diagnostic_telemetry.rows | json }}

      Investigation Protocol:
      1. Analyze the lock dependency hierarchy to find the root blocker session (pid holding ungranted locks while blocking downstream workers).
      2. Classify the incident verdict and evaluate operational severity (CRITICAL if core transactions are blocked for over 60 seconds).
      3. Generate the exact, safe remediation command (such as pg_cancel_backend or pg_terminate_backend) targeting only the root culprit PID.
      4. Provide architectural recommendations to prevent this lock contention pattern in future release migrations.

  - id: evaluate_severity_gate
    type: io.kestra.plugin.core.flow.If
    description: Branches resolution workflow based on whether investigation
      identified a critical blocker lock.
    condition: "{{ fromJson(outputs.investigate_incident.text).severity == 'CRITICAL' }}"
    then:
      - id: dispatch_sre_incident_card
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        description: Dispatches actionable diagnostic investigation card and copy-paste
          fix to SRE Slack channel.
        payload: |
          {
            "text": "🚨 *Database Incident Triage: {{ fromJson(outputs.investigate_incident.text).incident_verdict }}*\n*Severity:* `{{ fromJson(outputs.investigate_incident.text).severity }}` | *Engine:* `{{ inputs.database_engine }}`\n*Root Cause:* {{ fromJson(outputs.investigate_incident.text).root_cause_analysis }}\n*Culprit Process:* PID `{{ fromJson(outputs.investigate_incident.text).culprit_pid }}`\n*Offending Query:* \n```sql\n{{ fromJson(outputs.investigate_incident.text).culprit_query }}\n```\n*Immediate Remediation Command:* \n```sql\n{{ fromJson(outputs.investigate_incident.text).immediate_remediation_sql }}\n```\n*Preventive Guidance:* {{ fromJson(outputs.investigate_incident.text).preventive_recommendations }}"
          }
    else:
      - id: log_routine_health
        type: io.kestra.plugin.core.log.Log
        description: Logs advisory analysis when incident does not require immediate
          manual kill intervention.
        message: "Database investigation completed. Severity: {{
          fromJson(outputs.investigate_incident.text).severity }}. Verdict: {{
          fromJson(outputs.investigate_incident.text).incident_verdict }}."

triggers:
  - id: scheduled_health_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic scheduled audit of database session lock contention.
      Shipped disabled by default.
    cron: "0 */4 * * *"
    disabled: true

  - id: alert_webhook_trigger
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook endpoint triggered by Datadog, Prometheus, or
      CloudWatch alarms.
    key: "{{ secret('WEBHOOK_KEY') }}"

errors:
  - id: alert_investigation_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    description: Sends alert notice if autonomous investigation agent encounters an
      unhandled exception.
    payload: |
      {
        "text": "⚠️ Execution {{ execution.id }} failed in flow {{ flow.id }}. Autonomous database investigation agent encountered an unhandled exception."
      }