Add basic opentracing support

changelog.d/5544.misc

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

docs/sample_config.yaml

@@ -1352,3 +1352,18 @@ password_config:
 #  module: "my_custom_project.SuperRulesSet"
 #  config:
 #    example_option: 'things'
+
+
+## Tracer ##
+
+#tracer:
+#  # Enable / disable tracer
+#  tracer_enabled: false
+#  # The list of users who's requests will be traced
+#  # The list is a list of regex which is matched against the user_id
+#  user_whitelist:
+#    - "*"
+#  # The list of homeservers we wish to trace across
+#  # The list is a list of regex which is matched against the homeserver name
+#  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.config.tracer.init_tracing(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,85 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 New Vector Ltd
+#
+# 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
+
+from jaeger_client import Config as JaegerConfig
+
+from synapse.util.scopecontextmanager import LogContextScopeManager
+
+from ._base import Config
+
+logger = logging.getLogger(__name__)
+
+
+class TracerConfig(Config):
+    def read_config(self, config, **kwargs):
+        self.tracer_config = config.get("tracer")
+
+        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("user_whitelist", ["*"])
+            self.tracer_config.setdefault("homeserver_whitelist", ["*"])
+
+            if type(self.tracer_config.get("user_whitelist")) != list:
+                raise RuntimeError("Tracer user_whitelist config is malformed")
+            if type(self.tracer_config.get("homeserver_whitelist")) != list:
+                raise RuntimeError("Tracer homesererver_whitelist config is malformed")
+
+    def generate_config_section(cls, **kwargs):
+        return """\
+        ## Tracer ##
+
+        #tracer:
+        #  # Enable / disable tracer
+        #  tracer_enabled: false
+        #  # The list of users who's requests will be traced
+        #  # The list is a list of regex which is matched against the user_id
+        #  user_whitelist:
+        #    - "*"
+        #  # The list of homeservers we wish to trace across
+        #  # The list is a list of regex which is matched against the homeserver name
+        #  homeserver_whitelist:
+        #    - "*"
+        """
+
+
+def init_tracing(config):
+    """Initialise the JaegerClient tracer
+
+    Args:
+        config (Config)
+        The config used by the homserver. Here it's used to set the service
+        name to the homeserver's.
+    """
+
+    if config.tracer_config.get("tracer_enabled", False):
+        jaeger_config = JaegerConfig(
+            config={"sampler": {"type": "const", "param": 1}, "logging": True},
+            service_name=config.server_name,
+            scope_manager=LogContextScopeManager(config),
+        )
+    else:  # The tracer is not configured so we instantiate a noop tracer
+        jaeger_config = JaegerConfig(
+            config={"sampler": {"type": "const", "param": 0}},
+            service_name=config.server_name,
+        )
+
+    return jaeger_config.initialize_tracer()

synapse/http/matrixfederationclient.py

@@ -23,8 +23,10 @@
 from six.moves import urllib
 
 import attr
+import opentracing
 import treq
 from canonicaljson import encode_canonical_json
+from opentracing.propagation import Format
 from prometheus_client import Counter
 from signedjson.sign import sign_json
 from zope.interface import implementer
@@ -339,9 +341,26 @@ def _send_request(
         else:
             query_bytes = b""
 
-        headers_dict = {b"User-Agent": [self.version_string_bytes]}
+        # Retreive current span
+        scope = opentracing.tracer.start_active_span(
+            "outgoing-federation-request",
+            tags={
+                "span.kind": "client",  # With respect to this request's role in an rpc
+                "peer.address": request.destination,
+                "http.method": request.method,
+                "http.url": request.path,
+            },
+            finish_on_close=True,
+        )
+
+        # Inject the span into the headers
+        headers_dict = {}
+        opentracing.tracer.inject(scope.span, Format.HTTP_HEADERS, headers_dict)
+        headers_dict = {k.encode(): [v.encode()] for k, v in headers_dict.items()}
+
+        headers_dict[b"User-Agent"] = [self.version_string_bytes]
 
-        with limiter:
+        with limiter, scope:
             # XXX: Would be much nicer to retry only at the transaction-layer
             # (once we have reliable transactions in place)
             if long_retries:
@@ -419,6 +438,12 @@ def _send_request(
                         response.phrase.decode("ascii", errors="replace"),
                     )
 
+                    logger.info(
+                        "Setting response code on span {} *********".format(
+                            opentracing.tracer.active_span
+                        )
+                    )
+                    scope.span.set_tag("http.status_code", response.code)
                     if 200 <= response.code < 300:
                         pass
                     else:
@@ -498,9 +523,15 @@ def _send_request(
                         url_str,
                         _flatten_response_never_received(e),
                     )
+                    logger.info(
+                        "Setting response code on span {} *********".format(
+                            opentracing.tracer.active_span
+                        )
+                    )
+                    # scope.span.set_tag("error", True)
                     raise
 
-            defer.returnValue(response)
+        defer.returnValue(response)
 
     def build_auth_headers(
         self, destination, method, url_bytes, content=None, destination_is=None

synapse/http/site.py

@@ -15,11 +15,14 @@
 import logging
 import time
 
+import opentracing
+
 from twisted.web.server import Request, Site
 
 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 extract_span_context
 
 logger = logging.getLogger(__name__)
 
@@ -211,6 +214,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 +237,20 @@ def _started_processing(self, servlet_name):
             self.get_redacted_uri(),
         )
 
+        # Start a span
+        span_context = extract_span_context(self.requestHeaders)
+        opentracing.tracer.start_active_span(
+            "incoming-federation-request",
+            tags={
+                "request_id": self.get_request_id(),
+                "span.kind": "server",
+                "http.method": self.get_method(),
+                "http.url": self.get_redacted_uri(),
+                "peer.ipv6": self.getClientIP(),
+            },
+            child_of=span_context,
+        )
+
     def _finished_processing(self):
         """Log the completion of this request and update the metrics
         """
@@ -252,7 +270,7 @@ def _finished_processing(self):
         # the time between receiving the request and the request handler finishing
         processing_time = self._processing_finished_time - self.start_time
 
-        # the time between the request handler finishing and the response being sent
+        # the time between the reb'\x80\x03ctwisted.web.http_headers\nHeaders\nq\x00)\x81q\x01}q\x02X\x0b\x00\x00\x00_rawHeadersq\x03}q\x04(C\x04hostq\x05]q\x06C\x0elocalhost:8081q\x07aC\nuser-agentq\x08]q\tCLMozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:67.0) Gecko/20100101 Firefox/67.0q\naC\x06acceptq\x0b]q\x0cC\x10application/jsonq\raC\x0faccept-languageq\x0e]q\x0fC\x0een-GB,en;q=0.5q\x10aC\x0faccept-encodingq\x11]q\x12C\rgzip, deflateq\x13aC\x06originq\x14]q\x15C\x04nullq\x16aC\nconnectionq\x17]q\x18C\nkeep-aliveq\x19ausb.'quest handler finishing and the response being sent
         # to the client (nb may be negative)
         response_send_time = self.finish_time - self._processing_finished_time
 
@@ -302,6 +320,12 @@ def _finished_processing(self):
             usage.evt_db_fetch_count,
         )
 
+        scope = opentracing.tracer.scope_manager.active
+        if scope is not None:
+            # finish the span if it's there.
+            scope.span.set_tag("peer.address", authenticated_entity)
+            scope.__exit__(None, None, None)
+
         try:
             self.request_metrics.stop(self.finish_time, self.code, self.sentLength)
         except Exception as e:

synapse/python_dependencies.py

@@ -72,6 +72,7 @@
     # Twisted 18.7.0 requires attrs>=17.4.0
     "attrs>=17.4.0",
     "netaddr>=0.7.18",
+    "jaeger-client>=4.0.0",
     "Jinja2>=2.9",
     "bleach>=1.4.3",
 ]

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)