feat(celery): allow custom monitor config for beat tasks

[alexon1234]

Description

Celery Beat auto-instrumentation (CeleryIntegration(monitor_beat_tasks=True)) derives the cron monitor config from the schedule only (schedule + timezone), so there is no way to set max_runtime, checkin_margin, failure_issue_threshold, recovery_threshold, or owner on the monitors the SDK creates.

Consequence: a task that can legitimately run longer than Sentry's default 30-minute max_runtime is reported as a timed-out cron failure even though it completes successfully. The only workaround is to exclude it from auto-instrumentation and re-declare the schedule in a manual @sentry_sdk.monitor(..., monitor_config=...) call, which is awkward when the schedule differs per environment.

This adds a beat_task_monitor_config option to CeleryIntegration: a mapping of Celery Beat task name to a partial MonitorConfig that is merged over the config derived from the schedule.

import sentry_sdk
from sentry_sdk.integrations.celery import CeleryIntegration

sentry_sdk.init(
    integrations=[
        CeleryIntegration(
            monitor_beat_tasks=True,
            beat_task_monitor_config={
                "nightly-cleanup": {"max_runtime": 120, "checkin_margin": 10},
            },
        ),
    ],
)

The SDK still derives schedule and timezone; the override only needs the extra fields, and can still override the derived values if explicitly provided.

Changes:

• CeleryIntegration: new beat_task_monitor_config option.
• _get_monitor_config: accepts and merges monitor config overrides.
• _apply_crons_data_to_schedule_entry: looks up the override for the schedule entry and passes it through (covers both the default scheduler and RedBeat, which share this path).
• Tests: merge into the derived config, schedule override, and per-task matching (unmatched tasks are unaffected).

Issues

• resolves: Celery Beat auto-instrumentation: allow configuring monitor limits (max_runtime, checkin_margin, ...) #7838

Reminders

• Please add tests to validate your changes, and lint your code using uv run ruff.
• Add GH Issue ID & Linear ID (if applicable)
• PR title should use conventional commit style (feat:, fix:, ref:, meta:)
• For external contributors: CONTRIBUTING.md, Sentry SDK development docs, Discord community

[ericapisani]

Thanks for your contribution!

This is a great start, I've left some notes that we'll need to tackle before this gets merged in. Let me know if you have any questions about what I've left here.

[ericapisani]

Since we need a monitor_name provided as part of the configuration for any overrides to work, we can set the default value to an empty dictionary instead of None here.

Suggested change

	beat_task_monitor_config: "Optional[Dict[str, MonitorConfig]]" = None,
	beat_task_monitor_config: "Optional[Dict[str, MonitorConfig]]" = {},

This would allow us to update the (integration.beat_task_monitor_config or {}).get(monitor_name) conditional to either:

integration.beat_task_monitor_config.get(monitor_name)

or

getattr(integration.beat_task_monitor_config, monitor_name, None)

depending on your preference.

[ericapisani]

Let's remove this test

[ericapisani]

Within the SDK, we're trying to move towards testing functionality via the public APIs rather than the private methods and use mocks as sparingly as we can.

Instead of this approach, can we look to instead do something similar to the test_beat_task_crons_success in test_celery_beat_cron_monitoring.py (which creates a celery app and asserts that the envelopes contain the right data)?

[ericapisani]

We shouldn't support overriding the schedule (or its type) via this new property as there's validation that we do within the SDK to ensure the provided values are valid.

Instead, let's do the following:

1️⃣ Introduce a new type in _types.py to a subset of what's needed to support long-running tasks, etc. and update the relevant spots in this PR

MonitorConfigOverrides = TypedDict(
        "MonitorConfigOverrides",
        {
            "checkin_margin": int,
            "max_runtime": int,
            "failure_issue_threshold": int,
            "recovery_threshold": int,
            "owner": str,
        },
        total=False,
    )

2️⃣ Validate the overrides in the __init__ of the integration to ensure that everything's valid

In celery/beat.py:

_ALLOWED_MONITOR_CONFIG_OVERRIDE_KEYS = frozenset(
    {
        "checkin_margin",
        "max_runtime",
        "failure_issue_threshold",
        "recovery_threshold",
        "owner",
    }
)


def _validate_beat_task_monitor_config(
    beat_task_monitor_config: "Dict[str, MonitorConfigOverrides]",
) -> None:
    for task_name, overrides in beat_task_monitor_config.items():
        invalid = set(overrides) - _ALLOWED_MONITOR_CONFIG_OVERRIDE_KEYS
        if invalid:
            raise ValueError(
                f"Unsupported keys in beat_task_monitor_config for '{task_name}': "
                f"{sorted(invalid)}. Allowed keys: "
                f"{sorted(_ALLOWED_MONITOR_CONFIG_OVERRIDE_KEYS)}"
            )

and then invoke _validate_beat_task_monitor_config(beat_task_monitor_config) in the init code in __init__.py before assigning it to self.beat_task_monitor_config on CeleryIntegration.