Prove Your MySQL Backups Restore to the Second, or to the Statement Before a Mistake

id: mysql-binlog-point-in-time-restore-drill
namespace: company.team
description: |
  Prove that your MySQL backups restore to the moment you choose. The drill takes the newest mysqldump and the archived
  binlogs, checks the binlog chain for gaps and bad checksums, restores into a throwaway MySQL, replays the binlogs up
  to a target time or right before a given statement (a DROP TABLE, a DELETE without WHERE), runs your checks and
  measures the restore time. Demo mode needs only Docker.

triggers:
  - id: after_backup
    type: io.kestra.plugin.core.trigger.Webhook
    description: Call it when the nightly mysqldump has been uploaded, so every new
      dump is proven restorable with the binlogs that follow it.
    key: "{{ secret('MYSQL_PITR_DRILL_WEBHOOK_KEY') }}"
  - id: weekly_drill
    type: io.kestra.plugin.core.trigger.Schedule
    description: Safety net when the backup job does not call the webhook. Shipped disabled.
    cron: "20 5 * * 1"
    disabled: true

concurrency:
  limit: 1
  behavior: QUEUE

sla:
  - id: restore_deadline
    type: MAX_DURATION
    duration: PT1H
    behavior: FAIL
    labels:
      drill: deadline_exceeded

inputs:
  - id: mode
    type: SELECT
    values: [ demo, s3 ]
    defaults: demo
    description: demo builds a source database, dumps it, keeps writing, then drops
      a table by mistake, and recovers to the statement before the DROP. s3
      drills your own dumps and binlogs.
  - id: s3_bucket
    type: STRING
    required: false
  - id: dump_prefix
    type: STRING
    defaults: mysql/dumps/
    description: Where the mysqldump files are (*.sql.gz). The newest one is restored.
  - id: binlog_prefix
    type: STRING
    defaults: mysql/binlogs/
    description: Where the archived binlogs are, under their original names
      (binlog.000042), as written by mysqlbinlog --read-from-remote-server
      --raw.
  - id: s3_endpoint
    type: STRING
    defaults: https://s3.us-east-1.amazonaws.com
    description: For MinIO, Ceph, SeaweedFS or R2, their S3 URL.
  - id: target_time
    type: STRING
    required: false
    description: Recover to this moment, UTC, "2026-10-16 10:15:00". mysqlbinlog
      stops at the first event at or after this second, so nothing from the
      target second itself is replayed; binlog timestamps have one-second
      resolution. Use stop_before_statement to stop exactly before a mistake.
      Empty replays every archived binlog.
  - id: stop_before_statement
    type: STRING
    required: false
    description: A regular expression. The replay stops right before the first
      transaction whose statement matches it, for example "DROP TABLE
      `?shop`?.`?orders" or "DELETE FROM `shop`.`orders`" (row events are
      matched on mysqlbinlog -v pseudo-SQL). The demo uses "DROP TABLE".
  - id: checks
    type: ARRAY
    itemType: STRING
    required: false
    description: SQL statements run on the restored server. Each returns one row
      with columns ok (1 or 0) and detail. The demo adds its own row count
      check.
  - id: archive_lag_minutes
    type: INT
    defaults: 30
    description: The newest archived binlog event must be younger than this. It is
      how much data you can lose if the server dies now.
  - id: rto_minutes
    type: INT
    defaults: 30
    description: The restore (dump load and binlog replay) must finish within this
      many minutes.
  - id: mysql_image
    type: STRING
    defaults: percona/percona-server:8.0
    description: Needs mysqld, mysqldump and mysqlbinlog of your server's major
      version. The official mysql image no longer ships mysqlbinlog; Percona
      Server is a drop-in MySQL build that does, and restores MySQL Community
      dumps and binlogs.

variables:
  last_good_key: mysql_pitr_drill_last_good

tasks:
  - id: s3_mode
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.mode == 's3' }}"
    then:
      - id: check_bucket
        type: io.kestra.plugin.core.flow.If
        condition: "{{ (inputs.s3_bucket ?? '') == '' }}"
        then:
          - id: missing_bucket
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "DRILL FAILED: s3 mode needs s3_bucket."
      - id: list_dumps
        type: io.kestra.plugin.aws.s3.List
        endpointOverride: "{{ inputs.s3_endpoint }}"
        forcePathStyle: true
        compatibilityMode: true
        region: us-east-1
        accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
        bucket: "{{ inputs.s3_bucket }}"
        prefix: "{{ inputs.dump_prefix }}"
        regexp: ".*\\.sql\\.gz$"
        filter: FILES
      - id: dump_found
        type: io.kestra.plugin.core.flow.If
        condition: "{{ (outputs.list_dumps.objects ?? []) | length == 0 }}"
        then:
          - id: no_dump
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "DRILL FAILED: no *.sql.gz under s3://{{ inputs.s3_bucket }}/{{
              inputs.dump_prefix }}. The dump job is not uploading where the
              drill looks, or not running."
      - id: download_dump
        type: io.kestra.plugin.aws.s3.Download
        description: The newest dump by upload time.
        endpointOverride: "{{ inputs.s3_endpoint }}"
        forcePathStyle: true
        compatibilityMode: true
        region: us-east-1
        accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
        bucket: "{{ inputs.s3_bucket }}"
        key: "{{ outputs.list_dumps.objects | jq('sort_by(.lastModified) | last | .key')
          | first }}"
      - id: download_binlogs
        type: io.kestra.plugin.aws.s3.Downloads
        endpointOverride: "{{ inputs.s3_endpoint }}"
        forcePathStyle: true
        compatibilityMode: true
        region: us-east-1
        accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
        bucket: "{{ inputs.s3_bucket }}"
        prefix: "{{ inputs.binlog_prefix }}"
        regexp: ".*\\.[0-9]{6}$"
        action: NONE

  - id: restore
    type: io.kestra.plugin.scripts.shell.Commands
    description: |
      One throwaway container: a scratch MySQL with the source's GTID mode, the dump, the binlog chain from the dump's
      coordinates on, replayed in a single mysqlbinlog call, then the checks. Reports JSON whether it passes or not.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      pullPolicy: IF_NOT_PRESENT
      user: root
    containerImage: "{{ inputs.mysql_image }}"
    inputFiles: "{{ inputs.mode == 's3' ? (outputs.download_binlogs.outputFiles |
      merge({'dump.sql.gz': outputs.download_dump.uri})) : {} }}"
    env:
      MODE: "{{ inputs.mode }}"
      DUMP_KEY: "{{ inputs.mode == 's3' ? (outputs.list_dumps.objects |
        jq('sort_by(.lastModified) | last | .key') | first) : 'demo' }}"
      TARGET_TIME: "{{ inputs.target_time ?? '' }}"
      STOP_RE: "{{ inputs.stop_before_statement ?? '' }}"
      CHECKS: "{{ (inputs.checks ?? []) | join('

        ') }}"
      LAG_MIN: "{{ inputs.archive_lag_minutes }}"
      TZ: UTC
    commands:
      - |
        set -eu
        W=/tmp/drill; rm -rf $W; mkdir -p $W/archive $W/srcbin; chown -R mysql $W
        fail() { echo "DRILL FAILED: $*" >&2; exit 1; }
        problems=""; problem() { problems="$problems$*|"; }
        start_mysqld() { # <datadir> <socket> [options]
          d=$1; s=$2; shift 2
          mysqld --initialize-insecure --user=mysql --datadir=$d > $d.init.log 2>&1 || { tail -n 5 $d.init.log; fail "mysqld --initialize failed"; }
          mysqld --user=mysql --datadir=$d --socket=$s --skip-networking --mysqlx=OFF --pid-file=$d.pid --log-error=$d.err "$@" &
          for i in $(seq 1 120); do mysqladmin -S $s -uroot ping > /dev/null 2>&1 && return 0; sleep 0.5; done
          tail -n 5 $d.err; fail "mysqld did not start: $(grep ERROR $d.err | grep -v Aborting | head -n 1 | sed 's/.*\] //')"
        }

        if [ "$MODE" = demo ]; then
          # A source with GTIDs and binlogs, a dump, more writes over three binlog files, then the accident.
          start_mysqld $W/src $W/src.sock --server-id=1 --log-bin=$W/srcbin/binlog --gtid-mode=ON --enforce-gtid-consistency=ON
          SRC="mysql -S $W/src.sock -uroot"
          $SRC -e "CREATE DATABASE shop; CREATE TABLE shop.orders (id INT PRIMARY KEY, total DECIMAL(10,2)); INSERT INTO shop.orders WITH RECURSIVE s(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM s WHERE i < 1000) SELECT i, i * 1.5 FROM s"
          mysqldump -S $W/src.sock -uroot --single-transaction --source-data=2 --set-gtid-purged=ON --databases shop 2> /dev/null | gzip > dump.sql.gz
          for b in 1001 1501; do
            $SRC -e "INSERT INTO shop.orders WITH RECURSIVE s(i) AS (SELECT $b UNION ALL SELECT i + 1 FROM s WHERE i < $b + 499) SELECT i, i * 1.5 FROM s; FLUSH BINARY LOGS"
          done
          $SRC -e "DROP TABLE shop.orders; FLUSH BINARY LOGS"
          cp $W/srcbin/binlog.0* $W/archive/
          mysqladmin -S $W/src.sock -uroot shutdown
          [ -n "$STOP_RE" ] || STOP_RE="DROP TABLE"
          CHECKS="SELECT count(*) = 2000 AS ok, CONCAT(count(*), ' orders recovered (2000 before the accidental DROP TABLE)') AS detail FROM shop.orders
        $CHECKS"
          echo "demo: 1000 orders dumped, 1000 more written across 3 binlogs, then DROP TABLE shop.orders in the same second"
        else
          find . -path ./dump.sql.gz -prune -o -type f -regex '.*\.[0-9][0-9][0-9][0-9][0-9][0-9]$' -exec cp {} $W/archive/ \;
          [ -s dump.sql.gz ] || fail "the dump $DUMP_KEY is empty"
        fi

        # The dump's binlog coordinates: where the replay must start.
        head=$(zcat dump.sql.gz 2> $W/zerr | head -n 60) || true
        [ -n "$head" ] || fail "$DUMP_KEY is not a readable gzip file: $(cat $W/zerr)"
        coord=$(echo "$head" | grep -oE "(SOURCE|MASTER)_LOG_FILE='[^']+', *(SOURCE|MASTER)_LOG_POS=[0-9]+" | head -n 1)
        [ -n "$coord" ] || fail "$DUMP_KEY has no binlog coordinates: make the dump with --source-data=2 (or --master-data=2 before MySQL 8.0.26)"
        first=$(echo "$coord" | cut -d"'" -f2); pos=$(echo "$coord" | grep -oE '[0-9]+$')
        gtid=OFF; echo "$head" | grep -q 'GTID_PURGED' && gtid=ON
        zcat dump.sql.gz | tail -n 1 | grep -q 'Dump completed' || fail "$DUMP_KEY ends without '-- Dump completed': the dump was cut short"

        # The chain: every file from the dump's file on, no number missing, every checksum good.
        base=${first%.*}; n0=$(echo "${first##*.}" | sed 's/^0*//')
        last=$(ls $W/archive | grep -E "^$base\.[0-9]{6}$" | sort | tail -n 1)
        [ -n "$last" ] || fail "no archived binlog named $base.* (the dump points at $first)"
        n1=$(echo "${last##*.}" | sed 's/^0*//'); files=""; gap=""
        for n in $(seq $n0 $n1); do
          f=$(printf "%s.%06d" "$base" "$n")
          if [ ! -f $W/archive/$f ]; then gap="$f"; break; fi
          mysqlbinlog --verify-binlog-checksum $W/archive/$f > /dev/null 2> $W/crc || fail "$f is corrupt: $(head -n 1 $W/crc)"
          files="$files $W/archive/$f"
        done
        [ -n "$files" ] || fail "$first, the binlog the dump starts from, is not in the archive"
        lastfile=$(echo $files | awk '{print $NF}')
        ts() { mysqlbinlog "$@" 2> /dev/null | grep -oE '^#[0-9]{6} +[0-9]{1,2}:[0-9]{2}:[0-9]{2}' | tail -n 1 | sed -E 's/^#([0-9]{2})([0-9]{2})([0-9]{2}) +/20\1-\2-\3 /'; }
        end_time=$(ts $lastfile)
        [ -z "$gap" ] || problem "$gap is missing from the archive: nothing after $(basename $lastfile) ($end_time) can be replayed"
        if [ "$MODE" = s3 ]; then
          lag=$(( ($(date +%s) - $(date -d "$end_time" +%s)) / 60 ))
          [ "$lag" -le "$LAG_MIN" ] || problem "the newest archived binlog event is from $end_time UTC, $lag min ago, over the $LAG_MIN min limit: the binlog archiver has stopped"
        fi
        [ -z "$TARGET_TIME" ] || [ "$(date -d "$TARGET_TIME" +%s)" -le "$(date -d "$end_time" +%s)" ] || problem "the target $TARGET_TIME is after the last archived event ($end_time): whatever was written in between is not in the archive"

        # An exact stop: the start of the first transaction whose statement matches STOP_RE.
        stop_pos=""; stop_dt=""; stop_note=""
        if [ -n "$STOP_RE" ]; then
          for f in $files; do
            sp=""; [ "$f" = "$W/archive/$first" ] && sp="--start-position=$pos"
            at=$(mysqlbinlog -v --base64-output=DECODE-ROWS $sp $f 2> /dev/null | awk -v re="$STOP_RE" '
              /^# at [0-9]+/ { at = $3 }
              index($0, "SET @@SESSION.GTID_NEXT=") == 1 && !/AUTOMATIC/ { txn = at }
              (!/^#/ || /^### /) && tolower($0) ~ tolower(re) { print (txn != "" ? txn : at); exit }')
            if [ -n "$at" ]; then stopfile=$f; break; fi
          done
          [ -n "${at:-}" ] || fail "no statement matching /$STOP_RE/ in the binlogs after the dump"
          files=$(echo $files | awk -v s="$stopfile" '{ for (i = 1; i <= NF; i++) { printf "%s ", $i; if ($i == s) exit } }')
          stop_pos="$at"; stop_note="right before /$STOP_RE/ ($(basename $stopfile) position $at)"
        elif [ -n "$TARGET_TIME" ]; then
          stop_dt="$TARGET_TIME"; stop_note="to $TARGET_TIME UTC"
        else
          stop_note="to the end of the archive ($end_time UTC)"
        fi

        # Restore: the scratch server has no binlog of its own, and the source's GTID mode so GTID_PURGED and the
        # binlogs' GTID_NEXT apply. All files go through one mysqlbinlog call, as temporary tables and
        # multi-file transactions need.
        started=$(date +%s)
        start_mysqld $W/r $W/r.sock --server-id=2 --skip-log-bin --gtid-mode=$gtid --enforce-gtid-consistency=$gtid
        R="mysql -S $W/r.sock -uroot"
        zcat dump.sql.gz | $R 2> $W/load.err || fail "loading $DUMP_KEY failed: $(head -n 1 $W/load.err)"
        dump_s=$(( $(date +%s) - started ))
        set -- --start-position=$pos
        [ -z "$stop_pos" ] || set -- "$@" --stop-position=$stop_pos
        [ -z "$stop_dt" ] || set -- "$@" "--stop-datetime=$stop_dt"
        mysqlbinlog "$@" $files 2> $W/mb.err | $R 2> $W/replay.err || fail "the binlog replay failed: $(head -n 1 $W/replay.err)"
        ! grep -qiE 'error|could not' $W/mb.err || fail "mysqlbinlog could not read the binlogs: $(grep -m1 -iE 'error|could not' $W/mb.err)"
        restore_s=$(( $(date +%s) - started ))
        replayed_last=$(basename $(echo $files | awk '{print $NF}'))

        results=""; failed_checks=0; total_checks=0
        printf '%s\n' "$CHECKS" | grep -v '^ *$' > $W/checks.sql || true
        while IFS= read -r q; do
          total_checks=$((total_checks + 1))
          row=$($R -N -B -e "$q" 2>&1 | head -n 1) || row="0	check could not run: $row"
          case "$row" in ERROR*) row="0	check could not run: $row";; esac
          ok=$(printf '%s' "$row" | cut -f1); detail=$(printf '%s' "$row" | cut -f2- | tr -d '"\\')
          [ "$ok" = 1 ] || { failed_checks=$((failed_checks + 1)); problem "check failed: $detail"; }
          results="$results$([ "$ok" = 1 ] && echo PASS || echo FAIL): $detail; "
        done < $W/checks.sql
        mysqladmin -S $W/r.sock -uroot shutdown > /dev/null 2>&1 || true

        echo "::$(printf '{"outputs":{"ok":%s,"dump":"%s","binlogs":%s,"first_binlog":"%s","replayed_to":"%s","last_binlog":"%s","archive_end":"%s","gtid_mode":"%s","stopped":"%s","dump_seconds":%s,"restore_seconds":%s,"checks":"%s","problems":"%s"}}' \
          "$([ -z "$problems" ] && echo true || echo false)" "$DUMP_KEY" "$(echo $files | wc -w)" "$first" "$replayed_last" "$(basename $lastfile)" "$end_time" "$gtid" \
          "$stop_note" "$dump_s" "$restore_s" "${results%; }" "$(echo "${problems%|}" | tr -d '"\\')")::"

  - id: verdict
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.restore.vars.ok and outputs.restore.vars.restore_seconds
      <= inputs.rto_minutes * 60 }}"
    then:
      - id: remember_good_restore
        type: io.kestra.plugin.core.kv.Set
        key: "{{ vars.last_good_key }}"
        kvType: JSON
        value: "{{ {'at': execution.startDate, 'dump': outputs.restore.vars.dump,
          'archive_end': outputs.restore.vars.archive_end} | toJson }}"
      - id: report_pass
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          MySQL point-in-time restore drill PASSED ({{ inputs.mode }}): `{{ outputs.restore.vars.dump }}` plus {{ outputs.restore.vars.binlogs }} binlog(s) {{ outputs.restore.vars.first_binlog }} to {{ outputs.restore.vars.replayed_to }} (GTID {{ outputs.restore.vars.gtid_mode }}; the archive ends at {{ outputs.restore.vars.last_binlog }}, {{ outputs.restore.vars.archive_end }} UTC), replayed {{ outputs.restore.vars.stopped }}, in {{ outputs.restore.vars.restore_seconds }} s (dump {{ outputs.restore.vars.dump_seconds }} s; RTO {{ inputs.rto_minutes }} min).
          Checks: {{ outputs.restore.vars.checks }}
    else:
      - id: drill_failed
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "DRILL FAILED: {% if outputs.restore.vars.problems != '' %}{{
          outputs.restore.vars.problems | replace({'|': '; '}) }}{% else %}the
          restore took {{ outputs.restore.vars.restore_seconds }} s, over the {{
          inputs.rto_minutes }} min RTO{% endif %}"

errors:
  - id: unless_deadline
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (labels.drill ?? '') != 'deadline_exceeded' }}"
    then:
      - id: alert_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          MySQL point-in-time restore drill FAILED ({{ inputs.mode }}) in execution {{ execution.id }}{% if errorLogs() | length > 0 %} at {{ errorLogs() | jq('([.[] | select(.message | test("DRILL FAILED"))] + .)[0] | "`" + .taskId + "`: " + (.message | sub("^.*DRILL FAILED: "; "") | split("\n")[0])') | first }}{% endif %}
          Last good drill: {{ (kv(vars.last_good_key, errorOnMissing=false) ?? {}).at ?? 'never' }}

afterExecution:
  - id: deadline_alert
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (labels.drill ?? '') == 'deadline_exceeded' }}"
    then:
      - id: alert_deadline
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: "MySQL point-in-time restore drill did not finish within 1 hour
          (execution {{ execution.id }}) and was stopped by its SLA. A download
          or the replay hung; the restore is not proven."

outputs:
  - id: result
    type: JSON
    value: "{{ outputs.restore.vars ?? {} }}"