Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/enterprise_billing/billing_metrics.py: 48%
142 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Push-based OTLP metering for enterprise litellm deployments.
4Owns a dedicated OpenTelemetry meter provider and an OTLP/HTTP exporter
5authenticated to our global collector with a TLS client certificate. The
6collector front end terminates mutual TLS: the client certificate presented
7here is validated against our CA at the edge, and the verified subject is
8what identifies the deployment. It is intentionally isolated from the global
9meter provider so the customer's own OTEL metrics are untouched and ours
10never leak into their backend.
12The deployment's identity rides on the TLS client certificate, not on the
13payload; the secret license key is never sent as an attribute or header.
14"""
16import os
17import tempfile
18from dataclasses import dataclass
19from typing import TYPE_CHECKING, Final, Optional
21from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
22from opentelemetry.metrics import Counter
23from opentelemetry.sdk.metrics import MeterProvider
24from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
25from opentelemetry.sdk.resources import Resource
27from litellm._logging import verbose_proxy_logger
28from litellm.proxy.middleware.billable_request_metrics_middleware import (
29 BillableCategory,
30)
32if TYPE_CHECKING: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true
33 from litellm.proxy._types import EnterpriseLicenseData
35ENDPOINT_ENV: Final = "LITELLM_BILLING_METRICS_ENDPOINT"
36CLIENT_CERT_ENV: Final = "LITELLM_BILLING_METRICS_CLIENT_CERT"
37CLIENT_KEY_ENV: Final = "LITELLM_BILLING_METRICS_CLIENT_KEY"
38CA_CERT_ENV: Final = "LITELLM_BILLING_METRICS_CA_CERT"
39EXPORT_INTERVAL_ENV: Final = "LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS"
40DEFAULT_EXPORT_INTERVAL_MS: Final = 60_000
41SHUTDOWN_FLUSH_TIMEOUT_MS: Final = 5_000
42_METRICS_PATH: Final = "/v1/metrics"
44# The cert env vars take a path or the PEM itself. Secret stores that inject
45# values as env content cannot mount them as files, so inline PEM is written out.
46_PEM_PREFIX: Final = "-----BEGIN"
47_PEM_DIR_PREFIX: Final = "litellm-billing-mtls-"
48_PEM_FILE_MODE: Final = 0o600
49_CLIENT_CERT_FILENAME: Final = "client.crt"
50_CLIENT_KEY_FILENAME: Final = "client.key"
51_CA_CERT_FILENAME: Final = "ca.crt"
53METRIC_NAME: Final = "litellm.enterprise.billable_requests"
54METER_NAME: Final = "litellm.enterprise.billing"
56AttributeValue = str | int
59@dataclass(frozen=True, slots=True)
60class BillingMetricsConfig:
61 endpoint: str
62 client_cert_path: str
63 client_key_path: str
64 ca_cert_path: str | None
65 export_interval_ms: int
66 litellm_version: str
67 license_id: str | None
70def _metrics_endpoint(endpoint: str) -> str:
71 """The OTLP/HTTP metric exporter wants the full URL including the signal path."""
72 trimmed: Final = endpoint.rstrip("/")
73 return trimmed if trimmed.endswith(_METRICS_PATH) else f"{trimmed}{_METRICS_PATH}"
76def _resource_attributes(config: BillingMetricsConfig) -> dict[str, AttributeValue]:
77 base: Final[dict[str, AttributeValue]] = {
78 "service.name": "litellm-proxy",
79 "litellm.version": config.litellm_version,
80 }
81 license_attr: dict[str, AttributeValue] = {"litellm.license.id": config.license_id} if config.license_id else {}
82 return {**base, **license_attr}
85def _billable_attributes(
86 category: BillableCategory, route: str, status_code: int, model_id: str | None
87) -> dict[str, AttributeValue]:
88 base: Final[dict[str, AttributeValue]] = {
89 "litellm.endpoint.category": category.value,
90 "http.route": route,
91 "http.response.status_code": status_code,
92 }
93 model_attr: Final[dict[str, AttributeValue]] = {"litellm.model_id": model_id} if model_id else {}
94 return {**base, **model_attr}
97def build_mtls_meter_provider(config: BillingMetricsConfig) -> MeterProvider:
98 """OTLP/HTTP exporter presenting a TLS client certificate.
100 The collector's load balancer terminates mutual TLS and validates the client
101 certificate against our CA. Server verification uses the system trust store
102 (the collector presents a public web-PKI certificate); ca_cert_path overrides
103 it only for private/test collectors.
104 """
105 exporter: Final = OTLPMetricExporter(
106 endpoint=_metrics_endpoint(config.endpoint),
107 # None -> exporter falls back to the system trust store.
108 certificate_file=config.ca_cert_path,
109 client_certificate_file=config.client_cert_path,
110 client_key_file=config.client_key_path,
111 )
112 reader: Final = PeriodicExportingMetricReader(exporter, export_interval_millis=config.export_interval_ms)
113 return MeterProvider(metric_readers=[reader], resource=Resource.create(_resource_attributes(config)))
116class BillingMetricsRecorder:
117 """Increments one OTLP counter per billable request. The meter provider is injected (see the factory)."""
119 def __init__(self, provider: MeterProvider) -> None:
120 self._provider = provider
121 self._counter: Counter = provider.get_meter(METER_NAME).create_counter(
122 name=METRIC_NAME,
123 unit="{request}",
124 description="Count of 2xx HTTP requests to billable LLM/MCP/A2A endpoints",
125 )
127 def record(self, *, category: BillableCategory, route: str, status_code: int, model_id: str | None) -> None:
128 self._counter.add(1, _billable_attributes(category, route, status_code, model_id))
130 def shutdown(self) -> None:
131 """Final flush + exporter-thread stop. Without this, up to one export
132 interval of billable counts is dropped on every proxy restart."""
133 self._provider.shutdown(timeout_millis=SHUTDOWN_FLUSH_TIMEOUT_MS)
136def _export_interval_ms() -> int:
137 raw: Final = os.getenv(EXPORT_INTERVAL_ENV)
138 if raw is None:
139 return DEFAULT_EXPORT_INTERVAL_MS
140 try:
141 return int(raw)
142 except ValueError:
143 verbose_proxy_logger.warning(
144 "Invalid %s=%r, falling back to %d ms", EXPORT_INTERVAL_ENV, raw, DEFAULT_EXPORT_INTERVAL_MS
145 )
146 return DEFAULT_EXPORT_INTERVAL_MS
149@dataclass(frozen=True, slots=True)
150class _CredentialPaths:
151 client_cert_path: str
152 client_key_path: str
153 ca_cert_path: str | None
156def _is_pem_content(value: str) -> bool:
157 return value.lstrip().startswith(_PEM_PREFIX)
160def _write_pem(directory: str, filename: str, pem: str) -> str:
161 path: Final = os.path.join(directory, filename)
162 with open(path, "w", encoding="utf-8") as handle:
163 handle.write(pem if pem.endswith("\n") else f"{pem}\n")
164 os.chmod(path, _PEM_FILE_MODE)
165 return path
168def _resolve_credential_paths(*, client_cert: str, client_key: str, ca_cert: str | None) -> _CredentialPaths:
169 """
170 Accept either a filesystem path or inline PEM content for each credential.
172 Secret stores that inject values as environment content rather than mounted
173 files (ECS tasks reading AWS Secrets Manager, Cloud Run reading Secret
174 Manager) can only deliver the certificate as a string. The OTLP exporter
175 takes paths, so inline PEM is written to a private directory once, when the
176 recorder is built. Raises OSError if that write fails; the caller disables
177 metering rather than propagating.
178 """
179 inline: Final = tuple(value for value in (client_cert, client_key, ca_cert) if value and _is_pem_content(value))
180 if not inline:
181 return _CredentialPaths(client_cert, client_key, ca_cert)
183 # mkdtemp is 0o700, so the 0o600 key file it holds is unreachable by other users.
184 directory: Final = tempfile.mkdtemp(prefix=_PEM_DIR_PREFIX)
185 return _CredentialPaths(
186 client_cert_path=(
187 _write_pem(directory, _CLIENT_CERT_FILENAME, client_cert) if _is_pem_content(client_cert) else client_cert
188 ),
189 client_key_path=(
190 _write_pem(directory, _CLIENT_KEY_FILENAME, client_key) if _is_pem_content(client_key) else client_key
191 ),
192 ca_cert_path=(
193 _write_pem(directory, _CA_CERT_FILENAME, ca_cert) if ca_cert and _is_pem_content(ca_cert) else ca_cert
194 ),
195 )
198def load_billing_metrics_config(
199 *, license_data: Optional["EnterpriseLicenseData"], litellm_version: str
200) -> BillingMetricsConfig | None:
201 endpoint: Final = os.getenv(ENDPOINT_ENV)
202 client_cert: Final = os.getenv(CLIENT_CERT_ENV)
203 client_key: Final = os.getenv(CLIENT_KEY_ENV)
204 # Optional: only for private/test collectors whose server cert is not on the
205 # public web PKI. The production collector needs no CA override.
206 ca_cert: Final = os.getenv(CA_CERT_ENV)
208 missing: Final = [
209 name
210 for name, value in (
211 (ENDPOINT_ENV, endpoint),
212 (CLIENT_CERT_ENV, client_cert),
213 (CLIENT_KEY_ENV, client_key),
214 )
215 if not value
216 ]
217 if not endpoint or not client_cert or not client_key:
218 verbose_proxy_logger.warning(
219 "Enterprise billing metrics disabled: licensed deployment missing config (%s)",
220 ", ".join(missing),
221 )
222 return None
224 try:
225 paths: Final = _resolve_credential_paths(client_cert=client_cert, client_key=client_key, ca_cert=ca_cert)
226 except OSError as exc:
227 verbose_proxy_logger.warning(
228 "Enterprise billing metrics disabled: could not write inline certificate content to disk: %s", exc
229 )
230 return None
232 # Report the variable names, never their values. A value that is neither a
233 # readable path nor recognizable PEM is still secret material, and this
234 # warning would otherwise copy a client key straight into the proxy logs.
235 unreadable: Final = [
236 env_name
237 for env_name, path in (
238 (CLIENT_CERT_ENV, paths.client_cert_path),
239 (CLIENT_KEY_ENV, paths.client_key_path),
240 (CA_CERT_ENV, paths.ca_cert_path),
241 )
242 if path and not os.path.isfile(path)
243 ]
244 if unreadable:
245 verbose_proxy_logger.warning(
246 "Enterprise billing metrics disabled: %s did not resolve to a readable certificate file. "
247 "Set each to a file path, or to inline PEM content beginning with '%s'.",
248 ", ".join(unreadable),
249 _PEM_PREFIX,
250 )
251 return None
253 return BillingMetricsConfig(
254 endpoint=endpoint,
255 client_cert_path=paths.client_cert_path,
256 client_key_path=paths.client_key_path,
257 ca_cert_path=paths.ca_cert_path,
258 export_interval_ms=_export_interval_ms(),
259 litellm_version=litellm_version,
260 license_id=(license_data or {}).get("user_id"),
261 )
264class _ActiveRecorderRegistry:
265 """One-slot registry linking the factory-built recorder to the shutdown
266 hook; the middleware instance holding the recorder is not reachable from
267 proxy_shutdown_event."""
269 def __init__(self) -> None:
270 self._recorder: BillingMetricsRecorder | None = None
272 def set(self, recorder: BillingMetricsRecorder) -> None:
273 self._recorder = recorder
275 def pop(self) -> BillingMetricsRecorder | None:
276 recorder: Final = self._recorder
277 self._recorder = None
278 return recorder
281_ACTIVE_RECORDER: Final = _ActiveRecorderRegistry()
284def build_billing_metrics_recorder(
285 *, premium: bool, license_data: Optional["EnterpriseLicenseData"], litellm_version: str
286) -> BillingMetricsRecorder | None:
287 """Build the recorder, or None when the deployment is not licensed or metering is unconfigured."""
288 if not premium: 288 ↛ 294line 288 didn't jump to line 294 because the condition on line 288 was always true
289 # Debug, not warning: unlicensed is the common case and a warning here
290 # would be noise on every OSS proxy. Every other disable path warns.
291 verbose_proxy_logger.debug("Enterprise billing metrics disabled: deployment is not licensed")
292 return None
294 config: Final = load_billing_metrics_config(license_data=license_data, litellm_version=litellm_version)
295 if config is None:
296 return None
298 try:
299 recorder: Final = BillingMetricsRecorder(build_mtls_meter_provider(config))
300 except Exception as exc: # noqa: BLE001 -- metering must never break proxy startup
301 verbose_proxy_logger.warning("Enterprise billing metrics disabled: failed to initialize exporter: %s", exc)
302 return None
303 _ACTIVE_RECORDER.set(recorder)
304 # The only positive signal that this component meters. Without it, a silent
305 # return above is indistinguishable from a working exporter in the logs, and
306 # a component that carries the cert but no license would look healthy.
307 verbose_proxy_logger.info(
308 "Enterprise billing metrics enabled: exporting to %s every %d ms",
309 config.endpoint,
310 config.export_interval_ms,
311 )
312 return recorder
315def shutdown_billing_metrics_recorder() -> None:
316 """Flush and stop the active recorder, if any. Idempotent; never raises."""
317 recorder: Final = _ACTIVE_RECORDER.pop()
318 if recorder is None: 318 ↛ 320line 318 didn't jump to line 320 because the condition on line 318 was always true
319 return
320 try:
321 recorder.shutdown()
322 except Exception as exc: # noqa: BLE001 -- shutdown must never block or fail proxy exit
323 verbose_proxy_logger.warning("Enterprise billing metrics: final flush failed: %s", exc)