feat(rq): Add an option to monitor rq-scheduler jobs (#8053)

Part of a series adding automatic Crons monitoring for the most-used job
schedulers in each SDK. Check-ins for a monitor that doesn't exist yet
are dropped, so today each monitor has to be created by hand with its
schedule copied from the code. Sending the job's schedule as the monitor
config lets Sentry create the monitor and keep it in sync.

This adds `monitor_cron_jobs` to `RqIntegration` for recurring jobs from
rq-scheduler's `scheduler.cron(...)` and `scheduler.schedule(...,
interval=...)`. Only the worker needs the option, and it sends the cron
string or interval as the monitor config. The slug is the job `id`, and
jobs without one share a monitor named after the function's import path.
It's off by default.

```python
sentry_sdk.init(
    integrations=[RqIntegration(monitor_cron_jobs=True, exclude_cron_jobs=["cleanup"])],
)
```

Jobs aren't monitored when Sentry may read the cron string differently
(6 fields, `@daily`, `MON`, both day fields set), the interval isn't
whole minutes, the job has `repeat`, or the worker's local timezone
isn't UTC. RQ's built-in `rq cron` scheduler isn't covered.

The rq test envs now install `rq-scheduler` for RQ 2.x.

Docs: getsentry/sentry-docs#20058

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: wedamija

scripts/populate_tox/config.py

@@ -471,6 +471,7 @@
             # The default "addr" value was removed in https://github.com/cunla/fakeredis-py/commit/0441288fb22c8c191fc716b561e0001cf512abe5.
             # from fakeredis.
             ">=1.1.14": ["fakeredis<2.36.0"],
+            ">=2.0": ["rq-scheduler"],
             "<0.9": ["fakeredis<1.0", "redis<3.2.2"],
             ">=0.9,<0.14": ["fakeredis>=1.0,<1.7.4"],
             "py3.6,py3.7": ["fakeredis!=2.26.0"],

scripts/populate_tox/package_dependencies.jsonl

@@ -0,0 +1,29 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Optional, Tuple
+
+    from sentry_sdk._types import MonitorConfigSchedule, MonitorConfigScheduleUnit
+
+
+_INTERVAL_UNITS: "Tuple[Tuple[MonitorConfigScheduleUnit, int], ...]" = (
+    ("day", 60 * 60 * 24),
+    ("hour", 60 * 60),
+    ("minute", 60),
+)
+
+
+def _get_interval_schedule(seconds: float) -> "Optional[MonitorConfigSchedule]":
+    """
+    Express an interval in the largest unit that divides it evenly.
+
+    Returns None if the interval isn't a whole number of minutes.
+    """
+    for unit, unit_seconds in _INTERVAL_UNITS:
+        if seconds >= unit_seconds and seconds % unit_seconds == 0:
+            return {
+                "type": "interval",
+                "value": int(seconds // unit_seconds),
+                "unit": unit,
+            }
+    return None

sentry_sdk/crons/utils.py

@@ -1,9 +1,13 @@
 import functools
+import re
+import time
 import weakref
 
 import sentry_sdk
 from sentry_sdk.api import continue_trace
 from sentry_sdk.consts import OP, SPANDATA
+from sentry_sdk.crons import MonitorStatus, capture_checkin
+from sentry_sdk.crons.utils import _get_interval_schedule
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
 from sentry_sdk.integrations.logging import ignore_logger
 from sentry_sdk.scope import Scope, should_send_default_pii
@@ -16,6 +20,9 @@
     event_from_exception,
     format_timestamp,
     has_data_collection_enabled,
+    logger,
+    match_regex_list,
+    now,
     parse_version,
 )
 
@@ -39,18 +46,33 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from typing import Any, Callable
+    from typing import Any, Callable, List, Optional, Tuple
 
     from rq.job import Job
 
-    from sentry_sdk._types import Event, EventProcessor
+    from sentry_sdk._types import Event, EventProcessor, MonitorConfig
     from sentry_sdk.utils import ExcInfo
 
+# rq generates `str(uuid4())` job ids.
+_GENERATED_JOB_ID = re.compile(
+    r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
+)
+# `*`, `*/n`, `n`, `n/n`, `n-n` and `n-n/n`.
+_CRONTAB_ITEM = re.compile(r"\*(?:/\d+)?|(\d+)(?:-(\d+)(?:/\d+)?|/\d+)?")
+
 
 class RqIntegration(Integration):
     identifier = "rq"
     origin = f"auto.queue.{identifier}"
 
+    def __init__(
+        self,
+        monitor_cron_jobs: bool = False,
+        exclude_cron_jobs: "Optional[List[str]]" = None,
+    ) -> None:
+        self.monitor_cron_jobs = monitor_cron_jobs
+        self.exclude_cron_jobs = exclude_cron_jobs
+
     @staticmethod
     def setup_once() -> None:
         version = parse_version(RQ_VERSION)
@@ -69,64 +91,87 @@ def sentry_patched_perform_job(
             self: "Any", job: "Job", queue: "Queue", *args: "Any", **kwargs: "Any"
         ) -> bool:
             client = sentry_sdk.get_client()
-            if client.get_integration(RqIntegration) is None:
+            integration = client.get_integration(RqIntegration)
+            if integration is None:
                 return old_perform_job(self, job, queue, *args, **kwargs)
 
-            with sentry_sdk.new_scope() as scope:
-                scope.clear_breadcrumbs()
-                scope.add_event_processor(_make_event_processor(weakref.ref(job)))
-
-                if has_span_streaming_enabled(client.options):
-                    sentry_sdk.traces.continue_trace(
-                        job.meta.get("_sentry_trace_headers") or {}
-                    )
-
-                    Scope.set_custom_sampling_context({"rq_job": job})
-
-                    func_name = None
-                    with capture_internal_exceptions():
-                        func_name = job.func_name
-
-                    with sentry_sdk.traces.start_span(
-                        name="unknown RQ task" if func_name is None else func_name,
-                        attributes={
-                            "sentry.op": OP.QUEUE_TASK_RQ,
-                            "sentry.origin": RqIntegration.origin,
-                            "sentry.segment.name.source": SegmentNameSource.TASK,
-                            SPANDATA.MESSAGING_MESSAGE_ID: job.id,
-                            SPANDATA.MESSAGING_DESTINATION_NAME: queue.name,
-                        },
-                        parent_span=None,
-                    ) as span:
-                        if func_name is not None:
-                            span.set_attribute(SPANDATA.CODE_FUNCTION_NAME, func_name)
-
-                        rv = old_perform_job(self, job, queue, *args, **kwargs)
-                else:
-                    transaction = continue_trace(
-                        job.meta.get("_sentry_trace_headers") or {},
-                        op=OP.QUEUE_TASK_RQ,
-                        name="unknown RQ task",
-                        source=TransactionSource.TASK,
-                        origin=RqIntegration.origin,
-                    )
-
+            check_in = None
+            with capture_internal_exceptions():
+                check_in = _start_check_in(integration, job)
+
+            status = MonitorStatus.ERROR
+            try:
+                with sentry_sdk.new_scope() as scope:
+                    scope.clear_breadcrumbs()
+                    scope.add_event_processor(_make_event_processor(weakref.ref(job)))
+
+                    if has_span_streaming_enabled(client.options):
+                        sentry_sdk.traces.continue_trace(
+                            job.meta.get("_sentry_trace_headers") or {}
+                        )
+
+                        Scope.set_custom_sampling_context({"rq_job": job})
+
+                        func_name = None
+                        with capture_internal_exceptions():
+                            func_name = job.func_name
+
+                        with sentry_sdk.traces.start_span(
+                            name="unknown RQ task" if func_name is None else func_name,
+                            attributes={
+                                "sentry.op": OP.QUEUE_TASK_RQ,
+                                "sentry.origin": RqIntegration.origin,
+                                "sentry.segment.name.source": SegmentNameSource.TASK,
+                                SPANDATA.MESSAGING_MESSAGE_ID: job.id,
+                                SPANDATA.MESSAGING_DESTINATION_NAME: queue.name,
+                            },
+                            parent_span=None,
+                        ) as span:
+                            if func_name is not None:
+                                span.set_attribute(
+                                    SPANDATA.CODE_FUNCTION_NAME, func_name
+                                )
+
+                            rv = old_perform_job(self, job, queue, *args, **kwargs)
+                    else:
+                        transaction = continue_trace(
+                            job.meta.get("_sentry_trace_headers") or {},
+                            op=OP.QUEUE_TASK_RQ,
+                            name="unknown RQ task",
+                            source=TransactionSource.TASK,
+                            origin=RqIntegration.origin,
+                        )
+
+                        with capture_internal_exceptions():
+                            transaction.name = job.func_name
+
+                        with sentry_sdk.start_transaction(
+                            transaction,
+                            custom_sampling_context={"rq_job": job},
+                        ) as span:
+                            span.set_data(
+                                SPANDATA.MESSAGING_DESTINATION_NAME, queue.name
+                            )
+
+                            rv = old_perform_job(self, job, queue, *args, **kwargs)
+                status = MonitorStatus.OK if rv else MonitorStatus.ERROR
+            finally:
+                if check_in is not None:
                     with capture_internal_exceptions():
-                        transaction.name = job.func_name
-
-                    with sentry_sdk.start_transaction(
-                        transaction,
-                        custom_sampling_context={"rq_job": job},
-                    ) as span:
-                        span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, queue.name)
-
-                        rv = old_perform_job(self, job, queue, *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.
-                sentry_sdk.get_client().flush()
+                        monitor_slug, monitor_config, check_in_id, start = check_in
+                        capture_checkin(
+                            monitor_slug=monitor_slug,
+                            monitor_config=monitor_config,
+                            check_in_id=check_in_id,
+                            duration=now() - start,
+                            status=status,
+                        )
+
+                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.
+                    sentry_sdk.get_client().flush()
 
             return rv
 
@@ -229,3 +274,90 @@ def _capture_exception(exc_info: "ExcInfo", **kwargs: "Any") -> None:
     )
 
     sentry_sdk.capture_event(event, hint=hint)
+
+
+def _is_portable_crontab_field(field: str, is_weekday: bool) -> bool:
+    for item in field.split(","):
+        match = _CRONTAB_ITEM.fullmatch(item)
+        if match is None:
+            return False
+        first, last = match.groups()
+        # rq-scheduler's crontab parser and Sentry read these differently.
+        if last is not None and int(last) <= int(first):
+            return False
+        if is_weekday and first is not None and last is None and "/" in item:
+            return False
+    return True
+
+
+def _get_crontab_config(cron_string: str) -> "Optional[MonitorConfig]":
+    fields = cron_string.split()
+    if len(fields) != 5 or not all(
+        _is_portable_crontab_field(field, i == 4) for i, field in enumerate(fields)
+    ):
+        return None
+    # Day of month and day of week are matched differently by different
+    # cron parsers when both are set.
+    if fields[2] != "*" and fields[4] != "*":
+        return None
+    # rq-scheduler evaluates crontabs in the system's local time.
+    if time.timezone != 0 or (time.daylight and time.altzone != 0):
+        return None
+    return {
+        "schedule": {"type": "crontab", "value": " ".join(fields)},
+        "timezone": "UTC",
+    }
+
+
+def _get_monitor_config(job: "Job") -> "Optional[MonitorConfig]":
+    # Jobs with `repeat` stop after a number of runs.
+    if job.meta.get("repeat") is not None:
+        return None
+
+    cron_string = job.meta.get("cron_string")
+    if isinstance(cron_string, str):
+        return _get_crontab_config(cron_string)
+
+    interval = job.meta.get("interval")
+    if isinstance(interval, int):
+        schedule = _get_interval_schedule(interval)
+        if schedule is not None:
+            return {"schedule": schedule}
+    return None
+
+
+def _start_check_in(
+    integration: "RqIntegration", job: "Job"
+) -> "Optional[Tuple[str, MonitorConfig, Optional[str], float]]":
+    """
+    Send an `in_progress` check-in for jobs that rq-scheduler's `cron()` and
+    `schedule(interval=...)` re-enqueue. It stores their schedule in the meta.
+    """
+    if not integration.monitor_cron_jobs:
+        return None
+    if "cron_string" not in job.meta and "interval" not in job.meta:
+        return None
+
+    # Without an explicit `id`, every scheduled job gets a random one.
+    if _GENERATED_JOB_ID.fullmatch(job.id):
+        monitor_slug = job.func_name
+    else:
+        monitor_slug = job.id
+    if match_regex_list(monitor_slug, integration.exclude_cron_jobs):
+        return None
+
+    monitor_config = _get_monitor_config(job)
+    if monitor_config is None:
+        logger.debug(
+            "Not monitoring RQ job %r: its schedule can't be expressed as a "
+            "Sentry Crons schedule.",
+            monitor_slug,
+        )
+        return None
+
+    check_in_id = capture_checkin(
+        monitor_slug=monitor_slug,
+        monitor_config=monitor_config,
+        status=MonitorStatus.IN_PROGRESS,
+    )
+    return monitor_slug, monitor_config, check_in_id, now()