feat(huey): Add cron monitoring for Huey periodic tasks

sentry_sdk/integrations/huey.py

@@ -5,6 +5,7 @@
 import sentry_sdk
 from sentry_sdk.api import continue_trace, get_baggage, get_traceparent
 from sentry_sdk.consts import OP, SPANDATA, SPANSTATUS
+from sentry_sdk.crons import monitor
 from sentry_sdk.integrations import DidNotEnable, Integration
 from sentry_sdk.scope import should_send_default_pii
 from sentry_sdk.traces import SegmentNameSource, SpanStatus, StreamedSpan
@@ -21,19 +22,21 @@
     ensure_integration_enabled,
     event_from_exception,
     has_data_collection_enabled,
+    logger,
+    match_regex_list,
     reraise,
 )
 
 if TYPE_CHECKING:
-    from typing import Any, Callable, Optional, TypeVar, Union
+    from typing import Any, Callable, List, Optional, TypeVar, Union
 
-    from sentry_sdk._types import Event, EventProcessor, Hint
+    from sentry_sdk._types import Event, EventProcessor, Hint, MonitorConfig
     from sentry_sdk.utils import ExcInfo
 
     F = TypeVar("F", bound=Callable[..., Any])
 
 try:
-    from huey.api import Huey, PeriodicTask, Result, ResultGroup, Task
+    from huey.api import Huey, PeriodicTask, Result, ResultGroup, Task, crontab
     from huey.exceptions import CancelExecution, RetryTask, TaskLockedException
 except ImportError:
     raise DidNotEnable("Huey is not installed")
@@ -53,6 +56,14 @@
     identifier = "huey"
     origin = f"auto.queue.{identifier}"
 
+    def __init__(
+        self,
+        monitor_periodic_tasks: bool = False,
+        exclude_periodic_tasks: "Optional[List[str]]" = None,
+    ) -> None:
+        self.monitor_periodic_tasks = monitor_periodic_tasks
+        self.exclude_periodic_tasks = exclude_periodic_tasks
+
     @staticmethod
     def setup_once() -> None:
         patch_enqueue()
@@ -249,9 +260,93 @@
 
             if not getattr(task, "_sentry_is_patched", False):
                 task.execute = _wrap_task_execute(task.execute)
+                with capture_internal_exceptions():
+                    task.execute = _monitor_periodic_task(self, task, task.execute)
                 task._sentry_is_patched = True
 
             with span_ctx:
                 return old_execute(self, task, timestamp)
 
     Huey._execute = _sentry_execute
+
+
+def _monitor_periodic_task(
+    huey: "Huey", task: "Task", execute: "Callable[..., Any]"
+) -> "Callable[..., Any]":
+    integration = sentry_sdk.get_client().get_integration(HueyIntegration)
+    if (
+        integration is None
+        or not integration.monitor_periodic_tasks
+        or not isinstance(task, PeriodicTask)
+        or match_regex_list(task.name, integration.exclude_periodic_tasks)
+    ):
+        return execute
+
+    monitor_config = _get_monitor_config(huey, task)
+    if monitor_config is None:
+        logger.debug(
+            "Huey periodic task %r can't be expressed as a Sentry Crons "
+            "schedule. Sending check-ins without a monitor config.",
+            task.name,
+        )
+
+    return monitor(monitor_slug=task.name, monitor_config=monitor_config)(execute)
+
+
+_CRONTAB_CODE = crontab().__code__
+
+
+def _get_closure_var(func: "Any", name: str) -> "Any":
+    code = getattr(func, "__code__", None)
+    closure = getattr(func, "__closure__", None)
+    if code is None or closure is None or name not in code.co_freevars:
+        return None
+    return closure[code.co_freevars.index(name)].cell_contents
+
+
+def _get_crontab_field(values: "List[int]", first: int, last: int) -> str:
+    full = list(range(first, last + 1))
+    if values == full:
+        return "*"
+    if len(values) > 2:
+        step = values[1] - values[0]
+        if values == list(range(values[0], values[-1] + 1, step)):
+            if step == 1:
+                return "%d-%d" % (values[0], values[-1])
+            if values == full[::step]:
+                return "*/%d" % step
+            return "%d-%d/%d" % (values[0], values[-1], step)
+    return ",".join(str(v) for v in values)
+
+
+def _get_monitor_config(huey: "Huey", task: "Task") -> "Optional[MonitorConfig]":
+    # Huey only evaluates crontabs in UTC when `utc=True` (the default).
+    if not huey.utc:
+        return None
+
+    # `periodic_task` wraps the `crontab()` validator in a method.
+    validate = _get_closure_var(type(task).validate_datetime, "validate_datetime")
+    if getattr(validate, "__code__", None) is not _CRONTAB_CODE:
+        return None
+    months, days, weekdays, hours, minutes = _get_closure_var(validate, "cron_settings")
+    if not (months and days and weekdays and hours and minutes):
+        return None
+
+    day = _get_crontab_field(days, 1, 31)
+    # Huey 3 expands "*" to 0-7, with 7 also meaning Sunday.
+    weekday = _get_crontab_field(sorted({v % 7 for v in weekdays}), 0, 6)
+    # Huey requires both to match, cron requires either.
+    if day != "*" and weekday != "*":
+        return None
+
+    fields = (
+        _get_crontab_field(minutes, 0, 59),
+        _get_crontab_field(hours, 0, 23),
+        day,
+        _get_crontab_field(months, 1, 12),
+        weekday,
+    )
+    return {
+        "schedule": {"type": "crontab", "value": " ".join(fields)},
+        "timezone": "UTC",
+    }

tests/integrations/huey/test_huey_crons.py

@@ -0,0 +1,193 @@
+from datetime import datetime
+
+import pytest
+from huey import crontab
+from huey.api import MemoryHuey
+
+from sentry_sdk.integrations.huey import HueyIntegration, _get_monitor_config
+
+
+def _check_ins(envelopes):
+    return [
+        item.payload.json
+        for envelope in envelopes
+        for item in envelope.items
+        if item.headers.get("type") == "check_in"
+    ]
+
+
+def _run_periodic_tasks(huey, timestamp=datetime(2026, 1, 1)):
+    """Run the periodic tasks due at `timestamp` the way the consumer does."""
+    for task in huey.read_periodic(timestamp):
+        huey.enqueue(task)
+    while True:
+        task = huey.dequeue()
+        if task is None:
+            break
+        huey.execute(task)
+
+
+def test_periodic_task_ok(sentry_init, capture_envelopes):
+    sentry_init(integrations=[HueyIntegration(monitor_periodic_tasks=True)])
+    envelopes = capture_envelopes()
+    huey = MemoryHuey()
+
+    @huey.periodic_task(crontab(minute="*/5"))
+    def cleanup():
+        pass
+
+    _run_periodic_tasks(huey)
+
+    in_progress, ok = _check_ins(envelopes)
+    assert in_progress["monitor_slug"] == "cleanup"
+    assert in_progress["status"] == "in_progress"
+    assert in_progress["monitor_config"] == {
+        "schedule": {"type": "crontab", "value": "*/5 * * * *"},
+        "timezone": "UTC",
+    }
+    assert ok["monitor_slug"] == "cleanup"
+    assert ok["status"] == "ok"
+    assert ok["check_in_id"] == in_progress["check_in_id"]
+    assert ok["duration"] >= 0
+
+
+def test_periodic_task_error(sentry_init, capture_envelopes):
+    sentry_init(integrations=[HueyIntegration(monitor_periodic_tasks=True)])
+    envelopes = capture_envelopes()
+    huey = MemoryHuey()
+
+    @huey.periodic_task(crontab(minute="0", hour="0"), name="nightly")
+    def failing():
+        raise ValueError("oops")
+
+    _run_periodic_tasks(huey)
+
+    check_ins = _check_ins(envelopes)
+    assert [c["status"] for c in check_ins] == ["in_progress", "error"]
+    assert {c["monitor_slug"] for c in check_ins} == {"nightly"}
+
+    events = [envelope.get_event() for envelope in envelopes]
+    (error,) = [e for e in events if e is not None and e.get("exception")]
+    assert error["exception"]["values"][0]["type"] == "ValueError"
+
+
+def test_regular_tasks_not_monitored(sentry_init, capture_envelopes):
+    sentry_init(integrations=[HueyIntegration(monitor_periodic_tasks=True)])
+    envelopes = capture_envelopes()
+    huey = MemoryHuey()
+
+    @huey.task()
+    def add(a, b):
+        return a + b
+
+    add(1, 2)
+    huey.execute(huey.dequeue())
+
+    assert _check_ins(envelopes) == []
+
+
+def test_periodic_tasks_not_monitored_by_default(sentry_init, capture_envelopes):
+    sentry_init(integrations=[HueyIntegration()])
+    envelopes = capture_envelopes()
+    huey = MemoryHuey()
+
+    @huey.periodic_task(crontab())
+    def cleanup():
+        pass
+
+    _run_periodic_tasks(huey)
+
+    assert _check_ins(envelopes) == []
+
+
+def test_exclude_periodic_tasks(sentry_init, capture_envelopes):
+    sentry_init(
+        integrations=[
+            HueyIntegration(
+                monitor_periodic_tasks=True, exclude_periodic_tasks=["clean.*"]
+            )
+        ]
+    )
+    envelopes = capture_envelopes()
+    huey = MemoryHuey()
+
+    @huey.periodic_task(crontab())
+    def cleanup():
+        pass
+
+    @huey.periodic_task(crontab())
+    def report():
+        pass
+
+    _run_periodic_tasks(huey)
+
+    assert {c["monitor_slug"] for c in _check_ins(envelopes)} == {"report"}
+
+
+def test_periodic_task_without_config(sentry_init, capture_envelopes):
+    sentry_init(integrations=[HueyIntegration(monitor_periodic_tasks=True)])
+    envelopes = capture_envelopes()
+    huey = MemoryHuey()
+
+    def every_minute(timestamp):
+        return True
+
+    @huey.periodic_task(every_minute)
+    def cleanup():
+        pass
+
+    _run_periodic_tasks(huey)
+
+    check_ins = _check_ins(envelopes)
+    assert [c["status"] for c in check_ins] == ["in_progress", "ok"]
+    assert all("monitor_config" not in c for c in check_ins)
+
+
+def _config(huey, validate_datetime):
+    @huey.periodic_task(validate_datetime)
+    def task():
+        pass
+
+    return _get_monitor_config(huey, task.task_class())
+
+
+@pytest.mark.parametrize(
+    "kwargs, value",
+    [
+        ({}, "* * * * *"),
+        ({"minute": "0"}, "0 * * * *"),
+        ({"minute": 30, "hour": "9"}, "30 9 * * *"),
+        ({"minute": "*/15"}, "*/15 * * * *"),
+        ({"minute": "0", "hour": "*/6"}, "0 */6 * * *"),
+        ({"minute": "0", "hour": "9-17"}, "0 9-17 * * *"),
+        ({"minute": "0", "hour": "9,17"}, "0 9,17 * * *"),
+        ({"minute": "0", "hour": "8,12,16,20"}, "0 8-20/4 * * *"),
+        ({"minute": "0", "hour": "1,2,5"}, "0 1,2,5 * * *"),
+        ({"minute": "0", "hour": "0", "day": "1"}, "0 0 1 * *"),
+        ({"minute": "0", "hour": "0", "day": "1-31"}, "0 0 * * *"),
+        ({"minute": "0", "hour": "0", "day": "1", "month": "*/3"}, "0 0 1 */3 *"),
+        ({"minute": "0", "hour": "0", "day_of_week": "1-5"}, "0 0 * * 1-5"),
+        ({"minute": "0", "hour": "0", "day_of_week": "0,6"}, "0 0 * * 0,6"),
+    ],
+)
+def test_get_monitor_config(kwargs, value):
+    assert _config(MemoryHuey(), crontab(**kwargs)) == {
+        "schedule": {"type": "crontab", "value": value},
+        "timezone": "UTC",
+    }
+
+
+@pytest.mark.skipif(
+    not hasattr(crontab, "hourly"), reason="crontab.hourly not available"
+)
+def test_get_monitor_config_hourly():
+    assert _config(MemoryHuey(), crontab.hourly())["schedule"]["value"] == ("0 * * * *")
+
+
+def test_get_monitor_config_day_and_weekday():
+    # Huey requires both to match, cron either.
+    assert _config(MemoryHuey(), crontab(day="1", day_of_week="1")) is None
+
+
+def test_get_monitor_config_local_time():
+    assert _config(MemoryHuey(utc=False), crontab()) is None