RSDK-14017: Python tracing

pyproject.toml

@@ -44,6 +44,10 @@ mlmodel = [
 	"numpy",
 	"packaging",
 ]
+tracing = [
+	"opentelemetry-api>=1.27.0",
+	"opentelemetry-sdk>=1.27.0",
+]
 
 [build-system]
 requires = ["hatchling"]

src/viam/module/module.py

@@ -34,6 +34,7 @@
 from viam.robot.client import RobotClient
 from viam.rpc.dial import DialOptions, _host_port_from_url
 from viam.rpc.server import Server
+from viam.rpc.tracing import install_parent_send_exporter
 
 # These imports are required to register built-in resources with the registry
 from ..components.arm import Arm  # noqa: F401
@@ -171,6 +172,7 @@ async def _connect_to_parent(self):
             )
             self.logger.debug("Starting module logging")
             logging.setParent(self.parent)
+            install_parent_send_exporter(self.parent)
 
     async def _get_resource(self, name: ResourceName) -> ResourceBase:
         await self._connect_to_parent()

src/viam/rpc/dial.py

@@ -22,6 +22,7 @@
 from viam.errors import InsecureConnectionError, ViamError
 from viam.proto.rpc.auth import AuthenticateRequest, AuthServiceStub
 from viam.proto.rpc.auth import Credentials as PBCredentials
+from viam.rpc.tracing import inject_trace_context
 from viam.version_metadata import API_VERSION, SDK_VERSION
 
 LOGGER = logging.getLogger(__name__)
@@ -373,6 +374,7 @@ def _create_chan(path: str) -> Channel:
 async def _dial_inner(address: str, options: Optional[DialOptions] = None) -> ViamChannel:
     async def send_request(event: SendRequest):
         event.metadata["viam_client"] = f"python;v{SDK_VERSION};v{API_VERSION}"
+        inject_trace_context(event.metadata)
 
     opts = options if options else DialOptions()
     if opts.disable_webrtc:

src/viam/rpc/server.py

@@ -17,6 +17,7 @@
 from viam.resource.registry import Registry
 from viam.resource.rpc_service_base import ResourceRPCServiceBase
 from viam.robot.service import RobotService
+from viam.rpc.tracing import start_server_span
 
 from .signaling import SignalingService
 
@@ -63,23 +64,25 @@ async def _grpc_recvrequest_handler(self, event: RecvRequest):
             host = address[0]
             port = address[1]
         method_func = event.method_func
+        method_name = event.method_name
 
         async def log_resource_name(stream: Stream):
             recv_msg = stream.recv_message
 
             async def rcv_and_log_msg():
                 msg = await recv_msg()
-                log_msg = f"[gRPC] Received message from {host or 'xxxx'}:{port or 'xxxx'} - {event.method_name}"
+                log_msg = f"[gRPC] Received message from {host or 'xxxx'}:{port or 'xxxx'} - {method_name}"
                 if msg and hasattr(msg, "name"):
                     log_msg += f" for resource named: {msg.name}"
                 LOGGER.debug(log_msg)
                 return msg
 
             stream.recv_message = rcv_and_log_msg
-            try:
-                return await method_func(stream)
-            finally:
-                LOGGER.debug(f"[gRPC] Finished call from {host or 'xxxx'}:{port or 'xxxx'} - {event.method_name}")
+            with start_server_span(method_name, stream.metadata or {}):
+                try:
+                    return await method_func(stream)
+                finally:
+                    LOGGER.debug(f"[gRPC] Finished call from {host or 'xxxx'}:{port or 'xxxx'} - {method_name}")
 
         event.method_func = log_resource_name