diff --git a/src/sentry/workflow_engine/processors/evaluations/base.py b/src/sentry/workflow_engine/processors/evaluations/base.py index 9167b3309076..473d21bd847c 100644 --- a/src/sentry/workflow_engine/processors/evaluations/base.py +++ b/src/sentry/workflow_engine/processors/evaluations/base.py @@ -1,9 +1,17 @@ +from abc import abstractmethod from collections.abc import Callable, Iterable from dataclasses import dataclass, field, replace -from typing import Any +from typing import TYPE_CHECKING, Any, ClassVar from sentry.workflow_engine.types import ConditionError +if TYPE_CHECKING: + from logging import Logger + +# The shared, static prefix for every evaluation log. Combined with each evaluation's +# `log_name` to build a stable, searchable event name (e.g. `workflow_engine.evaluation.condition`). +_LOG_PREFIX = "workflow_engine.evaluation" + def _find_error( items: list["BaseWorkflowEngineEvaluation[Any, Any]"], @@ -38,6 +46,29 @@ class BaseWorkflowEngineEvaluation[R, D]: triggered: bool = field(compare=False) + # The static suffix for this evaluation type, combined with `_LOG_PREFIX` to build the + # log event name in `to_log`. Set by each concrete subclass. + log_name: ClassVar[str] + + @abstractmethod + def to_artifact(self) -> dict[str, Any]: + """ + Serialize the result, data, error, and triggered state into a flat, log-safe dict + that can be emitted in a log or stored as an artifact. + + Implementations MUST NOT include the raw evaluated value (it may be large or contain + PII) - use metadata (ids, types) and the derived result instead. + """ + ... + + def to_log(self, logger: "Logger") -> None: + """ + Emit this evaluation as a structured log line under a stable, searchable event name + (`{_LOG_PREFIX}.{log_name}`), passing `to_artifact()` as the log `extra`. + """ + event = f"{_LOG_PREFIX}.{self.log_name}" + logger.debug(event, extra=self.to_artifact()) + def is_tainted(self) -> bool: """ Returns True if this result is less trustworthy due to an error during @@ -45,6 +76,10 @@ def is_tainted(self) -> bool: """ return self.error is not None + def error_message(self) -> str | None: + """The error's message for serialization in `to_artifact`, or None when untainted.""" + return self.error.msg if self.error else None + def with_error(self, error: ConditionError) -> "BaseWorkflowEngineEvaluation[R, D]": """ Returns a copy of this evaluation with the given error. If the evaluation is diff --git a/src/sentry/workflow_engine/processors/evaluations/condition.py b/src/sentry/workflow_engine/processors/evaluations/condition.py index ec18998d89a9..79ab3a64ddc7 100644 --- a/src/sentry/workflow_engine/processors/evaluations/condition.py +++ b/src/sentry/workflow_engine/processors/evaluations/condition.py @@ -40,5 +40,17 @@ class DataConditionEvaluation( - triggered: bool - If the evaluation should consider this condition "triggered" or not. """ + log_name = "condition" + result: DataConditionResult = None condition: DataCondition + + def to_artifact(self) -> dict[str, Any]: + return { + "condition_id": self.condition.id, + "type": self.condition.type, + # DataConditionResult may be a DetectorPriorityLevel enum; unwrap to its value. + "result": getattr(self.result, "value", self.result), + "triggered": self.triggered, + "error": self.error_message(), + } diff --git a/src/sentry/workflow_engine/processors/evaluations/condition_group.py b/src/sentry/workflow_engine/processors/evaluations/condition_group.py index 598ac12af071..d647f6e6f98e 100644 --- a/src/sentry/workflow_engine/processors/evaluations/condition_group.py +++ b/src/sentry/workflow_engine/processors/evaluations/condition_group.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING, TypedDict +from typing import TYPE_CHECKING, Any, TypedDict from .base import BaseWorkflowEngineEvaluation from .condition import DataConditionEvaluation @@ -31,4 +31,17 @@ class DataConditionGroupEvaluation(BaseWorkflowEngineEvaluation[bool, GroupEvalu - triggered: bool - whether the group's conditions passed """ - pass + log_name = "condition_group" + + def to_artifact(self) -> dict[str, Any]: + logic_type = self.data["logic_type"] + return { + # logic_type may be a DataConditionGroup.Type enum or a raw string; unwrap to its value. + "logic_type": getattr(logic_type, "value", logic_type), + "result": self.result, + "triggered": self.triggered, + "error": self.error_message(), + "condition_evaluations": [ + condition.to_artifact() for condition in self.data["condition_evaluations"] + ], + } diff --git a/src/sentry/workflow_engine/processors/evaluations/detector.py b/src/sentry/workflow_engine/processors/evaluations/detector.py index d65e22bd1e76..8e1fa76d147e 100644 --- a/src/sentry/workflow_engine/processors/evaluations/detector.py +++ b/src/sentry/workflow_engine/processors/evaluations/detector.py @@ -37,5 +37,19 @@ class DetectorEvaluation( - triggered: bool - If there is an event that should trigger the next phase in the system. """ + log_name = "detector" + result: DetectorResult = None priority: DetectorPriorityLevel + + def to_artifact(self) -> dict[str, Any]: + return { + "group_key": self.data["group_key"], + # priority is a DetectorPriorityLevel enum; unwrap to its value. + "priority": getattr(self.priority, "value", self.priority), + # result is an IssueOccurrence / StatusChangeMessage / None; log its type, not the payload. + "result": type(self.result).__name__ if self.result is not None else None, + "triggered": self.triggered, + "error": self.error_message(), + "trigger_group_evaluation": self.data["trigger_group_evaluation"].to_artifact(), + } diff --git a/src/sentry/workflow_engine/processors/evaluations/workflow.py b/src/sentry/workflow_engine/processors/evaluations/workflow.py index 209d4d5c9b63..7ba2f2ac5762 100644 --- a/src/sentry/workflow_engine/processors/evaluations/workflow.py +++ b/src/sentry/workflow_engine/processors/evaluations/workflow.py @@ -3,7 +3,7 @@ import random from collections.abc import Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING, TypedDict +from typing import TYPE_CHECKING, Any, TypedDict from sentry_sdk import logger as sentry_logger @@ -63,7 +63,27 @@ class WorkflowEvaluation( - `triggered`: bool - Whether the workflow's trigger (WHEN) conditions passed. """ - pass + log_name = "workflow" + + def to_artifact(self) -> dict[str, Any]: + result = self.result + # result is either the "deferred" sentinel (a str) or the sequence of actions that fired. + if isinstance(result, str): + result_kind = "deferred" + triggered_action_ids = None + else: + result_kind = "actions" + triggered_action_ids = [action.id for action in result] + return { + "triggered": self.triggered, + "error": self.error_message(), + "result": result_kind, + "triggered_action_ids": triggered_action_ids, + "trigger_group_eval": self.data["trigger_group_eval"].to_artifact(), + "filter_group_evals": [ + filter_eval.to_artifact() for filter_eval in self.data["filter_group_evals"] + ], + } class WorkflowEvaluationSnapshot(TypedDict): @@ -101,7 +121,7 @@ class GroupedWorkflowEvaluationResult: result: dict[WorkflowId, WorkflowEvaluation] tainted: bool - # Batch-level context used by log_to / get_snapshot / consumers. + # Batch-level context used by to_log / to_artifact / get_snapshot / consumers. organization: Organization event: GroupEvent | Activity group: Group | None = None @@ -160,7 +180,42 @@ def get_snapshot(self) -> WorkflowEvaluationSnapshot: "triggered_actions": triggered_actions, } - def log_to(self, logger: Logger) -> bool: + def to_artifact(self) -> dict[str, Any]: + """ + Flatten the batch-level context into a structured, log-safe dict (ids, counts, and + debug info) and embed each workflow's own `to_artifact()` under `workflow_evaluations` + so a log search can answer *why* a given workflow did or didn't trigger. + + This is the `extra` payload emitted by `to_log`. Empty per-workflow evaluations for the + sentinel early-return paths, where `result` is `{}`. + """ + data_snapshot = self.get_snapshot() + detection_type = ( + data_snapshot["associated_detector"]["type"] + if data_snapshot["associated_detector"] + else None + ) + group_id = data_snapshot["group"].id if data_snapshot["group"] else None + triggered_workflows = data_snapshot["triggered_workflows"] or [] + action_filter_conditions = data_snapshot["action_filter_conditions"] or [] + triggered_actions = data_snapshot["triggered_actions"] or [] + return { + "event_id": data_snapshot["event_id"], + "group_id": group_id, + "detection_type": detection_type, + "workflow_ids": data_snapshot["workflow_ids"], + "triggered_workflow_ids": [w["id"] for w in triggered_workflows], + "delayed_conditions": data_snapshot["delayed_conditions"], + "action_filter_group_ids": [afg["id"] for afg in action_filter_conditions], + "triggered_action_ids": [a["id"] for a in triggered_actions], + "debug_msg": self.msg, + "workflow_evaluations": { + str(workflow_id): evaluation.to_artifact() + for workflow_id, evaluation in self.result.items() + }, + } + + def to_log(self, logger: Logger) -> bool: """ Logs workflow evaluation data. Logging may be skipped if the organization isn't opted in and logs are being @@ -189,27 +244,7 @@ def log_to(self, logger: Logger) -> bool: else: log_str = f"{log_str}.actions.triggered" - data_snapshot = self.get_snapshot() - detection_type = ( - data_snapshot["associated_detector"]["type"] - if data_snapshot["associated_detector"] - else None - ) - group_id = data_snapshot["group"].id if data_snapshot["group"] else None - triggered_workflows = data_snapshot["triggered_workflows"] or [] - action_filter_conditions = data_snapshot["action_filter_conditions"] or [] - triggered_actions = data_snapshot["triggered_actions"] or [] - extra = { - "event_id": data_snapshot["event_id"], - "group_id": group_id, - "detection_type": detection_type, - "workflow_ids": data_snapshot["workflow_ids"], - "triggered_workflow_ids": [w["id"] for w in triggered_workflows], - "delayed_conditions": data_snapshot["delayed_conditions"], - "action_filter_group_ids": [afg["id"] for afg in action_filter_conditions], - "triggered_action_ids": [a["id"] for a in triggered_actions], - "debug_msg": self.msg, - } + extra = self.to_artifact() if direct_to_sentry: sentry_logger.info(log_str, attributes=extra) diff --git a/src/sentry/workflow_engine/tasks/workflows.py b/src/sentry/workflow_engine/tasks/workflows.py index 0761779cdd7d..5e04075b0ecf 100644 --- a/src/sentry/workflow_engine/tasks/workflows.py +++ b/src/sentry/workflow_engine/tasks/workflows.py @@ -75,7 +75,7 @@ def process_workflow_activity(activity_id: int, group_id: int, detector_id: Dete batch_client, event_data, event_start_time=activity.datetime, detector=detector ) - evaluation.log_to(logger) + evaluation.to_log(logger) metrics.incr( "workflow_engine.tasks.process_workflows.activity_update.executed", @@ -172,7 +172,7 @@ def _process_workflows_event( and len(evaluation.triggered_actions) > 0, ) - evaluation.log_to(logger) + evaluation.to_log(logger) duration = time.time() - start_time is_slow = duration > 1.0 # We want full coverage for particularly slow cases, plus a random sampling. diff --git a/tests/sentry/workflow_engine/processors/evaluations/test_base.py b/tests/sentry/workflow_engine/processors/evaluations/test_base.py index a5dd2c8f6e82..bec743b1d7b3 100644 --- a/tests/sentry/workflow_engine/processors/evaluations/test_base.py +++ b/tests/sentry/workflow_engine/processors/evaluations/test_base.py @@ -1,7 +1,17 @@ -from sentry.workflow_engine.models import DataConditionGroup -from sentry.workflow_engine.processors.evaluations import DataConditionGroupEvaluation +from unittest import mock + +from sentry.workflow_engine.models import DataCondition, DataConditionGroup +from sentry.workflow_engine.processors.evaluations import ( + DataConditionEvaluation, + DataConditionGroupEvaluation, + DetectorEvaluation, +) from sentry.workflow_engine.processors.evaluations.base import BaseWorkflowEngineEvaluation -from sentry.workflow_engine.types import ConditionError +from sentry.workflow_engine.types import ( + ConditionError, + DataConditionResult, + DetectorPriorityLevel, +) ERR = ConditionError(msg="test error") OTHER_ERR = ConditionError(msg="other error") @@ -155,3 +165,119 @@ def test_returns_first_when_neither_tainted(self) -> None: a, b = _ev(True), _ev(False) assert BaseWorkflowEngineEvaluation.choose_tainted(a, b) is a assert BaseWorkflowEngineEvaluation.choose_tainted(b, a) is b + + +def _condition_eval( + *, + result: DataConditionResult = True, + triggered: bool = True, + error: ConditionError | None = None, +) -> DataConditionEvaluation: + # Unsaved model instance: to_artifact only reads `id`/`type`, no DB access. + return DataConditionEvaluation( + condition=DataCondition(id=42, type="eq"), + result=result, + triggered=triggered, + error=error, + data="the-raw-value", # never surfaced in the artifact + ) + + +class TestDataConditionEvaluationArtifact: + def test_serializes_metadata_result_and_triggered(self) -> None: + artifact = _condition_eval(result=True, triggered=True).to_artifact() + assert artifact == { + "condition_id": 42, + "type": "eq", + "result": True, + "triggered": True, + "error": None, + } + + def test_omits_raw_value(self) -> None: + # The evaluated value may be large / contain PII; it must never appear in the artifact. + assert "the-raw-value" not in _condition_eval().to_artifact().values() + + def test_unwraps_enum_result(self) -> None: + artifact = _condition_eval(result=DetectorPriorityLevel.HIGH).to_artifact() + assert artifact["result"] == DetectorPriorityLevel.HIGH.value + + def test_serializes_error_message(self) -> None: + artifact = _condition_eval(triggered=False, error=ERR).to_artifact() + assert artifact["triggered"] is False + assert artifact["error"] == "test error" + + +class TestDataConditionGroupEvaluationArtifact: + def test_embeds_condition_artifacts_and_unwraps_logic_type(self) -> None: + condition = _condition_eval() + group = DataConditionGroupEvaluation( + result=True, + triggered=True, + error=None, + data={ + "condition_evaluations": [condition], + "logic_type": DataConditionGroup.Type.ANY, + }, + ) + assert group.to_artifact() == { + "logic_type": "any", + "result": True, + "triggered": True, + "error": None, + "condition_evaluations": [condition.to_artifact()], + } + + def test_handles_raw_string_logic_type(self) -> None: + group = DataConditionGroupEvaluation( + result=False, + triggered=False, + error=None, + data={"condition_evaluations": [], "logic_type": "not-a-real-type"}, + ) + assert group.to_artifact()["logic_type"] == "not-a-real-type" + + +class TestDetectorEvaluationArtifact: + def test_embeds_trigger_group_and_unwraps_priority(self) -> None: + trigger_group = DataConditionGroupEvaluation( + result=True, + triggered=True, + error=None, + data={ + "condition_evaluations": [_condition_eval()], + "logic_type": DataConditionGroup.Type.ANY, + }, + ) + detector = DetectorEvaluation( + result=None, + priority=DetectorPriorityLevel.HIGH, + triggered=True, + error=None, + data={ + "group_key": "group-1", + "trigger_group_evaluation": trigger_group, + "event_data": None, + }, + ) + assert detector.to_artifact() == { + "group_key": "group-1", + "priority": DetectorPriorityLevel.HIGH.value, + "result": None, + "triggered": True, + "error": None, + "trigger_group_evaluation": trigger_group.to_artifact(), + } + + +class TestToLog: + def test_logs_under_static_prefix_with_artifact_extra(self) -> None: + evaluation = _ev(True) + mock_logger = mock.MagicMock() + + evaluation.to_log(mock_logger) + + mock_logger.debug.assert_called_once_with( + "workflow_engine.evaluation.condition_group", + extra=evaluation.to_artifact(), + ) diff --git a/tests/sentry/workflow_engine/processors/evaluations/test_workflow.py b/tests/sentry/workflow_engine/processors/evaluations/test_workflow.py index d4b98ae529f1..356e67207a07 100644 --- a/tests/sentry/workflow_engine/processors/evaluations/test_workflow.py +++ b/tests/sentry/workflow_engine/processors/evaluations/test_workflow.py @@ -3,11 +3,26 @@ from sentry.testutils.cases import TestCase from sentry.testutils.helpers.features import Feature from sentry.testutils.helpers.options import override_options +from sentry.workflow_engine.models import Action, DataConditionGroup +from sentry.workflow_engine.processors.evaluations import ( + DataConditionGroupEvaluation, + WorkflowEvaluation, +) from sentry.workflow_engine.processors.evaluations.workflow import GroupedWorkflowEvaluationResult +from sentry.workflow_engine.types import ConditionError, WorkflowEventData LOG_TO_MODULE = "sentry.workflow_engine.processors.evaluations.workflow" +def _trigger_group() -> DataConditionGroupEvaluation: + return DataConditionGroupEvaluation( + result=True, + triggered=True, + error=None, + data={"condition_evaluations": [], "logic_type": DataConditionGroup.Type.ANY}, + ) + + class TestGroupedWorkflowEvaluationResultLogTo(TestCase): def setUp(self) -> None: super().setUp() @@ -35,7 +50,7 @@ def test_log_to_always_logs_with_feature_enabled(self) -> None: "workflow_engine.evaluation_logs_direct_to_sentry": False, } ): - assert evaluation.log_to(mock_logger) is True + assert evaluation.to_log(mock_logger) is True def test_log_to_respects_sample_rate_when_feature_disabled(self) -> None: evaluation = self._build_result() @@ -49,7 +64,7 @@ def test_log_to_respects_sample_rate_when_feature_disabled(self) -> None: } ): with mock.patch(f"{LOG_TO_MODULE}.random.random", return_value=0.5): - assert not evaluation.log_to(mock_logger) + assert not evaluation.to_log(mock_logger) with override_options( { @@ -58,7 +73,7 @@ def test_log_to_respects_sample_rate_when_feature_disabled(self) -> None: } ): with mock.patch(f"{LOG_TO_MODULE}.random.random", return_value=0.5): - assert evaluation.log_to(mock_logger) + assert evaluation.to_log(mock_logger) def test_log_to_samples_correctly(self) -> None: evaluation = self._build_result() @@ -72,10 +87,10 @@ def test_log_to_samples_correctly(self) -> None: } ): with mock.patch(f"{LOG_TO_MODULE}.random.random", return_value=0.05): - assert evaluation.log_to(mock_logger) + assert evaluation.to_log(mock_logger) with mock.patch(f"{LOG_TO_MODULE}.random.random", return_value=0.15): - assert not evaluation.log_to(mock_logger) + assert not evaluation.to_log(mock_logger) def test_log_to_sentry_logger_when_direct_to_sentry_enabled(self) -> None: evaluation = self._build_result() @@ -84,7 +99,7 @@ def test_log_to_sentry_logger_when_direct_to_sentry_enabled(self) -> None: with Feature({"organizations:workflow-engine-log-evaluations": True}): with override_options({"workflow_engine.evaluation_logs_direct_to_sentry": True}): with mock.patch(f"{LOG_TO_MODULE}.sentry_logger") as mock_sentry_logger: - assert evaluation.log_to(mock_logger) + assert evaluation.to_log(mock_logger) mock_sentry_logger.info.assert_called_once() mock_logger.info.assert_not_called() @@ -95,6 +110,87 @@ def test_log_to_regular_logger_when_direct_to_sentry_disabled(self) -> None: with Feature({"organizations:workflow-engine-log-evaluations": True}): with override_options({"workflow_engine.evaluation_logs_direct_to_sentry": False}): with mock.patch(f"{LOG_TO_MODULE}.sentry_logger") as mock_sentry_logger: - assert evaluation.log_to(mock_logger) + assert evaluation.to_log(mock_logger) mock_logger.info.assert_called_once() mock_sentry_logger.info.assert_not_called() + + +class TestWorkflowEvaluationArtifact(TestCase): + def test_actions_result_lists_action_ids(self) -> None: + trigger_group = _trigger_group() + filter_group = _trigger_group() + # Unsaved Action instances: to_artifact only reads `id`, no DB access. + evaluation = WorkflowEvaluation( + result=[Action(id=7), Action(id=9)], + triggered=True, + error=None, + data={ + "trigger_group_eval": trigger_group, + "filter_group_evals": [filter_group], + "event": WorkflowEventData(event=mock.MagicMock(), group=mock.MagicMock()), + }, + ) + assert evaluation.to_artifact() == { + "triggered": True, + "error": None, + "result": "actions", + "triggered_action_ids": [7, 9], + "trigger_group_eval": trigger_group.to_artifact(), + "filter_group_evals": [filter_group.to_artifact()], + } + + def test_deferred_result_has_no_action_ids(self) -> None: + trigger_group = _trigger_group() + evaluation = WorkflowEvaluation( + result="deferred", + triggered=True, + error=ConditionError(msg="boom"), + data={ + "trigger_group_eval": trigger_group, + "filter_group_evals": [], + "event": WorkflowEventData(event=mock.MagicMock(), group=mock.MagicMock()), + }, + ) + artifact = evaluation.to_artifact() + assert artifact["result"] == "deferred" + assert artifact["triggered_action_ids"] is None + assert artifact["error"] == "boom" + assert artifact["filter_group_evals"] == [] + + +class TestGroupedWorkflowEvaluationResultArtifact(TestCase): + def setUp(self) -> None: + super().setUp() + self.organization = self.create_organization() + self.project = self.create_project(organization=self.organization) + self.event = self.store_event(data={}, project_id=self.project.id) + + def test_embeds_per_workflow_artifacts_keyed_by_id(self) -> None: + trigger_group = _trigger_group() + workflow_eval = WorkflowEvaluation( + result="deferred", + triggered=True, + error=None, + data={ + "trigger_group_eval": trigger_group, + "filter_group_evals": [], + "event": WorkflowEventData(event=mock.MagicMock(), group=mock.MagicMock()), + }, + ) + result = GroupedWorkflowEvaluationResult( + result={123: workflow_eval}, + tainted=True, + organization=self.organization, + event=self.event, + ) + artifact = result.to_artifact() + assert artifact["workflow_evaluations"] == {"123": workflow_eval.to_artifact()} + + def test_empty_result_yields_empty_workflow_evaluations(self) -> None: + result = GroupedWorkflowEvaluationResult( + result={}, + tainted=True, + organization=self.organization, + event=self.event, + ) + assert result.to_artifact()["workflow_evaluations"] == {} diff --git a/tests/sentry/workflow_engine/test_task.py b/tests/sentry/workflow_engine/test_task.py index 1fb933df818b..d4f640ee7725 100644 --- a/tests/sentry/workflow_engine/test_task.py +++ b/tests/sentry/workflow_engine/test_task.py @@ -78,6 +78,7 @@ def test_process_workflow_activity__no_workflows(self, mock_logger: mock.MagicMo "triggered_workflow_ids": [], "delayed_conditions": None, "debug_msg": "No workflows are associated with the detector in the event", + "workflow_evaluations": {}, }, ) @@ -129,6 +130,7 @@ def test_process_workflow_activity__workflows__no_actions( "triggered_workflow_ids": [], "delayed_conditions": None, "debug_msg": "No items were triggered or queued for slow evaluation", + "workflow_evaluations": {}, }, ) @@ -223,6 +225,22 @@ def test_process_workflow_activity__success_logs( "triggered_workflow_ids": [self.workflow.id], "delayed_conditions": None, "debug_msg": None, + "workflow_evaluations": { + str(self.workflow.id): { + "triggered": True, + "error": None, + "result": "actions", + "triggered_action_ids": [self.action.id], + "trigger_group_eval": { + "logic_type": "any", + "result": True, + "triggered": True, + "error": None, + "condition_evaluations": [], + }, + "filter_group_evals": mock.ANY, + } + }, }, )