feat: --drop-unresolvable-principals flag + drop-aware connection scaffolding (1/4)

datadog_sync/commands/shared/options.py

@@ -499,6 +499,21 @@ 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 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(
         "--cleanup",
         default="False",

datadog_sync/model/dashboards.py

@@ -4,11 +4,18 @@
 # Copyright 2019 Datadog, Inc.
 
 from __future__ import annotations
+from collections import defaultdict
 from copy import deepcopy
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, cast
 
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource, check_diff, prep_resource
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult
+from datadog_sync.utils.resource_utils import (
+    CustomClientHTTPError,
+    SkipResource,
+    check_diff,
+    find_attr,
+    prep_resource,
+)
 
 if TYPE_CHECKING:
     from datadog_sync.utils.custom_client import CustomClient
@@ -172,5 +179,36 @@ async def delete_resource(self, _id: str) -> None:
             self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """Drop-aware override.
+
+        Widget connections (monitor alert_ids, powerpacks, slos) keep the generic
+        find_attr/connect_id path. The flat `restricted_roles` list goes through the shared
+        drop-aware filter so a permanently-stale role can be dropped (under
+        --drop-unresolvable-principals) while an emptied list still hard-fails as an
+        access-elevation guard.
+        """
+        if not self.resource_config.resource_connections:
+            return ResourceConnectionResult()
+
+        failed_connections_dict = defaultdict(list)
+        for resource_to_connect, attrs in self.resource_config.resource_connections.items():
+            for attr_connection in attrs:
+                if attr_connection == "restricted_roles":
+                    continue  # handled by the drop-aware filter below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource, "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, roles_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         return super(Dashboards, self).connect_id(key, r_obj, resource_to_connect)

datadog_sync/model/monitors.py

@@ -4,14 +4,15 @@
 # Copyright 2019 Datadog, Inc.
 
 from __future__ import annotations
+from collections import defaultdict
 import logging
 import re
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, cast
 
 from datadog_sync.constants import LOGGER_NAME
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, TaggingConfig
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult, TaggingConfig
 from datadog_sync.utils.custom_client import PaginationConfig
-from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource
+from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource, find_attr
 
 _log = logging.getLogger(LOGGER_NAME)
 
@@ -238,6 +239,49 @@ async def delete_resource(self, _id: str) -> None:
             params={"force": "true"},
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """Drop-aware override.
+
+        query / composite-monitor / slo-alert connections keep the generic
+        find_attr/connect_id path. The two access-control shapes -- the flat
+        `restricted_roles` list and the `restriction_policy.bindings.principals` composites
+        -- go through the shared drop-aware filters so permanently-stale references can be
+        dropped (under --drop-unresolvable-principals) while an emptied binding/list still
+        hard-fails as an access-elevation guard.
+        """
+        if not self.resource_config.resource_connections:
+            return ResourceConnectionResult()
+
+        failed_connections_dict = defaultdict(list)
+        for resource_to_connect, attrs in self.resource_config.resource_connections.items():
+            for attr_connection in attrs:
+                if attr_connection in ("restricted_roles", "restriction_policy.bindings.principals"):
+                    continue  # handled by the drop-aware filters below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        empty_risk = False
+        restriction_policy = resource.get("restriction_policy")
+        if restriction_policy:
+            principal_failed, binding_risk = self._filter_stale_binding_principals(
+                _id, restriction_policy.get("bindings")
+            )
+            for rt, ids in principal_failed.items():
+                failed_connections_dict[rt].extend(ids)
+            empty_risk = empty_risk or binding_risk
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource, "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+        empty_risk = empty_risk or roles_risk
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, empty_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         monitors = self.config.state.destination[resource_to_connect]
 
@@ -308,6 +352,9 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
 
     def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         # Mirror of connect_id -- keep in sync when connect_id changes.
+        # Intentionally UNAFFECTED by --drop-unresolvable-principals: must keep returning
+        # every referenced id (including ones connect_resources will later drop) so
+        # --minimize-reads lazy-loading can look them up in source to make the drop decision.
         if key == "query" and r_obj.get("type") == "composite" and resource_to_connect != "service_level_objectives":
             return re.findall("[0-9]+", r_obj[key])
         elif key == "query" and resource_to_connect == "service_level_objectives" and r_obj.get("type") == "slo alert":

datadog_sync/model/restriction_policies.py

@@ -4,10 +4,15 @@
 # Copyright 2019 Datadog, Inc.
 
 from __future__ import annotations
+from collections import defaultdict
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple
 
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult
+from datadog_sync.utils.resource_utils import (
+    CustomClientHTTPError,
+    SkipResource,
+    find_attr,
+)
 
 if TYPE_CHECKING:
     from datadog_sync.utils.custom_client import CustomClient
@@ -147,6 +152,38 @@ async def delete_resource(self, _id: str) -> None:
             self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """Drop-aware override.
+
+        The "id" connections (dashboards/slos/notebooks) keep the generic
+        find_attr/connect_id hard-fail path. The bindings' composite principals go through
+        the shared per-binding filter so that principals permanently absent from source can
+        be dropped (under --drop-unresolvable-principals) instead of failing the whole
+        policy, while an emptied binding still hard-fails as an access-elevation guard.
+        """
+        if not self.resource_config.resource_connections:
+            return ResourceConnectionResult()
+
+        failed_connections_dict = defaultdict(list)
+        for resource_to_connect, attrs in self.resource_config.resource_connections.items():
+            for attr_connection in attrs:
+                if attr_connection == "attributes.bindings.principals":
+                    continue  # handled per-binding below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        bindings = (resource.get("attributes") or {}).get("bindings")
+        principal_failed, empty_binding_risk = self._filter_stale_binding_principals(_id, bindings)
+        for rt, ids in principal_failed.items():
+            failed_connections_dict[rt].extend(ids)
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, empty_binding_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         dashboards = self.config.state.destination["dashboards"]
         slos = self.config.state.destination["service_level_objectives"]
@@ -198,6 +235,11 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
 
     def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         # Mirror of connect_id -- keep in sync when connect_id changes.
+        # Intentionally UNAFFECTED by --drop-unresolvable-principals: this must keep
+        # returning every referenced id (including ones connect_resources will later drop),
+        # because --minimize-reads lazy-loading relies on it to decide WHICH principals to
+        # look up in source in order to make the drop decision. Do NOT "fix" this to match
+        # connect_resources' filtering.
         if key == "id":
             _type, _id = r_obj[key].split(":", 1)
             type_map = {"dashboard": "dashboards", "slo": "service_level_objectives", "notebook": "notebooks"}

datadog_sync/model/synthetics_private_locations.py

@@ -6,11 +6,12 @@
 from __future__ import annotations
 import json
 import re
+from collections import defaultdict
 
 from typing import TYPE_CHECKING, List, Dict, Optional, Tuple
 
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, TaggingConfig
-from datadog_sync.utils.resource_utils import SkipResource
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult, TaggingConfig
+from datadog_sync.utils.resource_utils import SkipResource, find_attr
 
 if TYPE_CHECKING:
     from datadog_sync.utils.custom_client import CustomClient
@@ -102,5 +103,34 @@ async def delete_resource(self, _id: str) -> None:
             self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """Drop-aware override for the flat `metadata.restricted_roles` list.
+
+        A permanently-stale role can be dropped (under --drop-unresolvable-principals) while
+        an emptied list still hard-fails as an access-elevation guard. Any future
+        non-restricted_roles connection keeps the generic find_attr/connect_id path.
+        """
+        if not self.resource_config.resource_connections:
+            return ResourceConnectionResult()
+
+        failed_connections_dict = defaultdict(list)
+        for resource_to_connect, attrs in self.resource_config.resource_connections.items():
+            for attr_connection in attrs:
+                if attr_connection == "metadata.restricted_roles":
+                    continue  # handled by the drop-aware filter below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource.get("metadata"), "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, roles_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         return super(SyntheticsPrivateLocations, self).connect_id(key, r_obj, resource_to_connect)

datadog_sync/model/synthetics_tests.py

@@ -9,11 +9,13 @@
 import certifi
 import json
 import ssl
+from collections import defaultdict
 from copy import deepcopy
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, cast
 from yarl import URL
 
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, TaggingConfig
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult, TaggingConfig
+from datadog_sync.utils.resource_utils import find_attr
 from datadog_sync.model.synthetics_mobile_applications_versions import SyntheticsMobileApplicationsVersions
 
 if TYPE_CHECKING:
@@ -422,6 +424,49 @@ async def delete_resource(self, _id: str) -> None:
         dest_resource = self.config.state.destination[self.resource_type][_id]
         await self._delete_test(destination_client, dest_resource.get("type"), dest_resource["public_id"])
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """Drop-aware override.
+
+        All non-access-control connections (private locations, subtests, global variables,
+        rum/mobile apps) keep the generic find_attr/connect_id path. The flat
+        `options.restricted_roles` list and the `restriction_policy.bindings.principals`
+        composites go through the shared drop-aware filters so permanently-stale references
+        can be dropped (under --drop-unresolvable-principals) while an emptied binding/list
+        still hard-fails as an access-elevation guard.
+        """
+        if not self.resource_config.resource_connections:
+            return ResourceConnectionResult()
+
+        failed_connections_dict = defaultdict(list)
+        for resource_to_connect, attrs in self.resource_config.resource_connections.items():
+            for attr_connection in attrs:
+                if attr_connection in ("options.restricted_roles", "restriction_policy.bindings.principals"):
+                    continue  # handled by the drop-aware filters below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        empty_risk = False
+        restriction_policy = resource.get("restriction_policy")
+        if restriction_policy:
+            principal_failed, binding_risk = self._filter_stale_binding_principals(
+                _id, restriction_policy.get("bindings")
+            )
+            for rt, ids in principal_failed.items():
+                failed_connections_dict[rt].extend(ids)
+            empty_risk = empty_risk or binding_risk
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource.get("options"), "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+        empty_risk = empty_risk or roles_risk
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, empty_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         failed_connections: List[str] = []
         if resource_to_connect == "synthetics_private_locations":
@@ -504,6 +549,9 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
 
     def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         # Mirror of connect_id -- keep in sync when connect_id changes.
+        # Intentionally UNAFFECTED by --drop-unresolvable-principals: must keep returning
+        # every referenced id (including ones connect_resources will later drop) so
+        # --minimize-reads lazy-loading can look them up in source to make the drop decision.
         # Only synthetics_private_locations and mobile application versions need special handling.
         # rum_applications, synthetics_tests (subtests), synthetics_global_variables, roles, and
         # synthetics_mobile_applications (applicationId key) all use plain IDs at the leaf —