MySQL Stopped Replication Recovery Guard with Kestra

id: mysql-stopped-replication-recovery-guard
namespace: company.team

inputs:
  - id: replica_url
    type: STRING
    defaults: jdbc:mysql://localhost:3306/
    description: JDBC URL of the MySQL read replica to guard. Point this at the
      replica, not the source.

  - id: use_secret_auth
    type: BOOL
    defaults: true
    description: Leave true in production so the connection reads MYSQL_USERNAME and
      MYSQL_PASSWORD from secrets. Set false only to pass credentials inline in
      a disposable test environment.

  - id: mysql_username
    type: STRING
    defaults: root
    description: Username used only when use_secret_auth is false. Ignored in production.

  - id: mysql_password
    type: STRING
    defaults: ""
    description: Password used only when use_secret_auth is false. Ignored in production.

  - id: max_lag_seconds
    type: INT
    defaults: 30
    description: Lag ceiling in seconds. When both threads run but
      Seconds_Behind_Source is above this, the replica is reported as lagging.

  - id: dry_run
    type: BOOL
    defaults: true
    description: When true, report the replication state without restarting anything.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true and dry_run is false, restart a transient stop behind an
      approval gate, then re-read status.

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "*/5 * * * *"
    description: Poll replica health every five minutes. Disabled by default and
      pinned to safe mode so it reports without restarting until an operator
      enables it.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: detect
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Read SHOW REPLICA STATUS from the replica. Zero rows means the
      target is not configured as a replica.
    url: "{{ inputs.replica_url }}"
    username: "{{ inputs.use_secret_auth ? secret('MYSQL_USERNAME') :
      inputs.mysql_username }}"
    password: "{{ inputs.use_secret_auth ? secret('MYSQL_PASSWORD') :
      inputs.mysql_password }}"
    fetchType: FETCH
    sql: SHOW REPLICA STATUS

  - id: classify
    type: io.kestra.plugin.core.output.OutputValues
    description: Classify the replica state from the status row. A thread that is
      not Yes with Last_SQL_Errno zero is a transient stop. A thread that is not
      Yes with a non-zero Last_SQL_Errno is a data error that a restart must
      never paper over.
    values:
      classification: >-
        {{ outputs.detect.rows | jq('if length == 0 then "NOT_A_REPLICA" else
        .[0] as $r | ($r.Replica_IO_Running // "No") as $io |
        ($r.Replica_SQL_Running // "No") as $sql | ((($r.Last_SQL_Errno // 0) |
        tostring) | tonumber) as $errno | ((($r.Seconds_Behind_Source // -1) |
        tostring) | tonumber) as $lag | if ($io == "Yes" and $sql == "Yes") then
        (if $lag > ' ~ inputs.max_lag_seconds ~ ' then "LAGGING" else "HEALTHY"
        end) elif ($errno != 0) then "STOPPED_DATA_ERROR" else
        "STOPPED_TRANSIENT" end end') | first }}
      io_running: >-
        {{ outputs.detect.rows | jq('.[0].Replica_IO_Running // "No"') | first }}
      sql_running: >-
        {{ outputs.detect.rows | jq('.[0].Replica_SQL_Running // "No"') | first
        }}
      last_sql_errno: >-
        {{ outputs.detect.rows | jq('(.[0].Last_SQL_Errno // 0) | tostring') |
        first }}
      last_io_errno: >-
        {{ outputs.detect.rows | jq('(.[0].Last_IO_Errno // 0) | tostring') |
        first }}
      seconds_behind: >-
        {{ outputs.detect.rows | jq('(.[0].Seconds_Behind_Source // -1) |
        tostring') | first }}
      last_error: >-
        {{ outputs.detect.rows | jq('((.[0].Last_SQL_Error // "") + " " +
        (.[0].Last_IO_Error // ""))') | first }}

  - id: log_status
    type: io.kestra.plugin.core.log.Log
    message: |
      MySQL replica {{ inputs.replica_url }} classification {{ outputs.classify.values.classification }}.
      Replica_IO_Running={{ outputs.classify.values.io_running }} Replica_SQL_Running={{ outputs.classify.values.sql_running }} Last_IO_Errno={{ outputs.classify.values.last_io_errno }} Last_SQL_Errno={{ outputs.classify.values.last_sql_errno }} Seconds_Behind_Source={{ outputs.classify.values.seconds_behind }}.
      Mode dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}. Last error text: {{ outputs.classify.values.last_error }}

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify.values.classification }}"
    cases:
      HEALTHY:
        - id: healthy_log
          type: io.kestra.plugin.core.log.Log
          message: "Replica is healthy. Both threads run and lag is within {{
            inputs.max_lag_seconds }} second(s). Nothing to do."
      LAGGING:
        - id: lagging_alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alert that both threads run but the replica is behind. This guard
            does not restart a lagging replica because the threads are already
            running.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {"text": "MySQL replication guard: replica {{ inputs.replica_url }} is lagging. Seconds_Behind_Source {{ outputs.classify.values.seconds_behind }}, threshold {{ inputs.max_lag_seconds }}. Both threads run, so no restart is attempted."}
        - id: lagging_log
          type: io.kestra.plugin.core.log.Log
          message: "Replica is lagging by {{ outputs.classify.values.seconds_behind }}
            second(s). Both threads run, so this guard only alerts."
      STOPPED_DATA_ERROR:
        - id: data_error_alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Loud alert for a data-error stop. This path never restarts
            replication because a restart would skip the failing event and drift
            the replica from the source.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {"text": "MySQL replication guard: replica {{ inputs.replica_url }} STOPPED on a DATA ERROR (Last_SQL_Errno {{ outputs.classify.values.last_sql_errno }}). A restart is refused on purpose. A human must resolve the conflict. Execution {{ execution.id }}."}
        - id: data_error_fail
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: >-
            Replica {{ inputs.replica_url }} stopped on a data error
            (Last_SQL_Errno {{ outputs.classify.values.last_sql_errno }}). This
            guard refuses to restart a data-error stop because START REPLICA
            would leave the failing event unapplied and silently drift the
            replica from the source. Resolve the conflict by hand.
      NOT_A_REPLICA:
        - id: not_replica_alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: The target returned no SHOW REPLICA STATUS row, so it is not
            configured as a replica.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {"text": "MySQL replication guard: target {{ inputs.replica_url }} returned no SHOW REPLICA STATUS row. It is not configured as a read replica. Check the connection target. Execution {{ execution.id }}."}
        - id: not_replica_fail
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: >-
            Target {{ inputs.replica_url }} returned no SHOW REPLICA STATUS row,
            so it is not a configured replica. Point replica_url at the replica
            instance.
      STOPPED_TRANSIENT:
        - id: transient_alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alert on every transient stop, in dry run and in remediate mode, so
            a human sees the outage even when nothing is restarted.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {"text": "MySQL replication guard: replica {{ inputs.replica_url }} has a STOPPED thread with no data error (Replica_IO_Running {{ outputs.classify.values.io_running }}, Replica_SQL_Running {{ outputs.classify.values.sql_running }}, Last_SQL_Errno {{ outputs.classify.values.last_sql_errno }}). Transient stop. dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}."}
        - id: mode
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
          then:
            - id: approval_gate
              type: io.kestra.plugin.core.flow.Pause
              description: Hold the run until a human approves the restart. START REPLICA
                never runs on its own.
              pauseDuration: PT1H
              behavior: FAIL
              onResume:
                - id: approved
                  type: BOOL
                  defaults: false
                  description: Set true to run START REPLICA. Anything else leaves replication
                    stopped.
            - id: apply
              type: io.kestra.plugin.core.flow.If
              condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
              then:
                - id: restart_replica
                  type: io.kestra.plugin.jdbc.mysql.Query
                  description: Restart replication on the replica. allowFailure so a restart error
                    does not stop the pass and the re-read stays the real gate.
                  allowFailure: true
                  url: "{{ inputs.replica_url }}"
                  username: "{{ inputs.use_secret_auth ? secret('MYSQL_USERNAME') :
                    inputs.mysql_username }}"
                  password: "{{ inputs.use_secret_auth ? secret('MYSQL_PASSWORD') :
                    inputs.mysql_password }}"
                  fetchType: NONE
                  sql: START REPLICA
                - id: settle
                  type: io.kestra.plugin.core.flow.Sleep
                  duration: PT8S
                - id: reverify
                  type: io.kestra.plugin.jdbc.mysql.Query
                  description: Re-read SHOW REPLICA STATUS after the restart. This read is the
                    proof, not the START REPLICA call.
                  url: "{{ inputs.replica_url }}"
                  username: "{{ inputs.use_secret_auth ? secret('MYSQL_USERNAME') :
                    inputs.mysql_username }}"
                  password: "{{ inputs.use_secret_auth ? secret('MYSQL_PASSWORD') :
                    inputs.mysql_password }}"
                  fetchType: FETCH
                  sql: SHOW REPLICA STATUS
                - id: verify_threads
                  type: io.kestra.plugin.core.output.OutputValues
                  values:
                    threads_up: >-
                      {{ outputs.reverify.rows | jq('(length > 0) and
                      ((.[0].Replica_IO_Running // "No") == "Yes") and
                      ((.[0].Replica_SQL_Running // "No") == "Yes")') | first }}
                    rv_io: >-
                      {{ outputs.reverify.rows | jq('.[0].Replica_IO_Running //
                      "No"') | first }}
                    rv_sql: >-
                      {{ outputs.reverify.rows | jq('.[0].Replica_SQL_Running //
                      "No"') | first }}
                    rv_errno: >-
                      {{ outputs.reverify.rows | jq('(.[0].Last_SQL_Errno // 0)
                      | tostring') | first }}
                - id: verify_result
                  type: io.kestra.plugin.core.flow.If
                  condition: "{{ outputs.verify_threads.values.threads_up == 'true' }}"
                  then:
                    - id: remediated
                      type: io.kestra.plugin.core.log.Log
                      message: "Restart confirmed. Before: IO={{ outputs.classify.values.io_running }}
                        SQL={{ outputs.classify.values.sql_running }}. After:
                        IO={{ outputs.verify_threads.values.rv_io }} SQL={{
                        outputs.verify_threads.values.rv_sql }}
                        Last_SQL_Errno={{ outputs.verify_threads.values.rv_errno
                        }}."
                  else:
                    - id: remediation_failed
                      type: io.kestra.plugin.core.execution.Fail
                      errorMessage: >-
                        Replica {{ inputs.replica_url }} did not resume after
                        START REPLICA. After the restart Replica_IO_Running is
                        {{ outputs.verify_threads.values.rv_io }},
                        Replica_SQL_Running is {{
                        outputs.verify_threads.values.rv_sql }}, Last_SQL_Errno
                        is {{ outputs.verify_threads.values.rv_errno }}. The
                        source may be unreachable or a data error surfaced on
                        apply.
              else:
                - id: not_approved
                  type: io.kestra.plugin.core.log.Log
                  message: "Approval not granted, so START REPLICA was not run. Replica {{
                    inputs.replica_url }} is still stopped."
          else:
            - id: report_only
              type: io.kestra.plugin.core.log.Log
              message: "Report only (dry_run={{ inputs.dry_run }}, auto_remediate={{
                inputs.auto_remediate }}). A transient stop was found and START
                REPLICA would be run. No change made."
    defaults:
      - id: unexpected_fail
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Unexpected classification {{
          outputs.classify.values.classification }} for replica {{
          inputs.replica_url }}."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post to Slack on any failure, including a connection error, a
      refused data-error stop or a replica that did not resume after a restart.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "MySQL replication guard FAILED for replica {{ inputs.replica_url }} (execution {{ execution.id }}). A connection error, a refused data-error stop or a replica that did not resume after a restart. Check the run."}

A MySQL read replica can lose its IO or SQL thread after a transient fault such as a network blip or a dropped connection to the source. The threads stay down, the replica falls behind in silence and keeps serving stale reads or breaks a failover. This guard reads the replica state, restarts replication only when the stop is transient and safe, then proves the threads came back.

This blueprint was created by zkasuran.

Safety

Read-only by default. dry_run is true and auto_remediate is false, so a plain run reports and never restarts.
A restart happens only when auto_remediate is true and dry_run is false. It then waits for a human to resume the Pause with approved set true. An unanswered gate reads as not approved.
Only a transient stop is restarted. A stop that carries a non-zero Last_SQL_Errno is a data error, so the guard refuses the restart and fails loudly. START REPLICA on a data error would skip the failing event and drift the replica from the source.
The guard never sets sql_slave_skip_counter and never skips an event. It only runs START REPLICA and only for the transient case.

How it works

detect (io.kestra.plugin.jdbc.mysql.Query) reads SHOW REPLICA STATUS from the replica. Zero rows means the target is not a replica.
classify (io.kestra.plugin.core.output.OutputValues) reads the status row and sorts it into one state. HEALTHY is both threads Yes with lag within the ceiling. LAGGING is both threads Yes with lag over the ceiling. STOPPED_TRANSIENT is a thread not Yes with Last_SQL_Errno zero. STOPPED_DATA_ERROR is a thread not Yes with a non-zero Last_SQL_Errno. NOT_A_REPLICA is zero rows.
log_status (io.kestra.plugin.core.log.Log) records the classification with the raw status fields.
route (io.kestra.plugin.core.flow.Switch) branches on the classification. HEALTHY logs and ends. LAGGING alerts and ends because both threads already run. STOPPED_DATA_ERROR and NOT_A_REPLICA alert and fail.
For STOPPED_TRANSIENT, transient_alert posts the outage. mode (io.kestra.plugin.core.flow.If) enters remediation only when auto_remediate is true and dry_run is false. Otherwise report_only logs what would run.
approval_gate (io.kestra.plugin.core.flow.Pause) holds the run until a human resumes it with approved. restart_replica runs START REPLICA only when approved is true.
reverify re-reads SHOW REPLICA STATUS. verify_result fails the run through io.kestra.plugin.core.execution.Fail when the two threads are not both running again.

What you get

A recovery step that acts only on a safe transient stop and leaves a data-error stop for a human.
A gate that is safe to leave scheduled. It reports until an operator turns on remediation and approves each run.
A real verification step. The run passes only when the second read of SHOW REPLICA STATUS shows both threads running.
Slack alerts on every stop, on a lagging replica and on any failure.

Who it's for

Platform engineers who run MySQL read replicas and want a safe self-heal for the common transient stop.
DBAs who need the dangerous data-error stop surfaced loudly rather than papered over by an auto-skip.
Teams evaluating the MySQL plugin who want an example that reads replica telemetry and runs gated recovery.

Why orchestrate this with Kestra

MySQL has no scheduler, no approval step and no way to prove a restart worked. Kestra adds the missing pieces. It polls the replica, gates the restart behind dry-run and approval, re-reads SHOW REPLICA STATUS to turn a hopeful restart into a verified one, applies one connection to every task and posts a Slack alert the moment a thread stops or a step fails.

Tested end to end

Run on Kestra 2.0.5 against a MySQL 8.0 source and replica pair with GTID replication.

Detector can fail: a healthy pair returned HEALTHY with both threads Yes, the same replica after STOP REPLICA returned STOPPED_TRANSIENT.
Dry run on a transient stop detected it, alerted and left both threads down (confirmed with the mysql client, No then No).
Remediate with approval ran START REPLICA and the re-read reported both threads Yes (No then Yes in SHOW REPLICA STATUS).
The approval gate held. A run resumed with approved false ran no restart.
A data-error stop (duplicate key, Last_SQL_Errno 1062) was refused. No START REPLICA ran and Exec_Source_Log_Pos stayed put.
A re-read after a restart against an unreachable source failed the run rather than reporting success.
An unreachable replica failed the detect step and fired the Slack alert.

Prerequisites

A MySQL read replica reachable over JDBC and a user that can run SHOW REPLICA STATUS and START REPLICA. The connecting user needs the REPLICATION CLIENT privilege and the REPLICATION_SLAVE_ADMIN privilege.

Secrets

MYSQL_USERNAME: MySQL user, read when use_secret_auth is true.
MYSQL_PASSWORD: password for that user.
SLACK_WEBHOOK_URL: used by the alerts and the errors block.
In Kestra OSS, provide each secret as a base64-encoded environment variable prefixed SECRET_.

Inputs

Name	Type	Default	Description
`replica_url`	STRING	`jdbc:mysql://localhost:3306/`	JDBC URL of the replica to guard.
`use_secret_auth`	BOOL	`true`	Read credentials from secrets. Set false only for inline test credentials.
`mysql_username`	STRING	`root`	Username used only when use_secret_auth is false.
`mysql_password`	STRING	``	Password used only when use_secret_auth is false.
`max_lag_seconds`	INT	`30`	Lag ceiling in seconds for the lagging state.
`dry_run`	BOOL	`true`	Report without restarting.
`auto_remediate`	BOOL	`false`	Restart a transient stop behind the approval gate when dry_run is false.

Quick start

Set MYSQL_USERNAME, MYSQL_PASSWORD and SLACK_WEBHOOK_URL as namespace secrets.
Import this flow.
Run it with replica_url pointed at a read replica. The default dry run reports the replica state.
To recover a transient stop, run again with dry_run false and auto_remediate true, then resume the paused execution with approved true.

How to extend

Loop over several replicas from one flow by moving the detect and restart steps into a replica loop.
Add a channel name to the SQL for a multi-source replica, for example SHOW REPLICA STATUS FOR CHANNEL 'ch1'.
Open a ticket on a STOPPED_DATA_ERROR classification so the human handoff is tracked.
Fail the run as soon as detection finds any stop, for a read-only drift check with no remediation.

MySQL Stopped Replication Recovery Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MySQL Stopped Replication Recovery Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

More Infrastructure Blueprints

New to Kestra?