Merge remote-tracking branch 'origin/develop' into dbkr/email_notifs

3 files changed, 219 insertions, 39 deletions

diff --git a/synapse/http/client.py b/synapse/http/client.py
index 6c89b20984..902ae7a203 100644
--- a/synapse/http/client.py
+++ b/synapse/http/client.py
@@ -462,5 +462,8 @@ class InsecureInterceptableContextFactory(ssl.ContextFactory):
         self._context = SSL.Context(SSL.SSLv23_METHOD)
         self._context.set_verify(VERIFY_NONE, lambda *_: None)
 
-    def getContext(self, hostname, port):
+    def getContext(self, hostname=None, port=None):
         return self._context
+
+    def creatorForNetloc(self, hostname, port):
+        return self
diff --git a/synapse/http/server.py b/synapse/http/server.py
index b82196fd5e..f705abab94 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -74,7 +74,12 @@ response_db_txn_duration = metrics.register_distribution(
 _next_request_id = 0
 
 
-def request_handler(request_handler):
+def request_handler(report_metrics=True):
+    """Decorator for ``wrap_request_handler``"""
+    return lambda request_handler: wrap_request_handler(request_handler, report_metrics)
+
+
+def wrap_request_handler(request_handler, report_metrics):
     """Wraps a method that acts as a request handler with the necessary logging
     and exception handling.
 
@@ -96,7 +101,12 @@ def request_handler(request_handler):
         global _next_request_id
         request_id = "%s-%s" % (request.method, _next_request_id)
         _next_request_id += 1
+
         with LoggingContext(request_id) as request_context:
+            if report_metrics:
+                request_metrics = RequestMetrics()
+                request_metrics.start(self.clock)
+
             request_context.request = request_id
             with request.processing():
                 try:
@@ -133,6 +143,14 @@ def request_handler(request_handler):
                         },
                         send_cors=True
                     )
+                finally:
+                    try:
+                        if report_metrics:
+                            request_metrics.stop(
+                                self.clock, request, self.__class__.__name__
+                            )
+                    except:
+                        pass
     return wrapped_request_handler
 
 
@@ -197,19 +215,23 @@ class JsonResource(HttpServer, resource.Resource):
         self._async_render(request)
         return server.NOT_DONE_YET
 
-    @request_handler
+    # Disable metric reporting because _async_render does its own metrics.
+    # It does its own metric reporting because _async_render dispatches to
+    # a callback and it's the class name of that callback we want to report
+    # against rather than the JsonResource itself.
+    @request_handler(report_metrics=False)
     @defer.inlineCallbacks
     def _async_render(self, request):
         """ This gets called from render() every time someone sends us a request.
             This checks if anyone has registered a callback for that method and
             path.
         """
-        start = self.clock.time_msec()
         if request.method == "OPTIONS":
             self._send_response(request, 200, {})
             return
 
-        start_context = LoggingContext.current_context()
+        request_metrics = RequestMetrics()
+        request_metrics.start(self.clock)
 
         # Loop through all the registered callbacks to check if the method
         # and path regex match
@@ -241,40 +263,7 @@ class JsonResource(HttpServer, resource.Resource):
                 self._send_response(request, code, response)
 
             try:
-                context = LoggingContext.current_context()
-
-                tag = ""
-                if context:
-                    tag = context.tag
-
-                    if context != start_context:
-                        logger.warn(
-                            "Context have unexpectedly changed %r, %r",
-                            context, self.start_context
-                        )
-                        return
-
-                incoming_requests_counter.inc(request.method, servlet_classname, tag)
-
-                response_timer.inc_by(
-                    self.clock.time_msec() - start, request.method,
-                    servlet_classname, tag
-                )
-
-                ru_utime, ru_stime = context.get_resource_usage()
-
-                response_ru_utime.inc_by(
-                    ru_utime, request.method, servlet_classname, tag
-                )
-                response_ru_stime.inc_by(
-                    ru_stime, request.method, servlet_classname, tag
-                )
-                response_db_txn_count.inc_by(
-                    context.db_txn_count, request.method, servlet_classname, tag
-                )
-                response_db_txn_duration.inc_by(
-                    context.db_txn_duration, request.method, servlet_classname, tag
-                )
+                request_metrics.stop(self.clock, request, servlet_classname)
             except:
                 pass
 
@@ -307,6 +296,48 @@ class JsonResource(HttpServer, resource.Resource):
         )
 
 
+class RequestMetrics(object):
+    def start(self, clock):
+        self.start = clock.time_msec()
+        self.start_context = LoggingContext.current_context()
+
+    def stop(self, clock, request, servlet_classname):
+        context = LoggingContext.current_context()
+
+        tag = ""
+        if context:
+            tag = context.tag
+
+            if context != self.start_context:
+                logger.warn(
+                    "Context have unexpectedly changed %r, %r",
+                    context, self.start_context
+                )
+                return
+
+        incoming_requests_counter.inc(request.method, servlet_classname, tag)
+
+        response_timer.inc_by(
+            clock.time_msec() - self.start, request.method,
+            servlet_classname, tag
+        )
+
+        ru_utime, ru_stime = context.get_resource_usage()
+
+        response_ru_utime.inc_by(
+            ru_utime, request.method, servlet_classname, tag
+        )
+        response_ru_stime.inc_by(
+            ru_stime, request.method, servlet_classname, tag
+        )
+        response_db_txn_count.inc_by(
+            context.db_txn_count, request.method, servlet_classname, tag
+        )
+        response_db_txn_duration.inc_by(
+            context.db_txn_duration, request.method, servlet_classname, tag
+        )
+
+
 class RootRedirect(resource.Resource):
     """Redirects the root '/' path to another path."""
 
diff --git a/synapse/http/site.py b/synapse/http/site.py
new file mode 100644
index 0000000000..4b09d7ee66
--- /dev/null
+++ b/synapse/http/site.py
@@ -0,0 +1,146 @@
+# Copyright 2016 OpenMarket 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.
+
+from synapse.util.logcontext import LoggingContext
+from twisted.web.server import Site, Request
+
+import contextlib
+import logging
+import re
+import time
+
+ACCESS_TOKEN_RE = re.compile(r'(\?.*access(_|%5[Ff])token=)[^&]*(.*)$')
+
+
+class SynapseRequest(Request):
+    def __init__(self, site, *args, **kw):
+        Request.__init__(self, *args, **kw)
+        self.site = site
+        self.authenticated_entity = None
+        self.start_time = 0
+
+    def __repr__(self):
+        # We overwrite this so that we don't log ``access_token``
+        return '<%s at 0x%x method=%s uri=%s clientproto=%s site=%s>' % (
+            self.__class__.__name__,
+            id(self),
+            self.method,
+            self.get_redacted_uri(),
+            self.clientproto,
+            self.site.site_tag,
+        )
+
+    def get_redacted_uri(self):
+        return ACCESS_TOKEN_RE.sub(
+            r'\1<redacted>\3',
+            self.uri
+        )
+
+    def get_user_agent(self):
+        return self.requestHeaders.getRawHeaders("User-Agent", [None])[-1]
+
+    def started_processing(self):
+        self.site.access_logger.info(
+            "%s - %s - Received request: %s %s",
+            self.getClientIP(),
+            self.site.site_tag,
+            self.method,
+            self.get_redacted_uri()
+        )
+        self.start_time = int(time.time() * 1000)
+
+    def finished_processing(self):
+
+        try:
+            context = LoggingContext.current_context()
+            ru_utime, ru_stime = context.get_resource_usage()
+            db_txn_count = context.db_txn_count
+            db_txn_duration = context.db_txn_duration
+        except:
+            ru_utime, ru_stime = (0, 0)
+            db_txn_count, db_txn_duration = (0, 0)
+
+        self.site.access_logger.info(
+            "%s - %s - {%s}"
+            " Processed request: %dms (%dms, %dms) (%dms/%d)"
+            " %sB %s \"%s %s %s\" \"%s\"",
+            self.getClientIP(),
+            self.site.site_tag,
+            self.authenticated_entity,
+            int(time.time() * 1000) - self.start_time,
+            int(ru_utime * 1000),
+            int(ru_stime * 1000),
+            int(db_txn_duration * 1000),
+            int(db_txn_count),
+            self.sentLength,
+            self.code,
+            self.method,
+            self.get_redacted_uri(),
+            self.clientproto,
+            self.get_user_agent(),
+        )
+
+    @contextlib.contextmanager
+    def processing(self):
+        self.started_processing()
+        yield
+        self.finished_processing()
+
+
+class XForwardedForRequest(SynapseRequest):
+    def __init__(self, *args, **kw):
+        SynapseRequest.__init__(self, *args, **kw)
+
+    """
+    Add a layer on top of another request that only uses the value of an
+    X-Forwarded-For header as the result of C{getClientIP}.
+    """
+    def getClientIP(self):
+        """
+        @return: The client address (the first address) in the value of the
+            I{X-Forwarded-For header}.  If the header is not present, return
+            C{b"-"}.
+        """
+        return self.requestHeaders.getRawHeaders(
+            b"x-forwarded-for", [b"-"])[0].split(b",")[0].strip()
+
+
+class SynapseRequestFactory(object):
+    def __init__(self, site, x_forwarded_for):
+        self.site = site
+        self.x_forwarded_for = x_forwarded_for
+
+    def __call__(self, *args, **kwargs):
+        if self.x_forwarded_for:
+            return XForwardedForRequest(self.site, *args, **kwargs)
+        else:
+            return SynapseRequest(self.site, *args, **kwargs)
+
+
+class SynapseSite(Site):
+    """
+    Subclass of a twisted http Site that does access logging with python's
+    standard logging
+    """
+    def __init__(self, logger_name, site_tag, config, resource, *args, **kwargs):
+        Site.__init__(self, resource, *args, **kwargs)
+
+        self.site_tag = site_tag
+
+        proxied = config.get("x_forwarded", False)
+        self.requestFactory = SynapseRequestFactory(self, proxied)
+        self.access_logger = logging.getLogger(logger_name)
+
+    def log(self, request):
+        pass