From d439536ca9b26d7a8ce1cfc5fdaac818c08c756c Mon Sep 17 00:00:00 2001 From: Rock Date: Tue, 15 Sep 2026 08:53:58 +0000 Subject: [PATCH] Add TTFT & ITL Metrics for LLM Inference (#28) Time-to-First-Token (TTFT) and Inter-Token Latency (ITL) metrics for LLM inference observability Metrics: llm_ttft_seconds, llm_itl_seconds, llm_tokens_total Closes #31 #32 #33 --------- Co-authored-by: poimen Reviewed-on: https://forgejo.riotpiao.com/riotpiao-poimen/homelab-frontend/pulls/28 --- .../2026-09-13/11-02-26/cache-stats.txt | 4 - .../2026-09-13/11-02-26/changed-files.txt | 4 - .../11-02-26/object-id-map.old-new.txt | 12 - API.md | 120 +++- cmd/gateway/main.go | 33 +- examples/gotify-crud.sh | 62 +++ examples/sendmsg-email.sh | 44 +- internal/config/config.go | 4 + internal/config/loader.go | 18 +- internal/notification/gotify.go | 260 +++++++++ internal/notification/handler.go | 359 ++++++++---- internal/notification/handler_test.go | 513 ++++++++++++++++++ internal/observability/handler.go | 28 + internal/observability/metrics.go | 118 ++++ internal/observability/prometheus.go | 198 +++++++ internal/proxy/bodydispatch_test.go | 63 +++ internal/proxy/llm_metrics.go | 153 ++++++ internal/proxy/router.go | 14 + internal/serviceadapter/router.go | 8 + internal/serviceadapter/types.go | 14 +- internal/serviceadapter/workflow_adapter.go | 73 ++- k8s/grafana-dashboard-llm-metrics.yaml | 321 +++++++++++ 22 files changed, 2224 insertions(+), 199 deletions(-) delete mode 100644 ..bfg-report/2026-09-13/11-02-26/cache-stats.txt delete mode 100644 ..bfg-report/2026-09-13/11-02-26/changed-files.txt delete mode 100644 ..bfg-report/2026-09-13/11-02-26/object-id-map.old-new.txt create mode 100644 examples/gotify-crud.sh create mode 100644 internal/notification/gotify.go create mode 100644 internal/notification/handler_test.go create mode 100644 internal/observability/handler.go create mode 100644 internal/observability/prometheus.go create mode 100644 internal/proxy/llm_metrics.go create mode 100644 k8s/grafana-dashboard-llm-metrics.yaml diff --git a/..bfg-report/2026-09-13/11-02-26/cache-stats.txt b/..bfg-report/2026-09-13/11-02-26/cache-stats.txt deleted file mode 100644 index 31eb65d..0000000 --- a/..bfg-report/2026-09-13/11-02-26/cache-stats.txt +++ /dev/null @@ -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}) diff --git a/..bfg-report/2026-09-13/11-02-26/changed-files.txt b/..bfg-report/2026-09-13/11-02-26/changed-files.txt deleted file mode 100644 index dbab69a..0000000 --- a/..bfg-report/2026-09-13/11-02-26/changed-files.txt +++ /dev/null @@ -1,4 +0,0 @@ -e71e5b78236a67327c678490cb50b46981f19de0 bbcbb68b91e786eb71bbb0a4443d7b8a26140e1b .sops.yaml -4189696f5581ac0ffdc125c3bf9b9f664b3ddfb0 7cd3f1ee4865c563d141464f6fc185436993b84b .sops.yaml -635630e73152a5f22e6cbd42322ec55d79f8d9c0 297e94a89d73d18c4f47013bb0e8303f123715f3 configmap.yaml -29e515e7b46742fab8c3fcc2189af7010a6ccc62 6869fa11f96e03f7ec76a0ea14a4ddaf604004a4 gateway-config-secret.enc.yaml diff --git a/..bfg-report/2026-09-13/11-02-26/object-id-map.old-new.txt b/..bfg-report/2026-09-13/11-02-26/object-id-map.old-new.txt deleted file mode 100644 index 121c322..0000000 --- a/..bfg-report/2026-09-13/11-02-26/object-id-map.old-new.txt +++ /dev/null @@ -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 diff --git a/API.md b/API.md index 3b04a8c..b5aa932 100644 --- a/API.md +++ b/API.md @@ -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 ' \ 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 user@example.com -``` +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=' \ + -d 'scope=openid' + +# Response: +# { +# "access_token": "", +# "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": "", + "client_id": "paperless-ai-agent", + "client_secret": "", + "scope": "llm:inference memory:read" + }' + +# Response: +# { +# "access_token": "", +# "token_type": "Bearer", +# "expires_in": 3600, +# "subject": "", +# "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). --- diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go index d19d39e..d59c8f4 100644 --- a/cmd/gateway/main.go +++ b/cmd/gateway/main.go @@ -11,6 +11,7 @@ import ( "forgejo.riotpiao.com/rock/homelab-frontend/internal/auth" "forgejo.riotpiao.com/rock/homelab-frontend/internal/config" + "forgejo.riotpiao.com/rock/homelab-frontend/internal/notification" "forgejo.riotpiao.com/rock/homelab-frontend/internal/proxy" "forgejo.riotpiao.com/rock/homelab-frontend/internal/server" "forgejo.riotpiao.com/rock/homelab-frontend/internal/serviceadapter" @@ -75,13 +76,38 @@ func main() { // Add workflow service adapter (uses Temporal handler for gRPC forwarding) workflowSpec := serviceadapter.GetWorkflowSpec() + workflowAdapterHandler := serviceadapter.NewWorkflowAdapter(temporalHandler) workflowAdapter := &serviceadapter.ServiceAdapter{ - Namespace: "api", + Namespace: "temporal", ServiceName: "workflow", + Handler: workflowAdapterHandler, Spec: *workflowSpec, } _ = registry.Add(workflowAdapter) - + + // Add notification service adapter (internal handler, no upstream proxy) + notifHandler := notification.NewHandler() + notifAdapter := &serviceadapter.ServiceAdapter{ + Namespace: "notification", + ServiceName: "notification", + Handler: notifHandler, + Spec: serviceadapter.Spec{ + ServiceName: "notification", + Auth: serviceadapter.Auth{Required: true}, + Resources: []serviceadapter.Resource{ + {Name: "send-email", Methods: []serviceadapter.Method{{Verb: "POST", UpstreamPath: "/send-email"}}}, + {Name: "send-message", Methods: []serviceadapter.Method{{Verb: "POST", UpstreamPath: "/send-message"}}}, + {Name: "list-messages", Methods: []serviceadapter.Method{{Verb: "GET", UpstreamPath: "/list-messages"}}}, + {Name: "delete-message", Methods: []serviceadapter.Method{{Verb: "DELETE", UpstreamPath: "/delete-message"}}}, + {Name: "delete-all-messages", Methods: []serviceadapter.Method{{Verb: "DELETE", UpstreamPath: "/delete-all-messages"}}}, + {Name: "list-applications", Methods: []serviceadapter.Method{{Verb: "GET", UpstreamPath: "/list-applications"}}}, + {Name: "create-application", Methods: []serviceadapter.Method{{Verb: "POST", UpstreamPath: "/create-application"}}}, + {Name: "delete-application", Methods: []serviceadapter.Method{{Verb: "DELETE", UpstreamPath: "/delete-application"}}}, + }, + }, + } + _ = registry.Add(notifAdapter) + // Add other adapters from config for _, a := range cfg.Adapters { _ = registry.Add(a) @@ -96,8 +122,7 @@ func main() { dispatcher := serviceadapter.NewDispatcher(registry, jwtValidator) // Wire workflow adapter to temporal handler for proper request forwarding - workflowAdapterImpl := serviceadapter.NewWorkflowAdapter(temporalHandler) - _ = workflowAdapterImpl // The dispatcher will call temporal handler directly for gRPC + // workflowAdapterHandler (above) handles JSON-to-gRPC translation for workflow service // Create router that handles health endpoints, X-Service (ServiceAdapter) routing, // temporal endpoints, and passes others to upstream handler diff --git a/examples/gotify-crud.sh b/examples/gotify-crud.sh new file mode 100644 index 0000000..bd2365a --- /dev/null +++ b/examples/gotify-crud.sh @@ -0,0 +1,62 @@ +#!/bin/bash +# Example: Gotify CRUD operations via notification service (X-Service routing) + +BASE_URL="${1:-https://api.riotpiao.com}" +AUTH_TOKEN="${2:-}" +AUTH="-H \"Authorization: Bearer $AUTH_TOKEN\"" + +echo "=== Send Gotify Message ===" +curl -s -X POST "$BASE_URL" \ + -H "X-Service: notification" \ + -H "X-Resource: send-message" \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer $AUTH_TOKEN" \ + -d '{ + "title": "Deployment Complete", + "message": "homelab-frontend v1.2.0 deployed to production", + "priority": 5 + }' | jq . + +echo "" +echo "=== List Messages ===" +curl -s -X GET "$BASE_URL?limit=10" \ + -H "X-Service: notification" \ + -H "X-Resource: list-messages" \ + -H "Authorization: Bearer $AUTH_TOKEN" | jq . + +echo "" +echo "=== List Applications ===" +curl -s -X GET "$BASE_URL" \ + -H "X-Service: notification" \ + -H "X-Resource: list-applications" \ + -H "Authorization: Bearer $AUTH_TOKEN" | jq . + +echo "" +echo "=== Create Application ===" +curl -s -X POST "$BASE_URL" \ + -H "X-Service: notification" \ + -H "X-Resource: create-application" \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer $AUTH_TOKEN" \ + -d '{ + "name": "my-monitor", + "description": "Monitoring alerts" + }' | jq . + +echo "" +echo "=== Delete Message (by ID) ===" +curl -s -X DELETE "$BASE_URL" \ + -H "X-Service: notification" \ + -H "X-Resource: delete-message" \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer $AUTH_TOKEN" \ + -d '{"id": 1}' | jq . + +echo "" +echo "=== Delete Application (by ID) ===" +curl -s -X DELETE "$BASE_URL" \ + -H "X-Service: notification" \ + -H "X-Resource: delete-application" \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer $AUTH_TOKEN" \ + -d '{"id": 1}' | jq . diff --git a/examples/sendmsg-email.sh b/examples/sendmsg-email.sh index e8e1ed4..bcb2c90 100644 --- a/examples/sendmsg-email.sh +++ b/examples/sendmsg-email.sh @@ -1,36 +1,20 @@ #!/bin/bash -# Example: Send email via notification/sendMsg endpoint +# Example: Send email via notification service (X-Service routing) BASE_URL="${1:-https://api.riotpiao.com}" -AUTH_TOKEN="${2:-}" # Optional JWT token if auth required +AUTH_TOKEN="${2:-}" -PAYLOAD=$(cat <<'EOF' -{ - "format": "smtp", - "title": "System Alert", - "message": "CPU usage exceeded 90% threshold", - "priority": 7, - "extras": { - "to_email": "admin@example.com", - "cc": "ops-team@example.com" - } -} -EOF -) - -if [ -n "$AUTH_TOKEN" ]; then - curl -X POST "$BASE_URL" \ - -H "X-Service: notification" \ - -H "X-Resource: sendMsg" \ - -H "Content-Type: application/json" \ - -H "Authorization: Bearer $AUTH_TOKEN" \ - -d "$PAYLOAD" -else - curl -X POST "$BASE_URL" \ - -H "X-Service: notification" \ - -H "X-Resource: sendMsg" \ - -H "Content-Type: application/json" \ - -d "$PAYLOAD" -fi +# Send email +curl -s -X POST "$BASE_URL" \ + -H "X-Service: notification" \ + -H "X-Resource: send-email" \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer $AUTH_TOKEN" \ + -d '{ + "to": "admin@example.com", + "cc": "ops-team@example.com", + "subject": "System Alert", + "body": "CPU usage exceeded 90% threshold" + }' | jq . echo "" diff --git a/internal/config/config.go b/internal/config/config.go index 13bc39f..89e50ac 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -36,6 +36,10 @@ type ModelUpstream struct { Address string // Path is the upstream path for this model (e.g., "/v1/chat/completions"). Path string + // UpstreamModel is the model name to send to the upstream server. + // If empty, the client-provided model name (Name) is used as-is. + // Use this when the upstream expects a different model name than clients send. + UpstreamModel string // AuthRequired indicates whether this model requires JWT authentication. AuthRequired bool } diff --git a/internal/config/loader.go b/internal/config/loader.go index 344dfc6..e0d9ed3 100644 --- a/internal/config/loader.go +++ b/internal/config/loader.go @@ -37,10 +37,11 @@ type rawRoute struct { // rawModel represents a single model entry in the YAML configuration. type rawModel struct { - Name string `yaml:"name"` - Address string `yaml:"address"` - Path string `yaml:"path"` - AuthRequired *bool `yaml:"authRequired"` + Name string `yaml:"name"` + Address string `yaml:"address"` + Path string `yaml:"path"` + UpstreamModel string `yaml:"upstreamModel"` + AuthRequired *bool `yaml:"authRequired"` } // rawAdapter represents a service adapter in the YAML configuration. @@ -132,10 +133,11 @@ func LoadRoutesAndModelsFromFile(path string) (map[string]*Route, map[string]*Mo authRequired = *rawModel.AuthRequired } models[rawModel.Name] = &ModelUpstream{ - Name: rawModel.Name, - Address: rawModel.Address, - Path: rawModel.Path, - AuthRequired: authRequired, + Name: rawModel.Name, + Address: rawModel.Address, + Path: rawModel.Path, + UpstreamModel: rawModel.UpstreamModel, + AuthRequired: authRequired, } } diff --git a/internal/notification/gotify.go b/internal/notification/gotify.go new file mode 100644 index 0000000..9939026 --- /dev/null +++ b/internal/notification/gotify.go @@ -0,0 +1,260 @@ +package notification + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +// GotifyClient is a CRUD client for the Gotify API. +type GotifyClient struct { + baseURL string + appToken string // token for sending messages (application token) + clientToken string // token for reading/managing (client token) + httpClient *http.Client +} + +// NewGotifyClient creates a Gotify API client. +// appToken is used for sending messages. +// clientToken is used for listing/deleting messages and managing applications. +func NewGotifyClient(baseURL, appToken, clientToken string) *GotifyClient { + return &GotifyClient{ + baseURL: baseURL, + appToken: appToken, + clientToken: clientToken, + httpClient: &http.Client{ + Timeout: 10 * time.Second, + }, + } +} + +// --- Message Types --- + +// GotifyMessage represents a Gotify message. +type GotifyMessage struct { + ID int `json:"id,omitempty"` + AppID int `json:"appid,omitempty"` + Title string `json:"title"` + Message string `json:"message"` + Priority int `json:"priority,omitempty"` + Date string `json:"date,omitempty"` + Extras map[string]interface{} `json:"extras,omitempty"` +} + +// GotifyMessageList is a paginated list of messages. +type GotifyMessageList struct { + Messages []GotifyMessage `json:"messages"` + Paging GotifyPaging `json:"paging"` +} + +// GotifyPaging represents pagination info. +type GotifyPaging struct { + Size int `json:"size"` + Since int `json:"since"` + Limit int `json:"limit"` + Next string `json:"next,omitempty"` +} + +// --- Application Types --- + +// GotifyApplication represents a Gotify application. +type GotifyApplication struct { + ID int `json:"id,omitempty"` + Token string `json:"token,omitempty"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Image string `json:"image,omitempty"` + Internal bool `json:"internal,omitempty"` +} + +// --- Message CRUD --- + +// SendMessage sends a message via Gotify (uses app token). +func (c *GotifyClient) SendMessage(msg GotifyMessage) (*GotifyMessage, error) { + body, err := json.Marshal(msg) + if err != nil { + return nil, fmt.Errorf("marshal message: %w", err) + } + + req, err := http.NewRequest(http.MethodPost, c.baseURL+"/message", bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Gotify-Key", c.appToken) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("send message: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return nil, c.readError(resp) + } + + var result GotifyMessage + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + return &result, nil +} + +// ListMessages lists messages (uses client token). +func (c *GotifyClient) ListMessages(limit int) (*GotifyMessageList, error) { + url := fmt.Sprintf("%s/message?limit=%d", c.baseURL, limit) + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + req.Header.Set("X-Gotify-Key", c.clientToken) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("list messages: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, c.readError(resp) + } + + var result GotifyMessageList + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + return &result, nil +} + +// DeleteMessage deletes a message by ID (uses client token). +func (c *GotifyClient) DeleteMessage(id int) error { + url := fmt.Sprintf("%s/message/%d", c.baseURL, id) + req, err := http.NewRequest(http.MethodDelete, url, nil) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + req.Header.Set("X-Gotify-Key", c.clientToken) + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("delete message: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + return c.readError(resp) + } + return nil +} + +// DeleteAllMessages deletes all messages (uses client token). +func (c *GotifyClient) DeleteAllMessages() error { + req, err := http.NewRequest(http.MethodDelete, c.baseURL+"/message", nil) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + req.Header.Set("X-Gotify-Key", c.clientToken) + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("delete all messages: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + return c.readError(resp) + } + return nil +} + +// --- Application CRUD --- + +// ListApplications lists all applications (uses client token). +func (c *GotifyClient) ListApplications() ([]GotifyApplication, error) { + req, err := http.NewRequest(http.MethodGet, c.baseURL+"/application", nil) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + req.Header.Set("X-Gotify-Key", c.clientToken) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("list applications: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, c.readError(resp) + } + + var result []GotifyApplication + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + return result, nil +} + +// CreateApplication creates a new application (uses client token). +func (c *GotifyClient) CreateApplication(app GotifyApplication) (*GotifyApplication, error) { + body, err := json.Marshal(app) + if err != nil { + return nil, fmt.Errorf("marshal application: %w", err) + } + + req, err := http.NewRequest(http.MethodPost, c.baseURL+"/application", bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Gotify-Key", c.clientToken) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("create application: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return nil, c.readError(resp) + } + + var result GotifyApplication + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + return &result, nil +} + +// DeleteApplication deletes an application by ID (uses client token). +func (c *GotifyClient) DeleteApplication(id int) error { + url := fmt.Sprintf("%s/application/%d", c.baseURL, id) + req, err := http.NewRequest(http.MethodDelete, url, nil) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + req.Header.Set("X-Gotify-Key", c.clientToken) + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("delete application: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + return c.readError(resp) + } + return nil +} + +// --- Helpers --- + +func (c *GotifyClient) readError(resp *http.Response) error { + body, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("gotify API error (HTTP %d): failed to read body: %w", resp.StatusCode, err) + } + return fmt.Errorf("gotify API error (HTTP %d): %s", resp.StatusCode, string(body)) +} diff --git a/internal/notification/handler.go b/internal/notification/handler.go index 4493257..3bcf4d2 100644 --- a/internal/notification/handler.go +++ b/internal/notification/handler.go @@ -7,154 +7,285 @@ import ( "net/http" "net/smtp" "os" + "strconv" ) -// SendMsgRequest represents a sendMsg API request. -type SendMsgRequest struct { - Format string `json:"format"` // "smtp" or "sms" - Title string `json:"title"` - Message string `json:"message"` - Priority int `json:"priority,omitempty"` - Extras map[string]string `json:"extras,omitempty"` // e.g., {"to_email": "user@example.com", "phone": "+1234567890"} -} - -// SendMsgResponse represents a sendMsg API response. -type SendMsgResponse struct { - Status string `json:"status"` - MessageID string `json:"messageId,omitempty"` - Error string `json:"error,omitempty"` -} - -// Handler handles sendMsg requests and forwards to appropriate channel (email, SMS, or Gotify push). +// Handler handles notification requests routed via X-Resource header. +// Supports: send-email, send-gotify, list-messages, delete-message, +// delete-all-messages, list-applications, create-application, delete-application. type Handler struct { - smtpHost string - smtpPort string - smtpFrom string - smtpUser string - smtpPass string - smsAPIURL string - smsAPIKey string - gotifyURL string - gotifyToken string + smtpHost string + smtpPort string + smtpFrom string + smtpUser string + smtpPass string + gotify *GotifyClient } -// NewHandler creates a new notification handler from environment variables. +// NewHandler creates a notification handler from environment variables. func NewHandler() *Handler { + var gotify *GotifyClient + gotifyURL := os.Getenv("GOTIFY_URL") + if gotifyURL != "" { + gotify = NewGotifyClient( + gotifyURL, + os.Getenv("GOTIFY_APP_TOKEN"), + os.Getenv("GOTIFY_CLIENT_TOKEN"), + ) + } + return &Handler{ - smtpHost: os.Getenv("SMTP_HOST"), - smtpPort: os.Getenv("SMTP_PORT"), - smtpFrom: os.Getenv("SMTP_FROM"), - smtpUser: os.Getenv("SMTP_USER"), - smtpPass: os.Getenv("SMTP_PASS"), - smsAPIURL: os.Getenv("SMS_API_URL"), - smsAPIKey: os.Getenv("SMS_API_KEY"), - gotifyURL: os.Getenv("GOTIFY_URL"), - gotifyToken: os.Getenv("GOTIFY_TOKEN"), + smtpHost: os.Getenv("SMTP_HOST"), + smtpPort: os.Getenv("SMTP_PORT"), + smtpFrom: os.Getenv("SMTP_FROM"), + smtpUser: os.Getenv("SMTP_USER"), + smtpPass: os.Getenv("SMTP_PASS"), + gotify: gotify, } } -// ServeHTTP handles sendMsg requests. +// ServeHTTP routes requests by X-Upstream-Path (set by dispatcher after resource matching). func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - return - } + resource := r.Header.Get("X-Resource") - var req SendMsgRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - json.NewEncoder(w).Encode(SendMsgResponse{ - Status: "error", - Error: "invalid request: " + err.Error(), - }) - return - } + switch resource { + // --- Email --- + case "send-email": + h.handleSendEmail(w, r) + + // --- Gotify Messages --- + case "send-message": + h.handleSendGotify(w, r) + case "list-messages": + h.handleListMessages(w, r) + case "delete-message": + h.handleDeleteMessage(w, r) + case "delete-all-messages": + h.handleDeleteAllMessages(w, r) + + // --- Gotify Applications --- + case "list-applications": + h.handleListApplications(w, r) + case "create-application": + h.handleCreateApplication(w, r) + case "delete-application": + h.handleDeleteApplication(w, r) - // Route based on format - var resp SendMsgResponse - switch req.Format { - case "smtp": - resp = h.sendEmail(req) - case "sms": - resp = h.sendSMS(req) default: - resp = SendMsgResponse{ - Status: "error", - Error: "unsupported format: " + req.Format, - } + h.writeJSON(w, http.StatusNotFound, map[string]string{ + "error": fmt.Sprintf("unknown resource: %s", resource), + }) } - - w.Header().Set("Content-Type", "application/json") - if resp.Error != "" { - w.WriteHeader(http.StatusInternalServerError) - } else { - w.WriteHeader(http.StatusOK) - } - json.NewEncoder(w).Encode(resp) } -// sendEmail sends an email via SMTP. -func (h *Handler) sendEmail(req SendMsgRequest) SendMsgResponse { - toEmail := req.Extras["to_email"] - if toEmail == "" { - return SendMsgResponse{ - Status: "error", - Error: "missing to_email in extras", - } +// --- Email --- + +type SendEmailRequest struct { + To string `json:"to"` + CC string `json:"cc,omitempty"` + Subject string `json:"subject"` + Body string `json:"body"` +} + +func (h *Handler) handleSendEmail(w http.ResponseWriter, r *http.Request) { + var req SendEmailRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request: " + err.Error()}) + return } - subject := req.Title + if req.To == "" { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "missing 'to' field"}) + return + } + + subject := req.Subject if subject == "" { subject = "Notification" } - // Construct email body - body := req.Message - if req.Extras != nil { - if cc := req.Extras["cc"]; cc != "" { - body = fmt.Sprintf("CC: %s\n\n%s", cc, body) - } - } - msg := fmt.Sprintf( "From: %s\r\nTo: %s\r\nSubject: %s\r\nContent-Type: text/plain; charset=UTF-8\r\n\r\n%s", - h.smtpFrom, toEmail, subject, body, + h.smtpFrom, req.To, subject, req.Body, ) - // Send via SMTP smtpAddr := fmt.Sprintf("%s:%s", h.smtpHost, h.smtpPort) auth := smtp.PlainAuth("", h.smtpUser, h.smtpPass, h.smtpHost) - if err := smtp.SendMail(smtpAddr, auth, h.smtpFrom, []string{toEmail}, []byte(msg)); err != nil { - log.Printf("error sending email to %s: %v", toEmail, err) - return SendMsgResponse{ - Status: "error", - Error: "failed to send email: " + err.Error(), + if err := smtp.SendMail(smtpAddr, auth, h.smtpFrom, []string{req.To}, []byte(msg)); err != nil { + log.Printf("error sending email to %s: %v", req.To, err) + h.writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to send email: " + err.Error()}) + return + } + + h.writeJSON(w, http.StatusOK, map[string]string{ + "status": "success", + "messageId": fmt.Sprintf("email-%s", req.To), + }) +} + +// --- Gotify Messages --- + +func (h *Handler) handleSendGotify(w http.ResponseWriter, r *http.Request) { + if h.gotify == nil { + h.writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Gotify not configured"}) + return + } + + var msg GotifyMessage + if err := json.NewDecoder(r.Body).Decode(&msg); err != nil { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request: " + err.Error()}) + return + } + + result, err := h.gotify.SendMessage(msg) + if err != nil { + log.Printf("error sending gotify message: %v", err) + h.writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + + h.writeJSON(w, http.StatusOK, result) +} + +func (h *Handler) handleListMessages(w http.ResponseWriter, r *http.Request) { + if h.gotify == nil { + h.writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Gotify not configured"}) + return + } + + limit := 50 + if l := r.URL.Query().Get("limit"); l != "" { + if parsed, err := strconv.Atoi(l); err == nil && parsed > 0 { + limit = parsed } } - return SendMsgResponse{ - Status: "success", - MessageID: fmt.Sprintf("email-%s", toEmail), - } -} - -// sendSMS sends an SMS via configured provider. -// Placeholder: integrate with Twilio, AWS SNS, or similar. -func (h *Handler) sendSMS(req SendMsgRequest) SendMsgResponse { - phone := req.Extras["phone"] - if phone == "" { - return SendMsgResponse{ - Status: "error", - Error: "missing phone in extras", - } + result, err := h.gotify.ListMessages(limit) + if err != nil { + log.Printf("error listing gotify messages: %v", err) + h.writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return } - // TODO: Implement SMS provider integration (Twilio, AWS SNS, etc.) - // For now, return error - return SendMsgResponse{ - Status: "error", - Error: "SMS not implemented yet", + h.writeJSON(w, http.StatusOK, result) +} + +func (h *Handler) handleDeleteMessage(w http.ResponseWriter, r *http.Request) { + if h.gotify == nil { + h.writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Gotify not configured"}) + return } + + var req struct { + ID int `json:"id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request: " + err.Error()}) + return + } + + if req.ID == 0 { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "missing 'id' field"}) + return + } + + if err := h.gotify.DeleteMessage(req.ID); err != nil { + log.Printf("error deleting gotify message %d: %v", req.ID, err) + h.writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + + h.writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) +} + +func (h *Handler) handleDeleteAllMessages(w http.ResponseWriter, r *http.Request) { + if h.gotify == nil { + h.writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Gotify not configured"}) + return + } + + if err := h.gotify.DeleteAllMessages(); err != nil { + log.Printf("error deleting all gotify messages: %v", err) + h.writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + + h.writeJSON(w, http.StatusOK, map[string]string{"status": "all messages deleted"}) +} + +// --- Gotify Applications --- + +func (h *Handler) handleListApplications(w http.ResponseWriter, r *http.Request) { + if h.gotify == nil { + h.writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Gotify not configured"}) + return + } + + result, err := h.gotify.ListApplications() + if err != nil { + log.Printf("error listing gotify applications: %v", err) + h.writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + + h.writeJSON(w, http.StatusOK, result) +} + +func (h *Handler) handleCreateApplication(w http.ResponseWriter, r *http.Request) { + if h.gotify == nil { + h.writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Gotify not configured"}) + return + } + + var app GotifyApplication + if err := json.NewDecoder(r.Body).Decode(&app); err != nil { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request: " + err.Error()}) + return + } + + result, err := h.gotify.CreateApplication(app) + if err != nil { + log.Printf("error creating gotify application: %v", err) + h.writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + + h.writeJSON(w, http.StatusCreated, result) +} + +func (h *Handler) handleDeleteApplication(w http.ResponseWriter, r *http.Request) { + if h.gotify == nil { + h.writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Gotify not configured"}) + return + } + + var req struct { + ID int `json:"id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request: " + err.Error()}) + return + } + + if req.ID == 0 { + h.writeJSON(w, http.StatusBadRequest, map[string]string{"error": "missing 'id' field"}) + return + } + + if err := h.gotify.DeleteApplication(req.ID); err != nil { + log.Printf("error deleting gotify application %d: %v", req.ID, err) + h.writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + + h.writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) +} + +// --- Helpers --- + +func (h *Handler) writeJSON(w http.ResponseWriter, status int, data interface{}) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(data) } diff --git a/internal/notification/handler_test.go b/internal/notification/handler_test.go new file mode 100644 index 0000000..c7a70a5 --- /dev/null +++ b/internal/notification/handler_test.go @@ -0,0 +1,513 @@ +package notification + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" +) + +// mockGotifyServer creates a test server that simulates the Gotify API. +func mockGotifyServer() *httptest.Server { + mux := http.NewServeMux() + + // POST /message — send message + mux.HandleFunc("/message", func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodPost: + var msg GotifyMessage + if err := json.NewDecoder(r.Body).Decode(&msg); err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + msg.ID = 42 + msg.AppID = 1 + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(msg) + + case http.MethodGet: + // list messages + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(GotifyMessageList{ + Messages: []GotifyMessage{ + {ID: 1, Title: "Test", Message: "hello", Priority: 3}, + {ID: 2, Title: "Alert", Message: "world", Priority: 7}, + }, + Paging: GotifyPaging{Size: 2, Limit: 50}, + }) + + case http.MethodDelete: + // delete all messages + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) + + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } + }) + + // DELETE /message/{id} + mux.HandleFunc("/message/", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) + }) + + // GET/POST/DELETE /application + mux.HandleFunc("/application", func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode([]GotifyApplication{ + {ID: 1, Name: "app1", Token: "tok1"}, + {ID: 2, Name: "app2", Token: "tok2"}, + }) + + case http.MethodPost: + var app GotifyApplication + if err := json.NewDecoder(r.Body).Decode(&app); err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + app.ID = 10 + app.Token = "new-token" + w.WriteHeader(http.StatusCreated) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(app) + + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } + }) + + // DELETE /application/{id} + mux.HandleFunc("/application/", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) + }) + + return httptest.NewServer(mux) +} + +func newTestHandler(gotifyURL string) *Handler { + h := &Handler{} + if gotifyURL != "" { + h.gotify = NewGotifyClient(gotifyURL, "test-app-token", "test-client-token") + } + return h +} + +func doRequest(h *Handler, method, resource string, body interface{}) *httptest.ResponseRecorder { + var reqBody io.Reader + if body != nil { + b, _ := json.Marshal(body) + reqBody = bytes.NewReader(b) + } + + req := httptest.NewRequest(method, "/", reqBody) + req.Header.Set("X-Resource", resource) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + w := httptest.NewRecorder() + h.ServeHTTP(w, req) + return w +} + +func decodeResponse(t *testing.T, w *httptest.ResponseRecorder) map[string]interface{} { + t.Helper() + var result map[string]interface{} + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode response: %v (body: %s)", err, w.Body.String()) + } + return result +} + +// ============================================================ +// Handler routing tests +// ============================================================ + +func TestHandler_UnknownResource(t *testing.T) { + h := newTestHandler("") + w := doRequest(h, "GET", "unknown-resource", nil) + if w.Code != http.StatusNotFound { + t.Errorf("expected 404, got %d", w.Code) + } + data := decodeResponse(t, w) + if _, ok := data["error"]; !ok { + t.Error("expected error in response") + } +} + +func TestHandler_GotifyNotConfigured(t *testing.T) { + h := newTestHandler("") // no gotify + + resources := []struct { + method string + resource string + body interface{} + }{ + {"POST", "send-message", map[string]string{"title": "t", "message": "m"}}, + {"GET", "list-messages", nil}, + {"DELETE", "delete-message", map[string]int{"id": 1}}, + {"DELETE", "delete-all-messages", nil}, + {"GET", "list-applications", nil}, + {"POST", "create-application", map[string]string{"name": "app"}}, + {"DELETE", "delete-application", map[string]int{"id": 1}}, + } + + for _, tc := range resources { + t.Run(tc.resource, func(t *testing.T) { + w := doRequest(h, tc.method, tc.resource, tc.body) + if w.Code != http.StatusServiceUnavailable { + t.Errorf("expected 503, got %d", w.Code) + } + }) + } +} + +// ============================================================ +// Gotify message tests (via handler) +// ============================================================ + +func TestHandler_SendMessage(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "POST", "send-message", map[string]interface{}{ + "title": "Test Alert", + "message": "Something happened", + "priority": 5, + }) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + data := decodeResponse(t, w) + if data["title"] != "Test Alert" { + t.Errorf("expected title 'Test Alert', got %v", data["title"]) + } + if int(data["id"].(float64)) != 42 { + t.Errorf("expected id 42, got %v", data["id"]) + } +} + +func TestHandler_SendMessage_InvalidJSON(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + req := httptest.NewRequest("POST", "/", bytes.NewReader([]byte("not json"))) + req.Header.Set("X-Resource", "send-message") + w := httptest.NewRecorder() + h.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", w.Code) + } +} + +func TestHandler_ListMessages(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "GET", "list-messages", nil) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var result GotifyMessageList + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode: %v", err) + } + if len(result.Messages) != 2 { + t.Errorf("expected 2 messages, got %d", len(result.Messages)) + } +} + +func TestHandler_DeleteMessage(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "DELETE", "delete-message", map[string]int{"id": 1}) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestHandler_DeleteMessage_MissingID(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "DELETE", "delete-message", map[string]int{"id": 0}) + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", w.Code) + } +} + +func TestHandler_DeleteAllMessages(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "DELETE", "delete-all-messages", nil) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } +} + +// ============================================================ +// Gotify application tests (via handler) +// ============================================================ + +func TestHandler_ListApplications(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "GET", "list-applications", nil) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var apps []GotifyApplication + if err := json.Unmarshal(w.Body.Bytes(), &apps); err != nil { + t.Fatalf("decode: %v", err) + } + if len(apps) != 2 { + t.Errorf("expected 2 apps, got %d", len(apps)) + } +} + +func TestHandler_CreateApplication(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "POST", "create-application", map[string]string{ + "name": "my-app", + "description": "test app", + }) + if w.Code != http.StatusCreated { + t.Fatalf("expected 201, got %d: %s", w.Code, w.Body.String()) + } + + data := decodeResponse(t, w) + if data["name"] != "my-app" { + t.Errorf("expected name 'my-app', got %v", data["name"]) + } +} + +func TestHandler_DeleteApplication(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "DELETE", "delete-application", map[string]int{"id": 1}) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestHandler_DeleteApplication_MissingID(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + h := newTestHandler(srv.URL) + + w := doRequest(h, "DELETE", "delete-application", map[string]int{"id": 0}) + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", w.Code) + } +} + +// ============================================================ +// Email tests (routing only, no SMTP) +// ============================================================ + +func TestHandler_SendEmail_MissingTo(t *testing.T) { + h := newTestHandler("") + + w := doRequest(h, "POST", "send-email", map[string]string{ + "subject": "Test", + "body": "Hello", + }) + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", w.Code) + } +} + +func TestHandler_SendEmail_InvalidJSON(t *testing.T) { + h := newTestHandler("") + + req := httptest.NewRequest("POST", "/", bytes.NewReader([]byte("{bad"))) + req.Header.Set("X-Resource", "send-email") + w := httptest.NewRecorder() + h.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", w.Code) + } +} + +// ============================================================ +// GotifyClient direct tests +// ============================================================ + +func TestGotifyClient_SendMessage(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + msg, err := client.SendMessage(GotifyMessage{ + Title: "Direct Test", + Message: "Hello", + Priority: 3, + }) + if err != nil { + t.Fatalf("send: %v", err) + } + if msg.ID != 42 { + t.Errorf("expected id 42, got %d", msg.ID) + } + if msg.Title != "Direct Test" { + t.Errorf("expected title 'Direct Test', got %s", msg.Title) + } +} + +func TestGotifyClient_ListMessages(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + list, err := client.ListMessages(50) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(list.Messages) != 2 { + t.Errorf("expected 2 messages, got %d", len(list.Messages)) + } +} + +func TestGotifyClient_DeleteMessage(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + if err := client.DeleteMessage(1); err != nil { + t.Fatalf("delete: %v", err) + } +} + +func TestGotifyClient_DeleteAllMessages(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + if err := client.DeleteAllMessages(); err != nil { + t.Fatalf("delete all: %v", err) + } +} + +func TestGotifyClient_ListApplications(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + apps, err := client.ListApplications() + if err != nil { + t.Fatalf("list: %v", err) + } + if len(apps) != 2 { + t.Errorf("expected 2 apps, got %d", len(apps)) + } +} + +func TestGotifyClient_CreateApplication(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + app, err := client.CreateApplication(GotifyApplication{ + Name: "new-app", + Description: "test", + }) + if err != nil { + t.Fatalf("create: %v", err) + } + if app.ID != 10 { + t.Errorf("expected id 10, got %d", app.ID) + } +} + +func TestGotifyClient_DeleteApplication(t *testing.T) { + srv := mockGotifyServer() + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + if err := client.DeleteApplication(1); err != nil { + t.Fatalf("delete: %v", err) + } +} + +func TestGotifyClient_ErrorResponse(t *testing.T) { + // Server that returns 500 for everything + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + fmt.Fprint(w, "internal error") + })) + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + _, err := client.SendMessage(GotifyMessage{Title: "test"}) + if err == nil { + t.Fatal("expected error") + } + + _, err = client.ListMessages(10) + if err == nil { + t.Fatal("expected error") + } + + _, err = client.ListApplications() + if err == nil { + t.Fatal("expected error") + } +} + +func TestGotifyClient_RateLimited(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTooManyRequests) + fmt.Fprint(w, "rate limited") + })) + defer srv.Close() + client := NewGotifyClient(srv.URL, "app-token", "client-token") + + _, err := client.SendMessage(GotifyMessage{Title: "test"}) + if err == nil { + t.Fatal("expected error on 429") + } +} + +func TestGotifyClient_InvalidURL(t *testing.T) { + client := NewGotifyClient("http://localhost:1", "app-token", "client-token") + + _, err := client.SendMessage(GotifyMessage{Title: "test"}) + if err == nil { + t.Fatal("expected connection error") + } +} diff --git a/internal/observability/handler.go b/internal/observability/handler.go new file mode 100644 index 0000000..7506ab7 --- /dev/null +++ b/internal/observability/handler.go @@ -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())) +} diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index b085abb..608d1ad 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -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) } diff --git a/internal/observability/prometheus.go b/internal/observability/prometheus.go new file mode 100644 index 0000000..6c2d040 --- /dev/null +++ b/internal/observability/prometheus.go @@ -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 +} diff --git a/internal/proxy/bodydispatch_test.go b/internal/proxy/bodydispatch_test.go index 0e01828..eede0ac 100644 --- a/internal/proxy/bodydispatch_test.go +++ b/internal/proxy/bodydispatch_test.go @@ -381,3 +381,66 @@ func TestBodySizeCappedDispatch(t *testing.T) { t.Errorf("expected 200 for reasonable body, got %d", resp.StatusCode) } } + +// TestUpstreamModelRewrite verifies that the model field is rewritten when upstreamModel is set. +func TestUpstreamModelRewrite(t *testing.T) { + var receivedModel string + + upstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var payload map[string]interface{} + json.Unmarshal(body, &payload) + receivedModel = payload["model"].(string) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, `{}`) + })) + defer upstreamServer.Close() + + upstreamAddr := strings.TrimPrefix(upstreamServer.URL, "http://") + + cfg := &config.Config{ + Models: map[string]*config.ModelUpstream{ + // Client sends "ornith:35b", upstream expects "qwen2.5:72b-instruct" + "ornith:35b": { + Name: "ornith:35b", + Address: upstreamAddr, + UpstreamModel: "qwen2.5:72b-instruct", + }, + // No rewrite - upstream model same as client model + "reasoning": { + Name: "reasoning", + Address: upstreamAddr, + }, + }, + Routes: make(map[string]*config.Route), + } + + handler := New(cfg) + defer handler.Close() + + server := httptest.NewServer(handler) + defer server.Close() + + // Test 1: Model should be rewritten + requestBody := `{"model":"ornith:35b","messages":[{"role":"user","content":"hi"}]}` + resp, err := http.Post(server.URL+"/v1/chat/completions", "application/json", strings.NewReader(requestBody)) + if err != nil { + t.Fatalf("request failed: %v", err) + } + resp.Body.Close() + + if receivedModel != "qwen2.5:72b-instruct" { + t.Errorf("expected upstream to receive model 'qwen2.5:72b-instruct', got '%s'", receivedModel) + } + + // Test 2: No rewrite when upstreamModel is empty + receivedModel = "" + requestBody = `{"model":"reasoning","messages":[{"role":"user","content":"hi"}]}` + resp, _ = http.Post(server.URL+"/v1/chat/completions", "application/json", strings.NewReader(requestBody)) + resp.Body.Close() + + if receivedModel != "reasoning" { + t.Errorf("expected upstream to receive model 'reasoning', got '%s'", receivedModel) + } +} diff --git a/internal/proxy/llm_metrics.go b/internal/proxy/llm_metrics.go new file mode 100644 index 0000000..bddc7c7 --- /dev/null +++ b/internal/proxy/llm_metrics.go @@ -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") +} + + diff --git a/internal/proxy/router.go b/internal/proxy/router.go index d35ab58..da95efa 100644 --- a/internal/proxy/router.go +++ b/internal/proxy/router.go @@ -122,6 +122,20 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { } } + // If upstream expects a different model name, rewrite the body + if modelUpstream.UpstreamModel != "" && modelUpstream.UpstreamModel != modelName { + payload["model"] = modelUpstream.UpstreamModel + newBody, err := json.Marshal(payload) + if err != nil { + return nil, &modelValidationError{ + Kind: "invalid_request", + Message: fmt.Sprintf("failed to rewrite model name: %v", err), + } + } + r.Body = io.NopCloser(bytes.NewReader(newBody)) + r.ContentLength = int64(len(newBody)) + } + // Determine the upstream path based on the request path upstreamPath := path if path == "/v1/rerank" { diff --git a/internal/serviceadapter/router.go b/internal/serviceadapter/router.go index 9d2e183..4c090d3 100644 --- a/internal/serviceadapter/router.go +++ b/internal/serviceadapter/router.go @@ -80,6 +80,14 @@ func (d *Dispatcher) Dispatch(w http.ResponseWriter, r *http.Request) { } } + // Internal handler: dispatch directly without reverse proxy + if adapter.Handler != nil { + // Set X-Upstream-Path so the handler knows which method was matched + r.Header.Set("X-Upstream-Path", method.UpstreamPath) + adapter.Handler.ServeHTTP(w, r) + return + } + upstreamURL := adapter.Spec.Upstream.URL if strings.HasPrefix(upstreamURL, "grpc://") { d.dispatchGRPC(w, r, upstreamURL, method, adapter) diff --git a/internal/serviceadapter/types.go b/internal/serviceadapter/types.go index a6b0303..95fc4b1 100644 --- a/internal/serviceadapter/types.go +++ b/internal/serviceadapter/types.go @@ -1,6 +1,7 @@ package serviceadapter import ( + "net/http" "time" ) @@ -49,11 +50,14 @@ type Status struct { } // ServiceAdapter is a gateway service adapter. +// When Handler is set, the dispatcher routes directly to the internal handler +// instead of reverse-proxying to Spec.Upstream.URL. type ServiceAdapter struct { - Name string // namespace/name - Namespace string + Name string // namespace/name + Namespace string ServiceName string - Spec Spec - Status Status - CreatedAt time.Time + Spec Spec + Status Status + CreatedAt time.Time + Handler http.Handler `json:"-" yaml:"-"` // internal handler (skip serialization) } diff --git a/internal/serviceadapter/workflow_adapter.go b/internal/serviceadapter/workflow_adapter.go index e4cd346..dee9a33 100644 --- a/internal/serviceadapter/workflow_adapter.go +++ b/internal/serviceadapter/workflow_adapter.go @@ -84,6 +84,68 @@ func (wa *WorkflowAdapter) HandleUpdate(w http.ResponseWriter, r *http.Request) wa.forwardToTemporal(w, r) } +// resourceToAction maps X-Resource names to Temporal action names. +var resourceToAction = map[string]string{ + "execute": "START_WORKFLOW", + "describe": "DESCRIBE_WORKFLOW", + "list": "LIST_WORKFLOWS", + "history": "GET_WORKFLOW_HISTORY", + "terminate": "TERMINATE_WORKFLOW", + "cancel": "CANCEL_WORKFLOW", + "signal": "SIGNAL_WORKFLOW", + "query": "QUERY_WORKFLOW", + "reset": "RESET_WORKFLOW", + "update": "UPDATE_WORKFLOW", +} + +// ServeHTTP implements http.Handler for X-Service: workflow routing. +// Maps X-Resource header to Temporal action, injects action into body, +// and forwards to the temporal handler. +func (wa *WorkflowAdapter) ServeHTTP(w http.ResponseWriter, r *http.Request) { + resource := r.Header.Get("X-Resource") + action, ok := resourceToAction[resource] + if !ok { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + fmt.Fprintf(w, `{"error":"unknown workflow resource: %s"}`, resource) + return + } + + // Read body, inject action, forward + body, err := io.ReadAll(r.Body) + if err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + fmt.Fprintf(w, `{"error":"failed to read body: %s"}`, err.Error()) + return + } + + var payload map[string]interface{} + if len(body) > 0 { + if err := json.Unmarshal(body, &payload); err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + fmt.Fprintf(w, `{"error":"invalid JSON: %s"}`, err.Error()) + return + } + } else { + payload = make(map[string]interface{}) + } + + // Inject action into body for temporal handler + payload["action"] = action + if _, ok := payload["namespace"]; !ok { + payload["namespace"] = "default" + } + + newBody, _ := json.Marshal(payload) + r.Body = io.NopCloser(bytes.NewReader(newBody)) + r.ContentLength = int64(len(newBody)) + r.URL.Path = "/workflow" + + wa.temporalHandler.ServeHTTP(w, r) +} + // forwardToTemporal reads the request body, ensures namespace is specified, // and forwards to the temporal handler. func (wa *WorkflowAdapter) forwardToTemporal(w http.ResponseWriter, r *http.Request) { @@ -134,7 +196,7 @@ func GetWorkflowSpec() *Spec { Retryable: true, Resources: []Resource{ { - Name: "start", + Name: "execute", Methods: []Method{ { Verb: "POST", @@ -151,6 +213,9 @@ func GetWorkflowSpec() *Spec { { Name: "describe", Methods: []Method{ + { + Verb: "GET", + }, { Verb: "POST", UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/DescribeWorkflowExecution", @@ -166,6 +231,9 @@ func GetWorkflowSpec() *Spec { { Name: "list", Methods: []Method{ + { + Verb: "GET", + }, { Verb: "POST", UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/ListWorkflowExecutions", @@ -181,6 +249,9 @@ func GetWorkflowSpec() *Spec { { Name: "history", Methods: []Method{ + { + Verb: "GET", + }, { Verb: "POST", UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/GetWorkflowExecutionHistory", diff --git a/k8s/grafana-dashboard-llm-metrics.yaml b/k8s/grafana-dashboard-llm-metrics.yaml new file mode 100644 index 0000000..88ba84e --- /dev/null +++ b/k8s/grafana-dashboard-llm-metrics.yaml @@ -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 + }