Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
03f3239cd0 | ||
|
|
d5c7440655 | ||
|
|
bff46fefe7 | ||
|
|
c93bfca6a8 | ||
|
|
44ec3502dc | ||
|
|
6608f1a8d5 | ||
|
|
c6cd41fd4b | ||
|
|
b0f608f145 | ||
|
|
cc9a32f53a |
@@ -1,4 +0,0 @@
|
||||
(apply,CacheStats{hitCount=337, missCount=199, loadSuccessCount=199, loadExceptionCount=0, totalLoadTime=581291927, evictionCount=0})
|
||||
(tree,CacheStats{hitCount=986, missCount=352, loadSuccessCount=299, loadExceptionCount=0, totalLoadTime=821650758, evictionCount=0})
|
||||
(commit,CacheStats{hitCount=108, missCount=107, loadSuccessCount=107, loadExceptionCount=0, totalLoadTime=78983052, evictionCount=0})
|
||||
(tag,CacheStats{hitCount=0, missCount=2, loadSuccessCount=2, loadExceptionCount=0, totalLoadTime=319542, evictionCount=0})
|
||||
@@ -1,4 +0,0 @@
|
||||
e71e5b78236a67327c678490cb50b46981f19de0 bbcbb68b91e786eb71bbb0a4443d7b8a26140e1b .sops.yaml
|
||||
4189696f5581ac0ffdc125c3bf9b9f664b3ddfb0 7cd3f1ee4865c563d141464f6fc185436993b84b .sops.yaml
|
||||
635630e73152a5f22e6cbd42322ec55d79f8d9c0 297e94a89d73d18c4f47013bb0e8303f123715f3 configmap.yaml
|
||||
29e515e7b46742fab8c3fcc2189af7010a6ccc62 6869fa11f96e03f7ec76a0ea14a4ddaf604004a4 gateway-config-secret.enc.yaml
|
||||
@@ -1,12 +0,0 @@
|
||||
0a95af80c0051bacbeb8483c1632e47acd3db5be 40207e487cfb63409a976fb2a0b9e1e62c8b1513
|
||||
27428d910111299d0699f429190284a9ca6e50b7 3318daf758349402aef43b095482743ab96b37f9
|
||||
329a495af4c935529fdae17229314101c0c77876 67f24ea76359c8dba4b56267790aad76bbc58464
|
||||
4c8bc6c920b6b75399555827022f69ef0c4f7d15 1fa839b41975fa3f0ac9052355ffb625f5a8f324
|
||||
528545f414c83217408edfea234dcd1f3edee0c2 b8f95506ca1545b876b5531cd385172e9ca5b4b0
|
||||
81038e1cf7567a9133d7c233a97b1e2f19fa1c82 4a00312906ba725f3968187656fde2663b1763ab
|
||||
a5b3b5c44a406896bcb414df6c6426c277715706 2ab47a9dbe5ba36dfa0e275991ef7b7656908410
|
||||
ce27643667a0399115cd1f2b6d38123fdcf2b4f1 6ff0a50de8efbad105fa588245f22fdb26afddc4
|
||||
d49756886a46542b38533b913a1f776b5145f5ec d82cc5a6970a1fb32e21dda9a737b987a8668111
|
||||
db3a30fbcf1f139c667fb68a91762582c49b8cee 04619a269fed9eeea53ab4d4d73131e3713f40a0
|
||||
eb54715e4dec0fb35402576fcc224a09808b00c1 d53b7632cf9646dda1c978a5f94615dc9eaed5e8
|
||||
ef72b5bbccf2df89aa1c86dee29311c63f33bf62 ba55d184fefef1a73a50409ca4fb1f7b27f5b075
|
||||
@@ -571,42 +571,128 @@ curl -X GET https://api.riotpiao.com/ \
|
||||
|
||||
## Authentication
|
||||
|
||||
All operations except `/healthz` and `/readyz` require JWT authentication.
|
||||
|
||||
### Bearer Token (JWT)
|
||||
|
||||
All operations except `/healthz` and `/readyz` require authentication.
|
||||
Provide JWT in Authorization header:
|
||||
|
||||
```bash
|
||||
curl -H 'Authorization: Bearer <jwt-token>' \
|
||||
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
|
||||
|
||||
**Via Authentik OIDC (human login):**
|
||||
#### User Login (OIDC Device Code Flow)
|
||||
|
||||
```bash
|
||||
core auth login --username [email protected]
|
||||
```
|
||||
export USER_TOKEN=$(cat ~/.cache/talos/authentik_id_token)
|
||||
|
||||
**Via service account (programmatic):**
|
||||
```bash
|
||||
core mwinit login --username service-account --password secret
|
||||
export RIOTPIAO_TOKEN=$(cat ~/.talos/.riotpiao-auth)
|
||||
|
||||
curl -H "Authorization: Bearer $RIOTPIAO_TOKEN" \
|
||||
curl -H "Authorization: Bearer $USER_TOKEN" \
|
||||
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)
|
||||
|
||||
Tokens embed capabilities in claims. Required capabilities:
|
||||
JWT claims contain permission arrays. Required capabilities:
|
||||
|
||||
- `llm:inference` — `/v1/*` chat/embeddings/rerank
|
||||
- `workflow:execute` — `/workflow` operations
|
||||
- `memory:read` — Memory queries
|
||||
- `memory:write` — Memory ingest
|
||||
- `sqs:access` — Queue operations
|
||||
- `s3:access` — S3 operations
|
||||
- `iam:admin` — IAM management
|
||||
| Capability | Used For |
|
||||
|------------|----------|
|
||||
| `llm:inference` | `/v1/chat/completions`, `/v1/embeddings`, `/v1/rerank` |
|
||||
| `workflow:execute` | `/workflow` (Temporal operations) |
|
||||
| `memory:read` | `/memory` query operations |
|
||||
| `memory:write` | `/memory` ingest operations |
|
||||
| `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).
|
||||
|
||||
---
|
||||
|
||||
|
||||
+1
-1
@@ -76,7 +76,7 @@ func main() {
|
||||
// Add workflow service adapter (uses Temporal handler for gRPC forwarding)
|
||||
workflowSpec := serviceadapter.GetWorkflowSpec()
|
||||
workflowAdapter := &serviceadapter.ServiceAdapter{
|
||||
Namespace: "api",
|
||||
Namespace: "temporal",
|
||||
ServiceName: "workflow",
|
||||
Spec: *workflowSpec,
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"net/http"
|
||||
"net/smtp"
|
||||
"os"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// SendMsgRequest represents a sendMsg API request.
|
||||
@@ -78,6 +79,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
resp = h.sendEmail(req)
|
||||
case "sms":
|
||||
resp = h.sendSMS(req)
|
||||
case "gotify":
|
||||
resp = h.sendGotify(req)
|
||||
default:
|
||||
resp = SendMsgResponse{
|
||||
Status: "error",
|
||||
@@ -158,3 +161,72 @@ func (h *Handler) sendSMS(req SendMsgRequest) SendMsgResponse {
|
||||
Error: "SMS not implemented yet",
|
||||
}
|
||||
}
|
||||
|
||||
// sendGotify sends a notification via Gotify server.
|
||||
func (h *Handler) sendGotify(req SendMsgRequest) SendMsgResponse {
|
||||
if h.gotifyURL == "" || h.gotifyToken == "" {
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "Gotify not configured (missing GOTIFY_URL or GOTIFY_TOKEN)",
|
||||
}
|
||||
}
|
||||
|
||||
// Construct Gotify message
|
||||
gotifyReq := map[string]interface{}{
|
||||
"title": req.Title,
|
||||
"message": req.Message,
|
||||
"priority": req.Priority,
|
||||
}
|
||||
|
||||
// Marshal to JSON
|
||||
body, err := json.Marshal(gotifyReq)
|
||||
if err != nil {
|
||||
log.Printf("error marshaling Gotify request: %v", err)
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "failed to marshal request: " + err.Error(),
|
||||
}
|
||||
}
|
||||
|
||||
// POST to Gotify
|
||||
gotifyEndpoint := fmt.Sprintf("%s/message?token=%s", h.gotifyURL, h.gotifyToken)
|
||||
resp, err := http.Post(gotifyEndpoint, "application/json", strings.NewReader(string(body)))
|
||||
if err != nil {
|
||||
log.Printf("error sending to Gotify: %v", err)
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "failed to send to Gotify: " + err.Error(),
|
||||
}
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// Check response status
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
|
||||
log.Printf("Gotify returned status %d", resp.StatusCode)
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: fmt.Sprintf("Gotify returned status %d", resp.StatusCode),
|
||||
}
|
||||
}
|
||||
|
||||
// Parse response
|
||||
var gotifyResp map[string]interface{}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&gotifyResp); err != nil {
|
||||
log.Printf("error decoding Gotify response: %v", err)
|
||||
return SendMsgResponse{
|
||||
Status: "success",
|
||||
MessageID: "gotify-sent",
|
||||
}
|
||||
}
|
||||
|
||||
// Extract message ID if available
|
||||
msgID := "gotify-sent"
|
||||
if id, ok := gotifyResp["id"]; ok {
|
||||
msgID = fmt.Sprintf("gotify-%v", id)
|
||||
}
|
||||
|
||||
return SendMsgResponse{
|
||||
Status: "success",
|
||||
MessageID: msgID,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
streamingResponsesTotal 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.
|
||||
@@ -41,6 +49,9 @@ func NewMetrics() *Metrics {
|
||||
upstreamHealth: make(map[string]int),
|
||||
streamingResponsesTotal: 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,
|
||||
"streaming_responses_total": m.streamingResponsesTotal,
|
||||
"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).
|
||||
func (m *Metrics) Reset() {
|
||||
m.mu.Lock()
|
||||
@@ -162,4 +277,7 @@ func (m *Metrics) Reset() {
|
||||
m.upstreamHealth = make(map[string]int)
|
||||
m.streamingResponsesTotal = 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,198 @@
|
||||
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)
|
||||
|
||||
// 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"
|
||||
"net/http"
|
||||
"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")
|
||||
}
|
||||
|
||||
|
||||
@@ -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