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

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
         )