feat(workflow_engine): Add evaluation logs to detectors

src/sentry/workflow_engine/caches/detector.py

@@ -46,7 +46,7 @@ def _query_detectors(source_id: str, query_type: str) -> list[Detector]:
             data_sources__source_id=source_id,
             data_sources__type=query_type,
         )
-        .select_related("workflow_condition_group")
+        .select_related("project__organization", "workflow_condition_group")
         .prefetch_related("workflow_condition_group__conditions")
         .distinct()
         .order_by("id")

src/sentry/workflow_engine/processors/__init__.py

@@ -2,6 +2,12 @@
     "DataConditionEvaluation",
     "DataConditionGroupEvaluation",
     "DetectorEvaluation",
+    "ProcessDetectorsResult",
 ]
 
-from .evaluations import DataConditionEvaluation, DataConditionGroupEvaluation, DetectorEvaluation
+from .evaluations import (
+    DataConditionEvaluation,
+    DataConditionGroupEvaluation,
+    DetectorEvaluation,
+    ProcessDetectorsResult,
+)

src/sentry/workflow_engine/processors/detector.py

@@ -10,6 +10,7 @@
 from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
 from sentry.models.activity import Activity
 from sentry.models.group import Group
+from sentry.models.organization import Organization
 from sentry.services.eventstore.models import GroupEvent
 from sentry.utils import metrics
 from sentry.utils.cache import cache
@@ -21,7 +22,8 @@
 )
 from sentry.workflow_engine.models import DataPacket, Detector
 from sentry.workflow_engine.models.detector_group import DetectorGroup
-from sentry.workflow_engine.processors import DetectorEvaluation
+from sentry.workflow_engine.processors import DetectorEvaluation, ProcessDetectorsResult
+from sentry.workflow_engine.processors.evaluation_logging import emit_detector_evaluation_logs
 from sentry.workflow_engine.types import (
     DetectorGroupKey,
     DetectorId,
@@ -271,6 +273,16 @@
     )
 
 
+def _get_detector_organization(detector: Detector) -> Organization:
+    if detector.project_id is not None:
+        return detector.linked_project.organization
+
+    organization_id = detector.config.get("organization_id")
+    if not isinstance(organization_id, int):
+        raise ValueError("Organization-scoped detector is missing organization_id")
+    return Organization.objects.get_from_cache(id=organization_id)
+
+
 @trace
 def process_detectors[T](
     data_packet: DataPacket[T], detectors: list[Detector]
@@ -293,6 +305,17 @@
         ):
             detector_results = handler.evaluate(data_packet)
 
+        emit_detector_evaluation_logs(
+            logger,
+            organization=_get_detector_organization(detector),
+            result=ProcessDetectorsResult(
+                detector_id=detector.id,
+                detector_type=detector.type,
+                project_id=detector.project_id,
+                evaluations=detector_results,
+            ),
+        )
+
         for result in detector_results.values():
             logger_extra = {
                 "detector": detector.id,

src/sentry/workflow_engine/processors/evaluation_logging.py

@@ -2,19 +2,65 @@
 
 import random
 from logging import Logger
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from sentry import features, options
 from sentry.utils.sdk import sdk_logger
+from sentry.workflow_engine.processors.evaluations.detector import ProcessDetectorsResult
 from sentry.workflow_engine.processors.evaluations.workflow import ProcessWorkflowsResult
 
 if TYPE_CHECKING:
     from sentry.models.organization import Organization
 
 
+DETECTOR_EVALUATION_LOG_PREFIX = "workflow_engine.process_detectors.evaluation"
 WORKFLOW_EVALUATION_LOG_PREFIX = "workflow_engine.process_workflows.evaluation"
 
 
+def _should_emit_evaluation_logs(organization: Organization) -> bool:
+    if features.has("organizations:workflow-engine-log-evaluations", organization):
+        return True
+    sample_rate = cast(float, options.get("workflow_engine.evaluation_log_sample_rate"))
+    return random.random() < sample_rate
+
+
+def _emit_evaluation_artifacts(
+    logger: Logger,
+    *,
+    organization_id: int,
+    artifacts: list[dict[str, object]],
+    log_prefix: str,
+) -> None:
+    direct_to_sentry = options.get("workflow_engine.evaluation_logs_direct_to_sentry")
+    for artifact in artifacts:
+        artifact["organization_id"] = organization_id
+
+        if direct_to_sentry:
+            sdk_logger.info(log_prefix, attributes=artifact)
+        else:
+            logger.info(log_prefix, extra=artifact)
+
+
+def emit_detector_evaluation_logs(
+    logger: Logger,
+    *,
+    organization: Organization,
+    result: ProcessDetectorsResult,
+    log_prefix: str = DETECTOR_EVALUATION_LOG_PREFIX,
+) -> bool:
+    """Sample a detector and emit one self-contained artifact per grouped evaluation."""
+    if not _should_emit_evaluation_logs(organization):
+        return False
+
+    _emit_evaluation_artifacts(
+        logger,
+        organization_id=organization.id,
+        artifacts=result.evaluation_artifacts(),
+        log_prefix=log_prefix,
+    )
+    return True
+
+
 def emit_workflow_evaluation_logs(
     logger: Logger,
     *,
@@ -23,25 +69,18 @@ def emit_workflow_evaluation_logs(
     log_prefix: str = WORKFLOW_EVALUATION_LOG_PREFIX,
 ) -> bool:
     """Sample a batch and emit one self-contained artifact per workflow evaluation."""
-    should_log = features.has("organizations:workflow-engine-log-evaluations", organization)
-    if not should_log:
-        should_log = random.random() < options.get("workflow_engine.evaluation_log_sample_rate")
-
-    if not should_log:
+    if not _should_emit_evaluation_logs(organization):
         return False
 
-    direct_to_sentry = options.get("workflow_engine.evaluation_logs_direct_to_sentry")
     artifacts = (
         [evaluation.to_artifact() for evaluation in result.evaluations.values()]
         if result.evaluations
         else [result.to_artifact()]
     )
-    for artifact in artifacts:
-        artifact["organization_id"] = organization.id
-
-        if direct_to_sentry:
-            sdk_logger.info(log_prefix, attributes=artifact)
-        else:
-            logger.info(log_prefix, extra=artifact)
-
+    _emit_evaluation_artifacts(
+        logger,
+        organization_id=organization.id,
+        artifacts=artifacts,
+        log_prefix=log_prefix,
+    )
     return True

src/sentry/workflow_engine/processors/evaluations/__init__.py

@@ -4,7 +4,9 @@
     "DataConditionGroupEvaluation",
     "DetectorEvaluation",
     "DetectorEvaluationData",
+    "DetectorEvaluationOutcome",
     "DeferredWorkflowEvaluationResult",
+    "ProcessDetectorsResult",
     "ProcessWorkflowsResult",
     "WorkflowEvaluation",
     "WorkflowEvaluationData",
@@ -13,7 +15,12 @@
 
 from .condition import DataConditionEvaluation, DataConditionEvaluationException
 from .condition_group import DataConditionGroupEvaluation
-from .detector import DetectorEvaluation, DetectorEvaluationData
+from .detector import (
+    DetectorEvaluation,
+    DetectorEvaluationData,
+    DetectorEvaluationOutcome,
+    ProcessDetectorsResult,
+)
 from .workflow import (
     DeferredWorkflowEvaluationResult,
     ProcessWorkflowsResult,

src/sentry/workflow_engine/processors/evaluations/detector.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from enum import StrEnum
 from typing import Any, TypedDict
 
 from sentry.workflow_engine.types import DetectorGroupKey, DetectorPriorityLevel, DetectorResult
@@ -13,6 +14,11 @@ class DetectorEvaluationData(TypedDict):
     event_data: dict[str, Any] | None  # TODO - improve this typing, for now migrating
 
 
+class DetectorEvaluationOutcome(StrEnum):
+    COMPLETED = "completed"
+    NO_RESULTS = "no_results"
+
+
 @dataclass(frozen=True, kw_only=True)
 class DetectorEvaluation(
     BaseWorkflowEngineEvaluation[
@@ -49,3 +55,35 @@ def artifact_fields(self) -> dict[str, Any]:
             "priority": self.priority.value,
             "trigger_group_evaluation": self.data["trigger_group_evaluation"].to_artifact(),
         }
+
+
+@dataclass(frozen=True, kw_only=True)
+class ProcessDetectorsResult:
+    detector_id: int
+    detector_type: str
+    project_id: int | None
+    evaluations: dict[DetectorGroupKey, DetectorEvaluation]
+
+    @property
+    def outcome(self) -> DetectorEvaluationOutcome:
+        if self.evaluations:
+            return DetectorEvaluationOutcome.COMPLETED
+        return DetectorEvaluationOutcome.NO_RESULTS
+
+    def to_artifact(self) -> dict[str, object]:
+        return {
+            "detector_id": self.detector_id,
+            "detector_type": self.detector_type,
+            "project_id": self.project_id,
+            "outcome": self.outcome,
+        }
+
+    def evaluation_artifacts(self) -> list[dict[str, object]]:
+        detector_artifact = self.to_artifact()
+        if not self.evaluations:
+            return [detector_artifact]
+
+        return [
+            {**detector_artifact, **evaluation.to_artifact()}
+            for evaluation in self.evaluations.values()
+        ]