From 4588cddd8428f07d8f4b38f615518fcca12a92bf Mon Sep 17 00:00:00 2001 From: Dan Fuller Date: Fri, 16 Oct 2026 17:46:00 -0700 Subject: [PATCH] feat(huey): Add cron monitoring for Huey periodic tasks Send check-ins around each run of a Huey periodic task when HueyIntegration(monitor_periodic_tasks=True) is set. Huey crontab() schedules are sent as the monitor config when Huey runs in UTC, so Sentry creates the monitor and keeps it in sync. Co-Authored-By: Claude --- sentry_sdk/integrations/huey.py | 101 ++++++++++- tests/integrations/huey/test_huey_crons.py | 193 +++++++++++++++++++++ 2 files changed, 291 insertions(+), 3 deletions(-) create mode 100644 tests/integrations/huey/test_huey_crons.py diff --git a/sentry_sdk/integrations/huey.py b/sentry_sdk/integrations/huey.py index e1d34f05d7..e7467f2498 100644 --- a/sentry_sdk/integrations/huey.py +++ b/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 @@ 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 +260,93 @@ 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( + "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", + } diff --git a/tests/integrations/huey/test_huey_crons.py b/tests/integrations/huey/test_huey_crons.py new file mode 100644 index 0000000000..6ebba04f5e --- /dev/null +++ b/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