Skip to content

Commit 02f1c73

Browse files
wedamijaclaude
andcommitted
fix(rq): Don't monitor jobs with unsupported schedules
Match the Celery beat integration: skip jobs whose schedule can't be expressed as a Sentry Crons monitor config instead of sending check-ins without one. Also close the in_progress check-in with an error if perform_job raises. Co-Authored-By: Claude <noreply@anthropic.com>
1 parent d1549d3 commit 02f1c73

2 files changed

Lines changed: 105 additions & 28 deletions

File tree

‎sentry_sdk/integrations/rq.py‎

Lines changed: 44 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,32 @@ def sentry_patched_perform_job(
9999
with capture_internal_exceptions():
100100
check_in = _start_check_in(integration, job)
101101

102+
try:
103+
rv = _perform_job(self, job, queue, *args, **kwargs)
104+
except BaseException:
105+
if check_in is not None:
106+
_finish_check_in(check_in, MonitorStatus.ERROR)
107+
if self.is_horse:
108+
sentry_sdk.get_client().flush()
109+
raise
110+
111+
if check_in is not None:
112+
_finish_check_in(
113+
check_in, MonitorStatus.OK if rv else MonitorStatus.ERROR
114+
)
115+
116+
if self.is_horse:
117+
# We're inside of a forked process and RQ is
118+
# about to call `os._exit`. Make sure that our
119+
# events get sent out.
120+
sentry_sdk.get_client().flush()
121+
122+
return rv
123+
124+
def _perform_job(
125+
self: "Any", job: "Job", queue: "Queue", *args: "Any", **kwargs: "Any"
126+
) -> bool:
127+
client = sentry_sdk.get_client()
102128
with sentry_sdk.new_scope() as scope:
103129
scope.clear_breadcrumbs()
104130
scope.add_event_processor(_make_event_processor(weakref.ref(job)))
@@ -149,23 +175,6 @@ def sentry_patched_perform_job(
149175

150176
rv = old_perform_job(self, job, queue, *args, **kwargs)
151177

152-
if check_in is not None:
153-
with capture_internal_exceptions():
154-
monitor_slug, monitor_config, check_in_id, start = check_in
155-
capture_checkin(
156-
monitor_slug=monitor_slug,
157-
monitor_config=monitor_config,
158-
check_in_id=check_in_id,
159-
duration=now() - start,
160-
status=MonitorStatus.OK if rv else MonitorStatus.ERROR,
161-
)
162-
163-
if self.is_horse:
164-
# We're inside of a forked process and RQ is
165-
# about to call `os._exit`. Make sure that our
166-
# events get sent out.
167-
sentry_sdk.get_client().flush()
168-
169178
return rv
170179

171180
worker_cls.perform_job = sentry_patched_perform_job
@@ -319,9 +328,23 @@ def _get_monitor_config(job: "Job") -> "Optional[MonitorConfig]":
319328
return None
320329

321330

331+
def _finish_check_in(
332+
check_in: "Tuple[str, MonitorConfig, Optional[str], float]", status: str
333+
) -> None:
334+
with capture_internal_exceptions():
335+
monitor_slug, monitor_config, check_in_id, start = check_in
336+
capture_checkin(
337+
monitor_slug=monitor_slug,
338+
monitor_config=monitor_config,
339+
check_in_id=check_in_id,
340+
duration=now() - start,
341+
status=status,
342+
)
343+
344+
322345
def _start_check_in(
323346
integration: "RqIntegration", job: "Job"
324-
) -> "Optional[Tuple[str, Optional[MonitorConfig], Optional[str], float]]":
347+
) -> "Optional[Tuple[str, MonitorConfig, Optional[str], float]]":
325348
"""
326349
Send an `in_progress` check-in for jobs that rq-scheduler's `cron()` and
327350
`schedule(interval=...)` re-enqueue. It stores their schedule in the meta.
@@ -342,10 +365,11 @@ def _start_check_in(
342365
monitor_config = _get_monitor_config(job)
343366
if monitor_config is None:
344367
logger.debug(
345-
"Schedule of RQ job %r can't be expressed as a Sentry Crons "
346-
"schedule. Sending check-ins without a monitor config.",
368+
"Not monitoring RQ job %r: its schedule can't be expressed as a "
369+
"Sentry Crons schedule.",
347370
monitor_slug,
348371
)
372+
return None
349373

350374
check_in_id = capture_checkin(
351375
monitor_slug=monitor_slug,

‎tests/integrations/rq/test_rq_crons.py‎

Lines changed: 61 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -103,8 +103,6 @@ def test_cron_job_generated_id(sentry_init, capture_envelopes):
103103
(300, {"type": "interval", "value": 5, "unit": "minute"}),
104104
(7200, {"type": "interval", "value": 2, "unit": "hour"}),
105105
(86400, {"type": "interval", "value": 1, "unit": "day"}),
106-
(30, None),
107-
(90, None),
108106
],
109107
)
110108
def test_interval_job(sentry_init, capture_envelopes, interval, schedule):
@@ -121,20 +119,75 @@ def test_interval_job(sentry_init, capture_envelopes, interval, schedule):
121119
assert [c["status"] for c in check_ins] == ["in_progress", "ok"]
122120
assert {c["monitor_slug"] for c in check_ins} == {"poll"}
123121
for check_in in check_ins:
124-
assert check_in.get("monitor_config") == (
125-
None if schedule is None else {"schedule": schedule}
122+
assert check_in["monitor_config"] == {"schedule": schedule}
123+
124+
125+
@pytest.mark.parametrize("interval", [30, 90])
126+
def test_unsupported_interval_not_monitored(sentry_init, capture_envelopes, interval):
127+
sentry_init(integrations=[RqIntegration(monitor_cron_jobs=True)])
128+
envelopes = capture_envelopes()
129+
130+
_run(
131+
lambda s: s.schedule(
132+
datetime.now(timezone.utc), ok_job, interval=interval, id="poll"
126133
)
134+
)
135+
136+
assert _check_ins(envelopes) == []
137+
138+
139+
def test_unsupported_cron_string_not_monitored(sentry_init, capture_envelopes):
140+
sentry_init(integrations=[RqIntegration(monitor_cron_jobs=True)])
141+
envelopes = capture_envelopes()
142+
143+
_run(lambda s: s.cron("0 0 1 * 1", func=ok_job, id="first-or-monday"))
144+
145+
assert _check_ins(envelopes) == []
146+
147+
148+
def test_local_time_cron_not_monitored(monkeypatch, sentry_init, capture_envelopes):
149+
_set_local_timezone(monkeypatch, "America/New_York")
150+
sentry_init(integrations=[RqIntegration(monitor_cron_jobs=True)])
151+
envelopes = capture_envelopes()
152+
153+
_run(lambda s: s.cron("0 * * * *", func=ok_job, id="hourly"))
154+
155+
assert _check_ins(envelopes) == []
127156

128157

129-
def test_repeat_job_without_config(sentry_init, capture_envelopes):
158+
def test_repeat_job_not_monitored(sentry_init, capture_envelopes):
130159
sentry_init(integrations=[RqIntegration(monitor_cron_jobs=True)])
131160
envelopes = capture_envelopes()
132161

133162
_run(lambda s: s.cron("0 * * * *", func=ok_job, id="limited", repeat=3))
134163

135-
check_ins = _check_ins(envelopes)
136-
assert [c["status"] for c in check_ins] == ["in_progress", "ok"]
137-
assert all("monitor_config" not in c for c in check_ins)
164+
assert _check_ins(envelopes) == []
165+
166+
167+
def test_perform_job_raises(monkeypatch, sentry_init, capture_envelopes):
168+
sentry_init(integrations=[RqIntegration(monitor_cron_jobs=True)])
169+
envelopes = capture_envelopes()
170+
171+
def handle_execution_ended(*args, **kwargs):
172+
raise ConnectionError("redis down")
173+
174+
connection = FakeStrictRedis()
175+
queue = rq.Queue(connection=connection)
176+
scheduler = rq_scheduler.Scheduler(queue=queue, connection=connection)
177+
job = scheduler.cron("0 * * * *", func=ok_job, id="hourly")
178+
scheduler.enqueue_job(job)
179+
worker = rq.SimpleWorker([queue], connection=connection)
180+
# Raises from rq's own exception handler too.
181+
monkeypatch.setattr(worker, "handle_execution_ended", handle_execution_ended)
182+
183+
with pytest.raises(ConnectionError):
184+
worker.perform_job(job, queue)
185+
186+
in_progress, error = _check_ins(envelopes)
187+
assert in_progress["status"] == "in_progress"
188+
assert error["status"] == "error"
189+
assert error["check_in_id"] == in_progress["check_in_id"]
190+
assert error["monitor_config"] == in_progress["monitor_config"]
138191

139192

140193
def test_one_off_jobs_not_monitored(sentry_init, capture_envelopes):

0 commit comments

Comments
 (0)