MySQL Replica IO Thread Stagnation and Lag Sentinel

id: mysql-replica-io-thread-stagnation-sentinel
namespace: company.data
description: |
  Inspects MySQL replica Seconds_Behind_Master and IO/SQL replication threads via SHOW REPLICA STATUS,
  alerting DBAs before replication drift causes stale read anomalies or failover hazards.
inputs:
  - id: database_host
    type: STRING
    displayName: MySQL Replica Host
    defaults: mysql-replica-01.internal
  - id: database_port
    type: INT
    displayName: Database Port
    defaults: 3306
  - id: database_name
    type: STRING
    displayName: Database Name
    defaults: production
  - id: database_username
    type: STRING
    displayName: Database Username
    defaults: replica_monitor
  - id: max_seconds_behind_master_threshold
    type: INT
    displayName: Max Replication Lag Seconds Threshold
    defaults: 300

triggers:
  - id: periodic_replication_lag_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    disabled: true

tasks:
  - id: check_replica_status
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Executes SHOW REPLICA STATUS to fetch current replication threads
      and lag metrics
    url: "jdbc:mysql://{{ inputs.database_host }}:{{ inputs.database_port }}/{{
      inputs.database_name }}"
    username: "{{ inputs.database_username }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    sql: "SHOW REPLICA STATUS;"
    fetchType: FETCH_ONE

  - id: evaluate_replication_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates if IO/SQL threads are running and lag is below threshold
    condition: "{{ (outputs.check_replica_status.row.Seconds_Behind_Master ?? 99999)
      >= inputs.max_seconds_behind_master_threshold or
      outputs.check_replica_status.row.Replica_IO_Running != 'Yes' or
      outputs.check_replica_status.row.Replica_SQL_Running != 'Yes' }}"
    then:
      - id: alert_dba_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alerts DBA team on Slack regarding replication disruption
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "text": "🚨 *MySQL Replication Lag / Thread Alert*\n*Host:* `{{ inputs.database_host }}`\n*Seconds Behind Master:* *{{ outputs.check_replica_status.row.Seconds_Behind_Master ?? 'NULL (Broken)' }}* (Threshold: `{{ inputs.max_seconds_behind_master_threshold }}s`)\n*Replica IO Running:* `{{ outputs.check_replica_status.row.Replica_IO_Running }}`\n*Replica SQL Running:* `{{ outputs.check_replica_status.row.Replica_SQL_Running }}`\n*Last Error:* `{{ outputs.check_replica_status.row.Last_Error ?? 'None' }}`"
          }
    else:
      - id: log_replication_healthy
        type: io.kestra.plugin.core.log.Log
        message: "MySQL replication on {{ inputs.database_host }} is healthy (Lag: {{
          outputs.check_replica_status.row.Seconds_Behind_Master ?? 0 }}s)."

errors:
  - id: alert_check_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts Slack on connection or query execution failure
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "⚠️ *MySQL Replication Sentinel Failed* during execution `{{ execution.id }}` on host `{{ inputs.database_host }}`."
      }