feat(rq): Add an option to monitor 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/crons/utils.py

@@ -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/integrations/rq.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 @@
             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 @@
     )
 
     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()