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

@@ -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

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."""
 

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)

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),
     )

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),)

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))

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

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),
+    )

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)

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

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, ...]

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, ...]

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

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

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)

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 == []

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}