Add basic opentracing support

changelog.d/5544.misc

@@ -0,0 +1 @@
+Added opentracing and configuration options.

docs/sample_config.yaml

@@ -1352,3 +1352,20 @@ password_config:
 #  module: "my_custom_project.SuperRulesSet"
 #  config:
 #    example_option: 'things'
+
+
+## Opentracing ##
+# These settings enable opentracing which implements distributed tracing
+# This allows you to observe the causal chain of events across servers
+# including requests, key lookups etc. across any server running
+# synapse or any other other services which supports opentracing.
+# (specifically those implemented with jaeger)
+
+#opentracing:
+#  # Enable / disable tracer
+#  tracer_enabled: false
+#  # The list of homeservers we wish to expose our current traces to.
+#  # The list is a list of regexes which are matched against the
+#  # servername of the homeserver
+#  homeserver_whitelist:
+#    - ".*"

synapse/app/_base.py

@@ -240,6 +240,9 @@ def handle_sighup(*args, **kwargs):
         # Load the certificate from disk.
         refresh_certificate(hs)
 
+        # Start the tracer
+        synapse.util.tracerutils.TracerUtil.init_tracer(hs.config)
+
         # It is now safe to start your Synapse.
         hs.start_listening(listeners)
         hs.get_datastore().start_profiling()

synapse/config/homeserver.py

@@ -40,6 +40,7 @@
 from .stats import StatsConfig
 from .third_party_event_rules import ThirdPartyRulesConfig
 from .tls import TlsConfig
+from .tracer import TracerConfig
 from .user_directory import UserDirectoryConfig
 from .voip import VoipConfig
 from .workers import WorkerConfig
@@ -75,5 +76,6 @@ class HomeServerConfig(
     ServerNoticesConfig,
     RoomDirectoryConfig,
     ThirdPartyRulesConfig,
+    TracerConfig,
 ):
     pass

synapse/config/tracer.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 The Matrix.org Foundation C.I.C.d
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ._base import Config, ConfigError
+
+
+class TracerConfig(Config):
+    def read_config(self, config, **kwargs):
+        self.tracer_config = config.get("opentracing")
+
+        if self.tracer_config is None:
+            # If the tracer is not configured we assume it is disabled
+            self.tracer_config = {"tracer_enabled": False}
+
+        if self.tracer_config.get("tracer_enabled", False):
+            # The tracer is enabled so sanitize the config
+            # If no whitelists are given
+            self.tracer_config.setdefault("homeserver_whitelist", [])
+
+            if not isinstance(self.tracer_config.get("homeserver_whitelist"), list):
+                raise ConfigError("Tracer homesererver_whitelist config is malformed")
+
+    def generate_config_section(cls, **kwargs):
+        return """\
+        ## Opentracing ##
+        # These settings enable opentracing which implements distributed tracing
+        # This allows you to observe the causal chain of events across servers
+        # including requests, key lookups etc. across any server running
+        # synapse or any other other services which supports opentracing.
+        # (specifically those implemented with jaeger)
+
+        #opentracing:
+        #  # Enable / disable tracer
+        #  tracer_enabled: false
+        #  # The list of homeservers we wish to expose our current traces to.
+        #  # The list is a list of regexes which are matched against the
+        #  # servername of the homeserver
+        #  homeserver_whitelist:
+        #    - ".*"
+        """

synapse/http/matrixfederationclient.py

@@ -51,6 +51,7 @@
 from synapse.util.async_helpers import timeout_deferred
 from synapse.util.logcontext import make_deferred_yieldable
 from synapse.util.metrics import Measure
+from synapse.util.tracerutils import TracerUtil
 
 logger = logging.getLogger(__name__)
 
@@ -339,7 +340,23 @@ def _send_request(
         else:
             query_bytes = b""
 
-        headers_dict = {b"User-Agent": [self.version_string_bytes]}
+        # Retreive current span
+        TracerUtil.start_active_span(
+            "outgoing-federation-request",
+            tags={
+                TracerUtil.Tags.SPAN_KIND: TracerUtil.Tags.SPAN_KIND_RPC_CLIENT,
+                TracerUtil.Tags.PEER_ADDRESS: request.destination,
+                TracerUtil.Tags.HTTP_METHOD: request.method,
+                TracerUtil.Tags.HTTP_URL: request.path,
+            },
+            finish_on_close=True,
+        )
+
+        # Inject the span into the headers
+        headers_dict = {}
+        TracerUtil.inject_active_span_byte_dict(headers_dict, request.destination)
+
+        headers_dict[b"User-Agent"] = [self.version_string_bytes]
 
         with limiter:
             # XXX: Would be much nicer to retry only at the transaction-layer
@@ -419,6 +436,8 @@ def _send_request(
                         response.phrase.decode("ascii", errors="replace"),
                     )
 
+                    TracerUtil.set_tag(TracerUtil.Tags.HTTP_STATUS_CODE, response.code)
+
                     if 200 <= response.code < 300:
                         pass
                     else:
@@ -499,8 +518,8 @@ def _send_request(
                         _flatten_response_never_received(e),
                     )
                     raise
-
-            defer.returnValue(response)
+        TracerUtil.close_active_span()
+        defer.returnValue(response)
 
     def build_auth_headers(
         self, destination, method, url_bytes, content=None, destination_is=None

synapse/http/site.py

@@ -20,6 +20,7 @@
 from synapse.http import redact_uri
 from synapse.http.request_metrics import RequestMetrics, requests_counter
 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
+from synapse.util.tracerutils import TracerUtil
 
 logger = logging.getLogger(__name__)
 
@@ -211,6 +212,7 @@ def _started_processing(self, servlet_name):
 
         This will log the request's arrival. Once the request completes,
         be sure to call finished_processing.
+        It will also start a span for this request.
 
         Args:
             servlet_name (str): the name of the servlet which will be
@@ -233,6 +235,19 @@ def _started_processing(self, servlet_name):
             self.get_redacted_uri(),
         )
 
+        # Start a span
+        TracerUtil.start_active_span_from_context(
+            self.requestHeaders,
+            "incoming-federation-request",
+            tags={
+                "request_id": self.get_request_id(),
+                TracerUtil.Tags.SPAN_KIND: TracerUtil.Tags.SPAN_KIND_RPC_SERVER,
+                TracerUtil.Tags.HTTP_METHOD: self.get_method(),
+                TracerUtil.Tags.HTTP_URL: self.get_redacted_uri(),
+                TracerUtil.Tags.PEER_HOST_IPV6: self.getClientIP(),
+            },
+        )
+
     def _finished_processing(self):
         """Log the completion of this request and update the metrics
         """
@@ -302,6 +317,10 @@ def _finished_processing(self):
             usage.evt_db_fetch_count,
         )
 
+        # finish the span if it's there.
+        TracerUtil.set_tag("peer.address", authenticated_entity)
+        TracerUtil.close_active_span()
+
         try:
             self.request_metrics.stop(self.finish_time, self.code, self.sentLength)
         except Exception as e:

synapse/python_dependencies.py

@@ -95,6 +95,7 @@
     "url_preview": ["lxml>=3.5.0"],
     "test": ["mock>=2.0", "parameterized"],
     "sentry": ["sentry-sdk>=0.7.2"],
+    "opentracing": ["jaeger-client>=4.0.0", "opentracing>=2.2.0"],
 }
 
 ALL_OPTIONAL_REQUIREMENTS = set()

synapse/util/logcontext.py

@@ -176,6 +176,7 @@ class LoggingContext(object):
         "alive",
         "request",
         "tag",
+        "scope",
     ]
 
     thread_local = threading.local()
@@ -228,6 +229,7 @@ def __init__(self, name=None, parent_context=None, request=None):
         self.request = None
         self.tag = ""
         self.alive = True
+        self.scope = None
 
         self.parent_context = parent_context
 
@@ -312,10 +314,12 @@ def copy_to(self, record):
         another LoggingContext
         """
 
-        # 'request' is the only field we currently use in the logger, so that's
-        # all we need to copy
+        # we track the current request
         record.request = self.request
 
+        # we also track the current scope:
+        record.scope = self.scope
+
     def start(self):
         if threading.current_thread() is not self.main_thread:
             logger.warning("Started logcontext %s on different thread", self)

synapse/util/scopecontextmanager.py

@@ -0,0 +1,135 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 The Matrix.org Foundation C.I.C.d
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.import logging
+
+import logging
+
+from opentracing import Scope, ScopeManager
+
+from .logcontext import LoggingContext, nested_logging_context
+
+logger = logging.getLogger(__name__)
+
+
+class LogContextScopeManager(ScopeManager):
+    """
+    The LogContextScopeManager tracks the active scope in opentracing
+    by using the log contexts which are native to synapse. This is so
+    that the basic opentracing api can be used across twisted defereds.
+    (I would love to break logcontexts and this into an OS package. but
+    let's wait for twisted's contexts to be released.)
+    """
+
+    def __init__(self, config):
+        # Set the whitelists
+        logger.info(config.tracer_config)
+        self._homeserver_whitelist = config.tracer_config["homeserver_whitelist"]
+
+    @property
+    def active(self):
+        """
+        Returns the currently active Scope which can be used to access the
+        currently active Scope.span.
+        If there is a non-null Scope, its wrapped Span
+        becomes an implicit parent of any newly-created Span at
+        Tracer.start_active_span() time.
+
+        Return:
+            (Scope) : the Scope that is active, or None if not
+            available.
+        """
+        ctx = LoggingContext.current_context()
+        if ctx is LoggingContext.sentinel:
+            return None
+        else:
+            return ctx.scope
+
+    def activate(self, span, finish_on_close):
+        """
+        Makes a Span active.
+        Args
+            span (Span): the span that should become active.
+            finish_on_close (Boolean): whether Span should be automatically
+                finished when Scope.close() is called.
+
+        Returns:
+            Scope to control the end of the active period for
+            *span*. It is a programming error to neglect to call
+            Scope.close() on the returned instance.
+        """
+
+        enter_logcontext = False
+        ctx = LoggingContext.current_context()
+
+        if ctx is LoggingContext.sentinel:
+            # We don't want this scope to affect.
+            logger.error("Tried to activate scope outside of loggingcontext")
+            return Scope(None, span)
+        elif ctx.scope is not None:
+            # We want the logging scope to look exactly the same so we give it
+            # a blank suffix
+            ctx = nested_logging_context("")
+            enter_logcontext = True
+
+        scope = _LogContextScope(self, span, ctx, enter_logcontext, finish_on_close)
+        ctx.scope = scope
+        return scope
+
+
+class _LogContextScope(Scope):
+    """
+    A custom opentracing scope. The only significant difference is that it will
+    close the log context it's related to if the logcontext was created specifically
+    for this scope.
+    """
+
+    def __init__(self, manager, span, logcontext, enter_logcontext, finish_on_close):
+        """
+        Args:
+            manager (LogContextScopeManager):
+                the manager that is responsible for this scope.
+            span (Span):
+                the opentracing span which this scope represents the local
+                lifetime for.
+            logcontext (LogContext):
+                the logcontext to which this scope is attached.
+            enter_logcontext (Boolean):
+                if True the logcontext will be entered and exited when the scope
+                is entered and exited respectively
+            finish_on_close (Boolean):
+                if True finish the span when the scope is closed
+        """
+        super(_LogContextScope, self).__init__(manager, span)
+        self.logcontext = logcontext
+        self._finish_on_close = finish_on_close
+        self._enter_logcontext = enter_logcontext
+
+    def __enter__(self):
+        if self._enter_logcontext:
+            self.logcontext.__enter__()
+
+    def __exit__(self, type, value, traceback):
+        super(_LogContextScope, self).__exit__(type, value, traceback)
+        if self._enter_logcontext:
+            self.logcontext.__exit__(type, value, traceback)
+        else:  # the logcontext existed before the creation of the scope
+            self.logcontext.scope = None
+
+    def close(self):
+        if self.manager.active is not self:
+            logger.error("Tried to close a none active scope!")
+            return
+
+        if self._finish_on_close:
+            self.span.finish()