Implement start_active_span_from_edu for OTEL

AFAICT, this never worked before because everything was serialized into `content["org.matrix.opentracing_context"]`
but `start_active_span_from_edu` read from `content["opentracing"]`.
See #5852 (comment)

Do we even still want this?

synapse/api/constants.py

@@ -193,6 +193,9 @@ class LimitBlockingTypes:
 class EventContentFields:
     """Fields found in events' content, regardless of type."""
 
+    # Synapse internal content field for tracing
+    TRACING_CONTEXT: Final = "org.matrix.tracing_context"
+
     # Labels for the event, cf https://github.com/matrix-org/matrix-doc/pull/2326
     LABELS: Final = "org.matrix.labels"
 

synapse/federation/federation_server.py

@@ -1399,7 +1399,7 @@ async def on_edu(self, edu_type: str, origin: str, content: dict) -> None:
         # Check if we have a handler on this instance
         handler = self.edu_handlers.get(edu_type)
         if handler:
-            with start_active_span_from_edu(content, "handle_edu"):
+            with start_active_span_from_edu("handle_edu", edu_content=content):
                 try:
                     await handler(origin, content)
                 except SynapseError as e:

synapse/federation/units.py

@@ -21,6 +21,7 @@
 
 import attr
 
+from synapse.api.constants import EventContentFields
 from synapse.types import JsonDict
 
 logger = logging.getLogger(__name__)
@@ -55,10 +56,12 @@ def get_internal_dict(self) -> JsonDict:
         }
 
     def get_context(self) -> str:
-        return getattr(self, "content", {}).get("org.matrix.tracing_context", "{}")
+        return getattr(self, "content", {}).get(
+            EventContentFields.TRACING_CONTEXT, "{}"
+        )
 
     def strip_context(self) -> None:
-        getattr(self, "content", {})["org.matrix.tracing_context"] = "{}"
+        getattr(self, "content", {})[EventContentFields.TRACING_CONTEXT] = "{}"
 
 
 def _none_to_list(edus: Optional[List[JsonDict]]) -> List[JsonDict]:

synapse/handlers/devicemessage.py

@@ -15,7 +15,7 @@
 import logging
 from typing import TYPE_CHECKING, Any, Dict
 
-from synapse.api.constants import EduTypes, ToDeviceEventTypes
+from synapse.api.constants import EduTypes, EventContentFields, ToDeviceEventTypes
 from synapse.api.errors import SynapseError
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.logging.context import run_in_background
@@ -273,7 +273,7 @@ async def send_device_message(
                 "sender": sender_user_id,
                 "type": message_type,
                 "message_id": message_id,
-                "org.matrix.tracing_context": json_encoder.encode(context),
+                EventContentFields.TRACING_CONTEXT: json_encoder.encode(context),
             }
 
         # Add messages to the database.

synapse/logging/tracing.py

@@ -177,7 +177,6 @@ def set_fates(clotho, lachesis, atropos, father="Zues", mother="Themis"):
     Dict,
     Generator,
     Iterable,
-    Iterator,
     List,
     Optional,
     Pattern,
@@ -194,7 +193,9 @@ def set_fates(clotho, lachesis, atropos, father="Zues", mother="Themis"):
 from twisted.web.http import Request
 from twisted.web.http_headers import Headers
 
+from synapse.api.constants import EventContentFields
 from synapse.config import ConfigError
+from synapse.util import json_decoder
 
 if TYPE_CHECKING:
     from synapse.http.site import SynapseRequest
@@ -555,20 +556,28 @@ def start_active_span(
 
 
 def start_active_span_from_edu(
-    edu_content: Dict[str, Any],
     operation_name: str,
-) -> Iterator["opentelemetry.trace.span.Span"]:
+    *,
+    edu_content: Dict[str, Any],
+) -> ContextManager["opentelemetry.trace.span.Span"]:
     """
     Extracts a span context from an edu and uses it to start a new active span
 
     Args:
+        operation_name: The label for the chunk of time used to process the given edu.
         edu_content: an edu_content with a `context` field whose value is
-        canonical json for a dict which contains opentracing information.
-
-        For the other args see opentracing.tracer
+            canonical json for a dict which contains opentracing information.
     """
-    # TODO
-    pass
+    if opentelemetry is None:
+        return contextlib.nullcontext()  # type: ignore[unreachable]
+
+    carrier = json_decoder.decode(edu_content.get("context", "{}")).get(
+        EventContentFields.TRACING_CONTEXT, {}
+    )
+
+    context = extract_text_map(carrier)
+
+    return start_active_span(name=operation_name, context=context)
 
 
 # OpenTelemetry setters for attributes, logs, etc
@@ -588,12 +597,12 @@ def set_attribute(key: str, value: Union[str, bool, int, float]) -> None:
 
 @ensure_active_span("set the status")
 def set_status(
-    status: "opentelemetry.trace.status.StatusCode", exc: Optional[Exception]
+    status_code: "opentelemetry.trace.status.StatusCode", exc: Optional[Exception]
 ) -> None:
     """Sets a tag on the active span"""
     active_span = get_active_span()
     assert active_span is not None
-    active_span.set_status(status)
+    active_span.set_status(opentelemetry.trace.status.Status(status_code=status_code))
     if exc:
         active_span.record_exception(exc)
 

synapse/storage/databases/main/devices.py

@@ -30,7 +30,7 @@
 
 from typing_extensions import Literal
 
-from synapse.api.constants import EduTypes
+from synapse.api.constants import EduTypes, EventContentFields
 from synapse.api.errors import Codes, StoreError
 from synapse.logging.tracing import (
     get_active_span_text_map,
@@ -537,7 +537,7 @@ async def _get_device_update_edus_by_remote(
                     "device_id": device_id,
                     "prev_id": [prev_id] if prev_id else [],
                     "stream_id": stream_id,
-                    "org.matrix.tracing_context": tracing_context,
+                    EventContentFields.TRACING_CONTEXT: tracing_context,
                 }
 
                 prev_id = stream_id