started to add get_snapshot to the models, this will simplify the layers of trying to get debuggable output.

Author: saponifi3d

src/sentry/workflow_engine/models/action.py

@@ -3,7 +3,7 @@
 import builtins
 import logging
 from enum import StrEnum
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from django.db import models
 from django.db.models import Q
@@ -112,6 +112,12 @@ class Meta:
             ),
         ]
 
+    def get_snapshot(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "type": self.type,
+        }
+
     def get_handler(self) -> builtins.type[ActionHandler]:
         action_type = Action.Type(self.type)
         return action_handler_registry.get(action_type)

src/sentry/workflow_engine/models/data_condition_group.py

@@ -1,5 +1,5 @@
 from enum import StrEnum
-from typing import ClassVar, Self
+from typing import Any, ClassVar, Self
 
 from django.db import models
 
@@ -36,3 +36,14 @@ class Type(StrEnum):
         max_length=200, choices=[(t.value, t.value) for t in Type], default=Type.ANY
     )
     organization = models.ForeignKey("sentry.Organization", on_delete=models.CASCADE)
+
+    def get_snapshot(self) -> dict[str, Any]:
+        conditions = []
+        if hasattr(self, "conditions"):
+            conditions = [cond.get_snapshot() for cond in self.conditions.all()]
+
+        return {
+            "id": self.id,
+            "logic_type": self.logic_type,
+            "conditions": conditions,
+        }

src/sentry/workflow_engine/models/detector.py

@@ -141,6 +141,18 @@ def settings(self) -> DetectorSettings:
 
         return settings
 
+    def get_snapshot(self) -> dict[str, Any]:
+        trigger_conditions = None
+        if self.workflow_condition_group:
+            trigger_conditions = self.workflow_condition_group.get_snapshot()
+
+        return {
+            "id": self.id,
+            "enabled": self.enabled,
+            "status": self.status,
+            "trigger_conditions": trigger_conditions,
+        }
+
     def get_audit_log_data(self) -> dict[str, Any]:
         return {"name": self.name}
 

src/sentry/workflow_engine/models/workflow.py

@@ -92,6 +92,23 @@ class Meta:
     def get_audit_log_data(self) -> dict[str, Any]:
         return {"name": self.name}
 
+    def get_snapshot(self) -> dict[str, Any]:
+        when_condition_group = None
+        if self.when_condition_group:
+            when_condition_group = self.when_condition_group.get_snapshot()
+
+        environment_id = None
+        if self.environment:
+            environment_id = self.environment.id
+
+        return {
+            "id": self.id,
+            "enabled": self.enabled,
+            "environment_id": environment_id,
+            "status": self.status,
+            "triggers": when_condition_group,
+        }
+
     def evaluate_trigger_conditions(
         self, event_data: WorkflowEventData, when_data_conditions: list[DataCondition] | None = None
     ) -> tuple[TriggerResult, list[DataCondition]]:

src/sentry/workflow_engine/types.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from dataclasses import asdict, dataclass, field
+from dataclasses import dataclass, field
 from enum import IntEnum, StrEnum
 from logging import Logger
 from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypedDict, TypeVar
@@ -96,6 +96,43 @@ class WorkflowEvaluationData:
     triggered_workflows: set[Workflow] | None = None
     associated_detector: Detector | None = None
 
+    def get_snapshot(self) -> dict[str, Any]:
+        """
+        This method will take the complex data structures, like models / list of models,
+        and turn them into the critical attributes of a model or lists of IDs.
+        """
+
+        associated_detector = None
+        if self.associated_detector:
+            associated_detector = self.associated_detector.get_snapshot()
+
+        workflow_ids = []
+        if self.workflows:
+            workflow_ids = [workflow.id for workflow in self.workflows]
+
+        triggered_workflows = []
+        if self.triggered_workflows:
+            triggered_workflows = [workflow.get_snapshot() for workflow in self.triggered_workflows]
+
+        action_filter_conditions = []
+        if self.action_groups:
+            action_filter_conditions = [group.get_snapshot() for group in self.action_groups]
+
+        triggered_actions = []
+        if self.triggered_actions:
+            triggered_actions = [action.get_snapshot() for action in self.triggered_actions]
+
+        return {
+            "workflow_ids": workflow_ids,
+            "associated_detector": associated_detector,
+            "event": self.event,
+            "group": self.event.group,
+            "event_data": self.event.data,
+            "action_filter_conditions": action_filter_conditions,
+            "triggered_actions": triggered_actions,
+            "triggered_workflows": triggered_workflows,
+        }
+
 
 @dataclass(frozen=True)
 class WorkflowEvaluation:
@@ -134,15 +171,7 @@ def to_log(self, logger: Logger) -> None:
         else:
             log_str = f"{log_str}.actions.triggered"
 
-        logger.info(
-            log_str,
-            extra={
-                **asdict(self.data),
-                "debug_msg": self.msg,
-                "group": self.data.event.group,
-                "data": self.data.event.data,
-            },
-        )
+        logger.info(log_str, extra={**self.data.get_snapshot(), "debug_msg": self.msg})
 
 
 class ConfigTransformer(ABC):

tests/sentry/workflow_engine/test_task.py

@@ -254,6 +254,51 @@ def test_process_workflow_activity(
             },
         )
 
+    @mock.patch("sentry.workflow_engine.processors.workflow.evaluate_workflow_triggers")
+    @mock.patch("sentry.workflow_engine.tasks.workflows.logger")
+    def test_process_workflow_activity__success_logs(
+        self, mock_logger, mock_evaluate_workflow_triggers
+    ) -> None:
+        self.workflow = self.create_workflow(organization=self.organization)
+        self.workflow.when_condition_group = self.create_data_condition_group()
+        self.create_data_condition(condition_group=self.workflow.when_condition_group)
+        self.workflow.save()
+
+        self.action_group = self.create_data_condition_group(logic_type="any-short")
+        self.action = self.create_action()
+        self.create_data_condition_group_action(
+            condition_group=self.action_group,
+            action=self.action,
+        )
+        self.create_workflow_data_condition_group(self.workflow, self.action_group)
+
+        self.create_detector_workflow(
+            detector=self.detector,
+            workflow=self.workflow,
+        )
+
+        mock_evaluate_workflow_triggers.return_value = ({self.workflow}, {})
+        process_workflow_activity(
+            activity_id=self.activity.id,
+            group_id=self.group.id,
+            detector_id=self.detector.id,
+        )
+
+        mock_logger.info.assert_called_once_with(
+            "workflow_engine.process_workflows.evaluation.actions.triggered",
+            extra={
+                "workflow_ids": [self.workflow.id],
+                "associated_detector": self.detector.get_snapshot(),
+                "event": self.activity,
+                "group": self.activity.group,
+                "event_data": self.activity.data,
+                "action_filter_conditions": [self.action_group.get_snapshot()],
+                "triggered_actions": [self.action.get_snapshot()],
+                "triggered_workflows": [self.workflow.get_snapshot()],
+                "debug_msg": None,
+            },
+        )
+
     @mock.patch(
         "sentry.workflow_engine.models.incident_groupopenperiod.update_incident_based_on_open_period_status_change"
     )  # rollout code that is independently tested