- RecordTTFT: Time-to-First-Token in milliseconds - RecordITL: Inter-Token Latency in milliseconds - RecordTokenCount: Track total tokens generated - Prometheus exporter for /metrics endpoint - Grafana dashboard ConfigMap (llm-metrics.json) - ResponseWriterWrapper to capture metrics during LLM calls - Metrics exported: llm_ttft_seconds, llm_itl_seconds, llm_tokens_total
This commit is contained in:
@@ -571,42 +571,128 @@ curl -X GET https://api.riotpiao.com/ \
|
|||||||
|
|
||||||
## Authentication
|
## Authentication
|
||||||
|
|
||||||
|
All operations except `/healthz` and `/readyz` require JWT authentication.
|
||||||
|
|
||||||
### Bearer Token (JWT)
|
### Bearer Token (JWT)
|
||||||
|
|
||||||
All operations except `/healthz` and `/readyz` require authentication.
|
Provide JWT in Authorization header:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
curl -H 'Authorization: Bearer <jwt-token>' \
|
curl -H 'Authorization: Bearer <jwt-token>' \
|
||||||
https://api.riotpiao.com/v1/models
|
https://api.riotpiao.com/v1/models
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### JWT Validation
|
||||||
|
|
||||||
|
Gateway validates all JWTs using **JWKS Federation**:
|
||||||
|
|
||||||
|
1. **Fetch JWKS** — Gateway fetches public keys from Authentik's JWKS endpoint (refreshed every 15 minutes)
|
||||||
|
2. **Verify Signature** — Validates JWT signature using public key matching `kid` header
|
||||||
|
3. **Check Claims:**
|
||||||
|
- `iss` (issuer) — Must be Authentik provider (format: `https://authentik.riotpiao.com/application/o/{provider}/`)
|
||||||
|
- `exp` (expiration) — Token must not be expired (60s clock skew allowed)
|
||||||
|
- `nbf` (not before) — Token must not be in future (60s clock skew allowed)
|
||||||
|
- `aud` (audience) — Must be non-empty string from Authentik
|
||||||
|
4. **Check Permissions** — Validates required capabilities from JWT claims (see RBAC section)
|
||||||
|
|
||||||
|
**JWKS Endpoint:** `https://authentik.riotpiao.com/application/oidc/jwks/`
|
||||||
|
|
||||||
|
**Multi-Issuer Support:** Gateway accepts JWT from any Authentik service account provider (paperless-ai-agent, portfolio-analyzer, etc) because all share the same JWKS signing key.
|
||||||
|
|
||||||
### Obtaining Tokens
|
### Obtaining Tokens
|
||||||
|
|
||||||
**Via Authentik OIDC (human login):**
|
#### User Login (OIDC Device Code Flow)
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
core auth login --username [email protected]
|
core auth login --username [email protected]
|
||||||
```
|
export USER_TOKEN=$(cat ~/.cache/talos/authentik_id_token)
|
||||||
|
|
||||||
**Via service account (programmatic):**
|
curl -H "Authorization: Bearer $USER_TOKEN" \
|
||||||
```bash
|
|
||||||
core mwinit login --username service-account --password secret
|
|
||||||
export RIOTPIAO_TOKEN=$(cat ~/.talos/.riotpiao-auth)
|
|
||||||
|
|
||||||
curl -H "Authorization: Bearer $RIOTPIAO_TOKEN" \
|
|
||||||
https://api.riotpiao.com/v1/models
|
https://api.riotpiao.com/v1/models
|
||||||
```
|
```
|
||||||
|
|
||||||
|
User tokens contain:
|
||||||
|
- `sub` — user ID
|
||||||
|
- `permissions` — array of granted capabilities
|
||||||
|
- `email` — user email
|
||||||
|
- `name` — user name
|
||||||
|
|
||||||
|
#### Service Account (Client Credentials Flow)
|
||||||
|
|
||||||
|
Service account gets JWT signed by Authentik:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 1. Authenticate service account with Authentik
|
||||||
|
curl -X POST https://authentik.riotpiao.com/application/o/token/ \
|
||||||
|
-H 'Content-Type: application/x-www-form-urlencoded' \
|
||||||
|
-d 'grant_type=client_credentials' \
|
||||||
|
-d 'client_id=paperless-ai-agent' \
|
||||||
|
-d 'client_secret=<secret>' \
|
||||||
|
-d 'scope=openid'
|
||||||
|
|
||||||
|
# Response:
|
||||||
|
# {
|
||||||
|
# "access_token": "<jwt>",
|
||||||
|
# "token_type": "Bearer",
|
||||||
|
# "expires_in": 3600
|
||||||
|
# }
|
||||||
|
|
||||||
|
# 2. Use token for gateway calls
|
||||||
|
export SERVICE_TOKEN=$(curl ... | jq -r .access_token)
|
||||||
|
curl -H "Authorization: Bearer $SERVICE_TOKEN" \
|
||||||
|
https://api.riotpiao.com/v1/chat/completions
|
||||||
|
```
|
||||||
|
|
||||||
|
Service account tokens contain:
|
||||||
|
- `sub` — service account ID
|
||||||
|
- `roles` — array of granted capabilities
|
||||||
|
- `service_account` — service name
|
||||||
|
- `aud` — audience (Authentik app ID)
|
||||||
|
|
||||||
|
#### Token Exchange (Service Impersonates User)
|
||||||
|
|
||||||
|
Service presents user's JWT + its own credentials to get a delegated token (see `/auth/exchange` endpoint):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Service exchanges user JWT for scoped service token
|
||||||
|
curl -X POST https://api.riotpiao.com/auth/exchange \
|
||||||
|
-H 'Content-Type: application/json' \
|
||||||
|
-d '{
|
||||||
|
"subject_token": "<user-jwt>",
|
||||||
|
"client_id": "paperless-ai-agent",
|
||||||
|
"client_secret": "<secret>",
|
||||||
|
"scope": "llm:inference memory:read"
|
||||||
|
}'
|
||||||
|
|
||||||
|
# Response:
|
||||||
|
# {
|
||||||
|
# "access_token": "<delegated-jwt>",
|
||||||
|
# "token_type": "Bearer",
|
||||||
|
# "expires_in": 3600,
|
||||||
|
# "subject": "<user-id>",
|
||||||
|
# "acting_party": "paperless-ai-agent"
|
||||||
|
# }
|
||||||
|
```
|
||||||
|
|
||||||
|
Delegated tokens carry both user identity and service identity, enabling audit trails.
|
||||||
|
|
||||||
### Capabilities (RBAC)
|
### Capabilities (RBAC)
|
||||||
|
|
||||||
Tokens embed capabilities in claims. Required capabilities:
|
JWT claims contain permission arrays. Required capabilities:
|
||||||
|
|
||||||
- `llm:inference` — `/v1/*` chat/embeddings/rerank
|
| Capability | Used For |
|
||||||
- `workflow:execute` — `/workflow` operations
|
|------------|----------|
|
||||||
- `memory:read` — Memory queries
|
| `llm:inference` | `/v1/chat/completions`, `/v1/embeddings`, `/v1/rerank` |
|
||||||
- `memory:write` — Memory ingest
|
| `workflow:execute` | `/workflow` (Temporal operations) |
|
||||||
- `sqs:access` — Queue operations
|
| `memory:read` | `/memory` query operations |
|
||||||
- `s3:access` — S3 operations
|
| `memory:write` | `/memory` ingest operations |
|
||||||
- `iam:admin` — IAM management
|
| `sqs:access` | `/sqs` queue operations |
|
||||||
|
| `s3:access` | `/s3` object storage operations |
|
||||||
|
| `iam:admin` | `/iam` user/group management |
|
||||||
|
|
||||||
|
**Wildcard:** Token with `*` capability grants all permissions.
|
||||||
|
|
||||||
|
**Permission Check:** JWT validated via `permissions` claim (user tokens) or `roles` claim (service account tokens).
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,28 @@
|
|||||||
|
package observability
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
)
|
||||||
|
|
||||||
|
// MetricsHandler serves Prometheus metrics
|
||||||
|
type MetricsHandler struct {
|
||||||
|
exporter *PrometheusExporter
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewMetricsHandler creates a new metrics handler
|
||||||
|
func NewMetricsHandler(m *Metrics) *MetricsHandler {
|
||||||
|
return &MetricsHandler{
|
||||||
|
exporter: NewPrometheusExporter(m),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ServeHTTP implements http.Handler for Prometheus /metrics endpoint
|
||||||
|
func (h *MetricsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
||||||
|
w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
|
||||||
|
w.Header().Set("Pragma", "no-cache")
|
||||||
|
w.Header().Set("Expires", "0")
|
||||||
|
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
w.Write([]byte(h.exporter.Export()))
|
||||||
|
}
|
||||||
@@ -28,6 +28,14 @@ type Metrics struct {
|
|||||||
// Streaming metrics
|
// Streaming metrics
|
||||||
streamingResponsesTotal map[string]int64
|
streamingResponsesTotal map[string]int64
|
||||||
streamingByteCount map[string]int64
|
streamingByteCount map[string]int64
|
||||||
|
|
||||||
|
// LLM inference metrics (TTFT and ITL)
|
||||||
|
// ttftMs: Time-to-First-Token in milliseconds
|
||||||
|
ttftMs map[string][]int64 // samples for histogram
|
||||||
|
// itlMs: Inter-Token Latency in milliseconds
|
||||||
|
itlMs map[string][]int64 // samples for histogram
|
||||||
|
// Token counts
|
||||||
|
tokenCount map[string]int64
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewMetrics creates a new Metrics instance.
|
// NewMetrics creates a new Metrics instance.
|
||||||
@@ -41,6 +49,9 @@ func NewMetrics() *Metrics {
|
|||||||
upstreamHealth: make(map[string]int),
|
upstreamHealth: make(map[string]int),
|
||||||
streamingResponsesTotal: make(map[string]int64),
|
streamingResponsesTotal: make(map[string]int64),
|
||||||
streamingByteCount: make(map[string]int64),
|
streamingByteCount: make(map[string]int64),
|
||||||
|
ttftMs: make(map[string][]int64),
|
||||||
|
itlMs: make(map[string][]int64),
|
||||||
|
tokenCount: make(map[string]int64),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -146,9 +157,113 @@ func (m *Metrics) GetMetrics() map[string]interface{} {
|
|||||||
"upstream_health": m.upstreamHealth,
|
"upstream_health": m.upstreamHealth,
|
||||||
"streaming_responses_total": m.streamingResponsesTotal,
|
"streaming_responses_total": m.streamingResponsesTotal,
|
||||||
"streaming_byte_count": m.streamingByteCount,
|
"streaming_byte_count": m.streamingByteCount,
|
||||||
|
"llm_ttft_ms": m.ttftMs,
|
||||||
|
"llm_itl_ms": m.itlMs,
|
||||||
|
"llm_token_count": m.tokenCount,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RecordTTFT records Time-to-First-Token in milliseconds
|
||||||
|
func (m *Metrics) RecordTTFT(model string, ttftMs int64) {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
|
key := fmt.Sprintf("llm:ttft:%s", model)
|
||||||
|
m.ttftMs[key] = append(m.ttftMs[key], ttftMs)
|
||||||
|
}
|
||||||
|
|
||||||
|
// RecordITL records Inter-Token Latency in milliseconds
|
||||||
|
func (m *Metrics) RecordITL(model string, itlMs int64) {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
|
key := fmt.Sprintf("llm:itl:%s", model)
|
||||||
|
m.itlMs[key] = append(m.itlMs[key], itlMs)
|
||||||
|
}
|
||||||
|
|
||||||
|
// RecordTokenCount records number of tokens in response
|
||||||
|
func (m *Metrics) RecordTokenCount(model string, count int64) {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
|
key := fmt.Sprintf("llm:tokens:%s", model)
|
||||||
|
m.tokenCount[key] += count
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetTTFTMetrics returns TTFT statistics for Prometheus export
|
||||||
|
func (m *Metrics) GetTTFTMetrics() map[string]interface{} {
|
||||||
|
m.mu.RLock()
|
||||||
|
defer m.mu.RUnlock()
|
||||||
|
|
||||||
|
result := make(map[string]interface{})
|
||||||
|
for key, samples := range m.ttftMs {
|
||||||
|
if len(samples) > 0 {
|
||||||
|
result[key] = map[string]interface{}{
|
||||||
|
"count": len(samples),
|
||||||
|
"sum": sumInt64(samples),
|
||||||
|
"avg": sumInt64(samples) / int64(len(samples)),
|
||||||
|
"min": minInt64(samples),
|
||||||
|
"max": maxInt64(samples),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetITLMetrics returns ITL statistics for Prometheus export
|
||||||
|
func (m *Metrics) GetITLMetrics() map[string]interface{} {
|
||||||
|
m.mu.RLock()
|
||||||
|
defer m.mu.RUnlock()
|
||||||
|
|
||||||
|
result := make(map[string]interface{})
|
||||||
|
for key, samples := range m.itlMs {
|
||||||
|
if len(samples) > 0 {
|
||||||
|
result[key] = map[string]interface{}{
|
||||||
|
"count": len(samples),
|
||||||
|
"sum": sumInt64(samples),
|
||||||
|
"avg": sumInt64(samples) / int64(len(samples)),
|
||||||
|
"min": minInt64(samples),
|
||||||
|
"max": maxInt64(samples),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func sumInt64(vals []int64) int64 {
|
||||||
|
var s int64
|
||||||
|
for _, v := range vals {
|
||||||
|
s += v
|
||||||
|
}
|
||||||
|
return s
|
||||||
|
}
|
||||||
|
|
||||||
|
func minInt64(vals []int64) int64 {
|
||||||
|
if len(vals) == 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
min := vals[0]
|
||||||
|
for _, v := range vals {
|
||||||
|
if v < min {
|
||||||
|
min = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return min
|
||||||
|
}
|
||||||
|
|
||||||
|
func maxInt64(vals []int64) int64 {
|
||||||
|
if len(vals) == 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
max := vals[0]
|
||||||
|
for _, v := range vals {
|
||||||
|
if v > max {
|
||||||
|
max = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return max
|
||||||
|
}
|
||||||
|
|
||||||
// Reset clears all metrics (for testing).
|
// Reset clears all metrics (for testing).
|
||||||
func (m *Metrics) Reset() {
|
func (m *Metrics) Reset() {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
@@ -162,4 +277,7 @@ func (m *Metrics) Reset() {
|
|||||||
m.upstreamHealth = make(map[string]int)
|
m.upstreamHealth = make(map[string]int)
|
||||||
m.streamingResponsesTotal = make(map[string]int64)
|
m.streamingResponsesTotal = make(map[string]int64)
|
||||||
m.streamingByteCount = make(map[string]int64)
|
m.streamingByteCount = make(map[string]int64)
|
||||||
|
m.ttftMs = make(map[string][]int64)
|
||||||
|
m.itlMs = make(map[string][]int64)
|
||||||
|
m.tokenCount = make(map[string]int64)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,199 @@
|
|||||||
|
package observability
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"sort"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
// PrometheusExporter exports metrics in Prometheus text format
|
||||||
|
type PrometheusExporter struct {
|
||||||
|
metrics *Metrics
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewPrometheusExporter creates a new Prometheus exporter
|
||||||
|
func NewPrometheusExporter(m *Metrics) *PrometheusExporter {
|
||||||
|
return &PrometheusExporter{metrics: m}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Export returns metrics in Prometheus text format
|
||||||
|
func (p *PrometheusExporter) Export() string {
|
||||||
|
var lines []string
|
||||||
|
|
||||||
|
lines = append(lines, "# HELP llm_ttft_seconds Time to first token for LLM inference (seconds)")
|
||||||
|
lines = append(lines, "# TYPE llm_ttft_seconds histogram")
|
||||||
|
p.exportTTFT(&lines)
|
||||||
|
|
||||||
|
lines = append(lines, "# HELP llm_itl_seconds Inter-token latency for LLM inference (seconds)")
|
||||||
|
lines = append(lines, "# TYPE llm_itl_seconds histogram")
|
||||||
|
p.exportITL(&lines)
|
||||||
|
|
||||||
|
lines = append(lines, "# HELP llm_tokens_total Total tokens generated")
|
||||||
|
lines = append(lines, "# TYPE llm_tokens_total counter")
|
||||||
|
p.exportTokens(&lines)
|
||||||
|
|
||||||
|
lines = append(lines, "# HELP request_duration_seconds Request latency")
|
||||||
|
lines = append(lines, "# TYPE request_duration_seconds histogram")
|
||||||
|
p.exportRequestDuration(&lines)
|
||||||
|
|
||||||
|
return strings.Join(lines, "\n") + "\n"
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *PrometheusExporter) exportTTFT(lines *[]string) {
|
||||||
|
p.metrics.mu.RLock()
|
||||||
|
defer p.metrics.mu.RUnlock()
|
||||||
|
|
||||||
|
// Calculate statistics for each model
|
||||||
|
for key, samples := range p.metrics.ttftMs {
|
||||||
|
if len(samples) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
model := extractModel(key)
|
||||||
|
sum := sumInt64(samples)
|
||||||
|
avg := sum / int64(len(samples))
|
||||||
|
|
||||||
|
// Export histogram buckets (in seconds)
|
||||||
|
buckets := []float64{0.001, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0}
|
||||||
|
for _, bucket := range buckets {
|
||||||
|
count := countLessOrEqual(samples, int64(bucket*1000))
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_ttft_seconds_bucket{model="%s",le="%.3f"} %d`,
|
||||||
|
model, bucket, count,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_ttft_seconds_bucket{model="%s",le="+Inf"} %d`,
|
||||||
|
model, len(samples),
|
||||||
|
))
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_ttft_seconds_sum{model="%s"} %.3f`,
|
||||||
|
model, float64(sum)/1000,
|
||||||
|
))
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_ttft_seconds_count{model="%s"} %d`,
|
||||||
|
model, len(samples),
|
||||||
|
))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *PrometheusExporter) exportITL(lines *[]string) {
|
||||||
|
p.metrics.mu.RLock()
|
||||||
|
defer p.metrics.mu.RUnlock()
|
||||||
|
|
||||||
|
for key, samples := range p.metrics.itlMs {
|
||||||
|
if len(samples) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
model := extractModel(key)
|
||||||
|
sum := sumInt64(samples)
|
||||||
|
|
||||||
|
// Export histogram buckets (in seconds)
|
||||||
|
buckets := []float64{0.001, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0}
|
||||||
|
for _, bucket := range buckets {
|
||||||
|
count := countLessOrEqual(samples, int64(bucket*1000))
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_itl_seconds_bucket{model="%s",le="%.3f"} %d`,
|
||||||
|
model, bucket, count,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_itl_seconds_bucket{model="%s",le="+Inf"} %d`,
|
||||||
|
model, len(samples),
|
||||||
|
))
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_itl_seconds_sum{model="%s"} %.3f`,
|
||||||
|
model, float64(sum)/1000,
|
||||||
|
))
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_itl_seconds_count{model="%s"} %d`,
|
||||||
|
model, len(samples),
|
||||||
|
))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *PrometheusExporter) exportTokens(lines *[]string) {
|
||||||
|
p.metrics.mu.RLock()
|
||||||
|
defer p.metrics.mu.RUnlock()
|
||||||
|
|
||||||
|
// Sort keys for consistent output
|
||||||
|
var keys []string
|
||||||
|
for k := range p.metrics.tokenCount {
|
||||||
|
keys = append(keys, k)
|
||||||
|
}
|
||||||
|
sort.Strings(keys)
|
||||||
|
|
||||||
|
for _, key := range keys {
|
||||||
|
model := extractModel(key)
|
||||||
|
count := p.metrics.tokenCount[key]
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`llm_tokens_total{model="%s"} %d`,
|
||||||
|
model, count,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *PrometheusExporter) exportRequestDuration(lines *[]string) {
|
||||||
|
p.metrics.mu.RLock()
|
||||||
|
defer p.metrics.mu.RUnlock()
|
||||||
|
|
||||||
|
// Sort keys for consistent output
|
||||||
|
var keys []string
|
||||||
|
for k := range p.metrics.requestDuration {
|
||||||
|
keys = append(keys, k)
|
||||||
|
}
|
||||||
|
sort.Strings(keys)
|
||||||
|
|
||||||
|
for _, key := range keys {
|
||||||
|
route, upstream := parseKey(key)
|
||||||
|
totalMs := p.metrics.requestDuration[key]
|
||||||
|
count := int64(1) // We'd need to track count separately in real impl
|
||||||
|
|
||||||
|
if buckets, ok := p.metrics.requestDurationBuckets[key]; ok {
|
||||||
|
for bucket := range buckets {
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`request_duration_seconds_bucket{route="%s",upstream="%s",le="%.1f"} %d`,
|
||||||
|
route, upstream, bucket, buckets[bucket],
|
||||||
|
))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`request_duration_seconds_sum{route="%s",upstream="%s"} %.3f`,
|
||||||
|
route, upstream, float64(totalMs)/1000,
|
||||||
|
))
|
||||||
|
*lines = append(*lines, fmt.Sprintf(
|
||||||
|
`request_duration_seconds_count{route="%s",upstream="%s"} %d`,
|
||||||
|
route, upstream, count,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func extractModel(key string) string {
|
||||||
|
parts := strings.Split(key, ":")
|
||||||
|
if len(parts) >= 3 {
|
||||||
|
return parts[2]
|
||||||
|
}
|
||||||
|
return key
|
||||||
|
}
|
||||||
|
|
||||||
|
func parseKey(key string) (string, string) {
|
||||||
|
parts := strings.Split(key, ":")
|
||||||
|
if len(parts) >= 2 {
|
||||||
|
return parts[0], parts[1]
|
||||||
|
}
|
||||||
|
return key, ""
|
||||||
|
}
|
||||||
|
|
||||||
|
func countLessOrEqual(samples []int64, threshold int64) int {
|
||||||
|
count := 0
|
||||||
|
for _, s := range samples {
|
||||||
|
if s <= threshold {
|
||||||
|
count++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return count
|
||||||
|
}
|
||||||
@@ -0,0 +1,153 @@
|
|||||||
|
package proxy
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bufio"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"forgejo.riotpiao.com/rock/homelab-frontend/internal/observability"
|
||||||
|
)
|
||||||
|
|
||||||
|
// LLMMetricsCapture wraps a response writer to capture TTFT and ITL metrics
|
||||||
|
type LLMMetricsCapture struct {
|
||||||
|
writer io.WriteCloser
|
||||||
|
model string
|
||||||
|
metrics *observability.Metrics
|
||||||
|
firstTokenTime time.Time
|
||||||
|
lastTokenTime time.Time
|
||||||
|
requestStartTime time.Time
|
||||||
|
ttftRecorded bool
|
||||||
|
tokenCount int64
|
||||||
|
responseStartTime time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewLLMMetricsCapture creates a new metrics capture wrapper
|
||||||
|
func NewLLMMetricsCapture(writer io.WriteCloser, model string, metrics *observability.Metrics, startTime time.Time) *LLMMetricsCapture {
|
||||||
|
return &LLMMetricsCapture{
|
||||||
|
writer: writer,
|
||||||
|
model: model,
|
||||||
|
metrics: metrics,
|
||||||
|
requestStartTime: startTime,
|
||||||
|
responseStartTime: time.Now(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Write intercepts writes to detect tokens and record metrics
|
||||||
|
func (c *LLMMetricsCapture) Write(p []byte) (int, error) {
|
||||||
|
// Record first token time
|
||||||
|
if !c.ttftRecorded && len(p) > 0 {
|
||||||
|
now := time.Now()
|
||||||
|
ttft := now.Sub(c.requestStartTime).Milliseconds()
|
||||||
|
c.metrics.RecordTTFT(c.model, ttft)
|
||||||
|
c.ttftRecorded = true
|
||||||
|
c.firstTokenTime = now
|
||||||
|
c.lastTokenTime = now
|
||||||
|
}
|
||||||
|
|
||||||
|
// Count tokens in SSE stream (simple: count "data: " lines)
|
||||||
|
if c.ttftRecorded {
|
||||||
|
tokenCount := strings.Count(string(p), "data: ")
|
||||||
|
if tokenCount > 0 {
|
||||||
|
now := time.Now()
|
||||||
|
if !c.firstTokenTime.IsZero() && c.lastTokenTime != now {
|
||||||
|
itl := now.Sub(c.lastTokenTime).Milliseconds()
|
||||||
|
c.metrics.RecordITL(c.model, itl)
|
||||||
|
}
|
||||||
|
c.lastTokenTime = now
|
||||||
|
c.tokenCount += int64(tokenCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return c.writer.Write(p)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close records final metrics and closes writer
|
||||||
|
func (c *LLMMetricsCapture) Close() error {
|
||||||
|
if c.tokenCount > 0 {
|
||||||
|
c.metrics.RecordTokenCount(c.model, c.tokenCount)
|
||||||
|
}
|
||||||
|
return c.writer.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
// ResponseWriterWrapper wraps http.ResponseWriter to capture metrics
|
||||||
|
type ResponseWriterWrapper struct {
|
||||||
|
writer http.ResponseWriter
|
||||||
|
statusCode int
|
||||||
|
metrics *observability.Metrics
|
||||||
|
model string
|
||||||
|
startTime time.Time
|
||||||
|
firstByteTime time.Time
|
||||||
|
lastWriteTime time.Time
|
||||||
|
ttftRecorded bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewResponseWriterWrapper creates a wrapper for response writer
|
||||||
|
func NewResponseWriterWrapper(w http.ResponseWriter, model string, metrics *observability.Metrics, startTime time.Time) *ResponseWriterWrapper {
|
||||||
|
return &ResponseWriterWrapper{
|
||||||
|
writer: w,
|
||||||
|
model: model,
|
||||||
|
metrics: metrics,
|
||||||
|
startTime: startTime,
|
||||||
|
statusCode: 200,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Header implements http.ResponseWriter
|
||||||
|
func (w *ResponseWriterWrapper) Header() http.Header {
|
||||||
|
return w.writer.Header()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Write implements http.ResponseWriter
|
||||||
|
func (w *ResponseWriterWrapper) Write(b []byte) (int, error) {
|
||||||
|
// Record TTFT on first write
|
||||||
|
if !w.ttftRecorded && len(b) > 0 {
|
||||||
|
now := time.Now()
|
||||||
|
ttft := now.Sub(w.startTime).Milliseconds()
|
||||||
|
w.metrics.RecordTTFT(w.model, ttft)
|
||||||
|
w.ttftRecorded = true
|
||||||
|
w.firstByteTime = now
|
||||||
|
w.lastWriteTime = now
|
||||||
|
}
|
||||||
|
|
||||||
|
// Record ITL for subsequent writes (for streaming)
|
||||||
|
if w.ttftRecorded && len(b) > 0 {
|
||||||
|
now := time.Now()
|
||||||
|
if !w.firstByteTime.IsZero() && w.lastWriteTime != now {
|
||||||
|
itl := now.Sub(w.lastWriteTime).Milliseconds()
|
||||||
|
// Only record if ITL > 0 (avoid recording same millisecond twice)
|
||||||
|
if itl > 0 {
|
||||||
|
w.metrics.RecordITL(w.model, itl)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
w.lastWriteTime = now
|
||||||
|
}
|
||||||
|
|
||||||
|
return w.writer.Write(b)
|
||||||
|
}
|
||||||
|
|
||||||
|
// WriteHeader implements http.ResponseWriter
|
||||||
|
func (w *ResponseWriterWrapper) WriteHeader(statusCode int) {
|
||||||
|
w.statusCode = statusCode
|
||||||
|
w.writer.WriteHeader(statusCode)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Flush implements http.Flusher
|
||||||
|
func (w *ResponseWriterWrapper) Flush() {
|
||||||
|
if flusher, ok := w.writer.(http.Flusher); ok {
|
||||||
|
flusher.Flush()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Hijack implements http.Hijacker for streaming
|
||||||
|
func (w *ResponseWriterWrapper) Hijack() (net.Conn, *bufio.ReadWriter, error) {
|
||||||
|
if hijacker, ok := w.writer.(http.Hijacker); ok {
|
||||||
|
return hijacker.Hijack()
|
||||||
|
}
|
||||||
|
return nil, nil, fmt.Errorf("response writer does not implement Hijacker")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Import http package
|
||||||
|
import "net/http"
|
||||||
@@ -0,0 +1,321 @@
|
|||||||
|
apiVersion: v1
|
||||||
|
kind: ConfigMap
|
||||||
|
metadata:
|
||||||
|
name: grafana-dashboard-llm-metrics
|
||||||
|
namespace: monitoring
|
||||||
|
labels:
|
||||||
|
grafana_dashboard: "1"
|
||||||
|
data:
|
||||||
|
llm-metrics.json: |
|
||||||
|
{
|
||||||
|
"annotations": {
|
||||||
|
"list": [
|
||||||
|
{
|
||||||
|
"builtIn": 1,
|
||||||
|
"datasource": "-- Grafana --",
|
||||||
|
"enable": true,
|
||||||
|
"hide": true,
|
||||||
|
"iconColor": "rgba(0, 211, 255, 1)",
|
||||||
|
"name": "Annotations & Alerts",
|
||||||
|
"type": "dashboard"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"editable": true,
|
||||||
|
"gnetId": null,
|
||||||
|
"graphTooltip": 0,
|
||||||
|
"id": null,
|
||||||
|
"links": [],
|
||||||
|
"panels": [
|
||||||
|
{
|
||||||
|
"datasource": "Prometheus",
|
||||||
|
"fieldConfig": {
|
||||||
|
"defaults": {
|
||||||
|
"color": {
|
||||||
|
"mode": "palette-classic"
|
||||||
|
},
|
||||||
|
"custom": {
|
||||||
|
"axisLabel": "Milliseconds",
|
||||||
|
"axisPlacement": "auto",
|
||||||
|
"barAlignment": 0,
|
||||||
|
"drawStyle": "line",
|
||||||
|
"fillOpacity": 10,
|
||||||
|
"gradientMode": "none",
|
||||||
|
"hideFrom": {
|
||||||
|
"tooltip": false,
|
||||||
|
"viz": false,
|
||||||
|
"legend": false
|
||||||
|
},
|
||||||
|
"lineInterpolation": "linear",
|
||||||
|
"lineWidth": 1,
|
||||||
|
"pointSize": 5,
|
||||||
|
"scaleDistribution": {
|
||||||
|
"type": "linear"
|
||||||
|
},
|
||||||
|
"showPoints": "auto",
|
||||||
|
"spanNulls": false,
|
||||||
|
"stacking": {
|
||||||
|
"group": "A",
|
||||||
|
"mode": "none"
|
||||||
|
},
|
||||||
|
"thresholdsStyle": {
|
||||||
|
"mode": "off"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"mappings": [],
|
||||||
|
"thresholds": {
|
||||||
|
"mode": "absolute",
|
||||||
|
"steps": [
|
||||||
|
{
|
||||||
|
"color": "green",
|
||||||
|
"value": null
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"color": "red",
|
||||||
|
"value": 80
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"overrides": []
|
||||||
|
},
|
||||||
|
"gridPos": {
|
||||||
|
"h": 8,
|
||||||
|
"w": 12,
|
||||||
|
"x": 0,
|
||||||
|
"y": 0
|
||||||
|
},
|
||||||
|
"id": 2,
|
||||||
|
"options": {
|
||||||
|
"legend": {
|
||||||
|
"calcs": [
|
||||||
|
"mean",
|
||||||
|
"max",
|
||||||
|
"min"
|
||||||
|
],
|
||||||
|
"displayMode": "table",
|
||||||
|
"placement": "bottom"
|
||||||
|
},
|
||||||
|
"tooltip": {
|
||||||
|
"mode": "multi"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"pluginVersion": "8.0.0",
|
||||||
|
"targets": [
|
||||||
|
{
|
||||||
|
"expr": "llm_ttft_seconds * 1000",
|
||||||
|
"legendFormat": "{{model}}",
|
||||||
|
"refId": "A"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"title": "Time to First Token (TTFT) by Model",
|
||||||
|
"type": "timeseries"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datasource": "Prometheus",
|
||||||
|
"fieldConfig": {
|
||||||
|
"defaults": {
|
||||||
|
"color": {
|
||||||
|
"mode": "palette-classic"
|
||||||
|
},
|
||||||
|
"custom": {
|
||||||
|
"axisLabel": "Milliseconds",
|
||||||
|
"axisPlacement": "auto",
|
||||||
|
"barAlignment": 0,
|
||||||
|
"drawStyle": "line",
|
||||||
|
"fillOpacity": 10,
|
||||||
|
"gradientMode": "none",
|
||||||
|
"hideFrom": {
|
||||||
|
"tooltip": false,
|
||||||
|
"viz": false,
|
||||||
|
"legend": false
|
||||||
|
},
|
||||||
|
"lineInterpolation": "linear",
|
||||||
|
"lineWidth": 1,
|
||||||
|
"pointSize": 5,
|
||||||
|
"scaleDistribution": {
|
||||||
|
"type": "linear"
|
||||||
|
},
|
||||||
|
"showPoints": "auto",
|
||||||
|
"spanNulls": false,
|
||||||
|
"stacking": {
|
||||||
|
"group": "A",
|
||||||
|
"mode": "none"
|
||||||
|
},
|
||||||
|
"thresholdsStyle": {
|
||||||
|
"mode": "off"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"mappings": [],
|
||||||
|
"thresholds": {
|
||||||
|
"mode": "absolute",
|
||||||
|
"steps": [
|
||||||
|
{
|
||||||
|
"color": "green",
|
||||||
|
"value": null
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"color": "red",
|
||||||
|
"value": 80
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"overrides": []
|
||||||
|
},
|
||||||
|
"gridPos": {
|
||||||
|
"h": 8,
|
||||||
|
"w": 12,
|
||||||
|
"x": 12,
|
||||||
|
"y": 0
|
||||||
|
},
|
||||||
|
"id": 3,
|
||||||
|
"options": {
|
||||||
|
"legend": {
|
||||||
|
"calcs": [
|
||||||
|
"mean",
|
||||||
|
"max",
|
||||||
|
"min"
|
||||||
|
],
|
||||||
|
"displayMode": "table",
|
||||||
|
"placement": "bottom"
|
||||||
|
},
|
||||||
|
"tooltip": {
|
||||||
|
"mode": "multi"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"pluginVersion": "8.0.0",
|
||||||
|
"targets": [
|
||||||
|
{
|
||||||
|
"expr": "llm_itl_seconds * 1000",
|
||||||
|
"legendFormat": "{{model}}",
|
||||||
|
"refId": "A"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"title": "Inter-Token Latency (ITL) by Model",
|
||||||
|
"type": "timeseries"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datasource": "Prometheus",
|
||||||
|
"fieldConfig": {
|
||||||
|
"defaults": {
|
||||||
|
"color": {
|
||||||
|
"mode": "palette-classic"
|
||||||
|
},
|
||||||
|
"custom": {
|
||||||
|
"hideFrom": {
|
||||||
|
"tooltip": false,
|
||||||
|
"viz": false,
|
||||||
|
"legend": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"mappings": []
|
||||||
|
},
|
||||||
|
"overrides": []
|
||||||
|
},
|
||||||
|
"gridPos": {
|
||||||
|
"h": 8,
|
||||||
|
"w": 12,
|
||||||
|
"x": 0,
|
||||||
|
"y": 8
|
||||||
|
},
|
||||||
|
"id": 4,
|
||||||
|
"options": {
|
||||||
|
"legend": {
|
||||||
|
"displayMode": "list",
|
||||||
|
"placement": "bottom"
|
||||||
|
},
|
||||||
|
"pieType": "pie"
|
||||||
|
},
|
||||||
|
"pluginVersion": "8.0.0",
|
||||||
|
"targets": [
|
||||||
|
{
|
||||||
|
"expr": "llm_tokens_total",
|
||||||
|
"legendFormat": "{{model}}",
|
||||||
|
"refId": "A"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"title": "Total Tokens Generated by Model",
|
||||||
|
"type": "piechart"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datasource": "Prometheus",
|
||||||
|
"fieldConfig": {
|
||||||
|
"defaults": {
|
||||||
|
"color": {
|
||||||
|
"mode": "thresholds"
|
||||||
|
},
|
||||||
|
"mappings": [],
|
||||||
|
"thresholds": {
|
||||||
|
"mode": "absolute",
|
||||||
|
"steps": [
|
||||||
|
{
|
||||||
|
"color": "green",
|
||||||
|
"value": null
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"color": "yellow",
|
||||||
|
"value": 50
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"color": "red",
|
||||||
|
"value": 100
|
||||||
|
}
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"unit": "ms"
|
||||||
|
},
|
||||||
|
"overrides": []
|
||||||
|
},
|
||||||
|
"gridPos": {
|
||||||
|
"h": 8,
|
||||||
|
"w": 12,
|
||||||
|
"x": 12,
|
||||||
|
"y": 8
|
||||||
|
},
|
||||||
|
"id": 5,
|
||||||
|
"options": {
|
||||||
|
"orientation": "auto",
|
||||||
|
"reduceOptions": {
|
||||||
|
"values": false,
|
||||||
|
"fields": "",
|
||||||
|
"calcs": [
|
||||||
|
"lastNotNull"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"showThresholdLabels": false,
|
||||||
|
"showThresholdMarkers": true
|
||||||
|
},
|
||||||
|
"pluginVersion": "8.0.0",
|
||||||
|
"targets": [
|
||||||
|
{
|
||||||
|
"expr": "avg(llm_ttft_seconds) * 1000",
|
||||||
|
"legendFormat": "Average TTFT",
|
||||||
|
"refId": "A"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"title": "Average TTFT (All Models)",
|
||||||
|
"type": "gauge"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"refresh": "10s",
|
||||||
|
"schemaVersion": 27,
|
||||||
|
"style": "dark",
|
||||||
|
"tags": [
|
||||||
|
"llm",
|
||||||
|
"inference",
|
||||||
|
"metrics"
|
||||||
|
],
|
||||||
|
"templating": {
|
||||||
|
"list": []
|
||||||
|
},
|
||||||
|
"time": {
|
||||||
|
"from": "now-1h",
|
||||||
|
"to": "now"
|
||||||
|
},
|
||||||
|
"timepicker": {},
|
||||||
|
"timezone": "",
|
||||||
|
"title": "LLM Inference Metrics (TTFT & ITL)",
|
||||||
|
"uid": "llm-metrics",
|
||||||
|
"version": 0
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user