MySQL Read Replica Replication Lag Monitoring with Kestra

id: mysql-replication-lag-sentinel
namespace: company.database

inputs:
  - id: mysql_url
    type: STRING
    displayName: MySQL JDBC URL
    description: JDBC connection string targeting the MySQL read replica instance.
    defaults: "jdbc:mysql://localhost:3306/production"

  - id: max_lag_seconds
    type: INT
    displayName: Maximum Lag Threshold (Seconds)
    description: Threshold in seconds beyond which replication delay is classified
      as an SLA breach.
    defaults: 60

  - id: fail_on_breach
    type: BOOL
    displayName: Fail Execution on SLA Breach
    description: When enabled, terminates the execution as FAILED if lag breaches
      SLA or replication threads halt.
    defaults: false

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Send a DBA incident notification card to Slack when replication lag
      or thread errors are detected.
    defaults: true

concurrency:
  limit: 1

triggers:
  - id: scheduled_replication_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Recurring replication health check executed every 5 minutes.
    cron: "*/5 * * * *"

tasks:
  - id: check_replica_status
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Query MySQL replica telemetry via SHOW REPLICA STATUS to inspect
      IO/SQL threads and replication lag.
    url: "{{ inputs.mysql_url }}"
    username: "{{ secret('MYSQL_USERNAME') }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: SHOW REPLICA STATUS;

  - id: evaluate_replica_health
    type: io.kestra.plugin.scripts.python.Script
    description: Parse replication telemetry, inspect thread health across modern
      and legacy MySQL conventions, and measure lag.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install --no-cache-dir kestra
    script: |
      import json
      from kestra import Kestra

      # Ingest query output row
      raw_row = {{ outputs.check_replica_status.row | toJson }} or {}

      if not raw_row:
          print("Warning: SHOW REPLICA STATUS returned no rows. Is this instance configured as a replica?")
          Kestra.outputs({
              "is_configured_replica": False,
              "threads_healthy": False,
              "is_lagging": True,
              "lag_seconds": -1,
              "source_host": "N/A",
              "binlog_file": "N/A",
              "error_message": "Instance is not configured as a MySQL read replica (SHOW REPLICA STATUS returned 0 rows)."
          })
          exit(0)

      # Support both MySQL 8.0.22+ and legacy MySQL 5.7 / MariaDB column names
      io_running = raw_row.get("Replica_IO_Running") or raw_row.get("Slave_IO_Running") or "No"
      sql_running = raw_row.get("Replica_SQL_Running") or raw_row.get("Slave_SQL_Running") or "No"
      source_host = raw_row.get("Source_Host") or raw_row.get("Master_Host") or "unknown"
      binlog_file = raw_row.get("Source_Log_File") or raw_row.get("Master_Log_File") or "unknown"
      binlog_pos = raw_row.get("Read_Source_Log_Pos") or raw_row.get("Read_Master_Log_Pos") or 0

      # Seconds_Behind_Source / Seconds_Behind_Master is NULL if SQL thread is stopped
      raw_lag = raw_row.get("Seconds_Behind_Source")
      if raw_lag is None:
          raw_lag = raw_row.get("Seconds_Behind_Master")

      try:
          lag_seconds = int(raw_lag) if raw_lag is not None else 999999
      except (ValueError, TypeError):
          lag_seconds = 999999

      last_io_error = raw_row.get("Last_IO_Error") or ""
      last_sql_error = raw_row.get("Last_SQL_Error") or ""
      combined_error = f"{last_io_error} {last_sql_error}".strip() or "None"

      threads_healthy = (io_running == "Yes" and sql_running == "Yes")
      max_lag = int("{{ inputs.max_lag_seconds }}")
      is_lagging = (lag_seconds > max_lag) or not threads_healthy

      print(f"Replication Status: IO={io_running}, SQL={sql_running}, Lag={lag_seconds}s (Max: {max_lag}s)")
      print(f"Source: {source_host}, Binlog: {binlog_file}:{binlog_pos}")

      Kestra.outputs({
          "is_configured_replica": True,
          "threads_healthy": threads_healthy,
          "is_lagging": is_lagging,
          "lag_seconds": lag_seconds if lag_seconds != 999999 else -1,
          "source_host": str(source_host),
          "binlog_file": f"{binlog_file}:{binlog_pos}",
          "error_message": combined_error
      })

  - id: evaluate_sla_breach
    type: io.kestra.plugin.core.flow.If
    description: Branch based on whether replication is lagging beyond SLA or thread
      errors occurred.
    condition: "{{ outputs.evaluate_replica_health.vars.is_lagging or not
      outputs.evaluate_replica_health.vars.threads_healthy }}"
    then:
      - id: notify_slack_alert
        type: io.kestra.plugin.core.flow.If
        description: Dispatch critical replication lag incident card to Slack if
          notifications are enabled.
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_dba_alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Send structured DBA alert card to Slack.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "CRITICAL: MySQL Read Replica Replication Breach!\n• Replica URL: `{{ inputs.mysql_url }}`\n• Source Host: `{{ outputs.evaluate_replica_health.vars.source_host }}`\n• Replication Lag: {{ outputs.evaluate_replica_health.vars.lag_seconds }}s (SLA Threshold: {{ inputs.max_lag_seconds }}s)\n• Threads Healthy: `{{ outputs.evaluate_replica_health.vars.threads_healthy }}`\n• Binlog Position: `{{ outputs.evaluate_replica_health.vars.binlog_file }}`\n• Error Detail: {{ outputs.evaluate_replica_health.vars.error_message }}\n\nExecution: {{ execution.id }}"
              }

      - id: check_fail_condition
        type: io.kestra.plugin.core.flow.If
        description: Conditionally terminate execution with failure status if
          fail_on_breach is enabled.
        condition: "{{ inputs.fail_on_breach }}"
        then:
          - id: fail_execution
            type: io.kestra.plugin.core.execution.Fail
            description: Terminate execution due to replication SLA breach.
            errorMessage: "MySQL replication SLA breach: Lag is {{
              outputs.evaluate_replica_health.vars.lag_seconds }}s (threshold:
              {{ inputs.max_lag_seconds }}s) or threads stopped."
    else:
      - id: log_healthy_status
        type: io.kestra.plugin.core.log.Log
        description: Record healthy replica sync status in execution logs.
        message: >-
          MySQL read replica is healthy and in sync with '{{
          outputs.evaluate_replica_health.vars.source_host }}'. Lag: {{
          outputs.evaluate_replica_health.vars.lag_seconds }}s. Coordinates: {{
          outputs.evaluate_replica_health.vars.binlog_file }}. All IO and SQL
          threads running normally.

errors:
  - id: on_connection_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Catch database connection timeouts or authentication failures and
      alert the DBA team.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "MySQL connection failure in flow `{{ flow.id }}`: Unable to query read replica at `{{ inputs.mysql_url }}`. Check network routing and database credentials. Execution: {{ execution.id }}."
      }

outputs:
  - id: threads_healthy
    type: BOOLEAN
    description: Indicates whether both IO and SQL replication threads are running ('Yes').
    value: "{{ outputs.evaluate_replica_health.vars.threads_healthy }}"
  - id: is_lagging
    type: BOOLEAN
    description: Indicates whether replication delay exceeds the configured
      max_lag_seconds threshold.
    value: "{{ outputs.evaluate_replica_health.vars.is_lagging }}"
  - id: lag_seconds
    type: INT
    description: Current replication delay in seconds behind the primary.
    value: "{{ outputs.evaluate_replica_health.vars.lag_seconds }}"
  - id: source_host
    type: STRING
    description: Upstream primary source hostname.
    value: "{{ outputs.evaluate_replica_health.vars.source_host }}"
  - id: binlog_coordinates
    type: STRING
    description: Current read binlog file and position.
    value: "{{ outputs.evaluate_replica_health.vars.binlog_file }}"

Diagram unavailable

We could not build the topology for this blueprint. The flow itself is valid, use the YAML on the left to run it.

MySQL read replicas form the backbone of modern database scalability. However, heavy write batches, DDL locks, or network latency can cause replicas to fall severely behind the primary:

Stale Reads: Application queries directed to the replica return outdated or missing data, impacting user experience.
Dangerous Failovers: If an automated failover promotes a lagging replica during an incident, unapplied transactions are lost.
Silent Thread Halts: When a replication error occurs (e.g. key collision or schema mismatch), the SQL applier thread silently halts (Replica_SQL_Running: No), causing lag to accumulate indefinitely.

This blueprint provides automated, continuous replication observability for MySQL and MariaDB:

Queries SHOW REPLICA STATUS every 5 minutes via the MySQL JDBC plugin.
Automatically supports both modern MySQL 8.0+ column definitions (Replica_IO_Running, Seconds_Behind_Source) and legacy MySQL 5.7 / MariaDB columns (Slave_IO_Running, Seconds_Behind_Master).
Evaluates IO thread status, SQL applier status, and numeric replication delay in seconds.
Dispatches rich Slack incident cards when lag breaches configurable SLA thresholds or when replication crashes.

How it works

scheduled_replication_check triggers recurring checks every 5 minutes (or on demand).
check_replica_status executes SHOW REPLICA STATUS against the target replica instance via io.kestra.plugin.jdbc.mysql.Query.
evaluate_replica_health runs a Python script to parse telemetry, normalize column differences across MySQL versions, extract error messages, and calculate lag metrics.
evaluate_sla_breach (io.kestra.plugin.core.flow.If) branches:
- If threads are stopped or lag exceeds max_lag_seconds: dispatches a Slack alert with binlog coordinates and error diagnostics, and optionally triggers fail_execution.
- If healthy: logs in-sync verification metrics.
An errors block catches connection drops or bad credentials.

What you get

Proactive SLA Breach Detection: Catch replication delays before users experience stale reads.
Cross-Version Compatibility: Works seamlessly across MySQL 5.7, MySQL 8.0+, and MariaDB.
Actionable Diagnostic Context: Alerts include exact binlog filenames, coordinates, and error strings for rapid troubleshooting.
Zero Credential Leaks: Database passwords and webhook tokens remain secured via Kestra namespace secrets.

Prerequisites

A MySQL (5.7+, 8.0+) or MariaDB instance configured as a read replica.
Database user with REPLICATION CLIENT (or REPLICATION SLAVE) privileges.
An incoming Slack webhook URL for notifications.

Secrets

MYSQL_USERNAME: Database username with replication inspection privileges.
MYSQL_PASSWORD: Database password.
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL.

Quick start

Add MYSQL_USERNAME, MYSQL_PASSWORD, and SLACK_WEBHOOK_URL to your Kestra namespace secrets.
Configure mysql_url with your replica's JDBC connection string (e.g. jdbc:mysql://replica.internal:3306/production).
Adjust max_lag_seconds to match your application's replication SLA (default: 60s).
Trigger a manual execution to test initial connectivity and replication telemetry.

How to extend

Automated Traffic Shunting: Integrate with ProxySQL or AWS Route 53 to remove a lagging replica from the active read pool automatically.
PagerDuty Escalation: Add io.kestra.plugin.pagerduty.Alert alongside Slack for high-severity thread halt incidents.
Multi-Replica Sweep: Wrap the query task in an io.kestra.plugin.core.flow.Loop over a list of replica endpoints to audit an entire read replica cluster.

MySQL Read Replica Replication Lag and Health Sentinel

How it works

What you get

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MySQL Read Replica Replication Lag and Health Sentinel

How it works

What you get

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Alert When Streaming Windows Ship With Late or Missing Rows

Hold Every Release to a Lighthouse Score Budget

Kinesis Stream Throughput Quota Gate

Langfuse Daily LLM Trace Quality and Cost Sentinel

Kubernetes Secret Exposure Sentinel

KeyDB Active-Replica Multi-Master Peering Sentinel

More Infrastructure Blueprints

Retire Unused Kestra Flows with Quarantine, Approval and Restore

Gate a Kestra upgrade against every live flow

KMS Key Rotation Compliance Gate

Kong Ingress Upstream Target Unhealthy and Circuit Breaker Sentinel

Kinesis Stream Throughput Quota Gate

Knative Service Revision Failure and Scale Sentinel

New to Kestra?