id: mysql-binlog-retention-safety-guard
namespace: company.team
description: |
Decide which MySQL binary logs are safe to delete, using every replica's own read position
instead of a retention clock. Reads SHOW BINARY LOGS, SHOW REPLICAS and the retention settings
on the primary, then asks each listed replica for SHOW REPLICA STATUS. The purge boundary is the
oldest file any replica still needs or the file the last backup needs, whichever is older, minus
a margin of files. Alerts Slack. Behind auto_remediate, dry_run and an approval it runs
PURGE BINARY LOGS TO that file, then re-reads the server and fails if a needed log went missing or
the footprint did not drop. It refuses to purge at all while a replica is stalled or unaccounted
for, because that replica is the reason the logs are piling up.
inputs:
- id: primary_url
type: STRING
displayName: Primary JDBC URL
defaults: "jdbc:mysql://localhost:3306/mysql"
- id: replicas
type: JSON
displayName: Replicas of this primary
description: >-
Every replica that reads this primary, as a JSON array of objects with a
name and a JDBC url. The guard asks each replica for its own position,
because the primary cannot see the read position of a replica whose IO
thread is stopped. List at least one.
defaults: [ { "name": "replica-1", "url": "jdbc:mysql://replica-1:3306/mysql" } ]
- id: backup_required_binlog
type: STRING
displayName: Binary log the last backup needs
description: >-
The binary log file your last full backup recorded as the start of its
recovery stream. The boundary never goes past it. Leave empty when point
in time recovery is not in use.
defaults: ""
- id: pitr_window_hours
type: INT
displayName: Point in time recovery window (hours)
description: The recovery window you promise. A shorter
binlog_expire_logs_seconds cannot keep it.
defaults: 72
- id: binlog_budget_mb
type: INT
displayName: Binary log budget (MiB)
description: Total size of the binary logs that triggers a reclaim.
defaults: 20480
- id: datadir_capacity_mb
type: INT
displayName: Data directory capacity (MiB)
description: >-
MySQL reports no filesystem free space over SQL, so give the volume size
here to get a share of disk check as well as the absolute budget. 0 turns
it off.
defaults: 0
- id: datadir_pct_threshold
type: INT
displayName: Share of the volume the logs may use (percent)
defaults: 60
- id: safety_margin_files
type: INT
displayName: Files kept below the oldest needed log
description: Extra files left in place under the computed boundary, for a
replica that reconnects behind its last reported position.
defaults: 2
- id: lag_files_threshold
type: INT
displayName: Replica lag that blocks a purge (files)
defaults: 20
- id: auto_remediate
type: BOOL
displayName: Auto remediate
defaults: false
- id: dry_run
type: BOOL
displayName: Dry run
description: Must be false, together with auto_remediate, for PURGE BINARY LOGS to run.
defaults: true
variables:
logs_values: >-
{{ outputs.read_binlogs.rows | jq('to_entries | map(select(.value.Log_name |
test("^[A-Za-z0-9._-]+$")) | "SELECT \(.key + 1) AS ord,
\u0027\(.value.Log_name)\u0027 AS log_name, \(.value.File_size) AS
file_size") | if length == 0 then "SELECT 0 AS ord, \u0027\u0027 AS
log_name, 0 AS file_size FROM DUAL WHERE 1 = 0" else join(" UNION ALL ")
end') | first }}
reg_values: >-
{{ outputs.read_registered.rows | jq('map("SELECT \(.Server_Id) AS
server_id, \u0027\(.Host | gsub("\u0027"; "\u0027\u0027"))\u0027 AS host,
\(.Port) AS port") | if length == 0 then "SELECT 0 AS server_id,
\u0027\u0027 AS host, 0 AS port FROM DUAL WHERE 1 = 0" else join(" UNION ALL
") end') | first }}
reps_values: >-
{{ (outputs.probe_replicas.outputs ?? []) | jq('map(.outputs) | map("SELECT
\u0027\(.rname | gsub("\u0027"; "\u0027\u0027"))\u0027 AS rname,
\((.server_id | tonumber?) // 0) AS server_id, \u0027\(if .io_running ==
"Yes" then "Yes" else "No" end)\u0027 AS io_running, \u0027\(if .sql_running
== "Yes" then "Yes" else "No" end)\u0027 AS sql_running, \u0027\(if
(.source_file | test("^[A-Za-z0-9._-]+$")) then .source_file else "UNKNOWN"
end)\u0027 AS source_file, \u0027\(if (.relay_file |
test("^[A-Za-z0-9._-]+$")) then .relay_file else "UNKNOWN" end)\u0027 AS
relay_file, \((.seconds_behind | tonumber?) // -1) AS seconds_behind") | if
length == 0 then "SELECT \u0027\u0027 AS rname, 0 AS server_id,
\u0027No\u0027 AS io_running, \u0027No\u0027 AS sql_running,
\u0027UNKNOWN\u0027 AS source_file, \u0027UNKNOWN\u0027 AS relay_file, -1 AS
seconds_behind FROM DUAL WHERE 1 = 0" else join(" UNION ALL ") end') | first
}}
after_logs_values: >-
{{ outputs.reverify_binlogs.rows | jq('to_entries |
map(select(.value.Log_name | test("^[A-Za-z0-9._-]+$")) | "SELECT \(.key +
1) AS ord, \u0027\(.value.Log_name)\u0027 AS log_name, \(.value.File_size)
AS file_size") | if length == 0 then "SELECT 0 AS ord, \u0027\u0027 AS
log_name, 0 AS file_size FROM DUAL WHERE 1 = 0" else join(" UNION ALL ")
end') | first }}
after_reps_values: >-
{{ (outputs.reverify_replicas.outputs ?? []) | jq('map(.outputs) |
map("SELECT \u0027\(.rname | gsub("\u0027"; "\u0027\u0027"))\u0027 AS rname,
\u0027\(if .io_running == "Yes" then "Yes" else "No" end)\u0027 AS
io_running, \u0027\(if .sql_running == "Yes" then "Yes" else "No" end)\u0027
AS sql_running, \u0027\(if (.source_file | test("^[A-Za-z0-9._-]+$")) then
.source_file else "UNKNOWN" end)\u0027 AS source_file, \u0027\(if
(.relay_file | test("^[A-Za-z0-9._-]+$")) then .relay_file else "UNKNOWN"
end)\u0027 AS relay_file") | if length == 0 then "SELECT \u0027\u0027 AS
rname, \u0027No\u0027 AS io_running, \u0027No\u0027 AS sql_running,
\u0027UNKNOWN\u0027 AS source_file, \u0027UNKNOWN\u0027 AS relay_file FROM
DUAL WHERE 1 = 0" else join(" UNION ALL ") end') | first }}
plan_sql: |
WITH logs AS (
{{ render(vars.logs_values) }}
), reps AS (
{{ render(vars.reps_values) }}
), reg AS (
{{ render(vars.reg_values) }}
), cfg AS (
SELECT
CAST(COALESCE(MAX(CASE WHEN VARIABLE_NAME = 'binlog_expire_logs_seconds' THEN VARIABLE_VALUE END), '0') AS SIGNED) AS expire_seconds,
CAST(COALESCE(MAX(CASE WHEN VARIABLE_NAME = 'expire_logs_days' THEN VARIABLE_VALUE END), '0') AS SIGNED) AS expire_days,
CAST(COALESCE(MAX(CASE WHEN VARIABLE_NAME = 'max_binlog_size' THEN VARIABLE_VALUE END), '0') AS SIGNED) AS max_binlog_size,
COALESCE(MAX(CASE WHEN VARIABLE_NAME = 'log_bin' THEN VARIABLE_VALUE END), 'OFF') AS log_bin,
COALESCE(MAX(CASE WHEN VARIABLE_NAME = 'binlog_format' THEN VARIABLE_VALUE END), 'UNKNOWN') AS binlog_format,
COALESCE(MAX(CASE WHEN VARIABLE_NAME = 'datadir' THEN VARIABLE_VALUE END), 'UNKNOWN') AS datadir
FROM performance_schema.global_variables
WHERE VARIABLE_NAME IN ('binlog_expire_logs_seconds', 'expire_logs_days', 'max_binlog_size', 'log_bin', 'binlog_format', 'datadir')
), bk AS (
SELECT '{{ inputs.backup_required_binlog | replace({"'": "''"}) }}' AS backup_file
), need AS (
SELECT r.rname, r.server_id, r.io_running, r.sql_running, r.seconds_behind, r.source_file, r.relay_file,
CASE
WHEN r.source_file = 'UNKNOWN' OR r.relay_file = 'UNKNOWN' THEN 'NOT_REPLICATING'
WHEN ls.ord IS NULL OR lr.ord IS NULL THEN 'NEEDS_PURGED_FILE'
WHEN r.io_running <> 'Yes' OR r.sql_running <> 'Yes' THEN 'THREAD_STOPPED'
ELSE 'HEALTHY'
END AS rep_state,
LEAST(COALESCE(ls.ord, 999999999), COALESCE(lr.ord, 999999999)) AS need_ord
FROM reps r
LEFT JOIN logs ls ON ls.log_name = r.source_file
LEFT JOIN logs lr ON lr.log_name = r.relay_file
), base AS (
SELECT
(SELECT COUNT(*) FROM logs) AS file_count,
(SELECT CAST(COALESCE(SUM(file_size), 0) AS SIGNED) FROM logs) AS total_bytes,
(SELECT MAX(ord) FROM logs) AS active_ord,
(SELECT log_name FROM logs ORDER BY ord LIMIT 1) AS oldest_file,
(SELECT log_name FROM logs ORDER BY ord DESC LIMIT 1) AS active_file,
(SELECT COUNT(*) FROM reps) AS replica_count,
(SELECT COUNT(*) FROM reg) AS registered_count,
(SELECT COUNT(*) FROM need WHERE rep_state = 'NEEDS_PURGED_FILE') AS blocked_count,
(SELECT COUNT(*) FROM need WHERE rep_state = 'THREAD_STOPPED') AS stopped_count,
(SELECT COUNT(*) FROM need WHERE rep_state = 'NOT_REPLICATING') AS notrep_count,
(SELECT MIN(need_ord) FROM need WHERE rep_state IN ('HEALTHY', 'THREAD_STOPPED')) AS replica_need_ord,
(SELECT COUNT(*) FROM reg g WHERE NOT EXISTS (SELECT 1 FROM reps r WHERE r.server_id = g.server_id)) AS undeclared_count,
(SELECT COALESCE(GROUP_CONCAT(CONCAT(g.host, ':', g.port, ' server_id=', g.server_id) ORDER BY g.server_id SEPARATOR ', '), 'none') FROM reg g WHERE NOT EXISTS (SELECT 1 FROM reps r WHERE r.server_id = g.server_id)) AS undeclared_hosts,
(SELECT COALESCE(GROUP_CONCAT(CONCAT(n.rname, ' ', n.rep_state, ' io=', n.io_running, ' sql=', n.sql_running, ' read=', n.source_file, ' applied=', n.relay_file, ' behind=', n.seconds_behind, 's') ORDER BY n.rname SEPARATOR '; '), 'no replicas listed') FROM need n) AS replica_detail,
(SELECT l.ord FROM logs l, bk WHERE l.log_name = bk.backup_file) AS backup_ord,
(SELECT backup_file FROM bk) AS backup_file,
cfg.expire_seconds, cfg.expire_days, cfg.max_binlog_size, cfg.log_bin, cfg.binlog_format, cfg.datadir
FROM cfg
), calc AS (
SELECT b.*,
(b.backup_file <> '' AND b.backup_ord IS NULL) AS backup_missing,
(b.backup_file <> '' AND b.backup_file NOT REGEXP '^[A-Za-z0-9._-]+$') AS backup_invalid,
LEAST(COALESCE(b.replica_need_ord, 999999999), COALESCE(b.backup_ord, 999999999), b.active_ord) AS need_ord,
(b.total_bytes > {{ inputs.binlog_budget_mb }} * 1048576
OR ({{ inputs.datadir_capacity_mb }} > 0
AND b.total_bytes * 100 >= {{ inputs.datadir_capacity_mb }} * 1048576 * {{ inputs.datadir_pct_threshold }})) AS disk_at_risk,
GREATEST(b.active_ord - COALESCE(b.replica_need_ord, b.active_ord), 0) AS files_behind
FROM base b
), bound AS (
SELECT c.*, GREATEST(c.need_ord - {{ inputs.safety_margin_files }}, 1) AS purge_to_ord
FROM calc c
), decided AS (
SELECT p.*,
(SELECT log_name FROM logs WHERE ord = p.purge_to_ord) AS purge_to_file,
p.purge_to_ord - 1 AS files_to_purge,
(SELECT CAST(COALESCE(SUM(file_size), 0) AS SIGNED) FROM logs WHERE ord < p.purge_to_ord) AS bytes_to_purge,
(p.log_bin = 'ON'
AND p.blocked_count = 0 AND p.notrep_count = 0 AND p.undeclared_count = 0 AND p.stopped_count = 0
AND p.files_behind < {{ inputs.lag_files_threshold }}
AND NOT p.backup_missing AND NOT p.backup_invalid
AND NOT (p.replica_count = 0 AND p.backup_file = '')) AS safe
FROM bound p
)
SELECT
file_count, total_bytes, oldest_file, active_file, active_ord,
replica_count, registered_count, undeclared_count, undeclared_hosts, replica_detail,
blocked_count, stopped_count, notrep_count, files_behind,
expire_seconds, expire_days, max_binlog_size, log_bin, binlog_format, datadir,
backup_file, COALESCE(backup_ord, 0) AS backup_ord,
need_ord, purge_to_ord, COALESCE(purge_to_file, '') AS purge_to_file, files_to_purge, bytes_to_purge,
IF(disk_at_risk, 'YES', 'NO') AS disk_at_risk,
IF(safe, 'YES', 'NO') AS safe_to_purge,
CASE
WHEN log_bin <> 'ON' THEN 'BINLOG_OFF'
WHEN blocked_count > 0 THEN 'REPLICA_BLOCKED'
WHEN notrep_count > 0 OR undeclared_count > 0 OR backup_invalid THEN 'TOPOLOGY_UNVERIFIED'
WHEN stopped_count > 0 OR files_behind >= {{ inputs.lag_files_threshold }} THEN 'REPLICA_LAGGING_PINS_LOGS'
WHEN backup_missing OR expire_days > 0
OR (expire_seconds > 0 AND expire_seconds < {{ inputs.pitr_window_hours }} * 3600) THEN 'UNSAFE_RETENTION'
WHEN disk_at_risk THEN 'DISK_AT_RISK'
ELSE 'OK'
END AS state,
CASE
WHEN NOT safe THEN 'REFUSE'
WHEN files_to_purge <= 0 THEN 'NOTHING_TO_PURGE'
WHEN NOT disk_at_risk THEN 'NO_DISK_PRESSURE'
ELSE 'PURGE'
END AS purge_decision,
COALESCE(NULLIF(TRIM(CONCAT_WS(' ',
NULLIF(IF(log_bin <> 'ON', 'Binary logging is off on this server.', ''), ''),
NULLIF(IF(replica_count = 0 AND backup_file = '', 'No replicas are listed and no backup_required_binlog is set, so there is no boundary to compute.', ''), ''),
NULLIF(IF(blocked_count > 0, CONCAT(blocked_count, ' replica(s) already need a binary log that is gone, so they need a rebuild rather than a purge.'), ''), ''),
NULLIF(IF(notrep_count > 0, CONCAT(notrep_count, ' listed server(s) returned no replica status, so the topology is unverified.'), ''), ''),
NULLIF(IF(undeclared_count > 0, CONCAT('Replicas connected to this primary are missing from the replicas input: ', undeclared_hosts, '.'), ''), ''),
NULLIF(IF(stopped_count > 0, CONCAT(stopped_count, ' replica(s) have a stopped IO or SQL thread, so their position cannot advance.'), ''), ''),
NULLIF(IF(files_behind >= {{ inputs.lag_files_threshold }}, CONCAT('A replica is ', files_behind, ' files behind the active log, which is what is filling the volume.'), ''), ''),
NULLIF(IF(backup_missing, CONCAT('The last backup needs ', backup_file, ' and it is not on disk any more.'), ''), ''),
NULLIF(IF(backup_invalid, 'backup_required_binlog is not a plain binary log file name.', ''), '')
)), ''), 'none') AS refuse_reason
FROM decided
verify_sql: |
WITH logs AS (
{{ render(vars.after_logs_values) }}
), reps AS (
{{ render(vars.after_reps_values) }}
)
SELECT
(SELECT COUNT(*) FROM logs) AS file_count,
(SELECT CAST(COALESCE(SUM(file_size), 0) AS SIGNED) FROM logs) AS total_bytes,
(SELECT log_name FROM logs ORDER BY ord LIMIT 1) AS oldest_file,
(SELECT '{{ outputs.plan.row.purge_to_file }}') AS boundary_file,
(SELECT COUNT(*) FROM reps r
WHERE r.relay_file NOT IN (SELECT l.log_name FROM logs l)
AND r.relay_file < '{{ outputs.plan.row.purge_to_file }}') AS missing_count,
(SELECT COUNT(*) FROM reps r
WHERE r.source_file NOT IN (SELECT l.log_name FROM logs l)) AS unreadable_count,
(SELECT COUNT(*) FROM reps WHERE io_running <> 'Yes' OR sql_running <> 'Yes') AS stopped_count,
(SELECT COALESCE(GROUP_CONCAT(CONCAT(rname, ' io=', io_running, ' sql=', sql_running, ' read=', source_file, ' applied=', relay_file) ORDER BY rname SEPARATOR '; '), 'none') FROM reps) AS replica_detail
tasks:
- id: read_binlogs
type: io.kestra.plugin.jdbc.mysql.Query
description: Read-only. File names and sizes of every binary log still on the
primary, oldest first.
url: "{{ inputs.primary_url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH
sql: SHOW BINARY LOGS
- id: read_registered
type: io.kestra.plugin.jdbc.mysql.Query
description: Read-only. Replicas with a live connection to this primary, used to
catch one that is not in the replicas input.
url: "{{ inputs.primary_url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH
sql: SHOW REPLICAS
- id: probe_replicas
type: io.kestra.plugin.core.flow.Loop
description: Read-only. One connection per listed replica, for the file it has
read and the file it has applied.
concurrencyLimit: 1
values: "{{ inputs.replicas | toJson }}"
outputs:
- id: rname
type: STRING
value: "{{ fromJson(item.value).name }}"
- id: server_id
type: STRING
value: "{{ outputs.ident.row.server_id }}"
- id: io_running
type: STRING
value: "{{ outputs.status.rows | jq('.[0].Replica_IO_Running // \"UNKNOWN\"') |
first }}"
- id: sql_running
type: STRING
value: "{{ outputs.status.rows | jq('.[0].Replica_SQL_Running // \"UNKNOWN\"') |
first }}"
- id: source_file
type: STRING
value: "{{ outputs.status.rows | jq('.[0].Source_Log_File // \"UNKNOWN\"') |
first }}"
- id: relay_file
type: STRING
value: "{{ outputs.status.rows | jq('.[0].Relay_Source_Log_File // \"UNKNOWN\"')
| first }}"
- id: seconds_behind
type: STRING
value: "{{ outputs.status.rows | jq('.[0].Seconds_Behind_Source // -1') | first
}}"
tasks:
- id: status
type: io.kestra.plugin.jdbc.mysql.Query
url: "{{ fromJson(item.value).url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH
sql: SHOW REPLICA STATUS
- id: ident
type: io.kestra.plugin.jdbc.mysql.Query
url: "{{ fromJson(item.value).url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH_ONE
sql: SELECT @@server_id AS server_id
- id: plan
type: io.kestra.plugin.jdbc.mysql.Query
description: >-
Read-only. Joins the binary log list, the replica positions and the
retention settings in one statement, then returns the state, the safe
boundary file and the reason when it refuses.
url: "{{ inputs.primary_url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "{{ render(vars.plan_sql) }}"
- id: report
type: io.kestra.plugin.core.log.Log
message: >-
MySQL binlog retention on {{ inputs.primary_url }}: state {{
outputs.plan.row.state }}, {{ outputs.plan.row.file_count }} files holding
{{ outputs.plan.row.total_bytes }} bytes from {{
outputs.plan.row.oldest_file }} to {{ outputs.plan.row.active_file }} in
{{ outputs.plan.row.datadir }}. Settings: binlog_expire_logs_seconds={{
outputs.plan.row.expire_seconds }}, expire_logs_days={{
outputs.plan.row.expire_days }}, max_binlog_size={{
outputs.plan.row.max_binlog_size }}, log_bin={{ outputs.plan.row.log_bin
}}, binlog_format={{ outputs.plan.row.binlog_format }}. Replicas listed {{
outputs.plan.row.replica_count }}, connected {{
outputs.plan.row.registered_count }}, unaccounted {{
outputs.plan.row.undeclared_count }} ({{ outputs.plan.row.undeclared_hosts
}}). Positions: {{ outputs.plan.row.replica_detail }}. Oldest needed file
is number {{ outputs.plan.row.need_ord }}, replicas are {{
outputs.plan.row.files_behind }} files behind the active log, backup needs
"{{ outputs.plan.row.backup_file }}" (number {{
outputs.plan.row.backup_ord }}). Decision {{
outputs.plan.row.purge_decision }}, safe_to_purge={{
outputs.plan.row.safe_to_purge }}, disk_at_risk={{
outputs.plan.row.disk_at_risk }}. It would run PURGE BINARY LOGS TO "{{
outputs.plan.row.purge_to_file }}", freeing {{
outputs.plan.row.files_to_purge }} files and {{
outputs.plan.row.bytes_to_purge }} bytes. Why not: {{
outputs.plan.row.refuse_reason }}.
- id: notify
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.plan.row.state != 'OK' }}"
then:
- id: alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': 'MySQL binlog retention ' ~ outputs.plan.row.state ~ ' on ' ~ inputs.primary_url ~ ': ' ~ outputs.plan.row.file_count ~ ' binary logs holding ' ~ outputs.plan.row.total_bytes ~ ' bytes. Oldest needed file number ' ~ outputs.plan.row.need_ord ~ ', replicas ' ~ outputs.plan.row.files_behind ~ ' files behind. Positions: ' ~ outputs.plan.row.replica_detail ~ '. Decision ' ~ outputs.plan.row.purge_decision ~ ' with boundary ' ~ outputs.plan.row.purge_to_file ~ ' (' ~ outputs.plan.row.files_to_purge ~ ' files, ' ~ outputs.plan.row.bytes_to_purge ~ ' bytes). Why not: ' ~ outputs.plan.row.refuse_reason ~ '.'} | toJson }}
else:
- id: clean
type: io.kestra.plugin.core.log.Log
message: >-
Binary logs are within budget and every listed replica is caught up.
Nothing to report on {{ inputs.primary_url }}.
- id: act
type: io.kestra.plugin.core.flow.Switch
value: "{{ inputs.auto_remediate and not inputs.dry_run ?
outputs.plan.row.purge_decision : 'REPORT_ONLY' }}"
cases:
PURGE:
- id: approve
type: io.kestra.plugin.core.flow.Pause
description: >-
PURGE BINARY LOGS cannot be undone, so a person confirms the
boundary file by name. No answer in four hours resumes with an empty
confirmation, which purges nothing.
pauseDuration: PT4H
behavior: RESUME
onResume:
- id: confirm_purge_to
type: STRING
required: false
description: Type the boundary file name from the report to confirm. Anything
else purges nothing.
- id: confirm
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.approve.onResume.confirm_purge_to ?? '') ==
outputs.plan.row.purge_to_file }}"
then:
- id: purge
type: io.kestra.plugin.jdbc.mysql.Query
description: Deletes every binary log before the boundary file. The boundary
itself stays.
url: "{{ inputs.primary_url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: NONE
sql: "PURGE BINARY LOGS TO '{{ outputs.plan.row.purge_to_file }}'"
- id: reverify_binlogs
type: io.kestra.plugin.jdbc.mysql.Query
url: "{{ inputs.primary_url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH
sql: SHOW BINARY LOGS
- id: reverify_replicas
type: io.kestra.plugin.core.flow.Loop
description: Asks every listed replica again for the file it needs now.
concurrencyLimit: 1
values: "{{ inputs.replicas | toJson }}"
outputs:
- id: rname
type: STRING
value: "{{ fromJson(item.value).name }}"
- id: io_running
type: STRING
value: "{{ outputs.status_after.rows | jq('.[0].Replica_IO_Running //
\"UNKNOWN\"') | first }}"
- id: sql_running
type: STRING
value: "{{ outputs.status_after.rows | jq('.[0].Replica_SQL_Running //
\"UNKNOWN\"') | first }}"
- id: source_file
type: STRING
value: "{{ outputs.status_after.rows | jq('.[0].Source_Log_File // \"UNKNOWN\"')
| first }}"
- id: relay_file
type: STRING
value: "{{ outputs.status_after.rows | jq('.[0].Relay_Source_Log_File //
\"UNKNOWN\"') | first }}"
tasks:
- id: status_after
type: io.kestra.plugin.jdbc.mysql.Query
url: "{{ fromJson(item.value).url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH
sql: SHOW REPLICA STATUS
- id: verify
type: io.kestra.plugin.jdbc.mysql.Query
description: Counts the logs that are left and any replica whose needed file is
gone.
url: "{{ inputs.primary_url }}"
username: "{{ secret('MYSQL_USERNAME') }}"
password: "{{ secret('MYSQL_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "{{ render(vars.verify_sql) }}"
- id: verify_gate
type: io.kestra.plugin.core.flow.If
condition: >-
{{ outputs.verify.row.missing_count > 0 or
outputs.verify.row.unreadable_count > 0
or outputs.verify.row.stopped_count > 0
or outputs.verify.row.total_bytes >= outputs.plan.row.total_bytes }}
then:
- id: purge_broke_replication
type: io.kestra.plugin.core.execution.Fail
errorMessage: >-
Purge check failed on {{ inputs.primary_url }}. {{
outputs.verify.row.missing_count }} replica(s) have an
applied position that the purge went past, {{
outputs.verify.row.unreadable_count }} can no longer read
the log they are on, {{ outputs.verify.row.stopped_count }}
have a stopped thread and the footprint went from {{
outputs.plan.row.total_bytes }} to {{
outputs.verify.row.total_bytes }} bytes across {{
outputs.verify.row.file_count }} files. Positions now: {{
outputs.verify.row.replica_detail }}.
else:
- id: purged
type: io.kestra.plugin.core.log.Log
message: >-
Purged to {{ outputs.plan.row.purge_to_file }} on {{
inputs.primary_url }}. Files {{ outputs.plan.row.file_count
}} to {{ outputs.verify.row.file_count }}, bytes {{
outputs.plan.row.total_bytes }} to {{
outputs.verify.row.total_bytes }}, oldest log now {{
outputs.verify.row.oldest_file }}. Every listed replica
still has the file it needs: {{
outputs.verify.row.replica_detail }}.
else:
- id: not_confirmed
type: io.kestra.plugin.core.log.Log
message: >-
Approval did not confirm {{ outputs.plan.row.purge_to_file }},
so nothing was purged on {{ inputs.primary_url }}. {{
outputs.plan.row.file_count }} files still hold {{
outputs.plan.row.total_bytes }} bytes.
REFUSE:
- id: refused
type: io.kestra.plugin.core.log.Log
message: >-
Refusing to purge on {{ inputs.primary_url }}. {{
outputs.plan.row.refuse_reason }} State {{ outputs.plan.row.state
}}, positions {{ outputs.plan.row.replica_detail }}. The safe
boundary would have been {{ outputs.plan.row.purge_to_file }}.
- id: refused_alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': 'MySQL binlog purge REFUSED on ' ~ inputs.primary_url ~ '. ' ~ outputs.plan.row.refuse_reason ~ ' State ' ~ outputs.plan.row.state ~ ', ' ~ outputs.plan.row.file_count ~ ' files holding ' ~ outputs.plan.row.total_bytes ~ ' bytes. Positions: ' ~ outputs.plan.row.replica_detail ~ '. Fix replication first, a purge here would break it.'} | toJson }}
NOTHING_TO_PURGE:
- id: nothing_to_purge
type: io.kestra.plugin.core.log.Log
message: >-
Nothing below the safe boundary on {{ inputs.primary_url }}. The
oldest needed file is number {{ outputs.plan.row.need_ord }} and the
margin of {{ inputs.safety_margin_files }} files covers everything
older.
NO_DISK_PRESSURE:
- id: no_disk_pressure
type: io.kestra.plugin.core.log.Log
message: >-
{{ outputs.plan.row.files_to_purge }} files holding {{
outputs.plan.row.bytes_to_purge }} bytes could be purged safely, but
the total of {{ outputs.plan.row.total_bytes }} bytes is inside the
{{ inputs.binlog_budget_mb }} MiB budget, so they stay for recovery.
defaults:
- id: report_only
type: io.kestra.plugin.core.log.Log
message: >-
Read-only run on {{ inputs.primary_url }} (auto_remediate={{
inputs.auto_remediate }}, dry_run={{ inputs.dry_run }}). Nothing was
deleted. With both flags set to act it would {{
outputs.plan.row.purge_decision }} at {{
outputs.plan.row.purge_to_file }}, freeing {{
outputs.plan.row.files_to_purge }} files and {{
outputs.plan.row.bytes_to_purge }} bytes.
errors:
- id: alert_flow_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': 'MySQL binlog retention safety guard failed on ' ~ inputs.primary_url ~ ' (execution ' ~ execution.id ~ '). A primary or replica query did not answer, so no binary log was purged. Check the logs.'} | toJson }}
triggers:
- id: daily
type: io.kestra.plugin.core.trigger.Schedule
description: Shipped disabled and always in safe mode.
cron: "0 3 * * *"
disabled: true
inputs:
auto_remediate: false
dry_run: true
