feat(proxy): auth_v2 authz observability metrics

Adds an in-process, dependency-free metrics collector so the authz layer is
observable: decision counts (keyed decision/resource/action), authz latency
(count + sum over the casbin enforce call), and policy-cache hit/miss.

- metrics.py exposes a singleton with observe_decision / observe_latency /
  record_cache and a typed snapshot() for a /metrics export at the edge.
- The authorizer counts every allow/deny/loud-open and times the enforce call;
  the entry point does the same for the model-call decision; the policy store
  records cache hit vs miss. No hot-path dependency - just counters.

Fully typed (no new bare Any), mypy clean on 23 files, 141 tests green, still
imports with casbin/authlib absent.

litellm/proxy/auth/v2/__init__.py

@@ -1,4 +1,5 @@
 from .audit import AuthzDecision, Decision, register_sink, reset_sinks
+from .metrics import MetricsSnapshot, metrics
 from .context import (
     AuthMethod,
     RequestAuthContext,
@@ -25,4 +26,6 @@ __all__ = [
     "Decision",
     "register_sink",
     "reset_sinks",
+    "metrics",
+    "MetricsSnapshot",
 ]

litellm/proxy/auth/v2/authz/authorizer.py

@@ -1,7 +1,9 @@
 import logging
+import time
 from typing import Any, Dict, Optional
 
 from ..audit import AuthzDecision, Decision, record
+from ..metrics import metrics
 from ..principal import Principal
 from ..protocols import SupportsEnforce
 from .route_map import GovernedRoute, match_route
@@ -65,13 +67,18 @@ def authorize(
                 auth_method=auth_method,
             )
         )
+        metrics.observe_decision(Decision.LOUD_OPEN, "route", "*")
         return
 
     obj = _build_object(rule, request_data)
+    start = time.perf_counter()
     allowed = enforcer.enforce(principal.subject, principal.domain, obj, rule.action)
+    metrics.observe_latency(time.perf_counter() - start)
+    decision = Decision.ALLOW if allowed else Decision.DENY
+    metrics.observe_decision(decision, rule.resource, rule.action)
     record(
         AuthzDecision(
-            decision=Decision.ALLOW if allowed else Decision.DENY,
+            decision=decision,
             subject=principal.subject,
             domain=principal.domain,
             obj=obj,

litellm/proxy/auth/v2/authz/policy_store.py

@@ -1,6 +1,7 @@
 import time
 from typing import List, Optional, Tuple
 
+from ..metrics import metrics
 from ..protocols import CasbinRuleRow, PolicyDB
 
 # Always-present bootstrap: principals carrying the proxy_admin role keep full
@@ -68,8 +69,10 @@ async def load_policy_snapshot(
     global _cache
     now = time.monotonic()
     if _cache is not None and (now - _cache[0]) < _CACHE_TTL_SECONDS:
+        metrics.record_cache(hit=True)
         return _cache[1], _cache[2], _cache[3], _cache[4]
 
+    metrics.record_cache(hit=False)
     rows: List[CasbinRuleRow] = []
     if prisma_client is not None:
         rows = await prisma_client.db.litellm_casbinrule.find_many()

litellm/proxy/auth/v2/entry.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import time
 from typing import TYPE_CHECKING, Optional, Tuple, cast
 
 from fastapi import HTTPException, Request, status
@@ -7,6 +8,7 @@ from fastapi import HTTPException, Request, status
 from litellm.integrations.otel.runtime import seed_request_identity
 
 from .audit import AuthzDecision, Decision, record
+from .metrics import metrics
 from .authn.authenticators import AuthContext, AuthResult, authenticate
 from .authz.authorizer import AuthorizationDenied, authorize
 from .authz.enforcer import CasbinEnforcer
@@ -198,10 +200,14 @@ async def user_api_key_auth_v2(
         if requested_model:
             enforcer = await _build_enforcer(principal, prisma_client)
             obj = f"model:{requested_model}"
+            start = time.perf_counter()
             allowed = enforcer.enforce(principal.subject, principal.domain, obj, "call")
+            metrics.observe_latency(time.perf_counter() - start)
+            decision = Decision.ALLOW if allowed else Decision.DENY
+            metrics.observe_decision(decision, "model", "call")
             record(
                 AuthzDecision(
-                    decision=Decision.ALLOW if allowed else Decision.DENY,
+                    decision=decision,
                     subject=principal.subject,
                     domain=principal.domain,
                     obj=obj,

litellm/proxy/auth/v2/metrics.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, Tuple
+
+from .audit import Decision
+
+
+@dataclass(frozen=True)
+class MetricsSnapshot:
+    """A point-in-time read of the auth_v2 authz metrics, for a /metrics export."""
+
+    # "decision:resource:action" -> count
+    decisions: Dict[str, int]
+    authz_latency_count: int
+    authz_latency_sum_seconds: float
+    policy_cache_hits: int
+    policy_cache_misses: int
+
+
+class _Metrics:
+    """In-process authz metrics. Dependency-free so it never burdens the hot path;
+    a Prometheus/OTel exporter reads snapshot() at the edge."""
+
+    def __init__(self) -> None:
+        self._decisions: Dict[Tuple[str, str, str], int] = {}
+        self._latency_count: int = 0
+        self._latency_sum: float = 0.0
+        self._cache_hits: int = 0
+        self._cache_misses: int = 0
+
+    def observe_decision(self, decision: Decision, resource: str, action: str) -> None:
+        key = (decision.value, resource, action)
+        self._decisions[key] = self._decisions.get(key, 0) + 1
+
+    def observe_latency(self, seconds: float) -> None:
+        self._latency_count += 1
+        self._latency_sum += seconds
+
+    def record_cache(self, hit: bool) -> None:
+        if hit:
+            self._cache_hits += 1
+        else:
+            self._cache_misses += 1
+
+    def snapshot(self) -> MetricsSnapshot:
+        return MetricsSnapshot(
+            decisions={
+                f"{d}:{r}:{a}": count for (d, r, a), count in self._decisions.items()
+            },
+            authz_latency_count=self._latency_count,
+            authz_latency_sum_seconds=self._latency_sum,
+            policy_cache_hits=self._cache_hits,
+            policy_cache_misses=self._cache_misses,
+        )
+
+    def reset(self) -> None:
+        self._decisions.clear()
+        self._latency_count = 0
+        self._latency_sum = 0.0
+        self._cache_hits = 0
+        self._cache_misses = 0
+
+
+metrics = _Metrics()

tests/test_litellm/proxy/auth/v2/test_metrics.py

@@ -0,0 +1,86 @@
+import pytest
+
+from litellm.proxy.auth.v2.audit import Decision
+from litellm.proxy.auth.v2.authz.authorizer import AuthorizationDenied, authorize
+from litellm.proxy.auth.v2.metrics import metrics
+from litellm.proxy.auth.v2.principal import Principal
+
+PRINCIPAL = Principal(subject="user:u1", domain="*", groupings=[])
+
+
+@pytest.fixture(autouse=True)
+def _reset_metrics():
+    metrics.reset()
+    yield
+    metrics.reset()
+
+
+class _Enforcer:
+    def __init__(self, ret: bool):
+        self.ret = ret
+
+    def enforce(self, *_args: object) -> bool:
+        return self.ret
+
+
+def test_decisions_are_counted_by_decision_resource_action():
+    metrics.observe_decision(Decision.ALLOW, "model", "read")
+    metrics.observe_decision(Decision.ALLOW, "model", "read")
+    metrics.observe_decision(Decision.DENY, "model", "write")
+    snap = metrics.snapshot()
+    assert snap.decisions["allow:model:read"] == 2
+    assert snap.decisions["deny:model:write"] == 1
+
+
+def test_latency_accumulates_count_and_sum():
+    metrics.observe_latency(0.1)
+    metrics.observe_latency(0.3)
+    snap = metrics.snapshot()
+    assert snap.authz_latency_count == 2
+    assert snap.authz_latency_sum_seconds == pytest.approx(0.4)
+
+
+def test_cache_hit_rate_tracking():
+    metrics.record_cache(hit=True)
+    metrics.record_cache(hit=True)
+    metrics.record_cache(hit=False)
+    snap = metrics.snapshot()
+    assert snap.policy_cache_hits == 2
+    assert snap.policy_cache_misses == 1
+
+
+def test_reset_clears_everything():
+    metrics.observe_decision(Decision.ALLOW, "model", "read")
+    metrics.observe_latency(0.1)
+    metrics.record_cache(hit=True)
+    metrics.reset()
+    snap = metrics.snapshot()
+    assert snap.decisions == {}
+    assert snap.authz_latency_count == 0
+    assert snap.policy_cache_hits == 0
+
+
+def test_authorize_feeds_decision_and_latency_metrics():
+    # A governed allow records one decision and one latency observation.
+    authorize(PRINCIPAL, "/model/new", {}, _Enforcer(True))
+    snap = metrics.snapshot()
+    assert snap.decisions["allow:model:write"] == 1
+    assert snap.authz_latency_count == 1
+
+
+def test_authorize_deny_is_counted():
+    with pytest.raises(AuthorizationDenied):
+        authorize(PRINCIPAL, "/model/delete", {"model_id": "m9"}, _Enforcer(False))
+    assert metrics.snapshot().decisions["deny:model:delete"] == 1
+
+
+def test_loud_open_is_counted_without_latency():
+    # Ungoverned routes don't reach the enforcer, so no latency is observed.
+    class _Exploding:
+        def enforce(self, *_args: object) -> bool:
+            raise AssertionError("must not enforce")
+
+    authorize(PRINCIPAL, "/chat/completions", {}, _Exploding())
+    snap = metrics.snapshot()
+    assert snap.decisions["loud_open:route:*"] == 1
+    assert snap.authz_latency_count == 0