diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py index db227a0f..7a96ca29 100644 --- a/datadog_sync/utils/resources_handler.py +++ b/datadog_sync/utils/resources_handler.py @@ -740,7 +740,22 @@ async def import_resources_without_saving(self) -> None: await self.worker.schedule_workers() import_failures = self.worker.counter.failure if get_failures == 0 and import_failures == 0: - self.config.state.mark_source_authoritative(self.config.resources_arg) + # Only types WITHOUT an id-file scoping are authoritative after this + # import — types with id_payload only fetched the requested subset, + # so state.source[type] is intentionally partial and cannot drive + # stale-file pruning. Marking a partial subset as authoritative + # would over-prune every non-listed file on the next dump_state. + id_scoped_types = set(self.config.id_payload or {}) + authoritative_types = [ + rt for rt in self.config.resources_arg if rt not in id_scoped_types + ] + if authoritative_types: + self.config.state.mark_source_authoritative(authoritative_types) + if id_scoped_types & set(self.config.resources_arg): + self.config.logger.debug( + "source state not marked authoritative for id-file-scoped types: %s", + sorted(id_scoped_types & set(self.config.resources_arg)), + ) else: self.config.logger.debug( "source state not marked authoritative after import: get_failures=%s import_failures=%s", diff --git a/datadog_sync/utils/state.py b/datadog_sync/utils/state.py index 6738a4f7..0847d7f5 100644 --- a/datadog_sync/utils/state.py +++ b/datadog_sync/utils/state.py @@ -3,6 +3,7 @@ # This product includes software developed at Datadog (https://www.datadoghq.com/). # Copyright 2019 Datadog, Inc. import logging +import os import time from typing import Any, Dict, List, Set, Tuple @@ -232,12 +233,107 @@ def reload_destination(self, resource_types: List[str]) -> Dict[str, int]: def dump_state(self, origin: Origin = Origin.ALL) -> None: dump_start = time.perf_counter() self._storage.put(origin, self._data) + pruned = self._prune_stale_files_after_put(origin) log.info( - "sync-cli-timing phase=dump_state origin=%s wall_ms=%d", + "sync-cli-timing phase=dump_state origin=%s prune_status=%s " + "pruned_source=%d pruned_destination=%d wall_ms=%d", origin.value, + pruned.get("status", "ok"), + pruned.get(Origin.SOURCE, 0), + pruned.get(Origin.DESTINATION, 0), int((time.perf_counter() - dump_start) * 1000), ) + def _prune_stale_files_after_put(self, origin: Origin) -> Dict[Any, Any]: + """Delete per-resource cache files whose IDs are no longer in-memory. + + Restores what the legacy monolithic-cache layout achieved incidentally: + a source ID that drops between two ``dump_state`` calls leaves no + stale cache file behind. Under ``--resource-per-file`` the write path + only writes files for currently-live IDs, so without this step stale + files accumulate and get re-read as authoritative source content on + the next ``load_state`` — see prune subcommand and its docstrings for + the pre-existing manual escape hatch this makes automatic. + + Gate discipline: only prunes types marked authoritative via + ``mark_source_authoritative``. Callers that write partial or + filtered state (e.g. mid-run intermediate saves, ``--minimize-reads`` + type-scoped loads, ``--id-file`` targeted paths) do not mark + authoritative for the scoped types, so this method silently no-ops + for them and cannot over-prune. + + Destination-side pruning uses the same source-ID-keyed filename + convention (state.py:275-277: "Destination state files are keyed by + source IDs") and the same authoritative-source guard, so a source ID + that dropped after a successful destination API DELETE + in-memory + pop is cleaned from both sides on the same dump. + + Only active under ``--resource-per-file``. Legacy monolithic mode is + naturally self-cleaning and needs no change. + + Kill switch: set ``DD_SYNC_CLI_DISABLE_DUMP_PRUNE=1`` in the + environment to disable the new behavior at runtime without a code + revert. Intended as an emergency escape hatch; not for routine use. + + Returns a dict with per-origin file-delete counts and a ``status`` + string that discriminates why prune returned zero (``ok``, + ``gated_kill_switch``, ``gated_legacy_layout``, + ``gated_no_authoritative``, ``no_stale``, ``compute_error``, + ``delete_error``). Never raises: a best-effort cleanup; per-file + failures are surfaced via ``delete_many`` return values and logged + at DEBUG in ``delete_stale_files``. + """ + result: Dict[Any, Any] = {Origin.SOURCE: 0, Origin.DESTINATION: 0, "status": "ok"} + if os.environ.get("DD_SYNC_CLI_DISABLE_DUMP_PRUNE"): + result["status"] = "gated_kill_switch" + return result + if not getattr(self._storage, "resource_per_file", False): + result["status"] = "gated_legacy_layout" + return result + if not self._authoritative_source_types: + result["status"] = "gated_no_authoritative" + return result + + # Only prune origins that were actually written on this call. + target_origins: List[Origin] = [] + if origin in (Origin.SOURCE, Origin.ALL): + target_origins.append(Origin.SOURCE) + if origin in (Origin.DESTINATION, Origin.ALL): + target_origins.append(Origin.DESTINATION) + if not target_origins: + result["status"] = "no_stale" + return result + + types = sorted(self._authoritative_source_types) + try: + stale = self.compute_stale_files(target_origins, types) + except ValueError as e: + # Race with concurrent authoritative-clear or missing source + # state: skip pruning this cycle rather than raising into + # dump_state's happy path. Next dump will retry. Log at WARNING + # so a genuine invariant break (per compute_stale_files docstring) + # is visible to operators rather than silently swallowed. + log.warning("dump_state prune skipped due to compute_stale_files: %s", e) + result["status"] = "compute_error" + return result + if not any(stale.values()): + result["status"] = "no_stale" + return result + + try: + counts = self.delete_stale_files(stale) + except Exception as e: + # Backend errors during delete (transient GCS/S3 blip, permission + # change mid-run) must not abort dump_state's happy path — the + # write has already succeeded and callers rely on returning. + # Stale files remain on disk; next dump_state retries. + log.warning("dump_state prune failed during delete_stale_files: %s", e) + result["status"] = "delete_error" + return result + for (o, _rt), (ok, _fail) in counts.items(): + result[o] = result.get(o, 0) + ok + return result + def get_all_resources(self, resources_types: List[str]) -> Dict[Tuple[str, str], Any]: """Returns all resources of the given types. diff --git a/tests/unit/test_state_dump_cleanup.py b/tests/unit/test_state_dump_cleanup.py new file mode 100644 index 00000000..5676adec --- /dev/null +++ b/tests/unit/test_state_dump_cleanup.py @@ -0,0 +1,300 @@ +# Unless explicitly stated otherwise all files in this repository are licensed +# under the 3-clause BSD style license (see LICENSE). +# This product includes software developed at Datadog (https://www.datadoghq.com/). +# Copyright 2019 Datadog, Inc. + +"""Tests for the delete-on-drop behavior in State.dump_state. + +Under --resource-per-file the write path historically only wrote files for +currently-live IDs. Files for IDs that dropped between two dump_state calls +survived on the storage backend and got re-read into state.source on the +next load_state, resurrecting deleted-source records as apparent-authoritative +content. The fix plumbs the existing compute_stale_files / delete_stale_files +primitives into dump_state, gated on the same authoritative-source marker +that compute_stale_files itself uses. +""" + +from datadog_sync.constants import Origin +from datadog_sync.utils.state import State +from datadog_sync.utils.storage.storage_types import StorageType + + +def _make_state(tmp_path, resource_per_file=True): + src = tmp_path / "source" + dst = tmp_path / "dest" + src.mkdir() + dst.mkdir() + state = State( + type_=StorageType.LOCAL_FILE, + resource_per_file=resource_per_file, + source_resources_path=str(src), + destination_resources_path=str(dst), + ) + return state, src, dst + + +def _list_source_files(src, resource_type): + return sorted(p.name for p in src.iterdir() if p.name.startswith(f"{resource_type}.")) + + +def _list_destination_files(dst, resource_type): + return sorted(p.name for p in dst.iterdir() if p.name.startswith(f"{resource_type}.")) + + +# ---------- create path (baseline, unchanged) ---------- + + +def test_dump_state_creates_per_id_files_under_resource_per_file(tmp_path): + state, src, _dst = _make_state(tmp_path) + state.source["monitors"] = {"a": {"id": "a"}, "b": {"id": "b"}} + state.mark_source_authoritative(["monitors"]) + + state.dump_state(Origin.SOURCE) + + assert _list_source_files(src, "monitors") == ["monitors.a.json", "monitors.b.json"] + + +# ---------- update path (baseline, unchanged) ---------- + + +def test_dump_state_overwrites_existing_files_for_current_ids(tmp_path): + state, src, _dst = _make_state(tmp_path) + (src / "monitors.a.json").write_text('{"a": {"id": "a", "stale": true}}') + + state.source["monitors"] = {"a": {"id": "a", "stale": False}} + state.mark_source_authoritative(["monitors"]) + state.dump_state(Origin.SOURCE) + + assert '"stale": false' in (src / "monitors.a.json").read_text() + + +# ---------- new: source-ID-drop cleanup ---------- + + +def test_dump_state_deletes_source_file_for_dropped_id(tmp_path): + state, src, _dst = _make_state(tmp_path) + # Simulate a prior dump that produced two files. + (src / "monitors.keep.json").write_text('{"keep": {"id": "keep"}}') + (src / "monitors.drop.json").write_text('{"drop": {"id": "drop"}}') + + # New in-memory state has only one of the two IDs (the other was deleted + # on source and the import phase dropped it from state.source). + state.source["monitors"] = {"keep": {"id": "keep"}} + state.mark_source_authoritative(["monitors"]) + + state.dump_state(Origin.SOURCE) + + assert _list_source_files(src, "monitors") == ["monitors.keep.json"] + + +def test_dump_state_deletes_destination_file_for_dropped_source_id(tmp_path): + """Destination state files are keyed by source ID (state.py:275-277). + + A source ID that dropped from state.source therefore has its + destination-side cache file cleaned on the same dump. This is what allows + a subsequent load_state to correctly compute + ``state.destination - state.source`` for the --cleanup path. + """ + state, src, dst = _make_state(tmp_path) + (src / "monitors.keep.json").write_text('{"keep": {"id": "keep"}}') + (dst / "monitors.keep.json").write_text('{"keep": {"id": "dest_1"}}') + (dst / "monitors.dropped.json").write_text('{"dropped": {"id": "dest_2"}}') + + state.source["monitors"] = {"keep": {"id": "keep"}} + # Destination in-memory keeps dropped's mapping (as it would after a + # successful load_state on a wrapper-orchestrated per-type sync). + state.destination["monitors"] = { + "keep": {"id": "dest_1"}, + "dropped": {"id": "dest_2"}, + } + state.mark_source_authoritative(["monitors"]) + + state.dump_state(Origin.ALL) + + assert _list_source_files(src, "monitors") == ["monitors.keep.json"] + # dropped's destination file removed because its source ID is gone; + # keep's remains. + assert _list_destination_files(dst, "monitors") == ["monitors.keep.json"] + + +# ---------- safety gates ---------- + + +def test_dump_state_does_not_prune_when_type_is_not_authoritative(tmp_path): + """Partial / filtered / --minimize-reads state.source has no authoritative + marker. dump_state must not over-prune in that case. + """ + state, src, _dst = _make_state(tmp_path) + (src / "monitors.a.json").write_text('{"a": {"id": "a"}}') + (src / "monitors.b.json").write_text('{"b": {"id": "b"}}') + + # Only a subset in-memory; authoritative NOT marked. + state.source["monitors"] = {"a": {"id": "a"}} + # No mark_source_authoritative call. + + state.dump_state(Origin.SOURCE) + + # Both files still present — b.json was NOT pruned. + assert _list_source_files(src, "monitors") == ["monitors.a.json", "monitors.b.json"] + + +def test_dump_state_does_not_prune_under_legacy_layout(tmp_path): + """Legacy (monolithic-per-type) mode is naturally self-cleaning via + single-file overwrite; the new pruning step must be a no-op there. + """ + state, src, _dst = _make_state(tmp_path, resource_per_file=False) + state.source["monitors"] = {"a": {"id": "a"}} + state.mark_source_authoritative(["monitors"]) + + state.dump_state(Origin.SOURCE) + + # Under legacy layout the write path uses a single monitors.json; no + # per-ID files are produced and the prune step is bypassed. + assert (src / "monitors.json").exists() + assert not (src / "monitors.a.json").exists() + + +def test_dump_state_source_only_does_not_prune_destination(tmp_path): + """When dump_state is called with Origin.SOURCE, destination-side files + must be untouched even if they'd be considered stale — the caller is + only refreshing source. + """ + state, src, dst = _make_state(tmp_path) + (dst / "monitors.dropped.json").write_text('{"dropped": {"id": "dest_2"}}') + state.source["monitors"] = {"keep": {"id": "keep"}} + state.mark_source_authoritative(["monitors"]) + + state.dump_state(Origin.SOURCE) + + # Destination file survives because Origin.SOURCE was requested. + assert _list_destination_files(dst, "monitors") == ["monitors.dropped.json"] + + +def test_dump_state_prunes_only_marked_types(tmp_path): + """If only some types are marked authoritative, only those are pruned. + Untouched types keep their pre-existing files even if the in-memory + dict is empty.""" + state, src, _dst = _make_state(tmp_path) + (src / "monitors.stale.json").write_text('{"stale": {"id": "stale"}}') + (src / "dashboards.stale.json").write_text('{"stale": {"id": "stale"}}') + + state.source["monitors"] = {} # emptied + state.source["dashboards"] = {} # emptied + state.mark_source_authoritative(["monitors"]) # only monitors + + state.dump_state(Origin.SOURCE) + + # monitors got pruned; dashboards did not. + assert _list_source_files(src, "monitors") == [] + assert _list_source_files(src, "dashboards") == ["dashboards.stale.json"] + + +def test_dump_state_empty_in_memory_prunes_all_files_when_authoritative(tmp_path): + """Deliberate empty state + authoritative marker means "source has no + records of this type" and every on-disk file for the type is stale. + """ + state, src, _dst = _make_state(tmp_path) + for _id in ("a", "b", "c"): + (src / f"monitors.{_id}.json").write_text('{"%s": {"id": "%s"}}' % (_id, _id)) + + state.source["monitors"] = {} + state.mark_source_authoritative(["monitors"]) + + state.dump_state(Origin.SOURCE) + + assert _list_source_files(src, "monitors") == [] + + +def test_dump_state_does_not_raise_when_no_stale_files(tmp_path): + state, src, _dst = _make_state(tmp_path) + state.source["monitors"] = {"a": {"id": "a"}} + state.mark_source_authoritative(["monitors"]) + + # No pre-existing files, no stale set. Should be a clean no-op prune. + state.dump_state(Origin.SOURCE) + + assert _list_source_files(src, "monitors") == ["monitors.a.json"] + + +# ---------- id-file / subset-scope interaction ---------- + + +def test_dump_state_respects_partial_subset_when_only_subset_is_authoritative(tmp_path): + """Regression guard: an import scoped to a subset of source IDs (e.g. via + --id-file) must NOT cause the new prune to delete files for IDs outside + the subset. Callers that scope imports to a subset also refrain from + calling mark_source_authoritative for the scoped type — this test proves + that discipline is respected end-to-end at the State layer. + """ + state, src, _dst = _make_state(tmp_path) + # Simulate cache from a prior full-scope run: three monitor files exist. + for _id in ("a", "b", "c"): + (src / f"monitors.{_id}.json").write_text('{"%s": {"id": "%s"}}' % (_id, _id)) + + # Current run only imported one ID (as would happen with --id-file). The + # caller correctly does NOT mark monitors authoritative because only a + # subset was fetched. + state.source["monitors"] = {"a": {"id": "a"}} + # No mark_source_authoritative call for monitors. + + state.dump_state(Origin.SOURCE) + + # b.json and c.json survive because monitors is not authoritative for + # this dump. Data-loss on --id-file paths would fail this test. + assert _list_source_files(src, "monitors") == [ + "monitors.a.json", + "monitors.b.json", + "monitors.c.json", + ] + + +# ---------- resilience ---------- + + +def test_dump_state_kill_switch_disables_prune(tmp_path, monkeypatch): + """DD_SYNC_CLI_DISABLE_DUMP_PRUNE=1 must fully disable the new prune + behavior at runtime, without needing a code revert. Operator escape + hatch for a destructive-cleanup regression discovered in prod. + """ + state, src, _dst = _make_state(tmp_path) + (src / "monitors.stale.json").write_text('{"stale": {"id": "stale"}}') + state.source["monitors"] = {"keep": {"id": "keep"}} + state.mark_source_authoritative(["monitors"]) + + monkeypatch.setenv("DD_SYNC_CLI_DISABLE_DUMP_PRUNE", "1") + + state.dump_state(Origin.SOURCE) + + # Stale file survives because the kill switch was set. + assert "monitors.stale.json" in _list_source_files(src, "monitors") + assert "monitors.keep.json" in _list_source_files(src, "monitors") + + +def test_dump_state_swallows_backend_delete_errors(tmp_path, monkeypatch, caplog): + """A transient backend error during stale-file delete must not abort + dump_state's happy path — the write has already succeeded and callers + rely on returning normally. The stale files remain and next dump_state + retries.""" + import logging as _logging + + state, src, _dst = _make_state(tmp_path) + (src / "monitors.stale.json").write_text('{"stale": {"id": "stale"}}') + state.source["monitors"] = {"keep": {"id": "keep"}} + state.mark_source_authoritative(["monitors"]) + + def boom(*_a, **_kw): + raise RuntimeError("simulated backend transient") + + monkeypatch.setattr(state, "delete_stale_files", boom) + + # Should not raise despite delete_stale_files exploding. + with caplog.at_level(_logging.WARNING): + state.dump_state(Origin.SOURCE) + + # The keep file was written normally; the stale file survives (retry + # will handle it on the next dump_state cycle). + files = _list_source_files(src, "monitors") + assert "monitors.keep.json" in files + assert "monitors.stale.json" in files + # Warning was emitted so operators can see the swallowed failure. + assert any("delete_stale_files" in r.getMessage() for r in caplog.records)