From 410daafa11ab375af217ee23973983a3faa70060 Mon Sep 17 00:00:00 2001 From: yassin Date: Tue, 13 Oct 2026 21:03:54 +0000 Subject: [PATCH] feat(control_plane): fan-out primitive and global usage and fleet aggregates Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/proxy/control_plane/app.py | 7 +- .../proxy/control_plane/common/exceptions.py | 9 + litellm/proxy/control_plane/common/fanout.py | 88 ++++++++++ litellm/proxy/control_plane/container.py | 3 + .../control_plane/dataplanes/dependencies.py | 19 +- litellm/proxy/control_plane/fleet/__init__.py | 0 litellm/proxy/control_plane/fleet/router.py | 74 ++++++++ litellm/proxy/control_plane/fleet/service.py | 124 +++++++++++++ litellm/proxy/control_plane/usage/__init__.py | 0 litellm/proxy/control_plane/usage/router.py | 154 ++++++++++++++++ litellm/proxy/control_plane/usage/service.py | 164 ++++++++++++++++++ .../types/proxy/control_plane/dataplanes.py | 9 + litellm/types/proxy/control_plane/fleet.py | 55 ++++++ litellm/types/proxy/control_plane/usage.py | 72 ++++++++ .../proxy/control_plane/common/test_fanout.py | 85 +++++++++ .../proxy/control_plane/fleet/__init__.py | 0 .../proxy/control_plane/fleet/test_router.py | 115 ++++++++++++ tests/unit/proxy/control_plane/support.py | 85 +++++++++ .../proxy/control_plane/usage/__init__.py | 0 .../proxy/control_plane/usage/test_router.py | 146 ++++++++++++++++ .../proxy/control_plane/usage/test_service.py | 111 ++++++++++++ 21 files changed, 1317 insertions(+), 3 deletions(-) create mode 100644 litellm/proxy/control_plane/common/fanout.py create mode 100644 litellm/proxy/control_plane/fleet/__init__.py create mode 100644 litellm/proxy/control_plane/fleet/router.py create mode 100644 litellm/proxy/control_plane/fleet/service.py create mode 100644 litellm/proxy/control_plane/usage/__init__.py create mode 100644 litellm/proxy/control_plane/usage/router.py create mode 100644 litellm/proxy/control_plane/usage/service.py create mode 100644 litellm/types/proxy/control_plane/fleet.py create mode 100644 litellm/types/proxy/control_plane/usage.py create mode 100644 tests/unit/proxy/control_plane/common/test_fanout.py create mode 100644 tests/unit/proxy/control_plane/fleet/__init__.py create mode 100644 tests/unit/proxy/control_plane/fleet/test_router.py create mode 100644 tests/unit/proxy/control_plane/support.py create mode 100644 tests/unit/proxy/control_plane/usage/__init__.py create mode 100644 tests/unit/proxy/control_plane/usage/test_router.py create mode 100644 tests/unit/proxy/control_plane/usage/test_service.py diff --git a/litellm/proxy/control_plane/app.py b/litellm/proxy/control_plane/app.py index 2fab0c49557..3896fa680a4 100644 --- a/litellm/proxy/control_plane/app.py +++ b/litellm/proxy/control_plane/app.py @@ -9,9 +9,11 @@ from litellm.proxy.control_plane.auth.router import get_login_service from litellm.proxy.control_plane.auth.router import router as auth_router from litellm.proxy.control_plane.common.exceptions import install_exception_handlers from litellm.proxy.control_plane.container import Container -from litellm.proxy.control_plane.dataplanes.dependencies import get_dataplane_client, get_registry +from litellm.proxy.control_plane.dataplanes.dependencies import get_dataplane_client, get_fanout, get_registry from litellm.proxy.control_plane.dataplanes.router import router as dataplanes_router +from litellm.proxy.control_plane.fleet.router import router as fleet_router from litellm.proxy.control_plane.health.router import router as health_router +from litellm.proxy.control_plane.usage.router import router as usage_router TITLE: Final = "LiteLLM Control Plane" @@ -40,8 +42,11 @@ def create_app(container: Container, license_check: License) -> FastAPI: app.dependency_overrides[get_login_service] = lambda: container.login_service app.dependency_overrides[get_registry] = lambda: container.registry app.dependency_overrides[get_dataplane_client] = lambda: container.dataplane_client + app.dependency_overrides[get_fanout] = lambda: container.fanout install_exception_handlers(app) app.include_router(health_router) app.include_router(auth_router, prefix=container.settings.api_prefix) app.include_router(dataplanes_router, prefix=container.settings.api_prefix) + app.include_router(usage_router, prefix=container.settings.api_prefix) + app.include_router(fleet_router, prefix=container.settings.api_prefix) return app diff --git a/litellm/proxy/control_plane/common/exceptions.py b/litellm/proxy/control_plane/common/exceptions.py index 0c6dd7486a1..8f64ac15276 100644 --- a/litellm/proxy/control_plane/common/exceptions.py +++ b/litellm/proxy/control_plane/common/exceptions.py @@ -1,5 +1,6 @@ """RFC 9457 problem documents for the control plane, same contract as /management/v1.""" +from http import HTTPStatus from typing import Final from fastapi import FastAPI, Request @@ -23,6 +24,10 @@ def problem(slug: str, title: str, status: int, detail: str) -> ProblemDetail: return ProblemDetail(type=f"{PROBLEM_TYPE_BASE}{slug}", title=title, status=status, detail=detail) +def bad_request(detail: str) -> ProblemDetail: + return problem("bad-request", "Bad Request", HTTPStatus.BAD_REQUEST, detail) + + def unauthorized(detail: str) -> ProblemDetail: return problem("unauthorized", "Unauthorized", 401, detail) @@ -43,6 +48,10 @@ def bad_gateway(detail: str) -> ProblemDetail: return problem("bad-gateway", "Bad Gateway", 502, detail) +def service_unavailable(detail: str) -> ProblemDetail: + return problem("service-unavailable", "Service Unavailable", 503, detail) + + class ControlPlaneException(ManagementProblem): """A problem raised from a dependency or handler, rendered by the app-level handler below.""" diff --git a/litellm/proxy/control_plane/common/fanout.py b/litellm/proxy/control_plane/common/fanout.py new file mode 100644 index 00000000000..762f11b5f97 --- /dev/null +++ b/litellm/proxy/control_plane/common/fanout.py @@ -0,0 +1,88 @@ +"""Bounded scatter/gather over dataplanes. A dataplane that fails is a Missed value in the result, never an exception.""" + +import asyncio +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Final, Generic, TypeVar + +from pydantic import TypeAdapter, ValidationError + +from litellm.proxy.control_plane.dataplanes.client import CallResult, DataplaneReply, DataplaneUnreachable +from litellm.proxy.control_plane.dataplanes.credentials import CredentialUnavailable +from litellm.proxy.control_plane.dataplanes.registry import Dataplane + +T: Final = TypeVar("T") +M: Final = TypeVar("M") + + +@dataclass(frozen=True, slots=True) +class Failure: + reason: str + + +@dataclass(frozen=True, slots=True) +class Missed: + dataplane: Dataplane + reason: str + + +@dataclass(frozen=True, slots=True) +class Reached(Generic[T]): + dataplane: Dataplane + value: T + + +@dataclass(frozen=True, slots=True) +class Gathered(Generic[T]): + reached: tuple[Reached[T], ...] + missed: tuple[Missed, ...] + + @property + def partial(self) -> bool: + return bool(self.missed) + + @property + def nothing_answered(self) -> bool: + return bool(self.missed) and not self.reached + + +@dataclass(frozen=True, slots=True) +class Fanout: + concurrency: int + + async def gather( + self, dataplanes: tuple[Dataplane, ...], fetch: Callable[[Dataplane], Awaitable[T | Failure]] + ) -> Gathered[T]: + gate: Final = asyncio.Semaphore(self.concurrency) + + async def bounded(dataplane: Dataplane) -> Reached[T] | Missed: + async with gate: + outcome: Final = await fetch(dataplane) + if isinstance(outcome, Failure): + return Missed(dataplane, outcome.reason) + return Reached(dataplane, outcome) + + outcomes: Final = await asyncio.gather(*(bounded(dataplane) for dataplane in dataplanes)) + return Gathered( + reached=tuple(outcome for outcome in outcomes if isinstance(outcome, Reached)), + missed=tuple(outcome for outcome in outcomes if isinstance(outcome, Missed)), + ) + + +def parse_reply(result: CallResult, shape: TypeAdapter[M]) -> M | Failure: + match result: + case DataplaneReply(status=status, body=body) if status < 400: + try: + return shape.validate_python(body) + except ValidationError as error: + return Failure(f"unexpected response shape: {error.error_count()} validation error(s)") + case DataplaneReply(status=status, credential=credential): + return Failure(f"HTTP {status} with the {credential} credential") + case DataplaneUnreachable(detail=detail, attempts=attempts): + return Failure(f"unreachable after {attempts} attempt(s): {detail}") + case CredentialUnavailable(reason=reason): + return Failure(f"no usable credential: {reason}") + + +def describe_missed(missed: tuple[Missed, ...]) -> str: + return ", ".join(f"{miss.dataplane.id}: {miss.reason}" for miss in missed) diff --git a/litellm/proxy/control_plane/container.py b/litellm/proxy/control_plane/container.py index c62836704e3..5652321cd1f 100644 --- a/litellm/proxy/control_plane/container.py +++ b/litellm/proxy/control_plane/container.py @@ -24,6 +24,7 @@ from litellm.proxy.control_plane.auth.rbac import ( build_policy, parse_role_definition, ) +from litellm.proxy.control_plane.common.fanout import Fanout from litellm.proxy.control_plane.dataplanes.client import DataplaneClient, HttpTransport, RetryPolicy from litellm.proxy.control_plane.dataplanes.credentials import TokenExchanger, http_token_exchanger from litellm.proxy.control_plane.dataplanes.registry import DataplaneRegistry, build_registry @@ -39,6 +40,7 @@ class Container: login_service: LoginService registry: DataplaneRegistry dataplane_client: DataplaneClient + fanout: Fanout class InvalidRolePolicy(ValueError): @@ -141,4 +143,5 @@ def build_container( timeout_seconds=settings.dataplane_timeout_seconds, retry=RetryPolicy(attempts=settings.dataplane_retry_attempts), ), + fanout=Fanout(concurrency=settings.fanout_concurrency), ) diff --git a/litellm/proxy/control_plane/dataplanes/dependencies.py b/litellm/proxy/control_plane/dataplanes/dependencies.py index c9ea7441c37..79938b85701 100644 --- a/litellm/proxy/control_plane/dataplanes/dependencies.py +++ b/litellm/proxy/control_plane/dataplanes/dependencies.py @@ -2,15 +2,16 @@ from typing import Annotated, Final -from fastapi import Depends +from fastapi import Depends, Query from fastapi.security import HTTPAuthorizationCredentials from litellm.proxy.control_plane.auth.dependencies import bearer_scheme, get_principal from litellm.proxy.control_plane.auth.rbac import Principal from litellm.proxy.control_plane.common.exceptions import ControlPlaneException, forbidden, not_found, unauthorized +from litellm.proxy.control_plane.common.fanout import Fanout from litellm.proxy.control_plane.dataplanes.client import DataplaneClient from litellm.proxy.control_plane.dataplanes.credentials import Caller -from litellm.proxy.control_plane.dataplanes.registry import Dataplane, DataplaneRegistry +from litellm.proxy.control_plane.dataplanes.registry import Dataplane, DataplaneRegistry, visible_to def get_registry() -> DataplaneRegistry: @@ -21,6 +22,10 @@ def get_dataplane_client() -> DataplaneClient: raise RuntimeError("get_dataplane_client is not bound, build the app with create_app") +def get_fanout() -> Fanout: + raise RuntimeError("get_fanout is not bound, build the app with create_app") + + async def get_caller( credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)], principal: Annotated[Principal, Depends(get_principal)], @@ -41,3 +46,13 @@ async def dataplane_access( if not principal.dataplanes.allows(dataplane_id): raise ControlPlaneException(forbidden(f"Not bound to dataplane {dataplane_id!r}.")) return dataplane + + +async def selected_dataplanes( + principal: Annotated[Principal, Depends(get_principal)], + registry: Annotated[DataplaneRegistry, Depends(get_registry)], + dataplane: Annotated[str | None, Query(description="Restrict a fleet-wide call to one dataplane id.")] = None, +) -> tuple[Dataplane, ...]: + if dataplane is None: + return visible_to(registry, principal) + return (await dataplane_access(dataplane, principal, registry),) diff --git a/litellm/proxy/control_plane/fleet/__init__.py b/litellm/proxy/control_plane/fleet/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/proxy/control_plane/fleet/router.py b/litellm/proxy/control_plane/fleet/router.py new file mode 100644 index 00000000000..4a5642a7212 --- /dev/null +++ b/litellm/proxy/control_plane/fleet/router.py @@ -0,0 +1,74 @@ +from typing import Annotated, Final, TypeAlias, TypeVar + +from fastapi import APIRouter, Depends, Security + +from litellm.proxy.control_plane.auth.dependencies import require_scopes +from litellm.proxy.control_plane.auth.rbac import Principal, Scope +from litellm.proxy.control_plane.common.exceptions import ControlPlaneException, service_unavailable +from litellm.proxy.control_plane.common.fanout import Fanout, Gathered, describe_missed +from litellm.proxy.control_plane.dataplanes.client import DataplaneClient +from litellm.proxy.control_plane.dataplanes.credentials import Caller +from litellm.proxy.control_plane.dataplanes.dependencies import ( + get_caller, + get_dataplane_client, + get_fanout, + selected_dataplanes, +) +from litellm.proxy.control_plane.dataplanes.registry import Dataplane +from litellm.proxy.control_plane.fleet.service import fetch_models, fetch_teams, group_models +from litellm.types.proxy.control_plane.dataplanes import UnreachableDataplane +from litellm.types.proxy.control_plane.fleet import FleetModel, FleetModelsResponse, FleetTeam, FleetTeamsResponse + +T: Final = TypeVar("T") + +router: Final = APIRouter(prefix="/fleet", tags=["fleet"]) + +FleetRead: TypeAlias = Annotated[Principal, Security(require_scopes, scopes=[Scope.FLEET_READ.value])] +Selected: TypeAlias = Annotated[tuple[Dataplane, ...], Depends(selected_dataplanes)] + + +def _or_503(gathered: Gathered[T]) -> Gathered[T]: + if gathered.nothing_answered: + raise ControlPlaneException(service_unavailable(f"No dataplane answered: {describe_missed(gathered.missed)}")) + return gathered + + +def _flatten(gathered: Gathered[tuple[T, ...]]) -> tuple[T, ...]: + return tuple(item for reached in gathered.reached for item in reached.value) # comprehension-ok: flatten + + +def _unreachable(gathered: Gathered[T]) -> tuple[UnreachableDataplane, ...]: + return tuple(UnreachableDataplane(id=miss.dataplane.id, reason=miss.reason) for miss in gathered.missed) + + +@router.get("/models", response_model=FleetModelsResponse) +async def fleet_models( + _principal: FleetRead, + dataplanes: Selected, + caller: Annotated[Caller, Depends(get_caller)], + client: Annotated[DataplaneClient, Depends(get_dataplane_client)], + fanout: Annotated[Fanout, Depends(get_fanout)], +) -> FleetModelsResponse: + """Every deployment across the fleet, plus which dataplanes serve each public model name.""" + gathered: Final[Gathered[tuple[FleetModel, ...]]] = _or_503( + await fanout.gather(dataplanes, lambda dataplane: fetch_models(client, dataplane, caller)) + ) + items: Final = _flatten(gathered) + return FleetModelsResponse( + items=items, groups=group_models(items), partial=gathered.partial, unreachable=_unreachable(gathered) + ) + + +@router.get("/teams", response_model=FleetTeamsResponse) +async def fleet_teams( + _principal: FleetRead, + dataplanes: Selected, + caller: Annotated[Caller, Depends(get_caller)], + client: Annotated[DataplaneClient, Depends(get_dataplane_client)], + fanout: Annotated[Fanout, Depends(get_fanout)], +) -> FleetTeamsResponse: + """Every team the caller may see on each dataplane.""" + gathered: Final[Gathered[tuple[FleetTeam, ...]]] = _or_503( + await fanout.gather(dataplanes, lambda dataplane: fetch_teams(client, dataplane, caller)) + ) + return FleetTeamsResponse(items=_flatten(gathered), partial=gathered.partial, unreachable=_unreachable(gathered)) diff --git a/litellm/proxy/control_plane/fleet/service.py b/litellm/proxy/control_plane/fleet/service.py new file mode 100644 index 00000000000..84af7b8380f --- /dev/null +++ b/litellm/proxy/control_plane/fleet/service.py @@ -0,0 +1,124 @@ +"""Fleet inventory: the union of what every dataplane serves, read through the caller's own identity.""" + +from http import HTTPStatus +from itertools import groupby +from types import MappingProxyType +from typing import Final + +from pydantic import BaseModel, ConfigDict, TypeAdapter + +from litellm.proxy.control_plane.common.fanout import Failure, parse_reply +from litellm.proxy.control_plane.dataplanes.client import CallResult, DataplaneClient, DataplaneReply +from litellm.proxy.control_plane.dataplanes.credentials import Caller +from litellm.proxy.control_plane.dataplanes.registry import Dataplane +from litellm.types.proxy.control_plane.fleet import FleetModel, FleetModelGroup, FleetTeam + +MODEL_INFO_PATH: Final = "/model/info" +TEAM_LIST_PATH: Final = "/team/list" + + +class _Lenient(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + +class _LitellmParams(_Lenient): + model: str | None = None + custom_llm_provider: str | None = None + + +class _ModelInfo(_Lenient): + id: str | None = None + litellm_provider: str | None = None + mode: str | None = None + + +class _ModelEntry(_Lenient): + model_name: str + litellm_params: _LitellmParams = _LitellmParams() + model_info: _ModelInfo = _ModelInfo() + + +class _ModelInfoBody(_Lenient): + data: tuple[_ModelEntry, ...] = () + + +class _TeamEntry(_Lenient): + team_id: str + team_alias: str | None = None + organization_id: str | None = None + spend: float | None = None + max_budget: float | None = None + members_with_roles: tuple[object, ...] = () + models: tuple[str, ...] = () + blocked: bool = False + + +_MODELS: Final = TypeAdapter(_ModelInfoBody) +_TEAMS: Final = TypeAdapter(tuple[_TeamEntry, ...]) + + +async def fetch_models( + client: DataplaneClient, dataplane: Dataplane, caller: Caller +) -> tuple[FleetModel, ...] | Failure: + body: Final = parse_reply(await client.get(dataplane, MODEL_INFO_PATH, caller), _MODELS) + if isinstance(body, Failure): + return body + return tuple( + FleetModel( + dataplane_id=dataplane.id, + model_name=entry.model_name, + litellm_model=entry.litellm_params.model, + provider=entry.litellm_params.custom_llm_provider or entry.model_info.litellm_provider, + model_id=entry.model_info.id, + mode=entry.model_info.mode, + ) + for entry in body.data + ) + + +NOT_ALLOWED_TO_LIST_ALL: Final = frozenset({HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN}) + + +def _refused(result: CallResult) -> bool: + return isinstance(result, DataplaneReply) and result.status in NOT_ALLOWED_TO_LIST_ALL + + +async def _list_teams(client: DataplaneClient, dataplane: Dataplane, caller: Caller) -> CallResult: + """Every team when the dataplane lets the caller see them all, else the teams the caller belongs to.""" + everything: Final = await client.get(dataplane, TEAM_LIST_PATH, caller) + if not _refused(everything): + return everything + own: Final = MappingProxyType({"user_id": caller.principal.subject}) + return await client.get(dataplane, TEAM_LIST_PATH, caller, params=own) + + +async def fetch_teams(client: DataplaneClient, dataplane: Dataplane, caller: Caller) -> tuple[FleetTeam, ...] | Failure: + teams: Final = parse_reply(await _list_teams(client, dataplane, caller), _TEAMS) + if isinstance(teams, Failure): + return teams + return tuple( + FleetTeam( + dataplane_id=dataplane.id, + team_id=team.team_id, + team_alias=team.team_alias, + organization_id=team.organization_id, + spend=team.spend or 0.0, + max_budget=team.max_budget, + members=len(team.members_with_roles), + models=team.models, + blocked=team.blocked, + ) + for team in teams + ) + + +def group_models(models: tuple[FleetModel, ...]) -> tuple[FleetModelGroup, ...]: + by_name: Final = sorted(models, key=_model_name) + return tuple( + FleetModelGroup(model_name=name, dataplanes=tuple(sorted({model.dataplane_id for model in served}))) + for name, served in groupby(by_name, key=_model_name) + ) + + +def _model_name(model: FleetModel) -> str: + return model.model_name diff --git a/litellm/proxy/control_plane/usage/__init__.py b/litellm/proxy/control_plane/usage/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/proxy/control_plane/usage/router.py b/litellm/proxy/control_plane/usage/router.py new file mode 100644 index 00000000000..1bef199de11 --- /dev/null +++ b/litellm/proxy/control_plane/usage/router.py @@ -0,0 +1,154 @@ +from datetime import UTC, date, datetime, timedelta +from typing import Annotated, Final, TypeAlias + +from fastapi import APIRouter, Depends, Query, Security + +from litellm.proxy.control_plane.auth.dependencies import require_scopes +from litellm.proxy.control_plane.auth.rbac import Principal, Scope +from litellm.proxy.control_plane.common.exceptions import ControlPlaneException, bad_request, service_unavailable +from litellm.proxy.control_plane.common.fanout import Fanout, Gathered, describe_missed +from litellm.proxy.control_plane.dataplanes.client import DataplaneClient +from litellm.proxy.control_plane.dataplanes.credentials import Caller +from litellm.proxy.control_plane.dataplanes.dependencies import ( + get_caller, + get_dataplane_client, + get_fanout, + selected_dataplanes, +) +from litellm.proxy.control_plane.dataplanes.registry import Dataplane +from litellm.proxy.control_plane.usage.service import ( + DataplaneDays, + UsageWindow, + daily_rows, + dataplane_totals, + fetch_daily_activity, + ranked, + totals, +) +from litellm.types.proxy.control_plane.dataplanes import UnreachableDataplane +from litellm.types.proxy.control_plane.usage import ( + DailyUsageResponse, + DailyUsageRow, + DataplaneUsage, + TopUsageResponse, + UsageGroupBy, + UsageRank, + UsageSummaryResponse, + UsageWindowResponse, +) +from litellm.types.proxy.management_endpoints.common_daily_activity import DailySpendData + +DEFAULT_WINDOW_DAYS: Final = 30 + +router: Final = APIRouter(prefix="/usage", tags=["usage"]) + +UsageRead: TypeAlias = Annotated[Principal, Security(require_scopes, scopes=[Scope.USAGE_READ.value])] +Selected: TypeAlias = Annotated[tuple[Dataplane, ...], Depends(selected_dataplanes)] +GroupBy: TypeAlias = Annotated[UsageGroupBy, Query(description="Dimension to split the fleet's usage by.")] + + +def usage_window( + start_date: Annotated[date | None, Query(description="First day, inclusive. Defaults to 30 days ago.")] = None, + end_date: Annotated[date | None, Query(description="Last day, inclusive. Defaults to today (UTC).")] = None, +) -> UsageWindow: + end: Final = end_date if end_date is not None else datetime.now(UTC).date() + start: Final = start_date if start_date is not None else end - timedelta(days=DEFAULT_WINDOW_DAYS - 1) + if start > end: + raise ControlPlaneException(bad_request(f"start_date {start} is after end_date {end}.")) + return UsageWindow(start=start, end=end) + + +Window: TypeAlias = Annotated[UsageWindow, Depends(usage_window)] + + +async def _gather_days( + fanout: Fanout, client: DataplaneClient, caller: Caller, dataplanes: tuple[Dataplane, ...], window: UsageWindow +) -> Gathered[tuple[DailySpendData, ...]]: + gathered: Final = await fanout.gather( + dataplanes, lambda dataplane: fetch_daily_activity(client, dataplane, caller, window) + ) + if gathered.nothing_answered: + raise ControlPlaneException(service_unavailable(f"No dataplane answered: {describe_missed(gathered.missed)}")) + return gathered + + +def _fleet(gathered: Gathered[tuple[DailySpendData, ...]]) -> tuple[DataplaneDays, ...]: + return tuple(DataplaneDays(reached.dataplane.id, reached.value) for reached in gathered.reached) + + +def _unreachable(gathered: Gathered[tuple[DailySpendData, ...]]) -> tuple[UnreachableDataplane, ...]: + return tuple(UnreachableDataplane(id=miss.dataplane.id, reason=miss.reason) for miss in gathered.missed) + + +def _window(window: UsageWindow) -> UsageWindowResponse: + return UsageWindowResponse(start_date=window.start, end_date=window.end) + + +@router.get("/summary", response_model=UsageSummaryResponse) +async def usage_summary( + _principal: UsageRead, + dataplanes: Selected, + window: Window, + caller: Annotated[Caller, Depends(get_caller)], + client: Annotated[DataplaneClient, Depends(get_dataplane_client)], + fanout: Annotated[Fanout, Depends(get_fanout)], +) -> UsageSummaryResponse: + """Spend and token totals across every dataplane the caller is bound to.""" + gathered: Final = await _gather_days(fanout, client, caller, dataplanes, window) + fleet: Final = _fleet(gathered) + return UsageSummaryResponse( + window=_window(window), + totals=totals(fleet), + by_dataplane=tuple(DataplaneUsage(id=days.dataplane_id, metrics=dataplane_totals(days)) for days in fleet), + partial=gathered.partial, + unreachable=_unreachable(gathered), + ) + + +@router.get("/daily", response_model=DailyUsageResponse) +async def usage_daily( + _principal: UsageRead, + dataplanes: Selected, + window: Window, + caller: Annotated[Caller, Depends(get_caller)], + client: Annotated[DataplaneClient, Depends(get_dataplane_client)], + fanout: Annotated[Fanout, Depends(get_fanout)], + group_by: GroupBy = "dataplane", +) -> DailyUsageResponse: + """Per-day usage across the fleet, split by dataplane, model, provider, key, team or endpoint.""" + gathered: Final = await _gather_days(fanout, client, caller, dataplanes, window) + return DailyUsageResponse( + window=_window(window), + group_by=group_by, + rows=tuple( + DailyUsageRow(date=day, group=group, metrics=metrics) + for (day, group), metrics in daily_rows(_fleet(gathered), group_by) + ), + partial=gathered.partial, + unreachable=_unreachable(gathered), + ) + + +@router.get("/top", response_model=TopUsageResponse) +async def usage_top( + _principal: UsageRead, + dataplanes: Selected, + window: Window, + caller: Annotated[Caller, Depends(get_caller)], + client: Annotated[DataplaneClient, Depends(get_dataplane_client)], + fanout: Annotated[Fanout, Depends(get_fanout)], + group_by: GroupBy = "model", + limit: Annotated[int, Query(ge=1, le=100)] = 10, +) -> TopUsageResponse: + """The biggest spenders in the window along one dimension, summed across the fleet.""" + gathered: Final = await _gather_days(fanout, client, caller, dataplanes, window) + return TopUsageResponse( + window=_window(window), + group_by=group_by, + limit=limit, + items=tuple( + UsageRank(group=group, metrics=metrics) for group, metrics in ranked(_fleet(gathered), group_by, limit) + ), + partial=gathered.partial, + unreachable=_unreachable(gathered), + ) diff --git a/litellm/proxy/control_plane/usage/service.py b/litellm/proxy/control_plane/usage/service.py new file mode 100644 index 00000000000..e00a5b96170 --- /dev/null +++ b/litellm/proxy/control_plane/usage/service.py @@ -0,0 +1,164 @@ +"""Fleet-wide usage: fetch each dataplane's daily activity, then pure reducers over SpendMetrics.""" + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from datetime import date +from functools import reduce +from itertools import groupby +from types import MappingProxyType +from typing import Final, TypeAlias + +from pydantic import TypeAdapter + +from litellm.proxy.control_plane.common.fanout import Failure, parse_reply +from litellm.proxy.control_plane.dataplanes.client import DataplaneClient +from litellm.proxy.control_plane.dataplanes.credentials import Caller +from litellm.proxy.control_plane.dataplanes.registry import Dataplane +from litellm.types.proxy.control_plane.usage import UsageGroupBy +from litellm.types.proxy.management_endpoints.common_daily_activity import ( + DailySpendData, + SpendAnalyticsPaginatedResponse, + SpendMetrics, +) + +DAILY_ACTIVITY_PATH: Final = "/user/daily/activity" +PAGE_SIZE: Final = 1000 +MAX_PAGES: Final = 10 +NO_TEAM: Final = "(no team)" + +_DAILY: Final = TypeAdapter(SpendAnalyticsPaginatedResponse) +_NUMBERS: Final = TypeAdapter(Mapping[str, float]) + +DatedGroup: TypeAlias = tuple[date, str] +Grouped: TypeAlias = tuple[tuple[DatedGroup, SpendMetrics], ...] + + +@dataclass(frozen=True, slots=True) +class UsageWindow: + start: date + end: date + + def params(self, page: int) -> Mapping[str, str]: + return MappingProxyType( + { + "start_date": self.start.isoformat(), + "end_date": self.end.isoformat(), + "page": str(page), + "page_size": str(PAGE_SIZE), + } + ) + + +@dataclass(frozen=True, slots=True) +class DataplaneDays: + dataplane_id: str + days: tuple[DailySpendData, ...] + + +async def fetch_daily_activity( + client: DataplaneClient, dataplane: Dataplane, caller: Caller, window: UsageWindow +) -> tuple[DailySpendData, ...] | Failure: + return await _collect(client, dataplane, caller, window, page=1, collected=()) + + +async def _collect( + client: DataplaneClient, + dataplane: Dataplane, + caller: Caller, + window: UsageWindow, + page: int, + collected: tuple[DailySpendData, ...], +) -> tuple[DailySpendData, ...] | Failure: + parsed: Final = parse_reply( + await client.get(dataplane, DAILY_ACTIVITY_PATH, caller, params=window.params(page)), _DAILY + ) + if isinstance(parsed, Failure): + return parsed + rows: Final = collected + tuple(parsed.results) + if not parsed.metadata.has_more: + return rows + if page >= MAX_PAGES: + return Failure(f"more than {MAX_PAGES} pages of daily activity, narrow the window") + return await _collect(client, dataplane, caller, window, page + 1, rows) + + +def add_metrics(left: SpendMetrics, right: SpendMetrics) -> SpendMetrics: + first: Final = _NUMBERS.validate_python(left.model_dump()) + second: Final = _NUMBERS.validate_python(right.model_dump()) + return SpendMetrics.model_validate({name: value + second[name] for name, value in first.items()}) + + +def sum_metrics(metrics: Iterable[SpendMetrics]) -> SpendMetrics: + return reduce(add_metrics, metrics, SpendMetrics()) + + +def _key_label(api_key: str, alias: str | None) -> str: + return alias if alias else api_key + + +def group_day(day: DailySpendData, group_by: UsageGroupBy, dataplane_id: str) -> Mapping[str, SpendMetrics]: + breakdown: Final = day.breakdown + match group_by: + case "dataplane": + return MappingProxyType({dataplane_id: day.metrics}) + case "model": + return MappingProxyType({name: entry.metrics for name, entry in breakdown.models.items()}) + case "model_group": + return MappingProxyType({name: entry.metrics for name, entry in breakdown.model_groups.items()}) + case "provider": + return MappingProxyType({name: entry.metrics for name, entry in breakdown.providers.items()}) + case "endpoint": + return MappingProxyType({name: entry.metrics for name, entry in breakdown.endpoints.items()}) + case "api_key": + by_label: Final = tuple( + (_key_label(key, entry.metadata.key_alias), entry.metrics) for key, entry in breakdown.api_keys.items() + ) + return MappingProxyType({label: metrics for (_, label), metrics in _merge_by_key(by_label, day.date)}) + case "team": + by_team: Final = tuple( + (entry.metadata.team_id or NO_TEAM, entry.metrics) for entry in breakdown.api_keys.values() + ) + return MappingProxyType({team: metrics for (_, team), metrics in _merge_by_key(by_team, day.date)}) + + +def _merge_by_key(entries: Iterable[tuple[str, SpendMetrics]], day: date) -> Grouped: + return merge_grouped(((day, group), metrics) for group, metrics in entries) + + +def merge_grouped(entries: Iterable[tuple[DatedGroup, SpendMetrics]]) -> Grouped: + ordered: Final = sorted(entries, key=_dated_group) + return tuple( + (key, sum_metrics(metrics for _, metrics in bucket)) for key, bucket in groupby(ordered, key=_dated_group) + ) + + +def _dated_group(entry: tuple[DatedGroup, SpendMetrics]) -> DatedGroup: + return entry[0] + + +def _dated_groups(days: DataplaneDays, group_by: UsageGroupBy) -> Iterable[tuple[DatedGroup, SpendMetrics]]: + for day in days.days: + for group, metrics in group_day(day, group_by, days.dataplane_id).items(): + yield (day.date, group), metrics + + +def daily_rows(fleet: Iterable[DataplaneDays], group_by: UsageGroupBy) -> Grouped: + return merge_grouped( + entry for days in fleet for entry in _dated_groups(days, group_by) + ) # comprehension-ok: flatten + + +def ranked(fleet: Iterable[DataplaneDays], group_by: UsageGroupBy, limit: int) -> tuple[tuple[str, SpendMetrics], ...]: + per_group: Final = merge_grouped( + ((date.min, group), metrics) for (_, group), metrics in daily_rows(fleet, group_by) + ) + by_spend: Final = sorted(per_group, key=lambda item: item[1].spend, reverse=True) + return tuple((group, metrics) for (_, group), metrics in by_spend[:limit]) + + +def totals(fleet: Iterable[DataplaneDays]) -> SpendMetrics: + return sum_metrics(day.metrics for days in fleet for day in days.days) # comprehension-ok: flatten + + +def dataplane_totals(days: DataplaneDays) -> SpendMetrics: + return sum_metrics(day.metrics for day in days.days) diff --git a/litellm/types/proxy/control_plane/dataplanes.py b/litellm/types/proxy/control_plane/dataplanes.py index a62d103878d..9948642f7c0 100644 --- a/litellm/types/proxy/control_plane/dataplanes.py +++ b/litellm/types/proxy/control_plane/dataplanes.py @@ -42,3 +42,12 @@ class DataplaneIdentityResponse(LiteLLMBaseModel): user_id: str | None user_email: str | None user_role: str | None + + +class UnreachableDataplane(LiteLLMBaseModel): + """A dataplane that did not answer a fleet-wide call, named so the caller knows what the numbers miss.""" + + model_config = ConfigDict(frozen=True) + + id: str + reason: str diff --git a/litellm/types/proxy/control_plane/fleet.py b/litellm/types/proxy/control_plane/fleet.py new file mode 100644 index 00000000000..90bb88f0a31 --- /dev/null +++ b/litellm/types/proxy/control_plane/fleet.py @@ -0,0 +1,55 @@ +from pydantic import ConfigDict + +from litellm.types.proxy.control_plane.dataplanes import UnreachableDataplane +from litellm.types.utils import LiteLLMBaseModel + + +class FleetModel(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + dataplane_id: str + model_name: str + litellm_model: str | None + provider: str | None + model_id: str | None + mode: str | None + + +class FleetModelGroup(LiteLLMBaseModel): + """One public model name and every dataplane that serves it.""" + + model_config = ConfigDict(frozen=True) + + model_name: str + dataplanes: tuple[str, ...] + + +class FleetModelsResponse(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + items: tuple[FleetModel, ...] + groups: tuple[FleetModelGroup, ...] + partial: bool + unreachable: tuple[UnreachableDataplane, ...] + + +class FleetTeam(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + dataplane_id: str + team_id: str + team_alias: str | None + organization_id: str | None + spend: float + max_budget: float | None + members: int + models: tuple[str, ...] + blocked: bool + + +class FleetTeamsResponse(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + items: tuple[FleetTeam, ...] + partial: bool + unreachable: tuple[UnreachableDataplane, ...] diff --git a/litellm/types/proxy/control_plane/usage.py b/litellm/types/proxy/control_plane/usage.py new file mode 100644 index 00000000000..181367678a1 --- /dev/null +++ b/litellm/types/proxy/control_plane/usage.py @@ -0,0 +1,72 @@ +from datetime import date +from typing import Literal, TypeAlias + +from pydantic import ConfigDict + +from litellm.types.proxy.control_plane.dataplanes import UnreachableDataplane +from litellm.types.proxy.management_endpoints.common_daily_activity import SpendMetrics +from litellm.types.utils import LiteLLMBaseModel + +UsageGroupBy: TypeAlias = Literal["dataplane", "model", "model_group", "provider", "api_key", "team", "endpoint"] + + +class UsageWindowResponse(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + start_date: date + end_date: date + + +class DataplaneUsage(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + id: str + metrics: SpendMetrics + + +class UsageSummaryResponse(LiteLLMBaseModel): + """Fleet-wide totals for the window, with the per-dataplane split and whoever did not answer.""" + + model_config = ConfigDict(frozen=True) + + window: UsageWindowResponse + totals: SpendMetrics + by_dataplane: tuple[DataplaneUsage, ...] + partial: bool + unreachable: tuple[UnreachableDataplane, ...] + + +class DailyUsageRow(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + date: date + group: str + metrics: SpendMetrics + + +class DailyUsageResponse(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + window: UsageWindowResponse + group_by: UsageGroupBy + rows: tuple[DailyUsageRow, ...] + partial: bool + unreachable: tuple[UnreachableDataplane, ...] + + +class UsageRank(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + group: str + metrics: SpendMetrics + + +class TopUsageResponse(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + window: UsageWindowResponse + group_by: UsageGroupBy + limit: int + items: tuple[UsageRank, ...] + partial: bool + unreachable: tuple[UnreachableDataplane, ...] diff --git a/tests/unit/proxy/control_plane/common/test_fanout.py b/tests/unit/proxy/control_plane/common/test_fanout.py new file mode 100644 index 00000000000..a0cbfc0d36b --- /dev/null +++ b/tests/unit/proxy/control_plane/common/test_fanout.py @@ -0,0 +1,85 @@ +import asyncio +from dataclasses import dataclass, field + +import pytest +from pydantic import BaseModel, TypeAdapter + +from litellm.proxy.control_plane.common.fanout import Failure, Fanout, Missed, parse_reply +from litellm.proxy.control_plane.dataplanes.client import DataplaneReply, DataplaneUnreachable +from litellm.proxy.control_plane.dataplanes.credentials import CredentialUnavailable, IssuedCredential +from litellm.proxy.control_plane.dataplanes.registry import Dataplane + + +class _NoCredential: + async def headers(self, caller) -> IssuedCredential: + raise AssertionError("fanout never issues credentials itself") + + +def _dataplane(dataplane_id: str) -> Dataplane: + return Dataplane(dataplane_id, dataplane_id, f"https://{dataplane_id}.test", _NoCredential(), ()) + + +@dataclass +class _InFlight: # mutable-ok: test probe counting concurrent fetches + now: int = 0 + peak: int = 0 + seen: list[str] = field(default_factory=list) + + +async def test_gather_bounds_concurrency_and_keeps_dataplane_order() -> None: + probe = _InFlight() + + async def fetch(dataplane: Dataplane) -> str | Failure: + probe.now += 1 + probe.peak = max(probe.peak, probe.now) + await asyncio.sleep(0) + probe.now -= 1 + probe.seen.append(dataplane.id) + return dataplane.id.upper() + + dataplanes = tuple(_dataplane(name) for name in ("a", "b", "c", "d", "e")) + gathered = await Fanout(concurrency=2).gather(dataplanes, fetch) + assert probe.peak == 2, f"semaphore of 2 let {probe.peak} fetches run at once" + assert [reached.value for reached in gathered.reached] == ["A", "B", "C", "D", "E"] + assert gathered.missed == () and not gathered.partial + + +async def test_a_failing_dataplane_is_a_missed_value_and_marks_the_result_partial() -> None: + async def fetch(dataplane: Dataplane) -> int | Failure: + return Failure("boom") if dataplane.id == "b" else 1 + + gathered = await Fanout(concurrency=4).gather(tuple(_dataplane(name) for name in "abc"), fetch) + assert [reached.dataplane.id for reached in gathered.reached] == ["a", "c"] + assert gathered.missed == (Missed(gathered.missed[0].dataplane, "boom"),) + assert gathered.missed[0].dataplane.id == "b" + assert gathered.partial and not gathered.nothing_answered + + +async def test_nothing_answered_only_when_every_dataplane_missed() -> None: + async def fetch(dataplane: Dataplane) -> int | Failure: + return Failure("down") + + all_down = await Fanout(concurrency=1).gather((_dataplane("a"),), fetch) + none_asked = await Fanout(concurrency=1).gather((), fetch) + assert all_down.nothing_answered and not none_asked.nothing_answered + + +class _Shape(BaseModel): + n: int + + +_SHAPE = TypeAdapter(_Shape) + + +@pytest.mark.parametrize( + ("result", "expected"), + [ + (DataplaneReply(200, {"n": 3}, "service_token"), _Shape(n=3)), + (DataplaneReply(200, {"n": "x"}, "service_token"), Failure("unexpected response shape: 1 validation error(s)")), + (DataplaneReply(403, {"error": "no"}, "oidc_exchange"), Failure("HTTP 403 with the oidc_exchange credential")), + (DataplaneUnreachable("ConnectError", 3), Failure("unreachable after 3 attempt(s): ConnectError")), + (CredentialUnavailable("service_token", "no token"), Failure("no usable credential: no token")), + ], +) +def test_parse_reply_turns_every_bad_outcome_into_a_failure(result, expected) -> None: + assert parse_reply(result, _SHAPE) == expected diff --git a/tests/unit/proxy/control_plane/fleet/__init__.py b/tests/unit/proxy/control_plane/fleet/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/proxy/control_plane/fleet/test_router.py b/tests/unit/proxy/control_plane/fleet/test_router.py new file mode 100644 index 00000000000..167a25e3737 --- /dev/null +++ b/tests/unit/proxy/control_plane/fleet/test_router.py @@ -0,0 +1,115 @@ +import httpx + +from litellm.proxy.control_plane.auth.rbac import USAGE_VIEWER_ROLE +from tests.unit.proxy.control_plane.support import response, rig + +EU_MODELS = "https://eu.test/model/info" +US_MODELS = "https://us.test/model/info" +EU_TEAMS = "https://eu.test/team/list" +US_TEAMS = "https://us.test/team/list" + + +def _model(name: str, provider: str, model_id: str, upstream: str | None = None) -> dict[str, object]: + return { + "model_name": name, + "litellm_params": {"model": upstream or f"{provider}/{name}", "custom_llm_provider": provider}, + "model_info": {"id": model_id, "mode": "chat", "db_model": True}, + } + + +def test_fleet_models_unions_deployments_and_groups_them_by_public_name() -> None: + client, _, minter = rig( + { + EU_MODELS: [ + response(200, {"data": [_model("gpt-4o", "azure", "m1"), _model("claude", "anthropic", "m2")]}) + ], + US_MODELS: [response(200, {"data": [_model("gpt-4o", "openai", "m3")]})], + } + ) + body = client.get("/api/v1/fleet/models", headers=minter.headers()).json() + assert [(m["dataplane_id"], m["model_name"], m["provider"], m["model_id"]) for m in body["items"]] == [ + ("eu", "gpt-4o", "azure", "m1"), + ("eu", "claude", "anthropic", "m2"), + ("us", "gpt-4o", "openai", "m3"), + ] + assert body["groups"] == [ + {"model_name": "claude", "dataplanes": ["eu"]}, + {"model_name": "gpt-4o", "dataplanes": ["eu", "us"]}, + ] + assert body["partial"] is False + + +def test_fleet_models_tolerates_an_unparseable_dataplane_and_reports_it() -> None: + client, _, minter = rig( + { + EU_MODELS: [response(200, {"data": [_model("gpt-4o", "azure", "m1")]})], + US_MODELS: [response(200, {"data": "x"})], + } + ) + body = client.get("/api/v1/fleet/models", headers=minter.headers()).json() + assert [m["model_id"] for m in body["items"]] == ["m1"] and body["partial"] is True + assert body["unreachable"][0]["id"] == "us" and "validation error" in body["unreachable"][0]["reason"] + + +def test_fleet_teams_counts_members_and_keeps_budget_fields() -> None: + team = { + "team_id": "t1", + "team_alias": "ml", + "organization_id": None, + "spend": 12.5, + "max_budget": 100.0, + "members_with_roles": [{"user_id": "a", "role": "admin"}, {"user_id": "b", "role": "user"}], + "models": ["gpt-4o"], + "blocked": False, + "metadata": {"ignored": True}, + } + client, _, minter = rig({EU_TEAMS: [response(200, [team])], US_TEAMS: [response(200, [])]}) + body = client.get("/api/v1/fleet/teams", headers=minter.headers()).json() + assert body["items"] == [ + { + "dataplane_id": "eu", + "team_id": "t1", + "team_alias": "ml", + "organization_id": None, + "spend": 12.5, + "max_budget": 100.0, + "members": 2, + "models": ["gpt-4o"], + "blocked": False, + } + ] + + +def test_fleet_teams_falls_back_to_the_callers_own_teams_when_the_dataplane_refuses_to_list_all() -> None: + own = {"team_id": "t-own", "members_with_roles": [{"user_id": "alice"}]} + client, transport, minter = rig( + { + EU_TEAMS: [ + response(401, {"detail": {"error": "Only admin users can query all teams"}}), + response(200, [own]), + ], + US_TEAMS: [response(200, [])], + } + ) + body = client.get("/api/v1/fleet/teams", headers=minter.headers()).json() + assert [(t["dataplane_id"], t["team_id"], t["members"]) for t in body["items"]] == [("eu", "t-own", 1)] + assert body["partial"] is False + eu_params = [params for url, params, _ in transport.calls if url == EU_TEAMS] + assert eu_params == [None, {"user_id": "alice"}] + + +def test_fleet_routes_need_fleet_read_and_503_when_nobody_answers() -> None: + client, _, minter = rig({EU_TEAMS: [httpx.ConnectError("down")] * 3, US_TEAMS: [response(502, "bad")] * 3}) + denied = client.get("/api/v1/fleet/teams", headers=minter.headers(role=USAGE_VIEWER_ROLE)) + assert denied.status_code == 403 and "fleet:read" in denied.json()["detail"] + assert client.get("/api/v1/fleet/teams", headers=minter.headers()).status_code == 503 + + +def test_a_team_with_null_spend_lists_as_zero_instead_of_failing_the_fleet() -> None: + fresh = {"team_id": "new", "spend": None} + client, _, minter = rig({EU_TEAMS: [response(200, [fresh])], US_TEAMS: [response(200, [])]}) + reply = client.get("/api/v1/fleet/teams", headers=minter.headers()) + assert reply.status_code == 200, reply.text + body = reply.json() + assert [(item["team_id"], item["spend"]) for item in body["items"]] == [("new", 0.0)] + assert body["partial"] is False diff --git a/tests/unit/proxy/control_plane/support.py b/tests/unit/proxy/control_plane/support.py new file mode 100644 index 00000000000..1988c17bb2f --- /dev/null +++ b/tests/unit/proxy/control_plane/support.py @@ -0,0 +1,85 @@ +"""Shared rig for control plane route tests: a routed fake transport and a licensed app over two dataplanes.""" + +from collections import defaultdict +from collections.abc import Iterator, Mapping +from dataclasses import dataclass, field +from datetime import timedelta +from typing import Final + +import httpx +from fastapi.testclient import TestClient + +from litellm.litellm_core_utils.oauth_token_exchange import OAuthTokenExchangeConfig +from litellm.proxy.control_plane.app import create_app +from litellm.proxy.control_plane.auth.jwt import SessionTokenCodec +from litellm.proxy.control_plane.auth.rbac import VIEWER_ROLE, RolePolicy, build_principal +from litellm.proxy.control_plane.container import build_container +from litellm.proxy.control_plane.settings import ControlPlaneSettings + +MASTER_KEY: Final = "sk-cp-master" +DATAPLANES: Final = { + "eu": {"url": "https://eu.test", "service_token": {"token": "sk-eu"}}, + "us": {"url": "https://us.test", "name": "US East", "service_token": {"token": "sk-us"}}, +} + + +class Licensed: + def is_premium(self) -> bool: + return True + + +@dataclass +class RoutedTransport: # mutable-ok: records calls for assertions + """Answers each URL from its own queue of outcomes, so fan-out order does not matter to a test.""" + + routes: Mapping[str, Iterator[httpx.Response | Exception]] + calls: list[tuple[str, dict[str, str] | None, dict[str, str] | None]] = field(default_factory=list) + + async def request( + self, + method: str, + url: str, + *, + params: dict[str, str] | None = None, + json: dict[str, object] | None = None, + headers: dict[str, str] | None = None, + timeout: float, + ) -> httpx.Response: + assert method == "GET", "fleet reads never write" + self.calls.append((url, dict(params) if params else None, headers)) + outcome = next(self.routes[url], None) + if outcome is None: + raise AssertionError(f"no scripted outcome left for {url}") + if isinstance(outcome, Exception): + raise outcome + return outcome + + +async def no_exchange(config: OAuthTokenExchangeConfig, subject_token: str) -> str: + raise AssertionError("no oidc dataplane configured") + + +def response(status: int, body: object = None) -> httpx.Response: + return httpx.Response(status, json=body, request=httpx.Request("GET", "https://x")) + + +@dataclass(frozen=True) +class Minter: + codec: SessionTokenCodec + policy: RolePolicy + + def headers(self, *dataplanes: str, role: str = VIEWER_ROLE) -> dict[str, str]: + principal = build_principal( + self.policy, "alice", "alice@x.test", frozenset({role}), None, frozenset(dataplanes), "session" + ) + return {"Authorization": f"Bearer {self.codec.mint(principal)}"} + + +def rig(routes: Mapping[str, list[httpx.Response | Exception]]) -> tuple[TestClient, RoutedTransport, Minter]: + settings = ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": MASTER_KEY, "dataplanes": DATAPLANES}) + queues: defaultdict[str, Iterator[httpx.Response | Exception]] = defaultdict(lambda: iter(())) + queues.update({url: iter(outcomes) for url, outcomes in routes.items()}) + transport = RoutedTransport(queues) + container = build_container(settings, transport=transport, exchanger=no_exchange) + codec = SessionTokenCodec(MASTER_KEY, settings.session_issuer, timedelta(hours=1), container.policy) + return TestClient(create_app(container, Licensed())), transport, Minter(codec, container.policy) diff --git a/tests/unit/proxy/control_plane/usage/__init__.py b/tests/unit/proxy/control_plane/usage/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/proxy/control_plane/usage/test_router.py b/tests/unit/proxy/control_plane/usage/test_router.py new file mode 100644 index 00000000000..e0012cc68d6 --- /dev/null +++ b/tests/unit/proxy/control_plane/usage/test_router.py @@ -0,0 +1,146 @@ +from datetime import date + +import httpx + +from litellm.proxy.control_plane.auth.rbac import USAGE_VIEWER_ROLE +from litellm.proxy.control_plane.usage.service import MAX_PAGES +from tests.unit.proxy.control_plane.support import MASTER_KEY, response, rig + +EU = "https://eu.test/user/daily/activity" +US = "https://us.test/user/daily/activity" +WINDOW = {"start_date": "2026-10-01", "end_date": "2026-10-02"} + + +def _page(rows: list[dict[str, object]], has_more: bool = False) -> dict[str, object]: + return {"results": rows, "metadata": {"has_more": has_more, "page": 1, "total_pages": 2 if has_more else 1}} + + +def _row(day: str, spend: float, models: dict[str, float] | None = None) -> dict[str, object]: + return { + "date": day, + "metrics": {"spend": spend, "api_requests": 1}, + "breakdown": {"models": {name: {"metrics": {"spend": cost}} for name, cost in (models or {}).items()}}, + } + + +def test_summary_adds_both_dataplanes_and_forwards_the_window_and_caller() -> None: + client, transport, minter = rig( + { + EU: [response(200, _page([_row("2026-10-01", 1.5), _row("2026-10-02", 0.5)]))], + US: [response(200, _page([_row("2026-10-01", 2.0)]))], + } + ) + body = client.get("/api/v1/usage/summary", params=WINDOW, headers=minter.headers(role=USAGE_VIEWER_ROLE)).json() + assert body["window"] == WINDOW + assert (body["totals"]["spend"], body["totals"]["api_requests"]) == (4.0, 3) + assert [(dp["id"], dp["metrics"]["spend"]) for dp in body["by_dataplane"]] == [("eu", 2.0), ("us", 2.0)] + assert body["partial"] is False and body["unreachable"] == [] + url, params, headers = transport.calls[0] + assert params == {**WINDOW, "page": "1", "page_size": "1000"} + assert headers is not None and headers["X-LiteLLM-Impersonate-User"] == "alice" + + +def test_summary_pages_through_a_dataplane_until_has_more_is_false() -> None: + client, transport, minter = rig( + { + EU: [ + response(200, _page([_row("2026-10-01", 1.0)], has_more=True)), + response(200, _page([_row("2026-10-02", 1.0)])), + ], + US: [response(200, _page([]))], + } + ) + body = client.get("/api/v1/usage/summary", params=WINDOW, headers=minter.headers()).json() + assert body["totals"]["spend"] == 2.0 + eu_pages = [params["page"] for url, params, _ in transport.calls if url == EU and params] + assert eu_pages == ["1", "2"] + + +def test_one_dataplane_down_gives_partial_numbers_and_names_it() -> None: + client, _, minter = rig( + {EU: [response(200, _page([_row("2026-10-01", 1.0)]))], US: [httpx.ConnectError("refused")] * 3} + ) + body = client.get("/api/v1/usage/summary", params=WINDOW, headers=minter.headers()).json() + assert body["totals"]["spend"] == 1.0 and body["partial"] is True + assert [u["id"] for u in body["unreachable"]] == ["us"] + assert "refused" in body["unreachable"][0]["reason"] + + +def test_every_dataplane_down_is_a_503_problem() -> None: + client, _, minter = rig({EU: [response(503, "maintenance")] * 3, US: [httpx.ReadTimeout("slow")] * 3}) + down = client.get("/api/v1/usage/summary", params=WINDOW, headers=minter.headers()) + assert down.status_code == 503 + assert down.json()["type"] == "urn:litellm:error:service-unavailable" + assert "eu:" in down.json()["detail"] and "us:" in down.json()["detail"] + + +def test_binding_and_dataplane_filter_narrow_the_fan_out() -> None: + client, transport, minter = rig( + { + EU: [response(200, _page([_row("2026-10-01", 1.0)]))] * 2, + US: [response(200, _page([_row("2026-10-01", 9.0)]))], + } + ) + bound = client.get("/api/v1/usage/summary", params=WINDOW, headers=minter.headers("eu")).json() + assert [dp["id"] for dp in bound["by_dataplane"]] == ["eu"] + filtered = client.get( + "/api/v1/usage/summary", params={**WINDOW, "dataplane": "eu"}, headers={"Authorization": f"Bearer {MASTER_KEY}"} + ).json() + assert [dp["id"] for dp in filtered["by_dataplane"]] == ["eu"] + assert all(url == EU for url, _, _ in transport.calls) + assert ( + client.get( + "/api/v1/usage/summary", params={**WINDOW, "dataplane": "us"}, headers=minter.headers("eu") + ).status_code + == 403 + ) + + +def test_daily_groups_by_model_across_dataplanes_and_top_ranks_them() -> None: + client, _, minter = rig( + { + EU: [response(200, _page([_row("2026-10-01", 3.0, {"gpt-4o": 2.0, "claude": 1.0})]))] * 2, + US: [response(200, _page([_row("2026-10-01", 5.0, {"gpt-4o": 5.0})]))] * 2, + } + ) + daily = client.get("/api/v1/usage/daily", params={**WINDOW, "group_by": "model"}, headers=minter.headers()).json() + assert [(row["date"], row["group"], row["metrics"]["spend"]) for row in daily["rows"]] == [ + ("2026-10-01", "claude", 1.0), + ("2026-10-01", "gpt-4o", 7.0), + ] + top = client.get("/api/v1/usage/top", params={**WINDOW, "limit": 1}, headers=minter.headers()).json() + assert top["group_by"] == "model" and top["limit"] == 1 + assert [(item["group"], item["metrics"]["spend"]) for item in top["items"]] == [("gpt-4o", 7.0)] + + +def test_usage_routes_need_the_usage_read_scope_and_default_the_window_to_a_month() -> None: + client, transport, minter = rig({EU: [response(200, _page([]))], US: [response(200, _page([]))]}) + assert client.get("/api/v1/usage/daily", params=WINDOW).status_code == 401 + body = client.get("/api/v1/usage/summary", headers=minter.headers()).json() + start, end = date.fromisoformat(body["window"]["start_date"]), date.fromisoformat(body["window"]["end_date"]) + assert (end - start).days == 29 + + +def test_a_dataplane_that_still_has_more_after_the_page_cap_is_reported_not_truncated() -> None: + client, transport, minter = rig( + { + EU: [response(200, _page([_row("2026-10-01", 1.0)], has_more=True)) for _ in range(MAX_PAGES)], + US: [response(200, _page([_row("2026-10-01", 2.0)]))], + } + ) + body = client.get("/api/v1/usage/summary", params=WINDOW, headers=minter.headers()).json() + assert body["totals"]["spend"] == 2.0, "eu's incomplete pages must not count as final numbers" + assert body["partial"] is True and [u["id"] for u in body["unreachable"]] == ["eu"] + assert "pages" in body["unreachable"][0]["reason"] + assert sum(1 for url, _, _ in transport.calls if url == EU) == MAX_PAGES + + +def test_reversed_window_is_a_400_before_any_dataplane_is_asked() -> None: + client, transport, minter = rig({EU: [], US: []}) + reversed_window = client.get( + "/api/v1/usage/summary", + params={"start_date": "2026-10-03", "end_date": "2026-10-01"}, + headers=minter.headers(), + ) + assert reversed_window.status_code == 400 and "after end_date" in reversed_window.json()["detail"] + assert transport.calls == [] diff --git a/tests/unit/proxy/control_plane/usage/test_service.py b/tests/unit/proxy/control_plane/usage/test_service.py new file mode 100644 index 00000000000..a6f41f19931 --- /dev/null +++ b/tests/unit/proxy/control_plane/usage/test_service.py @@ -0,0 +1,111 @@ +from datetime import date + +import pytest + +from litellm.proxy.control_plane.usage.service import ( + NO_TEAM, + DataplaneDays, + add_metrics, + daily_rows, + group_day, + ranked, + totals, +) +from litellm.types.proxy.management_endpoints.common_daily_activity import ( + BreakdownMetrics, + DailySpendData, + KeyMetadata, + KeyMetricWithMetadata, + MetricWithMetadata, + SpendMetrics, +) + +D1 = date(2026, 10, 1) +D2 = date(2026, 10, 2) + + +def _m(spend: float, requests: int = 1) -> SpendMetrics: + return SpendMetrics(spend=spend, api_requests=requests, successful_requests=requests, total_tokens=requests * 10) + + +def _day( + day: date, + spend: float, + models: dict[str, float] | None = None, + keys: dict[str, tuple[float, str | None, str | None]] | None = None, +) -> DailySpendData: + return DailySpendData( + date=day, + metrics=_m(spend), + breakdown=BreakdownMetrics( + models={name: MetricWithMetadata(metrics=_m(cost)) for name, cost in (models or {}).items()}, + api_keys={ + key: KeyMetricWithMetadata(metrics=_m(cost), metadata=KeyMetadata(key_alias=alias, team_id=team)) + for key, (cost, alias, team) in (keys or {}).items() + }, + ), + ) + + +def test_add_metrics_sums_every_numeric_field() -> None: + left = SpendMetrics.model_validate({name: 1 for name in SpendMetrics.model_fields}) + right = SpendMetrics.model_validate({name: 2 for name in SpendMetrics.model_fields}) + total = add_metrics(left, right) + assert total.model_dump() == {name: 3 for name in SpendMetrics.model_fields} + + +def test_totals_add_up_across_dataplanes_and_days() -> None: + fleet = ( + DataplaneDays("eu", (_day(D1, 1.0), _day(D2, 2.0))), + DataplaneDays("us", (_day(D1, 4.0),)), + ) + assert totals(fleet) == _m(7.0, requests=3) + + +def test_daily_rows_merge_the_same_model_on_the_same_day_from_two_dataplanes() -> None: + fleet = ( + DataplaneDays( + "eu", (_day(D1, 3.0, models={"gpt-4o": 2.0, "claude": 1.0}), _day(D2, 1.0, models={"claude": 1.0})) + ), + DataplaneDays("us", (_day(D1, 5.0, models={"gpt-4o": 5.0}),)), + ) + rows = daily_rows(fleet, "model") + assert [(day, group, metrics.spend) for (day, group), metrics in rows] == [ + (D1, "claude", 1.0), + (D1, "gpt-4o", 7.0), + (D2, "claude", 1.0), + ] + assert daily_rows(fleet, "dataplane") == ( + ((D1, "eu"), _m(3.0)), + ((D1, "us"), _m(5.0)), + ((D2, "eu"), _m(1.0)), + ) + + +def test_team_grouping_folds_keys_into_their_team_and_labels_keys_by_alias() -> None: + day = _day( + D1, + 6.0, + keys={"hash-a": (1.0, "alice-key", "t1"), "hash-b": (2.0, None, "t1"), "hash-c": (3.0, "svc", None)}, + ) + assert {team: metrics.spend for team, metrics in group_day(day, "team", "eu").items()} == {"t1": 3.0, NO_TEAM: 3.0} + assert set(group_day(day, "api_key", "eu")) == {"alice-key", "hash-b", "svc"} + + +def test_ranked_orders_by_fleet_spend_and_honours_the_limit() -> None: + fleet = ( + DataplaneDays("eu", (_day(D1, 0, models={"a": 1.0, "b": 5.0}), _day(D2, 0, models={"a": 1.0}))), + DataplaneDays("us", (_day(D1, 0, models={"c": 3.0, "a": 2.0}),)), + ) + assert [(group, metrics.spend) for group, metrics in ranked(fleet, "model", limit=2)] == [("b", 5.0), ("a", 4.0)] + assert len(ranked(fleet, "model", limit=10)) == 3 + + +@pytest.mark.parametrize("dimension", ["model_group", "provider", "endpoint"]) +def test_empty_breakdowns_group_to_nothing(dimension) -> None: + assert dict(group_day(_day(D1, 1.0), dimension, "eu")) == {} + + +def test_two_keys_sharing_an_alias_add_up_instead_of_overwriting() -> None: + day = _day(D1, 8.0, keys={"hash-a": (3.0, "shared", None), "hash-b": (5.0, "shared", None)}) + assert {label: metrics.spend for label, metrics in group_day(day, "api_key", "eu").items()} == {"shared": 8.0}