From a7514adbecd990207dfcf41c6e81baaf74af98de Mon Sep 17 00:00:00 2001 From: michael-richey Date: Tue, 14 Jul 2026 17:14:29 -0400 Subject: [PATCH 1/4] feat: add --drop-unresolvable-principals flag and drop-aware connection scaffolding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the opt-in `--drop-unresolvable-principals` flag plus the shared, inert-when-off machinery that later PRs wire into individual resource models. No behavior changes when the flag is absent (the default). - New `--drop-unresolvable-principals` CLI option (in `_diffs_options`, so it applies to sync/diffs/migrate) threaded through the `Configuration` dataclass. - `BaseResource._resolve_or_drop`: three-way resolver distinguishing destination-present, source-present ("not yet synced", hard-fail as today), and permanently-gone (absent from both — droppable only under the flag). - Shared `_filter_stale_binding_principals` / `_filter_stale_flat_roles` / `_raise_connection_error_if_any` helpers (rebuild lists rather than mutating in place, avoiding the enumerate/index-shift bug). - `ResourceConnectionError.empty_binding_risk` flag for the access-elevation case. - New `Counter` buckets (stale drops, empty-binding risks) + reset, surfaced by `_emit_apply_summary`; `config.counter` back-reference so models can record drops. - `_apply_resource_cb` adds a `risk:empty_restriction_policy` metric tag when a binding emptied out. Co-Authored-By: Claude Sonnet 5 --- datadog_sync/commands/shared/options.py | 13 ++ datadog_sync/utils/base_resource.py | 167 ++++++++++++++++++ datadog_sync/utils/configuration.py | 17 +- datadog_sync/utils/resource_utils.py | 7 +- datadog_sync/utils/resources_handler.py | 36 +++- datadog_sync/utils/workers.py | 31 +++- .../test_apply_resource_connection_error.py | 34 ++++ tests/unit/test_apply_summary_handler.py | 48 +++++ tests/unit/test_base_resource.py | 89 ++++++++++ .../test_drop_unresolvable_principals_cli.py | 43 +++++ tests/unit/test_resource_utils.py | 14 +- tests/unit/test_workers.py | 60 +++++++ 12 files changed, 551 insertions(+), 8 deletions(-) create mode 100644 tests/unit/test_base_resource.py create mode 100644 tests/unit/test_drop_unresolvable_principals_cli.py create mode 100644 tests/unit/test_workers.py diff --git a/datadog_sync/commands/shared/options.py b/datadog_sync/commands/shared/options.py index df944fccf..cac31365d 100644 --- a/datadog_sync/commands/shared/options.py +++ b/datadog_sync/commands/shared/options.py @@ -499,6 +499,19 @@ def click_config_file_provider(ctx: Context, opts: CustomOptionClass, value: Non help="Skip resource if resource connection fails.", cls=CustomOptionClass, ), + option( + "--drop-unresolvable-principals", + required=False, + is_flag=True, + default=False, + show_default=True, + help="For restriction policies and restricted_roles lists: drop principal/role " + "references that are absent from BOTH destination and source state (permanently " + "gone, e.g. deleted before the org's first import) instead of skipping the whole " + "resource. If dropping empties a binding/list that had entries at the source, the " + "resource is still skipped (access-elevation guard). Off by default.", + cls=CustomOptionClass, + ), option( "--cleanup", default="False", diff --git a/datadog_sync/utils/base_resource.py b/datadog_sync/utils/base_resource.py index 888675947..3b781bcb4 100644 --- a/datadog_sync/utils/base_resource.py +++ b/datadog_sync/utils/base_resource.py @@ -437,6 +437,173 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona return failed_connections + def _resolve_or_drop(self, plain_id: str, resource_to_connect: str) -> Tuple[Optional[str], bool]: + """Resolve plain_id (no "type:" prefix) against destination state. + + Returns (resolved_destination_id, is_permanently_stale): + - Present in state.destination[resource_to_connect]: returns + (destination_id, False) -- success path, caller keeps/remaps this entry. + - Absent from destination: calls ensure_resource_loaded() (for + --minimize-reads correctness) then checks state.source[resource_to_connect]. + - Present in source ("not yet synced"): returns (None, False) -- caller + must treat this as today's hard-fail (add plain_id to failed_connections); + NOT a drop. This is the legitimate "retry on a later sync" case. + - Absent from source (permanently gone -- deleted before this org's + first-ever import, or never existed): returns (None, True) -- caller + drops this entry ONLY if self.config.drop_unresolvable_principals is True; + if the flag is False, caller must treat this identically to the "not yet + synced" case (today's unchanged hard-fail). + + Callers own: parsing "type:id" composites to plain_id before calling, reassembling + "type:id" on success, deciding what "the list" means for their shape + (binding.principals vs. flat restricted_roles), and the "list is now empty" + access-elevation check. + + state.destination/state.source are defaultdict(dict), so indexing an unknown + resource_to_connect returns {} rather than raising -- plain `in` membership checks + are safe and no KeyError guard is needed. An exception raised by + ensure_resource_loaded (e.g. a storage error) propagates uncaught, matching every + other ensure_resource_loaded call site. + """ + destination = self.config.state.destination[resource_to_connect] + if plain_id in destination: + return destination[plain_id]["id"], False + + self.config.state.ensure_resource_loaded(resource_to_connect, plain_id) + if plain_id in self.config.state.source[resource_to_connect]: + return None, False + return None, True + + # Composite "type:id" principal type prefixes -> the resource_to_connect they map to. + _PRINCIPAL_TYPE_MAP: ClassVar[Dict[str, str]] = {"user": "users", "role": "roles", "team": "teams"} + + def _filter_stale_binding_principals( + self, _id: str, bindings: Optional[List[Dict]] + ) -> Tuple[Dict[str, List[str]], bool]: + """Resolve/drop/hard-fail composite "type:id" principals across restriction-policy + bindings (shared by restriction_policies / monitors / synthetics_tests). + + For each principal in each binding: resolve against destination (remap in place), + else consult source. Source-present ("not yet synced") or flag-off -> hard-fail + (add to failed_connections). Source-absent ("permanently gone") AND + --drop-unresolvable-principals -> drop it, WARN, and count it. + + Rebuilds each binding's "principals" as a NEW list (never del/pop/index-assign + during iteration) to avoid the enumerate/index-shift skip bug. Returns + (failed_connections_dict, empty_binding_risk) where empty_binding_risk is True if + any binding whose source list was non-empty became empty after dropping. + """ + failed_connections_dict: Dict[str, List[str]] = defaultdict(list) + empty_binding_risk = False + for binding in bindings or []: + principals = binding.get("principals") + if not principals: + continue + had_principals = len(principals) > 0 + kept: List[str] = [] + for policy_id in principals: + parts = policy_id.split(":", 1) + if len(parts) != 2: + kept.append(policy_id) + continue + _type, plain_id = parts + resource_to_connect = self._PRINCIPAL_TYPE_MAP.get(_type) + if resource_to_connect is None: + # org: (already remapped by pre_resource_action_hook) or any other + # non-user/role/team principal -> pass through untouched. + kept.append(policy_id) + continue + resolved, stale = self._resolve_or_drop(plain_id, resource_to_connect) + if resolved is not None: + kept.append(f"{_type}:{resolved}") + elif stale and self.config.drop_unresolvable_principals: + self.config.logger.warning( + f"dropping stale principal '{policy_id}': absent from source and " + "destination (likely deleted before the org's first sync)", + resource_type=self.resource_type, + _id=_id, + ) + if self.config.counter is not None: + self.config.counter.record_stale_principal_dropped(resource_type=self.resource_type, _id=_id) + else: + # Source-present ("not yet synced", retry later) or flag off: + # unchanged hard-fail semantics -- keep the original id and record it. + kept.append(policy_id) + failed_connections_dict[resource_to_connect].append(plain_id) + binding["principals"] = kept + if had_principals and not kept: + empty_binding_risk = True + return failed_connections_dict, empty_binding_risk + + def _filter_stale_flat_roles(self, _id: str, container: Optional[Dict], key: str) -> Tuple[List[str], bool]: + """Resolve/drop/hard-fail a flat list of role ids at container[key] + (restricted_roles / options.restricted_roles / metadata.restricted_roles). + + Same three-way logic as _filter_stale_binding_principals but for plain role ids + (no "type:" prefix). Rebuilds the list as a NEW list. Returns + (failed_role_ids, empty_list_risk) where empty_list_risk is True if a non-empty + source list became empty after dropping. + """ + failed: List[str] = [] + if not container: + return failed, False + roles = container.get(key) + if not roles: + return failed, False + had_roles = len(roles) > 0 + kept: List[str] = [] + for role in roles: + plain_id = str(role) + resolved, stale = self._resolve_or_drop(plain_id, "roles") + if resolved is not None: + kept.append(type(role)(resolved)) + elif stale and self.config.drop_unresolvable_principals: + self.config.logger.warning( + f"dropping stale role '{plain_id}' from {key}: absent from source and " + "destination (likely deleted before the org's first sync)", + resource_type=self.resource_type, + _id=_id, + ) + if self.config.counter is not None: + self.config.counter.record_stale_principal_dropped(resource_type=self.resource_type, _id=_id) + else: + kept.append(role) + failed.append(plain_id) + container[key] = kept + return failed, (had_roles and not kept) + + def _raise_connection_error_if_any( + self, _id: str, failed_connections_dict: Dict[str, List[str]], empty_binding_risk: bool = False + ) -> None: + """Shared terminal step for the drop-aware connect_resources overrides. + + Mirrors the base connect_resources raise/skip behavior: raise unless + --skip-failed-resource-connections is set. empty_binding_risk is threaded onto the + exception so _apply_resource_cb can tag the metric, and is logged at ERROR because + an empty binding is an access-elevation risk. + """ + if not failed_connections_dict and not empty_binding_risk: + return + e = ResourceConnectionError( + failed_connections_dict=failed_connections_dict, empty_binding_risk=empty_binding_risk + ) + if empty_binding_risk: + self.config.logger.error( + "access-elevation risk: a binding/list whose source had principals became " + "empty after dropping unresolvable references; refusing to sync", + resource_type=self.resource_type, + _id=_id, + ) + if not self.config.skip_failed_resource_connections: + self.config.logger.info( + f"skipping resource: {str(e)}", + _id=_id, + resource_type=self.resource_type, + ) + raise e + else: + self.config.logger.debug(f"{str(e)}", _id=_id, resource_type=self.resource_type) + def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]: """Extract dependency IDs referenced at r_obj[key] for resource_to_connect. diff --git a/datadog_sync/utils/configuration.py b/datadog_sync/utils/configuration.py index f91e95918..7050a72c7 100644 --- a/datadog_sync/utils/configuration.py +++ b/datadog_sync/utils/configuration.py @@ -8,7 +8,7 @@ import logging import sys import time -from typing import Any, Optional, Union, Dict, List +from typing import Any, Optional, TYPE_CHECKING, Union, Dict, List import click @@ -47,6 +47,9 @@ from datadog_sync.utils.state import State from datadog_sync.utils.storage.storage_types import StorageType +if TYPE_CHECKING: + from datadog_sync.utils.workers import Counter + @dataclass class Configuration(object): @@ -77,6 +80,16 @@ class Configuration(object): allow_self_lockout: bool datadog_host_override: Optional[str] = None emit_json: bool = False + # Opt-in: drop principal/role references that are absent from BOTH destination and + # source state (permanently gone -- e.g. deleted before this org's first-ever import) + # instead of hard-failing the whole resource. When False (default) behavior is + # byte-for-byte unchanged. Consumed by the restriction_policies/monitors/ + # synthetics_tests/dashboards/synthetics_private_locations models' connect logic. + drop_unresolvable_principals: bool = False + # Set at runtime by Workers.__init__ so resource models (which only hold self.config) + # can record dropped stale principals into the apply-run Counter from inside + # connect_id/connect_resources. None until a Workers is constructed for the command. + counter: Optional["Counter"] = None # Opt-in refresh of state.destination from storage before apply_resources # dispatches workers. Motivating case: an external orchestrator runs # sync-cli once per resource type as separate processes reading a shared @@ -407,6 +420,7 @@ def build_config(cmd: Command, **kwargs: Optional[Any]) -> Configuration: # Additional settings force_missing_dependencies = kwargs.get("force_missing_dependencies") or False skip_failed_resource_connections = kwargs.get("skip_failed_resource_connections") + drop_unresolvable_principals = kwargs.get("drop_unresolvable_principals") or False refresh_destination_state_before_apply = kwargs.get("refresh_destination_state_before_apply") or False max_workers = kwargs.get("max_workers") max_workers_per_type_raw = kwargs.get("max_workers_per_type") @@ -660,6 +674,7 @@ def build_config(cmd: Command, **kwargs: Optional[Any]) -> Configuration: filter_operator=filter_operator, force_missing_dependencies=force_missing_dependencies, skip_failed_resource_connections=skip_failed_resource_connections, + drop_unresolvable_principals=drop_unresolvable_principals, refresh_destination_state_before_apply=refresh_destination_state_before_apply, max_workers=max_workers, max_workers_per_type=max_workers_per_type, diff --git a/datadog_sync/utils/resource_utils.py b/datadog_sync/utils/resource_utils.py index 800d2dbb0..aa550b784 100644 --- a/datadog_sync/utils/resource_utils.py +++ b/datadog_sync/utils/resource_utils.py @@ -78,8 +78,13 @@ def __init__(self, _id: str, _type: str): class ResourceConnectionError(Exception): - def __init__(self, failed_connections_dict): + def __init__(self, failed_connections_dict, empty_binding_risk: bool = False): super(ResourceConnectionError, self).__init__(f"Failed to connect resource. {dict(failed_connections_dict)}") + # empty_binding_risk flags the access-elevation case: a restriction-policy + # binding (or a restricted_roles list) whose source-side list was non-empty + # but became empty after dropping unresolvable principals. Read by + # ResourcesHandler._apply_resource_cb to add a distinguishing metric tag. + self.empty_binding_risk = empty_binding_risk class CustomClientHTTPError(Exception): diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py index 752b7392d..27de8b90e 100644 --- a/datadog_sync/utils/resources_handler.py +++ b/datadog_sync/utils/resources_handler.py @@ -78,6 +78,30 @@ def _chunked_emit(rt: str, action_desc: str, ids: List[str]) -> None: ids = counter.skipped_missing_deps_by_type[rt] if ids: _chunked_emit(rt, "skipped for missing dependencies", ids) + # --drop-unresolvable-principals aggregate signals. Both dicts stay empty when the flag + # is off, so these lines never appear in the default path. getattr guards Counters + # constructed before these buckets existed. + for rt in sorted(getattr(counter, "stale_principals_dropped_by_type", {}).keys()): + ids = counter.stale_principals_dropped_by_type[rt] + if ids: + _chunked_emit(rt, "dropped stale principals", ids) + for rt in sorted(getattr(counter, "empty_binding_risk_by_type", {}).keys()): + ids = counter.empty_binding_risk_by_type[rt] + if ids: + total = len(ids) + for start in range(0, total, _SUMMARY_ID_CHUNK): + chunk = ids[start:start + _SUMMARY_ID_CHUNK] + end = min(start + _SUMMARY_ID_CHUNK, total) + logger.error( + "sync summary: %s skipped %d resource(s) for empty-binding " + "access-elevation risk [%d-%d of %d]: %s", + rt, + total, + start + 1, + end, + total, + ", ".join(chunk), + ) def _list_time_filter_passes(r_class, config, resource) -> bool: @@ -502,9 +526,15 @@ async def _apply_resource_cb(self, q_item: List) -> None: ) _reason, _fc = self._sanitize_reason(e) self._emit(resource_type, _id, "sync", "skipped", reason=_reason, failure_class=_fc) - await r_class._send_action_metrics( - Command.SYNC.value, _id, Status.SKIPPED.value, tags=["reason:connection_error"] - ) + # Distinguish the access-elevation case (a restriction-policy binding / + # restricted_roles list that emptied out after dropping stale principals) with + # a dedicated metric tag and a separate counter bucket for the end-of-run + # summary. getattr guards ResourceConnectionError instances without the attr. + extra_tags = ["reason:connection_error"] + if getattr(e, "empty_binding_risk", False): + extra_tags.append("risk:empty_restriction_policy") + self.worker.counter.record_empty_binding_risk(resource_type=resource_type, _id=_id) + await r_class._send_action_metrics(Command.SYNC.value, _id, Status.SKIPPED.value, tags=extra_tags) except Exception as e: # Track the source id in the counter's failed-ids bucket so # apply_resources can emit a targeted per-type summary at the diff --git a/datadog_sync/utils/workers.py b/datadog_sync/utils/workers.py index f67dea664..8a490e703 100644 --- a/datadog_sync/utils/workers.py +++ b/datadog_sync/utils/workers.py @@ -23,6 +23,10 @@ def __init__(self, config: Configuration) -> None: self.workers: List[Task] = [] self.work_queue: Queue = Queue() self.counter: Counter = Counter() + # Back-reference so resource models (which only hold self.config) can record + # dropped stale principals into the counter from inside connect_id/connect_resources. + # Last-writer-wins is fine: only one Workers drives a given command's apply pass. + config.counter = self.counter self.pbar: Optional[tqdm] = None self._running_workers_count: int = 0 self._loop: AbstractEventLoop = get_event_loop() @@ -119,6 +123,12 @@ class Counter: # actionable if the roles-sync run logged that X, Y, Z failed. failed_ids_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) skipped_missing_deps_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) + # Per-resource-type source IDs whose stale (source-absent) principal/role + # references were dropped from an access-control list, and resource IDs skipped + # because dropping left an empty binding/list (access-elevation risk). Populated + # only when --drop-unresolvable-principals is set. Surfaced by _emit_apply_summary. + stale_principals_dropped_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) + empty_binding_risk_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) def __str__(self): return ( @@ -129,6 +139,8 @@ def reset_counter(self) -> None: self.successes = self.failure = self.skipped = self.filtered = 0 self.failed_ids_by_type = defaultdict(list) self.skipped_missing_deps_by_type = defaultdict(list) + self.stale_principals_dropped_by_type = defaultdict(list) + self.empty_binding_risk_by_type = defaultdict(list) def increment_success(self) -> None: self.successes += 1 @@ -141,11 +153,26 @@ def increment_failure(self, resource_type: Optional[str] = None, _id: Optional[s if resource_type is not None and _id is not None: self.failed_ids_by_type[resource_type].append(str(_id)) - def increment_skipped(self, resource_type: Optional[str] = None, _id: Optional[str] = None, - missing_deps: bool = False) -> None: + def increment_skipped( + self, resource_type: Optional[str] = None, _id: Optional[str] = None, missing_deps: bool = False + ) -> None: self.skipped += 1 if missing_deps and resource_type is not None and _id is not None: self.skipped_missing_deps_by_type[resource_type].append(str(_id)) def increment_filtered(self) -> None: self.filtered += 1 + + # Two dedicated methods rather than the single-multi-purpose-method convention used by + # increment_skipped/increment_failure: the "dropped a stale principal but kept syncing" + # and "skipped for empty-binding access-elevation risk" cases are semantically distinct + # enough (one is a non-fatal drop, the other a hard-fail security signal) to warrant + # separate names. This deliberate deviation is called out here so it isn't flagged as + # an inconsistency. + def record_stale_principal_dropped(self, resource_type: Optional[str] = None, _id: Optional[str] = None) -> None: + if resource_type is not None and _id is not None: + self.stale_principals_dropped_by_type[resource_type].append(str(_id)) + + def record_empty_binding_risk(self, resource_type: Optional[str] = None, _id: Optional[str] = None) -> None: + if resource_type is not None and _id is not None: + self.empty_binding_risk_by_type[resource_type].append(str(_id)) diff --git a/tests/unit/test_apply_resource_connection_error.py b/tests/unit/test_apply_resource_connection_error.py index 535c7b2d5..4c477bd43 100644 --- a/tests/unit/test_apply_resource_connection_error.py +++ b/tests/unit/test_apply_resource_connection_error.py @@ -68,3 +68,37 @@ def test_resource_connection_error_still_increments_skipped(mock_config): emit_args, emit_kwargs = handler._emit.call_args assert emit_args[:4] == ("dashboards", "dash-1", "sync", "skipped") assert "reason" in emit_kwargs + + +def _metric_tags(r_class): + """Extract the tags passed to the (async) _send_action_metrics call.""" + _args, kwargs = r_class._send_action_metrics.call_args + return kwargs.get("tags", []) + + +def test_empty_binding_risk_adds_metric_tag_and_records(mock_config): + """empty_binding_risk=True must add the risk tag AND bump the dedicated counter.""" + exc = ResourceConnectionError({"roles": ["role-gone"]}, empty_binding_risk=True) + + handler = _drive_apply_cb(mock_config, "restriction_policies", "dashboard:dash-1", exc) + + r_class = mock_config.resources["restriction_policies"] + tags = _metric_tags(r_class) + assert "reason:connection_error" in tags + assert "risk:empty_restriction_policy" in tags + handler.worker.counter.record_empty_binding_risk.assert_called_once_with( + resource_type="restriction_policies", _id="dashboard:dash-1" + ) + + +def test_ordinary_connection_error_has_no_risk_tag(mock_config): + """Without empty_binding_risk the risk tag is absent and the counter is untouched.""" + exc = ResourceConnectionError({"roles": ["role-not-yet-synced"]}) + + handler = _drive_apply_cb(mock_config, "restriction_policies", "dashboard:xyz", exc) + + r_class = mock_config.resources["restriction_policies"] + tags = _metric_tags(r_class) + assert "reason:connection_error" in tags + assert "risk:empty_restriction_policy" not in tags + handler.worker.counter.record_empty_binding_risk.assert_not_called() diff --git a/tests/unit/test_apply_summary_handler.py b/tests/unit/test_apply_summary_handler.py index 45ea3b7e5..cfaa22990 100644 --- a/tests/unit/test_apply_summary_handler.py +++ b/tests/unit/test_apply_summary_handler.py @@ -36,6 +36,19 @@ def _rendered_warnings(logger): return out +def _rendered_errors(logger): + out = [] + for call in logger.error.call_args_list: + args = call.args + if not args: + continue + try: + out.append(args[0] % args[1:]) + except Exception: + out.append(str(args)) + return out + + def test_summary_emits_failed_ids_by_type(): counter = Counter() counter.increment_failure(resource_type="roles", _id="uuid-abc") @@ -123,6 +136,41 @@ def test_summary_uses_join_not_repr_for_greppability(): assert "sync summary" in joined +def test_summary_emits_stale_principals_dropped_at_warning(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="role:role-uuid-1") + + logger = _make_logger() + _emit_apply_summary(logger, counter) + + joined = "\n".join(_rendered_warnings(logger)) + assert "restriction_policies" in joined + assert "role:role-uuid-1" in joined + assert "dropped" in joined and "stale principal" in joined + + +def test_summary_emits_empty_binding_risk_at_error(): + counter = Counter() + counter.record_empty_binding_risk(resource_type="restriction_policies", _id="dashboard:dash-1") + + logger = _make_logger() + _emit_apply_summary(logger, counter) + + joined_err = "\n".join(_rendered_errors(logger)) + assert "restriction_policies" in joined_err + assert "dashboard:dash-1" in joined_err + assert "empty-binding" in joined_err and "access-elevation" in joined_err + + +def test_summary_silent_for_new_buckets_when_empty(): + counter = Counter() + logger = _make_logger() + _emit_apply_summary(logger, counter) + # No stale/empty-binding activity -> neither WARNING nor ERROR lines emitted. + assert logger.warning.call_count == 0 + assert logger.error.call_count == 0 + + def test_chunk_constant_is_reasonable(): assert 20 <= _SUMMARY_ID_CHUNK <= 200 diff --git a/tests/unit/test_base_resource.py b/tests/unit/test_base_resource.py new file mode 100644 index 000000000..899fbacd7 --- /dev/null +++ b/tests/unit/test_base_resource.py @@ -0,0 +1,89 @@ +# 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. + +"""Unit tests for BaseResource._resolve_or_drop. + +_resolve_or_drop is the shared three-way resolver behind --drop-unresolvable-principals: +it decides whether a plain dependency id resolves against destination state, is a +"not yet synced" source-present miss, or is a permanently-gone (source-absent) stale +reference. It reads only self.config, so we exercise it as an unbound method with a +stub self rather than instantiating the abstract BaseResource. +""" + +from collections import defaultdict +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from datadog_sync.utils.base_resource import BaseResource + + +def _make_config(): + config = MagicMock() + state = MagicMock() + state.source = defaultdict(dict) + state.destination = defaultdict(dict) + state.ensure_resource_loaded = MagicMock() + config.state = state + return config + + +def _call(config, plain_id, resource_to_connect): + stub = SimpleNamespace(config=config) + return BaseResource._resolve_or_drop(stub, plain_id, resource_to_connect) + + +def test_resolve_or_drop_destination_present_returns_dest_id_not_stale(): + config = _make_config() + config.state.destination["roles"]["src-role"] = {"id": "dst-role"} + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved == "dst-role" + assert stale is False + # No need to consult source when destination already has it. + config.state.ensure_resource_loaded.assert_not_called() + + +def test_resolve_or_drop_source_present_not_synced_returns_none_not_stale(): + config = _make_config() + config.state.source["roles"]["src-role"] = {"id": "src-role"} + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved is None + assert stale is False # legitimate "not yet synced" -> caller hard-fails, does NOT drop + config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role") + + +def test_resolve_or_drop_absent_from_both_is_permanently_stale(): + config = _make_config() + + resolved, stale = _call(config, "ghost-role", "roles") + + assert resolved is None + assert stale is True # absent from destination AND source -> permanently gone + config.state.ensure_resource_loaded.assert_called_once_with("roles", "ghost-role") + + +def test_resolve_or_drop_unknown_resource_type_does_not_raise(): + # state.destination/source are defaultdict(dict); an unknown type yields {} not KeyError. + config = _make_config() + + resolved, stale = _call(config, "whatever", "never_seen_type") + + assert resolved is None + assert stale is True + + +def test_resolve_or_drop_propagates_ensure_resource_loaded_exception(): + config = _make_config() + config.state.ensure_resource_loaded.side_effect = RuntimeError("storage down") + + # Not in destination -> ensure_resource_loaded is called -> its error must propagate, + # matching every other ensure_resource_loaded call site (none catch it). + with pytest.raises(RuntimeError, match="storage down"): + _call(config, "src-role", "roles") diff --git a/tests/unit/test_drop_unresolvable_principals_cli.py b/tests/unit/test_drop_unresolvable_principals_cli.py new file mode 100644 index 000000000..1dc11f690 --- /dev/null +++ b/tests/unit/test_drop_unresolvable_principals_cli.py @@ -0,0 +1,43 @@ +# 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. + +"""CLI round-trip test for --drop-unresolvable-principals. + +Mirrors test_force_missing_dependencies_cli.py. The flag lives in _diffs_options, +which is attached to sync, diffs, and migrate (the three commands that run +connect_resources). exit_code == 2 is click's "no such option" usage error, so +exit_code != 2 proves the option is recognized by each command. +""" + +import pytest +from click.testing import CliRunner + +from datadog_sync.cli import cli + + +@pytest.fixture +def runner(): + return CliRunner(mix_stderr=False) + + +def test_sync_accepts_drop_unresolvable_principals(runner): + result = runner.invoke(cli, ["sync", "--drop-unresolvable-principals", "--validate=false"]) + assert result.exit_code != 2 + + +def test_diffs_accepts_drop_unresolvable_principals(runner): + result = runner.invoke(cli, ["diffs", "--drop-unresolvable-principals"]) + assert result.exit_code != 2 + + +def test_migrate_accepts_drop_unresolvable_principals(runner): + result = runner.invoke(cli, ["migrate", "--drop-unresolvable-principals", "--validate=false"]) + assert result.exit_code != 2 + + +def test_import_rejects_drop_unresolvable_principals(runner): + # The flag is NOT attached to import (import does not run connect_resources). + result = runner.invoke(cli, ["import", "--drop-unresolvable-principals"]) + assert result.exit_code == 2 diff --git a/tests/unit/test_resource_utils.py b/tests/unit/test_resource_utils.py index c27803563..ca2431155 100644 --- a/tests/unit/test_resource_utils.py +++ b/tests/unit/test_resource_utils.py @@ -7,7 +7,19 @@ from unittest.mock import MagicMock, call from datadog_sync import models -from datadog_sync.utils.resource_utils import find_attr +from datadog_sync.utils.resource_utils import find_attr, ResourceConnectionError + + +def test_resource_connection_error_empty_binding_risk_defaults_false(): + exc = ResourceConnectionError({"roles": ["r1"]}) + assert exc.empty_binding_risk is False + + +def test_resource_connection_error_empty_binding_risk_settable(): + exc = ResourceConnectionError({"roles": ["r1"]}, empty_binding_risk=True) + assert exc.empty_binding_risk is True + # Message formatting is unchanged by the new kwarg. + assert "Failed to connect resource." in str(exc) @pytest.fixture(scope="class") diff --git a/tests/unit/test_workers.py b/tests/unit/test_workers.py new file mode 100644 index 000000000..e001b2249 --- /dev/null +++ b/tests/unit/test_workers.py @@ -0,0 +1,60 @@ +# 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. + +"""Unit tests for the Counter buckets added for --drop-unresolvable-principals. + +Guards two things: (1) the new fields default empty and record correctly, and +(2) reset_counter() zeroes them so repeated import/sync invocations in one process +don't double-count. +""" + +from datadog_sync.utils.workers import Counter + + +def test_new_counter_fields_default_empty(): + counter = Counter() + assert counter.stale_principals_dropped_by_type == {} + assert counter.empty_binding_risk_by_type == {} + + +def test_record_stale_principal_dropped_appends_by_type(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="dashboard:abc") + counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="dashboard:def") + counter.record_stale_principal_dropped(resource_type="monitors", _id="mon-1") + + assert counter.stale_principals_dropped_by_type["restriction_policies"] == [ + "dashboard:abc", + "dashboard:def", + ] + assert counter.stale_principals_dropped_by_type["monitors"] == ["mon-1"] + + +def test_record_empty_binding_risk_appends_by_type(): + counter = Counter() + counter.record_empty_binding_risk(resource_type="restriction_policies", _id="dashboard:dash-1") + + assert counter.empty_binding_risk_by_type["restriction_policies"] == ["dashboard:dash-1"] + + +def test_record_methods_ignore_none_args(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type=None, _id="x") + counter.record_stale_principal_dropped(resource_type="roles", _id=None) + counter.record_empty_binding_risk(resource_type=None, _id=None) + + assert counter.stale_principals_dropped_by_type == {} + assert counter.empty_binding_risk_by_type == {} + + +def test_reset_counter_clears_new_fields(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type="roles", _id="r1") + counter.record_empty_binding_risk(resource_type="roles", _id="r2") + + counter.reset_counter() + + assert counter.stale_principals_dropped_by_type == {} + assert counter.empty_binding_risk_by_type == {} From 5fba7a7cf13fffd7599d834e2b963dd36bc8149a Mon Sep 17 00:00:00 2001 From: michael-richey Date: Wed, 15 Jul 2026 08:37:10 -0400 Subject: [PATCH 2/4] fix: report suppressed empty-binding connection risks --- datadog_sync/commands/shared/options.py | 4 +- datadog_sync/utils/base_resource.py | 33 +++++++++++----- datadog_sync/utils/resources_handler.py | 38 +++++++++++++++++-- datadog_sync/utils/workers.py | 8 ++++ .../test_apply_resource_connection_error.py | 29 ++++++++++++++ tests/unit/test_apply_summary_handler.py | 14 +++++++ tests/unit/test_base_resource.py | 31 +++++++++++++++ tests/unit/test_workers.py | 12 ++++++ 8 files changed, 155 insertions(+), 14 deletions(-) diff --git a/datadog_sync/commands/shared/options.py b/datadog_sync/commands/shared/options.py index cac31365d..a519b6ec7 100644 --- a/datadog_sync/commands/shared/options.py +++ b/datadog_sync/commands/shared/options.py @@ -509,7 +509,9 @@ def click_config_file_provider(ctx: Context, opts: CustomOptionClass, value: Non "references that are absent from BOTH destination and source state (permanently " "gone, e.g. deleted before the org's first import) instead of skipping the whole " "resource. If dropping empties a binding/list that had entries at the source, the " - "resource is still skipped (access-elevation guard). Off by default.", + "resource connection fails normally. --skip-failed-resource-connections may suppress " + "that failure and continue syncing; an ERROR log and risk metric explicitly warn that " + "the destination resource may be unrestricted. Off by default.", cls=CustomOptionClass, ), option( diff --git a/datadog_sync/utils/base_resource.py b/datadog_sync/utils/base_resource.py index 3b781bcb4..ff6cd44af 100644 --- a/datadog_sync/utils/base_resource.py +++ b/datadog_sync/utils/base_resource.py @@ -574,26 +574,38 @@ def _filter_stale_flat_roles(self, _id: str, container: Optional[Dict], key: str def _raise_connection_error_if_any( self, _id: str, failed_connections_dict: Dict[str, List[str]], empty_binding_risk: bool = False - ) -> None: + ) -> bool: """Shared terminal step for the drop-aware connect_resources overrides. Mirrors the base connect_resources raise/skip behavior: raise unless --skip-failed-resource-connections is set. empty_binding_risk is threaded onto the - exception so _apply_resource_cb can tag the metric, and is logged at ERROR because - an empty binding is an access-elevation risk. + exception so _apply_resource_cb can tag skipped-resource metrics. When + --skip-failed-resource-connections suppresses the exception, returns True for an + empty-binding risk so the apply path can tag the successful action and include it in + the end-of-run escalation summary. Otherwise returns False. """ if not failed_connections_dict and not empty_binding_risk: - return + return False e = ResourceConnectionError( failed_connections_dict=failed_connections_dict, empty_binding_risk=empty_binding_risk ) if empty_binding_risk: - self.config.logger.error( - "access-elevation risk: a binding/list whose source had principals became " - "empty after dropping unresolvable references; refusing to sync", - resource_type=self.resource_type, - _id=_id, - ) + if self.config.skip_failed_resource_connections: + self.config.logger.error( + "access-elevation risk: a binding/list whose source had principals became " + "empty after dropping unresolvable references; " + "--skip-failed-resource-connections is enabled, continuing sync. " + "DESTINATION RESOURCE MAY BE UNRESTRICTED", + resource_type=self.resource_type, + _id=_id, + ) + else: + self.config.logger.error( + "access-elevation risk: a binding/list whose source had principals became " + "empty after dropping unresolvable references; refusing to sync", + resource_type=self.resource_type, + _id=_id, + ) if not self.config.skip_failed_resource_connections: self.config.logger.info( f"skipping resource: {str(e)}", @@ -603,6 +615,7 @@ def _raise_connection_error_if_any( raise e else: self.config.logger.debug(f"{str(e)}", _id=_id, resource_type=self.resource_type) + return empty_binding_risk def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]: """Extract dependency IDs referenced at r_obj[key] for resource_to_connect. diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py index 27de8b90e..a1333d47b 100644 --- a/datadog_sync/utils/resources_handler.py +++ b/datadog_sync/utils/resources_handler.py @@ -102,6 +102,24 @@ def _chunked_emit(rt: str, action_desc: str, ids: List[str]) -> None: total, ", ".join(chunk), ) + for rt in sorted(getattr(counter, "empty_binding_escalation_by_type", {}).keys()): + ids = counter.empty_binding_escalation_by_type[rt] + if ids: + total = len(ids) + for start in range(0, total, _SUMMARY_ID_CHUNK): + chunk = ids[start:start + _SUMMARY_ID_CHUNK] + end = min(start + _SUMMARY_ID_CHUNK, total) + logger.error( + "sync summary: %s synced %d resource(s) after an empty-binding " + "connection failure was suppressed [%d-%d of %d]: %s. " + "DESTINATION RESOURCE MAY BE UNRESTRICTED", + rt, + total, + start + 1, + end, + total, + ", ".join(chunk), + ) def _list_time_filter_passes(r_class, config, resource) -> bool: @@ -475,7 +493,11 @@ async def _apply_resource_cb(self, q_item: List) -> None: # Run hooks await r_class._pre_resource_action_hook(_id, resource) - r_class.connect_resources(_id, resource) + empty_binding_escalation = bool(r_class.connect_resources(_id, resource)) + + success_metric_tags = [] + if empty_binding_escalation: + success_metric_tags.append("risk:empty_restriction_policy") prep_resource(r_class.resource_config, resource) if _id in self.config.state.destination[resource_type]: @@ -487,16 +509,26 @@ async def _apply_resource_cb(self, q_item: List) -> None: self.config.logger.debug(f"Running update for {resource_type} with {_id}") await r_class._update_resource(_id, resource) + if empty_binding_escalation: + self.worker.counter.record_empty_binding_escalation(resource_type=resource_type, _id=_id) await r_class._send_action_metrics( - Command.SYNC.value, _id, Status.SUCCESS.value, tags=["action_sub_type:update"] + Command.SYNC.value, + _id, + Status.SUCCESS.value, + tags=["action_sub_type:update", *success_metric_tags], ) self.config.logger.debug(f"Finished update for {resource_type} with {_id}") self._emit(resource_type, _id, "sync", "success", "update") else: self.config.logger.debug(f"Running create for {resource_type} with id: {_id}") await r_class._create_resource(_id, resource) + if empty_binding_escalation: + self.worker.counter.record_empty_binding_escalation(resource_type=resource_type, _id=_id) await r_class._send_action_metrics( - Command.SYNC.value, _id, Status.SUCCESS.value, tags=["action_sub_type:create"] + Command.SYNC.value, + _id, + Status.SUCCESS.value, + tags=["action_sub_type:create", *success_metric_tags], ) self.config.logger.debug(f"finished create for {resource_type} with id: {_id}") self._emit(resource_type, _id, "sync", "success", "create") diff --git a/datadog_sync/utils/workers.py b/datadog_sync/utils/workers.py index 8a490e703..35704899a 100644 --- a/datadog_sync/utils/workers.py +++ b/datadog_sync/utils/workers.py @@ -129,6 +129,7 @@ class Counter: # only when --drop-unresolvable-principals is set. Surfaced by _emit_apply_summary. stale_principals_dropped_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) empty_binding_risk_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) + empty_binding_escalation_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) def __str__(self): return ( @@ -141,6 +142,7 @@ def reset_counter(self) -> None: self.skipped_missing_deps_by_type = defaultdict(list) self.stale_principals_dropped_by_type = defaultdict(list) self.empty_binding_risk_by_type = defaultdict(list) + self.empty_binding_escalation_by_type = defaultdict(list) def increment_success(self) -> None: self.successes += 1 @@ -176,3 +178,9 @@ def record_stale_principal_dropped(self, resource_type: Optional[str] = None, _i def record_empty_binding_risk(self, resource_type: Optional[str] = None, _id: Optional[str] = None) -> None: if resource_type is not None and _id is not None: self.empty_binding_risk_by_type[resource_type].append(str(_id)) + + def record_empty_binding_escalation( + self, resource_type: Optional[str] = None, _id: Optional[str] = None + ) -> None: + if resource_type is not None and _id is not None: + self.empty_binding_escalation_by_type[resource_type].append(str(_id)) diff --git a/tests/unit/test_apply_resource_connection_error.py b/tests/unit/test_apply_resource_connection_error.py index 4c477bd43..1c5d4eec4 100644 --- a/tests/unit/test_apply_resource_connection_error.py +++ b/tests/unit/test_apply_resource_connection_error.py @@ -14,6 +14,7 @@ import asyncio from unittest.mock import AsyncMock, MagicMock +from datadog_sync.utils.base_resource import ResourceConfig from datadog_sync.utils.resource_utils import ResourceConnectionError from datadog_sync.utils.resources_handler import ResourcesHandler @@ -102,3 +103,31 @@ def test_ordinary_connection_error_has_no_risk_tag(mock_config): assert "reason:connection_error" in tags assert "risk:empty_restriction_policy" not in tags handler.worker.counter.record_empty_binding_risk.assert_not_called() + + +def test_successful_update_after_suppressed_empty_binding_risk_is_tagged_and_recorded(mock_config): + resource_type = "restriction_policies" + _id = "dashboard:dash-1" + r_class = MagicMock() + r_class.resource_config = ResourceConfig(base_path="", skip_resource_mapping=True) + r_class.connect_resources.return_value = True + r_class._pre_resource_action_hook = AsyncMock() + r_class._update_resource = AsyncMock() + r_class._send_action_metrics = AsyncMock() + mock_config.resources = {resource_type: r_class} + mock_config.state.source[resource_type][_id] = {"id": _id, "bindings": []} + mock_config.state.destination[resource_type][_id] = {"id": _id, "bindings": [{"relation": "editor"}]} + + handler = ResourcesHandler(mock_config) + handler.worker = MagicMock() + handler.worker.counter = MagicMock() + handler.sorter = MagicMock() + handler._emit = MagicMock() + + asyncio.run(handler._apply_resource_cb([resource_type, _id])) + + assert "risk:empty_restriction_policy" in _metric_tags(r_class) + handler.worker.counter.record_empty_binding_escalation.assert_called_once_with( + resource_type=resource_type, _id=_id + ) + handler.worker.counter.record_empty_binding_risk.assert_not_called() diff --git a/tests/unit/test_apply_summary_handler.py b/tests/unit/test_apply_summary_handler.py index cfaa22990..1290e58f0 100644 --- a/tests/unit/test_apply_summary_handler.py +++ b/tests/unit/test_apply_summary_handler.py @@ -162,6 +162,20 @@ def test_summary_emits_empty_binding_risk_at_error(): assert "empty-binding" in joined_err and "access-elevation" in joined_err +def test_summary_emits_applied_empty_binding_escalation_at_error(): + counter = Counter() + counter.record_empty_binding_escalation(resource_type="restriction_policies", _id="dashboard:dash-1") + + logger = _make_logger() + _emit_apply_summary(logger, counter) + + joined_err = "\n".join(_rendered_errors(logger)) + assert "restriction_policies" in joined_err + assert "dashboard:dash-1" in joined_err + assert "connection failure was suppressed" in joined_err + assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" in joined_err + + def test_summary_silent_for_new_buckets_when_empty(): counter = Counter() logger = _make_logger() diff --git a/tests/unit/test_base_resource.py b/tests/unit/test_base_resource.py index 899fbacd7..ea4dcaf02 100644 --- a/tests/unit/test_base_resource.py +++ b/tests/unit/test_base_resource.py @@ -87,3 +87,34 @@ def test_resolve_or_drop_propagates_ensure_resource_loaded_exception(): # matching every other ensure_resource_loaded call site (none catch it). with pytest.raises(RuntimeError, match="storage down"): _call(config, "src-role", "roles") + + +def test_empty_binding_risk_continues_with_explicit_escalation_warning_when_connection_failures_are_skipped(): + config = _make_config() + config.skip_failed_resource_connections = True + config.logger = MagicMock() + stub = SimpleNamespace(config=config, resource_type="restriction_policies") + + escalation_risk = BaseResource._raise_connection_error_if_any(stub, "dashboard:abc", {}, True) + + assert escalation_risk is True + message = config.logger.error.call_args.args[0] + assert "continuing sync" in message + assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" in message + assert "refusing to sync" not in message + + +def test_empty_binding_risk_refuses_to_sync_when_connection_failures_are_not_skipped(): + from datadog_sync.utils.resource_utils import ResourceConnectionError + + config = _make_config() + config.skip_failed_resource_connections = False + config.logger = MagicMock() + stub = SimpleNamespace(config=config, resource_type="restriction_policies") + + with pytest.raises(ResourceConnectionError): + BaseResource._raise_connection_error_if_any(stub, "dashboard:abc", {}, True) + + message = config.logger.error.call_args.args[0] + assert "refusing to sync" in message + assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" not in message diff --git a/tests/unit/test_workers.py b/tests/unit/test_workers.py index e001b2249..bf923ac46 100644 --- a/tests/unit/test_workers.py +++ b/tests/unit/test_workers.py @@ -17,6 +17,7 @@ def test_new_counter_fields_default_empty(): counter = Counter() assert counter.stale_principals_dropped_by_type == {} assert counter.empty_binding_risk_by_type == {} + assert counter.empty_binding_escalation_by_type == {} def test_record_stale_principal_dropped_appends_by_type(): @@ -39,22 +40,33 @@ def test_record_empty_binding_risk_appends_by_type(): assert counter.empty_binding_risk_by_type["restriction_policies"] == ["dashboard:dash-1"] +def test_record_empty_binding_escalation_appends_by_type(): + counter = Counter() + counter.record_empty_binding_escalation(resource_type="restriction_policies", _id="dashboard:dash-1") + + assert counter.empty_binding_escalation_by_type["restriction_policies"] == ["dashboard:dash-1"] + + def test_record_methods_ignore_none_args(): counter = Counter() counter.record_stale_principal_dropped(resource_type=None, _id="x") counter.record_stale_principal_dropped(resource_type="roles", _id=None) counter.record_empty_binding_risk(resource_type=None, _id=None) + counter.record_empty_binding_escalation(resource_type=None, _id=None) assert counter.stale_principals_dropped_by_type == {} assert counter.empty_binding_risk_by_type == {} + assert counter.empty_binding_escalation_by_type == {} def test_reset_counter_clears_new_fields(): counter = Counter() counter.record_stale_principal_dropped(resource_type="roles", _id="r1") counter.record_empty_binding_risk(resource_type="roles", _id="r2") + counter.record_empty_binding_escalation(resource_type="roles", _id="r3") counter.reset_counter() assert counter.stale_principals_dropped_by_type == {} assert counter.empty_binding_risk_by_type == {} + assert counter.empty_binding_escalation_by_type == {} From be00928b91a23d271a8eda660e0ce607d57440c4 Mon Sep 17 00:00:00 2001 From: michael-richey Date: Wed, 15 Jul 2026 09:30:17 -0400 Subject: [PATCH 3/4] fix: recheck destination after lazy state load --- datadog_sync/utils/base_resource.py | 7 ++++++- tests/unit/test_base_resource.py | 31 +++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/datadog_sync/utils/base_resource.py b/datadog_sync/utils/base_resource.py index ff6cd44af..4e258267e 100644 --- a/datadog_sync/utils/base_resource.py +++ b/datadog_sync/utils/base_resource.py @@ -444,7 +444,8 @@ def _resolve_or_drop(self, plain_id: str, resource_to_connect: str) -> Tuple[Opt - Present in state.destination[resource_to_connect]: returns (destination_id, False) -- success path, caller keeps/remaps this entry. - Absent from destination: calls ensure_resource_loaded() (for - --minimize-reads correctness) then checks state.source[resource_to_connect]. + --minimize-reads correctness), then rechecks destination because the lazy load + may have populated its mapping, and only then checks source. - Present in source ("not yet synced"): returns (None, False) -- caller must treat this as today's hard-fail (add plain_id to failed_connections); NOT a drop. This is the legitimate "retry on a later sync" case. @@ -470,6 +471,10 @@ def _resolve_or_drop(self, plain_id: str, resource_to_connect: str) -> Tuple[Opt return destination[plain_id]["id"], False self.config.state.ensure_resource_loaded(resource_to_connect, plain_id) + destination = self.config.state.destination[resource_to_connect] + if plain_id in destination: + return destination[plain_id]["id"], False + if plain_id in self.config.state.source[resource_to_connect]: return None, False return None, True diff --git a/tests/unit/test_base_resource.py b/tests/unit/test_base_resource.py index ea4dcaf02..34feeeda4 100644 --- a/tests/unit/test_base_resource.py +++ b/tests/unit/test_base_resource.py @@ -48,6 +48,37 @@ def test_resolve_or_drop_destination_present_returns_dest_id_not_stale(): config.state.ensure_resource_loaded.assert_not_called() +def test_resolve_or_drop_rechecks_destination_after_lazy_load_populates_both_states(): + config = _make_config() + + def load_role(resource_type, plain_id): + config.state.source[resource_type][plain_id] = {"id": plain_id} + config.state.destination[resource_type][plain_id] = {"id": "dst-role"} + + config.state.ensure_resource_loaded.side_effect = load_role + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved == "dst-role" + assert stale is False + config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role") + + +def test_resolve_or_drop_rechecks_destination_after_lazy_load_populates_destination_only(): + config = _make_config() + + def load_role(resource_type, plain_id): + config.state.destination[resource_type][plain_id] = {"id": "dst-role"} + + config.state.ensure_resource_loaded.side_effect = load_role + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved == "dst-role" + assert stale is False + config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role") + + def test_resolve_or_drop_source_present_not_synced_returns_none_not_stale(): config = _make_config() config.state.source["roles"]["src-role"] = {"id": "src-role"} From 5f1462f85f2139796e119c6a0c2d3359823b3901 Mon Sep 17 00:00:00 2001 From: michael-richey Date: Wed, 15 Jul 2026 10:05:51 -0400 Subject: [PATCH 4/4] refactor: name resource connection results --- datadog_sync/utils/base_resource.py | 13 +++++++++++-- datadog_sync/utils/configuration.py | 4 ++-- datadog_sync/utils/resources_handler.py | 3 ++- tests/unit/test_apply_resource_connection_error.py | 4 ++-- tests/unit/test_base_resource.py | 11 ++++++++++- 5 files changed, 27 insertions(+), 8 deletions(-) diff --git a/datadog_sync/utils/base_resource.py b/datadog_sync/utils/base_resource.py index 4e258267e..8e6ed4860 100644 --- a/datadog_sync/utils/base_resource.py +++ b/datadog_sync/utils/base_resource.py @@ -141,6 +141,13 @@ def build_excluded_attributes(self) -> None: self.excluded_attributes[i] = "root" + "".join(["['{}']".format(v) for v in attr.split(".")]) +@dataclass(frozen=True) +class ResourceConnectionResult: + """Outcome metadata from resolving a resource's cross-resource connections.""" + + empty_binding_escalation: bool = False + + class BaseResource(abc.ABC): resource_type: str resource_config: ResourceConfig @@ -636,9 +643,9 @@ def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> return [str(v) for v in r_obj[key]] return [str(r_obj[key])] - def connect_resources(self, _id: str, resource: Dict) -> None: + def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult: if not self.resource_config.resource_connections: - return + return ResourceConnectionResult() failed_connections_dict = defaultdict(list) for resource_to_connect, v in self.resource_config.resource_connections.items(): @@ -660,6 +667,8 @@ def connect_resources(self, _id: str, resource: Dict) -> None: else: self.config.logger.debug(f"{str(e)}", _id=_id, resource_type=self.resource_type) + return ResourceConnectionResult() + def filter(self, resource: Dict) -> bool: if not self.config.filters or self.resource_type not in self.config.filters: return True diff --git a/datadog_sync/utils/configuration.py b/datadog_sync/utils/configuration.py index 7050a72c7..755897606 100644 --- a/datadog_sync/utils/configuration.py +++ b/datadog_sync/utils/configuration.py @@ -83,8 +83,8 @@ class Configuration(object): # Opt-in: drop principal/role references that are absent from BOTH destination and # source state (permanently gone -- e.g. deleted before this org's first-ever import) # instead of hard-failing the whole resource. When False (default) behavior is - # byte-for-byte unchanged. Consumed by the restriction_policies/monitors/ - # synthetics_tests/dashboards/synthetics_private_locations models' connect logic. + # byte-for-byte unchanged. The shared BaseResource drop-aware helpers consume this + # flag; model-specific connection wiring is intentionally separate. drop_unresolvable_principals: bool = False # Set at runtime by Workers.__init__ so resource models (which only hold self.config) # can record dropped stale principals into the apply-run Counter from inside diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py index a1333d47b..ad6bde846 100644 --- a/datadog_sync/utils/resources_handler.py +++ b/datadog_sync/utils/resources_handler.py @@ -493,7 +493,8 @@ async def _apply_resource_cb(self, q_item: List) -> None: # Run hooks await r_class._pre_resource_action_hook(_id, resource) - empty_binding_escalation = bool(r_class.connect_resources(_id, resource)) + connection_result = r_class.connect_resources(_id, resource) + empty_binding_escalation = connection_result.empty_binding_escalation success_metric_tags = [] if empty_binding_escalation: diff --git a/tests/unit/test_apply_resource_connection_error.py b/tests/unit/test_apply_resource_connection_error.py index 1c5d4eec4..2ecef0b22 100644 --- a/tests/unit/test_apply_resource_connection_error.py +++ b/tests/unit/test_apply_resource_connection_error.py @@ -14,7 +14,7 @@ import asyncio from unittest.mock import AsyncMock, MagicMock -from datadog_sync.utils.base_resource import ResourceConfig +from datadog_sync.utils.base_resource import ResourceConfig, ResourceConnectionResult from datadog_sync.utils.resource_utils import ResourceConnectionError from datadog_sync.utils.resources_handler import ResourcesHandler @@ -110,7 +110,7 @@ def test_successful_update_after_suppressed_empty_binding_risk_is_tagged_and_rec _id = "dashboard:dash-1" r_class = MagicMock() r_class.resource_config = ResourceConfig(base_path="", skip_resource_mapping=True) - r_class.connect_resources.return_value = True + r_class.connect_resources.return_value = ResourceConnectionResult(empty_binding_escalation=True) r_class._pre_resource_action_hook = AsyncMock() r_class._update_resource = AsyncMock() r_class._send_action_metrics = AsyncMock() diff --git a/tests/unit/test_base_resource.py b/tests/unit/test_base_resource.py index 34feeeda4..e3bf04460 100644 --- a/tests/unit/test_base_resource.py +++ b/tests/unit/test_base_resource.py @@ -18,7 +18,7 @@ import pytest -from datadog_sync.utils.base_resource import BaseResource +from datadog_sync.utils.base_resource import BaseResource, ResourceConnectionResult def _make_config(): @@ -36,6 +36,15 @@ def _call(config, plain_id, resource_to_connect): return BaseResource._resolve_or_drop(stub, plain_id, resource_to_connect) +def test_connect_resources_returns_named_default_result_when_no_connections(): + stub = SimpleNamespace(resource_config=SimpleNamespace(resource_connections=None)) + + result = BaseResource.connect_resources(stub, "resource-id", {}) + + assert result == ResourceConnectionResult() + assert result.empty_binding_escalation is False + + def test_resolve_or_drop_destination_present_returns_dest_id_not_stale(): config = _make_config() config.state.destination["roles"]["src-role"] = {"id": "dst-role"}