mirror of
https://github.com/element-hq/synapse
synced 2024-09-04 17:10:06 +00:00
Merge pull request #2095 from matrix-org/rav/cull_log_preserves
Cull spurious PreserveLoggingContexts
This commit is contained in:
commit
773e64cc1a
4 changed files with 31 additions and 43 deletions
|
@ -28,7 +28,7 @@ from synapse.api.constants import EventTypes, Membership, RejectedReason
|
||||||
from synapse.events.validator import EventValidator
|
from synapse.events.validator import EventValidator
|
||||||
from synapse.util import unwrapFirstError
|
from synapse.util import unwrapFirstError
|
||||||
from synapse.util.logcontext import (
|
from synapse.util.logcontext import (
|
||||||
PreserveLoggingContext, preserve_fn, preserve_context_over_deferred
|
preserve_fn, preserve_context_over_deferred
|
||||||
)
|
)
|
||||||
from synapse.util.metrics import measure_func
|
from synapse.util.metrics import measure_func
|
||||||
from synapse.util.logutils import log_function
|
from synapse.util.logutils import log_function
|
||||||
|
@ -394,11 +394,10 @@ class FederationHandler(BaseHandler):
|
||||||
target_user = UserID.from_string(target_user_id)
|
target_user = UserID.from_string(target_user_id)
|
||||||
extra_users.append(target_user)
|
extra_users.append(target_user)
|
||||||
|
|
||||||
with PreserveLoggingContext():
|
self.notifier.on_new_room_event(
|
||||||
self.notifier.on_new_room_event(
|
event, event_stream_id, max_stream_id,
|
||||||
event, event_stream_id, max_stream_id,
|
extra_users=extra_users
|
||||||
extra_users=extra_users
|
)
|
||||||
)
|
|
||||||
|
|
||||||
if event.type == EventTypes.Member:
|
if event.type == EventTypes.Member:
|
||||||
if event.membership == Membership.JOIN:
|
if event.membership == Membership.JOIN:
|
||||||
|
@ -916,11 +915,10 @@ class FederationHandler(BaseHandler):
|
||||||
origin, auth_chain, state, event
|
origin, auth_chain, state, event
|
||||||
)
|
)
|
||||||
|
|
||||||
with PreserveLoggingContext():
|
self.notifier.on_new_room_event(
|
||||||
self.notifier.on_new_room_event(
|
event, event_stream_id, max_stream_id,
|
||||||
event, event_stream_id, max_stream_id,
|
extra_users=[joinee]
|
||||||
extra_users=[joinee]
|
)
|
||||||
)
|
|
||||||
|
|
||||||
logger.debug("Finished joining %s to %s", joinee, room_id)
|
logger.debug("Finished joining %s to %s", joinee, room_id)
|
||||||
finally:
|
finally:
|
||||||
|
@ -1035,10 +1033,9 @@ class FederationHandler(BaseHandler):
|
||||||
target_user = UserID.from_string(target_user_id)
|
target_user = UserID.from_string(target_user_id)
|
||||||
extra_users.append(target_user)
|
extra_users.append(target_user)
|
||||||
|
|
||||||
with PreserveLoggingContext():
|
self.notifier.on_new_room_event(
|
||||||
self.notifier.on_new_room_event(
|
event, event_stream_id, max_stream_id, extra_users=extra_users
|
||||||
event, event_stream_id, max_stream_id, extra_users=extra_users
|
)
|
||||||
)
|
|
||||||
|
|
||||||
if event.type == EventTypes.Member:
|
if event.type == EventTypes.Member:
|
||||||
if event.content["membership"] == Membership.JOIN:
|
if event.content["membership"] == Membership.JOIN:
|
||||||
|
@ -1084,11 +1081,10 @@ class FederationHandler(BaseHandler):
|
||||||
)
|
)
|
||||||
|
|
||||||
target_user = UserID.from_string(event.state_key)
|
target_user = UserID.from_string(event.state_key)
|
||||||
with PreserveLoggingContext():
|
self.notifier.on_new_room_event(
|
||||||
self.notifier.on_new_room_event(
|
event, event_stream_id, max_stream_id,
|
||||||
event, event_stream_id, max_stream_id,
|
extra_users=[target_user],
|
||||||
extra_users=[target_user],
|
)
|
||||||
)
|
|
||||||
|
|
||||||
defer.returnValue(event)
|
defer.returnValue(event)
|
||||||
|
|
||||||
|
@ -1246,10 +1242,9 @@ class FederationHandler(BaseHandler):
|
||||||
target_user = UserID.from_string(target_user_id)
|
target_user = UserID.from_string(target_user_id)
|
||||||
extra_users.append(target_user)
|
extra_users.append(target_user)
|
||||||
|
|
||||||
with PreserveLoggingContext():
|
self.notifier.on_new_room_event(
|
||||||
self.notifier.on_new_room_event(
|
event, event_stream_id, max_stream_id, extra_users=extra_users
|
||||||
event, event_stream_id, max_stream_id, extra_users=extra_users
|
)
|
||||||
)
|
|
||||||
|
|
||||||
defer.returnValue(None)
|
defer.returnValue(None)
|
||||||
|
|
||||||
|
|
|
@ -612,7 +612,7 @@ class MessageHandler(BaseHandler):
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def _notify():
|
def _notify():
|
||||||
yield run_on_reactor()
|
yield run_on_reactor()
|
||||||
yield self.notifier.on_new_room_event(
|
self.notifier.on_new_room_event(
|
||||||
event, event_stream_id, max_stream_id,
|
event, event_stream_id, max_stream_id,
|
||||||
extra_users=extra_users
|
extra_users=extra_users
|
||||||
)
|
)
|
||||||
|
|
|
@ -202,7 +202,6 @@ class Notifier(object):
|
||||||
lambda: len(self.user_to_user_stream),
|
lambda: len(self.user_to_user_stream),
|
||||||
)
|
)
|
||||||
|
|
||||||
@preserve_fn
|
|
||||||
def on_new_room_event(self, event, room_stream_id, max_room_stream_id,
|
def on_new_room_event(self, event, room_stream_id, max_room_stream_id,
|
||||||
extra_users=[]):
|
extra_users=[]):
|
||||||
""" Used by handlers to inform the notifier something has happened
|
""" Used by handlers to inform the notifier something has happened
|
||||||
|
@ -216,15 +215,13 @@ class Notifier(object):
|
||||||
until all previous events have been persisted before notifying
|
until all previous events have been persisted before notifying
|
||||||
the client streams.
|
the client streams.
|
||||||
"""
|
"""
|
||||||
with PreserveLoggingContext():
|
self.pending_new_room_events.append((
|
||||||
self.pending_new_room_events.append((
|
room_stream_id, event, extra_users
|
||||||
room_stream_id, event, extra_users
|
))
|
||||||
))
|
self._notify_pending_new_room_events(max_room_stream_id)
|
||||||
self._notify_pending_new_room_events(max_room_stream_id)
|
|
||||||
|
|
||||||
self.notify_replication()
|
self.notify_replication()
|
||||||
|
|
||||||
@preserve_fn
|
|
||||||
def _notify_pending_new_room_events(self, max_room_stream_id):
|
def _notify_pending_new_room_events(self, max_room_stream_id):
|
||||||
"""Notify for the room events that were queued waiting for a previous
|
"""Notify for the room events that were queued waiting for a previous
|
||||||
event to be persisted.
|
event to be persisted.
|
||||||
|
@ -242,14 +239,16 @@ class Notifier(object):
|
||||||
else:
|
else:
|
||||||
self._on_new_room_event(event, room_stream_id, extra_users)
|
self._on_new_room_event(event, room_stream_id, extra_users)
|
||||||
|
|
||||||
@preserve_fn
|
|
||||||
def _on_new_room_event(self, event, room_stream_id, extra_users=[]):
|
def _on_new_room_event(self, event, room_stream_id, extra_users=[]):
|
||||||
"""Notify any user streams that are interested in this room event"""
|
"""Notify any user streams that are interested in this room event"""
|
||||||
# poke any interested application service.
|
# poke any interested application service.
|
||||||
self.appservice_handler.notify_interested_services(room_stream_id)
|
preserve_fn(self.appservice_handler.notify_interested_services)(
|
||||||
|
room_stream_id)
|
||||||
|
|
||||||
if self.federation_sender:
|
if self.federation_sender:
|
||||||
self.federation_sender.notify_new_events(room_stream_id)
|
preserve_fn(self.federation_sender.notify_new_events)(
|
||||||
|
room_stream_id
|
||||||
|
)
|
||||||
|
|
||||||
if event.type == EventTypes.Member and event.membership == Membership.JOIN:
|
if event.type == EventTypes.Member and event.membership == Membership.JOIN:
|
||||||
self._user_joined_room(event.state_key, event.room_id)
|
self._user_joined_room(event.state_key, event.room_id)
|
||||||
|
@ -260,7 +259,6 @@ class Notifier(object):
|
||||||
rooms=[event.room_id],
|
rooms=[event.room_id],
|
||||||
)
|
)
|
||||||
|
|
||||||
@preserve_fn
|
|
||||||
def on_new_event(self, stream_key, new_token, users=[], rooms=[]):
|
def on_new_event(self, stream_key, new_token, users=[], rooms=[]):
|
||||||
""" Used to inform listeners that something has happend event wise.
|
""" Used to inform listeners that something has happend event wise.
|
||||||
|
|
||||||
|
@ -287,7 +285,6 @@ class Notifier(object):
|
||||||
|
|
||||||
self.notify_replication()
|
self.notify_replication()
|
||||||
|
|
||||||
@preserve_fn
|
|
||||||
def on_new_replication_data(self):
|
def on_new_replication_data(self):
|
||||||
"""Used to inform replication listeners that something has happend
|
"""Used to inform replication listeners that something has happend
|
||||||
without waking up any of the normal user event streams"""
|
without waking up any of the normal user event streams"""
|
||||||
|
|
|
@ -334,12 +334,8 @@ def preserve_fn(f):
|
||||||
LoggingContext.set_current_context(LoggingContext.sentinel)
|
LoggingContext.set_current_context(LoggingContext.sentinel)
|
||||||
return result
|
return result
|
||||||
|
|
||||||
# XXX: why is this here rather than inside g? surely we want to preserve
|
|
||||||
# the context from the time the function was called, not when it was
|
|
||||||
# wrapped?
|
|
||||||
current = LoggingContext.current_context()
|
|
||||||
|
|
||||||
def g(*args, **kwargs):
|
def g(*args, **kwargs):
|
||||||
|
current = LoggingContext.current_context()
|
||||||
res = f(*args, **kwargs)
|
res = f(*args, **kwargs)
|
||||||
if isinstance(res, defer.Deferred) and not res.called:
|
if isinstance(res, defer.Deferred) and not res.called:
|
||||||
# The function will have reset the context before returning, so
|
# The function will have reset the context before returning, so
|
||||||
|
|
Loading…
Reference in a new issue