temporalio · gregbrowndev · Sep 7, 2024 · Nov 14, 2024 · Nov 14, 2024 · Nov 14, 2024
@@ -59,7 +59,7 @@ openai-agents = [
     "temporalio[openai-agents] >= 1.14.1",
 ]
 pydantic-converter = ["pydantic>=2.10.6,<3"]
-sentry = ["sentry-sdk>=1.11.0,<2"]
+sentry = ["sentry-sdk>=2.13.0"]
 trio-async = [
     "trio>=0.28.0,<0.29",
     "trio-asyncio>=0.15.0,<0.16",
@@ -157,5 +157,4 @@ ignore_errors = true
 
 [[tool.mypy.overrides]]
 module = "opentelemetry.*"
-ignore_errors = true
-
+ignore_errors = true
@@ -1,19 +1,40 @@
 # Sentry Sample
 
-This sample shows how to configure [Sentry](https://sentry.io) to intercept and capture errors from the Temporal SDK.
+This sample shows how to configure [Sentry](https://sentry.io) SDK (version 2) to intercept and capture errors from the Temporal SDK
+for workflows and activities. The integration adds some useful context to the errors, such as the activity type, task queue, etc.
+
+## Further details
+
+This is a small modification of the original example Sentry integration in this repo based on SDK v1. The integration
+didn't work properly with Sentry SDK v2 due to some internal changes in the Sentry SDK that broke the worker sandbox.
+Additionally, the v1 SDK has been deprecated and is only receiving security patches and will reach EOL some time in the future.
+If you still need to use Sentry SDK v1, check the original example at this [commit](https://github.com/temporalio/samples-python/blob/090b96d750bafc10d4aad5ad506bb2439c413d5e/sentry).
+
+## Running the Sample
 
 For this sample, the optional `sentry` dependency group must be included. To include, run:
 
-    uv sync --group sentry
+    uv sync --no-default-groups --dev --group sentry
+
+> Note: this integration breaks when `gevent` is installed (e.g. by the gevent sample) so make sure to only install
+> the `sentry` group and run the scripts below as described.
 
 To run, first see [README.md](../README.md) for prerequisites. Set `SENTRY_DSN` environment variable to the Sentry DSN.
 Then, run the following from the root directory to start the worker:
 
+    export SENTRY_DSN=  # You'll need a Sentry account to test against
+    export ENVIRONMENT=dev
     uv run sentry/worker.py
 
 This will start the worker. Then, in another terminal, run the following to execute the workflow:
 
     uv run sentry/starter.py
 
-The workflow should complete with the hello result. If you alter the workflow or the activity to raise an
-`ApplicationError` instead, it should appear in Sentry.
+You should see the activity fail causing an error to be reported to Sentry.
+
+## Screenshot
+
+The screenshot below shows the extra tags and context included in the
+Sentry error from the exception thrown in the activity.
+
+![Sentry screenshot](images/sentry.jpeg)
@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+
+from temporalio import activity
+
+
+@dataclass
+class WorkingActivityInput:
+    message: str
+
+
+@activity.defn
+async def working_activity(input: WorkingActivityInput) -> str:
+    activity.logger.info("Running activity with parameter %s" % input)
+    return "Success"
+
+
+@dataclass
+class BrokenActivityInput:
+    message: str
+
+
+@activity.defn
+async def broken_activity(input: BrokenActivityInput) -> str:
+    activity.logger.info("Running activity with parameter %s" % input)
+    raise Exception("Activity failed!")
@@ -1,5 +1,6 @@
+import logging
 from dataclasses import asdict, is_dataclass
-from typing import Any, Optional, Type, Union
+from typing import Any, Optional, Type
 
 from temporalio import activity, workflow
 from temporalio.worker import (
@@ -12,63 +13,65 @@
 )
 
 with workflow.unsafe.imports_passed_through():
-    from sentry_sdk import Hub, capture_exception, set_context, set_tag
+    import sentry_sdk
 
 
-def _set_common_workflow_tags(info: Union[workflow.Info, activity.Info]):
-    set_tag("temporal.workflow.type", info.workflow_type)
-    set_tag("temporal.workflow.id", info.workflow_id)
+logger = logging.getLogger(__name__)
 
 
 class _SentryActivityInboundInterceptor(ActivityInboundInterceptor):
     async def execute_activity(self, input: ExecuteActivityInput) -> Any:
         # https://docs.sentry.io/platforms/python/troubleshooting/#addressing-concurrency-issues
-        with Hub(Hub.current):
-            set_tag("temporal.execution_type", "activity")
-            set_tag("module", input.fn.__module__ + "." + input.fn.__qualname__)
-
+        with sentry_sdk.isolation_scope() as scope:
+            scope.set_tag("temporal.execution_type", "activity")
+            scope.set_tag("module", input.fn.__module__ + "." + input.fn.__qualname__)
             activity_info = activity.info()
-            _set_common_workflow_tags(activity_info)
-            set_tag("temporal.activity.id", activity_info.activity_id)
-            set_tag("temporal.activity.type", activity_info.activity_type)
-            set_tag("temporal.activity.task_queue", activity_info.task_queue)
-            set_tag("temporal.workflow.namespace", activity_info.workflow_namespace)
-            set_tag("temporal.workflow.run_id", activity_info.workflow_run_id)
+            scope.set_tag("temporal.workflow.type", activity_info.workflow_type)
+            scope.set_tag("temporal.workflow.id", activity_info.workflow_id)
+            scope.set_tag("temporal.activity.id", activity_info.activity_id)
+            scope.set_tag("temporal.activity.type", activity_info.activity_type)
+            scope.set_tag("temporal.activity.task_queue", activity_info.task_queue)
+            scope.set_tag(
+                "temporal.workflow.namespace", activity_info.workflow_namespace
+            )
+            scope.set_tag("temporal.workflow.run_id", activity_info.workflow_run_id)
             try:
                 return await super().execute_activity(input)
             except Exception as e:
                 if len(input.args) == 1:
                     [arg] = input.args
                     if is_dataclass(arg) and not isinstance(arg, type):
-                        set_context("temporal.activity.input", asdict(arg))
-                set_context("temporal.activity.info", activity.info().__dict__)
-                capture_exception()
+                        scope.set_context("temporal.activity.input", asdict(arg))
+                scope.set_context("temporal.activity.info", activity.info().__dict__)
+                scope.capture_exception()
                 raise e
 
 
 class _SentryWorkflowInterceptor(WorkflowInboundInterceptor):
     async def execute_workflow(self, input: ExecuteWorkflowInput) -> Any:
         # https://docs.sentry.io/platforms/python/troubleshooting/#addressing-concurrency-issues
-        with Hub(Hub.current):
-            set_tag("temporal.execution_type", "workflow")
-            set_tag("module", input.run_fn.__module__ + "." + input.run_fn.__qualname__)
+        with sentry_sdk.isolation_scope() as scope:
+            scope.set_tag("temporal.execution_type", "workflow")
+            scope.set_tag(
+                "module", input.run_fn.__module__ + "." + input.run_fn.__qualname__
+            )
             workflow_info = workflow.info()
-            _set_common_workflow_tags(workflow_info)
-            set_tag("temporal.workflow.task_queue", workflow_info.task_queue)
-            set_tag("temporal.workflow.namespace", workflow_info.namespace)
-            set_tag("temporal.workflow.run_id", workflow_info.run_id)
+            scope.set_tag("temporal.workflow.type", workflow_info.workflow_type)
+            scope.set_tag("temporal.workflow.id", workflow_info.workflow_id)
+            scope.set_tag("temporal.workflow.task_queue", workflow_info.task_queue)
+            scope.set_tag("temporal.workflow.namespace", workflow_info.namespace)
+            scope.set_tag("temporal.workflow.run_id", workflow_info.run_id)
             try:
                 return await super().execute_workflow(input)
             except Exception as e:
                 if len(input.args) == 1:
                     [arg] = input.args
                     if is_dataclass(arg) and not isinstance(arg, type):
-                        set_context("temporal.workflow.input", asdict(arg))
-                set_context("temporal.workflow.info", workflow.info().__dict__)
-
+                        scope.set_context("temporal.workflow.input", asdict(arg))
+                scope.set_context("temporal.workflow.info", workflow.info().__dict__)
                 if not workflow.unsafe.is_replaying():
                     with workflow.unsafe.sandbox_unrestricted():
-                        capture_exception()
+                        scope.capture_exception()
                 raise e
 
 
@@ -78,9 +81,6 @@ class SentryInterceptor(Interceptor):
     def intercept_activity(
         self, next: ActivityInboundInterceptor
     ) -> ActivityInboundInterceptor:
-        """Implementation of
-        :py:meth:`temporalio.worker.Interceptor.intercept_activity`.
-        """
         return _SentryActivityInboundInterceptor(super().intercept_activity(next))
 
     def workflow_interceptor_class(

@@ -1,23 +1,25 @@
 import asyncio
-import os
 
 from temporalio.client import Client
 
-from sentry.worker import GreetingWorkflow
+from sentry.workflow import SentryExampleWorkflow, SentryExampleWorkflowInput
 
 
 async def main():
     # Connect client
     client = await Client.connect("localhost:7233")
 
     # Run workflow
-    result = await client.execute_workflow(
-        GreetingWorkflow.run,
-        "World",
-        id="sentry-workflow-id",
-        task_queue="sentry-task-queue",
-    )
-    print(f"Workflow result: {result}")
+    try:
+        result = await client.execute_workflow(
+            SentryExampleWorkflow.run,
+            SentryExampleWorkflowInput(option="broken"),
+            id="sentry-workflow-id",
+            task_queue="sentry-task-queue",
+        )
+        print(f"Workflow result: {result}")
+    except Exception:
+        print("Workflow failed - check Sentry for details")
 
 
 if __name__ == "__main__":

@@ -1,64 +1,92 @@
 import asyncio
-import logging
 import os
-from dataclasses import dataclass
-from datetime import timedelta
 
 import sentry_sdk
-from temporalio import activity, workflow
+from sentry_sdk.integrations.asyncio import AsyncioIntegration
+from sentry_sdk.types import Event, Hint
 from temporalio.client import Client
 from temporalio.worker import Worker
+from temporalio.worker.workflow_sandbox import (
+    SandboxedWorkflowRunner,
+    SandboxRestrictions,
+)
 
+from sentry.activity import broken_activity, working_activity
 from sentry.interceptor import SentryInterceptor
+from sentry.workflow import SentryExampleWorkflow
 
+interrupt_event = asyncio.Event()
 
-@dataclass
-class ComposeGreetingInput:
-    greeting: str
-    name: str
 
+def before_send(event: Event, hint: Hint) -> Event | None:
+    # Filter out __ShutdownRequested events raised by the worker's internals
+    if str(hint.get("exc_info", [None])[0].__name__) == "_ShutdownRequested":
+        return None
 
-@activity.defn
-async def compose_greeting(input: ComposeGreetingInput) -> str:
-    activity.logger.info("Running activity with parameter %s" % input)
-    return f"{input.greeting}, {input.name}!"
+    return event
 
 
-@workflow.defn
-class GreetingWorkflow:
-    @workflow.run
-    async def run(self, name: str) -> str:
-        workflow.logger.info("Running workflow with parameter %s" % name)
-        return await workflow.execute_activity(
-            compose_greeting,
-            ComposeGreetingInput("Hello", name),
-            start_to_close_timeout=timedelta(seconds=10),
+def initialise_sentry() -> None:
+    sentry_dsn = os.environ.get("SENTRY_DSN")
+    if not sentry_dsn:
+        print(
+            "SENTRY_DSN environment variable is not set. Sentry will not be initialized."
         )
+        return
 
+    environment = os.environ.get("ENVIRONMENT")
+    sentry_sdk.init(
+        dsn=sentry_dsn,
+        environment=environment,
+        integrations=[
+            AsyncioIntegration(),
+        ],
+        attach_stacktrace=True,
+        before_send=before_send,
+    )
+    print(f"Sentry SDK initialized for environment: {environment!r}")
 
-async def main():
-    # Uncomment the line below to see logging
-    # logging.basicConfig(level=logging.INFO)
 
+async def main():
     # Initialize the Sentry SDK
-    sentry_sdk.init(
-        dsn=os.environ.get("SENTRY_DSN"),
-    )
+    initialise_sentry()
 
     # Start client
     client = await Client.connect("localhost:7233")
 
     # Run a worker for the workflow
-    worker = Worker(
+    async with Worker(
         client,
         task_queue="sentry-task-queue",
-        workflows=[GreetingWorkflow],
-        activities=[compose_greeting],
+        workflows=[SentryExampleWorkflow],
+        activities=[broken_activity, working_activity],
         interceptors=[SentryInterceptor()],  # Use SentryInterceptor for error reporting
-    )
-
-    await worker.run()
+        workflow_runner=SandboxedWorkflowRunner(
+            restrictions=SandboxRestrictions.default.with_passthrough_modules(
+                "sentry_sdk"
+            )
+        ),
+    ):
+        # Wait until interrupted
+        print("Worker started, ctrl+c to exit")
+        await interrupt_event.wait()
+        print("Shutting down")
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    # Note: "Addressing Concurrency Issues" section in Sentry docs recommends using
+    # the AsyncioIntegration: "If you do concurrency with asyncio coroutines, make
+    # sure to use the AsyncioIntegration which will clone the correct scope in your Tasks"
+    # See https://docs.sentry.io/platforms/python/troubleshooting/
+    #
+    # However, this captures all unhandled exceptions in the event loop.
+    # So handle shutdown gracefully to avoid CancelledError and KeyboardInterrupt
+    # exceptions being captured as errors. Sentry also captures the worker's
+    # _ShutdownRequested exception, which is probably not useful. We've filtered this
+    # out in Sentry's before_send function.
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(main())
+    except KeyboardInterrupt:
+        interrupt_event.set()
+        loop.run_until_complete(loop.shutdown_asyncgens())