MySQL Replica super_read_only Enforcement Guard

id: mysql-replica-super-read-only-enforcement-guard
namespace: company.team

inputs:
  - id: target_url
    type: STRING
    defaults: jdbc:mysql://localhost:3306/
    description: JDBC URL of the MySQL instance to check. Point this at a replica.
      An instance with no replication configured is reported and left untouched,
      so a source pointed at by mistake is never frozen.

  - 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: dry_run
    type: BOOL
    defaults: true
    description: When true, report the read-only state of the instance without changing it.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true and dry_run is false, set super_read_only ON behind an
      approval gate, then re-read the state and fail if the instance is still
      writable.

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "*/15 * * * *"
    description: Check replica write protection every fifteen minutes. Disabled by
      default and pinned to safe mode so it reports without changing anything
      until an operator enables it.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: detect_replication
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Read SHOW REPLICA STATUS. A returned row is the only proof that
      this instance is a replica. Zero rows means replication is not configured,
      which is how a source is recognised and left alone.
    url: "{{ inputs.target_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: detect_state
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Read the global write protection and the GTID set the instance has
      executed. super_read_only blocks writes from every client including a user
      holding CONNECTION_ADMIN, which is what a replica needs.
    url: "{{ inputs.target_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_ONE
    sql: >-
      SELECT IF(@@global.super_read_only, 'ON', 'OFF') AS sro_state,
      IF(@@global.read_only, 'ON', 'OFF') AS ro_state, @@global.server_id AS
      server_id, @@global.hostname AS host_name, REPLACE(@@global.gtid_executed,
      CHAR(10), ' ') AS gtid_executed

  - id: classify
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      Sort the instance into one state. NOT_A_REPLICA is zero SHOW REPLICA
      STATUS rows. COMPLIANT is a replica that already has super_read_only ON.
      WRITABLE_REPLICA is a replica with super_read_only OFF, which accepts a
      stray client write that only ever exists on that replica.
    values:
      classification: >-
        {{ {'repl_rows': (outputs.detect_replication.size ?? 0), 'sro':
        outputs.detect_state.row.sro_state} | jq('if .repl_rows == 0 then
        "NOT_A_REPLICA" elif .sro == "ON" then "COMPLIANT" else
        "WRITABLE_REPLICA" end') | first }}
      sro_before: "{{ outputs.detect_state.row.sro_state }}"
      ro_before: "{{ outputs.detect_state.row.ro_state }}"
      server_id: "{{ outputs.detect_state.row.server_id }}"
      host_name: "{{ outputs.detect_state.row.host_name }}"
      gtid_executed: "{{ outputs.detect_state.row.gtid_executed }}"
      replica_rows: "{{ outputs.detect_replication.size ?? 0 }}"
      source_host: >-
        {{ (outputs.detect_replication.rows ?? []) | jq('if length == 0 then
        "none" else (.[0].Source_Host // "unknown") end') | first }}
      io_running: >-
        {{ (outputs.detect_replication.rows ?? []) | jq('if length == 0 then
        "n/a" else (.[0].Replica_IO_Running // "No") end') | first }}
      sql_running: >-
        {{ (outputs.detect_replication.rows ?? []) | jq('if length == 0 then
        "n/a" else (.[0].Replica_SQL_Running // "No") end') | first }}

  - id: log_status
    type: io.kestra.plugin.core.log.Log
    message: |
      MySQL instance {{ outputs.classify.values.host_name }} (server_id {{ outputs.classify.values.server_id }}) at {{ inputs.target_url }} classified {{ outputs.classify.values.classification }}.
      super_read_only={{ outputs.classify.values.sro_before }} read_only={{ outputs.classify.values.ro_before }}. SHOW REPLICA STATUS rows={{ outputs.classify.values.replica_rows }} source_host={{ outputs.classify.values.source_host }} Replica_IO_Running={{ outputs.classify.values.io_running }} Replica_SQL_Running={{ outputs.classify.values.sql_running }}.
      Executed GTID set: {{ outputs.classify.values.gtid_executed }}
      Mode dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}.

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify.values.classification }}"
    cases:
      COMPLIANT:
        - id: compliant_log
          type: io.kestra.plugin.core.log.Log
          message: "Replica {{ outputs.classify.values.host_name }} already has
            super_read_only ON and read_only {{
            outputs.classify.values.ro_before }}. A client write is refused, so
            no errant transaction can be created here. Nothing to do."
      NOT_A_REPLICA:
        - id: not_a_replica_alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: The target returned no SHOW REPLICA STATUS row, so it is not a
            replica. Nothing is changed on it. Setting read_only on a source
            would stop application writes.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {"text": "MySQL super_read_only guard: {{ inputs.target_url }} (host {{ outputs.classify.values.host_name }}, server_id {{ outputs.classify.values.server_id }}) returned no SHOW REPLICA STATUS row, so it is not a replica. No change was made. It is still super_read_only {{ outputs.classify.values.sro_before }} and read_only {{ outputs.classify.values.ro_before }}. Point target_url at a replica. Execution {{ execution.id }}."}
        - id: not_a_replica_fail
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: >-
            Target {{ inputs.target_url }} (host {{
            outputs.classify.values.host_name }}, server_id {{
            outputs.classify.values.server_id }}) returned no SHOW REPLICA
            STATUS row, so replication is not configured on it. The guard
            refused to touch it and left super_read_only {{
            outputs.classify.values.sro_before }} with read_only {{
            outputs.classify.values.ro_before }}. Enforcing read-only on a
            source would stop every application write. Point target_url at a
            replica.
      WRITABLE_REPLICA:
        - id: writable_alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alert on every writable replica, in dry run as well as in remediate
            mode, so a human sees the exposure even when nothing is changed.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {"text": "MySQL super_read_only guard: replica {{ outputs.classify.values.host_name }} (server_id {{ outputs.classify.values.server_id }}) at {{ inputs.target_url }} is replicating from {{ outputs.classify.values.source_host }} with super_read_only OFF and read_only {{ outputs.classify.values.ro_before }}. A stray client write lands only here and becomes an errant transaction that diverges the GTID set from the source. dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}. Execution {{ execution.id }}."}
        - 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 change. SET GLOBAL
                super_read_only never runs on its own.
              pauseDuration: PT1H
              behavior: FAIL
              onResume:
                - id: approved
                  type: BOOL
                  defaults: false
                  description: Set true to run SET GLOBAL super_read_only = ON on this replica.
                    Anything else leaves the replica writable.
            - id: apply
              type: io.kestra.plugin.core.flow.If
              condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
              then:
                - id: enforce_super_read_only
                  type: io.kestra.plugin.jdbc.mysql.Query
                  description: Set super_read_only ON, which also sets read_only ON. allowFailure
                    so a rejected statement does not end the pass and the
                    re-read stays the real gate. Replication applier threads are
                    not affected by either variable, so the replica keeps
                    applying from the source.
                  allowFailure: true
                  url: "{{ inputs.target_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: SET GLOBAL super_read_only = ON
                - id: settle
                  type: io.kestra.plugin.core.flow.Sleep
                  duration: PT3S
                - id: reverify_state
                  type: io.kestra.plugin.jdbc.mysql.Query
                  description: Re-read both variables. This read is the proof, not the SET GLOBAL
                    call.
                  url: "{{ inputs.target_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_ONE
                  sql: >-
                    SELECT IF(@@global.super_read_only, 'ON', 'OFF') AS
                    sro_state, IF(@@global.read_only, 'ON', 'OFF') AS ro_state,
                    REPLACE(@@global.gtid_executed, CHAR(10), ' ') AS
                    gtid_executed
                - id: reverify_replication
                  type: io.kestra.plugin.jdbc.mysql.Query
                  description: Re-read SHOW REPLICA STATUS so the run also proves the change did
                    not disturb the replication threads.
                  url: "{{ inputs.target_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
                  type: io.kestra.plugin.core.output.OutputValues
                  values:
                    enforced: >-
                      {{ {'sro': outputs.reverify_state.row.sro_state, 'ro':
                      outputs.reverify_state.row.ro_state} | jq('.sro == "ON"
                      and .ro == "ON"') | first }}
                    sro_after: "{{ outputs.reverify_state.row.sro_state }}"
                    ro_after: "{{ outputs.reverify_state.row.ro_state }}"
                    gtid_after: "{{ outputs.reverify_state.row.gtid_executed }}"
                    io_after: >-
                      {{ (outputs.reverify_replication.rows ?? []) | jq('if
                      length == 0 then "n/a" else (.[0].Replica_IO_Running //
                      "No") end') | first }}
                    sql_after: >-
                      {{ (outputs.reverify_replication.rows ?? []) | jq('if
                      length == 0 then "n/a" else (.[0].Replica_SQL_Running //
                      "No") end') | first }}
                - id: verify_result
                  type: io.kestra.plugin.core.flow.If
                  condition: "{{ outputs.verify.values.enforced == 'true' }}"
                  then:
                    - id: remediated
                      type: io.kestra.plugin.core.log.Log
                      message: |
                        Write protection confirmed on {{ outputs.classify.values.host_name }}. Before: super_read_only={{ outputs.classify.values.sro_before }} read_only={{ outputs.classify.values.ro_before }}. After: super_read_only={{ outputs.verify.values.sro_after }} read_only={{ outputs.verify.values.ro_after }}.
                        Replication still running: Replica_IO_Running={{ outputs.verify.values.io_after }} Replica_SQL_Running={{ outputs.verify.values.sql_after }}. Executed GTID set after the change: {{ outputs.verify.values.gtid_after }}
                  else:
                    - id: remediation_failed
                      type: io.kestra.plugin.core.execution.Fail
                      errorMessage: >-
                        Replica {{ outputs.classify.values.host_name }} at {{
                        inputs.target_url }} is still writable after the change.
                        The re-read reports super_read_only {{
                        outputs.verify.values.sro_after }} and read_only {{
                        outputs.verify.values.ro_after }}. The connecting user
                        probably lacks SYSTEM_VARIABLES_ADMIN. Something may
                        also have set the variable back. The replica still
                        accepts client writes, so a stray write can still become
                        an errant transaction.
              else:
                - id: not_approved
                  type: io.kestra.plugin.core.log.Log
                  message: "Approval not granted, so SET GLOBAL super_read_only was not run.
                    Replica {{ outputs.classify.values.host_name }} is still
                    writable with super_read_only {{
                    outputs.classify.values.sro_before }}."
          else:
            - id: report_only
              type: io.kestra.plugin.core.log.Log
              message: "Report only (dry_run={{ inputs.dry_run }}, auto_remediate={{
                inputs.auto_remediate }}). Replica {{
                outputs.classify.values.host_name }} has super_read_only {{
                outputs.classify.values.sro_before }} and read_only {{
                outputs.classify.values.ro_before }}. SET GLOBAL super_read_only
                = ON would be run here. No change made."
    defaults:
      - id: unexpected_fail
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Unexpected classification {{
          outputs.classify.values.classification }} for {{ inputs.target_url
          }}."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post to Slack on any failure, including an unreachable instance, a
      target that is not a replica or a replica that is still writable after the
      change.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "MySQL super_read_only guard FAILED for {{ inputs.target_url }} (execution {{ execution.id }}). Either the instance is unreachable, it is not a replica or it is still writable after the change. Check the run."}