feat: RQ integration (#139)

* feat: RQ integration

* fix: Build matrix

Author: untitaker

sentry_sdk/integrations/celery.py

@@ -39,6 +39,7 @@ def _process_failure_signal(sender, task_id, einfo, **kw):
         return
 
     if isinstance(einfo.exception, SoftTimeLimitExceeded):
+        # TODO: Move this into event processor
         with hub.push_scope() as scope:
             scope.fingerprint = [
                 "celery",

sentry_sdk/integrations/rq.py

@@ -0,0 +1,81 @@
+from __future__ import absolute_import
+
+import weakref
+
+from sentry_sdk.hub import Hub
+from sentry_sdk.integrations import Integration
+from sentry_sdk.utils import capture_internal_exceptions, event_from_exception
+
+from rq.timeouts import JobTimeoutException
+from rq.worker import Worker
+
+
+class RqIntegration(Integration):
+    identifier = "rq"
+
+    @staticmethod
+    def setup_once():
+
+        old_perform_job = Worker.perform_job
+
+        def sentry_patched_perform_job(self, job, *args, **kwargs):
+            hub = Hub.current
+            integration = hub.get_integration(RqIntegration)
+
+            if integration is None:
+                return old_perform_job(self, job, *args, **kwargs)
+
+            with hub.push_scope() as scope:
+                scope.add_event_processor(_make_event_processor(weakref.ref(job)))
+                return old_perform_job(self, job, *args, **kwargs)
+
+        Worker.perform_job = sentry_patched_perform_job
+
+        old_handle_exception = Worker.handle_exception
+
+        def sentry_patched_handle_exception(self, job, *exc_info, **kwargs):
+            _capture_exception(exc_info)
+            return old_handle_exception(self, job, *exc_info, **kwargs)
+
+        Worker.handle_exception = sentry_patched_handle_exception
+
+
+def _make_event_processor(weak_job):
+    def event_processor(event, hint):
+        job = weak_job()
+        if job is not None:
+            with capture_internal_exceptions():
+                if "transaction" not in event:
+                    event["transaction"] = job.func_name
+
+            with capture_internal_exceptions():
+                extra = event.setdefault("extra", {})
+                extra["rq-job"] = {
+                    "job_id": job.id,
+                    "func": job.func_name,
+                    "args": job.args,
+                    "kwargs": job.kwargs,
+                    "description": job.description,
+                }
+
+        if "exc_info" in hint:
+            with capture_internal_exceptions():
+                if issubclass(hint["exc_info"][0], JobTimeoutException):
+                    event["fingerprint"] = ["rq", "JobTimeoutException", job.func_name]
+
+        return event
+
+    return event_processor
+
+
+def _capture_exception(exc_info, **kwargs):
+    hub = Hub.current
+    if hub.get_integration(RqIntegration) is None:
+        return
+    event, hint = event_from_exception(
+        exc_info,
+        client_options=hub.client.options,
+        mechanism={"type": "rq", "handled": False},
+    )
+
+    hub.capture_event(event, hint=hint)

tests/integrations/rq/__init__.py

@@ -0,0 +1,3 @@
+import pytest
+
+rq = pytest.importorskip("rq")

tests/integrations/rq/test_rq.py

@@ -0,0 +1,45 @@
+from sentry_sdk.integrations.rq import RqIntegration
+
+import pytest
+
+from fakeredis import FakeStrictRedis
+from rq import SimpleWorker, Queue
+
+
+@pytest.fixture
+def run_job():
+    queue = Queue(connection=FakeStrictRedis())
+    worker = SimpleWorker([queue], connection=queue.connection)
+
+    def inner(fn, *a, **kw):
+        job = queue.enqueue(fn, *a, **kw)
+        worker.work(burst=True)
+        return job
+
+    return inner
+
+
+def crashing_job(foo):
+    1 / 0
+
+
+def test_basic(sentry_init, capture_events, run_job):
+    sentry_init(integrations=[RqIntegration()])
+
+    events = capture_events()
+    run_job(crashing_job, foo=42)
+    event, = events
+
+    exception, = event["exception"]["values"]
+    assert exception["type"] == "ZeroDivisionError"
+    assert exception["mechanism"]["type"] == "rq"
+    assert exception["stacktrace"]["frames"][-1]["vars"]["foo"] == "42"
+
+    assert event["transaction"] == "tests.integrations.rq.test_rq.crashing_job"
+    assert event["extra"]["rq-job"] == {
+        "args": [],
+        "description": "tests.integrations.rq.test_rq.crashing_job(foo=42)",
+        "func": "tests.integrations.rq.test_rq.crashing_job",
+        "job_id": event["extra"]["rq-job"]["job_id"],
+        "kwargs": {"foo": 42},
+    }

tox.ini

@@ -30,6 +30,14 @@ envlist =
 
     {pypy,py2.7,py3.5,py3.6,py3.7,py3.8}-pyramid-{1.3,1.4,1.5,1.6,1.7,1.8,1.9}
 
+    {pypy,py2.7,py3.5,py3.6}-rq-0.6
+    {pypy,py2.7,py3.5,py3.6}-rq-0.7
+    {pypy,py2.7,py3.5,py3.6}-rq-0.8
+    {pypy,py2.7,py3.5,py3.6}-rq-0.9
+    {pypy,py2.7,py3.5,py3.6}-rq-0.10
+    {pypy,py2.7,py3.5,py3.6}-rq-0.11
+    {pypy,py2.7,py3.5,py3.6,py3.7,py3.8}-rq-0.12
+
 [testenv]
 deps =
     -r test-requirements.txt
@@ -70,6 +78,15 @@ deps =
     pyramid-1.8: pyramid>=1.8,<1.9
     pyramid-1.9: pyramid>=1.9,<1.10
 
+    rq: fakeredis
+    rq-0.6: rq>=0.6,<0.7
+    rq-0.7: rq>=0.7,<0.8
+    rq-0.8: rq>=0.8,<0.9
+    rq-0.9: rq>=0.9,<0.10
+    rq-0.10: rq>=0.10,<0.11
+    rq-0.11: rq>=0.11,<0.12
+    rq-0.12: rq>=0.12,<0.13
+
     linters: black
     linters: flake8
 setenv =
@@ -82,6 +99,7 @@ setenv =
     aws_lambda: TESTPATH=tests/integrations/aws_lambda
     sanic: TESTPATH=tests/integrations/sanic
     pyramid: TESTPATH=tests/integrations/pyramid
+    rq: TESTPATH=tests/integrations/rq
 passenv =
     AWS_ACCESS_KEY_ID
     AWS_SECRET_ACCESS_KEY