feat(aci): Track DataCondition errors and propagate such results as tainted (#103003)

DataCondition failures are currently reported to Sentry/logs
individually, then treated as a non-triggering.
Here we return a special error type, so that the caller knows the result
isn't necessarily reliable, and in DataConditionGroup logic we aggregate
these results in such a way that we know at the end if an untrustworthy
result influenced the outcome.

These "tainted" results are reported for each Detector evaluation, so we
can track how often our results were influenced by errors, giving us a
better high level picture of user impact than we'd have purely by
looking at individual reported error counts.

This pattern will be extended to cover over DataCondition evaluation
cases.

Author: kcons

src/sentry/grouping/grouptype.py

@@ -5,23 +5,20 @@
 from sentry.models.group import DEFAULT_TYPE_ID
 from sentry.types.group import PriorityLevel
 from sentry.workflow_engine.endpoints.validators.error_detector import ErrorDetectorValidator
-from sentry.workflow_engine.handlers.detector.base import DetectorHandler
-from sentry.workflow_engine.models.data_source import DataPacket
-from sentry.workflow_engine.types import (
-    DetectorEvaluationResult,
-    DetectorGroupKey,
-    DetectorSettings,
+from sentry.workflow_engine.handlers.detector.base import (
+    DetectorHandler,
+    GroupedDetectorEvaluationResult,
 )
+from sentry.workflow_engine.models.data_source import DataPacket
+from sentry.workflow_engine.types import DetectorSettings
 
 T = TypeVar("T")
 
 
 class ErrorDetectorHandler(DetectorHandler):
-    def evaluate(
-        self, data_packet: DataPacket[T]
-    ) -> dict[DetectorGroupKey, DetectorEvaluationResult]:
+    def evaluate_impl(self, data_packet: DataPacket[T]) -> GroupedDetectorEvaluationResult:
         # placeholder
-        return {}
+        return GroupedDetectorEvaluationResult(result={}, tainted=False)
 
 
 @dataclass(frozen=True)

src/sentry/workflow_engine/handlers/detector/__init__.py

@@ -4,8 +4,15 @@
     "DetectorHandler",
     "DetectorOccurrence",
     "DetectorStateData",
+    "GroupedDetectorEvaluationResult",
     "StatefulDetectorHandler",
 ]
 
-from .base import DataPacketEvaluationType, DataPacketType, DetectorHandler, DetectorOccurrence
+from .base import (
+    DataPacketEvaluationType,
+    DataPacketType,
+    DetectorHandler,
+    DetectorOccurrence,
+    GroupedDetectorEvaluationResult,
+)
 from .stateful import DetectorStateData, StatefulDetectorHandler

src/sentry/workflow_engine/handlers/detector/base.py

@@ -11,6 +11,7 @@
 from sentry.issues.grouptype import GroupType
 from sentry.issues.issue_occurrence import IssueEvidence, IssueOccurrence
 from sentry.types.actor import Actor
+from sentry.utils import metrics
 from sentry.workflow_engine.models import DataConditionGroup, DataPacket, Detector
 from sentry.workflow_engine.processors.data_condition_group import ProcessedDataConditionGroup
 from sentry.workflow_engine.types import (
@@ -77,6 +78,12 @@ def to_issue_occurrence(
         )
 
 
+@dataclass(frozen=True)
+class GroupedDetectorEvaluationResult:
+    result: dict[DetectorGroupKey, DetectorEvaluationResult]
+    tainted: bool
+
+
 # TODO - @saponifi3d - Change this class to be a pure ABC and remove the `__init__` method.
 # TODO - @saponifi3d - Once the change is made, we should introduce a `BaseDetector` class to evaluate simple cases
 class DetectorHandler(abc.ABC, Generic[DataPacketType, DataPacketEvaluationType]):
@@ -102,10 +109,27 @@ def __init__(self, detector: Detector):
         else:
             self.condition_group = None
 
-    @abc.abstractmethod
     def evaluate(
         self, data_packet: DataPacket[DataPacketType]
-    ) -> dict[DetectorGroupKey, DetectorEvaluationResult] | None:
+    ) -> dict[DetectorGroupKey, DetectorEvaluationResult]:
+        tags = {
+            "detector_type": self.detector.type,
+            "result": "unknown",
+        }
+        try:
+            value = self.evaluate_impl(data_packet)
+            tags["result"] = "tainted" if value.tainted else "success"
+            metrics.incr("workflow_engine_detector.evaluation", tags=tags, sample_rate=1.0)
+            return value.result
+        except Exception:
+            tags["result"] = "failure"
+            metrics.incr("workflow_engine_detector.evaluation", tags=tags, sample_rate=1.0)
+            raise
+
+    @abc.abstractmethod
+    def evaluate_impl(
+        self, data_packet: DataPacket[DataPacketType]
+    ) -> GroupedDetectorEvaluationResult:
         """
         This method is used to evaluate the data packet's value against the conditions on the detector.
         """

src/sentry/workflow_engine/handlers/detector/stateful.py

@@ -19,6 +19,7 @@
     DetectorHandler,
     DetectorOccurrence,
     EventData,
+    GroupedDetectorEvaluationResult,
 )
 from sentry.workflow_engine.models import DataPacket, Detector, DetectorState
 from sentry.workflow_engine.processors.data_condition_group import (
@@ -371,14 +372,16 @@ def _build_workflow_engine_evidence_data(
             ],
         }
 
-    def evaluate(
+    def evaluate_impl(
         self, data_packet: DataPacket[DataPacketType]
-    ) -> dict[DetectorGroupKey, DetectorEvaluationResult]:
+    ) -> GroupedDetectorEvaluationResult:
         dedupe_value = self.extract_dedupe_value(data_packet)
         group_data_values = self._extract_value_from_packet(data_packet)
         state = self.state_manager.get_state_data(list(group_data_values.keys()))
         results: dict[DetectorGroupKey, DetectorEvaluationResult] = {}
 
+        tainted = False
+
         for group_key, data_value in group_data_values.items():
             state_data: DetectorStateData = state[group_key]
             if dedupe_value <= state_data.dedupe_value:
@@ -391,7 +394,10 @@ def evaluate(
                 group_data_values[group_key]
             )
 
-            if condition_results is None or condition_results.logic_result is False:
+            if condition_results is not None and condition_results.logic_result.is_tainted():
+                tainted = True
+
+            if condition_results is None or condition_results.logic_result.triggered is False:
                 # Invalid condition result, nothing we can do
                 # Or if we didn't match any conditions in the evaluation
                 continue
@@ -441,7 +447,7 @@ def evaluate(
             )
 
         self.state_manager.commit_state_updates()
-        return results
+        return GroupedDetectorEvaluationResult(result=results, tainted=tainted)
 
     def _create_resolve_message(
         self,
@@ -624,7 +630,7 @@ def _evaluation_detector_conditions(
                 },
             )
 
-        if condition_evaluation.logic_result:
+        if condition_evaluation.logic_result.triggered:
             validated_condition_results: list[DetectorPriorityLevel] = [
                 condition_result.result
                 for condition_result in condition_evaluation.condition_results

src/sentry/workflow_engine/models/data_condition.py

@@ -14,7 +14,7 @@
 from sentry.db.models import DefaultFieldsModel, region_silo_model, sane_repr
 from sentry.utils import metrics, registry
 from sentry.workflow_engine.registry import condition_handler_registry
-from sentry.workflow_engine.types import DataConditionResult, DetectorPriorityLevel
+from sentry.workflow_engine.types import ConditionError, DataConditionResult, DetectorPriorityLevel
 from sentry.workflow_engine.utils import scopedstats
 
 logger = logging.getLogger(__name__)
@@ -145,7 +145,7 @@ def get_snapshot(self) -> dict[str, Any]:
             "condition_result": self.condition_result,
         }
 
-    def get_condition_result(self) -> DataConditionResult:
+    def get_condition_result(self) -> DataConditionResult | ConditionError:
         match self.condition_result:
             case float() | bool():
                 return self.condition_result
@@ -159,15 +159,15 @@ def get_condition_result(self) -> DataConditionResult:
                     "Invalid condition result",
                     extra={"condition_result": self.condition_result, "id": self.id},
                 )
+                return ConditionError(msg="Invalid condition result")
 
-        return None
-
-    def _evaluate_operator(self, condition_type: Condition, value: T) -> DataConditionResult:
+    def _evaluate_operator(
+        self, condition_type: Condition, value: T
+    ) -> DataConditionResult | ConditionError:
         # If the condition is a base type, handle it directly
         op = CONDITION_OPS[condition_type]
-        result = None
         try:
-            result = op(cast(Any, value), self.comparison)
+            return op(cast(Any, value), self.comparison)
         except TypeError:
             logger.exception(
                 "Invalid comparison for data condition",
@@ -178,19 +178,20 @@ def _evaluate_operator(self, condition_type: Condition, value: T) -> DataConditi
                     "condition_id": self.id,
                 },
             )
-
-        return result
+            return ConditionError(msg="Invalid comparison for data condition")
 
     @scopedstats.timer()
-    def _evaluate_condition(self, condition_type: Condition, value: T) -> DataConditionResult:
+    def _evaluate_condition(
+        self, condition_type: Condition, value: T
+    ) -> DataConditionResult | ConditionError:
         try:
             handler = condition_handler_registry.get(condition_type)
         except registry.NoRegistrationExistsError:
             logger.exception(
                 "No registration exists for condition",
                 extra={"type": self.type, "id": self.id},
             )
-            return None
+            return ConditionError(msg="No registration exists for condition")
 
         should_be_fast = not is_slow_condition(self)
         start_time = time.time()
@@ -212,7 +213,7 @@ def _evaluate_condition(self, condition_type: Condition, value: T) -> DataCondit
                     "error": str(e),
                 },
             )
-            return None
+            return ConditionError(msg=str(e))
         finally:
             duration = time.time() - start_time
             if should_be_fast and duration >= FAST_CONDITION_TOO_SLOW_THRESHOLD.total_seconds():
@@ -230,17 +231,17 @@ def _evaluate_condition(self, condition_type: Condition, value: T) -> DataCondit
 
         return result
 
-    def evaluate_value(self, value: T) -> DataConditionResult:
+    def evaluate_value(self, value: T) -> DataConditionResult | ConditionError:
         try:
             condition_type = Condition(self.type)
         except ValueError:
             logger.exception(
                 "Invalid condition type",
                 extra={"type": self.type, "id": self.id},
             )
-            return None
+            return ConditionError(msg="Invalid condition type")
 
-        result: DataConditionResult
+        result: DataConditionResult | ConditionError
         if condition_type in CONDITION_OPS:
             result = self._evaluate_operator(condition_type, value)
         else:

src/sentry/workflow_engine/models/workflow.py

@@ -18,7 +18,8 @@
 from sentry.models.owner_base import OwnerModel
 from sentry.workflow_engine.models.data_condition import DataCondition, is_slow_condition
 from sentry.workflow_engine.models.data_condition_group import DataConditionGroup
-from sentry.workflow_engine.types import WorkflowEventData
+from sentry.workflow_engine.processors.data_condition_group import TriggerResult
+from sentry.workflow_engine.types import ConditionError, WorkflowEventData
 
 from .json_config import JSONConfigBase
 
@@ -93,7 +94,7 @@ def get_audit_log_data(self) -> dict[str, Any]:
 
     def evaluate_trigger_conditions(
         self, event_data: WorkflowEventData, when_data_conditions: list[DataCondition] | None = None
-    ) -> tuple[bool, list[DataCondition]]:
+    ) -> tuple[TriggerResult, list[DataCondition]]:
         """
         Evaluate the conditions for the workflow trigger and return if the evaluation was successful.
         If there aren't any workflow trigger conditions, the workflow is considered triggered.
@@ -104,7 +105,7 @@ def evaluate_trigger_conditions(
         )
 
         if self.when_condition_group_id is None:
-            return True, []
+            return TriggerResult.TRUE, []
 
         workflow_event_data = replace(event_data, workflow_env=self.environment)
         try:
@@ -116,7 +117,7 @@ def evaluate_trigger_conditions(
                 "DataConditionGroup does not exist",
                 extra={"id": self.when_condition_group_id},
             )
-            return False, []
+            return TriggerResult(False, ConditionError(msg="DataConditionGroup does not exist")), []
         group_evaluation, remaining_conditions = process_data_condition_group(
             group, workflow_event_data, when_data_conditions
         )