diff --git a/products/alerts/backend/platform_investigation.py b/products/alerts/backend/platform_investigation.py index 3911a17a596a..25f99b211e05 100644 --- a/products/alerts/backend/platform_investigation.py +++ b/products/alerts/backend/platform_investigation.py @@ -8,6 +8,7 @@ write, and runs once the evaluation is recorded. """ +from dataclasses import replace from datetime import UTC, datetime from typing import Any, Final from uuid import UUID @@ -17,17 +18,42 @@ from posthog.dataclasses import frozen from posthog.temporal.alerts.investigation import ( InvestigationDecision, + carried_verdict_suppresses, decide_within_budget, investigates, refusing_investigations, ) from products.alerts.backend.investigation_episode import STARTED_INVESTIGATION_STATUSES -from products.alerts.backend.investigation_release import earlier_in_episode, last_verdict +from products.alerts.backend.investigation_release import ( + RUNNING_HOLD, + earlier_in_episode, + last_verdict, + previous_verdict_in_episode, + release_held, +) from products.alerts.backend.models.alert import AlertConfiguration, InvestigationStatus from products.alerts.backend.models.investigation import InsightAlertInvestigation +from products.alerts.backend.platform_investigation_detail import ( + INVESTIGATION_REQUESTED_KEY, + InvestigationClaims, + InvestigationStart, + requested_investigation, +) +from products.alerts_platform.backend.facade.contracts import ( + AlertEventKind, + HeldDelivery, + HeldDeliveryAddress, + PlatformAlertCheckInput, + PlatformAlertOutcome, + SourceBatchEvaluation, +) +from products.alerts_platform.backend.facade.enums import HoldReleasedBy +from products.alerts_platform.backend.facade.lifecycle import AlertState, NotificationAction _CLAIM_LOCK_SQL: Final = "SELECT pg_advisory_xact_lock(hashtext(%s))" +HOLD_REASON: Final = "investigation" +_EVALUATED_KEYS: Final = ("anomaly_scores", "triggered_points", "triggered_dates", "interval", "triggered_metadata") # A verdict-change follow-up is recorded under the key of the firing it follows up plus this suffix, # so it gets a delivery of its own. _VERDICT_CHANGE_KEY_SUFFIX: Final = "#verdict" @@ -111,3 +137,104 @@ def claim_investigation( evaluated=evaluated, ) return InvestigationClaim(investigation_id=investigation.id, claimed=not refused) + + +def with_investigation(check: PlatformAlertCheckInput, evaluation: SourceBatchEvaluation) -> SourceBatchEvaluation: + """The evaluation of a check the platform owns, with the insight stack's investigation rules applied. + + A firing the agent investigates asks for its claim in `source_detail`, and a gated alert holds the + firing's delivery until the verdict. A firing past the budget is muted when the firing's last + verdict suppresses it. Only an owned check, because a parallel run must never pay for an agent run. + """ + if not check.owned_by_platform or check.legacy_configuration_id is None or len(evaluation.outcomes) != 1: + return evaluation + outcome = evaluation.outcomes[0] + # A skipped check keeps a firing alert's state without evaluating it, so it has nothing to investigate. + if outcome.skipped or len(outcome.groups) != 1 or outcome.groups[0].new_state != AlertState.FIRING.value: + return evaluation + alert = AlertConfiguration.objects.filter(team_id=check.team_id, id=check.legacy_configuration_id).first() + if alert is None: + return evaluation + group = outcome.groups[0] + decision = decide_investigation( + alert, + firing=True, + configuration_id=check.id, + evaluation_key=outcome.evaluation_key, + episode_started_at=group.firing_episode.started_at if group.firing_episode else None, + ) + if decision.should_investigate: + group = replace(group, source_detail={**group.source_detail, INVESTIGATION_REQUESTED_KEY: str(alert.id)}) + held = None + if evaluation.deliveries and alert.investigation_gates_notifications: + (request,) = evaluation.deliveries + held = HeldDelivery( + request=request, reason=HOLD_REASON, release_after_seconds=int(RUNNING_HOLD.total_seconds()) + ) + return replace( + evaluation, + outcomes=(replace(outcome, groups=(group,), held_delivery=held),), + deliveries=() if held else evaluation.deliveries, + ) + if group.kind == AlertEventKind.FIRING and carried_verdict_suppresses(alert, decision): + # As the insight stack: the firing still moves the alert's state and `last_notified_at`. + group = replace(group, kind=AlertEventKind.CHECK, muted_notification=NotificationAction.FIRE.value) + return replace(evaluation, outcomes=(replace(outcome, groups=(group,)),), deliveries=()) + return evaluation + + +def claim_requested_investigations(team_id: int, outcomes: tuple[PlatformAlertOutcome, ...]) -> InvestigationClaims: + """Claims the investigation each recorded firing asked for, and returns the agent runs to start. + + A refused claim releases the firing's held delivery by the firing's last verdict, as the insight + stack mutes a check that runs no investigation of its own. + """ + starts: list[InvestigationStart] = [] + deliveries = [] + for outcome in outcomes: + requested = requested_investigation(outcome) + if requested is None: + continue + alert = AlertConfiguration.objects.filter(team_id=team_id, id=requested.alert_id).first() + if alert is None: + # The safety net sends the held firing at its deadline. + continue + group = outcome.groups[0] + claim = claim_investigation( + alert, + configuration_id=outcome.configuration_id, + evaluation_key=outcome.evaluation_key, + episode_started_at=requested.episode_started_at, + kind=InsightAlertInvestigation.Kind.AGENT, + evaluated={ + "value": group.value, + **{key: group.source_detail[key] for key in _EVALUATED_KEYS if key in group.source_detail}, + }, + ) + if claim.claimed: + starts.append( + InvestigationStart( + team_id=team_id, + alert_id=str(alert.id), + investigation_id=str(claim.investigation_id), + configuration_id=str(outcome.configuration_id), + evaluation_key=outcome.evaluation_key, + user_id=alert.created_by_id, + ) + ) + continue + if outcome.held_delivery is None: + continue + previous_verdict = previous_verdict_in_episode( + team_id, outcome.configuration_id, requested.episode_started_at, outcome.evaluation_key + ) + request = release_held( + HeldDeliveryAddress( + team_id=team_id, configuration_id=str(outcome.configuration_id), evaluation_key=outcome.evaluation_key + ), + suppress=carried_verdict_suppresses(alert, InvestigationDecision(previous_verdict=previous_verdict)), + released_by=HoldReleasedBy.CLAIM_REFUSED, + ) + if request is not None: + deliveries.append(request) + return InvestigationClaims(starts=tuple(starts), deliveries=tuple(deliveries)) diff --git a/products/alerts/backend/platform_investigation_detail.py b/products/alerts/backend/platform_investigation_detail.py new file mode 100644 index 000000000000..c31b4db8a3f2 --- /dev/null +++ b/products/alerts/backend/platform_investigation_detail.py @@ -0,0 +1,65 @@ +"""The investigation an insight firing asks for in its `source_detail`, claimed after the record. + +A check only decides that its firing gets an investigation, because its evaluation can be retried. +The evaluation workflow claims it once the firing is recorded, and starts the agent. This module +has no Django imports, so the workflow can call it. +""" + +from datetime import datetime + +from posthog.dataclasses import frozen + +from products.alerts_platform.backend.facade.contracts import AlertDeliveryRequest, PlatformAlertOutcome + +# The legacy alert whose agent investigates the firing. Recorded with the firing, so history shows +# which checks asked for an investigation. +INVESTIGATION_REQUESTED_KEY = "investigation_alert_id" + + +@frozen +class InvestigationStart: + """An agent run the claim granted, as the workflow starts it.""" + + team_id: int + alert_id: str + investigation_id: str + configuration_id: str + evaluation_key: str + user_id: int | None = None + + @property + def workflow_id(self) -> str: + return f"anomaly-investigation-{self.configuration_id}:{self.evaluation_key}" + + def workflow_inputs(self) -> dict[str, object]: + """The agent's `AnomalyInvestigationWorkflowInputs`, as JSON. The agent's module imports Django, + which the workflow sandbox refuses, so the inputs cross by name.""" + return { + "team_id": self.team_id, + "alert_id": self.alert_id, + "investigation_id": self.investigation_id, + "user_id": self.user_id, + } + + +@frozen +class InvestigationClaims: + starts: tuple[InvestigationStart, ...] = () + # Held firings whose claim was refused, released by the firing's last verdict and due to start. + deliveries: tuple[AlertDeliveryRequest, ...] = () + + +@frozen +class RequestedInvestigation: + alert_id: str + episode_started_at: datetime | None + + +def requested_investigation(outcome: PlatformAlertOutcome) -> RequestedInvestigation | None: + """The investigation an outcome asks for, or None. Only an ungrouped firing asks for one.""" + for group in outcome.groups: + alert_id = group.source_detail.get(INVESTIGATION_REQUESTED_KEY) + if alert_id is not None: + started_at = group.firing_episode.started_at if group.firing_episode else None + return RequestedInvestigation(alert_id=str(alert_id), episode_started_at=started_at) + return None diff --git a/products/alerts/backend/platform_source_cycle.py b/products/alerts/backend/platform_source_cycle.py index 8a22dfa868fc..f573af00090d 100644 --- a/products/alerts/backend/platform_source_cycle.py +++ b/products/alerts/backend/platform_source_cycle.py @@ -59,6 +59,7 @@ from products.alerts.backend.models.alert import AlertConfiguration from products.alerts.backend.platform_alert_wording import insight_alert_url from products.alerts.backend.platform_chart_detail import CHART_PENDING_KEY, RenderedChart +from products.alerts.backend.platform_investigation import with_investigation from products.alerts_platform.backend.facade.api import due_check_ids, due_checks, refresh_settings, slot_of from products.alerts_platform.backend.facade.contracts import ( SOURCE_CONDITION_KEY, @@ -495,10 +496,17 @@ def evaluate_insight_check( raise if outcome is None: return None - return _evaluation(checks[0], outcome) + evaluation = _evaluation(checks[0], outcome) finally: if not retrying: release_evaluation_slot(configuration_id, held_until=held_until, key=INFLIGHT_KEY) + # After the slot is free, because the investigation's reads need no ClickHouse slot. + try: + return with_investigation(checks[0], evaluation) + except Exception: + # The firing still goes out, uninvestigated, rather than the check being lost as errored. + logger.exception("Failed to decide a platform insight firing's investigation", check_id=configuration_id) + return evaluation def fail_insight_check( diff --git a/products/alerts/backend/temporal/platform_evaluate.py b/products/alerts/backend/temporal/platform_evaluate.py index b32797a376d6..a54cee05deb4 100644 --- a/products/alerts/backend/temporal/platform_evaluate.py +++ b/products/alerts/backend/temporal/platform_evaluate.py @@ -16,7 +16,7 @@ from temporalio import activity, workflow from temporalio.common import RetryPolicy -from temporalio.exceptions import ActivityError +from temporalio.exceptions import ActivityError, WorkflowAlreadyStartedError from posthog.dataclasses import frozen from posthog.temporal.common.base import PostHogWorkflow @@ -35,6 +35,11 @@ from posthog.temporal.common.utils import close_db_connections from products.alerts.backend.platform_chart_detail import RenderedChart, pending_chart_alert_id, with_chart + from products.alerts.backend.platform_investigation_detail import ( + InvestigationClaims, + InvestigationStart, + requested_investigation, + ) from products.alerts_platform.backend.facade.contracts import ( RECORD_OUTCOMES_ACTIVITY, AlertDeliveryRequest, @@ -83,6 +88,8 @@ # and never holds it back. It is best effort, so it gets what the batch budget leaves. STOP_QUERIES_START_TO_CLOSE = dt.timedelta(seconds=60) RECORD_START_TO_CLOSE = dt.timedelta(seconds=8) +CLAIM_START_TO_CLOSE = dt.timedelta(seconds=10) +CLAIM_SCHEDULE_TO_CLOSE = dt.timedelta(seconds=20) RECORD_SCHEDULE_TO_CLOSE = dt.timedelta(seconds=12) # The pool admits once per round, so a batch that admits only once evaluates at most one pool of # checks per tick. Rounds refill the slots that finished checks free, the way production's @@ -99,7 +106,8 @@ + max(CHECK_SCHEDULE_TO_CLOSE, REAL_TIME_CHECK_SCHEDULE_TO_CLOSE) + max(CHART_SCHEDULE_TO_CLOSE, FAIL_SCHEDULE_TO_CLOSE) + RECORD_SCHEDULE_TO_CLOSE - + max(FAIL_SCHEDULE_TO_CLOSE + RECORD_SCHEDULE_TO_CLOSE, STOP_QUERIES_START_TO_CLOSE) + + CLAIM_SCHEDULE_TO_CLOSE + + max(FAIL_SCHEDULE_TO_CLOSE + RECORD_SCHEDULE_TO_CLOSE + CLAIM_SCHEDULE_TO_CLOSE, STOP_QUERIES_START_TO_CLOSE) ) _PATCH_READMISSION = "platform-insight-readmission-2026-10" # Runs started before it recorded every outcome in one activity, and replay must schedule the same. @@ -115,6 +123,8 @@ _PATCH_RECORD_AS_FINISHED = "platform-insight-record-as-finished-2026-10" # Runs started before it returned when their first round found the pool full, and replay must not poll. _PATCH_WAIT_FOR_POOL = "platform-insight-wait-for-pool-2026-10" +# Runs started before it claimed investigations after a record, and replay must not schedule a claim. +_PATCH_INVESTIGATIONS = "platform-insight-investigations-2026-11" @frozen @@ -159,6 +169,12 @@ class FailedInsightCheckInputs: error_type: str | None = None +@frozen +class ClaimInsightInvestigationsInputs: + team_id: int + outcomes: tuple[PlatformAlertOutcome, ...] + + @frozen class StopInsightQueriesInputs: team_id: int @@ -335,6 +351,15 @@ def _release_platform_llm_check(thread: "asyncio.Future[SourceBatchEvaluation | thread.exception() +@activity.defn +@close_db_connections +def claim_platform_insight_investigations_activity(inputs: ClaimInsightInvestigationsInputs) -> InvestigationClaims: + """Claims the investigations recorded firings asked for. A retry gets the first claim's answer.""" + from products.alerts.backend.platform_investigation import claim_requested_investigations # noqa: PLC0415 + + return claim_requested_investigations(inputs.team_id, inputs.outcomes) + + @activity.defn @close_db_connections def render_platform_insight_chart_activity(inputs: InsightChartInputs) -> RenderedChart | None: @@ -368,6 +393,7 @@ async def run(self, inputs: SourceEvaluationInputs) -> int: # From the workflow's clock, so a retried plan reuses the expiry and gets its slots back. A # check releases its slot only while the slot still carries this expiry. expires_at = (workflow.now() + source_evaluation_timeout(SourceKind.INSIGHT)).timestamp() + self._investigating: list[asyncio.Task[None]] = [] if workflow.patched(_PATCH_READMISSION) and workflow.patched(_PATCH_RECORD_AS_FINISHED): return await self._evaluate_and_record_in_rounds(inputs, expires_at) if workflow.patched(_PATCH_READMISSION): @@ -405,6 +431,7 @@ async def run(self, inputs: SourceEvaluationInputs) -> int: if not outcomes: return 0 await self._record(inputs, evaluations, chunked=workflow.patched(_PATCH_CHUNKED_RECORD)) + await asyncio.gather(*self._investigating) return len(outcomes) async def _with_charts( @@ -597,6 +624,7 @@ async def _evaluate_and_record_in_rounds(self, inputs: SourceEvaluationInputs, e continue recorded += await self._record_waiting(inputs, waiting, last_chance=True) await asyncio.gather(*stopping, return_exceptions=True) + await asyncio.gather(*self._investigating) return recorded running = [task for task in unrecorded.values() if not task.done()] if not running: @@ -751,6 +779,45 @@ async def _record_chunk( # Started only after the record, because a delivery reads the history row the record writes. recorded = {str(outcome.configuration_id) for outcome in chunk} await start_deliveries([delivery for delivery in deliveries if delivery.configuration_id in recorded]) + investigated = tuple(outcome for outcome in chunk if requested_investigation(outcome) is not None) + if workflow.patched(_PATCH_INVESTIGATIONS) and investigated: + # A task, so a slow claim holds back only the firings it holds, not the next record or round. + self._investigating.append(asyncio.create_task(self._investigate(inputs, investigated))) + + async def _investigate(self, inputs: SourceEvaluationInputs, outcomes: tuple[PlatformAlertOutcome, ...]) -> None: + claims = await self._claim_investigations(inputs, outcomes) + await asyncio.gather(*(self._start_investigation(start) for start in claims.starts)) + await start_deliveries(list(claims.deliveries)) + + async def _claim_investigations( + self, inputs: SourceEvaluationInputs, outcomes: tuple[PlatformAlertOutcome, ...] + ) -> InvestigationClaims: + try: + return await workflow.execute_activity( + claim_platform_insight_investigations_activity, + ClaimInsightInvestigationsInputs(team_id=inputs.batch_key.team_id, outcomes=outcomes), + start_to_close_timeout=CLAIM_START_TO_CLOSE, + schedule_to_close_timeout=CLAIM_SCHEDULE_TO_CLOSE, + retry_policy=RetryPolicy(maximum_attempts=3), + ) + except ActivityError: + # The outcomes are recorded, so a failure here must not record them again. Each held + # firing goes out at its hold's deadline, uninvestigated, as a firing whose agent never ran. + workflow.logger.warning("Could not claim the investigations of recorded platform insight firings") + return InvestigationClaims() + + async def _start_investigation(self, start: InvestigationStart) -> None: + try: + await workflow.start_child_workflow( + "anomaly-investigation", + start.workflow_inputs(), + id=start.workflow_id, + task_queue=settings.MAX_AI_TASK_QUEUE, + parent_close_policy=workflow.ParentClosePolicy.ABANDON, + ) + except WorkflowAlreadyStartedError: + # A retried record claims again and gets the same investigation back. + pass def _maximum_attempts(check: AdmittedInsightCheck) -> int: @@ -770,6 +837,7 @@ def _maximum_attempts(check: AdmittedInsightCheck) -> int: plan_platform_insight_checks_activity, evaluate_platform_insight_check_activity, render_platform_insight_chart_activity, + claim_platform_insight_investigations_activity, fail_platform_insight_check_activity, stop_platform_insight_queries_activity, ] diff --git a/products/alerts/backend/test/test_platform_source_cycle.py b/products/alerts/backend/test/test_platform_source_cycle.py index ccb3ac6f1390..76c1c4e71ad6 100644 --- a/products/alerts/backend/test/test_platform_source_cycle.py +++ b/products/alerts/backend/test/test_platform_source_cycle.py @@ -38,9 +38,12 @@ LLMDetectorUnavailableError, ) from products.alerts.backend.logic.alert_email import INSIGHT_ALERT_ERRORED_EVENT_ID -from products.alerts.backend.models.alert import AlertCheck, AlertConfiguration +from products.alerts.backend.models.alert import AlertCheck, AlertConfiguration, InvestigationStatus +from products.alerts.backend.models.investigation import InsightAlertInvestigation from products.alerts.backend.platform_alert_wording import insight_alert_url from products.alerts.backend.platform_chart_detail import CHART_PENDING_KEY, RenderedChart +from products.alerts.backend.platform_investigation import claim_requested_investigations +from products.alerts.backend.platform_investigation_detail import requested_investigation from products.alerts.backend.platform_source_cycle import ( CAPACITY_REJECTED, CHART_RENDER_INFLIGHT_KEY, @@ -882,3 +885,126 @@ def test_the_pool_defers_what_it_cannot_hold_until_a_check_frees_its_slot(self) ) ) assert replanned == [AdmittedInsightCheck(configuration_id=str(second.id), real_time=True)] + + def _investigated_alert(self, **overrides: Any) -> tuple[AlertConfiguration, PlatformConfigurationSnapshot]: + alert = self._alert( + detector_config={"type": "zscore", "threshold": 0.95, "window": 30}, + investigation_agent_enabled=True, + **overrides, + ) + return alert, self._copy(alert, owner=PlatformAlertOwner.PLATFORM) + + def _fire(self, configuration: PlatformConfigurationSnapshot) -> PlatformAlertOutcome: + with patch(f"{_MODULE}.render_alert_insight_chart"): + outcome, _ = self._evaluate( + configuration, + result=AlertEvaluationResult( + value=150.0, breaches=["above 100"], triggered_dates=["2026-09-16"], triggered_points=[29] + ), + ) + assert outcome is not None + return outcome + + @parameterized.expand( + [ + ("a gated alert holds its firing", True, PlatformAlertOwner.PLATFORM, True, True), + ("an ungated alert sends its firing", False, PlatformAlertOwner.PLATFORM, True, False), + ("a parallel run never investigates", True, PlatformAlertOwner.SOURCE, False, False), + ] + ) + def test_an_owned_firing_asks_for_an_investigation( + self, _name: str, gates: bool, owner: str, investigates: bool, held: bool + ) -> None: + alert = self._alert( + detector_config={"type": "zscore", "threshold": 0.95, "window": 30}, + investigation_agent_enabled=True, + investigation_gates_notifications=gates, + ) + configuration = self._copy(alert, owner=owner) + + outcome = self._fire(configuration) + + assert (requested_investigation(outcome) is not None) is investigates + assert (outcome.held_delivery is not None) is held + if owner == PlatformAlertOwner.PLATFORM: + assert len(self.deliveries) == (0 if held else 1) + + def test_a_firing_past_its_budget_is_muted_by_the_firing_last_verdict(self) -> None: + alert, configuration = self._investigated_alert(investigation_gates_notifications=True) + for index, verdict in enumerate(["true_positive", "true_positive", "false_positive"]): + InsightAlertInvestigation.objects.for_team(self.team.id).create( + team=self.team, + alert_configuration=alert, + configuration_id=configuration.id, + evaluation_key=f"slot:earlier-{index}", + episode_started_at=CUTOFF, + kind=InsightAlertInvestigation.Kind.AGENT, + status=InvestigationStatus.DONE, + verdict=verdict, + ) + + outcome = self._fire(configuration) + + assert requested_investigation(outcome) is None + assert (outcome.groups[0].kind, outcome.groups[0].muted_notification) == (AlertEventKind.CHECK, "fire") + assert self.deliveries == () + + @parameterized.expand( + [ + ("claimed", None, None, True), + ("refused with no verdict yet", InvestigationStatus.RUNNING, None, False), + ("refused after a false positive", InvestigationStatus.DONE, "false_positive", False), + ] + ) + def test_a_recorded_firing_claims_its_investigation_or_releases_its_hold( + self, _name: str, earlier: str | None, verdict: str | None, claimed: bool + ) -> None: + alert, configuration = self._investigated_alert(investigation_gates_notifications=True) + if earlier is not None: + InsightAlertInvestigation.objects.for_team(self.team.id).create( + team=self.team, + alert_configuration=alert, + configuration_id=configuration.id, + evaluation_key="slot:earlier", + episode_started_at=CUTOFF, + kind=InsightAlertInvestigation.Kind.AGENT, + status=earlier, + verdict=verdict, + ) + outcome = self._fire(configuration) + record_outcomes(self.team.id, (outcome,), CUTOFF) + + claims = claim_requested_investigations(self.team.id, (outcome,)) + + assert claims == claim_requested_investigations(self.team.id, (outcome,)) + assert len(claims.starts) == (1 if claimed else 0) + if claimed: + investigation = InsightAlertInvestigation.objects.for_team(self.team.id).get( + id=claims.starts[0].investigation_id + ) + assert investigation.evaluated["value"] == 150.0 + assert investigation.evaluated["triggered_dates"] == ["2026-09-16"] + assert len(claims.deliveries) == (1 if not claimed and verdict is None else 0) + + def test_a_firing_whose_investigation_cannot_be_decided_still_goes_out(self) -> None: + _alert, configuration = self._investigated_alert(investigation_gates_notifications=True) + + with patch(f"{_MODULE}.with_investigation", side_effect=RuntimeError("database went away")): + outcome = self._fire(configuration) + + assert outcome.held_delivery is None + assert len(self.deliveries) == 1 + + def test_a_skipped_check_of_a_firing_alert_asks_for_no_investigation(self) -> None: + _alert, configuration = self._investigated_alert(investigation_gates_notifications=True) + record_outcomes(self.team.id, (self._fire(configuration),), CUTOFF) + with team_scope(self.team.id): + platform_testing.set_due_at(configuration.id, CUTOFF - timedelta(minutes=1)) + + outcome, _ = self._evaluate( + configuration, result=AlertEvaluationResult(value=None, breaches=[], skipped_reason="not enough data") + ) + + assert outcome is not None + assert outcome.skipped + assert requested_investigation(outcome) is None diff --git a/products/alerts_platform/backend/delivery/evaluation.py b/products/alerts_platform/backend/delivery/evaluation.py index 41818a80ab1f..396dbb0592ce 100644 --- a/products/alerts_platform/backend/delivery/evaluation.py +++ b/products/alerts_platform/backend/delivery/evaluation.py @@ -189,6 +189,17 @@ def deliver_evaluation(request: AlertDeliveryRequest) -> DeliveryOutcome: ) if announced is None: return DeliveryOutcome(live=True, sent=0, skipped_without_transport=0) + if stale := delivery_holds.stale_held_triggers(request, announced.transitions): + logger.info( + "alerts_platform_delivery_stale_trigger_dropped", + configuration_id=request.configuration_id, + evaluation_key=request.evaluation_key, + groups=len(stale), + ) + request = replace( + request, + incident_actions={key: action for key, action in request.incident_actions.items() if key not in stale}, + ) platform_destinations = _platform_destinations(request) native_announced, emissions = _split_by_path(request, announced, platform_destinations) emitter = delivery_paths.emitter_for(request.source) if emissions else None diff --git a/products/alerts_platform/backend/logic/delivery_holds.py b/products/alerts_platform/backend/logic/delivery_holds.py index 23cff87d7962..4923283d5e71 100644 --- a/products/alerts_platform/backend/logic/delivery_holds.py +++ b/products/alerts_platform/backend/logic/delivery_holds.py @@ -7,7 +7,7 @@ """ from collections.abc import Sequence -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from typing import Final from django.db import transaction @@ -16,11 +16,13 @@ from products.alerts_platform.backend.facade.contracts import ( AlertDeliveryRequest, + AnnouncedTransition, HeldDeliveryAddress, + IncidentAction, PlatformAlertOutcome, ) -from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy -from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold +from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy, PlatformAlertState +from products.alerts_platform.backend.models import PlatformAlert, PlatformAlertConfiguration, PlatformAlertDeliveryHold from products.alerts_platform.backend.models.platform_alert_events_sql import PLATFORM_ALERT_EVENTS_TTL_DAYS # As long as the history rows a hold annotates live, so a reader never finds one without the other. @@ -178,6 +180,46 @@ def blocks_delivery(team_id: int, configuration_id: str, evaluation_key: str) -> ) +def stale_held_triggers(request: AlertDeliveryRequest, transitions: Sequence[AnnouncedTransition]) -> frozenset[str]: + """The groups of a held evaluation whose incident trigger must not go out. + + A hold can release after a later evaluation resolved the firing it held, and nothing orders the + two sends. A trigger sent then would open an incident that no resolve follows. So a trigger goes + out only while its group is still in the firing the held evaluation saw. The message still goes, + as a late firing does. + """ + triggering = {key for key, action in request.incident_actions.items() if action == IncidentAction.TRIGGER} + if not triggering: + return frozenset() + held = ( + PlatformAlertDeliveryHold.objects.for_team(request.team_id) + .filter(configuration_id=request.configuration_id, evaluation_key=request.evaluation_key) + .exists() + ) + if not held: + return frozenset() + current = { + grouping_key: (state, _in_utc(firing_started_at)) + for grouping_key, state, firing_started_at in PlatformAlert.objects.for_team(request.team_id) + .filter(configuration_id=request.configuration_id, grouping_key__in=triggering) + .values_list("grouping_key", "state", "firing_started_at") + } + return frozenset( + transition.grouping_key + for transition in transitions + if transition.grouping_key in triggering + and current.get(transition.grouping_key) != (PlatformAlertState.FIRING, _in_utc(transition.episode_started_at)) + ) + + +def _in_utc(moment: datetime | None) -> datetime | None: + # ClickHouse can return a history row's times without a zone, and they are UTC. A naive time + # never equals an aware one, so every comparison with a Postgres time would say "changed". + if moment is None or moment.tzinfo is not None: + return moment + return moment.replace(tzinfo=UTC) + + def prune_holds(now: datetime, *, limit: int) -> int: """Deletes up to `limit` holds older than the history rows they annotate. Returns how many.""" stale = list( diff --git a/products/alerts_platform/backend/tests/test_delivery_evaluation.py b/products/alerts_platform/backend/tests/test_delivery_evaluation.py index 2a061635e35e..6eca12d4839e 100644 --- a/products/alerts_platform/backend/tests/test_delivery_evaluation.py +++ b/products/alerts_platform/backend/tests/test_delivery_evaluation.py @@ -1,6 +1,6 @@ from contextlib import ExitStack from dataclasses import replace -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import Any, cast from uuid import UUID @@ -36,7 +36,7 @@ SourceKind, ) from products.alerts_platform.backend.facade.enums import HoldRelease, PlatformAlertOwner -from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold +from products.alerts_platform.backend.models import PlatformAlert, PlatformAlertConfiguration, PlatformAlertDeliveryHold from products.alerts_platform.backend.tests.recorded_metrics import recorded_metrics _MODULE = "products.alerts_platform.backend.delivery.evaluation" @@ -537,6 +537,61 @@ def test_a_held_evaluation_sends_only_once_released_with_a_send( assert outcome.sent == sent + @parameterized.expand( + [ + ("still_firing", "firing", FIRING, FIRING, [IncidentAction.TRIGGER]), + # ClickHouse returns the episode start without a zone. + ( + "still_firing_read_without_a_zone", + "firing", + FIRING, + FIRING.replace(tzinfo=None), + [IncidentAction.TRIGGER], + ), + ("resolved_since", "not_firing", None, FIRING, []), + ("firing_again_since", "firing", FIRING + timedelta(hours=1), FIRING, []), + ] + ) + def test_a_released_hold_opens_an_incident_only_inside_the_firing_it_held( + self, + _name: str, + state: str, + firing_started_at: datetime | None, + episode_started_at: datetime, + incident: list[IncidentAction], + ) -> None: + configuration = PlatformAlertConfiguration.objects.for_team(self.team.id).create( + id=CONFIGURATION_ID, + team=self.team, + name="API errors", + source_kind=PlatformAlertConfiguration.SourceKind.LOGS, + check_interval_minutes=10, + ) + PlatformAlert.objects.for_team(self.team.id).create( + team=self.team, configuration=configuration, state=state, firing_started_at=firing_started_at + ) + PlatformAlertDeliveryHold.objects.for_team(self.team.id).create( + team=self.team, + configuration=configuration, + evaluation_key="eval-1", + reason="investigation", + request={}, + held_at=FIRING, + release_after=FIRING, + released_at=FIRING, + release_action=HoldRelease.SEND, + ) + + self._run( + _announcement(_transition(AlertEventKind.FIRING, episode_started_at=episode_started_at)), + {FIRING_EVENT: [_group(SLACK)], OPENED_EVENT: [_group(PAGERDUTY)]}, + incident_actions={"": IncidentAction.TRIGGER}, + owned_by_platform=True, + ) + + assert [channel for channel, _ in RecordingTransport.sends if channel == "C-1"] == ["C-1"] + assert [m.incident_action for channel, m in RecordingTransport.sends if channel == "R-1"] == incident + def test_a_team_without_the_flag_is_not_contacted(self) -> None: outcome = self._run( _announcement(_transition(AlertEventKind.FIRING)), {FIRING_EVENT: [_group(SLACK)]}, live=False