Cronjob monitoring isn't very ergonomic if you're not using an automatic integration

[Lexicality]

Problem Statement

The sentry.monitor() context manager is incredibly basic and doesn't set up transactions, tags or allow you to configure cronjobs via upsert

Solution Brainstorm

My current solution is this baseclass which works, but it's not great and I'm fairly wary about pulling unexported functions from inner modules.

from __future__ import annotations

import os
from typing import Optional

import sentry_sdk
from django.core.management import BaseCommand
from sentry_sdk.crons.api import capture_checkin
from sentry_sdk.crons.consts import MonitorStatus
from sentry_sdk.utils import now as timing_now


class BaseCronJob(BaseCommand):
    def run_cronjob(self):
        raise NotImplementedError("You must override the `run_cronjob` method!")

    name: Optional[str] = None

    def __init__(self, *args, **kwargs) -> None:
        name_override = os.environ.get("CRONJOB_NAME")
        if name_override is not None:
            self.name = name_override
        super().__init__(*args, **kwargs)

    def create_parser(self, prog_name: str, subcommand: str, **kwargs):
        # Cheeky hack for figuring out the cronjob name from the command name
        if self.name is None:
            self.name = subcommand.replace("_", "-")
        return super().create_parser(prog_name, subcommand, **kwargs)

    def handle(self, *args, **options):
        cronjob_schedule = os.getenv("CRONJOB_SCHEDULE")
        cronjob_startup = os.getenv("CRONJOB_STARTUP_DEADLINE")
        monitor_config = None
        if cronjob_schedule is not None:
            monitor_config = {
                "schedule": {
                    "value": cronjob_schedule,
                    "type": "crontab",
                },
                "timezone": "Etc/UTC",
            }
            if cronjob_startup is not None:
                monitor_config["checkin_margin"] = cronjob_startup

        with sentry_sdk.start_transaction(op="task", name=self.name):
            sentry_sdk.set_tag("cronjob", "true")
            sentry_sdk.set_context("monitor", {"slug": self.name})

            self.stdout.write(f"Starting cronjob {self.name}", self.style.NOTICE)

            start_timestamp = timing_now()
            check_in_id = capture_checkin(
                monitor_slug=self.name,
                status=MonitorStatus.IN_PROGRESS,
                monitor_config=monitor_config,
            )

            try:
                self.run_cronjob()
            except Exception:
                duration_s = timing_now() - start_timestamp
                if check_in_id:
                    capture_checkin(
                        monitor_slug=self.name,
                        check_in_id=check_in_id,
                        status=MonitorStatus.ERROR,
                        duration=duration_s,
                    )
                self.stdout.write(
                    f"Cronjob {self.name} failed after {duration_s:.2}s",
                    self.style.ERROR,
                )
                raise
            else:
                duration_s = timing_now() - start_timestamp
                if check_in_id:
                    capture_checkin(
                        monitor_slug=self.name,
                        check_in_id=check_in_id,
                        status=MonitorStatus.OK,
                        duration=duration_s,
                    )
                self.stdout.write(
                    f"Cronjob {self.name} succeeded after {duration_s:.2}s",
                    self.style.SUCCESS,
                )

[sentrivana]

Thanks @Lexicality, this is good feedback! Since you've had to sort of work around the current API, do you have any suggestions how you'd like the API to look?

[Lexicality]

@sentrivana The current context manager approach is great, however in the same way that you can configure transactions with sentry_sdk.start_transaction() I'd like to be able to configure the cronjob in sentry.monitor()

[Lexicality]

To be clearer, I'd prefer my custom baseclass to look like this:

class BaseCronJob(BaseCommand):
    def run_cronjob(self):
        raise NotImplementedError("You must override the `run_cronjob` method!")

    name: Optional[str] = None

    def __init__(self, *args, **kwargs) -> None:
        name_override = os.environ.get("CRONJOB_NAME")
        if name_override is not None:
            self.name = name_override
        super().__init__(*args, **kwargs)

    def create_parser(self, prog_name: str, subcommand: str, **kwargs):
        # Cheeky hack for figuring out the cronjob name from the command name
        if self.name is None:
            self.name = subcommand.replace("_", "-")
        return super().create_parser(prog_name, subcommand, **kwargs)

    def handle(self, *args, **options):
        with sentry_sdk.monitor(
            self.name,
            crontab_schedule=os.getenv("CRONJOB_SCHEDULE"),
            checkin_margin=os.getenv("CRONJOB_STARTUP_DEADLINE"),
        ):
            self.stdout.write(f"Starting cronjob {self.name}", self.style.NOTICE)
            try:
                self.run_cronjob()
            except Exception:
                self.stdout.write(f"Cronjob {self.name} failed", self.style.ERROR)
                raise
            else:
                self.stdout.write(f"Cronjob {self.name} succeeded", self.style.SUCCESS)

[sentrivana]

Thanks @Lexicality, that makes sense to me -- adding this to our backlog.