Initial pass at WorkflowNotProcessable, and adding logs. this will allow us to have a deterministic result from the evaluation, and simplified logging outside of it.

Author: saponifi3d

src/sentry/workflow_engine/processors/workflow.py

@@ -2,7 +2,7 @@
 from dataclasses import asdict, dataclass, replace
 from datetime import datetime
 from enum import StrEnum
-from typing import DefaultDict
+from typing import Any, DefaultDict
 
 import sentry_sdk
 from django.db import router, transaction
@@ -33,7 +33,7 @@
 )
 from sentry.workflow_engine.processors.detector import get_detector_by_event
 from sentry.workflow_engine.processors.workflow_fire_history import create_workflow_fire_histories
-from sentry.workflow_engine.types import WorkflowEventData
+from sentry.workflow_engine.types import WorkflowEventData, WorkflowNotProcessable
 from sentry.workflow_engine.utils import log_context, scopedstats
 from sentry.workflow_engine.utils.metrics import metrics_incr
 
@@ -464,7 +464,7 @@ def process_workflows(
     event_data: WorkflowEventData,
     event_start_time: datetime,
     detector: Detector | None = None,
-) -> set[Workflow]:
+) -> set[Workflow] | WorkflowNotProcessable:
     """
     This method will get the detector based on the event, and then gather the associated workflows.
     Next, it will evaluate the "when" (or trigger) conditions for each workflow, if the conditions are met,
@@ -478,6 +478,12 @@ def process_workflows(
         fire_actions,
     )
 
+    # This dictionary stores the data for WorkflowNotProcessable
+    # and is eventually unpacked to pass into the frozen dataclass
+    extra_data: dict[str, Any] = {
+        "group_event": event_data.event,
+    }
+
     try:
         if detector is None and isinstance(event_data.event, GroupEvent):
             detector = get_detector_by_event(event_data)
@@ -496,7 +502,12 @@ def process_workflows(
             )
         )
     except Detector.DoesNotExist:
-        return set()
+        return WorkflowNotProcessable(
+            message="No Detectors associated with the issue were found",
+            **extra_data,
+        )
+
+    extra_data["associated_detector"] = [detector]
 
     try:
         environment = get_environment_by_event(event_data)
@@ -510,30 +521,53 @@ def process_workflows(
             )
         )
     except Environment.DoesNotExist:
-        return set()
+        return WorkflowNotProcessable(
+            message="Environment for event not found",
+            **extra_data,
+        )
 
     if features.has("organizations:workflow-engine-process-workflows-logs", organization):
         log_context.set_verbose(True)
 
     workflows = _get_associated_workflows(detector, environment, event_data)
+    extra_data["workflows"] = workflows
+
     if not workflows:
-        # If there aren't any workflows, there's nothing to evaluate
-        return set()
+        return WorkflowNotProcessable(
+            message="No workflows are associated with the detector in the event",
+            **extra_data,
+        )
 
     triggered_workflows, queue_items_by_workflow_id = evaluate_workflow_triggers(
         workflows, event_data, event_start_time
     )
+
+    extra_data["triggered_workflows"] = triggered_workflows
+
     if not triggered_workflows and not queue_items_by_workflow_id:
-        # if there aren't any triggered workflows, there's no action filters to evaluate
-        return set()
+        return WorkflowNotProcessable(
+            message="No items were triggered or queued for slow evaluation",
+            **extra_data,
+        )
 
     actions_to_trigger, queue_items_by_workflow_id = evaluate_workflows_action_filters(
         triggered_workflows, event_data, queue_items_by_workflow_id, event_start_time
     )
+
     enqueue_workflows(batch_client, queue_items_by_workflow_id)
+
+    # TODO - refactor to have action evaluation separate from workflow evaluation
+    # The result of processing a workflow, should be a list of actions we want to trigger.
+    # From here on we are probably doing too many side effects and things related to actions.
     actions = filter_recently_fired_workflow_actions(actions_to_trigger, event_data)
     sentry_sdk.set_tag("workflow_engine.triggered_actions", len(actions))
 
+    extra_data = {
+        **extra_data,
+        "actions": actions_to_trigger,  # All the actions we found to trigger
+        "triggered_actions": actions,  # All the actions we intend to trigger
+    }
+
     if not actions:
         # If there aren't any actions on the associated workflows, there's nothing to trigger
         return triggered_workflows

src/sentry/workflow_engine/tasks/workflows.py

@@ -1,5 +1,6 @@
 import random
 import time
+from dataclasses import asdict
 from datetime import UTC, datetime
 from typing import Any
 
@@ -18,17 +19,47 @@
 from sentry.utils import metrics
 from sentry.utils.exceptions import quiet_redis_noise
 from sentry.utils.locking import UnableToAcquireLock
+from sentry.workflow_engine.buffer.batch_client import DelayedWorkflowClient
 from sentry.workflow_engine.models import DataConditionGroup, Detector
 from sentry.workflow_engine.tasks.utils import (
     EventNotFoundError,
     build_workflow_event_data_from_event,
 )
-from sentry.workflow_engine.types import WorkflowEventData
+from sentry.workflow_engine.types import WorkflowEventData, WorkflowNotProcessable
 from sentry.workflow_engine.utils import log_context, scopedstats
 
 logger = log_context.get_logger(__name__)
 
 
+# TODO - this seems like it should live in processors.workflow,
+# Then we can create a `process_new_issue` or `process_event` kind
+# of method. That will wrap a new `process_actions_with_logs`
+# (and `process_actions`) method. Allowing us to have a much cleaner
+# interface here, and between those processor methods.
+def process_workflows_with_logs(
+    batch_client: DelayedWorkflowClient,
+    event_data: WorkflowEventData,
+    event_start_time: datetime,
+    detector: Detector | None = None,
+) -> None:
+    """
+    This method is used to create improved logging for `process_workflows`,
+    where we can capture the `WorkflowNotProcessable` results, and log them.
+
+    This should create a log for each issue, so we can determine why a workflow
+    did or did not trigger.
+    """
+    from sentry.workflow_engine.processors.workflow import process_workflows
+
+    evaluation = process_workflows(batch_client, event_data, event_start_time, detector)
+
+    if isinstance(evaluation, WorkflowNotProcessable):
+        logger.info("process_workflows.evaluation", extra=asdict(evaluation))
+    else:
+        # TODO - Log the triggered workflows
+        pass
+
+
 @instrumented_task(
     name="sentry.workflow_engine.tasks.process_workflow_activity",
     namespace=namespaces.workflow_engine_tasks,
@@ -44,8 +75,6 @@ def process_workflow_activity(activity_id: int, group_id: int, detector_id: int)
     The task will get the Activity from the database, create a WorkflowEventData object,
     and then process the data in `process_workflows`.
     """
-    from sentry.workflow_engine.buffer.batch_client import DelayedWorkflowClient
-    from sentry.workflow_engine.processors.workflow import process_workflows
 
     with transaction.atomic(router.db_for_write(Detector)):
         try:
@@ -69,9 +98,10 @@ def process_workflow_activity(activity_id: int, group_id: int, detector_id: int)
     )
     with quiet_redis_noise():
         batch_client = DelayedWorkflowClient()
-        process_workflows(
+        process_workflows_with_logs(
             batch_client, event_data, event_start_time=activity.datetime, detector=detector
         )
+
     metrics.incr(
         "workflow_engine.tasks.process_workflows.activity_update.executed",
         tags={"activity_type": activity.type, "detector_type": detector.type},
@@ -103,11 +133,9 @@ def process_workflows_event(
     start_timestamp_seconds: float | None = None,
     **kwargs: dict[str, Any],
 ) -> None:
-    from sentry.workflow_engine.buffer.batch_client import DelayedWorkflowClient
-    from sentry.workflow_engine.processors.workflow import process_workflows
-
     recorder = scopedstats.Recorder()
     start_time = time.time()
+
     with recorder.record():
         try:
             event_data = build_workflow_event_data_from_event(
@@ -131,7 +159,7 @@ def process_workflows_event(
         )
         with quiet_redis_noise():
             batch_client = DelayedWorkflowClient()
-            process_workflows(batch_client, event_data, event_start_time=event_start_time)
+            process_workflows_with_logs(batch_client, event_data, event_start_time=event_start_time)
     duration = time.time() - start_time
     is_slow = duration > 1.0
     # We want full coverage for particularly slow cases, plus a random sampling.
@@ -158,7 +186,6 @@ def schedule_delayed_workflows(**kwargs: Any) -> None:
     """
     Schedule delayed workflow buffers in a batch.
     """
-    from sentry.workflow_engine.buffer.batch_client import DelayedWorkflowClient
     from sentry.workflow_engine.processors.schedule import process_buffered_workflows
 
     lock_name = "schedule_delayed_workflows"

src/sentry/workflow_engine/types.py

@@ -20,7 +20,7 @@
     from sentry.snuba.models import SnubaQueryEventType
     from sentry.workflow_engine.endpoints.validators.base import BaseDetectorTypeValidator
     from sentry.workflow_engine.handlers.detector import DetectorHandler
-    from sentry.workflow_engine.models import Action, Detector
+    from sentry.workflow_engine.models import Action, Detector, Workflow
     from sentry.workflow_engine.models.data_condition import Condition
 
 T = TypeVar("T")
@@ -72,6 +72,17 @@ class WorkflowEventData:
     workflow_env: Environment | None = None
 
 
+@dataclass(frozen=True)
+class WorkflowNotProcessable:
+    group_event: GroupEvent
+    message: str
+    actions: list[Action] | None = None
+    workflows: list[Workflow] | None = None
+    triggered_actions: list[Action] | None = None
+    triggered_workflows: list[Workflow] | None = None
+    associated_detectors: list[Detector] | None = None
+
+
 class ConfigTransformer(ABC):
     """
     A ConfigTransformer is used to transform the config between API and internal representations.