Celery Beat auto monitoring (#1967)

Automatically monitor Celery Beat tasks with Sentry Crons.

How we do this:
- we dynamically create a function that listens to the `celery_beat_init` hook. In the hook we do two things:
- 1.) patch existing scheduled tasks (in `sender.scheduler.schedule`):
    - Each scheduled task is patched to contain information about the Sentry monitor (the monitor slug and config (timezone, schedule, ...) in its headers.
    - We then stop Celery Beat and replace the scheduled tasks with the new patched scheduled tasks  
    - We restart Celery Beat to enable our patched tasks
- 2.) Connect each task to the following hooks to send information about the task to sentry: `task_prerun`, `task_success`, `task_failure`, `task_retry`. (config is sent by the tasks in its headers we set up in 1))

Author: antonpirker

sentry_sdk/crons.py

@@ -0,0 +1,3 @@
+from sentry_sdk.crons.api import capture_checkin  # noqa
+from sentry_sdk.crons.consts import MonitorStatus  # noqa
+from sentry_sdk.crons.decorator import monitor  # noqa

sentry_sdk/crons/__init__.py

@@ -0,0 +1,56 @@
+import uuid
+
+from sentry_sdk import Hub
+from sentry_sdk._types import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from typing import Any, Dict, Optional
+
+
+def _create_check_in_event(
+    monitor_slug=None,
+    check_in_id=None,
+    status=None,
+    duration_s=None,
+    monitor_config=None,
+):
+    # type: (Optional[str], Optional[str], Optional[str], Optional[float], Optional[Dict[str, Any]]) -> Dict[str, Any]
+    options = Hub.current.client.options if Hub.current.client else {}
+    check_in_id = check_in_id or uuid.uuid4().hex  # type: str
+
+    check_in = {
+        "type": "check_in",
+        "monitor_slug": monitor_slug,
+        "monitor_config": monitor_config or {},
+        "check_in_id": check_in_id,
+        "status": status,
+        "duration": duration_s,
+        "environment": options.get("environment", None),
+        "release": options.get("release", None),
+    }
+
+    return check_in
+
+
+def capture_checkin(
+    monitor_slug=None,
+    check_in_id=None,
+    status=None,
+    duration=None,
+    monitor_config=None,
+):
+    # type: (Optional[str], Optional[str], Optional[str], Optional[float], Optional[Dict[str, Any]]) -> str
+    hub = Hub.current
+
+    check_in_id = check_in_id or uuid.uuid4().hex
+    check_in_event = _create_check_in_event(
+        monitor_slug=monitor_slug,
+        check_in_id=check_in_id,
+        status=status,
+        duration_s=duration,
+        monitor_config=monitor_config,
+    )
+    hub.capture_event(check_in_event)
+
+    return check_in_event["check_in_id"]

sentry_sdk/crons/api.py

@@ -0,0 +1,4 @@
+class MonitorStatus:
+    IN_PROGRESS = "in_progress"
+    OK = "ok"
+    ERROR = "error"

sentry_sdk/crons/consts.py

@@ -0,0 +1,74 @@
+from functools import wraps
+import sys
+
+from sentry_sdk._compat import reraise
+from sentry_sdk._types import TYPE_CHECKING
+from sentry_sdk.crons import capture_checkin
+from sentry_sdk.crons.consts import MonitorStatus
+from sentry_sdk.utils import now
+
+
+if TYPE_CHECKING:
+    from typing import Any, Callable, Optional
+
+
+def monitor(monitor_slug=None):
+    # type: (Optional[str]) -> Callable[..., Any]
+    """
+    Decorator to capture checkin events for a monitor.
+
+    Usage:
+    ```
+    import sentry_sdk
+
+    app = Celery()
+
+    @app.task
+    @sentry_sdk.monitor(monitor_slug='my-fancy-slug')
+    def test(arg):
+        print(arg)
+    ```
+
+    This does not have to be used with Celery, but if you do use it with celery,
+    put the `@sentry_sdk.monitor` decorator below Celery's `@app.task` decorator.
+    """
+
+    def decorate(func):
+        # type: (Callable[..., Any]) -> Callable[..., Any]
+        if not monitor_slug:
+            return func
+
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            # type: (*Any, **Any) -> Any
+            start_timestamp = now()
+            check_in_id = capture_checkin(
+                monitor_slug=monitor_slug, status=MonitorStatus.IN_PROGRESS
+            )
+
+            try:
+                result = func(*args, **kwargs)
+            except Exception:
+                duration_s = now() - start_timestamp
+                capture_checkin(
+                    monitor_slug=monitor_slug,
+                    check_in_id=check_in_id,
+                    status=MonitorStatus.ERROR,
+                    duration=duration_s,
+                )
+                exc_info = sys.exc_info()
+                reraise(*exc_info)
+
+            duration_s = now() - start_timestamp
+            capture_checkin(
+                monitor_slug=monitor_slug,
+                check_in_id=check_in_id,
+                status=MonitorStatus.OK,
+                duration=duration_s,
+            )
+
+            return result
+
+        return wrapper
+
+    return decorate