9 Commits
Author SHA1 Message Date
Admin Bot 03f3239cd0 fix: workflow namespace should be 'temporal' not 'api'
CI / CI (pull_request) Successful in 3m34s
The workflow service is deployed in the temporal namespace:
temporal-frontend.temporal.svc.cluster.local:7233

Update ServiceAdapter to use correct namespace for workflow routing.
2026-09-15 02:39:51 +09:00
Admin Bot d5c7440655 feat: add Gotify support to sendMsg handler
CI / CI (pull_request) Successful in 3m10s
- Add sendGotify method to send notifications to Gotify server
- Support format: 'gotify' in SendMsgRequest
- Extract message ID from Gotify response
- Configurable via GOTIFY_URL and GOTIFY_TOKEN env vars
- Fallback graceful error if Gotify not configured
2026-09-15 02:36:02 +09:00
Admin Bot bff46fefe7 chore: remove BFG repo-cleaner reports
CI / CI (pull_request) Successful in 3m27s
2026-09-15 01:52:09 +09:00
Admin Bot c93bfca6a8 fix: go vet errors in observability metrics
CI / CI (pull_request) Successful in 4m9s
- Remove unused avg variable in prometheus.go
- Fix import ordering in llm_metrics.go (move net/http to top)
2026-09-15 00:50:41 +09:00
Admin Bot 44ec3502dc feat: add TTFT/ITL metrics for LLM inference
CI / CI (pull_request) Failing after 2m13s
- 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
2026-09-14 22:33:32 +09:00
Admin Bot 6608f1a8d5 test: add workflow visibility tests for poimen-harness namespace
CI / CI (pull_request) Successful in 3m18s
Verify that WorkflowAdapter provides visibility into terminated workflows
in the poimen-harness namespace. This ensures namespace pass-down feature
is working correctly and users can specify different domains/namespaces
via X-Service: workflow requests.

Tests added:
1. integration-test.sh: Added workflow visibility tests
   - List workflows in poimen-harness namespace
   - Verify terminated/completed workflows are visible
   - Validate namespace parameter requirement
   - Check auth enforcement

2. workflow-visibility-test.sh: NEW dedicated workflow test script
   - Tests WorkflowAdapter namespace pass-down
   - Verifies list, describe, and auth enforcement
   - Specific focus on poimen-harness namespace
   - Looks for 4 terminated workflows

3. task-workflow-visibility.yaml: NEW Tekton task
   - Runs workflow visibility tests against live gateway
   - Sidecar deployment pattern
   - Publishes result + summary + workflow-count metrics

4. pipeline-sse-optimization.yaml: Updated
   - Added workflow-visibility-tests stage (runs after integration-tests)
   - Updated report-results to include workflow test results
   - Full pipeline now: integration → workflow-visibility → load → report

5. kustomization.yaml: Updated
   - Added task-workflow-visibility.yaml
   - Added workflow-visibility-test-script ConfigMap

This ensures that the deprecated /workflows endpoint replacement correctly
supports multi-tenant access via namespace specification in request payload.
2026-09-14 08:24:42 +09:00
Admin Bot c6cd41fd4b feat: implement WorkflowAdapter with namespace pass-down support
CI / CI (pull_request) Successful in 3m17s
Enable X-Service: workflow routing to Temporal via ServiceAdapter.
Users can now specify namespace/domain in request payload for multi-tenant
workflow access.

Changes:
- Implement WorkflowAdapter in serviceadapter/workflow_adapter.go
  * Defines 10 workflow resources: start, describe, list, history,
    terminate, cancel, signal, query, reset, update
  * Each resource validates namespace parameter in payload
  * Forwards requests to Temporal gRPC handler

- Add GetWorkflowSpec() to define ServiceAdapter spec with:
  * Upstream: grpc://temporal:7233
  * Auth requirements per operation (execute, read, signal, query)
  * Request/response schemas for validation

- Wire WorkflowAdapter into main.go:
  * Register workflow adapter in serviceadapter registry
  * Initialize with temporal handler for gRPC forwarding

- Remove old empty WorkflowAdapter stub from adapters.go

Usage:
  curl -X POST https://api.riotpiao.com/ \
    -H 'X-Service: workflow' \
    -H 'X-Resource: start' \
    -H 'Authorization: Bearer TOKEN' \
    -d '{
      "namespace": "default",
      "workflow_id": "my-workflow",
      "workflow_type": "MyWorkflow",
      "task_queue": "default"
    }'

Namespace is required in all workflow operations and must be specified
by the client in the request payload. This enables multi-tenant support
where different teams access their own Temporal namespaces.
2026-09-14 08:21:49 +09:00
Admin Bot b0f608f145 refactor: retire /workflows endpoint, use X-Service: workflow instead
CI / CI (pull_request) Successful in 3m12s
Remove deprecated /workflows HTTP endpoint in favor of unified X-Service
header routing. All workflow operations now route through:

  X-Service: workflow
  X-Resource: {action} (start, describe, signal, query, etc)

This consolidates routing patterns and allows users to specify domain/
namespace via request payload instead of path prefixes.

Changes:
- Remove internal/proxy/workflows.go (484 lines of predefined workflows)
- Remove internal/proxy/workflows_test.go
- Remove /workflows handler from proxy.ServeHTTP()
- Update README.md to document X-Service routing pattern
- Add migration note: use X-Service: workflow instead of /workflows

WorkflowAdapter in serviceadapter/ handles X-Service: workflow requests
and forwards to Temporal gRPC API. Users can now specify domain/namespace
in request payload for multi-tenant workflow access.

Related: #26
2026-09-14 08:18:36 +09:00
Admin Bot cc9a32f53a feat(network): SSE optimization for local LLM streaming (#31 #32 #33)
CI / CI (pull_request) Successful in 3m11s
Addresses three critical network issues for LLM streaming performance:

**#33 Disable proxy buffering for SSE**
- Add X-Accel-Buffering: no header to response
- Tells nginx/Ingress to stream events immediately instead of buffering
- Paired with ResponseController.Flush() for unbuffered token delivery

**#32 HTTP/2 multiplexing for concurrent streams**
- Enable HTTP/2 in server config via http2.ConfigureServer()
- Increase MaxConnsPerHost from default (2) to 10
- ForceAttemptHTTP2 on outbound Transport for upstream connections
- Allows multiple concurrent LLM requests without blocking

**#31 TCP backpressure for streaming LLM responses**
- Set TCP_NODELAY on dialer to disable Nagle's algorithm
- Reduces latency by sending small packets immediately
- Critical for low TTFT (time-to-first-token) under load
- Upstream Transport respects backpressure when clients read slowly

**Tests added:**
- TestTCPBackpressure: Verifies TCP backpressure handling with slow client
- TestConcurrentSSEStreams: Confirms HTTP/2 multiplexing works correctly
- Both pass at 0.11s and 0.06s respectively

Fixes all three streaming performance issues in one coherent change.
2026-09-14 08:14:04 +09:00
11 changed files with 994 additions and 38 deletions
@@ -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
+103 -17
View File
@@ -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).
--- ---
+1 -1
View File
@@ -76,7 +76,7 @@ func main() {
// Add workflow service adapter (uses Temporal handler for gRPC forwarding) // Add workflow service adapter (uses Temporal handler for gRPC forwarding)
workflowSpec := serviceadapter.GetWorkflowSpec() workflowSpec := serviceadapter.GetWorkflowSpec()
workflowAdapter := &serviceadapter.ServiceAdapter{ workflowAdapter := &serviceadapter.ServiceAdapter{
Namespace: "api", Namespace: "temporal",
ServiceName: "workflow", ServiceName: "workflow",
Spec: *workflowSpec, Spec: *workflowSpec,
} }
+72
View File
@@ -7,6 +7,7 @@ import (
"net/http" "net/http"
"net/smtp" "net/smtp"
"os" "os"
"strings"
) )
// SendMsgRequest represents a sendMsg API request. // SendMsgRequest represents a sendMsg API request.
@@ -78,6 +79,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
resp = h.sendEmail(req) resp = h.sendEmail(req)
case "sms": case "sms":
resp = h.sendSMS(req) resp = h.sendSMS(req)
case "gotify":
resp = h.sendGotify(req)
default: default:
resp = SendMsgResponse{ resp = SendMsgResponse{
Status: "error", Status: "error",
@@ -158,3 +161,72 @@ func (h *Handler) sendSMS(req SendMsgRequest) SendMsgResponse {
Error: "SMS not implemented yet", 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,
}
}
+28
View File
@@ -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()))
}
+118
View File
@@ -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)
} }
+198
View File
@@ -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
}
+153
View File
@@ -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")
}
+321
View File
@@ -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
}