DEV: xray activation

Author: dandavison

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "nexus-rpc>=1.1.0",
     "protobuf>=3.20,<6",
     "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
+    "temporalio-xray",
     "types-protobuf>=3.20",
     "typing-extensions>=4.2.0,<5",
 ]
@@ -233,3 +234,4 @@ package = false
 
 [tool.uv.sources]
 nexus-rpc = { git = "https://github.com/nexus-rpc/sdk-python.git", rev = "35f574c711193a6e2560d3e6665732a5bb7ae92c" }
+temporalio-xray = { path = "../xray/sdks/python" }

temporalio/worker/_workflow_instance.py

@@ -44,6 +44,8 @@
 )
 
 import nexusrpc.handler
+import xray
+from google.protobuf.json_format import MessageToJson
 from nexusrpc import InputT, OutputT
 from typing_extensions import Self, TypeAlias, TypedDict
 
@@ -355,10 +357,30 @@ def get_thread_id(self) -> Optional[int]:
     # "_make_workflow_input", all other calls are "_apply_" + the job field
     # name.
 
+    # The commented code below had line numbers, which are now removed.
+
     def activate(
         self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation
     ) -> temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion:
-        print("activate -------------------------------------------------------------")
+        with xray.start_as_current_span(
+            actor=xray.Actor.WORKFLOW_USER,
+            workflow_id=self._info.workflow_id,
+            name="handle_activation",
+            request_payload=MessageToJson(act),
+        ) as span:
+            print(
+                f">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> WFT {self._info.workflow_id}\n"
+            )
+
+            completion = self._activate(act)
+            print("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
+            if span:
+                span.set_attribute("sdk.response.payload", MessageToJson(completion))
+        return completion
+
+    def _activate(
+        self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation
+    ) -> temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion:
         # Reset current completion, time, and whether replaying
         self._current_completion = (
             temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion()