From 664815d12d28517b0552c277f2ff28fa7e4903f8 Mon Sep 17 00:00:00 2001 From: Yucheng Zhu Date: Wed, 19 Aug 2026 18:03:42 -0700 Subject: [PATCH] fix(observability): stop skipped runs from masking a stalled job Two defects the review caught, both in the case this telemetry exists for. The last-run clock was refreshed on every event, including runs that never executed. A job repeatedly missed or skipped kept looking recently run, so the "time since last run" alert the docs recommend would stay quiet through exactly the outage it is meant to catch. Only a run that executed moves the clock now; the skip is still counted, just not as a run. Start times also leaked on a missed run. MAX_INSTANCES is emitted by the scheduler before it submits, so there is nothing to release, and the previous code treated MISSED the same way. MISSED comes from the executor after the submit (`apscheduler/executors/base.py`), so its start time was recorded and never freed, growing one entry per miss for the life of the process. The comment claiming neither follows a submission was wrong for MISSED. Also corrects the injected clock's annotation, which used `Final` in a parameter position where it is not valid, and adds the missing `Final` on the lock telemetry's logger lookup. --- litellm/integrations/prometheus.py | 3 +- .../common_utils/scheduled_job_metrics.py | 27 ++++++-- .../db_transaction_queue/pod_lock_manager.py | 2 +- .../test_prometheus_cronjob_lock_metrics.py | 40 ++++++++++++ .../test_scheduled_job_metrics.py | 65 +++++++++++++++++++ 5 files changed, 128 insertions(+), 9 deletions(-) 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) +