raven-python/raven/contrib/celery/__init__.py at e380432340a22a6909e9ee63c98018601f7a0b71 · getsentry/raven-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
"""
raven.contrib.celery
~~~~~~~~~~~~~~~~~~~~

:copyright: (c) 2010-2012 by the Sentry Team, see AUTHORS for more details.
:license: BSD, see LICENSE for more details.
"""
from __future__ import absolute_import

import logging

from celery.exceptions import SoftTimeLimitExceeded
from celery.signals import (
    after_setup_logger, after_setup_task_logger,
    task_failure, task_prerun, task_postrun)
from raven.handlers.logging import SentryHandler


class CeleryFilter(logging.Filter):
    def filter(self, record):
        # Context is fixed in Celery 3.x so use internal flag instead
        extra_data = getattr(record, 'data', {})
        if not isinstance(extra_data, dict):
            return record.funcName != '_log_error'
        # Fallback to funcName for Celery 2.5
        return extra_data.get('internal', record.funcName != '_log_error')


def register_signal(client, ignore_expected=False):
    SentryCeleryHandler(client, ignore_expected=ignore_expected).install()


def register_logger_signal(client, logger=None, loglevel=logging.ERROR):
    filter_ = CeleryFilter()

    handler = SentryHandler(client)
    handler.setLevel(loglevel)
    handler.addFilter(filter_)

    def process_logger_event(sender, logger, loglevel, logfile, format,
                             colorize, **kw):
        # Attempt to find an existing SentryHandler, and if it exists ensure
        # that the CeleryFilter is installed.
        # If one is found, we do not attempt to install another one.
        for h in logger.handlers:
            if isinstance(h, SentryHandler):
                h.addFilter(filter_)
                return False
        logger.addHandler(handler)

    def fix_task_logger(sender, logger=None, **kwargs):
        # If the task logger is set up to not propagate (as Celery is bound
        # to do), give it a SentryHandler from a parent logger. However, if
        # no parent is using Sentry, the task logger should not, either.
        if not logger or logger.propagate:
            return
        if any(isinstance(h, SentryHandler) for h in logger.handlers):
            return
        parent = logger.parent
        while parent:
            for handler in parent.handlers:
                if isinstance(handler, SentryHandler):
                    logger.addHandler(handler)
                    return
            parent = parent.parent

    after_setup_logger.connect(
        process_logger_event,
        weak=False,
        dispatch_uid='raven.contrib.celery.process_logger_event',
    )
    after_setup_task_logger.connect(
        fix_task_logger,
        weak=False,
        dispatch_uid='raven.contrib.celery.fix_task_logger',
    )


class SentryCeleryHandler(object):
    def __init__(self, client, ignore_expected=False):
        self.client = client
        self.ignore_expected = ignore_expected

    def install(self):
        task_prerun.connect(self.handle_task_prerun, weak=False)
        task_postrun.connect(self.handle_task_postrun, weak=False)
        task_failure.connect(self.process_failure_signal, weak=False)

    def uninstall(self):
        task_prerun.disconnect(self.handle_task_prerun)
        task_postrun.disconnect(self.handle_task_postrun)
        task_failure.disconnect(self.process_failure_signal)

    def process_failure_signal(self, sender, task_id, args, kwargs, einfo, **kw):
        if self.ignore_expected and hasattr(sender, 'throws') and isinstance(einfo.exception, sender.throws):
            return

        # This signal is fired inside the stack so let raven do its magic
        if isinstance(einfo.exception, SoftTimeLimitExceeded):
            fingerprint = ['celery', 'SoftTimeLimitExceeded', getattr(sender, 'name', sender)]
        else:
            fingerprint = None

        self.client.captureException(
            extra={
                'task_id': task_id,
                'task': sender,
                'args': args,
                'kwargs': kwargs,
            },
            fingerprint=fingerprint,
        )

    def handle_task_prerun(self, sender, task_id, task, **kw):
        self.client.context.activate()
        self.client.transaction.push(task.name)

    def handle_task_postrun(self, sender, task_id, task, **kw):
        self.client.transaction.pop(task.name)
        self.client.context.clear()