diff --git a/pyproject.toml b/pyproject.toml index b0baa53..94f407d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "requests", "pydantic>=2,<3", "workflows>=3.0", + "opentelemetry-api" ] [dependency-groups] diff --git a/src/zocalo/service/dispatcher.py b/src/zocalo/service/dispatcher.py index d6a30ee..465d1b7 100644 --- a/src/zocalo/service/dispatcher.py +++ b/src/zocalo/service/dispatcher.py @@ -11,9 +11,15 @@ from importlib.metadata import entry_points import workflows.recipe +from opentelemetry import trace from workflows.services.common_service import CommonService +def _extract_dcid(params: dict) -> int | None: + """Helper method to get dcid. Used for injecting it into current span""" + return params.get("ispyb_dcid") or params.get("dcid") + + class Dispatcher(CommonService): """ Single point of contact service that takes in job meta-information @@ -205,6 +211,14 @@ def process(self, rw, header, message): recipe_id = parameters.get("guid") or str(uuid.uuid4()) parameters["guid"] = recipe_id + # Extract DCID and set on trace span if OpenTelemetry is available + span = trace.get_current_span() + if span.is_recording(): + dcid = _extract_dcid(parameters) + if dcid: + span.set_attribute("dcid", dcid) + self.log.debug(f"Set DCID {dcid} on trace span") + if rw: # If we received a recipe wrapper then we already have a recipe_ID # attached to logs. Make a note of the downstream recipe ID so that