mirror of
https://git.anonymousland.org/anonymousland/synapse.git
synced 2024-09-07 04:41:57 +00:00
Gracefully handle a pending logging connection during shutdown. (#8685)
This commit is contained in:
parent
f21e24ffc2
commit
8b42a4eefd
1
changelog.d/8607.feature
Normal file
1
changelog.d/8607.feature
Normal file
@ -0,0 +1 @@
|
|||||||
|
Support generating structured logs via the standard logging configuration.
|
@ -1 +0,0 @@
|
|||||||
Re-organize the structured logging code to separate the TCP transport handling from the JSON formatting.
|
|
1
changelog.d/8685.feature
Normal file
1
changelog.d/8685.feature
Normal file
@ -0,0 +1 @@
|
|||||||
|
Support generating structured logs via the standard logging configuration.
|
@ -26,7 +26,7 @@ from typing_extensions import Deque
|
|||||||
from zope.interface import implementer
|
from zope.interface import implementer
|
||||||
|
|
||||||
from twisted.application.internet import ClientService
|
from twisted.application.internet import ClientService
|
||||||
from twisted.internet.defer import Deferred
|
from twisted.internet.defer import CancelledError, Deferred
|
||||||
from twisted.internet.endpoints import (
|
from twisted.internet.endpoints import (
|
||||||
HostnameEndpoint,
|
HostnameEndpoint,
|
||||||
TCP4ClientEndpoint,
|
TCP4ClientEndpoint,
|
||||||
@ -34,6 +34,7 @@ from twisted.internet.endpoints import (
|
|||||||
)
|
)
|
||||||
from twisted.internet.interfaces import IPushProducer, ITransport
|
from twisted.internet.interfaces import IPushProducer, ITransport
|
||||||
from twisted.internet.protocol import Factory, Protocol
|
from twisted.internet.protocol import Factory, Protocol
|
||||||
|
from twisted.python.failure import Failure
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@ -131,9 +132,11 @@ class RemoteHandler(logging.Handler):
|
|||||||
factory = Factory.forProtocol(Protocol)
|
factory = Factory.forProtocol(Protocol)
|
||||||
self._service = ClientService(endpoint, factory, clock=_reactor)
|
self._service = ClientService(endpoint, factory, clock=_reactor)
|
||||||
self._service.startService()
|
self._service.startService()
|
||||||
|
self._stopping = False
|
||||||
self._connect()
|
self._connect()
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
|
self._stopping = True
|
||||||
self._service.stopService()
|
self._service.stopService()
|
||||||
|
|
||||||
def _connect(self) -> None:
|
def _connect(self) -> None:
|
||||||
@ -146,17 +149,21 @@ class RemoteHandler(logging.Handler):
|
|||||||
|
|
||||||
self._connection_waiter = self._service.whenConnected(failAfterFailures=1)
|
self._connection_waiter = self._service.whenConnected(failAfterFailures=1)
|
||||||
|
|
||||||
@self._connection_waiter.addErrback
|
def fail(failure: Failure) -> None:
|
||||||
def fail(r):
|
# If the Deferred was cancelled (e.g. during shutdown) do not try to
|
||||||
r.printTraceback(file=sys.__stderr__)
|
# reconnect (this will cause an infinite loop of errors).
|
||||||
|
if failure.check(CancelledError) and self._stopping:
|
||||||
|
return
|
||||||
|
|
||||||
|
# For a different error, print the traceback and re-connect.
|
||||||
|
failure.printTraceback(file=sys.__stderr__)
|
||||||
self._connection_waiter = None
|
self._connection_waiter = None
|
||||||
self._connect()
|
self._connect()
|
||||||
|
|
||||||
@self._connection_waiter.addCallback
|
def writer(result: Protocol) -> None:
|
||||||
def writer(r):
|
|
||||||
# We have a connection. If we already have a producer, and its
|
# We have a connection. If we already have a producer, and its
|
||||||
# transport is the same, just trigger a resumeProducing.
|
# transport is the same, just trigger a resumeProducing.
|
||||||
if self._producer and r.transport is self._producer.transport:
|
if self._producer and result.transport is self._producer.transport:
|
||||||
self._producer.resumeProducing()
|
self._producer.resumeProducing()
|
||||||
self._connection_waiter = None
|
self._connection_waiter = None
|
||||||
return
|
return
|
||||||
@ -167,12 +174,14 @@ class RemoteHandler(logging.Handler):
|
|||||||
|
|
||||||
# Make a new producer and start it.
|
# Make a new producer and start it.
|
||||||
self._producer = LogProducer(
|
self._producer = LogProducer(
|
||||||
buffer=self._buffer, transport=r.transport, format=self.format,
|
buffer=self._buffer, transport=result.transport, format=self.format,
|
||||||
)
|
)
|
||||||
r.transport.registerProducer(self._producer, True)
|
result.transport.registerProducer(self._producer, True)
|
||||||
self._producer.resumeProducing()
|
self._producer.resumeProducing()
|
||||||
self._connection_waiter = None
|
self._connection_waiter = None
|
||||||
|
|
||||||
|
self._connection_waiter.addCallbacks(writer, fail)
|
||||||
|
|
||||||
def _handle_pressure(self) -> None:
|
def _handle_pressure(self) -> None:
|
||||||
"""
|
"""
|
||||||
Handle backpressure by shedding records.
|
Handle backpressure by shedding records.
|
||||||
|
@ -151,3 +151,19 @@ class RemoteHandlerTestCase(LoggerCleanupMixin, TestCase):
|
|||||||
+ ["warn %s" % (i,) for i in range(15, 20)],
|
+ ["warn %s" % (i,) for i in range(15, 20)],
|
||||||
logs,
|
logs,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def test_cancel_connection(self):
|
||||||
|
"""
|
||||||
|
Gracefully handle the connection being cancelled.
|
||||||
|
"""
|
||||||
|
handler = RemoteHandler(
|
||||||
|
"127.0.0.1", 9000, maximum_buffer=10, _reactor=self.reactor
|
||||||
|
)
|
||||||
|
logger = self.get_logger(handler)
|
||||||
|
|
||||||
|
# Send a message.
|
||||||
|
logger.info("Hello there, %s!", "wally")
|
||||||
|
|
||||||
|
# Do not accept the connection and shutdown. This causes the pending
|
||||||
|
# connection to be cancelled (and should not raise any exceptions).
|
||||||
|
handler.close()
|
||||||
|
Loading…
Reference in New Issue
Block a user