mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-15 04:19:08 +00:00
fix(caching): report Redis circuit-breaker metrics through the Prometheus services logger (#46484)
* fix(caching): collect the Redis circuit-breaker state gauge as one series per state across worker processes 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 <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(caching): register Redis circuit-breaker metrics only when LiteLLM prometheus logging is on The breaker collectors join the default registry once the prometheus or prometheus_system callback is configured or a Prometheus logger starts, with earlier breaker events tallied and seeded in, and litellm.disable_redis_circuit_breaker_metrics turns them off for apps that export their own metrics Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * refactor(caching): drop the disable_redis_circuit_breaker_metrics opt-out Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * refactor(caching): keep the Redis breaker metrics gate to a per-state count and a publish hook Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(caching): pin the Redis breaker metrics gate against an injected registry Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(caching): publish the breaker collectors before reading them in the existing metrics test Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(caching): report Redis circuit-breaker metrics through the Prometheus services logger Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(integrations): build the Redis breaker collectors with typed prometheus_client constructors Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu
GitHub
parent
83980c9d63
commit
f10134728d
7 changed files with 162 additions and 90 deletions
|
|
@ -152,6 +152,11 @@ class ServiceLogging(CustomLogger):
|
|||
except Exception as e:
|
||||
verbose_logger.exception("Error dispatching service event - %s", e)
|
||||
|
||||
def redis_breaker_hook(self, transitions_to: str | None, failure_class: str | None) -> None:
|
||||
prometheus_logger: Final[PrometheusServicesLogger | None] = getattr(self, "prometheusServicesLogger", None)
|
||||
if prometheus_logger is not None:
|
||||
prometheus_logger.redis_breaker_hook(transitions_to=transitions_to, failure_class=failure_class)
|
||||
|
||||
def service_success_hook(
|
||||
self,
|
||||
service: ServiceTypes,
|
||||
|
|
|
|||
|
|
@ -18,7 +18,9 @@ import json
|
|||
import logging
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Awaitable, Callable, Iterator, Sequence
|
||||
import weakref
|
||||
from collections import Counter
|
||||
from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence
|
||||
from contextvars import ContextVar
|
||||
from dataclasses import dataclass
|
||||
from datetime import timedelta
|
||||
|
|
@ -51,12 +53,12 @@ from .base_cache import BaseCache
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from opentelemetry.trace import Span as _Span
|
||||
from prometheus_client import Counter as _PromCounter
|
||||
from prometheus_client import Gauge as _PromGauge
|
||||
from redis.asyncio import Redis, RedisCluster
|
||||
from redis.asyncio.client import Pipeline
|
||||
from redis.asyncio.cluster import ClusterPipeline
|
||||
|
||||
from litellm._service_logger import ServiceLogging
|
||||
|
||||
pipeline = Pipeline
|
||||
cluster_pipeline = ClusterPipeline
|
||||
async_redis_client = Redis
|
||||
|
|
@ -243,7 +245,9 @@ class RedisCircuitBreaker:
|
|||
recovery_timeout: int,
|
||||
enabled: bool = True,
|
||||
timeout_min_duration: float = REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION,
|
||||
service_logger: "ServiceLogging | None" = None,
|
||||
) -> None:
|
||||
self._service_logger = service_logger
|
||||
self.failure_threshold = failure_threshold
|
||||
self.recovery_timeout = recovery_timeout
|
||||
self.enabled = enabled
|
||||
|
|
@ -255,7 +259,12 @@ class RedisCircuitBreaker:
|
|||
self._opened_at: float | None = None
|
||||
self._state = self.CLOSED
|
||||
self._generation = 0
|
||||
_breaker_metrics().record_state_change(None, self._state)
|
||||
_LIVE_BREAKERS.add(self)
|
||||
self._report(transitions_to=None, failure_class=None)
|
||||
|
||||
@property
|
||||
def state(self) -> str:
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def generation(self) -> int:
|
||||
|
|
@ -301,7 +310,7 @@ class RedisCircuitBreaker:
|
|||
self._timeout_count = 0
|
||||
self._timeout_streak_started_at = None
|
||||
self._opened_at = now
|
||||
_breaker_metrics().record_failure("timeout" if is_timeout else "connectivity")
|
||||
self._report(transitions_to=None, failure_class="timeout" if is_timeout else "connectivity")
|
||||
if self._should_open(now):
|
||||
if self._state != self.OPEN:
|
||||
verbose_logger.warning(
|
||||
|
|
@ -327,10 +336,13 @@ class RedisCircuitBreaker:
|
|||
def _set_state(self, state: str) -> None:
|
||||
if state == self._state:
|
||||
return
|
||||
_breaker_metrics().record_transition(state)
|
||||
_breaker_metrics().record_state_change(self._state, state)
|
||||
self._state = state
|
||||
self._generation += 1
|
||||
self._report(transitions_to=state, failure_class=None)
|
||||
|
||||
def _report(self, transitions_to: str | None, failure_class: str | None) -> None:
|
||||
if self._service_logger is not None:
|
||||
self._service_logger.redis_breaker_hook(transitions_to=transitions_to, failure_class=failure_class)
|
||||
|
||||
|
||||
_RedisCallResult = TypeVar("_RedisCallResult")
|
||||
|
|
@ -430,60 +442,6 @@ def is_redis_timeout_failure(exc: BaseException) -> bool:
|
|||
return any(isinstance(link, timeout_types) for link in _explicit_causes(exc))
|
||||
|
||||
|
||||
class _BreakerMetrics:
|
||||
"""Prometheus metrics for the Redis circuit breaker; no-ops when the client is absent.
|
||||
|
||||
Registered lazily on the default registry (which /metrics serves) via the module-level
|
||||
``_breaker_metrics`` singleton so repeated RedisCache construction never re-registers.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._state_gauge: _PromGauge | None = None
|
||||
self._transitions: _PromCounter | None = None
|
||||
self._failures: _PromCounter | None = None
|
||||
try:
|
||||
from prometheus_client import Counter as PromCounter
|
||||
from prometheus_client import Gauge
|
||||
except ImportError:
|
||||
return
|
||||
self._state_gauge = Gauge(
|
||||
"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",
|
||||
"Redis circuit breaker state transitions",
|
||||
labelnames=("state",),
|
||||
)
|
||||
self._failures = PromCounter(
|
||||
"litellm_redis_circuit_breaker_failures",
|
||||
"Redis health failures counted by the circuit breaker",
|
||||
labelnames=("failure_class",),
|
||||
)
|
||||
|
||||
def record_state_change(self, old_state: str | None, new_state: str) -> None:
|
||||
if self._state_gauge is None:
|
||||
return
|
||||
if old_state is not None:
|
||||
self._state_gauge.labels(old_state).dec()
|
||||
self._state_gauge.labels(new_state).inc()
|
||||
|
||||
def record_transition(self, state: str) -> None:
|
||||
if self._transitions is not None:
|
||||
self._transitions.labels(state).inc()
|
||||
|
||||
def record_failure(self, failure_class: str) -> None:
|
||||
if self._failures is not None:
|
||||
self._failures.labels(failure_class).inc()
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=1)
|
||||
def _breaker_metrics() -> _BreakerMetrics:
|
||||
return _BreakerMetrics()
|
||||
|
||||
|
||||
def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseException) -> None:
|
||||
"""Record a Redis failure that the calling method is about to swallow.
|
||||
|
||||
|
|
@ -499,6 +457,14 @@ def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseExcep
|
|||
_swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1)
|
||||
|
||||
|
||||
_LIVE_BREAKERS: Final[weakref.WeakSet[RedisCircuitBreaker]] = weakref.WeakSet()
|
||||
|
||||
|
||||
def redis_breaker_state_counts() -> Mapping[str, int]:
|
||||
"""How many live Redis circuit breakers sit in each state, for the Prometheus services logger."""
|
||||
return MappingProxyType(dict(Counter(breaker.state for breaker in _LIVE_BREAKERS)))
|
||||
|
||||
|
||||
class RedisCircuitBreakerOpenError(Exception):
|
||||
pass
|
||||
|
||||
|
|
@ -685,12 +651,9 @@ class RedisCache(BaseCache):
|
|||
redis_kwargs["socket_timeout"] = socket_timeout
|
||||
|
||||
### HEALTH MONITORING OBJECT ###
|
||||
if kwargs.get("service_logger_obj", None) is not None and isinstance(
|
||||
kwargs["service_logger_obj"], ServiceLogging
|
||||
):
|
||||
self.service_logger_obj: ServiceLogging = ServiceLogging()
|
||||
if isinstance(kwargs.get("service_logger_obj"), ServiceLogging):
|
||||
self.service_logger_obj = kwargs.pop("service_logger_obj")
|
||||
else:
|
||||
self.service_logger_obj = ServiceLogging()
|
||||
|
||||
redis_kwargs.update(kwargs)
|
||||
self.redis_client = get_redis_client(**redis_kwargs)
|
||||
|
|
@ -717,6 +680,7 @@ class RedisCache(BaseCache):
|
|||
failure_threshold=REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD,
|
||||
recovery_timeout=REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT,
|
||||
enabled=REDIS_CIRCUIT_BREAKER_ENABLED,
|
||||
service_logger=self.service_logger_obj,
|
||||
)
|
||||
|
||||
self._setup_health_pings()
|
||||
|
|
|
|||
|
|
@ -3,7 +3,9 @@
|
|||
# On success + failure, log events to Prometheus for litellm / adjacent services (litellm, redis, postgres, llm api providers)
|
||||
|
||||
|
||||
from typing import Final
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
import litellm
|
||||
from litellm._logging import print_verbose, verbose_logger
|
||||
|
|
@ -17,6 +19,42 @@ from litellm.types.services import (
|
|||
|
||||
FAILED_REQUESTS_LABELS: Final = ["error_class", "function_name"]
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from prometheus_client import Counter as PrometheusCounter
|
||||
from prometheus_client import Gauge
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _RedisBreakerCollectors:
|
||||
state: "Gauge"
|
||||
transitions: "PrometheusCounter"
|
||||
failures: "PrometheusCounter"
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _redis_breaker_collectors() -> _RedisBreakerCollectors:
|
||||
from prometheus_client import Counter as PrometheusCounter
|
||||
from prometheus_client import Gauge
|
||||
|
||||
return _RedisBreakerCollectors(
|
||||
state=Gauge(
|
||||
"litellm_redis_circuit_breaker_state",
|
||||
"Number of Redis circuit breakers currently in each state",
|
||||
labelnames=("state",),
|
||||
multiprocess_mode="livesum",
|
||||
),
|
||||
transitions=PrometheusCounter(
|
||||
"litellm_redis_circuit_breaker_transitions",
|
||||
"Redis circuit breaker state transitions",
|
||||
labelnames=("state",),
|
||||
),
|
||||
failures=PrometheusCounter(
|
||||
"litellm_redis_circuit_breaker_failures",
|
||||
"Redis health failures counted by the circuit breaker",
|
||||
labelnames=("failure_class",),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class PrometheusServicesLogger:
|
||||
# Class variables or attributes
|
||||
|
|
@ -41,6 +79,11 @@ class PrometheusServicesLogger:
|
|||
self.Counter = Counter
|
||||
self.Gauge = Gauge
|
||||
self.REGISTRY = REGISTRY
|
||||
breaker_collectors: Final = _redis_breaker_collectors()
|
||||
self.redis_breaker_state = breaker_collectors.state
|
||||
self.redis_breaker_transitions = breaker_collectors.transitions
|
||||
self.redis_breaker_failures = breaker_collectors.failures
|
||||
self._refresh_redis_breaker_gauge()
|
||||
|
||||
verbose_logger.debug("in init prometheus services metrics")
|
||||
|
||||
|
|
@ -98,6 +141,20 @@ class PrometheusServicesLogger:
|
|||
return DEFAULT_METRICS
|
||||
return metrics
|
||||
|
||||
def _refresh_redis_breaker_gauge(self) -> None:
|
||||
from litellm.caching.redis_cache import RedisCircuitBreaker, redis_breaker_state_counts
|
||||
|
||||
counts: Final = redis_breaker_state_counts()
|
||||
for state in (RedisCircuitBreaker.CLOSED, RedisCircuitBreaker.OPEN, RedisCircuitBreaker.HALF_OPEN):
|
||||
self.redis_breaker_state.labels(state).set(counts.get(state, 0))
|
||||
|
||||
def redis_breaker_hook(self, transitions_to: str | None, failure_class: str | None) -> None:
|
||||
self._refresh_redis_breaker_gauge()
|
||||
if transitions_to is not None:
|
||||
self.redis_breaker_transitions.labels(transitions_to).inc()
|
||||
if failure_class is not None:
|
||||
self.redis_breaker_failures.labels(failure_class).inc()
|
||||
|
||||
def is_metric_registered(self, metric_name) -> bool:
|
||||
# Use _names_to_collectors (O(1)) instead of REGISTRY.collect() (O(n)) to avoid
|
||||
# perf regression when a new Router is created per request (e.g. router_settings in DB).
|
||||
|
|
|
|||
|
|
@ -11,10 +11,12 @@ import sys
|
|||
from collections.abc import Iterator, Mapping
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
from typing import Final, Literal, TypeAlias
|
||||
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
Scenario: TypeAlias = Literal["callbacks_before_cache", "callbacks_after_cache", "no_callbacks"]
|
||||
|
||||
STATE_GAUGE: Final = "litellm_redis_circuit_breaker_state"
|
||||
WORKER_TIMEOUT_SECONDS: Final = 120
|
||||
|
||||
|
|
@ -29,9 +31,11 @@ class WorkerReport:
|
|||
_REPORT: Final = TypeAdapter(WorkerReport)
|
||||
|
||||
|
||||
def run_breaker_worker(api_base: str, prom_dir: Path, environment: Mapping[str, str] = os.environ) -> WorkerReport:
|
||||
def run_breaker_worker(
|
||||
scenario: Scenario, 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"]),
|
||||
(sys.executable, "-I", __file__, scenario, api_base, environment["REDIS_HOST"], environment["REDIS_PORT"]),
|
||||
env={**environment, "PROMETHEUS_MULTIPROC_DIR": str(prom_dir)},
|
||||
capture_output=True,
|
||||
text=True,
|
||||
|
|
@ -56,11 +60,15 @@ def _series(families: Iterator[object]) -> Iterator[tuple[tuple[tuple[str, str],
|
|||
yield tuple(sorted(sample.labels.items())), float(sample.value)
|
||||
|
||||
|
||||
def main(api_base: str, redis_host: str, redis_port: str) -> None:
|
||||
def main(scenario: str, api_base: str, redis_host: str, redis_port: str) -> None:
|
||||
import litellm
|
||||
from litellm.caching.caching import Cache
|
||||
|
||||
if scenario == "callbacks_before_cache":
|
||||
litellm.service_callback = ["prometheus_system"] # test-quality-ok: this worker process exits right after the report
|
||||
litellm.cache = Cache(type="redis", host=redis_host, port=redis_port) # test-quality-ok: throwaway worker process
|
||||
if scenario == "callbacks_after_cache":
|
||||
litellm.service_callback = ["prometheus_system"] # test-quality-ok: throwaway worker process
|
||||
response: Final = litellm.completion(
|
||||
model="openai/gpt-5.6",
|
||||
api_base=api_base,
|
||||
|
|
@ -76,4 +84,4 @@ def main(api_base: str, redis_host: str, redis_port: str) -> None:
|
|||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main(*sys.argv[1:4])
|
||||
main(*sys.argv[1:5])
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""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."""
|
||||
circuit-breaker state gauge must collect as one series per state across every worker, and it must stay off
|
||||
the registry entirely for an app that never turned on LiteLLM's own Prometheus logging."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -24,6 +25,12 @@ _CHAT_RESPONSE: Final = json.dumps(
|
|||
}
|
||||
).encode()
|
||||
_CLOSED: Final = (("state", "closed"),)
|
||||
_HALF_OPEN: Final = (("state", "half_open"),)
|
||||
_OPEN: Final = (("state", "open"),)
|
||||
|
||||
|
||||
def _all_closed(breakers: float) -> tuple[tuple[tuple[tuple[str, str], ...], float], ...]:
|
||||
return ((_CLOSED, breakers), (_HALF_OPEN, 0.0), (_OPEN, 0.0))
|
||||
|
||||
|
||||
def _reply(request: Request) -> Reply:
|
||||
|
|
@ -52,6 +59,22 @@ def _worked(report: WorkerReport) -> WorkerReport:
|
|||
|
||||
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))
|
||||
reports: Final = tuple(
|
||||
_worked(run_breaker_worker("callbacks_before_cache", 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
|
||||
assert _collected_state_series(tmp_path) == _all_closed(2.0), reports
|
||||
|
||||
|
||||
def test_prometheus_logging_turned_on_after_the_cache_still_reports_the_closed_breaker(tmp_path: Path) -> None:
|
||||
with wire_server(_reply) as wire:
|
||||
report: Final = _worked(run_breaker_worker("callbacks_after_cache", wire.url, tmp_path))
|
||||
assert report.state_series == _all_closed(1.0), report
|
||||
assert _collected_state_series(tmp_path) == _all_closed(1.0), report
|
||||
|
||||
|
||||
def test_app_without_litellm_prometheus_logging_gets_no_breaker_series(tmp_path: Path) -> None:
|
||||
with wire_server(_reply) as wire:
|
||||
report: Final = _worked(run_breaker_worker("no_callbacks", wire.url, tmp_path))
|
||||
assert report.state_series == (), report
|
||||
assert _collected_state_series(tmp_path) == (), report
|
||||
|
|
|
|||
|
|
@ -971,18 +971,17 @@ async def test_hard_failure_resets_timeout_streak_so_a_later_burst_must_earn_its
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_breaker_metrics_track_state_and_failure_class():
|
||||
async def test_breaker_metrics_track_state_and_failure_class(monkeypatch):
|
||||
"""Breaker accounting must be observable: failure class, transitions, and state."""
|
||||
from prometheus_client import REGISTRY
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
from litellm.caching.redis_cache import RedisCircuitBreaker, _breaker_metrics, is_redis_timeout_failure
|
||||
from litellm._service_logger import ServiceLogging
|
||||
from litellm.caching.redis_cache import RedisCircuitBreaker, is_redis_timeout_failure
|
||||
|
||||
metrics: Final = _breaker_metrics()
|
||||
for collector in (metrics._state_gauge, metrics._transitions, metrics._failures):
|
||||
if collector is not None and collector not in REGISTRY._collector_to_names:
|
||||
REGISTRY.register(collector)
|
||||
monkeypatch.setattr(litellm, "service_callback", ["prometheus_system"])
|
||||
service_logger: Final = ServiceLogging()
|
||||
|
||||
def sample(name, labels=None):
|
||||
return REGISTRY.get_sample_value(name, labels) or 0.0
|
||||
|
|
@ -993,7 +992,9 @@ async def test_breaker_metrics_track_state_and_failure_class():
|
|||
open_gauge_before = sample("litellm_redis_circuit_breaker_state", {"state": "open"})
|
||||
closed_gauge_before = sample("litellm_redis_circuit_breaker_state", {"state": "closed"})
|
||||
|
||||
breaker = RedisCircuitBreaker(failure_threshold=2, recovery_timeout=60, timeout_min_duration=5.0)
|
||||
breaker = RedisCircuitBreaker(
|
||||
failure_threshold=2, recovery_timeout=60, timeout_min_duration=5.0, service_logger=service_logger
|
||||
)
|
||||
breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("t")))
|
||||
breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused")))
|
||||
breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused")))
|
||||
|
|
@ -1011,6 +1012,24 @@ async def test_breaker_metrics_track_state_and_failure_class():
|
|||
assert sample("litellm_redis_circuit_breaker_state", {"state": "closed"}) == closed_gauge_before + 1
|
||||
|
||||
|
||||
def test_prometheus_services_logger_starts_from_the_breakers_already_built(monkeypatch):
|
||||
"""A breaker built before prometheus_system is on shows up in the state gauge once the services logger starts."""
|
||||
from prometheus_client import REGISTRY
|
||||
|
||||
from litellm._service_logger import ServiceLogging
|
||||
from litellm.caching.redis_cache import RedisCircuitBreaker, redis_breaker_state_counts
|
||||
|
||||
monkeypatch.setattr(litellm, "service_callback", [])
|
||||
early: Final = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, service_logger=ServiceLogging())
|
||||
assert early.is_open() is False
|
||||
|
||||
monkeypatch.setattr(litellm, "service_callback", ["prometheus_system"])
|
||||
ServiceLogging()
|
||||
assert REGISTRY.get_sample_value("litellm_redis_circuit_breaker_state", {"state": "closed"}) == (
|
||||
redis_breaker_state_counts()["closed"]
|
||||
)
|
||||
|
||||
|
||||
def test_sync_guard_counts_a_timeout_as_a_timeout():
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@ from prometheus_client import REGISTRY, Gauge
|
|||
from prometheus_client.registry import Collector
|
||||
|
||||
import litellm
|
||||
from litellm.caching.redis_cache import _breaker_metrics
|
||||
from litellm.integrations.prometheus import PrometheusLogger
|
||||
from litellm.integrations.prometheus_services import PrometheusServicesLogger
|
||||
from litellm.proxy.db.db_transaction_queue.spend_log_cleanup_metrics import SpendLogCleanupMetrics
|
||||
|
|
@ -62,10 +61,7 @@ def _lazy_owner_collectors() -> tuple[Collector, ...]:
|
|||
assert SpendLogCleanupMetrics.runs is not None
|
||||
in_flight: Final = InFlightRequestsMiddleware._get_gauge()
|
||||
assert in_flight is not None
|
||||
breaker: Final = _breaker_metrics()
|
||||
assert breaker._state_gauge is not None
|
||||
assert breaker._transitions is not None
|
||||
assert breaker._failures is not None
|
||||
services: Final = PrometheusServicesLogger()
|
||||
return (
|
||||
SpendLogCleanupMetrics.rows_deleted,
|
||||
SpendLogCleanupMetrics.batch_duration,
|
||||
|
|
@ -73,9 +69,9 @@ def _lazy_owner_collectors() -> tuple[Collector, ...]:
|
|||
SpendLogCleanupMetrics.batch_failures,
|
||||
SpendLogCleanupMetrics.runs,
|
||||
in_flight,
|
||||
breaker._state_gauge,
|
||||
breaker._transitions,
|
||||
breaker._failures,
|
||||
services.redis_breaker_state,
|
||||
services.redis_breaker_transitions,
|
||||
services.redis_breaker_failures,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue