diff --git a/CHANGELOG.md b/CHANGELOG.md index 26a8e14..1cbc2e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,54 @@ expose. ### Added +- **A durable run log in ClickHouse**, `_log`, written + beside the auxiliary table and **never truncated**. + + Why: pod logs are not a record. The kubelet rotates container output (10Mi + over 5 files by default), so `kubectl logs` cannot return the beginning of a + long run, and `ttlSecondsAfterFinished` deletes the Job and its pods along + with everything they printed. A cleanup that reclaimed terabytes left no + evidence of what it did once that window closed — the reported symptom was + simply "the job does not have all the log output". + + ClickHouse is the sink rather than a volume or the bucket: the connection, + credentials and grants already exist, a `readOnlyRootFilesystem` container + cannot write a file, and an object written under `S3PATH` would be listed by + the *next* collect, found absent from `system.remote_data_paths`, and become + a deletion candidate — s3gc would garbage-collect its own logs. + + Rows are self-describing evidence, not just text: `run_id`, `phase`, `event`, + `message`, running `objects`/`bytes`, and the scope the run was pointed at + (bucket, prefix, disk, cluster, dry-run, ClickHouse host, container + hostname). Events are phase start, throttled collect progress, per-sample + start, **per-delete-batch checkpoint**, finish with attempt totals, warnings + and errors. `S3GC_RUNID` defaults to a generated timestamped id and the Job + template passes `JOB_NAME`, so a row traces back to the Job that wrote it. + + Three constraints, each with a test, because this is bookkeeping attached to + an irreversible operation: + + 1. **Writes go on `ch_writer`, never `ch_client`.** `do_use()` holds + `ch_client`'s session for the whole anti-join stream, and a second query on + a held session is `SESSION_IS_LOCKED` (373) — the 0.6.0 defect, which + would now fire mid-delete at the worst possible moment. + 2. **A logging failure never fails the run.** One failure disables the run log + for the remainder of the process rather than retrying every batch, and a + missing `CREATE TABLE` grant degrades to stdout only with one warning. + 3. **Messages are redacted** through the existing `LogFormatter._filter` + before insert, so a credential cannot reach a table that outlives the run. + + Opt out with `--runlog false` / `S3GC_RUNLOG_FLAG=false`, or `RUNLOG=false` + in the renderer. Default on: durability is the point. + +- **Collect now reports progress at INFO**, throttled to every 100 000 objects. + Per-batch progress was logged at DEBUG only, so a multi-hour collect over + millions of objects emitted about four lines at `--verbose` — while `--debug` + emits one line *per object*, which on a large bucket exceeds the kubelet's + rotation limit and destroys the beginning of its own output. The operator's + real choice was "almost nothing" or "too much to retrieve". + + - **Regression tests pinning the deletion scope itself.** The suite already proved that a candidate list is deleted, batched and checkpointed correctly, but nothing proved the list contained only orphans: the ClickHouse fake diff --git a/README.md b/README.md index a3dc478..9e426dc 100644 --- a/README.md +++ b/README.md @@ -283,6 +283,31 @@ are in [deploy/kubernetes/README.md](deploy/kubernetes/README.md). A guarded collect, dry-run, and delete in one Job and still requires an explicit delete confirmation. +## Durable run history + +Each run appends structured events to `_log` in +ClickHouse, beside the auxiliary table: phase start, collect progress, +per-sample start, one row per confirmed delete batch, and a closing total, each +carrying the scope the run was pointed at. The table is never truncated. + +This exists because pod logs are ephemeral — the kubelet rotates them and a +deleted Job takes them with it — so a completed cleanup would otherwise leave no +evidence of what it removed. ClickHouse is the sink because the connection and +grants already exist; writing the log into the bucket would make the *next* +collect see it as an orphan and delete it. + +```sql +SELECT event_time, phase, event, objects, bytes, message +FROM _log +WHERE run_id = '' +ORDER BY event_time; +``` + +`--runid` labels the run (Kubernetes Jobs use the Job name automatically); +`--runlog false` turns the table off. It needs the same `CREATE TABLE` grant as +the auxiliary table, and a missing grant degrades to stdout only rather than +failing the run. + ## Testing Run all isolated unit tests: diff --git a/TODO.md b/TODO.md index cfff716..7985fe9 100644 --- a/TODO.md +++ b/TODO.md @@ -21,3 +21,12 @@ forward-looking backlog. - [ ] Quote `--useafter` as a SQL string literal (strict `xfail` in the suite). - [ ] Consider re-checking cluster topology per sample, not once per run, so a replica lost mid-run cannot widen the deletion scope. +- [ ] `print()` output is block-buffered because the image sets no + `PYTHONUNBUFFERED` and stdout is a pipe under Kubernetes. Logger records are + flushed per line, but the bare prints — including the closing `s3gc: OK` — + are lost when `activeDeadlineSeconds` fires and the kubelet sends SIGTERM. + The durable run log now covers the evidence case; this remains a stdout + fidelity gap, and it also makes the `dev-automation` phase markers interleave + wrongly against the shell's unbuffered `echo`. +- [ ] Consider a `TTL` on the run-log table. Volume is a few hundred rows per + run so it is not urgent, but it grows without bound across many cleanups. diff --git a/deploy/kubernetes/README.md b/deploy/kubernetes/README.md index 7e4c73a..92aae40 100644 --- a/deploy/kubernetes/README.md +++ b/deploy/kubernetes/README.md @@ -56,7 +56,7 @@ Create or provision a dedicated ClickHouse user for `s3gc`, then grant it: ```sql GRANT SELECT ON system.* TO s3gc; -- remote_data_paths, one, disks, tables -GRANT SELECT, INSERT, CREATE TABLE ON .* TO s3gc; -- the auxiliary table +GRANT SELECT, INSERT, CREATE TABLE ON .* TO s3gc; -- auxiliary + run-log tables ``` ### Values that vary per cluster, and bite when wrong @@ -136,6 +136,49 @@ Any failed stage stops the Job and later stages do not run; successful delete batches remain checkpointed. Do not use this phase for customer or production work because it removes the manual dry-run approval gate. +## Durable run history + +Pod logs are **not** a record. The kubelet rotates container output, so +`kubectl logs` cannot return the start of a long run, and +`ttlSecondsAfterFinished` deletes the Job and its pods along with everything +they printed. `kubectl logs -f` is for watching, not for evidence. + +So each run also appends to `_log` in +ClickHouse, beside the auxiliary table. That table is **never truncated**, and +`S3GC_RUNID` is set to the Job name, so a row traces back to the Job that wrote +it long after the pod is gone. + +What a delete actually did, on the same replica-pinned `CHHOST`: + +```sql +SELECT event_time, phase, event, objects, bytes, message +FROM ._log +WHERE run_id = '' +ORDER BY event_time; +``` + +How far a *failed* delete got before it died, which is what tells you whether to +start a replacement delete Job: + +```sql +SELECT max(objects) AS deleted, max(bytes) AS reclaimed +FROM ._log +WHERE run_id = '' AND event = 'checkpoint'; +``` + +Every phase of a cleanup, newest first: + +```sql +SELECT run_id, min(event_time) AS started, max(event_time) AS ended, + anyIf(message, event = 'error') AS error +FROM ._log +GROUP BY run_id ORDER BY started DESC; +``` + +Set `RUNLOG=false` to opt out. A missing `CREATE TABLE` grant degrades to +stdout only with one warning rather than failing the run, and a run-log write +that fails mid-delete disables the log instead of stopping the deletion. + ## Safety - Delete checks the local cluster macro and expected replica count before S3 diff --git a/deploy/kubernetes/example.env b/deploy/kubernetes/example.env index e435f77..fddff14 100644 --- a/deploy/kubernetes/example.env +++ b/deploy/kubernetes/example.env @@ -48,3 +48,9 @@ TTL_SECONDS_AFTER_FINISHED=604800 MEMORY_REQUEST=1Gi MEMORY_LIMIT=4Gi VERBOSE=true +# Record durable run history in _log in ClickHouse. +# Pod logs are rotated by the kubelet and deleted with the Job, so this table is +# the only evidence of what a cleanup did once TTL_SECONDS_AFTER_FINISHED +# expires. Needs the same CREATE TABLE grant as the auxiliary table; a missing +# grant degrades to stdout only instead of failing the run. +RUNLOG=true diff --git a/deploy/kubernetes/job.yaml.tmpl b/deploy/kubernetes/job.yaml.tmpl index 06de5aa..48e9178 100644 --- a/deploy/kubernetes/job.yaml.tmpl +++ b/deploy/kubernetes/job.yaml.tmpl @@ -96,3 +96,9 @@ spec: value: "${ACTIVE_DEADLINE_SECONDS}" - name: S3GC_VERBOSE_FLAG value: "${VERBOSE}" + - name: S3GC_RUNLOG_FLAG + value: "${RUNLOG}" + # The Job name is the run id, so a run-log row traces back to the + # Job that wrote it after the pod and its logs are gone. + - name: S3GC_RUNID + value: "${JOB_NAME}" diff --git a/deploy/kubernetes/render.py b/deploy/kubernetes/render.py index 4e4c7d4..a4b275b 100644 --- a/deploy/kubernetes/render.py +++ b/deploy/kubernetes/render.py @@ -45,7 +45,9 @@ # Optional keys and their defaults. An empty value renders no environment # variable at all, because s3gc parses S3GC_USETOTAL as an integer and would # reject an empty string. -OPTIONAL = {"USETOTAL": ""} +# RUNLOG defaults on: the durable run-log table is the only record that +# outlives the Job, whose pod and logs ttlSecondsAfterFinished deletes. +OPTIONAL = {"USETOTAL": "", "RUNLOG": "true"} # Environment variables dropped from the manifest when they render empty. OPTIONAL_ENV = ("S3GC_USETOTAL",) JOB_NAME_RE = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$") @@ -98,6 +100,8 @@ def validate(values: dict[str, str]) -> None: raise ValueError("VERBOSE must be true or false") if values["ORDER_BY_OBJPATH"] not in {"true", "false"}: raise ValueError("ORDER_BY_OBJPATH must be true or false") + if values["RUNLOG"] not in {"true", "false"}: + raise ValueError("RUNLOG must be true or false") if values["USETOTAL"] and ( not values["USETOTAL"].isdigit() or int(values["USETOTAL"]) < 1 ): diff --git a/s3gc.py b/s3gc.py index 39ebfb5..abf9ca1 100644 --- a/s3gc.py +++ b/s3gc.py @@ -33,6 +33,8 @@ import urllib3 import logging import datetime +import socket +import uuid usage = """ s3 garbage collector for ClickHouse @@ -501,6 +503,30 @@ def coerce_bool(value): help="list all command line options for internal purposes", ) +parser.add_argument( + "--runlog", + "--run-log", + dest="runlog_flag", + type=coerce_bool, + default=True, + help=( + "record durable run events in a ClickHouse table alongside the auxiliary " + "table. Pod logs are ephemeral: they are rotated by the kubelet and deleted " + "with the Job. Set false to keep stdout as the only record" + ), +) +parser.add_argument( + "--runid", + "--run-id", + dest="runid", + default="", + help=( + "identifier recorded with every run-log row. Defaults to a generated " + "timestamped id; the Kubernetes Job template passes the Job name so a row " + "traces back to the Job that wrote it" + ), +) + parser.add_argument("--cfg", action=ActionConfigFile) @@ -515,6 +541,7 @@ def coerce_bool(value): # can rely on real bools. BOOLEAN_DESTS = ( "s3secure_flag", + "runlog_flag", "use_remove_objects", "keepdata_flag", "collectonly_flag", @@ -635,6 +662,15 @@ def graceful_exit(): else: tname = f"`{dbparts[0]}{args.s3diskname}`" +# The run-log table lives beside the auxiliary table and follows the same naming +# convention, so one COLLECTTABLEPREFIX still identifies one cleanup. Unlike the +# auxiliary table it is NEVER truncated: it is the durable record of what the +# cleanup did after the pod and its logs are gone. +if dbname: + log_tname = f"{dbname}.`{dbparts[1]}{args.s3diskname}_log`" +else: + log_tname = f"`{dbparts[0]}{args.s3diskname}_log`" + minio_client = None ch_client = None # A second ClickHouse client, used only for writes issued while a result @@ -646,6 +682,142 @@ class S3DeletionError(RuntimeError): """A delete failed after successful deletions were checkpointed.""" +############################################################## +# Durable run log. +# +# Pod logs are not a record. The kubelet rotates container output (10Mi by +# default), so `kubectl logs` cannot return the beginning of a long run, and +# ttlSecondsAfterFinished deletes the Job and its pods along with everything +# they printed. A cleanup that reclaimed terabytes left no evidence of what it +# did once that window closed. +# +# So the same events are appended to a ClickHouse table. Three rules, each with +# a test, because this is bookkeeping attached to an irreversible operation: +# +# 1. Writes go on ch_writer, NEVER ch_client. do_use() holds ch_client's +# session for the whole anti-join stream, and a second query on a held +# session is SESSION_IS_LOCKED (373). Same reason tombstones live there. +# 2. A logging failure NEVER fails the run. One failure disables the run log +# for the remainder of the process rather than retrying every batch: a +# delete that is mid-flight must not die over its own audit trail. +# 3. Messages are redacted through LogFormatter._filter before insert, so a +# credential cannot reach a table that outlives the run. +############################################################## + +RUN_LOG_COLUMNS = [ + "event_time", + "run_id", + "phase", + "event", + "message", + "objects", + "bytes", + "s3bucket", + "s3path", + "s3diskname", + "clustername", + "dryrun", + "chhost", + "hostname", +] + +run_log_enabled = False +run_id = "" + + +def resolve_run_id(): + """A stable identifier for every row this process writes.""" + if args.runid: + return args.runid + stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return f"{stamp}-{uuid.uuid4().hex[:8]}" + + +def init_run_log(): + """Create the run-log table. Degrade to stdout-only rather than failing.""" + global run_log_enabled, run_id + if not args.runlog_flag: + logger.debug("run log disabled by --runlog false") + return + + run_id = resolve_run_id() + try: + ch_writer.command( + f"""CREATE TABLE IF NOT EXISTS {log_tname} ( + event_time DateTime64(3), + run_id String, + phase LowCardinality(String), + event LowCardinality(String), + message String, + objects UInt64, + bytes UInt64, + s3bucket String, + s3path String, + s3diskname LowCardinality(String), + clustername String, + dryrun Bool, + chhost String, + hostname String + ) ENGINE = MergeTree ORDER BY (run_id, event_time)""" + ) + except Exception as exc: + # Most likely a missing CREATE TABLE grant. The cleanup itself is + # unaffected, so say so once and carry on without the durable record. + logger.warning( + f"run log unavailable, continuing with stdout only: {exc}. " + f"Grant CREATE TABLE on {log_tname} to record durable run history, " + "or pass --runlog false to silence this." + ) + return + + run_log_enabled = True + logger.info(f"run log: {log_tname}, run_id={run_id}") + + +def run_log(event, message="", objects=0, bytes_=0, phase="run"): + """Append one durable event row. Never raises, never fails the run.""" + global run_log_enabled + if not run_log_enabled: + return + try: + ch_writer.insert( + log_tname, + [[ + datetime.datetime.now(datetime.timezone.utc), + run_id, + phase, + event, + LogFormatter._filter(str(message)), + int(objects), + int(bytes_), + args.s3bucket, + args.s3path, + args.s3diskname, + args.clustername, + bool(args.dryrun_flag), + args.chhost, + socket.gethostname(), + ]], + column_names=RUN_LOG_COLUMNS, + ) + except Exception as exc: + # Disable rather than retry: a per-batch failure would otherwise repeat + # for every batch of a multi-hour delete. + run_log_enabled = False + logger.warning(f"run log write failed, disabling run log for this run: {exc}") + + +def current_phase(): + """The phase name an operator would recognise from the Job that is running.""" + if args.collectonly_flag: + return "collect" + if args.dryrun_flag: + return "dry-run" + if args.usecollected_flag: + return "delete" + return "collect+use" + + def _query_single_value(query): result = ch_client.query(query) if not result.result_rows or not result.result_rows[0]: @@ -924,6 +1096,11 @@ def do_collect(): rest_row_nums = args.total # None if not set num_inserted = 0 total_size = 0 + # Progress is throttled rather than per-batch: at the default + # collectbatchsize=1024 a ten-million-object bucket would otherwise write + # ten thousand rows to say the same thing. + RUN_LOG_EVERY = 100_000 + next_progress = RUN_LOG_EVERY while go_on: objs = [] for batch_element in range(0, args.collectbatchsize): @@ -942,6 +1119,17 @@ def do_collect(): ch_client.insert(tname, objs, column_names=["objpath", "size", "last_modified", "active"]) logger.debug(f"{len(objs)} rows inserted in {tname}") num_inserted += len(objs) + if num_inserted >= next_progress: + # The collect phase is otherwise silent at INFO for hours. + logger.info(f"collect progress: {num_inserted} objects, {total_size} bytes") + run_log( + "progress", + f"{num_inserted} objects listed", + objects=num_inserted, + bytes_=total_size, + phase="collect", + ) + next_progress += RUN_LOG_EVERY if rest_row_nums is not None: rest_row_nums -= len(objs) if rest_row_nums == 0 or go_on == False: @@ -955,6 +1143,13 @@ def do_collect(): logger.info( f"information about {num_inserted} objects of total size {total_size} is inserted in {tname}" ) + run_log( + "finish", + f"collected {num_inserted} objects into {tname}", + objects=num_inserted, + bytes_=total_size, + phase="collect", + ) def check_samples_match_partitioning(): @@ -983,6 +1178,11 @@ def check_samples_match_partitioning(): f"partitioning ({partition_key}). Partition pruning will be lost; " "use the same --samples value that the collect phase used." ) + run_log( + "warning", + f"--samples {args.samples} does not match partitioning {partition_key}", + phase="use", + ) def do_use(): @@ -1051,6 +1251,7 @@ def make_antijoin(calc_only=False, sample=None): num_rows, total_size = result.result_rows[0] if num_rows == 0: logger.info("Nothing to do") + run_log("finish", "nothing to do", phase="use") graceful_exit() while True: @@ -1071,6 +1272,13 @@ def make_antijoin(calc_only=False, sample=None): for sample in range(0, args.samples): antijoin = make_antijoin(sample=sample) logger.info(f"antijoin {antijoin}") + run_log( + "progress", + f"sample {sample} of {args.samples} started", + objects=num_removed, + bytes_=total_size, + phase="use", + ) with ch_client.query_row_block_stream(antijoin) as stream: for block in stream: @@ -1131,8 +1339,24 @@ def make_antijoin(calc_only=False, sample=None): logger.info( f"delete checkpoint: {num_removed} objects / {total_size} bytes removed so far" ) + # The durable twin of the checkpoint above: if the pod is + # gone, this row is what says how far the delete got. + run_log( + "checkpoint", + f"sample {sample}: {len(successful_rows)} objects deleted in this batch", + objects=num_removed, + bytes_=total_size, + phase="use", + ) if errors: + run_log( + "error", + f"{len(errors)} S3 deletion error(s) in sample {sample}", + objects=num_removed, + bytes_=total_size, + phase="use", + ) raise S3DeletionError( f"{len(errors)} S3 deletion error(s); successful deletes were checkpointed" ) @@ -1145,6 +1369,14 @@ def make_antijoin(calc_only=False, sample=None): f"{'are removed' if not args.dryrun_flag else 'would be removed but for dryrun flag'} " "in this attempt" ) + run_log( + "finish", + f"{num_removed} objects " + f"{'removed' if not args.dryrun_flag else 'would be removed (dry run)'} in this attempt", + objects=num_removed, + bytes_=total_size, + phase="use", + ) if not args.dryrun_flag: try: cumulative = ch_client.query( @@ -1154,6 +1386,13 @@ def make_antijoin(calc_only=False, sample=None): f"cumulative for this auxiliary table: {cumulative[0]} objects / " f"{cumulative[1]} bytes tombstoned" ) + run_log( + "progress", + "cumulative tombstones for this auxiliary table", + objects=cumulative[0] or 0, + bytes_=cumulative[1] or 0, + phase="use", + ) except Exception as exc: # never fail a completed run over a status query logger.info(f"could not read cumulative tombstone count: {exc}") @@ -1165,6 +1404,8 @@ def make_antijoin(calc_only=False, sample=None): def main(): try: connect_to_ch() + init_run_log() + run_log("start", f"{current_phase()} phase started") if not (args.usecollected_flag and args.dryrun_flag): connect_to_s3() if not args.usecollected_flag: @@ -1172,13 +1413,21 @@ def main(): if not args.collectonly_flag: do_use() + run_log("finish", f"{current_phase()} phase completed") graceful_exit() except UserVisibleError as exc: + run_log("error", str(exc)) if args.debug_flag: logger.exception(str(exc)) else: logger.error(str(exc)) sys.exit(1) + except Exception as exc: + # A crash is exactly the case where the pod log is least likely to + # survive, so record it and then let it propagate unchanged. SystemExit + # is not an Exception, so graceful_exit() does not land here. + run_log("error", f"{type(exc).__name__}: {exc}") + raise if __name__ == "__main__": diff --git a/tests/test_s3gc.py b/tests/test_s3gc.py index fe5ec92..43bbd86 100644 --- a/tests/test_s3gc.py +++ b/tests/test_s3gc.py @@ -82,8 +82,14 @@ def make_args(**overrides): "interactive_flag": False, "use_remove_objects": True, "s3bucket": "bucket", + "s3path": "data/", + "chhost": "replica-0", "keepdata_flag": True, "silent_flag": True, + # Durable run log. run_log_enabled starts False at module level, so + # tests that do not opt in are unaffected by these. + "runlog_flag": True, + "runid": "test-run", # S3 auth surface (static | aws | iam) "s3auth": "static", "s3profile": "", @@ -1134,3 +1140,355 @@ def test_useafter_is_quoted_as_a_string_literal( sql = _antijoin_sql(s3gc_module, args_factory, monkeypatch, useafter="some/object") assert "s3o.objpath > 'some/object'" in sql + + +# --------------------------------------------------------------------------- +# Durable run log. +# +# Pod logs are not a record: the kubelet rotates container output and +# ttlSecondsAfterFinished deletes the Job with everything it printed. The same +# events are therefore appended to a ClickHouse table that outlives the pod. +# +# This is bookkeeping attached to an irreversible operation, so the tests below +# are mostly about what it must NOT do: never fail a run, never share the +# streaming session, never carry a credential into a table. +# --------------------------------------------------------------------------- + + +class RecordingCH(FakeCH): + """A client that records DDL and TRUNCATE instead of rejecting them.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.commands = [] + + def command(self, query): + self.commands.append(query) + if "COUNT(1)" in query: + return 1 + return None + + +def _enable_run_log(s3gc_module, monkeypatch, writer, **arg_overrides): + """Turn the run log on the way init_run_log() would, without a server.""" + namespace = s3gc_module["run_log"].__globals__ + monkeypatch.setitem(namespace, "ch_writer", writer) + monkeypatch.setitem(namespace, "run_log_enabled", True) + monkeypatch.setitem(namespace, "run_id", "test-run") + return namespace + + +def test_run_log_table_is_created_beside_the_auxiliary_table( + s3gc_module, args_factory, monkeypatch +): + """One COLLECTTABLEPREFIX still identifies one cleanup.""" + namespace = s3gc_module["init_run_log"].__globals__ + writer = RecordingCH() + monkeypatch.setitem(namespace, "args", args_factory()) + monkeypatch.setitem(namespace, "ch_writer", writer) + monkeypatch.setitem(namespace, "log_tname", "`s3objects_for_s3_log`") + + s3gc_module["init_run_log"]() + + ddl = " ".join(writer.commands[0].split()) + assert "CREATE TABLE IF NOT EXISTS `s3objects_for_s3_log`" in ddl + assert "ENGINE = MergeTree ORDER BY (run_id, event_time)" in ddl + assert namespace["run_log_enabled"] is True + + +def test_run_log_uses_the_configured_run_id(s3gc_module, args_factory, monkeypatch): + """The Kubernetes template passes JOB_NAME, so a row traces to its Job.""" + namespace = s3gc_module["init_run_log"].__globals__ + monkeypatch.setitem(namespace, "args", args_factory(runid="s3gc-delete-42")) + monkeypatch.setitem(namespace, "ch_writer", RecordingCH()) + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + + s3gc_module["init_run_log"]() + + assert namespace["run_id"] == "s3gc-delete-42" + + +def test_missing_create_grant_degrades_to_stdout_only( + s3gc_module, args_factory, monkeypatch, caplog +): + """A cleanup must not fail because it could not create its own log table.""" + namespace = s3gc_module["init_run_log"].__globals__ + + class NoGrantCH: + def command(self, query): + raise RuntimeError("Not enough privileges") + + monkeypatch.setitem(namespace, "args", args_factory()) + monkeypatch.setitem(namespace, "ch_writer", NoGrantCH()) + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + + with caplog.at_level("WARNING"): + s3gc_module["init_run_log"]() + + assert namespace["run_log_enabled"] is False + assert "run log unavailable" in caplog.text + + +def test_run_log_disabled_creates_no_table(s3gc_module, args_factory, monkeypatch): + """--runlog false must not touch the cluster at all.""" + namespace = s3gc_module["init_run_log"].__globals__ + writer = RecordingCH() + monkeypatch.setitem(namespace, "args", args_factory(runlog_flag=False)) + monkeypatch.setitem(namespace, "ch_writer", writer) + + s3gc_module["init_run_log"]() + + assert writer.commands == [] + assert namespace["run_log_enabled"] is False + + +def test_a_failed_run_log_write_never_fails_the_run( + s3gc_module, args_factory, monkeypatch, caplog +): + """The audit trail must not be able to kill a delete that is mid-flight.""" + attempts = [] + + class BrokenWriter: + def insert(self, *a, **k): + attempts.append(1) + raise RuntimeError("table went away") + + namespace = _enable_run_log(s3gc_module, monkeypatch, BrokenWriter()) + monkeypatch.setitem(namespace, "args", args_factory()) + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + + with caplog.at_level("WARNING"): + s3gc_module["run_log"]("checkpoint", "batch done") + s3gc_module["run_log"]("checkpoint", "another batch") + + # Disabled after the first failure, not retried once per batch for hours. + assert attempts == [1] + assert namespace["run_log_enabled"] is False + assert "disabling run log" in caplog.text + + +def test_run_log_redacts_secrets(s3gc_module, args_factory, monkeypatch): + """A credential must not reach a table that outlives the run.""" + writer = RecordingCH() + namespace = _enable_run_log(s3gc_module, monkeypatch, writer) + monkeypatch.setitem(namespace, "args", args_factory()) + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + monkeypatch.setattr( + s3gc_module["LogFormatter"], "filter_strings", ["super-secret-key"] + ) + + s3gc_module["run_log"]("error", "failed with key super-secret-key in it") + + message = writer.inserts[0][1][0][4] + assert "super-secret-key" not in message + assert "****" in message + + +def test_run_log_writes_off_the_streaming_session( + s3gc_module, args_factory, monkeypatch +): + """Same rule as tombstones: never ch_client while the anti-join streams. + + A run-log insert on the streaming client is a second query on a held + session, which ClickHouse rejects with SESSION_IS_LOCKED (373) — and it + would do so mid-delete, the worst possible moment. + """ + namespace = s3gc_module["do_use"].__globals__ + streaming = StreamingCH(blocks=[[("orphan-a", 10, "time")]]) + writer = RecordingCH() + + class SuccessfulMinio: + def remove_objects(self, bucket, objects): + return iter(()) + + monkeypatch.setitem(namespace, "args", args_factory()) + monkeypatch.setitem(namespace, "ch_client", streaming) + monkeypatch.setitem(namespace, "ch_writer", writer) + monkeypatch.setitem(namespace, "minio_client", SuccessfulMinio()) + monkeypatch.setitem(namespace, "run_log_enabled", True) + monkeypatch.setitem(namespace, "run_id", "test-run") + monkeypatch.setitem(namespace, "log_tname", "`s3objects_for_s3_log`") + + s3gc_module["do_use"]() + + assert streaming.inserts == [] + tables = [insert[0] for insert in writer.inserts] + assert "`s3objects_for_s3`" in tables # tombstone + assert "`s3objects_for_s3_log`" in tables # run-log rows + + +def test_delete_batches_are_recorded_durably(s3gc_module, args_factory, monkeypatch): + """If the pod is gone, these rows are what say how far the delete got.""" + namespace = s3gc_module["do_use"].__globals__ + client = RecordingCH(blocks=[[("orphan-a", 10, "time"), ("orphan-b", 20, "time")]]) + + class SuccessfulMinio: + def remove_objects(self, bucket, objects): + return iter(()) + + monkeypatch.setitem(namespace, "args", args_factory(deletebatchsize=1)) + monkeypatch.setitem(namespace, "ch_client", client) + monkeypatch.setitem(namespace, "ch_writer", client) + monkeypatch.setitem(namespace, "minio_client", SuccessfulMinio()) + monkeypatch.setitem(namespace, "run_log_enabled", True) + monkeypatch.setitem(namespace, "run_id", "test-run") + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + + s3gc_module["do_use"]() + + events = [ + (row[3], row[5], row[6]) + for table, rows, _ in client.inserts + if table == "`aux_log`" + for row in rows + ] + checkpoints = [event for event in events if event[0] == "checkpoint"] + assert len(checkpoints) == 2 + # Running totals, so a truncated log still shows how far it got. + assert checkpoints[0][1] == 1 and checkpoints[1][1] == 2 + assert checkpoints[-1][2] == 30 + assert any(event[0] == "finish" for event in events) + + +def test_run_log_table_is_never_truncated(s3gc_module, args_factory, monkeypatch): + """The aux table is scratch space; the run log is the record. Only one is wiped.""" + namespace = s3gc_module["do_use"].__globals__ + client = RecordingCH(blocks=[]) + + monkeypatch.setitem( + namespace, "args", args_factory(keepdata_flag=False, dryrun_flag=False) + ) + monkeypatch.setitem(namespace, "ch_client", client) + monkeypatch.setitem(namespace, "ch_writer", client) + monkeypatch.setitem(namespace, "run_log_enabled", True) + monkeypatch.setitem(namespace, "run_id", "test-run") + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + + s3gc_module["do_use"]() + + truncates = [query for query in client.commands if "TRUNCATE" in query] + assert truncates == ["TRUNCATE TABLE `s3objects_for_s3`"] + assert not any("aux_log" in query for query in truncates) + + +def test_run_log_records_the_scope_of_the_run(s3gc_module, args_factory, monkeypatch): + """Every row is self-describing evidence, not just a message.""" + writer = RecordingCH() + namespace = _enable_run_log(s3gc_module, monkeypatch, writer) + monkeypatch.setitem( + namespace, + "args", + args_factory(s3bucket="the-bucket", s3path="pre/fix/", s3diskname="gcs", + clustername="prod", dryrun_flag=True, chhost="replica-0"), + ) + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + + s3gc_module["run_log"]("start", "dry-run phase started", phase="run") + + row = dict(zip(s3gc_module["RUN_LOG_COLUMNS"], writer.inserts[0][1][0])) + assert row["run_id"] == "test-run" + assert row["phase"] == "run" and row["event"] == "start" + assert row["s3bucket"] == "the-bucket" and row["s3path"] == "pre/fix/" + assert row["s3diskname"] == "gcs" and row["clustername"] == "prod" + assert row["dryrun"] is True and row["chhost"] == "replica-0" + + +def test_collect_reports_progress_for_long_runs( + s3gc_module, args_factory, monkeypatch, caplog +): + """A multi-hour collect used to emit nothing at INFO until it finished.""" + import datetime + + namespace = s3gc_module["do_collect"].__globals__ + now = datetime.datetime.now(datetime.timezone.utc) + + class Obj: + def __init__(self, name): + self.object_name = name + self.size = 1 + self.last_modified = now - datetime.timedelta(days=2) + + class Minio: + def list_objects(self, bucket, prefix, recursive, start_after): + return iter(Obj(f"object-{index}") for index in range(200_000)) + + writer = RecordingCH() + monkeypatch.setitem( + namespace, + "args", + args_factory(age=0, collectbatchsize=50_000, total=None, collectafter="", + s3path="", createdatabase_flag=False, + drop_collecttable_flag=False, samples=4), + ) + monkeypatch.setitem(namespace, "minio_client", Minio()) + monkeypatch.setitem(namespace, "ch_client", RecordingCH()) + monkeypatch.setitem(namespace, "ch_writer", writer) + monkeypatch.setitem(namespace, "tname", "`aux`") + monkeypatch.setitem(namespace, "log_tname", "`aux_log`") + monkeypatch.setitem(namespace, "run_log_enabled", True) + monkeypatch.setitem(namespace, "run_id", "test-run") + + with caplog.at_level("INFO", logger="s3gc_test"): + s3gc_module["do_collect"]() + + assert "collect progress" in caplog.text + events = [ + row[3] for table, rows, _ in writer.inserts if table == "`aux_log`" for row in rows + ] + assert events.count("progress") == 2 # throttled to every 100k + assert events[-1] == "finish" + + +@pytest.mark.parametrize( + ("replacement", "message"), + [ + (("RUNLOG=true", "RUNLOG=maybe"), "RUNLOG must be true or false"), + (("RUNLOG=true", "RUNLOG="), "RUNLOG must be true or false"), + ], +) +def test_renderer_validates_runlog(tmp_path, replacement, message): + source = (ROOT / "deploy/kubernetes/example.env").read_text() + config_path = tmp_path / "runlog.env" + config_path.write_text(source.replace(*replacement)) + + result = subprocess.run( + [sys.executable, str(ROOT / "deploy/kubernetes/render.py"), config_path], + capture_output=True, text=True, check=False, + ) + + assert result.returncode == 64 + assert message in result.stderr + + +def test_renderer_defaults_runlog_on_for_legacy_env_files(tmp_path): + """Pre-existing env files must keep rendering, with the run log enabled.""" + source = (ROOT / "deploy/kubernetes/example.env").read_text() + without = "\n".join( + line for line in source.splitlines() if not line.startswith("RUNLOG=") + ) + config_path = tmp_path / "legacy.env" + config_path.write_text(without) + + result = subprocess.run( + [sys.executable, str(ROOT / "deploy/kubernetes/render.py"), config_path], + capture_output=True, text=True, check=False, + ) + + assert result.returncode == 0, result.stderr + assert "S3GC_RUNLOG_FLAG" in result.stdout + assert 'value: "true"' in result.stdout + + +def test_renderer_uses_the_job_name_as_the_run_id(tmp_path): + """A run-log row must trace back to the Job that wrote it.""" + config_path = tmp_path / "runid.env" + config_path.write_text((ROOT / "deploy/kubernetes/example.env").read_text()) + + result = subprocess.run( + [sys.executable, str(ROOT / "deploy/kubernetes/render.py"), config_path], + capture_output=True, text=True, check=False, + ) + + assert result.returncode == 0 + assert "- name: S3GC_RUNID" in result.stdout + assert 'value: "s3gc-example-dry-run"' in result.stdout