Add benchmarks for structured logging performance

changelog.d/6266.misc

@@ -0,0 +1 @@
+Add benchmarks for structured logging and improve output performance.

synapse/benchmarks/__init__.py

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 The Matrix.org Foundation C.I.C.
+#
+# 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 twisted.internet.defer import Deferred
+
+from synapse.config.homeserver import HomeServerConfig
+from synapse.util import Clock
+
+from tests.utils import default_config, setup_test_homeserver, setupdb
+
+DB_SETUP = False
+
+
+def setup_database():
+    global DB_SETUP
+    if not DB_SETUP:
+        setupdb()
+        DB_SETUP = True
+
+
+async def make_homeserver(reactor, config=None):
+    def wait(time):
+        d = Deferred()
+        reactor.callLater(time, d.callback, True)
+        return d
+
+    cleanup_tasks = []
+
+    clock = Clock(reactor)
+
+    if not config:
+        config = default_config("test")
+
+    config_obj = HomeServerConfig()
+    config_obj.parse_config_dict(config, "", "")
+
+    hs = await setup_test_homeserver(
+        cleanup_tasks.append, config=config_obj, reactor=reactor, clock=clock
+    )
+    stor = hs.get_datastore()
+
+    # Run the database background updates.
+    if hasattr(stor, "do_next_background_update"):
+        while not await stor.has_completed_background_updates():
+            await stor.do_next_background_update(1)
+
+    def cleanup():
+        for i in cleanup_tasks:
+            i()
+
+    return hs, wait, cleanup

synapse/benchmarks/__main__.py

@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 The Matrix.org Foundation C.I.C.
+#
+# 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 pyperf
+
+from twisted.python import reflect
+
+from synapse.benchmarks.suites import SUITES
+
+if __name__ == "__main__":
+
+    runner = pyperf.Runner(processes=5, values=1, warmups=0)
+    runner.parse_args()
+    runner.args.inherit_environ = ["SYNAPSE_POSTGRES"]
+
+    for suite, loops in SUITES:
+
+        func = reflect.namedAny("synapse.benchmarks.suites.%s.main" % (suite.lower(),))
+        runner.args.loops = loops
+        runner.bench_time_func(suite + "_" + str(loops), func)

synapse/benchmarks/suites/__init__.py

@@ -0,0 +1 @@
+SUITES = [("LOGGING", 1000), ("LOGGING", 10000)]

synapse/benchmarks/suites/logging.py

@@ -0,0 +1,140 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 The Matrix.org Foundation C.I.C.
+#
+# 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 warnings
+from contextlib import redirect_stderr
+from io import StringIO
+
+from mock import Mock
+
+from pyperf import perf_counter
+
+from twisted.internet.defer import ensureDeferred
+from twisted.internet.protocol import ServerFactory
+from twisted.logger import LogBeginner, Logger, LogPublisher, globalLogBeginner
+from twisted.protocols.basic import LineOnlyReceiver
+from twisted.python.failure import Failure
+
+from synapse.benchmarks import make_homeserver, setup_database
+from synapse.logging._structured import setup_structured_logging
+
+
+class LineCounter(LineOnlyReceiver):
+
+    delimiter = b"\n"
+
+    def __init__(self, *args, **kwargs):
+        self.count = 0
+        super().__init__(*args, **kwargs)
+
+    def lineReceived(self, line):
+        self.count += 1
+
+
+async def _main(reactor, loops):
+
+    servers = []
+
+    def protocol():
+        p = LineCounter()
+        servers.append(p)
+        return p
+
+    logger_factory = ServerFactory.forProtocol(protocol)
+    port = reactor.listenTCP(0, logger_factory, interface="127.0.0.1")
+
+    hs, wait, cleanup = await make_homeserver(reactor)
+
+    errors = StringIO()
+    publisher = LogPublisher()
+    mock_sys = Mock()
+    beginner = LogBeginner(
+        publisher, errors, mock_sys, warnings, initialBufferSize=loops
+    )
+
+    log_config = {
+        "loggers": {"synapse": {"level": "DEBUG"}},
+        "drains": {
+            "tersejson": {
+                "type": "network_json_terse",
+                "host": "127.0.0.1",
+                "port": port.getHost().port,
+                "maximum_buffer": 100,
+            }
+        },
+    }
+
+    logger = Logger(namespace="synapse.logging.test_terse_json", observer=publisher)
+
+    start = perf_counter()
+
+    logging_system = setup_structured_logging(
+        hs, hs.config, log_config, logBeginner=beginner, redirect_stdlib_logging=False
+    )
+
+    # Wait for it to connect...
+    await logging_system._observers[0]._service.whenConnected()
+
+    # Send a bunch of useful messages
+    for i in range(0, loops):
+        logger.info("test message %s" % (i,))
+
+        if (
+            len(logging_system._observers[0]._buffer)
+            == logging_system._observers[0].maximum_buffer
+        ):
+            while (
+                len(logging_system._observers[0]._buffer)
+                > logging_system._observers[0].maximum_buffer / 2
+            ):
+                await wait(0.01)
+
+    while servers[0].count != loops:
+        await wait(0.01)
+
+    end = perf_counter() - start
+
+    logging_system.stop()
+    port.stopListening()
+    cleanup()
+
+    return end
+
+
+def main(loops):
+
+    setup_database()
+
+    if globalLogBeginner._temporaryObserver:
+        globalLogBeginner.beginLoggingTo([lambda event: None])
+
+    file_out = StringIO()
+    with redirect_stderr(file_out):
+
+        from twisted.internet import epollreactor
+
+        reactor = epollreactor.EPollReactor()
+        d = ensureDeferred(_main(reactor, loops))
+
+        def on_done(_):
+            if isinstance(_, Failure):
+                _.printTraceback()
+            reactor.stop()
+            return _
+
+        d.addBoth(on_done)
+        reactor.run()
+
+    return d.result

synapse/logging/_structured.py

@@ -261,6 +261,13 @@ def parse_drain_configs(
             )
 
 
+class StoppableLogPublisher(LogPublisher):
+    def stop(self):
+        for obs in self._observers:
+            if hasattr(obs, "stop"):
+                obs.stop()
+
+
 def setup_structured_logging(
     hs,
     config,
@@ -336,7 +343,7 @@ def setup_structured_logging(
             # We should never get here, but, just in case, throw an error.
             raise ConfigError("%s drain type cannot be configured" % (observer.type,))
 
-    publisher = LogPublisher(*observers)
+    publisher = StoppableLogPublisher(*observers)
     log_filter = LogLevelFilterPredicate()
 
     for namespace, namespace_config in log_config.get(

synapse/logging/_terse_json.py

@@ -17,14 +17,14 @@
 Log formatters that output terse JSON.
 """
 
+import json
 import sys
 from collections import deque
 from ipaddress import IPv4Address, IPv6Address, ip_address
 from math import floor
 from typing import IO
 
 import attr
-from simplejson import dumps
 from zope.interface import implementer
 
 from twisted.application.internet import ClientService
@@ -33,9 +33,11 @@
     TCP4ClientEndpoint,
     TCP6ClientEndpoint,
 )
+from twisted.internet.interfaces import IPushProducer
 from twisted.internet.protocol import Factory, Protocol
 from twisted.logger import FileLogObserver, ILogObserver, Logger
-from twisted.python.failure import Failure
+
+_encoder = json.JSONEncoder(ensure_ascii=False, separators=(",", ":"))
 
 
 def flatten_event(event: dict, metadata: dict, include_time: bool = False):
@@ -141,11 +143,40 @@ def TerseJSONToConsoleLogObserver(outFile: IO[str], metadata: dict) -> FileLogOb
 
     def formatEvent(_event: dict) -> str:
         flattened = flatten_event(_event, metadata)
-        return dumps(flattened, ensure_ascii=False, separators=(",", ":")) + "\n"
+        return _encoder.encode(flattened) + "\n"
 
     return FileLogObserver(outFile, formatEvent)
 
 
+@attr.s
+@implementer(IPushProducer)
+class LogProducer(object):
+
+    _buffer = attr.ib()
+    _transport = attr.ib()
+    paused = attr.ib(default=False)
+
+    def pauseProducing(self):
+        self.paused = True
+
+    def stopProducing(self):
+        self.paused = True
+        self._buffer = None
+
+    def resumeProducing(self):
+        self.paused = False
+
+        while self.paused is False and (self._buffer and self._transport.connected):
+            try:
+                event = self._buffer.popleft()
+                self._transport.write(_encoder.encode(event).encode("utf8"))
+                self._transport.write(b"\n")
+            except Exception:
+                import traceback
+
+                traceback.print_exc(file=sys.__stderr__)
+
+
 @attr.s
 @implementer(ILogObserver)
 class TerseJSONToTCPLogObserver(object):
@@ -167,6 +198,7 @@ class TerseJSONToTCPLogObserver(object):
     _buffer = attr.ib(default=attr.Factory(deque), type=deque)
     _writer = attr.ib(default=None)
     _logger = attr.ib(default=attr.Factory(Logger))
+    _producer = attr.ib(default=None)
 
     def start(self) -> None:
 
@@ -188,34 +220,39 @@ def start(self) -> None:
         self._service = ClientService(endpoint, factory, clock=self.hs.get_reactor())
         self._service.startService()
 
+    def stop(self):
+        self._service.stopService()
+
     def _write_loop(self) -> None:
         """
         Implement the write loop.
         """
         if self._writer:
             return
 
+        if self._producer and self._producer._transport.connected:
+            self._producer.resumeProducing()
+            return
+
         self._writer = self._service.whenConnected()
 
-        @self._writer.addBoth
+        @self._writer.addErrback
+        def fail(r):
+            r.printTraceback(file=sys.__stderr__)
+            self._writer = None
+            self.hs.get_reactor().callLater(1, self._write_loop)
+            return
+
+        @self._writer.addCallback
         def writer(r):
-            if isinstance(r, Failure):
-                r.printTraceback(file=sys.__stderr__)
-                self._writer = None
+            def connectionLost(_self, reason):
+                self._producer.pauseProducing()
+                self._producer = None
                 self.hs.get_reactor().callLater(1, self._write_loop)
-                return
 
-            try:
-                for event in self._buffer:
-                    r.transport.write(
-                        dumps(event, ensure_ascii=False, separators=(",", ":")).encode(
-                            "utf8"
-                        )
-                    )
-                    r.transport.write(b"\n")
-                self._buffer.clear()
-            except Exception as e:
-                sys.__stderr__.write("Failed writing out logs with %s\n" % (str(e),))
+            self._producer = LogProducer(self._buffer, r.transport)
+            r.transport.registerProducer(self._producer, True)
+            self._producer.resumeProducing()
 
             self._writer = False
             self.hs.get_reactor().callLater(1, self._write_loop)