fix(caching): collect the Redis circuit-breaker state gauge as one series per state across worker processes (#46357)

The gauge declares multiprocess_mode=livesum, so PROMETHEUS_MULTIPROC_DIR apps get one series per state summed over
the live workers instead of one series per worker pid

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu 2026-10-14 11:45:49 -07:00 • committed by GitHub
parent 457b5d26c7
commit f0fedda722
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 137 additions and 0 deletions

View file

@ -450,6 +450,7 @@ class _BreakerMetrics:
"litellm_redis_circuit_breaker_state",
"Number of Redis circuit breakers currently in each state",
labelnames=("state",),
multiprocess_mode="livesum",
)
self._transitions = PromCounter(
"litellm_redis_circuit_breaker_transitions",

View file

@ -0,0 +1,79 @@
"""One SDK worker process for the Redis circuit-breaker metric cells: a Redis-cached completion against an owned
peer, then a report of the breaker state gauge on that process's own default registry. Run as a script by
``run_breaker_worker`` so ``PROMETHEUS_MULTIPROC_DIR`` is in place before ``prometheus_client`` is imported."""
from __future__ import annotations
import json
import os
import subprocess
import sys
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Final
from pydantic import TypeAdapter
STATE_GAUGE: Final = "litellm_redis_circuit_breaker_state"
WORKER_TIMEOUT_SECONDS: Final = 120
@dataclass(frozen=True, slots=True)
class WorkerReport:
pid: int
answer: str
state_series: tuple[tuple[tuple[tuple[str, str], ...], float], ...]
_REPORT: Final = TypeAdapter(WorkerReport)
def run_breaker_worker(api_base: str, prom_dir: Path, environment: Mapping[str, str] = os.environ) -> WorkerReport:
completed: Final = subprocess.run(
(sys.executable, "-I", __file__, api_base, environment["REDIS_HOST"], environment["REDIS_PORT"]),
env={**environment, "PROMETHEUS_MULTIPROC_DIR": str(prom_dir)},
capture_output=True,
text=True,
timeout=WORKER_TIMEOUT_SECONDS,
check=False,
)
assert completed.returncode == 0, completed.stderr[-3000:]
return _REPORT.validate_json(completed.stdout.splitlines()[-1])
def _state_series_on_default_registry() -> tuple[tuple[tuple[tuple[str, str], ...], float], ...]:
from prometheus_client import REGISTRY
return tuple(sorted(_series(REGISTRY.collect())))
def _series(families: Iterator[object]) -> Iterator[tuple[tuple[tuple[str, str], ...], float]]:
for family in families:
if getattr(family, "name", None) != STATE_GAUGE:
continue
for sample in getattr(family, "samples", ()):
yield tuple(sorted(sample.labels.items())), float(sample.value)
def main(api_base: str, redis_host: str, redis_port: str) -> None:
import litellm
from litellm.caching.caching import Cache
litellm.cache = Cache(type="redis", host=redis_host, port=redis_port) # test-quality-ok: throwaway worker process
response: Final = litellm.completion(
model="openai/gpt-5.6",
api_base=api_base,
api_key="synthetic-breaker-worker-key",
messages=[{"role": "user", "content": f"breaker worker {os.getpid()}"}],
)
report: Final = WorkerReport(
pid=os.getpid(),
answer=str(response.choices[0].message.content),
state_series=_state_series_on_default_registry(),
)
sys.stdout.write(json.dumps(_REPORT.dump_python(report)) + "\n")
if __name__ == "__main__":
main(*sys.argv[1:4])

View file

@ -0,0 +1,57 @@
"""An SDK app under a prefork worker pool (Celery, gunicorn) with ``PROMETHEUS_MULTIPROC_DIR`` set: the Redis
circuit-breaker state gauge must collect as one series per state across every worker, never one per worker pid."""
from __future__ import annotations
import json
from collections.abc import Iterator
from pathlib import Path
from typing import Final
from integration._support.redis_breaker_worker import STATE_GAUGE, WorkerReport, run_breaker_worker
from integration._support.wire import Reply, Request, wire_server
from prometheus_client import CollectorRegistry, multiprocess
_ANSWER: Final = "answered through the redis cache"
_CHAT_RESPONSE: Final = json.dumps(
{
"id": "chatcmpl-redis-breaker-worker",
"object": "chat.completion",
"created": 1700000000,
"model": "gpt-5.6",
"choices": [{"index": 0, "message": {"role": "assistant", "content": _ANSWER}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 5, "completion_tokens": 4, "total_tokens": 9},
}
).encode()
_CLOSED: Final = (("state", "closed"),)
def _reply(request: Request) -> Reply:
assert request.method == "POST", request.method
return Reply(body=_CHAT_RESPONSE)
def _state_samples(prom_dir: Path) -> Iterator[tuple[tuple[tuple[str, str], ...], float]]:
registry: Final = CollectorRegistry()
multiprocess.MultiProcessCollector(registry, path=str(prom_dir))
for family in registry.collect():
if family.name != STATE_GAUGE:
continue
for sample in family.samples:
yield tuple(sorted(sample.labels.items())), sample.value
def _collected_state_series(prom_dir: Path) -> tuple[tuple[tuple[tuple[str, str], ...], float], ...]:
return tuple(sorted(_state_samples(prom_dir)))
def _worked(report: WorkerReport) -> WorkerReport:
assert report.answer == _ANSWER, report
return report
def test_two_worker_processes_collect_as_one_state_series_per_state(tmp_path: Path) -> None:
with wire_server(_reply) as wire:
reports: Final = tuple(_worked(run_breaker_worker(wire.url, tmp_path)) for _ in range(2))
assert len({report.pid for report in reports}) == 2, reports
assert _collected_state_series(tmp_path) == ((_CLOSED, 2.0),), reports