Merge pull request #192 from matrix-org/erikj/fix_log_context

Fix log context when sending requests
This commit is contained in:
Erik Johnston 2015-06-19 16:21:40 +01:00
commit 7fa1363fb0
3 changed files with 68 additions and 52 deletions

View file

@ -109,7 +109,7 @@ class SimpleHttpClient(object):
bodyProducer=FileBodyProducer(StringIO(query_bytes))
)
body = yield readBody(response)
body = yield preserve_context_over_fn(readBody, response)
defer.returnValue(json.loads(body))
@ -128,7 +128,7 @@ class SimpleHttpClient(object):
bodyProducer=FileBodyProducer(StringIO(json_str))
)
body = yield readBody(response)
body = yield preserve_context_over_fn(readBody, response)
defer.returnValue(json.loads(body))
@ -161,7 +161,7 @@ class SimpleHttpClient(object):
})
)
body = yield readBody(response)
body = yield preserve_context_over_fn(readBody, response)
if 200 <= response.code < 300:
defer.returnValue(json.loads(body))
@ -204,7 +204,7 @@ class SimpleHttpClient(object):
bodyProducer=FileBodyProducer(StringIO(json_str))
)
body = yield readBody(response)
body = yield preserve_context_over_fn(readBody, response)
if 200 <= response.code < 300:
defer.returnValue(json.loads(body))
@ -238,7 +238,7 @@ class CaptchaServerHttpClient(SimpleHttpClient):
)
try:
body = yield readBody(response)
body = yield preserve_context_over_fn(readBody, response)
defer.returnValue(body)
except PartialDownloadError as e:
# twisted dislikes google's response, no content length.

View file

@ -127,7 +127,7 @@ class MatrixFederationHttpClient(object):
("", "", path_bytes, param_bytes, query_bytes, "",)
)
txn_id = "%s-%s" % (method, self._next_id)
txn_id = "%s-O-%s" % (method, self._next_id)
self._next_id = (self._next_id + 1) % (sys.maxint - 1)
outbound_logger.info(
@ -139,7 +139,9 @@ class MatrixFederationHttpClient(object):
# (once we have reliable transactions in place)
retries_left = 5
endpoint = self._getEndpoint(reactor, destination)
endpoint = preserve_context_over_fn(
self._getEndpoint, reactor, destination
)
log_result = None
try:
@ -149,21 +151,25 @@ class MatrixFederationHttpClient(object):
producer = body_callback(method, url_bytes, headers_dict)
try:
request_deferred = preserve_context_over_fn(
self.agent.request,
destination,
endpoint,
method,
path_bytes,
param_bytes,
query_bytes,
Headers(headers_dict),
producer
)
def send_request():
request_deferred = self.agent.request(
destination,
endpoint,
method,
path_bytes,
param_bytes,
query_bytes,
Headers(headers_dict),
producer
)
response = yield self.clock.time_bound_deferred(
request_deferred,
time_out=timeout/1000. if timeout else 60,
return self.clock.time_bound_deferred(
request_deferred,
time_out=timeout/1000. if timeout else 60,
)
response = yield preserve_context_over_fn(
send_request,
)
log_result = "%d %s" % (response.code, response.phrase,)
@ -212,7 +218,7 @@ class MatrixFederationHttpClient(object):
else:
# :'(
# Update transactions table?
body = yield readBody(response)
body = yield preserve_context_over_fn(readBody, response)
raise HttpResponseException(
response.code, response.phrase, body
)
@ -292,10 +298,7 @@ class MatrixFederationHttpClient(object):
"Content-Type not application/json"
)
logger.debug("Getting resp body")
body = yield readBody(response)
logger.debug("Got resp body")
body = yield preserve_context_over_fn(readBody, response)
defer.returnValue(json.loads(body))
@defer.inlineCallbacks
@ -338,9 +341,7 @@ class MatrixFederationHttpClient(object):
"Content-Type not application/json"
)
logger.debug("Getting resp body")
body = yield readBody(response)
logger.debug("Got resp body")
body = yield preserve_context_over_fn(readBody, response)
defer.returnValue(json.loads(body))
@ -398,9 +399,7 @@ class MatrixFederationHttpClient(object):
"Content-Type not application/json"
)
logger.debug("Getting resp body")
body = yield readBody(response)
logger.debug("Got resp body")
body = yield preserve_context_over_fn(readBody, response)
defer.returnValue(json.loads(body))
@ -443,7 +442,10 @@ class MatrixFederationHttpClient(object):
headers = dict(response.headers.getAllRawHeaders())
try:
length = yield _readBodyToFile(response, output_stream, max_size)
length = yield preserve_context_over_fn(
_readBodyToFile,
response, output_stream, max_size
)
except:
logger.exception("Failed to download body")
raise

View file

@ -140,6 +140,37 @@ class PreserveLoggingContext(object):
)
class _PreservingContextDeferred(defer.Deferred):
"""A deferred that ensures that all callbacks and errbacks are called with
the given logging context.
"""
def __init__(self, context):
self._log_context = context
defer.Deferred.__init__(self)
def addCallbacks(self, callback, errback=None,
callbackArgs=None, callbackKeywords=None,
errbackArgs=None, errbackKeywords=None):
callback = self._wrap_callback(callback)
errback = self._wrap_callback(errback)
return defer.Deferred.addCallbacks(
self, callback,
errback=errback,
callbackArgs=callbackArgs,
callbackKeywords=callbackKeywords,
errbackArgs=errbackArgs,
errbackKeywords=errbackKeywords,
)
def _wrap_callback(self, f):
def g(res, *args, **kwargs):
with PreserveLoggingContext():
LoggingContext.thread_local.current_context = self._log_context
res = f(res, *args, **kwargs)
return res
return g
def preserve_context_over_fn(fn, *args, **kwargs):
"""Takes a function and invokes it with the given arguments, but removes
and restores the current logging context while doing so.
@ -160,24 +191,7 @@ def preserve_context_over_deferred(deferred):
"""Given a deferred wrap it such that any callbacks added later to it will
be invoked with the current context.
"""
d = defer.Deferred()
current_context = LoggingContext.current_context()
def cb(res):
with PreserveLoggingContext():
LoggingContext.thread_local.current_context = current_context
res = d.callback(res)
return res
def eb(failure):
with PreserveLoggingContext():
LoggingContext.thread_local.current_context = current_context
res = d.errback(failure)
return res
if deferred.called:
return deferred
deferred.addCallbacks(cb, eb)
d = _PreservingContextDeferred(current_context)
deferred.chainDeferred(d)
return d