Audit MySQL Long-Running Uncommitted Transactions

id: mysql-long-running-uncommitted-transactions-sentinel
namespace: company.database

description: |
  Audits MySQL InnoDB transaction states to detect long-running uncommitted transactions, lock contention, and idle worker connections before thread pool saturation occurs.

inputs:
  - id: mysql_host
    type: STRING
    displayName: MySQL Host
    defaults: "mysql.internal"
    description: Hostname or IP address of the target MySQL server.

  - id: mysql_port
    type: INT
    displayName: MySQL Port
    defaults: 3306
    description: Port number for the MySQL connection (default 3306).

  - id: mysql_database
    type: STRING
    displayName: Database Name
    defaults: "information_schema"
    description: Target database name to establish the inspection session against.

  - id: max_trx_seconds_threshold
    type: INT
    displayName: Max Transaction Runtime (Seconds)
    defaults: 60
    description: Maximum allowable elapsed seconds an open InnoDB transaction may
      remain uncommitted before alerting.

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#database-alerts"
    description: Target Slack channel for database incident alerts and connection
      killing runbooks.

triggers:
  - id: recurring_transaction_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */2 * * *"
    disabled: true
    description: Automated schedule trigger to audit InnoDB transactions every two hours.

tasks:
  - id: query_long_transactions
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Queries information_schema.innodb_trx joined with processlist to
      identify transactions exceeding runtime thresholds.
    url: "jdbc:mysql://{{ inputs.mysql_host }}:{{ inputs.mysql_port }}/{{
      inputs.mysql_database }}"
    username: "{{ secret('MYSQL_USERNAME') }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
          t.trx_id,
          t.trx_state,
          t.trx_started,
          TIMESTAMPDIFF(SECOND, t.trx_started, NOW()) AS elapsed_seconds,
          t.trx_rows_locked,
          t.trx_rows_modified,
          p.id AS process_id,
          COALESCE(p.user, 'unknown') AS db_user,
          COALESCE(p.host, 'unknown') AS client_host,
          COALESCE(p.db, 'none') AS current_db,
          COALESCE(p.info, 'Transaction open without active running query') AS query_snippet
      FROM information_schema.innodb_trx t
      LEFT JOIN information_schema.processlist p
          ON t.trx_mysql_thread_id = p.id
      WHERE TIMESTAMPDIFF(SECOND, t.trx_started, NOW()) >= {{ inputs.max_trx_seconds_threshold }}
      ORDER BY elapsed_seconds DESC
      LIMIT 25;

  - id: evaluate_transaction_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether any InnoDB transactions exceed maximum runtime
      thresholds and branches into Slack alerting or nominal logging.
    condition: "{{ outputs.query_long_transactions.rows | length > 0 }}"
    then:
      - id: alert_transactions_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches actionable operational alert to Slack detailing
          long-running uncommitted MySQL transactions.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "attachments": [
              {
                "color": "#D32F2F",
                "title": "🚨 MySQL InnoDB Incident: Long-Running Uncommitted Transactions Detected",
                "text": "The Kestra MySQL Sentinel detected {{ outputs.query_long_transactions.size }} open transaction(s) exceeding {{ inputs.max_trx_seconds_threshold }}s on host `{{ inputs.mysql_host }}`.",
                "fields": [
                  {
                    "title": "Oldest Transaction ID",
                    "value": "`{{ outputs.query_long_transactions.rows[0].trx_id }}` (State: {{ outputs.query_long_transactions.rows[0].trx_state }})",
                    "short": true
                  },
                  {
                    "title": "Elapsed Duration",
                    "value": "*{{ outputs.query_long_transactions.rows[0].elapsed_seconds }} seconds*",
                    "short": true
                  },
                  {
                    "title": "Rows Locked / Modified",
                    "value": "{{ outputs.query_long_transactions.rows[0].trx_rows_locked }} locked / {{ outputs.query_long_transactions.rows[0].trx_rows_modified }} modified",
                    "short": true
                  },
                  {
                    "title": "Client Connection",
                    "value": "`{{ outputs.query_long_transactions.rows[0].db_user }}@{{ outputs.query_long_transactions.rows[0].client_host }}` (Process ID: {{ outputs.query_long_transactions.rows[0].process_id }})",
                    "short": true
                  },
                  {
                    "title": "Remediation Command",
                    "value": "`KILL {{ outputs.query_long_transactions.rows[0].process_id }};`",
                    "short": false
                  }
                ],
                "footer": "Kestra MySQL Sentinel • Execution ID: {{ execution.id }}"
              }
            ]
          }

      - id: log_transaction_details
        type: io.kestra.plugin.core.log.Log
        description: Emits diagnostic log message for on-call database engineers.
        message: "Uncommitted Transaction Alert: Process ID {{
          outputs.query_long_transactions.rows[0].process_id }} running for {{
          outputs.query_long_transactions.rows[0].elapsed_seconds }}s. Locked
          rows: {{ outputs.query_long_transactions.rows[0].trx_rows_locked }}."

    else:
      - id: log_nominal_state
        type: io.kestra.plugin.core.log.Log
        description: Emits confirmation that all active MySQL transactions are within
          nominal execution durations.
        message: "MySQL transaction health nominal. 0 transactions exceed runtime
          threshold of {{ inputs.max_trx_seconds_threshold }}s."

  - id: emit_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Records audit completion telemetry for compliance and operational tracking.
    message: "MySQL transaction audit completed for '{{ inputs.mysql_host }}:{{
      inputs.mysql_port }}'. Flagged transaction count: {{
      outputs.query_long_transactions.size }}."

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatches high-urgency Slack alert if the sentinel execution fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": "❌ *MySQL Sentinel Execution Failed*: Execution `{{ execution.id }}` crashed during transaction audit for host `{{ inputs.mysql_host }}`."
      }