Draft: Migrate to OpenTelemetry tracing

mypy.ini

@@ -164,9 +164,6 @@ ignore_missing_imports = True
 [mypy-pympler.*]
 ignore_missing_imports = True
 
-[mypy-rust_python_jaeger_reporter.*]
-ignore_missing_imports = True
-
 [mypy-saml2.*]
 ignore_missing_imports = True
 

pyproject.toml

@@ -175,13 +175,14 @@ authlib = { version = ">=0.14.0", optional = true }
 systemd-python = { version = ">=231", optional = true }
 lxml = { version = ">=4.2.0", optional = true }
 sentry-sdk = { version = ">=0.7.2", optional = true }
-opentracing = { version = ">=2.2.0", optional = true }
-jaeger-client = { version = ">=4.0.0", optional = true }
 txredisapi = { version = ">=1.4.7", optional = true }
 hiredis = { version = "*", optional = true }
 Pympler = { version = "*", optional = true }
 parameterized = { version = ">=0.7.4", optional = true }
 idna = { version = ">=2.5", optional = true }
+opentelemetry-api = {version = "^1.11.1", optional = true}
+opentelemetry-sdk = {version = "^1.11.1", optional = true}
+opentelemetry-exporter-jaeger = {version = "^1.11.1", optional = true}
 
 [tool.poetry.extras]
 # NB: Packages that should be part of `pip install matrix-synapse[all]` need to be specified
@@ -196,7 +197,7 @@ oidc = ["authlib"]
 systemd = ["systemd-python"]
 url_preview = ["lxml"]
 sentry = ["sentry-sdk"]
-opentracing = ["jaeger-client", "opentracing"]
+opentelemetry = ["opentelemetry-api", "opentelemetry-sdk", "opentelemetry-exporter-jaeger"]
 jwt = ["authlib"]
 # hiredis is not a *strict* dependency, but it makes things much faster.
 # (if it is not installed, we fall back to slow code.)
@@ -229,8 +230,8 @@ all = [
     "lxml",
     # sentry
     "sentry-sdk",
-    # opentracing
-    "jaeger-client", "opentracing",
+    # opentelemetry
+    "opentelemetry-api", "opentelemetry-sdk", "opentelemetry-exporter-jaeger",
     # redis
     "txredisapi", "hiredis",
     # cache_memory
@@ -254,7 +255,6 @@ mypy-zope = "*"
 types-bleach = ">=4.1.0"
 types-commonmark = ">=0.9.2"
 types-jsonschema = ">=3.2.0"
-types-opentracing = ">=2.4.2"
 types-Pillow = ">=8.3.4"
 types-psycopg2 = ">=2.9.9"
 types-pyOpenSSL = ">=20.0.7"

synapse/api/auth.py

@@ -30,7 +30,7 @@
 from synapse.appservice import ApplicationService
 from synapse.http import get_request_user_agent
 from synapse.http.site import SynapseRequest
-from synapse.logging.opentracing import active_span, force_tracing, start_active_span
+from synapse.logging.tracing import force_tracing, get_active_span, start_active_span
 from synapse.storage.databases.main.registration import TokenLookupResult
 from synapse.types import Requester, UserID, create_requester
 
@@ -132,7 +132,7 @@ async def get_user_by_req(
                 is invalid.
             AuthError if access is denied for the user in the access token
         """
-        parent_span = active_span()
+        parent_span = get_active_span()
         with start_active_span("get_user_by_req"):
             requester = await self._wrapped_get_user_by_req(
                 request, allow_guest, allow_expired
@@ -147,14 +147,14 @@ async def get_user_by_req(
                     # so we also force it on for that.
                     force_tracing()
                     force_tracing(parent_span)
-                parent_span.set_tag(
+                parent_span.set_attribute(
                     "authenticated_entity", requester.authenticated_entity
                 )
-                parent_span.set_tag("user_id", requester.user.to_string())
+                parent_span.set_attribute("user_id", requester.user.to_string())
                 if requester.device_id is not None:
-                    parent_span.set_tag("device_id", requester.device_id)
+                    parent_span.set_attribute("device_id", requester.device_id)
                 if requester.app_service is not None:
-                    parent_span.set_tag("appservice_id", requester.app_service.id)
+                    parent_span.set_attribute("appservice_id", requester.app_service.id)
             return requester
 
     async def _wrapped_get_user_by_req(

synapse/app/_base.py

@@ -62,7 +62,7 @@
 from synapse.events.third_party_rules import load_legacy_third_party_event_rules
 from synapse.handlers.auth import load_legacy_password_auth_providers
 from synapse.logging.context import PreserveLoggingContext
-from synapse.logging.opentracing import init_tracer
+from synapse.logging.tracing import init_tracer
 from synapse.metrics import install_gc_manager, register_threadpool
 from synapse.metrics.background_process_metrics import wrap_as_background_process
 from synapse.metrics.jemalloc import setup_jemalloc_stats

synapse/config/tracer.py

@@ -24,41 +24,43 @@ class TracerConfig(Config):
     section = "tracing"
 
     def read_config(self, config: JsonDict, **kwargs: Any) -> None:
-        opentracing_config = config.get("opentracing")
-        if opentracing_config is None:
-            opentracing_config = {}
+        opentelemetry_config = config.get("opentelemetry")
+        if opentelemetry_config is None:
+            opentelemetry_config = {}
 
-        self.opentracer_enabled = opentracing_config.get("enabled", False)
+        self.opentelemetry_enabled = opentelemetry_config.get("enabled", False)
 
-        self.jaeger_config = opentracing_config.get(
-            "jaeger_config",
-            {"sampler": {"type": "const", "param": 1}, "logging": False},
+        self.jaeger_exporter_config = opentelemetry_config.get(
+            "jaeger_exporter_config",
+            {},
         )
 
         self.force_tracing_for_users: Set[str] = set()
 
-        if not self.opentracer_enabled:
+        if not self.opentelemetry_enabled:
             return
 
-        check_requirements("opentracing")
+        check_requirements("opentelemetry")
 
         # The tracer is enabled so sanitize the config
 
-        self.opentracer_whitelist: List[str] = opentracing_config.get(
+        self.opentelemetry_whitelist: List[str] = opentelemetry_config.get(
             "homeserver_whitelist", []
         )
-        if not isinstance(self.opentracer_whitelist, list):
+        if not isinstance(self.opentelemetry_whitelist, list):
             raise ConfigError("Tracer homeserver_whitelist config is malformed")
 
-        force_tracing_for_users = opentracing_config.get("force_tracing_for_users", [])
+        force_tracing_for_users = opentelemetry_config.get(
+            "force_tracing_for_users", []
+        )
         if not isinstance(force_tracing_for_users, list):
             raise ConfigError(
-                "Expected a list", ("opentracing", "force_tracing_for_users")
+                "Expected a list", ("opentelemetry", "force_tracing_for_users")
             )
         for i, u in enumerate(force_tracing_for_users):
             if not isinstance(u, str):
                 raise ConfigError(
                     "Expected a string",
-                    ("opentracing", "force_tracing_for_users", f"index {i}"),
+                    ("opentelemetry", "force_tracing_for_users", f"index {i}"),
                 )
             self.force_tracing_for_users.add(u)

synapse/federation/federation_server.py

@@ -61,7 +61,7 @@
     nested_logging_context,
     run_in_background,
 )
-from synapse.logging.opentracing import log_kv, start_active_span_from_edu, trace
+from synapse.logging.tracing import log_kv, start_active_span_from_edu, trace
 from synapse.metrics.background_process_metrics import wrap_as_background_process
 from synapse.replication.http.federation import (
     ReplicationFederationSendEduRestServlet,

synapse/federation/sender/per_destination_queue.py

@@ -32,7 +32,7 @@
 from synapse.federation.units import Edu
 from synapse.handlers.presence import format_user_presence_state
 from synapse.logging import issue9533_logger
-from synapse.logging.opentracing import SynapseTags, set_tag
+from synapse.logging.tracing import SynapseTags, set_attribute
 from synapse.metrics import sent_transactions_counter
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.types import ReadReceipt
@@ -596,7 +596,7 @@ async def _get_to_device_message_edus(self, limit: int) -> Tuple[List[Edu], int]
             if not message_id:
                 continue
 
-            set_tag(SynapseTags.TO_DEVICE_MESSAGE_ID, message_id)
+            set_attribute(SynapseTags.TO_DEVICE_MESSAGE_ID, message_id)
 
         edus = [
             Edu(

synapse/federation/sender/transaction_manager.py

@@ -21,11 +21,12 @@
 from synapse.events import EventBase
 from synapse.federation.persistence import TransactionActions
 from synapse.federation.units import Edu, Transaction
-from synapse.logging.opentracing import (
+from synapse.logging.tracing import (
+    Link,
+    StatusCode,
     extract_text_map,
-    set_tag,
-    start_active_span_follows_from,
-    tags,
+    set_status,
+    start_active_span,
     whitelisted_homeserver,
 )
 from synapse.types import JsonDict
@@ -94,7 +95,10 @@ async def send_new_transaction(
             if keep_destination:
                 edu.strip_context()
 
-        with start_active_span_follows_from("send_transaction", span_contexts):
+        with start_active_span(
+            "send_transaction",
+            links=[Link(context) for context in span_contexts],
+        ):
             logger.debug("TX [%s] _attempt_new_transaction", destination)
 
             txn_id = str(self._next_txn_id)
@@ -166,7 +170,7 @@ def json_data_cb() -> JsonDict:
             except HttpResponseException as e:
                 code = e.code
 
-                set_tag(tags.ERROR, True)
+                set_status(StatusCode.ERROR, e)
 
                 logger.info("TX [%s] {%s} got %d response", destination, txn_id, code)
                 raise

synapse/federation/transport/server/_base.py

@@ -15,7 +15,6 @@
 import functools
 import logging
 import re
-import time
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional, Tuple, cast
 
@@ -25,12 +24,13 @@
 from synapse.http.servlet import parse_json_object_from_request
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import run_in_background
-from synapse.logging.opentracing import (
-    active_span,
-    set_tag,
+from synapse.logging.tracing import (
+    Link,
+    create_non_recording_span,
+    get_active_span,
+    set_attribute,
     span_context_from_request,
     start_active_span,
-    start_active_span_follows_from,
     whitelisted_homeserver,
 )
 from synapse.types import JsonDict
@@ -309,59 +309,65 @@ async def new_func(
                 raise
 
             # update the active opentracing span with the authenticated entity
-            set_tag("authenticated_entity", str(origin))
+            set_attribute("authenticated_entity", str(origin))
 
             # if the origin is authenticated and whitelisted, use its span context
             # as the parent.
-            context = None
+            origin_span_context = None
             if origin and whitelisted_homeserver(origin):
-                context = span_context_from_request(request)
-
-            if context:
-                servlet_span = active_span()
-                # a scope which uses the origin's context as a parent
-                processing_start_time = time.time()
-                scope = start_active_span_follows_from(
+                origin_span_context = span_context_from_request(request)
+
+            if origin_span_context:
+                local_servlet_span = get_active_span()
+                # Create a span which uses the `origin_span_context` as a parent
+                # so we can see how the incoming payload was processed while
+                # we're looking at the outgoing trace. Since the parent is set
+                # to a remote span (from the origin), it won't show up in the
+                # local trace which is why we create another span below for the
+                # local trace. A span can only have one parent so we have to
+                # create two separate ones.
+                remote_parent_span = start_active_span(
                     "incoming-federation-request",
-                    child_of=context,
-                    contexts=(servlet_span,),
-                    start_time=processing_start_time,
+                    context=origin_span_context,
+                    # Cross-link back to the local trace so we can jump
+                    # to the incoming side from the remote origin trace.
+                    links=[Link(local_servlet_span.get_span_context())],
+                )
+
+                # Create a local span to appear in the local trace
+                local_parent_span = start_active_span(
+                    "process-federation-request",
+                    # Cross-link back to the remote outgoing trace so we jump over
+                    # there.
+                    links=[Link(remote_parent_span.get_span_context())],
                 )
 
             else:
-                # just use our context as a parent
-                scope = start_active_span(
-                    "incoming-federation-request",
+                # Otherwise just use our local context as a parent
+                local_parent_span = start_active_span(
+                    "process-federation-request",
                 )
 
-            try:
-                with scope:
-                    if origin and self.RATELIMIT:
-                        with ratelimiter.ratelimit(origin) as d:
-                            await d
-                            if request._disconnected:
-                                logger.warning(
-                                    "client disconnected before we started processing "
-                                    "request"
-                                )
-                                return None
-                            response = await func(
-                                origin, content, request.args, *args, **kwargs
+                # Don't need to record anything for the remote
+                remote_parent_span = create_non_recording_span()
+
+            with remote_parent_span, local_parent_span:
+                if origin and self.RATELIMIT:
+                    with ratelimiter.ratelimit(origin) as d:
+                        await d
+                        if request._disconnected:
+                            logger.warning(
+                                "client disconnected before we started processing "
+                                "request"
                             )
-                    else:
+                            return None
                         response = await func(
                             origin, content, request.args, *args, **kwargs
                         )
-            finally:
-                # if we used the origin's context as the parent, add a new span using
-                # the servlet span as a parent, so that we have a link
-                if context:
-                    scope2 = start_active_span_follows_from(
-                        "process-federation_request",
-                        contexts=(scope.span,),
-                        start_time=processing_start_time,
+                else:
+                    response = await func(
+                        origin, content, request.args, *args, **kwargs
                     )
-                    scope2.close()
 
             return response