fix: Ensure drained queue when RQ kills job (#153)

* fix: Ensure drained queue when RQ kills job

* fix: callback may be none

* ref: Split up testcases

* fix: Fix tests

* fix: Encoding

Author: untitaker

sentry_sdk/integrations/rq.py

@@ -27,7 +27,18 @@ def sentry_patched_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)
+                rv = old_perform_job(self, job, *args, **kwargs)
+
+            if self.is_horse:
+                # We're inside of a forked process and RQ is
+                # about to call `os._exit`. Make sure that our
+                # events get sent out.
+                #
+                # Closing the client should not affect other jobs since
+                # we're in a different process
+                hub.client.close()
+
+            return rv
 
         Worker.perform_job = sentry_patched_perform_job
 

setup.py

@@ -24,19 +24,19 @@
     install_requires=["urllib3", "certifi"],
     extras_require={"flask": ["flask>=0.8", "blinker>=1.1"]},
     classifiers=[
-        'Development Status :: 5 - Production/Stable',
-        'Environment :: Web Environment',
-        'Intended Audience :: Developers',
-        'License :: OSI Approved :: BSD License',
-        'Operating System :: OS Independent',
-        'Programming Language :: Python',
-        'Programming Language :: Python :: 2',
-        'Programming Language :: Python :: 2.7',
-        'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.4',
-        'Programming Language :: Python :: 3.5',
-        'Programming Language :: Python :: 3.6',
-        'Programming Language :: Python :: 3.7',
-        'Topic :: Software Development :: Libraries :: Python Modules',
+        "Development Status :: 5 - Production/Stable",
+        "Environment :: Web Environment",
+        "Intended Audience :: Developers",
+        "License :: OSI Approved :: BSD License",
+        "Operating System :: OS Independent",
+        "Programming Language :: Python",
+        "Programming Language :: Python :: 2",
+        "Programming Language :: Python :: 2.7",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.4",
+        "Programming Language :: Python :: 3.5",
+        "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
+        "Topic :: Software Development :: Libraries :: Python Modules",
     ],
 )

tests/integrations/rq/test_rq.py

@@ -1,33 +1,28 @@
 from sentry_sdk.integrations.rq import RqIntegration
 
-import pytest
+import os
+import json
 
 from fakeredis import FakeStrictRedis
-from rq import SimpleWorker, Queue
+import rq
 
-
-@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
+from sentry_sdk import Hub
 
 
 def crashing_job(foo):
     1 / 0
 
 
-def test_basic(sentry_init, capture_events, run_job):
+def test_basic(sentry_init, capture_events):
     sentry_init(integrations=[RqIntegration()])
-
     events = capture_events()
-    run_job(crashing_job, foo=42)
+
+    queue = rq.Queue(connection=FakeStrictRedis())
+    worker = rq.SimpleWorker([queue], connection=queue.connection)
+
+    queue.enqueue(crashing_job, foo=42)
+    worker.work(burst=True)
+
     event, = events
 
     exception, = event["exception"]["values"]
@@ -43,3 +38,35 @@ def test_basic(sentry_init, capture_events, run_job):
         "job_id": event["extra"]["rq-job"]["job_id"],
         "kwargs": {"foo": 42},
     }
+
+
+def test_transport_shutdown(sentry_init):
+    sentry_init(integrations=[RqIntegration()])
+
+    events_r, events_w = os.pipe()
+    events_r = os.fdopen(events_r, "rb", 0)
+    events_w = os.fdopen(events_w, "wb", 0)
+
+    def capture_event(event):
+        events_w.write(json.dumps(event).encode("utf-8"))
+        events_w.write(b"\n")
+
+    def shutdown(timeout, callback=None):
+        events_w.write(b"shutdown\n")
+
+    Hub.current.client.transport.capture_event = capture_event
+    Hub.current.client.transport.shutdown = shutdown
+
+    queue = rq.Queue(connection=FakeStrictRedis())
+    worker = rq.Worker([queue], connection=queue.connection)
+
+    queue.enqueue(crashing_job, foo=42)
+    worker.work(burst=True)
+
+    event = events_r.readline()
+    shutdown = events_r.readline()
+    event = json.loads(event.decode("utf-8"))
+    assert shutdown == b"shutdown\n"
+
+    exception, = event["exception"]["values"]
+    assert exception["type"] == "ZeroDivisionError"