|
|
|
@ -742,8 +742,9 @@ class Worker(object):
|
|
|
|
|
|
|
|
|
|
def handle_exception(self, job, *exc_info):
|
|
|
|
|
"""Walks the exception handler stack to delegate exception handling."""
|
|
|
|
|
exc_string = ''.join(traceback.format_exception_only(*exc_info[:2]) +
|
|
|
|
|
traceback.format_exception(*exc_info))
|
|
|
|
|
exc_string = self._get_safe_exception_string(
|
|
|
|
|
traceback.format_exception_only(*exc_info[:2]) + traceback.format_exception(*exc_info)
|
|
|
|
|
)
|
|
|
|
|
self.log.error(exc_string, exc_info=True, extra={
|
|
|
|
|
'func': job.func_name,
|
|
|
|
|
'arguments': job.args,
|
|
|
|
@ -765,10 +766,16 @@ class Worker(object):
|
|
|
|
|
|
|
|
|
|
def move_to_failed_queue(self, job, *exc_info):
|
|
|
|
|
"""Default exception handler: move the job to the failed queue."""
|
|
|
|
|
exc_string = ''.join(traceback.format_exception(*exc_info))
|
|
|
|
|
exc_string = self._get_safe_exception_string(traceback.format_exception(*exc_info))
|
|
|
|
|
self.log.warning('Moving job to {0!r} queue'.format(self.failed_queue.name))
|
|
|
|
|
self.failed_queue.quarantine(job, exc_info=exc_string)
|
|
|
|
|
|
|
|
|
|
def _get_safe_exception_string(self, exc_strings):
|
|
|
|
|
"""Ensure list of exception strings is decoded on Python 2 and joined as one string safely."""
|
|
|
|
|
if sys.version_info[0] < 3:
|
|
|
|
|
exc_strings = [exc.decode("utf-8") for exc in exc_strings]
|
|
|
|
|
return ''.join(exc_strings)
|
|
|
|
|
|
|
|
|
|
def push_exc_handler(self, handler_func):
|
|
|
|
|
"""Pushes an exception handler onto the exc handler stack."""
|
|
|
|
|
self._exc_handlers.append(handler_func)
|
|
|
|
|