From 82d8d5993279df387ce5955024c70ad641cb0668 Mon Sep 17 00:00:00 2001 From: michael-richey <41595765+michael-richey@users.noreply.github.com> Date: Mon, 3 Aug 2026 13:21:22 -0400 Subject: [PATCH] fix(downtime_schedules): reconcile destination conflicts Recover an unambiguous duplicate create by fetching and persisting the existing destination downtime. Reject ambiguous duplicate responses so unmanaged resources are not silently accepted. When an update reports that its mapped downtime is gone, recreate it and replace the stale mapping. Other 404 responses continue through normal failure accounting. Keep delete 404 handling idempotent. Add regression coverage for state replacement, duplicate reconciliation, ambiguous conflicts, unrelated 404 responses, and sibling progress. --- datadog_sync/model/downtime_schedules.py | 155 ++++++++--- datadog_sync/utils/resource_utils.py | 1 + .../test_downtime_schedules_conflict_skip.py | 248 ++++++++++++++++++ 3 files changed, 368 insertions(+), 36 deletions(-) create mode 100644 tests/unit/test_downtime_schedules_conflict_skip.py diff --git a/datadog_sync/model/downtime_schedules.py b/datadog_sync/model/downtime_schedules.py index 5b2beaf4..b2ee07cb 100644 --- a/datadog_sync/model/downtime_schedules.py +++ b/datadog_sync/model/downtime_schedules.py @@ -3,17 +3,35 @@ # This product includes software developed at Datadog (https://www.datadoghq.com/). # Copyright 2019 Datadog, Inc. from __future__ import annotations +import json +import logging +import re from typing import TYPE_CHECKING, Optional, List, Dict, Tuple from datetime import datetime, timedelta, timezone from dateutil.parser import parse +from datadog_sync.constants import LOGGER_NAME from datadog_sync.utils.base_resource import BaseResource, ResourceConfig from datadog_sync.utils.custom_client import PaginationConfig -from datadog_sync.utils.resource_utils import DowntimeSchedulesDateOperator, SkipResource +from datadog_sync.utils.resource_utils import ( + CustomClientHTTPError, + DowntimeSchedulesDateOperator, + SkipResource, +) if TYPE_CHECKING: from datadog_sync.utils.custom_client import CustomClient +log = logging.getLogger(LOGGER_NAME) + +# Substring of the destination API's 400 body when a create collides with an +# equivalent existing downtime, e.g. +# {"errors":["The downtime being created is a duplicate of one or more +# existing downtimes: ['']"]} +_DUPLICATE_DOWNTIME_MARKER = "duplicate of one or more existing downtimes" +_DOWNTIME_NOT_FOUND_MARKER = "Downtime not found" +_SINGLE_DUPLICATE_ID_RE = re.compile(r"existing downtimes:\s*\[\s*(['\"])(?P[^'\"]+)\1\s*\]\s*$") + class DowntimeSchedules(BaseResource): resource_type = "downtime_schedules" @@ -87,35 +105,66 @@ def _parse_utc(value): def _iso_utc(dt) -> str: return dt.isoformat().replace("+00:00", "Z") + @staticmethod + def _http_error_messages(error: CustomClientHTTPError) -> List[str]: + """Return string messages from a JSON API error response.""" + body = error.response_body + if not isinstance(body, str): + return [] + try: + parsed = json.loads(body) + except (json.JSONDecodeError, TypeError): + return [] + errors = parsed.get("errors", []) if isinstance(parsed, dict) else [] + return [message for message in errors if isinstance(message, str)] + + @classmethod + def _single_duplicate_id(cls, error: CustomClientHTTPError) -> Optional[str]: + for message in cls._http_error_messages(error): + if _DUPLICATE_DOWNTIME_MARKER not in message: + continue + match = _SINGLE_DUPLICATE_ID_RE.search(message) + if match: + return match.group("id") + return None + + @classmethod + def _is_downtime_not_found(cls, error: CustomClientHTTPError) -> bool: + return any(_DOWNTIME_NOT_FOUND_MARKER in message for message in cls._http_error_messages(error)) + + def _normalize_create_schedule(self, _id: str, resource: Dict) -> None: + schedule = resource["attributes"].get("schedule") + if not schedule: + return + now = datetime.now(timezone.utc) + + # Past `end` means the maintenance window has already closed on the + # source. Replicating it to the destination would either invent a + # new customer-visible maintenance (if we shifted `end` forward) or + # 400 with "Downtime cannot be scheduled in the past". Skip: an + # ended downtime has nothing left to silence. + end_raw = schedule.get("end") + if end_raw: + end_dt = self._parse_utc(end_raw) + if end_dt <= now: + raise SkipResource( + str(_id), + self.resource_type, + "Downtime end is in the past.", + ) + + # Rewrite past `start` forward to now+60s. `end` (if present) is + # left as-is per customer intent — the window may shrink but its + # original end time is preserved. + start_raw = schedule.get("start") + if start_raw: + start_dt = self._parse_utc(start_raw) + if start_dt <= now: + schedule["start"] = self._iso_utc(now + timedelta(seconds=60)) + async def pre_resource_action_hook(self, _id, resource: Dict) -> None: if _id not in self.config.state.destination[self.resource_type]: - schedule = resource["attributes"].get("schedule") - if not schedule: - return - now = datetime.now(timezone.utc) - - # Past `end` means the maintenance window has already closed on the - # source. Replicating it to the destination would either invent a - # new customer-visible maintenance (if we shifted `end` forward) or - # 400 with "Downtime cannot be scheduled in the past". Skip: an - # ended downtime has nothing left to silence. - end_raw = schedule.get("end") - if end_raw: - end_dt = self._parse_utc(end_raw) - if end_dt <= now: - raise SkipResource( - str(_id), self.resource_type, - "Downtime end is in the past.", - ) - - # Rewrite past `start` forward to now+60s. `end` (if present) is - # left as-is per customer intent — the window may shrink but its - # original end time is preserved. - start_raw = schedule.get("start") - if start_raw: - start_dt = self._parse_utc(start_raw) - if start_dt <= now: - schedule["start"] = self._iso_utc(now + timedelta(seconds=60)) + self._normalize_create_schedule(_id, resource) else: # If start or end times of the resource are in the past, we set to the current destination `start` and `end` # this is to avoid unnecessary diff outputs @@ -139,7 +188,19 @@ async def pre_apply_hook(self) -> None: async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: destination_client = self.config.destination_client payload = {"data": resource} - resp = await destination_client.post(self.resource_config.base_path, payload) + try: + resp = await destination_client.post(self.resource_config.base_path, payload) + except CustomClientHTTPError as e: + duplicate_id = self._single_duplicate_id(e) if e.status_code == 400 else None + if duplicate_id is not None: + # The API identified exactly one equivalent destination + # downtime. Fetch and return it so BaseResource persists the + # recovered source-to-destination mapping. Ambiguous or + # malformed duplicate responses still propagate as failures. + existing = await destination_client.get(self.resource_config.base_path + f"/{duplicate_id}") + log.info(f"[downtime_schedules - {_id}] reconciled duplicate with existing destination downtime") + return _id, existing["data"] + raise return _id, resp["data"] @@ -147,18 +208,40 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]: destination_client = self.config.destination_client resource["id"] = self.config.state.destination[self.resource_type][_id]["id"] payload = {"data": resource} - resp = await destination_client.patch( - self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}", - payload, - ) + try: + resp = await destination_client.patch( + self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}", + payload, + ) + except CustomClientHTTPError as e: + if e.status_code == 404 and self._is_downtime_not_found(e): + # The mapped destination downtime was removed out-of-band, so + # the PATCH target no longer exists ("Downtime not found"). + # Recreate it now and return the new destination object so the + # BaseResource wrapper replaces the stale persisted mapping. + # Re-run create-only schedule normalization because the first + # pre-action hook took the update branch. + resource.pop("id", None) + self._normalize_create_schedule(_id, resource) + log.info(f"[downtime_schedules - {_id}] recreating missing mapped downtime on destination") + return await self.create_resource(_id, resource) + raise return _id, resp["data"] async def delete_resource(self, _id: str) -> None: destination_client = self.config.destination_client - await destination_client.delete( - self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}" - ) + try: + await destination_client.delete( + self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}" + ) + except CustomClientHTTPError as e: + if e.status_code == 404: + # Already gone on the destination: deleting a non-existent + # downtime is a successful no-op, not a failure. + log.info(f"[downtime_schedules - {_id}] already deleted on destination") + return + raise def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]: return super(DowntimeSchedules, self).connect_id(key, r_obj, resource_to_connect) diff --git a/datadog_sync/utils/resource_utils.py b/datadog_sync/utils/resource_utils.py index aa550b78..0dae7813 100644 --- a/datadog_sync/utils/resource_utils.py +++ b/datadog_sync/utils/resource_utils.py @@ -91,6 +91,7 @@ class CustomClientHTTPError(Exception): def __init__(self, response, message=None): super().__init__(f"{response.status} {response.message} - {message}") self.status_code = response.status + self.response_body = message class LogsPipelinesOrderIdsComparator(BaseOperator): diff --git a/tests/unit/test_downtime_schedules_conflict_skip.py b/tests/unit/test_downtime_schedules_conflict_skip.py new file mode 100644 index 00000000..d3450646 --- /dev/null +++ b/tests/unit/test_downtime_schedules_conflict_skip.py @@ -0,0 +1,248 @@ +# 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 downtime_schedules apply-time conflict/gone handling. + +Two deterministic destination-API rejections must not fail the whole +downtime_schedules resource for a batch: + +1. create path -> 400 "... duplicate of one or more existing downtimes ...": + reconcile an unambiguous existing downtime into destination state; reject + missing or multiple candidate IDs instead of accepting an unmanaged object. +2. update/delete path -> 404 "Downtime not found": the mapped destination + downtime was removed out-of-band. Recreate it on update so the stale mapping + is replaced, and treat delete as an idempotent no-op. + +Non-matching 4xx/5xx must still propagate so the retry layer and failure +accounting engage. +""" + +import asyncio +import logging +from types import SimpleNamespace + +import pytest + +from datadog_sync.constants import LOGGER_NAME +from datadog_sync.model.downtime_schedules import DowntimeSchedules +from datadog_sync.utils.resource_utils import CustomClientHTTPError + +DUPLICATE_BODY = ( + '{"errors":["The downtime being created is a duplicate of one or more ' + "existing downtimes: ['downtime-existing']\"]}" +) +NOT_FOUND_BODY = '{"errors":["Downtime not found"]}' + + +def _run(coro): + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +def _http_error(status, message): + return CustomClientHTTPError(SimpleNamespace(status=status, message="err"), message=message) + + +def _make_resource(): + return {"attributes": {"schedule": {"start": "2999-01-01T00:00:00Z"}}} + + +@pytest.fixture +def downtime(mock_config): + return DowntimeSchedules(mock_config) + + +# --- create path: duplicate 400 -------------------------------------------- + + +def test_create_duplicate_400_reconciles_single_existing_downtime(downtime): + downtime.config.destination_client.post = _http_error_raiser(400, DUPLICATE_BODY) + + async def _get(path): + assert path == "/api/v2/downtime/downtime-existing" + return {"data": {"id": "downtime-existing", "type": "downtime"}} + + downtime.config.destination_client.get = _get + _run(downtime._create_resource("src-id", _make_resource())) + + assert downtime.config.state.destination["downtime_schedules"]["src-id"] == { + "id": "downtime-existing", + "type": "downtime", + } + + +def test_create_duplicate_400_logs_reconciliation_at_info(downtime, caplog): + downtime.config.destination_client.post = _http_error_raiser(400, DUPLICATE_BODY) + + async def _get(_path): + return {"data": {"id": "downtime-existing", "type": "downtime"}} + + downtime.config.destination_client.get = _get + with caplog.at_level(logging.INFO, logger=LOGGER_NAME): + _run(downtime.create_resource("src-id", _make_resource())) + recs = [r for r in caplog.records if r.name == LOGGER_NAME and "src-id" in r.getMessage()] + assert recs, "expected an INFO reconciliation log identifying the downtime" + assert all(r.levelno == logging.INFO for r in recs) + + +@pytest.mark.parametrize( + "body", + [ + '{"errors":["duplicate of one or more existing downtimes"]}', + '{"errors":["The downtime being created is a duplicate of one or more existing downtimes: ' + "['dest-one', 'dest-two']\"]}", + ], +) +def test_create_duplicate_without_one_unambiguous_id_propagates(downtime, body): + downtime.config.destination_client.post = _http_error_raiser(400, body) + with pytest.raises(CustomClientHTTPError) as exc: + _run(downtime.create_resource("src-id", _make_resource())) + assert exc.value.status_code == 400 + + +def test_create_non_duplicate_400_propagates(downtime): + downtime.config.destination_client.post = _http_error_raiser( + 400, '{"errors":["Downtime cannot be scheduled in the past"]}' + ) + with pytest.raises(CustomClientHTTPError) as exc: + _run(downtime.create_resource("src-id", _make_resource())) + assert exc.value.status_code == 400 + + +def test_create_500_propagates(downtime): + downtime.config.destination_client.post = _http_error_raiser(500, "Internal Server Error") + with pytest.raises(CustomClientHTTPError) as exc: + _run(downtime.create_resource("src-id", _make_resource())) + assert exc.value.status_code == 500 + + +def test_create_200_unchanged(downtime): + async def _ok(_path, _payload): + return {"data": {"id": "dest-id", "type": "downtime"}} + + downtime.config.destination_client.post = _ok + _id, data = _run(downtime.create_resource("src-id", _make_resource())) + assert _id == "src-id" + assert data == {"id": "dest-id", "type": "downtime"} + + +# --- update path: not-found 404 -------------------------------------------- + + +def _seed_dest(downtime, _id="src-id", dest_id="dest-id"): + downtime.config.state.destination["downtime_schedules"][_id] = {"id": dest_id} + + +def test_update_not_found_404_recreates_and_replaces_stale_mapping(downtime): + _seed_dest(downtime) + downtime.config.destination_client.patch = _http_error_raiser(404, NOT_FOUND_BODY) + posted = [] + + async def _post(_path, payload): + posted.append(payload) + return {"data": {"id": "replacement-id", "type": "downtime"}} + + downtime.config.destination_client.post = _post + _run(downtime._update_resource("src-id", _make_resource())) + + assert posted[0]["data"].get("id") is None + assert downtime.config.state.destination["downtime_schedules"]["src-id"] == { + "id": "replacement-id", + "type": "downtime", + } + + +def test_update_not_found_404_logs_recreation_at_info(downtime, caplog): + _seed_dest(downtime) + downtime.config.destination_client.patch = _http_error_raiser(404, NOT_FOUND_BODY) + + async def _post(_path, _payload): + return {"data": {"id": "replacement-id", "type": "downtime"}} + + downtime.config.destination_client.post = _post + with caplog.at_level(logging.INFO, logger=LOGGER_NAME): + _run(downtime.update_resource("src-id", _make_resource())) + recs = [r for r in caplog.records if r.name == LOGGER_NAME and "src-id" in r.getMessage()] + assert recs and all(r.levelno == logging.INFO for r in recs) + + +def test_update_unrelated_404_propagates(downtime): + _seed_dest(downtime) + downtime.config.destination_client.patch = _http_error_raiser(404, '{"errors":["Route not found"]}') + with pytest.raises(CustomClientHTTPError) as exc: + _run(downtime.update_resource("src-id", _make_resource())) + assert exc.value.status_code == 404 + + +def test_update_400_propagates(downtime): + _seed_dest(downtime) + downtime.config.destination_client.patch = _http_error_raiser(400, "Bad Request") + with pytest.raises(CustomClientHTTPError) as exc: + _run(downtime.update_resource("src-id", _make_resource())) + assert exc.value.status_code == 400 + + +def test_update_200_unchanged(downtime): + _seed_dest(downtime) + + async def _ok(_path, _payload): + return {"data": {"id": "dest-id", "type": "downtime"}} + + downtime.config.destination_client.patch = _ok + _id, data = _run(downtime.update_resource("src-id", _make_resource())) + assert _id == "src-id" + assert data == {"id": "dest-id", "type": "downtime"} + + +# --- delete path: not-found 404 is a no-op --------------------------------- + + +def test_delete_404_is_noop(downtime): + _seed_dest(downtime) + downtime.config.destination_client.delete = _http_error_raiser(404, NOT_FOUND_BODY) + # Already gone == delete succeeded; must not raise. + assert _run(downtime.delete_resource("src-id")) is None + + +def test_delete_500_propagates(downtime): + _seed_dest(downtime) + downtime.config.destination_client.delete = _http_error_raiser(500, "Internal Server Error") + with pytest.raises(CustomClientHTTPError) as exc: + _run(downtime.delete_resource("src-id")) + assert exc.value.status_code == 500 + + +# --- batch resilience: one duplicate doesn't stop siblings ------------------ + + +def test_duplicate_does_not_block_siblings(downtime): + calls = {"n": 0} + + async def _post(_path, _payload): + calls["n"] += 1 + if calls["n"] == 2: + raise _http_error(400, DUPLICATE_BODY) + return {"data": {"id": f"dest-{calls['n']}"}} + + async def _get(_path): + return {"data": {"id": "downtime-existing"}} + + downtime.config.destination_client.post = _post + downtime.config.destination_client.get = _get + synced = [] + for src in ("a", "b", "c"): + _id, _ = _run(downtime.create_resource(src, _make_resource())) + synced.append(_id) + assert synced == ["a", "b", "c"] + + +def _http_error_raiser(status, message): + async def _raise(*_args, **_kwargs): + raise _http_error(status, message) + + return _raise