From 08539e4aeebd21a9a63cc615580b87b10c705ccb Mon Sep 17 00:00:00 2001 From: "riyaz.shiraguppi" Date: Tue, 25 Aug 2026 13:05:40 -0500 Subject: [PATCH 1/2] Emit typed bootstrap skip outcomes --- datadog_sync/model/logs_pipelines.py | 64 +++++++---- datadog_sync/model/metric_percentiles.py | 10 +- .../model/metric_tag_configurations.py | 12 ++- datadog_sync/model/metrics_metadata.py | 12 ++- datadog_sync/utils/resource_utils.py | 17 ++- datadog_sync/utils/resources_handler.py | 101 +++++++++++++----- datadog_sync/utils/sync_report.py | 17 +-- tests/unit/test_failure_class.py | 94 +++++++++++++++- tests/unit/test_logs_pipelines.py | 59 +++++++++- tests/unit/test_metric_percentiles.py | 16 ++- tests/unit/test_metric_tag_configurations.py | 16 ++- tests/unit/test_metrics_metadata.py | 12 ++- 12 files changed, 364 insertions(+), 66 deletions(-) diff --git a/datadog_sync/model/logs_pipelines.py b/datadog_sync/model/logs_pipelines.py index 7c306425..5fd88647 100644 --- a/datadog_sync/model/logs_pipelines.py +++ b/datadog_sync/model/logs_pipelines.py @@ -12,7 +12,12 @@ from datadog_sync.constants import LOGGER_NAME, Metrics from datadog_sync.utils.base_resource import BaseResource, ResourceConfig -from datadog_sync.utils.resource_utils import DEFAULT_TAGS, SkipResource, check_diff +from datadog_sync.utils.resource_utils import ( + DEFAULT_TAGS, + FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED, + SkipResource, + check_diff, +) if TYPE_CHECKING: from datadog_sync.utils.custom_client import CustomClient @@ -30,9 +35,14 @@ def _summarize_diff_keys(diff) -> List[str]: if not diff: return [] keys = set() - for change_type in ("values_changed", "type_changes", "iterable_item_added", - "iterable_item_removed", "dictionary_item_added", - "dictionary_item_removed"): + for change_type in ( + "values_changed", + "type_changes", + "iterable_item_added", + "iterable_item_removed", + "dictionary_item_added", + "dictionary_item_removed", + ): for path in diff.get(change_type, {}) or {}: # DeepDiff paths look like "root['is_enabled']" or "root['filter']['query']" # Extract the first bracketed segment. @@ -110,18 +120,33 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: "ddtags": ",".join(DEFAULT_TAGS), "message": f"[datadog-sync-cli] Triggering creation of '{resource['name']}' integration pipeline", } + outcome_details = { + "pipeline_name": resource["name"], + "ddsource": source, + } # Submit a log to the logs intake API to trigger the creation of the integration pipeline override_url = self.config.destination_logs_intake_url - if override_url: - await destination_client.post_unauthenticated(override_url, payload) - else: - subdomain = f"{self.logs_intake_subdomain}.{destination_client.url_object.subdomain}" - if destination_client.url_object.subdomain == "api": - subdomain = self.logs_intake_subdomain - elif destination_client.url_object.subdomain.startswith("api."): - subdomain = f"{self.logs_intake_subdomain}.{destination_client.url_object.subdomain[4:]}" - await destination_client.post(self.logs_intake_path, payload, subdomain=subdomain) + try: + if override_url: + await destination_client.post_unauthenticated(override_url, payload) + else: + subdomain = f"{self.logs_intake_subdomain}.{destination_client.url_object.subdomain}" + if destination_client.url_object.subdomain == "api": + subdomain = self.logs_intake_subdomain + elif destination_client.url_object.subdomain.startswith("api."): + subdomain = f"{self.logs_intake_subdomain}.{destination_client.url_object.subdomain[4:]}" + await destination_client.post(self.logs_intake_path, payload, subdomain=subdomain) + except Exception as e: + _log.debug("logs_pipelines: integration pipeline bootstrap intake post failed: %s", e) + raise SkipResource( + _id, + self.resource_type, + "Integration pipeline is not present on destination and requires bootstrap.", + failure_class=FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED, + reason=FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED, + outcome_details=outcome_details, + ) created = False for _ in range(12): @@ -134,9 +159,14 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: await sleep(5) if not created: - raise Exception( + raise SkipResource( + _id, + self.resource_type, f"Integration pipeline '{resource['name']}' is not created after x seconds. " - "It will be rechecked in the next sync." + "It will be rechecked in the next sync.", + failure_class=FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED, + reason=FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED, + outcome_details=outcome_details, ) self.config.state.destination[self.resource_type][_id] = self.destination_integration_pipelines[ @@ -234,9 +264,7 @@ async def _handle_read_only_diff(self, _id: str, resource: Dict, diff) -> None: ) except Exception as e: # Never let metric emission block the return path. - self.config.logger.debug( - f"logs_pipelines: failed to emit integration_diff_skipped metric: {e}" - ) + self.config.logger.debug(f"logs_pipelines: failed to emit integration_diff_skipped metric: {e}") async def delete_resource(self, _id: str) -> None: if self.config.state.destination[self.resource_type][_id]["is_read_only"]: diff --git a/datadog_sync/model/metric_percentiles.py b/datadog_sync/model/metric_percentiles.py index e89787a6..8ee3d2c7 100644 --- a/datadog_sync/model/metric_percentiles.py +++ b/datadog_sync/model/metric_percentiles.py @@ -6,7 +6,11 @@ from datadog_sync.utils.base_resource import BaseResource, ResourceConfig from datadog_sync.utils.custom_client import CustomClient -from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_DESTINATION_METRIC_MISSING, + CustomClientHTTPError, + SkipResource, +) def _error_body(error: CustomClientHTTPError) -> str: @@ -69,10 +73,14 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: await destination_client.patch(path, {"metric_names": [_id]}) except CustomClientHTTPError as e: if _is_metric_not_found_error(e): + operation = "percentiles_enable" if resource.get("include_percentiles") else "percentiles_disable" raise SkipResource( _id, self.resource_type, "Metric not present on destination; percentiles cannot attach.", + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + outcome_details={"metric_name": _id, "operation": operation}, ) raise diff --git a/datadog_sync/model/metric_tag_configurations.py b/datadog_sync/model/metric_tag_configurations.py index 2869115b..da5a8579 100644 --- a/datadog_sync/model/metric_tag_configurations.py +++ b/datadog_sync/model/metric_tag_configurations.py @@ -7,7 +7,11 @@ 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 +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_DESTINATION_METRIC_MISSING, + CustomClientHTTPError, + SkipResource, +) if TYPE_CHECKING: from datadog_sync.utils.custom_client import CustomClient @@ -69,6 +73,9 @@ async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: _id, self.resource_type, "Metric not present on destination; tag configuration cannot attach.", + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + outcome_details={"metric_name": _id, "operation": "tag_configuration_create"}, ) if not _is_existing_tag_config_conflict(e): raise @@ -96,6 +103,9 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: _id, self.resource_type, "Metric not present on destination; tag configuration cannot attach.", + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + outcome_details={"metric_name": _id, "operation": "tag_configuration_update"}, ) raise diff --git a/datadog_sync/model/metrics_metadata.py b/datadog_sync/model/metrics_metadata.py index b69367a2..1f6488d5 100644 --- a/datadog_sync/model/metrics_metadata.py +++ b/datadog_sync/model/metrics_metadata.py @@ -8,7 +8,11 @@ from datadog_sync.constants import LOGGER_NAME from datadog_sync.utils.base_resource import BaseResource, ResourceConfig from datadog_sync.utils.custom_client import CustomClient -from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_DESTINATION_METRIC_MISSING, + CustomClientHTTPError, + SkipResource, +) log = logging.getLogger(LOGGER_NAME) @@ -69,8 +73,7 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: raise SkipResource( _id, self.resource_type, - "distribution type is rejected by the destination metrics_metadata endpoint; " - "skipping public PUT", + "distribution type is rejected by the destination metrics_metadata endpoint; " "skipping public PUT", ) # metrics_metadata can only attach to a metric that already exists on @@ -95,6 +98,9 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: _id, self.resource_type, "Metric not present on destination; metadata cannot attach.", + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + outcome_details={"metric_name": _id, "operation": "metadata_update"}, ) raise diff --git a/datadog_sync/utils/resource_utils.py b/datadog_sync/utils/resource_utils.py index 0dae7813..a2a92c82 100644 --- a/datadog_sync/utils/resource_utils.py +++ b/datadog_sync/utils/resource_utils.py @@ -27,6 +27,9 @@ DEFAULT_TAGS = ["managed_by:datadog-sync"] +FAILURE_CLASS_DESTINATION_METRIC_MISSING = "destination_metric_missing" +FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED = "integration_pipeline_bootstrap_required" + # aiohttp timeout family — both have empty ``str()``. _TIMEOUT_EXC_TYPES = (asyncio.TimeoutError, aiohttp.ServerTimeoutError) @@ -53,7 +56,19 @@ def format_exc_for_log(exc: BaseException) -> str: class SkipResource(Exception): - def __init__(self, _id: str, _type: str, msg: str): + def __init__( + self, + _id: str, + _type: str, + msg: str, + *, + failure_class: str = "", + reason: Optional[str] = None, + outcome_details: Optional[Dict[str, str]] = None, + ): + self.failure_class = failure_class + self.outcome_reason = reason + self.outcome_details = outcome_details or {} super(SkipResource, self).__init__(f"Skipping {_type} with id: {_id}. {msg}") diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py index 7a96ca29..4bfbce59 100644 --- a/datadog_sync/utils/resources_handler.py +++ b/datadog_sync/utils/resources_handler.py @@ -55,10 +55,11 @@ def _emit_apply_summary(logger, counter) -> None: operators debugging a cascade usually want to see WHICH failure happened first, not lexicographic order. """ + def _chunked_emit(rt: str, action_desc: str, ids: List[str]) -> None: total = len(ids) for start in range(0, total, _SUMMARY_ID_CHUNK): - chunk = ids[start:start + _SUMMARY_ID_CHUNK] + chunk = ids[start : start + _SUMMARY_ID_CHUNK] end = min(start + _SUMMARY_ID_CHUNK, total) logger.warning( "sync summary: %s %s [%d-%d of %d]: %s", @@ -90,7 +91,7 @@ def _chunked_emit(rt: str, action_desc: str, ids: List[str]) -> None: if ids: total = len(ids) for start in range(0, total, _SUMMARY_ID_CHUNK): - chunk = ids[start:start + _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 " @@ -107,7 +108,7 @@ def _chunked_emit(rt: str, action_desc: str, ids: List[str]) -> None: if ids: total = len(ids) for start in range(0, total, _SUMMARY_ID_CHUNK): - chunk = ids[start:start + _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 " @@ -207,10 +208,15 @@ def _sanitize_reason(err: Exception) -> Tuple[str, str]: The generic fallback emits only the exception class name — the full detail is still logged at error/debug level by the caller. - Canonical failure_class values: + Common HTTP/transport failure_class values: http_4xx_403 http_4xx_404 http_4xx_429 http_4xx_other http_5xx http_timeout http_connection unknown + + Resource models may also attach domain-specific classes to SkipResource + for terminal-but-actionable skips, such as destination_metric_missing. """ + if isinstance(err, SkipResource): + return err.outcome_reason or type(err).__name__, err.failure_class or "unknown" if isinstance(err, CustomClientHTTPError): code = err.status_code reason = f"HTTP {code}" @@ -242,6 +248,7 @@ def _emit( action_sub_type: str = "", reason: str = "", failure_class: str = "", + details: Optional[Dict[str, str]] = None, ) -> None: if self.config.emit_json: _id_str = str(_id) if _id is not None else "" @@ -254,6 +261,7 @@ def _emit( action_sub_type=action_sub_type, reason=reason, failure_class=failure_class, + details=details or {}, ).emit() async def init_async(self) -> None: @@ -418,9 +426,7 @@ async def apply_resources(self) -> Tuple[int, int]: try: _emit_apply_summary(self.config.logger, self.worker.counter) except Exception as e: - self.config.logger.warning( - "sync summary emission failed: %s. State was already persisted.", e - ) + self.config.logger.warning("sync summary emission failed: %s. State was already persisted.", e) def _maybe_refresh_destination_state(self, resource_types) -> None: """Optional refresh of state.destination before workers dispatch. @@ -540,7 +546,15 @@ async def _apply_resource_cb(self, q_item: List) -> None: self.config.logger.info(f"skipping resource: {str(e)}", resource_type=resource_type, _id=_id) self.worker.counter.increment_skipped() _reason, _fc = self._sanitize_reason(e) - self._emit(resource_type, _id, "sync", "skipped", reason=_reason, failure_class=_fc) + self._emit( + resource_type, + _id, + "sync", + "skipped", + reason=_reason, + failure_class=_fc, + details=e.outcome_details, + ) await r_class._send_action_metrics(Command.SYNC.value, _id, Status.SKIPPED.value, tags=["reason:unknown"]) except ResourceConnectionError as e: self.config.logger.error( @@ -554,9 +568,7 @@ async def _apply_resource_cb(self, q_item: List) -> None: # separate process reading state from disk) can correlate the # cascade by matching failed source ids against the previous # process's summary log. - self.worker.counter.increment_skipped( - resource_type=resource_type, _id=_id, missing_deps=True - ) + self.worker.counter.increment_skipped(resource_type=resource_type, _id=_id, missing_deps=True) _reason, _fc = self._sanitize_reason(e) self._emit(resource_type, _id, "sync", "skipped", reason=_reason, failure_class=_fc) # Distinguish the access-elevation case (a restriction-policy binding / @@ -639,7 +651,15 @@ async def _diffs_worker_cb(self, q_item: List) -> None: self.config.logger.warning(f"skipping resource: resource_type:{resource_type} id:{_id}") self.config.logger.debug(str(e)) _reason, _fc = self._sanitize_reason(e) - self._emit(resource_type, _id, "sync", "skipped", reason=_reason, failure_class=_fc) + self._emit( + resource_type, + _id, + "sync", + "skipped", + reason=_reason, + failure_class=_fc, + details=e.outcome_details, + ) return try: @@ -746,9 +766,7 @@ async def import_resources_without_saving(self) -> None: # 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 - ] + 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): @@ -798,6 +816,7 @@ async def _import_get_resources_cb(self, resource_type: str, tmp_storage) -> Non # host_tags) don't have a working per-ID GET path and would produce # 100% permanent failures on this branch. from datadog_sync.utils.configuration import _ID_FILE_IMPORT_SUPPORTED_TYPES + if ( self.config.id_payload and resource_type in self.config.id_payload @@ -835,20 +854,14 @@ async def _import_get_resources_cb(self, resource_type: str, tmp_storage) -> Non # resource whose id downstream sync-cli invocations need to # correlate — a monitor that references this id will later # fail its connect_resources. - self.worker.counter.increment_skipped( - resource_type=resource_type, _id=mid, missing_deps=True - ) + self.worker.counter.increment_skipped(resource_type=resource_type, _id=mid, missing_deps=True) for eid, cls, reason in errored: if cls == "skipped": self._emit(resource_type, eid, "import", "skipped", reason=reason) - self.worker.counter.increment_skipped( - resource_type=resource_type, _id=eid, missing_deps=True - ) + self.worker.counter.increment_skipped(resource_type=resource_type, _id=eid, missing_deps=True) else: self._emit(resource_type, eid, "import", "failure", reason=reason) - self.worker.counter.increment_failure( - resource_type=resource_type, _id=eid - ) + self.worker.counter.increment_failure(resource_type=resource_type, _id=eid) # Threshold check — emit a log line containing the literal "rate limit" # substring so downstream consumers that scan subprocess output for a @@ -929,7 +942,15 @@ async def _import_resource(self, q_item: List) -> None: # a downstream cascade would not want to grep these ids. self.worker.counter.increment_skipped() _reason, _fc = self._sanitize_reason(e) - self._emit(resource_type, _id, "import", "skipped", reason=_reason, failure_class=_fc) + self._emit( + resource_type, + _id, + "import", + "skipped", + reason=_reason, + failure_class=_fc, + details=e.outcome_details, + ) await r_class._send_action_metrics(Command.IMPORT.value, _id, Status.SKIPPED.value) self.config.logger.info(f"skipping resource: {str(e)}", resource_type=resource_type, _id=_id) self.config.logger.debug(str(e)) @@ -1161,7 +1182,15 @@ async def _import_missing_dep_cb(self, q_item: Tuple[str, str]) -> None: self._emit(resource_type, _id, "import", "success") except SkipResource as e: _reason, _fc = self._sanitize_reason(e) - self._emit(resource_type, _id, "import", "skipped", reason=_reason, failure_class=_fc) + self._emit( + resource_type, + _id, + "import", + "skipped", + reason=_reason, + failure_class=_fc, + details=e.outcome_details, + ) self.config.logger.info(f"skipping dependency: {str(e)}", resource_type=resource_type, _id=_id) return except CustomClientHTTPError as e: @@ -1205,7 +1234,15 @@ async def _force_missing_dep_import_cb(self, q_item: List): self._emit(resource_type, _id, "import", "success") except SkipResource as e: _reason, _fc = self._sanitize_reason(e) - self._emit(resource_type, _id, "import", "skipped", reason=_reason, failure_class=_fc) + self._emit( + resource_type, + _id, + "import", + "skipped", + reason=_reason, + failure_class=_fc, + details=e.outcome_details, + ) self.config.logger.info(f"skipping dependency: {str(e)}", resource_type=resource_type, _id=_id) return except CustomClientHTTPError as e: @@ -1250,7 +1287,15 @@ async def _cleanup_worker(self, q_item: List) -> None: # cascade signal. Numeric-only accounting. self.worker.counter.increment_skipped() _reason, _fc = self._sanitize_reason(e) - self._emit(resource_type, _id, "delete", "skipped", reason=_reason, failure_class=_fc) + self._emit( + resource_type, + _id, + "delete", + "skipped", + reason=_reason, + failure_class=_fc, + details=e.outcome_details, + ) await r_class._send_action_metrics("delete", _id, Status.SKIPPED.value, tags=["reason:unknown"]) self.config.logger.info(f"skipping resource: {str(e)}", resource_type=resource_type, _id=_id) self.config.logger.info(f"skip deleting resource: {str(e)}", resource_type=resource_type, _id=_id) diff --git a/datadog_sync/utils/sync_report.py b/datadog_sync/utils/sync_report.py index 6c3c7275..4c5c2bf9 100644 --- a/datadog_sync/utils/sync_report.py +++ b/datadog_sync/utils/sync_report.py @@ -5,7 +5,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Dict, Literal from datadog_sync.utils.ndjson import write_ndjson_line @@ -32,6 +32,7 @@ class ResourceOutcome: status status:X success | skipped | failure | filtered action_sub_type action_sub_type:X create | update | "" (sync only) reason reason:X freetext explanation (truncated to 1024 chars) + details n/a optional machine-readable context Note: ``filtered`` is a JSON-only status. The CLI metric (``datadog.org-sync.action``) is not emitted for filtered resources, so this value has no metric-tag counterpart. @@ -56,17 +57,19 @@ class ResourceOutcome: status: Literal["success", "skipped", "failure", "filtered"] action_sub_type: Literal["create", "update", ""] # only populated on sync success reason: str # empty for success, explanation for skip/fail - # Optional structured failure category that consumers can branch on without - # pattern-matching the reason string. Omitted from the serialised NDJSON - # record when empty so consumers that don't know the field see no schema - # change. + # Optional structured failure/skip category that consumers can branch on + # without pattern-matching the reason string. Omitted from the serialised + # NDJSON record when empty so consumers that don't know the field see no + # schema change. failure_class: str = "" + # Optional machine-readable fields for typed outcomes. Omitted when empty. + details: Dict[str, str] = field(default_factory=dict) def __post_init__(self) -> None: if len(self.reason) > _REASON_MAX_LEN: self.reason = self.reason[:_REASON_MAX_LEN] + "...(truncated)" - def to_dict(self) -> Dict[str, str]: + def to_dict(self) -> Dict[str, object]: d = { "type": "outcome", "command": self.command, @@ -81,6 +84,8 @@ def to_dict(self) -> Dict[str, str]: # backward compatibility for downstream consumers that don't know this field. if self.failure_class: d["failure_class"] = self.failure_class + if self.details: + d["details"] = {str(k): str(v) for k, v in self.details.items()} return d def emit(self) -> None: diff --git a/tests/unit/test_failure_class.py b/tests/unit/test_failure_class.py index 71308f6d..11d81345 100644 --- a/tests/unit/test_failure_class.py +++ b/tests/unit/test_failure_class.py @@ -13,7 +13,12 @@ from io import StringIO from unittest.mock import MagicMock, patch -from datadog_sync.utils.resource_utils import CustomClientHTTPError, ResourceConnectionError +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_DESTINATION_METRIC_MISSING, + CustomClientHTTPError, + ResourceConnectionError, + SkipResource, +) from datadog_sync.utils.resources_handler import ResourcesHandler from datadog_sync.utils.sync_report import ResourceOutcome @@ -116,6 +121,25 @@ def test_resource_connection_error(self): assert reason == "connection_error" assert fc == "http_connection" + def test_plain_skip_resource_returns_unknown(self): + err = SkipResource("abc", "dashboards", "No differences detected.") + reason, fc = ResourcesHandler._sanitize_reason(err) + assert reason == "SkipResource" + assert fc == "unknown" + + def test_typed_skip_resource_returns_configured_failure_class(self): + err = SkipResource( + "custom.metric", + "metrics_metadata", + "Metric not present on destination.", + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + outcome_details={"metric_name": "custom.metric"}, + ) + reason, fc = ResourcesHandler._sanitize_reason(err) + assert reason == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert fc == FAILURE_CLASS_DESTINATION_METRIC_MISSING + def test_generic_exception_returns_class_name_and_unknown(self): err = ValueError("something unexpected") reason, fc = ResourcesHandler._sanitize_reason(err) @@ -188,6 +212,25 @@ def test_failure_class_present_when_set(self): assert "failure_class" in d assert d["failure_class"] == "http_5xx" + def test_details_omitted_when_empty(self): + outcome = ResourceOutcome("import", "monitors", "123", "import", "failure", "", "HTTP 500") + assert "details" not in outcome.to_dict() + + def test_details_present_when_set(self): + outcome = ResourceOutcome( + command="sync", + resource_type="metrics_metadata", + id="custom.metric", + action_type="sync", + status="skipped", + action_sub_type="", + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + details={"metric_name": "custom.metric", "operation": "metadata_update"}, + ) + d = outcome.to_dict() + assert d["details"] == {"metric_name": "custom.metric", "operation": "metadata_update"} + def test_all_7_canonical_values_round_trip(self): """All 7 canonical failure_class values survive to_dict().""" canonical = [ @@ -257,6 +300,26 @@ def test_emit_includes_failure_class_when_set(self): assert parsed["failure_class"] == "http_5xx" assert parsed["reason"] == "HTTP 500" + def test_emit_includes_details_when_set(self): + outcome = ResourceOutcome( + command="sync", + resource_type="metric_percentiles", + id="custom.metric", + action_type="sync", + status="skipped", + action_sub_type="", + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + details={"metric_name": "custom.metric", "operation": "percentiles_enable"}, + ) + buf = StringIO() + with patch("sys.stdout", buf): + outcome.emit() + + parsed = json.loads(buf.getvalue().strip()) + assert parsed["failure_class"] == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert parsed["details"] == {"metric_name": "custom.metric", "operation": "percentiles_enable"} + def test_emit_excludes_failure_class_when_empty(self): outcome = ResourceOutcome( command="import", @@ -411,6 +474,35 @@ def test_emit_sanitize_reason_connection_error_wires_failure_class(self): assert parsed["reason"] == "connection_error" assert parsed["failure_class"] == "http_connection" + def test_emit_typed_skip_resource_wires_failure_class_and_details(self): + handler = self._make_handler_for_emit() + err = SkipResource( + "custom.metric", + "metrics_metadata", + "Metric not present on destination.", + failure_class=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + reason=FAILURE_CLASS_DESTINATION_METRIC_MISSING, + outcome_details={"metric_name": "custom.metric", "operation": "metadata_update"}, + ) + _reason, _fc = ResourcesHandler._sanitize_reason(err) + + buf = StringIO() + with patch("sys.stdout", buf): + ResourcesHandler._emit( + handler, + "metrics_metadata", + "custom.metric", + "sync", + "skipped", + reason=_reason, + failure_class=_fc, + details=err.outcome_details, + ) + parsed = json.loads(buf.getvalue().strip()) + assert parsed["reason"] == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert parsed["failure_class"] == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert parsed["details"] == {"metric_name": "custom.metric", "operation": "metadata_update"} + def test_emit_noop_when_emit_json_false_no_output(self): """Even with failure_class set, no output when emit_json=False.""" handler = self._make_handler_for_emit() diff --git a/tests/unit/test_logs_pipelines.py b/tests/unit/test_logs_pipelines.py index c4ea1e7a..77465fea 100644 --- a/tests/unit/test_logs_pipelines.py +++ b/tests/unit/test_logs_pipelines.py @@ -6,7 +6,13 @@ import asyncio from unittest.mock import AsyncMock, patch +import pytest + from datadog_sync.model.logs_pipelines import LogsPipelines +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED, + SkipResource, +) def _make_integration_resource(name="nginx"): @@ -152,6 +158,53 @@ def test_invalid_pipeline_no_intake_post(mock_config): mock_config.destination_client.post.assert_not_awaited() +def test_integration_pipeline_intake_failure_raises_typed_bootstrap_skip(mock_config): + mock_config.destination_logs_intake_url = "https://evp.example/v2/track/logs/org/42" + + lp = LogsPipelines(mock_config) + lp.destination_integration_pipelines = {} + + mock_config.destination_client.post_unauthenticated = AsyncMock(side_effect=RuntimeError("intake unavailable")) + mock_config.destination_client.post = AsyncMock() + + with pytest.raises(SkipResource) as exc_info: + asyncio.run(lp.create_resource("src-nginx", _make_integration_resource())) + + assert exc_info.value.failure_class == FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED + assert exc_info.value.outcome_reason == FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED + assert exc_info.value.outcome_details == { + "pipeline_name": "nginx", + "ddsource": "nginx", + } + mock_config.destination_client.post_unauthenticated.assert_awaited_once() + mock_config.destination_client.post.assert_not_awaited() + + +def test_integration_pipeline_not_observed_after_intake_raises_typed_bootstrap_skip(mock_config): + mock_config.destination_logs_intake_url = None + mock_config.destination_client.url_object.subdomain = "api" + + lp = LogsPipelines(mock_config) + lp.destination_integration_pipelines = {} + + mock_config.destination_client.get = AsyncMock(return_value=[]) + mock_config.destination_client.post = AsyncMock() + mock_config.destination_client.post_unauthenticated = AsyncMock() + + with patch("datadog_sync.model.logs_pipelines.sleep", new_callable=AsyncMock): + with pytest.raises(SkipResource) as exc_info: + asyncio.run(lp.create_resource("src-nginx", _make_integration_resource())) + + assert exc_info.value.failure_class == FAILURE_CLASS_INTEGRATION_PIPELINE_BOOTSTRAP_REQUIRED + assert exc_info.value.outcome_details == { + "pipeline_name": "nginx", + "ddsource": "nginx", + } + mock_config.destination_client.post.assert_awaited_once() + mock_config.destination_client.post_unauthenticated.assert_not_awaited() + assert mock_config.destination_client.get.await_count == 12 + + # ── G/G 3: existing subdomain construction unchanged ───────────────────────── @@ -278,8 +331,7 @@ def test_integration_pipeline_with_no_diff_skips_update_and_no_warn(mock_config, mock_config.destination_client.put.assert_not_awaited() integ_warnings = [ - r for r in caplog.records - if r.levelname == "WARNING" and "integration pipeline" in r.getMessage() + r for r in caplog.records if r.levelname == "WARNING" and "integration pipeline" in r.getMessage() ] assert integ_warnings == [] mock_config.destination_client.send_metric.assert_not_awaited() @@ -425,8 +477,7 @@ def test_update_read_only_pipeline_no_diff_still_raises_skip(mock_config, caplog mock_config.destination_client.put.assert_not_awaited() # No diff → no divergence log or metric (guard fires silently). integ_warnings = [ - r for r in caplog.records - if r.levelname == "WARNING" and "integration pipeline" in r.getMessage() + r for r in caplog.records if r.levelname == "WARNING" and "integration pipeline" in r.getMessage() ] assert integ_warnings == [] mock_config.destination_client.send_metric.assert_not_awaited() diff --git a/tests/unit/test_metric_percentiles.py b/tests/unit/test_metric_percentiles.py index 4eb1ae0f..dcbd98e8 100644 --- a/tests/unit/test_metric_percentiles.py +++ b/tests/unit/test_metric_percentiles.py @@ -10,7 +10,11 @@ import pytest from datadog_sync.model.metric_percentiles import MetricPercentiles -from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_DESTINATION_METRIC_MISSING, + CustomClientHTTPError, + SkipResource, +) def _run(coro): @@ -78,6 +82,11 @@ def test_update_resource_missing_destination_metric_patch_raises_skip(metric_per assert "custom.metric" in str(exc_info.value) assert "not present on destination" in str(exc_info.value) + assert exc_info.value.failure_class == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert exc_info.value.outcome_details == { + "metric_name": "custom.metric", + "operation": "percentiles_enable", + } client.get.assert_not_awaited() client.patch.assert_awaited_once() @@ -97,6 +106,11 @@ def test_update_resource_metric_not_found_patch_raises_skip(metric_percentiles): assert "custom.metric" in str(exc_info.value) assert "not present on destination" in str(exc_info.value) + assert exc_info.value.failure_class == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert exc_info.value.outcome_details == { + "metric_name": "custom.metric", + "operation": "percentiles_enable", + } client.get.assert_not_awaited() client.patch.assert_awaited_once() diff --git a/tests/unit/test_metric_tag_configurations.py b/tests/unit/test_metric_tag_configurations.py index 93ef628f..a91aae0c 100644 --- a/tests/unit/test_metric_tag_configurations.py +++ b/tests/unit/test_metric_tag_configurations.py @@ -10,7 +10,11 @@ import pytest from datadog_sync.model.metric_tag_configurations import MetricTagConfigurations -from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_DESTINATION_METRIC_MISSING, + CustomClientHTTPError, + SkipResource, +) def _run(coro): @@ -69,6 +73,11 @@ def test_create_resource_missing_destination_metric_raises_skip(metric_tag_confi assert "missing.metric" in str(exc_info.value) assert "not present on destination" in str(exc_info.value) + assert exc_info.value.failure_class == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert exc_info.value.outcome_details == { + "metric_name": "missing.metric", + "operation": "tag_configuration_create", + } client.post.assert_awaited_once() client.get.assert_not_awaited() client.patch.assert_not_awaited() @@ -124,6 +133,11 @@ def test_update_resource_missing_destination_metric_raises_skip(metric_tag_confi assert "missing.metric" in str(exc_info.value) assert "not present on destination" in str(exc_info.value) + assert exc_info.value.failure_class == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert exc_info.value.outcome_details == { + "metric_name": "missing.metric", + "operation": "tag_configuration_update", + } client.patch.assert_awaited_once() diff --git a/tests/unit/test_metrics_metadata.py b/tests/unit/test_metrics_metadata.py index 54c0932b..04787cfd 100644 --- a/tests/unit/test_metrics_metadata.py +++ b/tests/unit/test_metrics_metadata.py @@ -12,7 +12,11 @@ import pytest from datadog_sync.model.metrics_metadata import MetricsMetadata -from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource +from datadog_sync.utils.resource_utils import ( + FAILURE_CLASS_DESTINATION_METRIC_MISSING, + CustomClientHTTPError, + SkipResource, +) def _http_error(status, message="err"): @@ -53,6 +57,12 @@ def test_update_resource_dest_missing_raises_skip(metrics_metadata): assert "missing.metric" in str(exc_info.value) assert "not present on destination" in str(exc_info.value) + assert exc_info.value.failure_class == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert exc_info.value.outcome_reason == FAILURE_CLASS_DESTINATION_METRIC_MISSING + assert exc_info.value.outcome_details == { + "metric_name": "missing.metric", + "operation": "metadata_update", + } client.put.assert_not_awaited() From 336bea12e730dc06f166ee8ee56c244bff462016 Mon Sep 17 00:00:00 2001 From: "riyaz.shiraguppi" Date: Tue, 25 Aug 2026 14:36:25 -0500 Subject: [PATCH 2/2] Omit empty outcome details --- datadog_sync/utils/resources_handler.py | 78 ++++++++----------------- 1 file changed, 24 insertions(+), 54 deletions(-) diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py index 4bfbce59..3705e149 100644 --- a/datadog_sync/utils/resources_handler.py +++ b/datadog_sync/utils/resources_handler.py @@ -546,15 +546,10 @@ async def _apply_resource_cb(self, q_item: List) -> None: self.config.logger.info(f"skipping resource: {str(e)}", resource_type=resource_type, _id=_id) self.worker.counter.increment_skipped() _reason, _fc = self._sanitize_reason(e) - self._emit( - resource_type, - _id, - "sync", - "skipped", - reason=_reason, - failure_class=_fc, - details=e.outcome_details, - ) + emit_kwargs = {"reason": _reason, "failure_class": _fc} + if e.outcome_details: + emit_kwargs["details"] = e.outcome_details + self._emit(resource_type, _id, "sync", "skipped", **emit_kwargs) await r_class._send_action_metrics(Command.SYNC.value, _id, Status.SKIPPED.value, tags=["reason:unknown"]) except ResourceConnectionError as e: self.config.logger.error( @@ -651,15 +646,10 @@ async def _diffs_worker_cb(self, q_item: List) -> None: self.config.logger.warning(f"skipping resource: resource_type:{resource_type} id:{_id}") self.config.logger.debug(str(e)) _reason, _fc = self._sanitize_reason(e) - self._emit( - resource_type, - _id, - "sync", - "skipped", - reason=_reason, - failure_class=_fc, - details=e.outcome_details, - ) + emit_kwargs = {"reason": _reason, "failure_class": _fc} + if e.outcome_details: + emit_kwargs["details"] = e.outcome_details + self._emit(resource_type, _id, "sync", "skipped", **emit_kwargs) return try: @@ -942,15 +932,10 @@ async def _import_resource(self, q_item: List) -> None: # a downstream cascade would not want to grep these ids. self.worker.counter.increment_skipped() _reason, _fc = self._sanitize_reason(e) - self._emit( - resource_type, - _id, - "import", - "skipped", - reason=_reason, - failure_class=_fc, - details=e.outcome_details, - ) + emit_kwargs = {"reason": _reason, "failure_class": _fc} + if e.outcome_details: + emit_kwargs["details"] = e.outcome_details + self._emit(resource_type, _id, "import", "skipped", **emit_kwargs) await r_class._send_action_metrics(Command.IMPORT.value, _id, Status.SKIPPED.value) self.config.logger.info(f"skipping resource: {str(e)}", resource_type=resource_type, _id=_id) self.config.logger.debug(str(e)) @@ -1182,15 +1167,10 @@ async def _import_missing_dep_cb(self, q_item: Tuple[str, str]) -> None: self._emit(resource_type, _id, "import", "success") except SkipResource as e: _reason, _fc = self._sanitize_reason(e) - self._emit( - resource_type, - _id, - "import", - "skipped", - reason=_reason, - failure_class=_fc, - details=e.outcome_details, - ) + emit_kwargs = {"reason": _reason, "failure_class": _fc} + if e.outcome_details: + emit_kwargs["details"] = e.outcome_details + self._emit(resource_type, _id, "import", "skipped", **emit_kwargs) self.config.logger.info(f"skipping dependency: {str(e)}", resource_type=resource_type, _id=_id) return except CustomClientHTTPError as e: @@ -1234,15 +1214,10 @@ async def _force_missing_dep_import_cb(self, q_item: List): self._emit(resource_type, _id, "import", "success") except SkipResource as e: _reason, _fc = self._sanitize_reason(e) - self._emit( - resource_type, - _id, - "import", - "skipped", - reason=_reason, - failure_class=_fc, - details=e.outcome_details, - ) + emit_kwargs = {"reason": _reason, "failure_class": _fc} + if e.outcome_details: + emit_kwargs["details"] = e.outcome_details + self._emit(resource_type, _id, "import", "skipped", **emit_kwargs) self.config.logger.info(f"skipping dependency: {str(e)}", resource_type=resource_type, _id=_id) return except CustomClientHTTPError as e: @@ -1287,15 +1262,10 @@ async def _cleanup_worker(self, q_item: List) -> None: # cascade signal. Numeric-only accounting. self.worker.counter.increment_skipped() _reason, _fc = self._sanitize_reason(e) - self._emit( - resource_type, - _id, - "delete", - "skipped", - reason=_reason, - failure_class=_fc, - details=e.outcome_details, - ) + emit_kwargs = {"reason": _reason, "failure_class": _fc} + if e.outcome_details: + emit_kwargs["details"] = e.outcome_details + self._emit(resource_type, _id, "delete", "skipped", **emit_kwargs) await r_class._send_action_metrics("delete", _id, Status.SKIPPED.value, tags=["reason:unknown"]) self.config.logger.info(f"skipping resource: {str(e)}", resource_type=resource_type, _id=_id) self.config.logger.info(f"skip deleting resource: {str(e)}", resource_type=resource_type, _id=_id)