feat(huey): Add an option to monitor Huey periodic tasks (#8052)

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_periodic_tasks` to `HueyIntegration`, matching
Celery's `monitor_beat_tasks`. When it's on, each `@huey.periodic_task`
run sends check-ins with its `crontab(...)` schedule in UTC, so Sentry
creates the monitor on the first run. The slug is the task name. It's
off by default, and regular tasks are unaffected.

```python
sentry_sdk.init(
    integrations=[HueyIntegration(monitor_periodic_tasks=True, exclude_periodic_tasks=["cleanup"])],
)
```

Tasks aren't monitored when they use a custom `validate_datetime`,
restrict both `day` and `day_of_week`, or run on a Huey instance with
`utc=False`.

Docs: getsentry/sentry-docs#20057

---------

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

Author: wedamija

sentry_sdk/integrations/huey.py

@@ -1,10 +1,12 @@
 import sys
 from datetime import datetime
+from functools import wraps
 from typing import TYPE_CHECKING
 
 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 +23,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 +57,14 @@ class HueyIntegration(Integration):
     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 +261,99 @@ def _sentry_execute(
 
             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(
+            "Not monitoring Huey periodic task %r: its schedule can't be "
+            "expressed as a Sentry Crons schedule.",
+            task.name,
+        )
+        return execute
+
+    @wraps(execute)
+    def _sentry_monitored_execute(*args: "Any", **kwargs: "Any") -> "Any":
+        with monitor(monitor_slug=task.name, monitor_config=monitor_config):
+            return execute(*args, **kwargs)
+
+    return _sentry_monitored_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,225 @@
+from datetime import datetime
+
+import pytest
+from huey import crontab
+from huey.api import MemoryHuey
+from huey.exceptions import CancelExecution, RetryTask, TaskLockedException
+
+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"
+
+
+@pytest.mark.parametrize("exception", [CancelExecution, RetryTask, TaskLockedException])
+def test_periodic_task_control_flow(sentry_init, capture_envelopes, exception):
+    sentry_init(integrations=[HueyIntegration(monitor_periodic_tasks=True)])
+    envelopes = capture_envelopes()
+    huey = MemoryHuey()
+
+    calls = []
+
+    @huey.periodic_task(crontab())
+    def cleanup():
+        if not calls:
+            calls.append(1)
+            raise exception()
+
+    _run_periodic_tasks(huey)
+
+    statuses = [c["status"] for c in _check_ins(envelopes)]
+    assert statuses[:2] == ["in_progress", "error"]
+
+
+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 _every_minute(timestamp):
+    return True
+
+
+@pytest.mark.parametrize(
+    "utc, validate_datetime",
+    [
+        (True, _every_minute),
+        (True, crontab(minute="0", hour="0", day="1", day_of_week="1")),
+        (False, crontab()),
+    ],
+)
+def test_unsupported_schedule_not_monitored(
+    sentry_init, capture_envelopes, utc, validate_datetime
+):
+    sentry_init(integrations=[HueyIntegration(monitor_periodic_tasks=True)])
+    envelopes = capture_envelopes()
+    huey = MemoryHuey(utc=utc)
+    calls = []
+
+    @huey.periodic_task(validate_datetime)
+    def cleanup():
+        calls.append(1)
+
+    _run_periodic_tasks(huey, timestamp=datetime(2026, 6, 1))
+
+    assert calls == [1]
+    assert _check_ins(envelopes) == []
+
+
+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