From 3d4945a26d2b592dc29c01e52d4dfacc1c44b3fd Mon Sep 17 00:00:00 2001
From: Itamar Turner-Trauring <itamar@pythonspeed.com>
Date: Tue, 12 Dec 2023 10:33:25 -0500
Subject: [PATCH] Write out Eliot messages in a thread.

---
 src/allmydata/test/test_eliotutil.py | 53 +++++++++-------------------
 src/allmydata/util/eliotutil.py      | 20 +++++------
 2 files changed, 26 insertions(+), 47 deletions(-)

diff --git a/src/allmydata/test/test_eliotutil.py b/src/allmydata/test/test_eliotutil.py
index 52d709e4c..5b191cd92 100644
--- a/src/allmydata/test/test_eliotutil.py
+++ b/src/allmydata/test/test_eliotutil.py
@@ -1,20 +1,7 @@
 """
 Tests for ``allmydata.util.eliotutil``.
-
-Ported to Python 3.
 """
 
-from __future__ import (
-    unicode_literals,
-    print_function,
-    absolute_import,
-    division,
-)
-
-from future.utils import PY2
-if PY2:
-    from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, list, object, range, str, max, min  # noqa: F401
-
 from sys import stdout
 import logging
 
@@ -67,6 +54,7 @@ from ..util.eliotutil import (
     _parse_destination_description,
     _EliotLogging,
 )
+from ..util.deferredutil import async_to_deferred
 
 from .common import (
     SyncTestCase,
@@ -214,13 +202,14 @@ class ParseDestinationDescriptionTests(SyncTestCase):
         )
 
 
-# Opt out of the great features of common.SyncTestCase because we're
-# interacting with Eliot in a very obscure, particular, fragile way. :/
-class EliotLoggingTests(TestCase):
+# We need AsyncTestCase because logging happens in a thread tied to the
+# reactor.
+class EliotLoggingTests(AsyncTestCase):
     """
     Tests for ``_EliotLogging``.
     """
-    def test_stdlib_event_relayed(self):
+    @async_to_deferred
+    async def test_stdlib_event_relayed(self):
         """
         An event logged using the stdlib logging module is delivered to the Eliot
         destination.
@@ -228,23 +217,16 @@ class EliotLoggingTests(TestCase):
         collected = []
         service = _EliotLogging([collected.append])
         service.startService()
-        self.addCleanup(service.stopService)
-
-        # The first destination added to the global log destinations gets any
-        # buffered messages delivered to it.  We don't care about those.
-        # Throw them on the floor.  Sorry.
-        del collected[:]
 
         logging.critical("oh no")
-        self.assertThat(
-            collected,
-            AfterPreprocessing(
-                len,
-                Equals(1),
-            ),
+        await service.stopService()
+
+        self.assertTrue(
+            "oh no" in str(collected[-1]), collected
         )
 
-    def test_twisted_event_relayed(self):
+    @async_to_deferred
+    async def test_twisted_event_relayed(self):
         """
         An event logged with a ``twisted.logger.Logger`` is delivered to the Eliot
         destination.
@@ -252,15 +234,13 @@ class EliotLoggingTests(TestCase):
         collected = []
         service = _EliotLogging([collected.append])
         service.startService()
-        self.addCleanup(service.stopService)
 
         from twisted.logger import Logger
         Logger().critical("oh no")
-        self.assertThat(
-            collected,
-            AfterPreprocessing(
-                len, Equals(1),
-            ),
+        await service.stopService()
+
+        self.assertTrue(
+            "oh no" in str(collected[-1]), collected
         )
 
     def test_validation_failure(self):
@@ -318,7 +298,6 @@ class EliotLoggingTests(TestCase):
         )
 
 
-
 class LogCallDeferredTests(TestCase):
     """
     Tests for ``log_call_deferred``.
diff --git a/src/allmydata/util/eliotutil.py b/src/allmydata/util/eliotutil.py
index 94d34f96f..9e3cf6ae0 100644
--- a/src/allmydata/util/eliotutil.py
+++ b/src/allmydata/util/eliotutil.py
@@ -36,7 +36,7 @@ from attr.validators import (
     optional,
     provides,
 )
-
+from twisted.internet import reactor
 from eliot import (
     ILogger,
     Message,
@@ -58,6 +58,7 @@ from eliot.twisted import (
     DeferredContext,
     inline_callbacks,
 )
+from eliot.logwriter import ThreadedWriter
 from twisted.python.usage import (
     UsageError,
 )
@@ -75,7 +76,7 @@ from twisted.logger import (
 from twisted.internet.defer import (
     maybeDeferred,
 )
-from twisted.application.service import Service
+from twisted.application.service import MultiService
 
 from .jsonbytes import AnyBytesJSONEncoder
 
@@ -144,7 +145,7 @@ def opt_help_eliot_destinations(self):
     raise SystemExit(0)
 
 
-class _EliotLogging(Service):
+class _EliotLogging(MultiService):
     """
     A service which adds stdout as an Eliot destination while it is running.
     """
@@ -153,23 +154,22 @@ class _EliotLogging(Service):
         :param list destinations: The Eliot destinations which will is added by this
             service.
         """
-        self.destinations = destinations
-
+        MultiService.__init__(self)
+        for destination in destinations:
+            service = ThreadedWriter(destination, reactor)
+            service.setServiceParent(self)
 
     def startService(self):
         self.stdlib_cleanup = _stdlib_logging_to_eliot_configuration(getLogger())
         self.twisted_observer = _TwistedLoggerToEliotObserver()
         globalLogPublisher.addObserver(self.twisted_observer)
-        add_destinations(*self.destinations)
-        return Service.startService(self)
+        return MultiService.startService(self)
 
 
     def stopService(self):
-        for dest in self.destinations:
-            remove_destination(dest)
         globalLogPublisher.removeObserver(self.twisted_observer)
         self.stdlib_cleanup()
-        return Service.stopService(self)
+        return MultiService.stopService(self)
 
 
 @implementer(ILogObserver)