feat(rq): Add cron monitoring for rq-scheduler jobs

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"],

sentry_sdk/integrations/rq.py

@@ -1,9 +1,12 @@
 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.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 +19,9 @@
     event_from_exception,
     format_timestamp,
     has_data_collection_enabled,
+    logger,
+    match_regex_list,
+    now,
     parse_version,
 )
 
@@ -39,18 +45,31 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from typing import Any, Callable
+    from typing import Any, Callable, List, Optional, Set, Tuple
 
     from rq.job import Job
 
-    from sentry_sdk._types import Event, EventProcessor
+    from sentry_sdk._types import (
+        Event,
+        EventProcessor,
+        MonitorConfig,
+        MonitorConfigScheduleUnit,
+    )
     from sentry_sdk.utils import ExcInfo
 
 
 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,9 +88,14 @@
             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)
 
+            check_in = None
+            with capture_internal_exceptions():
+                check_in = _start_check_in(integration, job)
+
             with sentry_sdk.new_scope() as scope:
                 scope.clear_breadcrumbs()
                 scope.add_event_processor(_make_event_processor(weakref.ref(job)))
@@ -122,6 +146,10 @@
 
                         rv = old_perform_job(self, job, queue, *args, **kwargs)
 
+            if check_in is not None:
+                with capture_internal_exceptions():
+                    _finish_check_in(check_in, rv)
+
             if self.is_horse:
                 # We're inside of a forked process and RQ is
                 # about to call `os._exit`. Make sure that our
@@ -229,3 +257,113 @@
     )
 
     sentry_sdk.capture_event(event, hint=hint)
+
+
+# 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}"
+)
+_warned_func_names: "Set[str]" = set()
+_CRONTAB_FIELD = re.compile(r"[0-9*,/-]+")
+_INTERVAL_UNITS: "Tuple[Tuple[MonitorConfigScheduleUnit, int], ...]" = (
+    ("day", 60 * 60 * 24),
+    ("hour", 60 * 60),
+    ("minute", 60),
+)
+
+
+def _get_crontab_config(cron_string: str) -> "Optional[MonitorConfig]":
+    fields = cron_string.split()
+    if len(fields) != 5 or not all(_CRONTAB_FIELD.fullmatch(f) for f in 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):
+        for unit, seconds in _INTERVAL_UNITS:
+            if interval >= seconds and interval % seconds == 0:
+                return {
+                    "schedule": {
+                        "type": "interval",
+                        "value": interval // seconds,
+                        "unit": unit,
+                    }
+                }
+    return None
+
+
+def _start_check_in(
+    integration: "RqIntegration", job: "Job"
+) -> "Optional[Tuple[str, Optional[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
+        if monitor_slug not in _warned_func_names:
+            _warned_func_names.add(monitor_slug)
+            logger.warning(
+                "RQ scheduled job %r has no explicit `id`, so its function name "
+                "is used as its monitor slug, and jobs for the same function "
+                "share a monitor. Pass `id` to give it its own monitor.",
+                monitor_slug,
+            )
+    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(
+            "Schedule of RQ job %r can't be expressed as a Sentry Crons "
+            "schedule. Sending check-ins without a monitor config.",
+            monitor_slug,
+        )
+
+    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()
+
+
+def _finish_check_in(
+    check_in: "Tuple[str, Optional[MonitorConfig], Optional[str], float]", ok: bool
+) -> None:
+    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=MonitorStatus.OK if ok else MonitorStatus.ERROR,
+    )