diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 3a847044e3b..0e34709d128 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -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", diff --git a/tests/integration/_support/redis_breaker_worker.py b/tests/integration/_support/redis_breaker_worker.py new file mode 100644 index 00000000000..f551baacf24 --- /dev/null +++ b/tests/integration/_support/redis_breaker_worker.py @@ -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]) diff --git a/tests/integration/sdk/test_redis_breaker_metrics_multiprocess_sdk.py b/tests/integration/sdk/test_redis_breaker_metrics_multiprocess_sdk.py new file mode 100644 index 00000000000..d1582dd59b9 --- /dev/null +++ b/tests/integration/sdk/test_redis_breaker_metrics_multiprocess_sdk.py @@ -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