diff --git a/litellm/integrations/prometheus.py b/litellm/integrations/prometheus.py index 2cbdf36cfdd..8061835f205 100644 --- a/litellm/integrations/prometheus.py +++ b/litellm/integrations/prometheus.py @@ -3207,7 +3207,8 @@ class PrometheusLogger(CustomLogger): def record_scheduled_job_run(self, run: JobRun) -> None: """Publish one completed run of a scheduled background job.""" self.litellm_scheduled_job_runs_total.labels(job_name=run.job_name, result=run.result.value).inc() - self.litellm_scheduled_job_last_run_timestamp.labels(job_name=run.job_name).set(time.time()) + if run.did_execute: + self.litellm_scheduled_job_last_run_timestamp.labels(job_name=run.job_name).set(time.time()) if run.duration_seconds is not None: self.litellm_scheduled_job_duration_seconds.labels(job_name=run.job_name).observe(run.duration_seconds) if run.items_processed: diff --git a/litellm/proxy/common_utils/scheduled_job_metrics.py b/litellm/proxy/common_utils/scheduled_job_metrics.py index 147d5c52f6d..40c24e77ef8 100644 --- a/litellm/proxy/common_utils/scheduled_job_metrics.py +++ b/litellm/proxy/common_utils/scheduled_job_metrics.py @@ -17,6 +17,7 @@ from __future__ import annotations import re import time +from collections.abc import Callable from dataclasses import dataclass from enum import Enum from typing import TYPE_CHECKING, Final @@ -65,6 +66,16 @@ class JobRun: duration_seconds: float | None items_processed: int | None + @property + def did_execute(self) -> bool: + """Whether the job body actually ran. + + A skipped run must not refresh the last-run clock, or a job that never + executes keeps looking recently run and the "time since last run" alert + stays quiet through exactly the outage it exists to catch. + """ + return self.result in (JobResult.SUCCESS, JobResult.ERROR) + def _label_for(job_id: str) -> str: """The job id, unless APScheduler generated it.""" @@ -95,7 +106,7 @@ class ScheduledJobMetricsListener: on the job id alone would let those runs consume each other's start times. """ - def __init__(self, *, monotonic: Final = time.monotonic) -> None: + def __init__(self, *, monotonic: Callable[[], float] = time.monotonic) -> None: self._monotonic: Final = monotonic self._started_at: dict[str, float] = {} # mutable-ok: start times arrive one scheduler event at a time @@ -124,14 +135,16 @@ class ScheduledJobMetricsListener: self._started_at[self._key(event.job_id, scheduled)] = now return None - # Neither of these follows a submission of its own. MAX_INSTANCES in - # particular arrives while the previous run is still going, so popping - # before this point would steal that run's start time and drop the - # duration of exactly the overruns worth measuring. - if event.code == EVENT_JOB_MISSED: - return JobRun(job_name, JobResult.MISSED, None, None) + # MAX_INSTANCES is emitted by the scheduler before it submits, while the + # previous run is still going, so popping here would steal that run's + # start time and drop the duration of exactly the overruns worth + # measuring. MISSED comes from the executor after the submit, so its + # start time was recorded and has to be released or it leaks. if event.code == EVENT_JOB_MAX_INSTANCES: return JobRun(job_name, JobResult.MAX_INSTANCES, None, None) + if event.code == EVENT_JOB_MISSED: + self._started_at.pop(self._key(event.job_id, getattr(event, "scheduled_run_time", None)), None) + return JobRun(job_name, JobResult.MISSED, None, None) started_at: Final = self._started_at.pop( self._key(event.job_id, getattr(event, "scheduled_run_time", None)), None diff --git a/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py b/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py index 9df4da3e552..a4033cbec3c 100644 --- a/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py +++ b/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py @@ -26,7 +26,7 @@ def _record_lock_attempt(cronjob_id: str, result: LockAttemptResult) -> None: try: from litellm.integrations.prometheus import PrometheusLogger - logger = PrometheusLogger.get_instance() + logger: Final = PrometheusLogger.get_instance() if logger is not None: logger.record_cronjob_lock_attempt(cronjob_id, result) except Exception as e: # noqa: BLE001 # telemetry must not stop a job from running diff --git a/tests/test_litellm/integrations/test_prometheus_cronjob_lock_metrics.py b/tests/test_litellm/integrations/test_prometheus_cronjob_lock_metrics.py index ffc818b7d06..f5db2686f65 100644 --- a/tests/test_litellm/integrations/test_prometheus_cronjob_lock_metrics.py +++ b/tests/test_litellm/integrations/test_prometheus_cronjob_lock_metrics.py @@ -35,3 +35,43 @@ def test_the_lock_metric_documents_every_result_it_can_emit(monkeypatch): documented = {value.strip() for value in advertised.group(1).split(",")} assert documented == {result.value for result in LockAttemptResult} + +def _fresh_logger(monkeypatch): + from prometheus_client import REGISTRY + + monkeypatch.setattr(litellm, "callbacks", []) + monkeypatch.setattr(litellm, "success_callback", []) + for collector in list(REGISTRY._collector_to_names.keys()): + try: + REGISTRY.unregister(collector) + except Exception: + pass + return PrometheusLogger() + + +def test_only_a_run_that_executed_refreshes_the_last_run_clock(monkeypatch): + """The recommended alert is time since last run, so a skipped run bumping the + clock would keep a job that never executes looking healthy.""" + from prometheus_client import REGISTRY + + from litellm.proxy.common_utils.scheduled_job_metrics import JobResult, JobRun + + logger = _fresh_logger(monkeypatch) + + def clock(): + return REGISTRY.get_sample_value( + "litellm_scheduled_job_last_run_timestamp", {"job_name": "job"} + ) + + logger.record_scheduled_job_run(JobRun("job", JobResult.SUCCESS, 1.0, None)) + after_success = clock() + assert after_success is not None and after_success > 0 + + logger.record_scheduled_job_run(JobRun("job", JobResult.MISSED, None, None)) + logger.record_scheduled_job_run(JobRun("job", JobResult.MAX_INSTANCES, None, None)) + + assert clock() == after_success, "a skipped run must not move the last-run clock" + assert REGISTRY.get_sample_value( + "litellm_scheduled_job_runs_total", {"job_name": "job", "result": "missed"} + ) == 1.0, "the skip is still counted, just not as a run" + diff --git a/tests/test_litellm/proxy/common_utils/test_scheduled_job_metrics.py b/tests/test_litellm/proxy/common_utils/test_scheduled_job_metrics.py index ed77335ea54..8897368be8a 100644 --- a/tests/test_litellm/proxy/common_utils/test_scheduled_job_metrics.py +++ b/tests/test_litellm/proxy/common_utils/test_scheduled_job_metrics.py @@ -17,6 +17,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler sys.path.insert(0, os.path.abspath("../../../..")) from litellm.proxy.common_utils.scheduled_job_metrics import ( + JobRun, JobResult, ScheduledJobMetricsListener, ) @@ -337,3 +338,67 @@ def test_handle_swallows_a_failure_so_the_scheduler_keeps_running(): with patch.object(ScheduledJobMetricsListener, "_publish", side_effect=RuntimeError("metrics down")): listener.handle(event) # must not raise +def test_a_skipped_run_does_not_refresh_the_last_run_clock(): + """The recommended alert is time since last run. If a skipped run bumped the + clock, a job that never executes would look recently run and the alert would + stay quiet through exactly the outage this telemetry exists to catch.""" + executed = JobRun("job", JobResult.SUCCESS, 1.0, None) + failed = JobRun("job", JobResult.ERROR, 1.0, None) + missed = JobRun("job", JobResult.MISSED, None, None) + overrun = JobRun("job", JobResult.MAX_INSTANCES, None, None) + + assert executed.did_execute is True + assert failed.did_execute is True, "a job that ran and raised still ran" + assert missed.did_execute is False + assert overrun.did_execute is False + + +def test_a_missed_run_releases_the_start_time_it_was_given(): + """APScheduler emits MISSED from the executor, after the scheduler already + emitted SUBMITTED, so a start time was recorded for that run. Dropping the + event without releasing it leaks one entry per miss for the life of the + process.""" + import datetime + + from apscheduler.events import ( + EVENT_JOB_MISSED, + EVENT_JOB_SUBMITTED, + JobExecutionEvent, + JobSubmissionEvent, + ) + + when = datetime.datetime(2026, 1, 1, 0, 0, 0) + listener = ScheduledJobMetricsListener() + + listener._to_run(JobSubmissionEvent(EVENT_JOB_SUBMITTED, "slow_job", "default", [when])) + assert len(listener._started_at) == 1, "the submit records a start time" + + run = listener._to_run(JobExecutionEvent(EVENT_JOB_MISSED, "slow_job", "default", when)) + + assert run is not None and run.result is JobResult.MISSED + assert listener._started_at == {}, "the missed run must not leave its start time behind" + + +def test_an_overrun_keeps_the_running_jobs_start_time(): + """MAX_INSTANCES is emitted before the submit, while the previous run is + still going, so it must not consume that run's start time.""" + import datetime + + from apscheduler.events import ( + EVENT_JOB_EXECUTED, + EVENT_JOB_MAX_INSTANCES, + EVENT_JOB_SUBMITTED, + JobExecutionEvent, + JobSubmissionEvent, + ) + + when = datetime.datetime(2026, 1, 1, 0, 0, 0) + ticks = iter([100.0, 104.0]) + listener = ScheduledJobMetricsListener(monotonic=lambda: next(ticks)) + + listener._to_run(JobSubmissionEvent(EVENT_JOB_SUBMITTED, "slow_job", "default", [when])) + listener._to_run(JobExecutionEvent(EVENT_JOB_MAX_INSTANCES, "slow_job", "default", when)) + finished = listener._to_run(JobExecutionEvent(EVENT_JOB_EXECUTED, "slow_job", "default", when, retval=None)) + + assert finished is not None and finished.duration_seconds == pytest.approx(4.0) +