fix: reconcile state.destination on skip-without-write outcomes

datadog_sync/utils/base_resource.py

@@ -394,17 +394,62 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         pass
 
     async def _create_resource(self, _id: str, resource: Dict) -> None:
-        _id, r = await self.create_resource(_id, resource)
+        try:
+            _id, r = await self.create_resource(_id, resource)
+        except SkipResource:
+            self._reconcile_destination_if_absent(_id, resource)
+            raise
         self.config.state.destination[self.resource_type][_id] = r
 
     @abc.abstractmethod
     async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         pass
 
     async def _update_resource(self, _id: str, resource: Dict) -> None:
-        _id, r = await self.update_resource(_id, resource)
+        try:
+            _id, r = await self.update_resource(_id, resource)
+        except SkipResource:
+            self._reconcile_destination_if_absent(_id, resource)
+            raise
         self.config.state.destination[self.resource_type][_id] = r
 
+    def _reconcile_destination_if_absent(self, _id: str, resource: Dict) -> None:
+        """Best-effort: record a skipped-but-existing destination resource in state.
+
+        When ``create_resource``/``update_resource`` discovers (via
+        ``_existing_resources_map``) that a resource already exists on the
+        destination and raises ``SkipResource`` without writing
+        ``state.destination``, the bucket view diverges from destination truth:
+        no state file is persisted, so downstream consumers that trust the bucket
+        count the id as failed even though the resource is confirmed present via
+        the live API.
+
+        This reconciles that gap by writing the discovered destination resource
+        into ``state.destination`` keyed by the source id (the convention used
+        throughout state.py / models). Insert-if-absent only: an entry already
+        present is never overwritten, preserving delegate-then-skip resources
+        that wrote ``state.destination`` before delegating to ``update_resource``.
+
+        Best-effort: any unexpected error during key extraction/lookup is logged
+        at debug and swallowed so a reconcile failure can never break a skip's
+        counter/metrics accounting or turn a skip into an unhandled error.
+        """
+        try:
+            if _id in self.config.state.destination[self.resource_type]:
+                return
+            key = self.get_resource_mapping_key(resource)
+            if key is not None and key in self._existing_resources_map:
+                self.config.state.destination[self.resource_type][_id] = self._existing_resources_map[key]
+        except Exception as e:
+            # Pre-format the message (f-string) rather than passing positional
+            # %s args: the NDJSON log backend (utils/log.py Log.debug) does not
+            # interpolate positional args in JSON mode, so %s placeholders would
+            # be emitted literally. Pre-formatting keeps the diagnostic readable
+            # in both plain and NDJSON modes.
+            self.config.logger.debug(
+                f"destination reconcile skipped for {self.resource_type} {_id}: {e}"
+            )
+
     @abc.abstractmethod
     async def delete_resource(self, _id: str) -> None:
         pass

datadog_sync/utils/resources_handler.py

@@ -544,8 +544,19 @@ async def _apply_resource_cb(self, q_item: List) -> None:
                 await sem.acquire()
                 sem_acquired = True
 
-            # Run hooks
-            await r_class._pre_resource_action_hook(_id, resource)
+            # Run hooks. Reconcile a pre-hook SkipResource the same way the
+            # _create_resource/_update_resource wrappers do: if the resource is
+            # present in _existing_resources_map, record it in state.destination
+            # (insert-if-absent) before re-raising. Without this, a pre-hook skip
+            # for a resource that already exists on the destination (e.g. an
+            # immutable or deprecated security monitoring rule whose matching
+            # destination rule is in the map) would leave no state.destination
+            # entry -- the same bucket-view divergence this change fixes elsewhere.
+            try:
+                await r_class._pre_resource_action_hook(_id, resource)
+            except SkipResource:
+                r_class._reconcile_destination_if_absent(_id, resource)
+                raise
             connection_result = r_class.connect_resources(_id, resource)
             empty_binding_escalation = connection_result.empty_binding_escalation
 

tests/unit/test_logs_indexes_skip_reconcile.py

@@ -0,0 +1,93 @@
+# 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.
+
+"""logs_indexes: regression guard + framework safety-net proof.
+
+logs_indexes' exists-path is already correct (write state.destination, delegate to
+update_resource PUT). The framework is a pure no-op for it; the safety-net test
+proves the framework WOULD reconcile logs_indexes if its exists-path regressed.
+logs_indexes keys _existing_resources_map by name, and import_resource returns
+resource["name"] as the source id, so source _id == name == map key.
+
+All identifiers are obviously synthetic (``index-test``).
+"""
+
+import asyncio
+from collections import defaultdict
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from datadog_sync.model.logs_indexes import LogsIndexes
+from datadog_sync.utils.resource_utils import SkipResource
+
+
+def _index(name):
+    return {
+        "name": name,
+        "daily_limit": None,
+    }
+
+
+def _make_logs_indexes(existing_map=None, destination_client=None):
+    config = MagicMock()
+    config.destination_client = destination_client or MagicMock()
+    config.state = MagicMock()
+    config.state.source = defaultdict(dict)
+    config.state.destination = defaultdict(dict)
+    config.logger = MagicMock()
+    li = LogsIndexes(config=config)
+    li._existing_resources_map = existing_map or {}
+    return li, config
+
+
+class TestLogsIndexesSkipReconcile:
+    def test_existing_resource_create_writes_state_destination(self):
+        # source _id == name == map key.
+        _id = "index-test"
+        dest = _index("index-test")
+        put_resp = {"name": "index-test", "daily_limit": 42}
+        destination_client = MagicMock()
+        destination_client.put = AsyncMock(return_value=put_resp)
+        li, config = _make_logs_indexes(
+            existing_map={"index-test": dest},
+            destination_client=destination_client,
+        )
+        resource = _index("index-test")
+
+        asyncio.run(li._create_resource(_id, resource))
+
+        destination_client.put.assert_called_once()
+        # update_resource does state.destination[_id].update(resp); wrapper rewrites.
+        assert config.state.destination["logs_indexes"][_id]["daily_limit"] == 42
+
+    def test_framework_reconciles_if_create_raised_skip(self):
+        _id = "index-test"
+        dest = _index("index-test")
+        li, config = _make_logs_indexes(existing_map={"index-test": dest})
+        li.create_resource = AsyncMock(side_effect=SkipResource(_id, "logs_indexes", "exists"))
+        resource = _index("index-test")
+
+        with pytest.raises(SkipResource):
+            asyncio.run(li._create_resource(_id, resource))
+
+        assert config.state.destination["logs_indexes"][_id] == dest
+
+    def test_create_non_existing_still_posts(self):
+        _id = "index-test"
+        posted = _index("index-test")
+        destination_client = MagicMock()
+        destination_client.post = AsyncMock(return_value=posted)
+        li, config = _make_logs_indexes(
+            existing_map={},
+            destination_client=destination_client,
+        )
+        resource = _index("index-test")
+
+        out_id, out_r = asyncio.run(li.create_resource(_id, resource))
+
+        destination_client.post.assert_called_once()
+        assert out_id == _id
+        assert out_r == posted

tests/unit/test_logs_metrics_skip_reconcile.py

@@ -0,0 +1,92 @@
+# 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.
+
+"""logs_metrics: regression guard + framework safety-net proof.
+
+logs_metrics' exists-path is already correct (write state.destination, delegate to
+update_resource PATCH). The framework is a pure no-op for it; the safety-net test
+proves the framework WOULD reconcile logs_metrics if its exists-path regressed.
+
+All identifiers are obviously synthetic (``metric-src``, ``metric-dst``).
+"""
+
+import asyncio
+from collections import defaultdict
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from datadog_sync.model.logs_metrics import LogsMetrics
+from datadog_sync.utils.resource_utils import SkipResource
+
+
+def _metric(metric_id, name=None):
+    return {
+        "id": metric_id,
+        "type": "logs_metrics",
+        "attributes": {"name": name or metric_id},
+    }
+
+
+def _make_logs_metrics(existing_map=None, destination_client=None):
+    config = MagicMock()
+    config.destination_client = destination_client or MagicMock()
+    config.state = MagicMock()
+    config.state.source = defaultdict(dict)
+    config.state.destination = defaultdict(dict)
+    config.logger = MagicMock()
+    lm = LogsMetrics(config=config)
+    lm._existing_resources_map = existing_map or {}
+    return lm, config
+
+
+class TestLogsMetricsSkipReconcile:
+    def test_existing_resource_create_writes_state_destination(self):
+        # logs_metrics ids are stable across orgs, so source _id == map key.
+        _id = "metric-test"
+        dest = _metric("metric-test")
+        patched = _metric("metric-test")
+        patched["attributes"]["marker"] = "patched"
+        destination_client = MagicMock()
+        destination_client.patch = AsyncMock(return_value={"data": patched})
+        lm, config = _make_logs_metrics(
+            existing_map={"metric-test": dest},
+            destination_client=destination_client,
+        )
+        resource = _metric("metric-test")
+
+        asyncio.run(lm._create_resource(_id, resource))
+
+        destination_client.patch.assert_called_once()
+        assert config.state.destination["logs_metrics"][_id] == patched
+
+    def test_framework_reconciles_if_create_raised_skip(self):
+        _id = "metric-test"
+        dest = _metric("metric-test")
+        lm, config = _make_logs_metrics(existing_map={"metric-test": dest})
+        lm.create_resource = AsyncMock(side_effect=SkipResource(_id, "logs_metrics", "exists"))
+        resource = _metric("metric-test")
+
+        with pytest.raises(SkipResource):
+            asyncio.run(lm._create_resource(_id, resource))
+
+        assert config.state.destination["logs_metrics"][_id] == dest
+
+    def test_create_non_existing_still_posts(self):
+        _id = "metric-test"
+        posted = _metric("metric-test")
+        destination_client = MagicMock()
+        destination_client.post = AsyncMock(return_value={"data": posted})
+        lm, config = _make_logs_metrics(
+            existing_map={},
+            destination_client=destination_client,
+        )
+        resource = _metric("metric-test")
+
+        out_id, out_r = asyncio.run(lm.create_resource(_id, resource))
+
+        destination_client.post.assert_called_once()
+        assert out_id == _id
+        assert out_r == posted