Emit typed bootstrap skip outcomes

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"]:

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
 

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
 

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
 

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}")