mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
feat: expose proxy OTEL metrics and Langfuse status
Wire the proxy's operational metrics facade into the new observability layer, expand built-in Prometheus export, surface OTEL and Langfuse status in /stats, and document the split between anonymous telemetry, OTEL metrics, and Langfuse traces. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
JerrettDavis
parent
ebd482c0a0
commit
be6aa14110
9 changed files with 657 additions and 96 deletions
127
docs/metrics.md
127
docs/metrics.md
|
|
@ -145,28 +145,113 @@ curl http://localhost:8787/metrics
|
|||
```
|
||||
|
||||
```prometheus
|
||||
# HELP headroom_requests_total Total requests processed
|
||||
headroom_requests_total{mode="optimize"} 1234
|
||||
# HELP headroom_requests_total Total number of requests
|
||||
headroom_requests_total 1234
|
||||
|
||||
# HELP headroom_tokens_saved_total Total tokens saved
|
||||
# HELP headroom_latency_ms_count Count of observed request latencies
|
||||
headroom_latency_ms_count 1234
|
||||
|
||||
# HELP headroom_tokens_saved_total Tokens saved by optimization
|
||||
headroom_tokens_saved_total 5678900
|
||||
|
||||
# HELP headroom_compression_ratio Compression ratio histogram
|
||||
headroom_compression_ratio_bucket{le="0.5"} 890
|
||||
headroom_compression_ratio_bucket{le="0.7"} 1100
|
||||
headroom_compression_ratio_bucket{le="0.9"} 1200
|
||||
# HELP headroom_requests_by_provider Requests by provider
|
||||
headroom_requests_by_provider{provider="anthropic"} 800
|
||||
headroom_requests_by_provider{provider="openai"} 434
|
||||
|
||||
# HELP headroom_latency_seconds Request latency histogram
|
||||
headroom_latency_seconds_bucket{le="0.01"} 800
|
||||
headroom_latency_seconds_bucket{le="0.1"} 1150
|
||||
# HELP headroom_transform_timing_ms_sum Sum of transform timing in milliseconds
|
||||
headroom_transform_timing_ms_sum{transform="router"} 5123.7
|
||||
|
||||
# HELP headroom_cache_hits_total Cache hit counter
|
||||
headroom_cache_hits_total 456
|
||||
|
||||
# HELP headroom_cache_misses_total Cache miss counter
|
||||
headroom_cache_misses_total 778
|
||||
# HELP headroom_cache_write_ttl_tokens_total Provider cache write tokens by observed TTL bucket
|
||||
headroom_cache_write_ttl_tokens_total{provider="anthropic",ttl="5m"} 20000
|
||||
headroom_cache_write_ttl_tokens_total{provider="anthropic",ttl="1h"} 50000
|
||||
```
|
||||
|
||||
The built-in Prometheus endpoint exposes the proxy's in-memory operational state, including:
|
||||
|
||||
- request counters
|
||||
- token totals and savings
|
||||
- latency / overhead / TTFB summaries
|
||||
- per-provider and per-model request counts
|
||||
- per-stage pipeline timing
|
||||
- waste signal token totals
|
||||
- provider cache read/write and TTL-bucket counters
|
||||
- cache bust counters
|
||||
|
||||
### OTEL Metrics
|
||||
|
||||
Headroom now emits the same operational events through a shared OTEL metrics facade.
|
||||
|
||||
There are two integration modes:
|
||||
|
||||
1. **Ambient OTEL app setup** - if your application already configures a global OTEL meter provider, Headroom records into that provider automatically.
|
||||
2. **Headroom-managed export** - if you want the proxy to configure its own OTEL metrics exporter, install:
|
||||
|
||||
```bash
|
||||
pip install "headroom-ai[proxy,otel]"
|
||||
```
|
||||
|
||||
Then set:
|
||||
|
||||
```bash
|
||||
HEADROOM_OTEL_METRICS_ENABLED=1
|
||||
HEADROOM_OTEL_METRICS_EXPORTER=otlp_http
|
||||
HEADROOM_OTEL_METRICS_ENDPOINT=http://127.0.0.1:4318/v1/metrics
|
||||
HEADROOM_OTEL_SERVICE_NAME=headroom-proxy
|
||||
HEADROOM_OTEL_RESOURCE_ATTRIBUTES=deployment.environment=dev,service.namespace=headroom
|
||||
```
|
||||
|
||||
For local validation without a collector:
|
||||
|
||||
```bash
|
||||
HEADROOM_OTEL_METRICS_ENABLED=1
|
||||
HEADROOM_OTEL_METRICS_EXPORTER=console
|
||||
headroom proxy
|
||||
```
|
||||
|
||||
The proxy's `/stats` response now includes an `otel` block that reports whether Headroom is managing an OTEL exporter for the current process.
|
||||
|
||||
Headroom's managed OTEL exporters are intentionally scoped to Headroom's own instrumentation. If you already manage global OTEL providers in your app, keep using those and let Headroom record into the ambient providers instead of enabling `HEADROOM_OTEL_*`.
|
||||
|
||||
### OTEL Environment Variables
|
||||
|
||||
| Variable | Default | Description |
|
||||
|----------|---------|-------------|
|
||||
| `HEADROOM_OTEL_METRICS_ENABLED` | `0` | Enables Headroom-managed OTEL metric export |
|
||||
| `HEADROOM_OTEL_METRICS_EXPORTER` | `otlp_http` | Exporter type: `otlp_http` or `console` |
|
||||
| `HEADROOM_OTEL_METRICS_ENDPOINT` | unset | OTLP HTTP metrics endpoint |
|
||||
| `HEADROOM_OTEL_METRICS_HEADERS` | unset | Comma-separated `key=value` headers for OTLP export |
|
||||
| `HEADROOM_OTEL_METRICS_EXPORT_INTERVAL_MS` | `10000` | Periodic export interval in milliseconds |
|
||||
| `HEADROOM_OTEL_SERVICE_NAME` | `headroom-proxy` in proxy mode | OTEL `service.name` |
|
||||
| `HEADROOM_OTEL_RESOURCE_ATTRIBUTES` | unset | Comma-separated resource attributes |
|
||||
|
||||
### Anonymous Telemetry vs OTEL
|
||||
|
||||
Headroom has two separate systems:
|
||||
|
||||
- `HEADROOM_TELEMETRY` / `--no-telemetry` controls the privacy-preserving anonymous data-flywheel beacon and TOIN-related aggregate reporting.
|
||||
- `HEADROOM_OTEL_*` controls operational OTEL metric export.
|
||||
|
||||
They are independent by design so you can disable the anonymous beacon while keeping OTEL metrics enabled, or vice versa.
|
||||
|
||||
### Langfuse
|
||||
|
||||
Langfuse fits next to this implementation as a **trace backend**, not as a metrics backend.
|
||||
|
||||
- Headroom metrics continue to go to `/metrics` and/or your OTEL metrics exporter.
|
||||
- Langfuse receives OTLP traces for Headroom's compression pipeline.
|
||||
- Headroom's `/stats` response includes a `langfuse` block when Headroom is managing Langfuse trace export for the process.
|
||||
|
||||
Enable it with:
|
||||
|
||||
```bash
|
||||
HEADROOM_LANGFUSE_ENABLED=1
|
||||
LANGFUSE_PUBLIC_KEY=pk-lf-...
|
||||
LANGFUSE_SECRET_KEY=sk-lf-...
|
||||
LANGFUSE_BASE_URL=https://cloud.langfuse.com
|
||||
```
|
||||
|
||||
For self-hosted Langfuse, set `LANGFUSE_BASE_URL` to your instance URL.
|
||||
|
||||
### Health Check
|
||||
|
||||
```bash
|
||||
|
|
@ -293,19 +378,19 @@ Example Grafana dashboard configuration for Prometheus metrics:
|
|||
"targets": [{"expr": "headroom_tokens_saved_total"}]
|
||||
},
|
||||
{
|
||||
"title": "Compression Ratio",
|
||||
"title": "Average Request Latency (ms)",
|
||||
"type": "gauge",
|
||||
"targets": [{"expr": "histogram_quantile(0.5, headroom_compression_ratio_bucket)"}]
|
||||
"targets": [{"expr": "headroom_latency_ms_sum / clamp_min(headroom_latency_ms_count, 1)"}]
|
||||
},
|
||||
{
|
||||
"title": "Request Latency (p99)",
|
||||
"title": "Max Request Latency (ms)",
|
||||
"type": "graph",
|
||||
"targets": [{"expr": "histogram_quantile(0.99, headroom_latency_seconds_bucket)"}]
|
||||
"targets": [{"expr": "headroom_latency_ms_max"}]
|
||||
},
|
||||
{
|
||||
"title": "Cache Hit Rate",
|
||||
"title": "Provider Cache Hit Rate",
|
||||
"type": "gauge",
|
||||
"targets": [{"expr": "headroom_cache_hits_total / (headroom_cache_hits_total + headroom_cache_misses_total)"}]
|
||||
"targets": [{"expr": "headroom_provider_cache_hit_requests_total / clamp_min(headroom_provider_cache_requests_total, 1)"}]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,7 +21,29 @@ headroom proxy \
|
|||
--budget 100.0
|
||||
```
|
||||
|
||||
Telemetry is enabled by default. Opt out with `HEADROOM_TELEMETRY=off` or `headroom proxy --no-telemetry`. Downstream apps can set `HEADROOM_SDK=headroom-app` to override the anonymous telemetry `sdk` label; the default remains `proxy`.
|
||||
Anonymous aggregate telemetry is enabled by default. Opt out with `HEADROOM_TELEMETRY=off` or `headroom proxy --no-telemetry`. Downstream apps can set `HEADROOM_SDK=headroom-app` to override the anonymous telemetry `sdk` label; the default remains `proxy`.
|
||||
|
||||
Operational OTEL metrics are configured separately and are **off by default**. Install `headroom-ai[proxy,otel]` and set:
|
||||
|
||||
```bash
|
||||
HEADROOM_OTEL_METRICS_ENABLED=1
|
||||
HEADROOM_OTEL_METRICS_EXPORTER=otlp_http
|
||||
HEADROOM_OTEL_METRICS_ENDPOINT=http://127.0.0.1:4318/v1/metrics
|
||||
HEADROOM_OTEL_SERVICE_NAME=headroom-proxy
|
||||
```
|
||||
|
||||
Use `HEADROOM_OTEL_METRICS_EXPORTER=console` for local smoke testing. `HEADROOM_TELEMETRY` controls the anonymous data-flywheel beacon only; it does not disable or enable OTEL export.
|
||||
|
||||
Langfuse can be enabled alongside this OTEL path for **trace ingestion**. Langfuse does **not** ingest OTEL metrics, so Headroom keeps metrics and Langfuse traces as complementary signals:
|
||||
|
||||
```bash
|
||||
HEADROOM_LANGFUSE_ENABLED=1
|
||||
LANGFUSE_PUBLIC_KEY=pk-lf-...
|
||||
LANGFUSE_SECRET_KEY=sk-lf-...
|
||||
LANGFUSE_BASE_URL=https://cloud.langfuse.com
|
||||
```
|
||||
|
||||
When configured, Headroom emits OTLP traces for the shared compression pipeline to Langfuse while continuing to expose metrics through `/metrics` and OTEL metric exporters.
|
||||
|
||||
## Command Line Options
|
||||
|
||||
|
|
@ -174,6 +196,8 @@ curl "http://localhost:8787/stats-history?format=csv&series=monthly"
|
|||
curl http://localhost:8787/metrics
|
||||
```
|
||||
|
||||
`/metrics` remains the built-in Prometheus-formatted operational view. The proxy now also emits the same operational events through the OTEL facade when OTEL metrics are configured.
|
||||
|
||||
### LLM APIs
|
||||
|
||||
The proxy supports both Anthropic and OpenAI API formats:
|
||||
|
|
|
|||
|
|
@ -389,7 +389,7 @@ class AnthropicHandlerMixin:
|
|||
rate_key = f"{api_key[:16]}:{client_ip}" if api_key else client_ip
|
||||
allowed, wait_seconds = await self.rate_limiter.check_request(rate_key)
|
||||
if not allowed:
|
||||
await self.metrics.record_rate_limited()
|
||||
await self.metrics.record_rate_limited(provider="anthropic")
|
||||
raise HTTPException(
|
||||
status_code=429,
|
||||
detail=f"Rate limited. Retry after {wait_seconds:.1f}s",
|
||||
|
|
@ -1443,7 +1443,7 @@ class AnthropicHandlerMixin:
|
|||
)
|
||||
|
||||
except Exception as e:
|
||||
await self.metrics.record_failed()
|
||||
await self.metrics.record_failed(provider="anthropic")
|
||||
# Log full error details internally for debugging
|
||||
logger.error(f"[{request_id}] Request failed: {type(e).__name__}: {e}")
|
||||
|
||||
|
|
@ -1721,7 +1721,7 @@ class AnthropicHandlerMixin:
|
|||
)
|
||||
|
||||
except Exception as e:
|
||||
await self.metrics.record_failed()
|
||||
await self.metrics.record_failed(provider="anthropic")
|
||||
logger.error(f"[{request_id}] Batch request failed: {type(e).__name__}: {e}")
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
|
|
|
|||
|
|
@ -810,7 +810,7 @@ class BatchHandlerMixin:
|
|||
|
||||
except Exception as e:
|
||||
logger.error(f"[{request_id}] Batch creation failed: {type(e).__name__}: {e}")
|
||||
await self.metrics.record_failed()
|
||||
await self.metrics.record_failed(provider="batch")
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content={
|
||||
|
|
|
|||
|
|
@ -175,7 +175,7 @@ class GeminiHandlerMixin:
|
|||
rate_key = headers.get("x-goog-api-key", "default")[:20]
|
||||
allowed, wait_seconds = await self.rate_limiter.check_request(rate_key)
|
||||
if not allowed:
|
||||
await self.metrics.record_rate_limited()
|
||||
await self.metrics.record_rate_limited(provider="gemini")
|
||||
raise HTTPException(
|
||||
status_code=429,
|
||||
detail=f"Rate limited. Retry after {wait_seconds:.1f}s",
|
||||
|
|
@ -406,7 +406,7 @@ class GeminiHandlerMixin:
|
|||
headers=response_headers,
|
||||
)
|
||||
except Exception as e:
|
||||
await self.metrics.record_failed()
|
||||
await self.metrics.record_failed(provider="gemini")
|
||||
logger.error(f"[{request_id}] Gemini request failed: {type(e).__name__}: {e}")
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
|
|
@ -650,7 +650,7 @@ class GeminiHandlerMixin:
|
|||
headers=response_headers,
|
||||
)
|
||||
except Exception as e:
|
||||
await self.metrics.record_failed()
|
||||
await self.metrics.record_failed(provider="gemini")
|
||||
logger.error(f"[{request_id}] Gemini countTokens failed: {type(e).__name__}: {e}")
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
|
|
|
|||
|
|
@ -213,7 +213,7 @@ class OpenAIHandlerMixin:
|
|||
rate_key = headers.get("authorization", "default")[:20]
|
||||
allowed, wait_seconds = await self.rate_limiter.check_request(rate_key)
|
||||
if not allowed:
|
||||
await self.metrics.record_rate_limited()
|
||||
await self.metrics.record_rate_limited(provider="openai")
|
||||
raise HTTPException(
|
||||
status_code=429,
|
||||
detail=f"Rate limited. Retry after {wait_seconds:.1f}s",
|
||||
|
|
@ -689,7 +689,7 @@ class OpenAIHandlerMixin:
|
|||
headers=response_headers,
|
||||
)
|
||||
except Exception as e:
|
||||
await self.metrics.record_failed()
|
||||
await self.metrics.record_failed(provider="openai")
|
||||
# Log full error details internally for debugging
|
||||
logger.error(f"[{request_id}] OpenAI request failed: {type(e).__name__}: {e}")
|
||||
# Return sanitized error message to client (don't expose internal details)
|
||||
|
|
@ -800,7 +800,7 @@ class OpenAIHandlerMixin:
|
|||
rate_key = headers.get("authorization", "default")[:20]
|
||||
allowed, wait_seconds = await self.rate_limiter.check_request(rate_key)
|
||||
if not allowed:
|
||||
await self.metrics.record_rate_limited()
|
||||
await self.metrics.record_rate_limited(provider="openai")
|
||||
raise HTTPException(
|
||||
status_code=429,
|
||||
detail=f"Rate limited. Retry after {wait_seconds:.1f}s",
|
||||
|
|
@ -950,7 +950,7 @@ class OpenAIHandlerMixin:
|
|||
headers=response_headers,
|
||||
)
|
||||
except Exception as e:
|
||||
await self.metrics.record_failed()
|
||||
await self.metrics.record_failed(provider="openai")
|
||||
logger.error(f"[{request_id}] OpenAI responses request failed: {type(e).__name__}: {e}")
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
|
|
|
|||
|
|
@ -16,13 +16,48 @@ from datetime import datetime
|
|||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from headroom.observability import HeadroomOtelMetrics
|
||||
from headroom.proxy.cost import CostTracker
|
||||
|
||||
from headroom.observability import get_otel_metrics
|
||||
from headroom.proxy.savings_tracker import SavingsTracker
|
||||
|
||||
logger = logging.getLogger("headroom.proxy")
|
||||
|
||||
|
||||
def _escape_label_value(value: str) -> str:
|
||||
return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"')
|
||||
|
||||
|
||||
def _format_labels(labels: dict[str, str] | None = None) -> str:
|
||||
if not labels:
|
||||
return ""
|
||||
|
||||
rendered = ",".join(
|
||||
f'{key}="{_escape_label_value(str(value))}"' for key, value in sorted(labels.items())
|
||||
)
|
||||
return f"{{{rendered}}}"
|
||||
|
||||
|
||||
def _append_metric(
|
||||
lines: list[str],
|
||||
*,
|
||||
name: str,
|
||||
metric_type: str,
|
||||
help_text: str,
|
||||
value: int | float,
|
||||
labels: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
lines.extend(
|
||||
[
|
||||
f"# HELP {name} {help_text}",
|
||||
f"# TYPE {name} {metric_type}",
|
||||
f"{name}{_format_labels(labels)} {value}",
|
||||
"",
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class PrometheusMetrics:
|
||||
"""Prometheus-compatible metrics."""
|
||||
|
||||
|
|
@ -30,6 +65,7 @@ class PrometheusMetrics:
|
|||
self,
|
||||
savings_tracker: SavingsTracker | None = None,
|
||||
cost_tracker: CostTracker | None = None,
|
||||
otel_metrics: HeadroomOtelMetrics | None = None,
|
||||
):
|
||||
self.requests_total = 0
|
||||
self.requests_by_provider: dict[str, int] = defaultdict(int)
|
||||
|
|
@ -115,6 +151,10 @@ class PrometheusMetrics:
|
|||
)
|
||||
|
||||
self._lock = asyncio.Lock()
|
||||
self._otel_metrics = otel_metrics
|
||||
|
||||
def _get_otel_metrics(self) -> HeadroomOtelMetrics:
|
||||
return self._otel_metrics or get_otel_metrics()
|
||||
|
||||
def _current_savings_tracker_totals(self) -> tuple[int, float]:
|
||||
total_input_tokens = self._savings_tracker_input_tokens_offset + self.tokens_input_total
|
||||
|
|
@ -260,77 +300,365 @@ class PrometheusMetrics:
|
|||
total_input_cost_usd=total_input_cost_usd,
|
||||
)
|
||||
|
||||
self._get_otel_metrics().record_proxy_request(
|
||||
provider=provider,
|
||||
model=model,
|
||||
input_tokens=input_tokens,
|
||||
output_tokens=output_tokens,
|
||||
tokens_saved=tokens_saved,
|
||||
latency_ms=latency_ms,
|
||||
cached=cached,
|
||||
overhead_ms=overhead_ms,
|
||||
ttfb_ms=ttfb_ms,
|
||||
cache_read_tokens=cache_read_tokens,
|
||||
cache_write_tokens=cache_write_tokens,
|
||||
cache_write_5m_tokens=cache_write_5m_tokens,
|
||||
cache_write_1h_tokens=cache_write_1h_tokens,
|
||||
uncached_input_tokens=uncached_input_tokens,
|
||||
)
|
||||
|
||||
async def record_cache_bust(self, tokens_lost: int) -> None:
|
||||
"""Record tokens that lost their cache discount due to compression."""
|
||||
async with self._lock:
|
||||
self.cache_bust_tokens_lost += tokens_lost
|
||||
self.cache_bust_count += 1
|
||||
self._get_otel_metrics().record_proxy_cache_bust(tokens_lost=tokens_lost)
|
||||
|
||||
async def record_rate_limited(self):
|
||||
async def record_rate_limited(self, *, provider: str | None = None, model: str | None = None):
|
||||
async with self._lock:
|
||||
self.requests_rate_limited += 1
|
||||
self._get_otel_metrics().record_proxy_rate_limited(provider=provider, model=model)
|
||||
|
||||
async def record_failed(self):
|
||||
async def record_failed(self, *, provider: str | None = None, model: str | None = None):
|
||||
async with self._lock:
|
||||
self.requests_failed += 1
|
||||
self._get_otel_metrics().record_proxy_failed(provider=provider, model=model)
|
||||
|
||||
async def export(self) -> str:
|
||||
"""Export metrics in Prometheus format."""
|
||||
async with self._lock:
|
||||
lines = [
|
||||
"# HELP headroom_requests_total Total number of requests",
|
||||
"# TYPE headroom_requests_total counter",
|
||||
f"headroom_requests_total {self.requests_total}",
|
||||
"",
|
||||
"# HELP headroom_requests_cached_total Cached request count",
|
||||
"# TYPE headroom_requests_cached_total counter",
|
||||
f"headroom_requests_cached_total {self.requests_cached}",
|
||||
"",
|
||||
"# HELP headroom_requests_rate_limited_total Rate limited requests",
|
||||
"# TYPE headroom_requests_rate_limited_total counter",
|
||||
f"headroom_requests_rate_limited_total {self.requests_rate_limited}",
|
||||
"",
|
||||
"# HELP headroom_requests_failed_total Failed requests",
|
||||
"# TYPE headroom_requests_failed_total counter",
|
||||
f"headroom_requests_failed_total {self.requests_failed}",
|
||||
"",
|
||||
"# HELP headroom_tokens_input_total Total input tokens",
|
||||
"# TYPE headroom_tokens_input_total counter",
|
||||
f"headroom_tokens_input_total {self.tokens_input_total}",
|
||||
"",
|
||||
"# HELP headroom_tokens_output_total Total output tokens",
|
||||
"# TYPE headroom_tokens_output_total counter",
|
||||
f"headroom_tokens_output_total {self.tokens_output_total}",
|
||||
"",
|
||||
"# HELP headroom_tokens_saved_total Tokens saved by optimization",
|
||||
"# TYPE headroom_tokens_saved_total counter",
|
||||
f"headroom_tokens_saved_total {self.tokens_saved_total}",
|
||||
"",
|
||||
"# HELP headroom_latency_ms_sum Sum of request latencies",
|
||||
"# TYPE headroom_latency_ms_sum counter",
|
||||
f"headroom_latency_ms_sum {self.latency_sum_ms:.2f}",
|
||||
]
|
||||
lines: list[str] = []
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_requests_total",
|
||||
metric_type="counter",
|
||||
help_text="Total number of requests",
|
||||
value=self.requests_total,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_requests_cached_total",
|
||||
metric_type="counter",
|
||||
help_text="Cached request count",
|
||||
value=self.requests_cached,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_requests_rate_limited_total",
|
||||
metric_type="counter",
|
||||
help_text="Rate limited requests",
|
||||
value=self.requests_rate_limited,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_requests_failed_total",
|
||||
metric_type="counter",
|
||||
help_text="Failed requests",
|
||||
value=self.requests_failed,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_tokens_input_total",
|
||||
metric_type="counter",
|
||||
help_text="Total input tokens",
|
||||
value=self.tokens_input_total,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_tokens_output_total",
|
||||
metric_type="counter",
|
||||
help_text="Total output tokens",
|
||||
value=self.tokens_output_total,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_tokens_saved_total",
|
||||
metric_type="counter",
|
||||
help_text="Tokens saved by optimization",
|
||||
value=self.tokens_saved_total,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_latency_ms_sum",
|
||||
metric_type="counter",
|
||||
help_text="Sum of request latencies in milliseconds",
|
||||
value=round(self.latency_sum_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_latency_ms_count",
|
||||
metric_type="counter",
|
||||
help_text="Count of observed request latencies",
|
||||
value=self.latency_count,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_latency_ms_min",
|
||||
metric_type="gauge",
|
||||
help_text="Minimum observed request latency in milliseconds",
|
||||
value=0 if self.latency_count == 0 else round(self.latency_min_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_latency_ms_max",
|
||||
metric_type="gauge",
|
||||
help_text="Maximum observed request latency in milliseconds",
|
||||
value=round(self.latency_max_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_overhead_ms_sum",
|
||||
metric_type="counter",
|
||||
help_text="Sum of Headroom processing overhead in milliseconds",
|
||||
value=round(self.overhead_sum_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_overhead_ms_count",
|
||||
metric_type="counter",
|
||||
help_text="Count of observed Headroom overhead samples",
|
||||
value=self.overhead_count,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_overhead_ms_min",
|
||||
metric_type="gauge",
|
||||
help_text="Minimum observed Headroom overhead in milliseconds",
|
||||
value=0 if self.overhead_count == 0 else round(self.overhead_min_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_overhead_ms_max",
|
||||
metric_type="gauge",
|
||||
help_text="Maximum observed Headroom overhead in milliseconds",
|
||||
value=round(self.overhead_max_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_ttfb_ms_sum",
|
||||
metric_type="counter",
|
||||
help_text="Sum of time to first byte in milliseconds",
|
||||
value=round(self.ttfb_sum_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_ttfb_ms_count",
|
||||
metric_type="counter",
|
||||
help_text="Count of observed time-to-first-byte samples",
|
||||
value=self.ttfb_count,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_ttfb_ms_min",
|
||||
metric_type="gauge",
|
||||
help_text="Minimum observed time to first byte in milliseconds",
|
||||
value=0 if self.ttfb_count == 0 else round(self.ttfb_min_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_ttfb_ms_max",
|
||||
metric_type="gauge",
|
||||
help_text="Maximum observed time to first byte in milliseconds",
|
||||
value=round(self.ttfb_max_ms, 2),
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_cache_bust_total",
|
||||
metric_type="counter",
|
||||
help_text="Requests that lost provider cache efficiency because of compression",
|
||||
value=self.cache_bust_count,
|
||||
)
|
||||
_append_metric(
|
||||
lines,
|
||||
name="headroom_cache_bust_tokens_lost_total",
|
||||
metric_type="counter",
|
||||
help_text="Tokens that lost provider cache discount because of compression",
|
||||
value=self.cache_bust_tokens_lost,
|
||||
)
|
||||
|
||||
# Per-provider metrics
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_requests_by_provider Requests by provider",
|
||||
"# TYPE headroom_requests_by_provider counter",
|
||||
]
|
||||
)
|
||||
for provider, count in self.requests_by_provider.items():
|
||||
lines.append(f'headroom_requests_by_provider{{provider="{provider}"}} {count}')
|
||||
lines.append("")
|
||||
|
||||
# Per-model metrics
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_requests_by_model Requests by model",
|
||||
"# TYPE headroom_requests_by_model counter",
|
||||
]
|
||||
)
|
||||
for model, count in self.requests_by_model.items():
|
||||
lines.append(f'headroom_requests_by_model{{model="{model}"}} {count}')
|
||||
lines.append("")
|
||||
|
||||
if self.transform_timing_sum:
|
||||
lines.extend(
|
||||
[
|
||||
"# HELP headroom_transform_timing_ms_sum Sum of transform timing in milliseconds",
|
||||
"# TYPE headroom_transform_timing_ms_sum counter",
|
||||
]
|
||||
)
|
||||
for name, total in self.transform_timing_sum.items():
|
||||
lines.append(
|
||||
f'headroom_transform_timing_ms_sum{{transform="{_escape_label_value(name)}"}} {round(total, 2)}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_transform_timing_ms_count Count of transform timing samples",
|
||||
"# TYPE headroom_transform_timing_ms_count counter",
|
||||
]
|
||||
)
|
||||
for name, count in self.transform_timing_count.items():
|
||||
lines.append(
|
||||
f'headroom_transform_timing_ms_count{{transform="{_escape_label_value(name)}"}} {count}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_transform_timing_ms_max Maximum transform timing in milliseconds",
|
||||
"# TYPE headroom_transform_timing_ms_max gauge",
|
||||
]
|
||||
)
|
||||
for name, max_value in self.transform_timing_max.items():
|
||||
lines.append(
|
||||
f'headroom_transform_timing_ms_max{{transform="{_escape_label_value(name)}"}} {round(max_value, 2)}'
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
if self.waste_signals_total:
|
||||
lines.extend(
|
||||
[
|
||||
"# HELP headroom_waste_signal_tokens_total Tokens attributed to detected waste signals",
|
||||
"# TYPE headroom_waste_signal_tokens_total counter",
|
||||
]
|
||||
)
|
||||
for signal_name, token_count in self.waste_signals_total.items():
|
||||
lines.append(
|
||||
f'headroom_waste_signal_tokens_total{{signal="{_escape_label_value(signal_name)}"}} {token_count}'
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
if self.cache_by_provider:
|
||||
lines.extend(
|
||||
[
|
||||
"# HELP headroom_cache_read_tokens_total Provider cache read tokens",
|
||||
"# TYPE headroom_cache_read_tokens_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_cache_read_tokens_total{{provider="{provider}"}} {stats["cache_read_tokens"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_cache_write_tokens_total Provider cache write tokens",
|
||||
"# TYPE headroom_cache_write_tokens_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_cache_write_tokens_total{{provider="{provider}"}} {stats["cache_write_tokens"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_cache_write_ttl_tokens_total Provider cache write tokens by observed TTL bucket",
|
||||
"# TYPE headroom_cache_write_ttl_tokens_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_cache_write_ttl_tokens_total{{provider="{provider}",ttl="5m"}} {stats["cache_write_5m_tokens"]}'
|
||||
)
|
||||
lines.append(
|
||||
f'headroom_cache_write_ttl_tokens_total{{provider="{provider}",ttl="1h"}} {stats["cache_write_1h_tokens"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_cache_write_ttl_requests_total Provider cache write requests by observed TTL bucket",
|
||||
"# TYPE headroom_cache_write_ttl_requests_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_cache_write_ttl_requests_total{{provider="{provider}",ttl="5m"}} {stats["cache_write_5m_requests"]}'
|
||||
)
|
||||
lines.append(
|
||||
f'headroom_cache_write_ttl_requests_total{{provider="{provider}",ttl="1h"}} {stats["cache_write_1h_requests"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_uncached_input_tokens_total Input tokens not served from provider cache",
|
||||
"# TYPE headroom_uncached_input_tokens_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_uncached_input_tokens_total{{provider="{provider}"}} {stats["uncached_input_tokens"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_provider_cache_requests_total Requests with provider cache observations",
|
||||
"# TYPE headroom_provider_cache_requests_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_provider_cache_requests_total{{provider="{provider}"}} {stats["requests"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_provider_cache_hit_requests_total Requests with provider cache reads",
|
||||
"# TYPE headroom_provider_cache_hit_requests_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_provider_cache_hit_requests_total{{provider="{provider}"}} {stats["hit_requests"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_provider_cache_bust_total Provider-specific cache bust count",
|
||||
"# TYPE headroom_provider_cache_bust_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_provider_cache_bust_total{{provider="{provider}"}} {stats["bust_count"]}'
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"# HELP headroom_provider_cache_bust_write_tokens_total Provider cache write tokens attributed to busts",
|
||||
"# TYPE headroom_provider_cache_bust_write_tokens_total counter",
|
||||
]
|
||||
)
|
||||
for provider, stats in self.cache_by_provider.items():
|
||||
lines.append(
|
||||
f'headroom_provider_cache_bust_write_tokens_total{{provider="{provider}"}} {stats["bust_write_tokens"]}'
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
|
|
|||
|
|
@ -78,6 +78,16 @@ from headroom.config import (
|
|||
SmartCrusherConfig,
|
||||
)
|
||||
from headroom.dashboard import get_dashboard_html
|
||||
from headroom.observability import (
|
||||
LangfuseTracingConfig,
|
||||
OTelMetricsConfig,
|
||||
configure_langfuse_tracing,
|
||||
configure_otel_metrics,
|
||||
get_langfuse_tracing_status,
|
||||
get_otel_metrics_status,
|
||||
shutdown_headroom_tracing,
|
||||
shutdown_otel_metrics,
|
||||
)
|
||||
from headroom.providers import AnthropicProvider, OpenAIProvider
|
||||
|
||||
# =============================================================================
|
||||
|
|
@ -824,7 +834,6 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
|
|||
from contextlib import asynccontextmanager
|
||||
|
||||
config = config or ProxyConfig()
|
||||
|
||||
proxy = HeadroomProxy(config)
|
||||
|
||||
# Telemetry beacon (anonymous aggregate stats).
|
||||
|
|
@ -890,32 +899,40 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
|
|||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI): # type: ignore[no-untyped-def]
|
||||
# Startup
|
||||
await proxy.startup()
|
||||
asyncio.create_task(_log_toin_stats_periodically())
|
||||
if proxy.usage_reporter:
|
||||
await proxy.usage_reporter.start(proxy)
|
||||
if proxy.traffic_learner:
|
||||
await proxy.traffic_learner.start()
|
||||
configure_otel_metrics(OTelMetricsConfig.from_env(default_service_name="headroom-proxy"))
|
||||
configure_langfuse_tracing(
|
||||
LangfuseTracingConfig.from_env(default_service_name="headroom-proxy")
|
||||
)
|
||||
|
||||
# Only start beacon if we acquire the lock (first worker wins)
|
||||
_beacon_is_owner[0] = _try_acquire_beacon_lock()
|
||||
if _beacon_is_owner[0]:
|
||||
await _beacon.start()
|
||||
else:
|
||||
logger.debug("Beacon: skipping (another worker owns the lock)")
|
||||
try:
|
||||
# Startup
|
||||
await proxy.startup()
|
||||
asyncio.create_task(_log_toin_stats_periodically())
|
||||
if proxy.usage_reporter:
|
||||
await proxy.usage_reporter.start(proxy)
|
||||
if proxy.traffic_learner:
|
||||
await proxy.traffic_learner.start()
|
||||
|
||||
yield
|
||||
# Only start beacon if we acquire the lock (first worker wins)
|
||||
_beacon_is_owner[0] = _try_acquire_beacon_lock()
|
||||
if _beacon_is_owner[0]:
|
||||
await _beacon.start()
|
||||
else:
|
||||
logger.debug("Beacon: skipping (another worker owns the lock)")
|
||||
|
||||
# Shutdown
|
||||
if _beacon_is_owner[0]:
|
||||
await _beacon.stop()
|
||||
_release_beacon_lock()
|
||||
if proxy.usage_reporter:
|
||||
await proxy.usage_reporter.stop()
|
||||
if proxy.traffic_learner:
|
||||
await proxy.traffic_learner.stop()
|
||||
await proxy.shutdown()
|
||||
yield
|
||||
finally:
|
||||
# Shutdown
|
||||
if _beacon_is_owner[0]:
|
||||
await _beacon.stop()
|
||||
_release_beacon_lock()
|
||||
if proxy.usage_reporter:
|
||||
await proxy.usage_reporter.stop()
|
||||
if proxy.traffic_learner:
|
||||
await proxy.traffic_learner.stop()
|
||||
await proxy.shutdown()
|
||||
shutdown_headroom_tracing()
|
||||
shutdown_otel_metrics()
|
||||
|
||||
app = FastAPI(
|
||||
title="Headroom Proxy",
|
||||
|
|
@ -1159,6 +1176,8 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
|
|||
"avg_compression_ratio": round(telemetry_stats.get("avg_compression_ratio", 0), 4),
|
||||
"avg_token_reduction": round(telemetry_stats.get("avg_token_reduction", 0), 4),
|
||||
},
|
||||
"otel": get_otel_metrics_status(),
|
||||
"langfuse": get_langfuse_tracing_status(),
|
||||
"feedback_loop": {
|
||||
"tools_tracked": feedback_stats.get("tools_tracked", 0),
|
||||
"total_compressions": feedback_stats.get("total_compressions", 0),
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import asyncio
|
|||
|
||||
import pytest
|
||||
|
||||
from headroom.observability import reset_headroom_tracing, reset_otel_metrics
|
||||
from headroom.proxy.cost import CostTracker, build_prefix_cache_stats
|
||||
from headroom.proxy.prometheus_metrics import PrometheusMetrics
|
||||
|
||||
|
|
@ -78,6 +79,40 @@ def test_prefix_cache_stats_include_observed_ttl_mix() -> None:
|
|||
assert stats["totals"]["observed_ttl_buckets"]["1h"]["tokens"] == 45
|
||||
|
||||
|
||||
def test_prometheus_metrics_export_includes_extended_fields() -> None:
|
||||
metrics = PrometheusMetrics()
|
||||
|
||||
asyncio.run(
|
||||
metrics.record_request(
|
||||
provider="anthropic",
|
||||
model="claude-opus-4-6",
|
||||
input_tokens=100,
|
||||
output_tokens=20,
|
||||
tokens_saved=5,
|
||||
latency_ms=12.5,
|
||||
overhead_ms=3.0,
|
||||
ttfb_ms=9.0,
|
||||
pipeline_timing={"router": 4.5},
|
||||
waste_signals={"json_bloat": 7},
|
||||
cache_read_tokens=40,
|
||||
cache_write_tokens=60,
|
||||
cache_write_5m_tokens=10,
|
||||
cache_write_1h_tokens=50,
|
||||
uncached_input_tokens=20,
|
||||
)
|
||||
)
|
||||
asyncio.run(metrics.record_cache_bust(11))
|
||||
|
||||
exported = asyncio.run(metrics.export())
|
||||
|
||||
assert "headroom_latency_ms_count 1" in exported
|
||||
assert 'headroom_transform_timing_ms_sum{transform="router"} 4.5' in exported
|
||||
assert 'headroom_waste_signal_tokens_total{signal="json_bloat"} 7' in exported
|
||||
assert 'headroom_cache_write_ttl_tokens_total{provider="anthropic",ttl="5m"} 10' in exported
|
||||
assert 'headroom_provider_cache_hit_requests_total{provider="anthropic"} 1' in exported
|
||||
assert "headroom_cache_bust_tokens_lost_total 11" in exported
|
||||
|
||||
|
||||
def test_streaming_parser_extracts_anthropic_ttl_bucket_usage() -> None:
|
||||
from headroom.proxy.server import HeadroomProxy, ProxyConfig
|
||||
|
||||
|
|
@ -145,3 +180,73 @@ def test_stats_endpoint_reports_observed_ttl_buckets() -> None:
|
|||
assert anthropic["observed_ttl_buckets"]["5m"]["tokens"] == 20
|
||||
assert anthropic["observed_ttl_buckets"]["1h"]["tokens"] == 50
|
||||
assert prefix_cache["totals"]["observed_ttl_mix"]["active_buckets"] == ["5m", "1h"]
|
||||
|
||||
|
||||
def test_stats_endpoint_reports_otel_configuration(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
pytest.importorskip("fastapi")
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from headroom.proxy.server import ProxyConfig, create_app
|
||||
|
||||
reset_otel_metrics()
|
||||
monkeypatch.setenv("HEADROOM_OTEL_METRICS_ENABLED", "1")
|
||||
monkeypatch.setenv("HEADROOM_OTEL_METRICS_EXPORTER", "console")
|
||||
|
||||
app = create_app(
|
||||
ProxyConfig(
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
log_requests=False,
|
||||
ccr_inject_tool=False,
|
||||
ccr_handle_responses=False,
|
||||
ccr_context_tracking=False,
|
||||
)
|
||||
)
|
||||
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/stats")
|
||||
|
||||
assert response.status_code == 200
|
||||
otel = response.json()["otel"]
|
||||
assert otel["configured"] is True
|
||||
assert otel["enabled"] is True
|
||||
assert otel["service_name"] == "headroom-proxy"
|
||||
assert otel["exporter"] == "console"
|
||||
|
||||
|
||||
def test_stats_endpoint_reports_langfuse_configuration(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
pytest.importorskip("fastapi")
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from headroom.proxy.server import ProxyConfig, create_app
|
||||
|
||||
reset_headroom_tracing()
|
||||
monkeypatch.setenv("HEADROOM_LANGFUSE_ENABLED", "1")
|
||||
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test")
|
||||
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test")
|
||||
monkeypatch.setenv("LANGFUSE_BASE_URL", "https://cloud.langfuse.com")
|
||||
|
||||
app = create_app(
|
||||
ProxyConfig(
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
log_requests=False,
|
||||
ccr_inject_tool=False,
|
||||
ccr_handle_responses=False,
|
||||
ccr_context_tracking=False,
|
||||
)
|
||||
)
|
||||
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/stats")
|
||||
|
||||
assert response.status_code == 200
|
||||
langfuse = response.json()["langfuse"]
|
||||
assert langfuse["configured"] is True
|
||||
assert langfuse["enabled"] is True
|
||||
assert langfuse["service_name"] == "headroom-proxy"
|
||||
assert langfuse["endpoint"] == "https://cloud.langfuse.com/api/public/otel/v1/traces"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue