Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Admin Bot
|
c1a720687e | ||
|
Admin Bot
|
152e4259ae | ||
|
Admin Bot
|
5c30fd4fb7 | ||
|
Admin Bot
|
3f89511bdd | ||
 rockandAdmin Bot
|
3edcb10310 | ||
|
Admin Bot
|
df553cc70d | ||
|
Admin Bot
|
f32ff08365 | ||
|
Admin Bot
|
c44a9b174f | ||
|
Admin Bot
|
f1030aae82 | ||
|
Admin Bot
|
dc97afb0ef | ||
|
Admin Bot
|
eb4c2ff13d | ||
|
Admin Bot
|
4463508f29 | ||
|
Admin Bot
|
11ca4a5902 |
+30
-90
@@ -5,25 +5,18 @@ on:
|
||||
branches: [main]
|
||||
pull_request:
|
||||
branches: [main]
|
||||
workflow_dispatch:
|
||||
|
||||
env:
|
||||
REGISTRY: forgejo.riotpiao.com
|
||||
IMAGE: forgejo.riotpiao.com/rock/api-gateway
|
||||
DOCKER_HOST: tcp://localhost:2375
|
||||
IMAGE: forgejo.riotpiao.com/rock/homelab-frontend
|
||||
|
||||
jobs:
|
||||
ci:
|
||||
name: CI
|
||||
test:
|
||||
name: Test
|
||||
runs-on: golang
|
||||
steps:
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
apt-get update
|
||||
apt-get install -y docker.io curl nodejs
|
||||
curl -sLO "https://dl.k8s.io/release/$(curl -sL https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
|
||||
chmod +x kubectl && mv kubectl /usr/local/bin/
|
||||
kubectl version --client
|
||||
- name: Install Node.js for actions runtime
|
||||
run: apt-get update && apt-get install -y nodejs
|
||||
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v4
|
||||
@@ -34,9 +27,25 @@ jobs:
|
||||
- name: Go test
|
||||
run: go test ./...
|
||||
|
||||
build-push:
|
||||
name: Build & Push Image
|
||||
needs: test
|
||||
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
|
||||
runs-on: golang
|
||||
steps:
|
||||
- name: Install Node.js and Docker
|
||||
run: |
|
||||
apt-get update
|
||||
apt-get install -y nodejs docker.io
|
||||
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Get short SHA
|
||||
id: sha
|
||||
run: echo "short_sha=$(git rev-parse --short HEAD)" >> $GITHUB_OUTPUT
|
||||
run: |
|
||||
SHORT_SHA=$(git rev-parse --short HEAD)
|
||||
echo "short_sha=${SHORT_SHA}" >> $GITHUB_OUTPUT
|
||||
|
||||
- name: Registry login
|
||||
run: |
|
||||
@@ -50,84 +59,15 @@ jobs:
|
||||
run: |
|
||||
docker build --no-cache \
|
||||
-t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \
|
||||
-f Dockerfile .
|
||||
-t "${IMAGE}:latest" \
|
||||
-f Dockerfile \
|
||||
.
|
||||
|
||||
- name: Push image (SHA tag)
|
||||
run: docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
|
||||
# ── Tekton integration tests ─────────────────────────────
|
||||
- name: Setup kubeconfig
|
||||
- name: Push Docker image
|
||||
run: |
|
||||
mkdir -p ~/.kube
|
||||
echo "${KUBECONFIG_B64}" | base64 -d > ~/.kube/config
|
||||
kubectl get pipelineruns -n api --no-headers | head -1 || echo 'No PipelineRuns yet'
|
||||
echo '✓ kubeconfig works'
|
||||
env:
|
||||
KUBECONFIG_B64: ${{ secrets.KUBECONFIG_B64 }}
|
||||
|
||||
- name: Trigger Tekton PipelineRun
|
||||
id: tekton
|
||||
run: |
|
||||
SHA="${{ steps.sha.outputs.short_sha }}"
|
||||
RUN_NAME="integration-test-${SHA}"
|
||||
|
||||
# Clean up any previous run with the same name
|
||||
kubectl delete taskrun "${RUN_NAME}" -n api --ignore-not-found
|
||||
|
||||
# Create TaskRun — spins up gateway sidecar + curl tests
|
||||
cat <<YAML | kubectl create -f -
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: TaskRun
|
||||
metadata:
|
||||
name: ${RUN_NAME}
|
||||
namespace: api
|
||||
labels:
|
||||
commit-sha: "${SHA}"
|
||||
spec:
|
||||
taskRef:
|
||||
name: integration-test
|
||||
params:
|
||||
- name: image
|
||||
value: "${IMAGE}:${SHA}"
|
||||
YAML
|
||||
|
||||
echo "✓ TaskRun created: ${RUN_NAME}"
|
||||
|
||||
# Wait for completion (Succeeded or Failed)
|
||||
echo "Waiting for tests (timeout 5m)..."
|
||||
if kubectl wait taskrun/"${RUN_NAME}" -n api \
|
||||
--for=condition=Succeeded --timeout=5m 2>/dev/null; then
|
||||
echo "result=pass" >> $GITHUB_OUTPUT
|
||||
else
|
||||
echo "result=fail" >> $GITHUB_OUTPUT
|
||||
fi
|
||||
|
||||
# Print logs + results
|
||||
echo ""
|
||||
echo "=== Test Logs ==="
|
||||
POD=$(kubectl get pod -n api -l tekton.dev/taskRun=${RUN_NAME} -o name | head -1)
|
||||
kubectl logs -n api "${POD}" -c step-run-tests 2>/dev/null || true
|
||||
echo ""
|
||||
REASON=$(kubectl get taskrun "${RUN_NAME}" -n api \
|
||||
-o jsonpath='{.status.conditions[0].reason}')
|
||||
SUMMARY=$(kubectl get taskrun "${RUN_NAME}" -n api \
|
||||
-o jsonpath='{.status.results[?(@.name=="summary")].value}')
|
||||
echo "Status: ${REASON}"
|
||||
echo "Summary: ${SUMMARY}"
|
||||
|
||||
- name: Gate on test result
|
||||
if: steps.tekton.outputs.result != 'pass'
|
||||
run: |
|
||||
echo "✗ Integration tests FAILED — image NOT promoted"
|
||||
exit 1
|
||||
|
||||
# ── Promote only after tests pass ────────────────────────
|
||||
- name: Promote image to latest
|
||||
run: |
|
||||
docker tag "${IMAGE}:${{ steps.sha.outputs.short_sha }}" "${IMAGE}:latest"
|
||||
docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
docker push "${IMAGE}:latest"
|
||||
echo "✓ Promoted to latest"
|
||||
echo "✓ Image pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
|
||||
|
||||
- name: Cleanup
|
||||
if: always()
|
||||
run: docker image prune -af 2>&1 | tail -3 || true
|
||||
- name: Prune unused images
|
||||
run: docker image prune -a --force 2>&1 | tail -3 || true
|
||||
|
||||
+2
-33
@@ -1,34 +1,3 @@
|
||||
# SOPS Configuration for secrets encryption
|
||||
# Public keys are safe to commit; private keys stay in cluster
|
||||
|
||||
creation_rules:
|
||||
# Encrypt secrets, configs, and sensitive files
|
||||
# Multiple public keys for key rotation support
|
||||
# Files matching these patterns will be encrypted automatically with `sops -e`
|
||||
- path_regex: k8s/(.*secret.*|.*config.*|.*deployment.*\.ya?ml)
|
||||
age:
|
||||
- age1e5fq3hwxy78psus2nfvmtmua36g0u3suk78ephw6246l974d2utsvn0hla
|
||||
- age1ryxmuwhecmdru786eqgek4cf8ppq585j2uqr7e87phya42w9s5wscn6tgp
|
||||
encrypted_regex: '^data|^stringData' # Only encrypt data fields, keep structure readable
|
||||
|
||||
# Fallback rule for .enc.yaml files
|
||||
- path_regex: '.*\.enc\.ya?ml'
|
||||
age:
|
||||
- age1e5fq3hwxy78psus2nfvmtmua36g0u3suk78ephw6246l974d2utsvn0hla
|
||||
- age1ryxmuwhecmdru786eqgek4cf8ppq585j2uqr7e87phya42w9s5wscn6tgp
|
||||
encrypted_regex: '^data|^stringData'
|
||||
|
||||
# To encrypt a file locally:
|
||||
# sops --encrypt k8s/configmap.yaml > k8s/configmap.yaml
|
||||
#
|
||||
# To decrypt and view:
|
||||
# sops k8s/configmap.yaml
|
||||
#
|
||||
# To decrypt to stdout:
|
||||
# sops --decrypt k8s/configmap.yaml
|
||||
#
|
||||
# The private age keys are stored in the cluster at:
|
||||
# kubectl -n argocd get secret sops-age -o jsonpath='{.data.key\.txt}' | base64 -d
|
||||
#
|
||||
# Key rotation: Multiple public keys can coexist for decryption
|
||||
# Only private keys MUST be kept secret (in cluster only)
|
||||
- path_regex: \.enc\.ya?ml$
|
||||
age: age1e5fq3hwxy78psus2nfvmtmua36g0u3suk78ephw6246l974d2utsvn0hla
|
||||
|
||||
@@ -571,128 +571,42 @@ curl -X GET https://api.riotpiao.com/ \
|
||||
|
||||
## Authentication
|
||||
|
||||
All operations except `/healthz` and `/readyz` require JWT authentication.
|
||||
|
||||
### Bearer Token (JWT)
|
||||
|
||||
Provide JWT in Authorization header:
|
||||
All operations except `/healthz` and `/readyz` require authentication.
|
||||
|
||||
```bash
|
||||
curl -H 'Authorization: Bearer <jwt-token>' \
|
||||
https://api.riotpiao.com/v1/models
|
||||
```
|
||||
|
||||
### JWT Validation
|
||||
|
||||
Gateway validates all JWTs using **JWKS Federation**:
|
||||
|
||||
1. **Fetch JWKS** — Gateway fetches public keys from Authentik's JWKS endpoint (refreshed every 15 minutes)
|
||||
2. **Verify Signature** — Validates JWT signature using public key matching `kid` header
|
||||
3. **Check Claims:**
|
||||
- `iss` (issuer) — Must be Authentik provider (format: `https://authentik.riotpiao.com/application/o/{provider}/`)
|
||||
- `exp` (expiration) — Token must not be expired (60s clock skew allowed)
|
||||
- `nbf` (not before) — Token must not be in future (60s clock skew allowed)
|
||||
- `aud` (audience) — Must be non-empty string from Authentik
|
||||
4. **Check Permissions** — Validates required capabilities from JWT claims (see RBAC section)
|
||||
|
||||
**JWKS Endpoint:** `https://authentik.riotpiao.com/application/oidc/jwks/`
|
||||
|
||||
**Multi-Issuer Support:** Gateway accepts JWT from any Authentik service account provider (paperless-ai-agent, portfolio-analyzer, etc) because all share the same JWKS signing key.
|
||||
|
||||
### Obtaining Tokens
|
||||
|
||||
#### User Login (OIDC Device Code Flow)
|
||||
|
||||
**Via Authentik OIDC (human login):**
|
||||
```bash
|
||||
core auth login --username [email protected]
|
||||
export USER_TOKEN=$(cat ~/.cache/talos/authentik_id_token)
|
||||
```
|
||||
|
||||
curl -H "Authorization: Bearer $USER_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" \
|
||||
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)
|
||||
|
||||
JWT claims contain permission arrays. Required capabilities:
|
||||
Tokens embed capabilities in claims. Required capabilities:
|
||||
|
||||
| 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).
|
||||
- `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
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -38,22 +38,22 @@ Production API gateway for the homelab cluster. Single entry point (`api.riotpia
|
||||
│ (routing, auth, limits) │
|
||||
└──────┬───────────────────────┘
|
||||
│
|
||||
┌──────┴───────────────────────────────────┐
|
||||
┌──────┴──────────────────────────────────┐
|
||||
│ │
|
||||
/v1/* X-Service header routing /
|
||||
(LLM) (workflow, sqs, s3, iam, memory) /
|
||||
│ │ │
|
||||
▼ ▼ ▼
|
||||
llm-serving temporal:7233 kmsvc/Kafka, MinIO,
|
||||
(vLLM, Ollama) (gRPC) Authentik, poimen-memory
|
||||
(TEI)
|
||||
/v1/* /workflow /sqs /
|
||||
(LLM) (Temporal gRPC) (Queues) (X-Service)
|
||||
│ │ │ │
|
||||
▼ ▼ ▼ ▼
|
||||
llm-serving temporal:7233 kmsvc/Kafka IAM, S3
|
||||
(vLLM, Ollama) (WorkflowService) Memory
|
||||
(TEI) (gRPC bridge) (poimen)
|
||||
```
|
||||
|
||||
**Design principles:**
|
||||
- ✅ Single hostname, unified X-Service + X-Resource header routing
|
||||
- ✅ HTTP REST gateway → gRPC Temporal bridge (via X-Service: workflow)
|
||||
- ✅ Single hostname, multiple path prefixes
|
||||
- ✅ HTTP REST gateway → gRPC Temporal bridge
|
||||
- ✅ Bearer token auth via Authentik (JWT + RBAC)
|
||||
- ✅ Streaming unbuffered (SSE, WebSocket, HTTP/2 multiplexing)
|
||||
- ✅ Streaming unbuffered (SSE, WebSocket)
|
||||
- ✅ Per-route timeouts & rate limits
|
||||
- ✅ No cluster credentials held by gateway
|
||||
|
||||
@@ -61,16 +61,16 @@ llm-serving temporal:7233 kmsvc/Kafka, MinIO,
|
||||
|
||||
## Services & Capabilities
|
||||
|
||||
| Service | Method | Upstream | Status |
|
||||
| Service | Prefix | Upstream | Status |
|
||||
|---------|--------|----------|--------|
|
||||
| **LLM Chat** | `POST /v1/chat/completions` | llm-serving (vLLM) | ✅ Live |
|
||||
| **Embeddings** | `POST /v1/embeddings` | llm-serving (TEI) | ✅ Live |
|
||||
| **Reranking** | `POST /v1/rerank` | llm-serving (TEI) | ✅ Live |
|
||||
| **Workflows** | `X-Service: workflow` + `X-Resource: {action}` | Temporal gRPC (7233) | ✅ Live (START, DESCRIBE, SIGNAL, QUERY, etc) |
|
||||
| **Queues** | `X-Service: sqs` + `X-Resource: {action}` | kmsvc/Kafka | ✅ Live |
|
||||
| **Memory** | `X-Service: memory` + `X-Resource: {action}` | poimen-memory | ✅ Live |
|
||||
| **IAM** | `X-Service: iam` + `X-Resource: {action}` | Authentik API | ✅ Live |
|
||||
| **S3** | `X-Service: s3` + `X-Resource: {action}` | MinIO | ✅ Live |
|
||||
| **LLM Chat** | `/v1/chat/completions` | llm-serving (vLLM) | ✅ Live |
|
||||
| **Embeddings** | `/v1/embeddings` | llm-serving (TEI) | ✅ Live |
|
||||
| **Reranking** | `/v1/rerank` | llm-serving (TEI) | ✅ Live |
|
||||
| **Workflows** | `/workflow` | Temporal gRPC (7233) | ✅ Live (START, DESCRIBE, SIGNAL, QUERY, etc) |
|
||||
| **Queues** | `/` + `X-Service: sqs` | kmsvc/Kafka | ⏳ Ready (ServiceAdapter) |
|
||||
| **Memory** | `/` + `X-Service: memory` | poimen-memory | ✅ Live |
|
||||
| **IAM** | `/` + `X-Service: iam` | Authentik API | ✅ Live |
|
||||
| **S3** | `/` + `X-Service: s3` | MinIO | ✅ Live |
|
||||
|
||||
---
|
||||
|
||||
@@ -102,17 +102,18 @@ curl -X POST https://api.riotpiao.com/v1/chat/completions \
|
||||
}'
|
||||
```
|
||||
|
||||
**Workflow (via X-Service header):**
|
||||
**Workflow:**
|
||||
```bash
|
||||
curl -X POST https://api.riotpiao.com/ \
|
||||
curl -X POST https://api.riotpiao.com/workflow \
|
||||
-H "Authorization: Bearer $TOKEN" \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: start" \
|
||||
-d '{
|
||||
"action": "START_WORKFLOW",
|
||||
"namespace": "default",
|
||||
"payload": {
|
||||
"workflow_id": "my-workflow",
|
||||
"workflow_type": "MyWorkflow",
|
||||
"task_queue": "default"
|
||||
}
|
||||
}'
|
||||
```
|
||||
|
||||
|
||||
+1
-23
@@ -9,7 +9,6 @@ import (
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/proxy"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/server"
|
||||
@@ -72,32 +71,11 @@ func main() {
|
||||
|
||||
// Create ServiceAdapter registry and dispatcher (phase 8)
|
||||
registry := serviceadapter.NewRegistry(nil)
|
||||
|
||||
// Add workflow service adapter (uses Temporal handler for gRPC forwarding)
|
||||
workflowSpec := serviceadapter.GetWorkflowSpec()
|
||||
workflowAdapter := &serviceadapter.ServiceAdapter{
|
||||
Namespace: "temporal",
|
||||
ServiceName: "workflow",
|
||||
Spec: *workflowSpec,
|
||||
}
|
||||
_ = registry.Add(workflowAdapter)
|
||||
|
||||
// Add other adapters from config
|
||||
for _, a := range cfg.Adapters {
|
||||
_ = registry.Add(a)
|
||||
}
|
||||
log.Printf("%d service adapters loaded", registry.Count())
|
||||
|
||||
// Create shared JWT validator for X-Service auth enforcement
|
||||
var jwtValidator *auth.Validator
|
||||
if cfg.Auth.Enabled && cfg.Auth.JWKSURL != "" {
|
||||
jwtValidator = auth.NewValidator(cfg.Auth.Issuer, cfg.Auth.Audience, cfg.Auth.JWKSURL)
|
||||
}
|
||||
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
|
||||
dispatcher := serviceadapter.NewDispatcher(registry)
|
||||
|
||||
// Create router that handles health endpoints, X-Service (ServiceAdapter) routing,
|
||||
// temporal endpoints, and passes others to upstream handler
|
||||
|
||||
@@ -1,36 +0,0 @@
|
||||
#!/bin/bash
|
||||
# Example: Send email via notification/sendMsg endpoint
|
||||
|
||||
BASE_URL="${1:-https://api.riotpiao.com}"
|
||||
AUTH_TOKEN="${2:-}" # Optional JWT token if auth required
|
||||
|
||||
PAYLOAD=$(cat <<'EOF'
|
||||
{
|
||||
"format": "smtp",
|
||||
"title": "System Alert",
|
||||
"message": "CPU usage exceeded 90% threshold",
|
||||
"priority": 7,
|
||||
"extras": {
|
||||
"to_email": "[email protected]",
|
||||
"cc": "[email protected]"
|
||||
}
|
||||
}
|
||||
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
|
||||
|
||||
echo ""
|
||||
@@ -1,104 +0,0 @@
|
||||
package config_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
|
||||
)
|
||||
|
||||
func TestLoadAuthConfig_TokenURLAndClientID(t *testing.T) {
|
||||
yaml := `
|
||||
routes: []
|
||||
models: []
|
||||
auth:
|
||||
enabled: true
|
||||
issuer: "https://authentik.example.com/application/o/api-gw/"
|
||||
audience: "api-gw"
|
||||
jwksUrl: "https://authentik.example.com/application/o/api-gw/jwks/"
|
||||
requiredCapability: "llm:inference"
|
||||
tokenUrl: "https://authentik.example.com/application/o/token/"
|
||||
clientId: "api-gw"
|
||||
`
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "config.yaml")
|
||||
if err := os.WriteFile(path, []byte(yaml), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Set env for client secret
|
||||
t.Setenv("AUTH_CLIENT_SECRET", "test-secret-value")
|
||||
|
||||
_, _, _, auth, err := config.LoadRoutesAndModelsFromFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
|
||||
if !auth.Enabled {
|
||||
t.Error("auth should be enabled")
|
||||
}
|
||||
if auth.TokenURL != "https://authentik.example.com/application/o/token/" {
|
||||
t.Errorf("tokenUrl = %q, want authentik token endpoint", auth.TokenURL)
|
||||
}
|
||||
if auth.ClientID != "api-gw" {
|
||||
t.Errorf("clientId = %q, want api-gw", auth.ClientID)
|
||||
}
|
||||
if auth.ClientSecret != "test-secret-value" {
|
||||
t.Errorf("clientSecret = %q, want test-secret-value", auth.ClientSecret)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadAuthConfig_ClientSecretFromEnvOnly(t *testing.T) {
|
||||
yaml := `
|
||||
routes: []
|
||||
models: []
|
||||
auth:
|
||||
enabled: true
|
||||
tokenUrl: "https://example.com/token/"
|
||||
clientId: "test"
|
||||
`
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "config.yaml")
|
||||
os.WriteFile(path, []byte(yaml), 0644)
|
||||
|
||||
// No AUTH_CLIENT_SECRET env set
|
||||
t.Setenv("AUTH_CLIENT_SECRET", "")
|
||||
|
||||
_, _, _, auth, err := config.LoadRoutesAndModelsFromFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
|
||||
if auth.ClientSecret != "" {
|
||||
t.Errorf("clientSecret should be empty when env not set, got %q", auth.ClientSecret)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadAuthConfig_BackwardCompatible(t *testing.T) {
|
||||
// Config without tokenUrl/clientId should still load (zero values)
|
||||
yaml := `
|
||||
routes: []
|
||||
models: []
|
||||
auth:
|
||||
enabled: true
|
||||
issuer: "https://example.com/"
|
||||
jwksUrl: "https://example.com/jwks/"
|
||||
requiredCapability: "llm:inference"
|
||||
`
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "config.yaml")
|
||||
os.WriteFile(path, []byte(yaml), 0644)
|
||||
|
||||
_, _, _, auth, err := config.LoadRoutesAndModelsFromFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
|
||||
if auth.TokenURL != "" {
|
||||
t.Errorf("tokenUrl should be empty, got %q", auth.TokenURL)
|
||||
}
|
||||
if auth.ClientID != "" {
|
||||
t.Errorf("clientId should be empty, got %q", auth.ClientID)
|
||||
}
|
||||
}
|
||||
@@ -24,8 +24,6 @@ type Config struct {
|
||||
Adapters []*serviceadapter.ServiceAdapter
|
||||
// Auth holds JWT authentication configuration for /v1/* endpoints.
|
||||
Auth AuthConfig
|
||||
// Temporal holds Temporal server configuration.
|
||||
Temporal TemporalConfig
|
||||
}
|
||||
|
||||
// ModelUpstream holds upstream configuration for a specific model.
|
||||
@@ -40,12 +38,6 @@ type ModelUpstream struct {
|
||||
AuthRequired bool
|
||||
}
|
||||
|
||||
// TemporalConfig holds Temporal server configuration.
|
||||
type TemporalConfig struct {
|
||||
// HostPort is the address of the Temporal server (host:port).
|
||||
HostPort string
|
||||
}
|
||||
|
||||
// AuthConfig holds JWT authentication configuration.
|
||||
type AuthConfig struct {
|
||||
// Enabled globally enables/disables auth for /v1/* endpoints.
|
||||
@@ -58,12 +50,6 @@ type AuthConfig struct {
|
||||
JWKSURL string
|
||||
// RequiredCapability is the permission required for LLM inference (e.g., "llm:inference").
|
||||
RequiredCapability string
|
||||
// TokenURL is the Authentik token endpoint for password/refresh grants.
|
||||
TokenURL string
|
||||
// ClientID is the OAuth2 client ID for token exchange.
|
||||
ClientID string
|
||||
// ClientSecret is the OAuth2 client secret (loaded from env, never from config file).
|
||||
ClientSecret string
|
||||
}
|
||||
|
||||
// Route represents a single route and its upstream configuration.
|
||||
@@ -146,12 +132,6 @@ func Load() (*Config, error) {
|
||||
authConfig = loadedAuth
|
||||
}
|
||||
|
||||
temporalHostPort := "localhost:7233"
|
||||
// Allow override via environment variable
|
||||
if hostPort, ok := os.LookupEnv("TEMPORAL_HOST_PORT"); ok {
|
||||
temporalHostPort = hostPort
|
||||
}
|
||||
|
||||
return &Config{
|
||||
ListenAddr: listenAddr,
|
||||
ShutdownTimeout: shutdownTimeout,
|
||||
@@ -159,8 +139,5 @@ func Load() (*Config, error) {
|
||||
Models: models,
|
||||
Adapters: adapters,
|
||||
Auth: authConfig,
|
||||
Temporal: TemporalConfig{
|
||||
HostPort: temporalHostPort,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -25,8 +25,6 @@ type rawAuth struct {
|
||||
Audience string `yaml:"audience"`
|
||||
JWKSURL string `yaml:"jwksUrl"`
|
||||
RequiredCapability string `yaml:"requiredCapability"`
|
||||
TokenURL string `yaml:"tokenUrl"`
|
||||
ClientID string `yaml:"clientId"`
|
||||
}
|
||||
|
||||
// rawRoute represents a single route in the YAML configuration.
|
||||
@@ -178,9 +176,6 @@ func LoadRoutesAndModelsFromFile(path string) (map[string]*Route, map[string]*Mo
|
||||
Audience: raw.Auth.Audience,
|
||||
JWKSURL: raw.Auth.JWKSURL,
|
||||
RequiredCapability: raw.Auth.RequiredCapability,
|
||||
TokenURL: raw.Auth.TokenURL,
|
||||
ClientID: raw.Auth.ClientID,
|
||||
ClientSecret: os.Getenv("AUTH_CLIENT_SECRET"),
|
||||
}
|
||||
|
||||
return routes, models, adapters, authConfig, nil
|
||||
|
||||
@@ -1,111 +0,0 @@
|
||||
// Package identity extracts authenticated user identity from JWT claims
|
||||
// and injects forwarding headers into proxied requests.
|
||||
//
|
||||
// Headers injected after JWT validation:
|
||||
//
|
||||
// X-Forwarded-User: subject (sub claim)
|
||||
// X-Forwarded-Roles: comma-separated roles or permissions
|
||||
// X-Acting-Service: authorized party (azp claim), only for service accounts
|
||||
// X-Auth-Verified: "true" when gateway validated the JWT
|
||||
//
|
||||
// Security contract: downstream services MUST only accept traffic from the
|
||||
// gateway (enforced by NetworkPolicy). They trust these headers because the
|
||||
// gateway is the sole ingress path.
|
||||
package identity
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
)
|
||||
|
||||
// Headers that the gateway controls. Incoming values from clients are
|
||||
// stripped to prevent spoofing.
|
||||
const (
|
||||
HeaderUser = "X-Forwarded-User"
|
||||
HeaderRoles = "X-Forwarded-Roles"
|
||||
HeaderActingService = "X-Acting-Service"
|
||||
HeaderAuthVerified = "X-Auth-Verified"
|
||||
)
|
||||
|
||||
// managed lists all headers this package owns. Used for stripping and cleanup.
|
||||
var managed = []string{
|
||||
HeaderUser,
|
||||
HeaderRoles,
|
||||
HeaderActingService,
|
||||
HeaderAuthVerified,
|
||||
}
|
||||
|
||||
// StripIncoming removes all gateway-managed identity headers from an
|
||||
// inbound request, preventing clients from spoofing identity.
|
||||
// Call this early in the handler chain, before any routing.
|
||||
func StripIncoming(r *http.Request) {
|
||||
for _, h := range managed {
|
||||
r.Header.Del(h)
|
||||
}
|
||||
}
|
||||
|
||||
// Inject extracts identity from validated JWT claims and sets the
|
||||
// corresponding forwarding headers on the request. Only call this
|
||||
// after successful JWT validation.
|
||||
func Inject(r *http.Request, claims jwt.MapClaims) {
|
||||
r.Header.Set(HeaderAuthVerified, "true")
|
||||
|
||||
if sub := claimString(claims, "sub"); sub != "" {
|
||||
r.Header.Set(HeaderUser, sub)
|
||||
}
|
||||
|
||||
if roles := claimStringSlice(claims, "roles"); len(roles) > 0 {
|
||||
r.Header.Set(HeaderRoles, strings.Join(roles, ","))
|
||||
} else if perms := claimStringSlice(claims, "permissions"); len(perms) > 0 {
|
||||
r.Header.Set(HeaderRoles, strings.Join(perms, ","))
|
||||
}
|
||||
|
||||
if azp := claimString(claims, "azp"); azp != "" {
|
||||
sub := claimString(claims, "sub")
|
||||
// Only set acting-service when azp differs from sub
|
||||
// (i.e., a service account acting, not the user themselves)
|
||||
if azp != sub {
|
||||
r.Header.Set(HeaderActingService, azp)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// claimString extracts a string value from claims, returning "" if
|
||||
// the key is missing or not a string.
|
||||
func claimString(claims jwt.MapClaims, key string) string {
|
||||
val, ok := claims[key]
|
||||
if !ok || val == nil {
|
||||
return ""
|
||||
}
|
||||
s, ok := val.(string)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// claimStringSlice extracts a []string from claims. JWT libraries
|
||||
// deserialize JSON arrays as []interface{}, so each element is
|
||||
// type-asserted individually. Non-string elements are skipped.
|
||||
func claimStringSlice(claims jwt.MapClaims, key string) []string {
|
||||
val, ok := claims[key]
|
||||
if !ok || val == nil {
|
||||
return nil
|
||||
}
|
||||
raw, ok := val.([]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(raw))
|
||||
for _, v := range raw {
|
||||
if s, ok := v.(string); ok && s != "" {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -1,205 +0,0 @@
|
||||
package identity
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
)
|
||||
|
||||
func TestStripIncoming_RemovesSpoofedHeaders(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set(HeaderUser, "evil-spoof")
|
||||
r.Header.Set(HeaderRoles, "admin:*")
|
||||
r.Header.Set(HeaderActingService, "fake-service")
|
||||
r.Header.Set(HeaderAuthVerified, "true")
|
||||
|
||||
StripIncoming(r)
|
||||
|
||||
for _, h := range managed {
|
||||
if got := r.Header.Get(h); got != "" {
|
||||
t.Errorf("header %s should be stripped, got %q", h, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStripIncoming_PreservesOtherHeaders(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("Authorization", "Bearer token")
|
||||
r.Header.Set("Content-Type", "application/json")
|
||||
r.Header.Set(HeaderUser, "spoof")
|
||||
|
||||
StripIncoming(r)
|
||||
|
||||
if got := r.Header.Get("Authorization"); got != "Bearer token" {
|
||||
t.Errorf("Authorization should be preserved, got %q", got)
|
||||
}
|
||||
if got := r.Header.Get("Content-Type"); got != "application/json" {
|
||||
t.Errorf("Content-Type should be preserved, got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInject_ServiceAccount(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "abc123-hashed-id",
|
||||
"azp": "portfolio-agent",
|
||||
"roles": []interface{}{"llm:inference", "memory:read"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
assertHeader(t, r, HeaderAuthVerified, "true")
|
||||
assertHeader(t, r, HeaderUser, "abc123-hashed-id")
|
||||
assertHeader(t, r, HeaderRoles, "llm:inference,memory:read")
|
||||
assertHeader(t, r, HeaderActingService, "portfolio-agent")
|
||||
}
|
||||
|
||||
func TestInject_HumanUser(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "user-hash-456",
|
||||
"azp": "api-gw",
|
||||
"permissions": []interface{}{"*"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
assertHeader(t, r, HeaderAuthVerified, "true")
|
||||
assertHeader(t, r, HeaderUser, "user-hash-456")
|
||||
assertHeader(t, r, HeaderRoles, "*")
|
||||
// azp != sub, so acting-service is set
|
||||
assertHeader(t, r, HeaderActingService, "api-gw")
|
||||
}
|
||||
|
||||
func TestInject_SameSubAndAzp_NoActingService(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "portfolio-agent",
|
||||
"azp": "portfolio-agent",
|
||||
"roles": []interface{}{"llm:inference"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
assertHeader(t, r, HeaderActingService, "")
|
||||
}
|
||||
|
||||
func TestInject_RolesOverPermissions(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "user-1",
|
||||
"roles": []interface{}{"llm:inference"},
|
||||
"permissions": []interface{}{"admin:*"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
// roles takes precedence over permissions
|
||||
assertHeader(t, r, HeaderRoles, "llm:inference")
|
||||
}
|
||||
|
||||
func TestInject_PermissionsFallback(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "user-1",
|
||||
"permissions": []interface{}{"grafana:read", "grafana:write"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
assertHeader(t, r, HeaderRoles, "grafana:read,grafana:write")
|
||||
}
|
||||
|
||||
func TestInject_EmptyClaims(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
assertHeader(t, r, HeaderAuthVerified, "true")
|
||||
assertHeader(t, r, HeaderUser, "")
|
||||
assertHeader(t, r, HeaderRoles, "")
|
||||
assertHeader(t, r, HeaderActingService, "")
|
||||
}
|
||||
|
||||
func TestInject_NilValuesInClaims(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": nil,
|
||||
"azp": nil,
|
||||
"roles": nil,
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
assertHeader(t, r, HeaderAuthVerified, "true")
|
||||
assertHeader(t, r, HeaderUser, "")
|
||||
assertHeader(t, r, HeaderRoles, "")
|
||||
}
|
||||
|
||||
func TestInject_WildcardPermission(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "admin-user",
|
||||
"permissions": []interface{}{"*"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
// Wildcard passed as literal, never expanded
|
||||
assertHeader(t, r, HeaderRoles, "*")
|
||||
}
|
||||
|
||||
func TestInject_MixedTypeRolesArray(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "user-1",
|
||||
"roles": []interface{}{"llm:inference", 42, nil, "", "memory:read"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
// Non-string and empty elements skipped
|
||||
assertHeader(t, r, HeaderRoles, "llm:inference,memory:read")
|
||||
}
|
||||
|
||||
func TestInject_EmptyRolesArray(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "user-1",
|
||||
"roles": []interface{}{},
|
||||
"permissions": []interface{}{"backup:read"},
|
||||
}
|
||||
|
||||
Inject(r, claims)
|
||||
|
||||
// Empty roles falls through to permissions
|
||||
assertHeader(t, r, HeaderRoles, "backup:read")
|
||||
}
|
||||
|
||||
func TestStripThenInject_OverwritesSpoof(t *testing.T) {
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set(HeaderUser, "evil-spoof")
|
||||
r.Header.Set(HeaderAuthVerified, "true")
|
||||
|
||||
StripIncoming(r)
|
||||
|
||||
claims := jwt.MapClaims{
|
||||
"sub": "real-user",
|
||||
"roles": []interface{}{"llm:inference"},
|
||||
}
|
||||
Inject(r, claims)
|
||||
|
||||
assertHeader(t, r, HeaderUser, "real-user")
|
||||
assertHeader(t, r, HeaderAuthVerified, "true")
|
||||
}
|
||||
|
||||
func assertHeader(t *testing.T, r *http.Request, key, want string) {
|
||||
t.Helper()
|
||||
got := r.Header.Get(key)
|
||||
if got != want {
|
||||
t.Errorf("header %s = %q, want %q", key, got, want)
|
||||
}
|
||||
}
|
||||
@@ -1,299 +0,0 @@
|
||||
//go:build integration
|
||||
|
||||
package integration
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
gatewayBaseURL = "http://api-gateway:8080"
|
||||
timeout = 30 * time.Second
|
||||
)
|
||||
|
||||
// TestIntegrationMemoryService tests memory adapter (ingest, query)
|
||||
func TestIntegrationMemoryService(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: timeout}
|
||||
|
||||
// Test 1: Ingest memory
|
||||
t.Log("Testing memory ingest...")
|
||||
ingestPayload := map[string]interface{}{
|
||||
"ingest_id": "test-ingest-" + fmt.Sprintf("%d", time.Now().Unix()),
|
||||
"project": "test-project",
|
||||
"title": "Integration Test Memory",
|
||||
"content": "This is a test memory entry from integration test",
|
||||
"tags": []string{"integration", "test"},
|
||||
"source": "integration-test",
|
||||
}
|
||||
|
||||
ingestBody, _ := json.Marshal(ingestPayload)
|
||||
req, _ := http.NewRequest("POST", gatewayBaseURL+"/memory/ingest", bytes.NewReader(ingestBody))
|
||||
req.Header.Set("X-Service", "memory")
|
||||
req.Header.Set("X-Resource", "ingest")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("memory ingest request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
t.Logf("Ingest response: %d - %s", resp.StatusCode, string(body))
|
||||
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
|
||||
t.Fatalf("memory ingest failed with status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
t.Log("✓ Memory ingest successful")
|
||||
|
||||
// Test 2: Query memory
|
||||
t.Log("Testing memory query...")
|
||||
queryPayload := map[string]interface{}{
|
||||
"query": "integration test",
|
||||
}
|
||||
|
||||
queryBody, _ := json.Marshal(queryPayload)
|
||||
req, _ = http.NewRequest("POST", gatewayBaseURL+"/memory/query", bytes.NewReader(queryBody))
|
||||
req.Header.Set("X-Service", "memory")
|
||||
req.Header.Set("X-Resource", "query")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err = client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("memory query request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, _ = io.ReadAll(resp.Body)
|
||||
t.Logf("Query response: %d - %s", resp.StatusCode, string(body))
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("memory query failed with status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
t.Log("✓ Memory query successful")
|
||||
}
|
||||
|
||||
// TestIntegrationS3Service tests S3 adapter (list, put, get)
|
||||
func TestIntegrationS3Service(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: timeout}
|
||||
|
||||
// Test 1: List objects
|
||||
t.Log("Testing S3 list objects...")
|
||||
req, _ := http.NewRequest("GET", gatewayBaseURL+"/", nil)
|
||||
req.Header.Set("X-Service", "s3")
|
||||
req.Header.Set("X-Resource", "list-objects")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("S3 list request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
t.Logf("List response: %d - %s", resp.StatusCode, string(body)[:100])
|
||||
|
||||
// S3 should respond with either 200 (list) or 403 (access denied) - both mean routing works
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusForbidden {
|
||||
t.Fatalf("S3 list failed with unexpected status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
t.Log("✓ S3 list objects successful")
|
||||
|
||||
// Test 2: Put object to dedicated test bucket
|
||||
t.Log("Testing S3 put object...")
|
||||
testContent := fmt.Sprintf("Integration test data - %d", time.Now().Unix())
|
||||
testKey := "test-file-" + fmt.Sprintf("%d", time.Now().Unix()) + ".txt"
|
||||
|
||||
req, _ = http.NewRequest("PUT", gatewayBaseURL+"/"+testKey, bytes.NewReader([]byte(testContent)))
|
||||
req.Header.Set("X-Service", "s3")
|
||||
req.Header.Set("X-Resource", "put-object")
|
||||
req.Header.Set("Content-Type", "text/plain")
|
||||
|
||||
resp, err = client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("S3 put request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, _ = io.ReadAll(resp.Body)
|
||||
t.Logf("Put response: %d - %s", resp.StatusCode, string(body)[:100])
|
||||
|
||||
// Put should respond - either success or S3 error (both mean routing works)
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 600 {
|
||||
t.Fatalf("S3 put failed with status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
t.Log("✓ S3 put object successful")
|
||||
}
|
||||
|
||||
// TestIntegrationSQSService tests SQS adapter (create queue, send, receive)
|
||||
func TestIntegrationSQSService(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: timeout}
|
||||
|
||||
// Test 1: List queues (no auth required in test, auth error is ok)
|
||||
t.Log("Testing SQS list queues...")
|
||||
req, _ := http.NewRequest("GET", gatewayBaseURL+"/sqs/queues", nil)
|
||||
req.Header.Set("X-Service", "sqs")
|
||||
req.Header.Set("X-Resource", "list-queues")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("SQS list request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
t.Logf("List queues response: %d - %s", resp.StatusCode, string(body))
|
||||
|
||||
// SQS requires auth, so 401 is expected but proves routing works
|
||||
if resp.StatusCode == http.StatusUnauthorized {
|
||||
t.Log("✓ SQS correctly requires authorization (routing works)")
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode == http.StatusOK {
|
||||
t.Log("✓ SQS list queues successful")
|
||||
return
|
||||
}
|
||||
|
||||
t.Fatalf("SQS list failed with unexpected status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
// TestIntegrationWorkflowService tests workflow adapter (list, describe)
|
||||
func TestIntegrationWorkflowService(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: timeout}
|
||||
|
||||
// Test 1: List workflows with gRPC
|
||||
t.Log("Testing workflow list (gRPC)...")
|
||||
req, _ := http.NewRequest("GET",
|
||||
gatewayBaseURL+"/temporal.api.workflowservice.v1.WorkflowService/ListWorkflowExecutions",
|
||||
nil)
|
||||
req.Header.Set("X-Service", "workflow")
|
||||
req.Header.Set("X-Resource", "list")
|
||||
req.Header.Set("Content-Type", "application/grpc")
|
||||
req.Header.Set("TE", "trailers")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("workflow list request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
|
||||
// gRPC responses are binary, but we can check status code
|
||||
t.Logf("List workflows response: %d (body length: %d bytes)", resp.StatusCode, len(body))
|
||||
|
||||
// Status 200 with gRPC binary data, or 501 if not yet implemented
|
||||
if resp.StatusCode == http.StatusOK {
|
||||
t.Log("✓ Workflow list successful (gRPC forwarding working)")
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode == http.StatusNotImplemented {
|
||||
t.Log("⚠ Workflow list: gRPC forwarding not yet implemented")
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode >= 400 && resp.StatusCode < 500 {
|
||||
// Client error might indicate routing works but request format issue
|
||||
t.Logf("✓ Workflow adapter routing confirmed (status %d)", resp.StatusCode)
|
||||
return
|
||||
}
|
||||
|
||||
t.Fatalf("Workflow list failed with status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
// TestIntegrationIAMService tests IAM adapter
|
||||
func TestIntegrationIAMService(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: timeout}
|
||||
|
||||
t.Log("Testing IAM list users...")
|
||||
req, _ := http.NewRequest("GET", gatewayBaseURL+"/api/v3/users", nil)
|
||||
req.Header.Set("X-Service", "iam")
|
||||
req.Header.Set("X-Resource", "list-users")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("IAM list request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
_, _ = io.ReadAll(resp.Body)
|
||||
t.Logf("IAM list users response: %d", resp.StatusCode)
|
||||
|
||||
// IAM (Authentik) should respond - 200, 404, or auth error all prove routing works
|
||||
if resp.StatusCode >= 200 && resp.StatusCode < 600 {
|
||||
t.Log("✓ IAM adapter routing successful")
|
||||
return
|
||||
}
|
||||
|
||||
t.Fatalf("IAM list failed with status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
// TestIntegrationHealthChecks tests gateway health endpoints
|
||||
func TestIntegrationHealthChecks(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: timeout}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
endpoint string
|
||||
}{
|
||||
{"liveness", "/healthz"},
|
||||
{"readiness", "/readyz"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
req, _ := http.NewRequest("GET", gatewayBaseURL+tt.endpoint, nil)
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("health check request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("health check failed with status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
var health map[string]string
|
||||
if err := json.NewDecoder(resp.Body).Decode(&health); err != nil {
|
||||
t.Fatalf("failed to decode health response: %v", err)
|
||||
}
|
||||
|
||||
t.Logf("✓ %s: %s", tt.name, health["status"])
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,232 +0,0 @@
|
||||
package notification
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/smtp"
|
||||
"os"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// 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": "[email protected]", "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).
|
||||
type Handler struct {
|
||||
smtpHost string
|
||||
smtpPort string
|
||||
smtpFrom string
|
||||
smtpUser string
|
||||
smtpPass string
|
||||
smsAPIURL string
|
||||
smsAPIKey string
|
||||
gotifyURL string
|
||||
gotifyToken string
|
||||
}
|
||||
|
||||
// NewHandler creates a new notification handler from environment variables.
|
||||
func NewHandler() *Handler {
|
||||
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"),
|
||||
}
|
||||
}
|
||||
|
||||
// ServeHTTP handles sendMsg requests.
|
||||
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
// Route based on format
|
||||
var resp SendMsgResponse
|
||||
switch req.Format {
|
||||
case "smtp":
|
||||
resp = h.sendEmail(req)
|
||||
case "sms":
|
||||
resp = h.sendSMS(req)
|
||||
case "gotify":
|
||||
resp = h.sendGotify(req)
|
||||
default:
|
||||
resp = SendMsgResponse{
|
||||
Status: "error",
|
||||
Error: "unsupported format: " + req.Format,
|
||||
}
|
||||
}
|
||||
|
||||
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",
|
||||
}
|
||||
}
|
||||
|
||||
subject := req.Title
|
||||
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,
|
||||
)
|
||||
|
||||
// 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(),
|
||||
}
|
||||
}
|
||||
|
||||
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",
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Implement SMS provider integration (Twilio, AWS SNS, etc.)
|
||||
// For now, return error
|
||||
return SendMsgResponse{
|
||||
Status: "error",
|
||||
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,
|
||||
}
|
||||
}
|
||||
@@ -1,28 +0,0 @@
|
||||
package observability
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// MetricsHandler serves Prometheus metrics
|
||||
type MetricsHandler struct {
|
||||
exporter *PrometheusExporter
|
||||
}
|
||||
|
||||
// NewMetricsHandler creates a new metrics handler
|
||||
func NewMetricsHandler(m *Metrics) *MetricsHandler {
|
||||
return &MetricsHandler{
|
||||
exporter: NewPrometheusExporter(m),
|
||||
}
|
||||
}
|
||||
|
||||
// ServeHTTP implements http.Handler for Prometheus /metrics endpoint
|
||||
func (h *MetricsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
||||
w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
|
||||
w.Header().Set("Pragma", "no-cache")
|
||||
w.Header().Set("Expires", "0")
|
||||
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(h.exporter.Export()))
|
||||
}
|
||||
@@ -28,14 +28,6 @@ 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.
|
||||
@@ -49,9 +41,6 @@ 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),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -157,113 +146,9 @@ 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()
|
||||
@@ -277,7 +162,4 @@ 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)
|
||||
}
|
||||
|
||||
@@ -1,198 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -1,160 +0,0 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/logging"
|
||||
)
|
||||
|
||||
// tokenRequest is the JSON body for POST /auth/token.
|
||||
type tokenRequest struct {
|
||||
Username string `json:"username"`
|
||||
Password string `json:"password"`
|
||||
Scope string `json:"scope,omitempty"`
|
||||
}
|
||||
|
||||
// refreshRequest is the JSON body for POST /auth/refresh.
|
||||
type refreshRequest struct {
|
||||
RefreshToken string `json:"refresh_token"`
|
||||
Scope string `json:"scope,omitempty"`
|
||||
}
|
||||
|
||||
// authClient handles token exchange with the upstream identity provider.
|
||||
// Extracted for testability — production uses http.DefaultClient,
|
||||
// tests inject a stub.
|
||||
type authClient interface {
|
||||
PostForm(url string, data url.Values) (*http.Response, error)
|
||||
}
|
||||
|
||||
// httpAuthClient wraps http.Client to implement authClient.
|
||||
type httpAuthClient struct {
|
||||
client *http.Client
|
||||
}
|
||||
|
||||
func (c *httpAuthClient) PostForm(url string, data url.Values) (*http.Response, error) {
|
||||
return c.client.PostForm(url, data)
|
||||
}
|
||||
|
||||
func newAuthClient() authClient {
|
||||
return &httpAuthClient{
|
||||
client: &http.Client{Timeout: 10 * time.Second},
|
||||
}
|
||||
}
|
||||
|
||||
// handleAuthToken exchanges username+password for a JWT via the upstream
|
||||
// identity provider's token endpoint using grant_type=password.
|
||||
func (h *Handler) handleAuthToken(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "POST" {
|
||||
writeProblemDetail(w, http.StatusMethodNotAllowed,
|
||||
"about:blank#method-not-allowed", "Method Not Allowed",
|
||||
"POST only", nil)
|
||||
return
|
||||
}
|
||||
|
||||
if h.config.Auth.TokenURL == "" || h.config.Auth.ClientID == "" {
|
||||
writeProblemDetail(w, http.StatusServiceUnavailable,
|
||||
"about:blank#not-configured", "Not Configured",
|
||||
"token endpoint not configured", nil)
|
||||
return
|
||||
}
|
||||
|
||||
var req tokenRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeProblemDetail(w, http.StatusBadRequest,
|
||||
"about:blank#bad-request", "Bad Request",
|
||||
"invalid JSON body", nil)
|
||||
return
|
||||
}
|
||||
|
||||
if req.Username == "" || req.Password == "" {
|
||||
writeProblemDetail(w, http.StatusBadRequest,
|
||||
"about:blank#bad-request", "Bad Request",
|
||||
"username and password are required", nil)
|
||||
return
|
||||
}
|
||||
|
||||
scope := req.Scope
|
||||
if scope == "" {
|
||||
scope = "openid roles permissions"
|
||||
}
|
||||
|
||||
form := url.Values{
|
||||
"grant_type": {"password"},
|
||||
"username": {req.Username},
|
||||
"password": {req.Password},
|
||||
"client_id": {h.config.Auth.ClientID},
|
||||
"client_secret": {h.config.Auth.ClientSecret},
|
||||
"scope": {scope},
|
||||
}
|
||||
|
||||
h.forwardTokenResponse(w, form, "/auth/token")
|
||||
}
|
||||
|
||||
// handleAuthRefresh exchanges a refresh token for a new JWT.
|
||||
func (h *Handler) handleAuthRefresh(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "POST" {
|
||||
writeProblemDetail(w, http.StatusMethodNotAllowed,
|
||||
"about:blank#method-not-allowed", "Method Not Allowed",
|
||||
"POST only", nil)
|
||||
return
|
||||
}
|
||||
|
||||
if h.config.Auth.TokenURL == "" || h.config.Auth.ClientID == "" {
|
||||
writeProblemDetail(w, http.StatusServiceUnavailable,
|
||||
"about:blank#not-configured", "Not Configured",
|
||||
"token endpoint not configured", nil)
|
||||
return
|
||||
}
|
||||
|
||||
var req refreshRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeProblemDetail(w, http.StatusBadRequest,
|
||||
"about:blank#bad-request", "Bad Request",
|
||||
"invalid JSON body", nil)
|
||||
return
|
||||
}
|
||||
|
||||
if req.RefreshToken == "" {
|
||||
writeProblemDetail(w, http.StatusBadRequest,
|
||||
"about:blank#bad-request", "Bad Request",
|
||||
"refresh_token is required", nil)
|
||||
return
|
||||
}
|
||||
|
||||
form := url.Values{
|
||||
"grant_type": {"refresh_token"},
|
||||
"refresh_token": {req.RefreshToken},
|
||||
"client_id": {h.config.Auth.ClientID},
|
||||
"client_secret": {h.config.Auth.ClientSecret},
|
||||
}
|
||||
|
||||
if req.Scope != "" {
|
||||
form.Set("scope", req.Scope)
|
||||
}
|
||||
|
||||
h.forwardTokenResponse(w, form, "/auth/refresh")
|
||||
}
|
||||
|
||||
// forwardTokenResponse posts form data to the token endpoint and
|
||||
// forwards the response verbatim to the client.
|
||||
func (h *Handler) forwardTokenResponse(w http.ResponseWriter, form url.Values, logPath string) {
|
||||
resp, err := h.authHTTP.PostForm(h.config.Auth.TokenURL, form)
|
||||
if err != nil {
|
||||
writeProblemDetail(w, http.StatusBadGateway,
|
||||
"about:blank#bad-gateway", "Bad Gateway",
|
||||
"identity provider unreachable", nil)
|
||||
logging.Errorf("auth upstream error", err, map[string]string{
|
||||
"path": logPath,
|
||||
})
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(resp.StatusCode)
|
||||
io.Copy(w, resp.Body)
|
||||
}
|
||||
@@ -1,315 +0,0 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
|
||||
)
|
||||
|
||||
// stubAuthClient captures the form data sent and returns a canned response.
|
||||
type stubAuthClient struct {
|
||||
lastForm url.Values
|
||||
statusCode int
|
||||
body string
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubAuthClient) PostForm(u string, data url.Values) (*http.Response, error) {
|
||||
s.lastForm = data
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
}
|
||||
return &http.Response{
|
||||
StatusCode: s.statusCode,
|
||||
Body: io.NopCloser(strings.NewReader(s.body)),
|
||||
Header: http.Header{"Content-Type": {"application/json"}},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func newTestHandler(tokenURL, clientID, clientSecret string, client authClient) *Handler {
|
||||
cfg := &config.Config{
|
||||
Auth: config.AuthConfig{
|
||||
TokenURL: tokenURL,
|
||||
ClientID: clientID,
|
||||
ClientSecret: clientSecret,
|
||||
},
|
||||
}
|
||||
h := &Handler{
|
||||
config: cfg,
|
||||
authHTTP: client,
|
||||
routes: make(map[string]*Route),
|
||||
transports: make(map[string]*http.Transport),
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func TestAuthToken_Success(t *testing.T) {
|
||||
stub := &stubAuthClient{
|
||||
statusCode: 200,
|
||||
body: `{"access_token":"jwt.token.here","refresh_token":"refresh","expires_in":3600}`,
|
||||
}
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", stub)
|
||||
|
||||
body := `{"username":"rock","password":"pass123"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if w.Code != 200 {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
// Verify form sent to upstream
|
||||
if stub.lastForm.Get("grant_type") != "password" {
|
||||
t.Errorf("grant_type = %q, want password", stub.lastForm.Get("grant_type"))
|
||||
}
|
||||
if stub.lastForm.Get("username") != "rock" {
|
||||
t.Errorf("username = %q, want rock", stub.lastForm.Get("username"))
|
||||
}
|
||||
if stub.lastForm.Get("client_id") != "api-gw" {
|
||||
t.Errorf("client_id = %q, want api-gw", stub.lastForm.Get("client_id"))
|
||||
}
|
||||
if stub.lastForm.Get("client_secret") != "secret" {
|
||||
t.Errorf("client_secret = %q, want secret", stub.lastForm.Get("client_secret"))
|
||||
}
|
||||
if stub.lastForm.Get("scope") != "openid roles permissions" {
|
||||
t.Errorf("scope = %q, want default scope", stub.lastForm.Get("scope"))
|
||||
}
|
||||
|
||||
// Verify response forwarded
|
||||
var resp map[string]interface{}
|
||||
json.NewDecoder(w.Body).Decode(&resp)
|
||||
if resp["access_token"] != "jwt.token.here" {
|
||||
t.Errorf("access_token not forwarded")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_CustomScope(t *testing.T) {
|
||||
stub := &stubAuthClient{statusCode: 200, body: `{}`}
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", stub)
|
||||
|
||||
body := `{"username":"rock","password":"pass","scope":"openid roles"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if stub.lastForm.Get("scope") != "openid roles" {
|
||||
t.Errorf("scope = %q, want custom scope", stub.lastForm.Get("scope"))
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_MissingUsername(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
|
||||
body := `{"password":"pass"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if w.Code != 400 {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_MissingPassword(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
|
||||
body := `{"username":"rock"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if w.Code != 400 {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_InvalidJSON(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader("not json"))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if w.Code != 400 {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_WrongMethod(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
|
||||
r := httptest.NewRequest("GET", "/auth/token", nil)
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if w.Code != 405 {
|
||||
t.Errorf("expected 405, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_NotConfigured(t *testing.T) {
|
||||
h := newTestHandler("", "", "", &stubAuthClient{})
|
||||
|
||||
body := `{"username":"rock","password":"pass"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if w.Code != 503 {
|
||||
t.Errorf("expected 503, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_UpstreamError(t *testing.T) {
|
||||
stub := &stubAuthClient{err: io.ErrUnexpectedEOF}
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", stub)
|
||||
|
||||
body := `{"username":"rock","password":"pass"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
if w.Code != 502 {
|
||||
t.Errorf("expected 502, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthToken_UpstreamRejectsCredentials(t *testing.T) {
|
||||
stub := &stubAuthClient{
|
||||
statusCode: 400,
|
||||
body: `{"error":"invalid_grant","error_description":"bad password"}`,
|
||||
}
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", stub)
|
||||
|
||||
body := `{"username":"rock","password":"wrong"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
// Upstream error forwarded verbatim
|
||||
if w.Code != 400 {
|
||||
t.Errorf("expected 400 (forwarded), got %d", w.Code)
|
||||
}
|
||||
if !strings.Contains(w.Body.String(), "invalid_grant") {
|
||||
t.Errorf("expected upstream error forwarded, got %s", w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// --- /auth/refresh tests ---
|
||||
|
||||
func TestAuthRefresh_Success(t *testing.T) {
|
||||
stub := &stubAuthClient{
|
||||
statusCode: 200,
|
||||
body: `{"access_token":"new.jwt","refresh_token":"new.refresh","expires_in":3600}`,
|
||||
}
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", stub)
|
||||
|
||||
body := `{"refresh_token":"old.refresh"}`
|
||||
r := httptest.NewRequest("POST", "/auth/refresh", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthRefresh(w, r)
|
||||
|
||||
if w.Code != 200 {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if stub.lastForm.Get("grant_type") != "refresh_token" {
|
||||
t.Errorf("grant_type = %q, want refresh_token", stub.lastForm.Get("grant_type"))
|
||||
}
|
||||
if stub.lastForm.Get("refresh_token") != "old.refresh" {
|
||||
t.Errorf("refresh_token not sent")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthRefresh_MissingToken(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
|
||||
r := httptest.NewRequest("POST", "/auth/refresh", strings.NewReader(`{}`))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthRefresh(w, r)
|
||||
|
||||
if w.Code != 400 {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthRefresh_WrongMethod(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
|
||||
r := httptest.NewRequest("GET", "/auth/refresh", nil)
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthRefresh(w, r)
|
||||
|
||||
if w.Code != 405 {
|
||||
t.Errorf("expected 405, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthRefresh_NotConfigured(t *testing.T) {
|
||||
h := newTestHandler("", "", "", &stubAuthClient{})
|
||||
|
||||
r := httptest.NewRequest("POST", "/auth/refresh", strings.NewReader(`{"refresh_token":"x"}`))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthRefresh(w, r)
|
||||
|
||||
if w.Code != 503 {
|
||||
t.Errorf("expected 503, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthRefresh_ExpiredToken(t *testing.T) {
|
||||
stub := &stubAuthClient{
|
||||
statusCode: 401,
|
||||
body: `{"error":"invalid_grant","error_description":"token expired"}`,
|
||||
}
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", stub)
|
||||
|
||||
r := httptest.NewRequest("POST", "/auth/refresh", strings.NewReader(`{"refresh_token":"expired"}`))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthRefresh(w, r)
|
||||
|
||||
if w.Code != 401 {
|
||||
t.Errorf("expected 401 (forwarded), got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify no credentials are leaked in response bodies
|
||||
func TestAuthToken_NoCredentialLeak(t *testing.T) {
|
||||
stub := &stubAuthClient{statusCode: 200, body: `{"access_token":"tok"}`}
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "super-secret", stub)
|
||||
|
||||
body := `{"username":"rock","password":"my-password"}`
|
||||
r := httptest.NewRequest("POST", "/auth/token", bytes.NewReader([]byte(body)))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthToken(w, r)
|
||||
|
||||
respBody := w.Body.String()
|
||||
if strings.Contains(respBody, "super-secret") {
|
||||
t.Error("client_secret leaked in response")
|
||||
}
|
||||
if strings.Contains(respBody, "my-password") {
|
||||
t.Error("password leaked in response")
|
||||
}
|
||||
}
|
||||
@@ -1,153 +0,0 @@
|
||||
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")
|
||||
}
|
||||
|
||||
|
||||
+7
-53
@@ -11,12 +11,10 @@ import (
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/identity"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/logging"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/tracing"
|
||||
)
|
||||
@@ -30,8 +28,6 @@ type Handler struct {
|
||||
config *config.Config
|
||||
// jwtValidator validates JWT tokens for authenticated endpoints
|
||||
jwtValidator *auth.Validator
|
||||
// authHTTP is the HTTP client for token exchange with the identity provider.
|
||||
authHTTP authClient
|
||||
// Default timeouts for synthesized routes (model-based dispatch)
|
||||
defaultConnectTimeout time.Duration
|
||||
defaultReadTimeout time.Duration
|
||||
@@ -89,10 +85,6 @@ func New(cfg *config.Config) *Handler {
|
||||
)
|
||||
}
|
||||
|
||||
if cfg.Auth.TokenURL != "" {
|
||||
h.authHTTP = newAuthClient()
|
||||
}
|
||||
|
||||
for name, route := range cfg.Routes {
|
||||
// Create a transport per unique upstream address for connection reuse
|
||||
transport := h.getOrCreateTransport(route.Upstream.Address, &route.Upstream)
|
||||
@@ -128,15 +120,6 @@ func (h *Handler) getOrCreateTransport(addr string, up *config.Upstream) *http.T
|
||||
dialer := &net.Dialer{
|
||||
Timeout: up.ConnectTimeout,
|
||||
KeepAlive: 30 * time.Second,
|
||||
// Issue #31: TCP_NODELAY disables Nagle's algorithm, reducing latency
|
||||
// for streaming responses by sending small packets immediately instead of
|
||||
// waiting for larger batches. Critical for low-latency LLM token streaming.
|
||||
Control: func(network, address string, c syscall.RawConn) error {
|
||||
return c.Control(func(fd uintptr) {
|
||||
// TCP_NODELAY disables Nagle's algorithm for immediate packet transmission
|
||||
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_NODELAY, 1)
|
||||
})
|
||||
},
|
||||
}
|
||||
|
||||
transport := &http.Transport{
|
||||
@@ -144,15 +127,8 @@ func (h *Handler) getOrCreateTransport(addr string, up *config.Upstream) *http.T
|
||||
DialContext: dialer.DialContext,
|
||||
MaxIdleConns: 100,
|
||||
IdleConnTimeout: 90 * time.Second,
|
||||
// Issue #32: Increase per-host connection limit to support HTTP/2 multiplexing.
|
||||
// With HTTP/2, we can serve many concurrent streams over fewer connections,
|
||||
// but we still allow more connections for better resource utilization.
|
||||
MaxConnsPerHost: 10,
|
||||
// Allow persistent connections
|
||||
DisableKeepAlives: false,
|
||||
// Issue #31: Enable HTTP/2 for client connections to support multiplexing.
|
||||
// This allows concurrent requests to stream simultaneously with better flow control.
|
||||
ForceAttemptHTTP2: true,
|
||||
}
|
||||
|
||||
// Store the upstream config for use in the handler
|
||||
@@ -251,28 +227,19 @@ func writeProblemDetail(w http.ResponseWriter, status int, problemType, title, d
|
||||
|
||||
// ServeHTTP implements http.Handler.
|
||||
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
// Auth endpoints — no JWT required (they issue tokens)
|
||||
if r.URL.Path == "/auth/token" {
|
||||
h.handleAuthToken(w, r)
|
||||
return
|
||||
}
|
||||
if r.URL.Path == "/auth/refresh" {
|
||||
h.handleAuthRefresh(w, r)
|
||||
return
|
||||
}
|
||||
if r.URL.Path == "/auth/exchange" {
|
||||
h.handleAuthExchange(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
// Handle /v1/models endpoint (no routing needed, derived from config)
|
||||
if r.URL.Path == "/v1/models" && r.Method == "GET" {
|
||||
h.handleModelsEndpoint(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
// Try to find a matching route (including body-based dispatch for /v1/chat/completions).
|
||||
// Note: /workflows endpoint is deprecated. Use X-Service: workflow + X-Resource headers instead.
|
||||
// Handle /workflows endpoint (workflow orchestration)
|
||||
if r.URL.Path == "/workflows" {
|
||||
h.handleWorkflow(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
// Try to find a matching route (including body-based dispatch for /v1/chat/completions)
|
||||
route, err := h.RouteRequest(r)
|
||||
|
||||
// Check if this is a model validation error (from body-based dispatch)
|
||||
@@ -336,10 +303,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Strip spoofed identity headers from all inbound requests.
|
||||
// Must happen before any routing — even unauthenticated paths.
|
||||
identity.StripIncoming(r)
|
||||
|
||||
// JWT Authentication for /v1/* endpoints
|
||||
if h.jwtValidator != nil && strings.HasPrefix(r.URL.Path, "/v1/") {
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
@@ -368,9 +331,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Inject identity headers for downstream services
|
||||
identity.Inject(r, claims)
|
||||
|
||||
// Check required capability if configured
|
||||
if h.config.Auth.RequiredCapability != "" {
|
||||
if !h.jwtValidator.CheckPermissions(claims, h.config.Auth.RequiredCapability, "*") {
|
||||
@@ -441,12 +401,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
// A streaming response that's continuously sending should not be cut off.
|
||||
// The Transport's socket read timeout (via Dialer) handles inactivity timeouts.
|
||||
|
||||
// For streaming responses (SSE, chunked), disable buffering to ensure events
|
||||
// reach clients immediately. Issue #33: X-Accel-Buffering:no tells nginx/Ingress
|
||||
// to stream instead of buffer. ResponseController.Flush() in upstream handler
|
||||
// pairs with this to deliver unbuffered chunks.
|
||||
w.Header().Set("X-Accel-Buffering", "no")
|
||||
|
||||
// Serve the request through the proxy
|
||||
proxy.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
@@ -476,214 +476,6 @@ func TestNoFullBuffering(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestTCPBackpressure verifies that TCP backpressure is respected during streaming.
|
||||
// When a client reads slowly, the upstream should experience backpressure on writes.
|
||||
func TestTCPBackpressure(t *testing.T) {
|
||||
// Track when upstream started writing and when each write completed
|
||||
var writeTimes []time.Time
|
||||
writesMu := sync.Mutex{}
|
||||
|
||||
upstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("X-Accel-Buffering", "no") // Issue #33: disable buffering
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
rc := http.NewResponseController(w)
|
||||
// Send many events to trigger backpressure
|
||||
for i := 0; i < 20; i++ {
|
||||
writesMu.Lock()
|
||||
writeTimes = append(writeTimes, time.Now())
|
||||
writesMu.Unlock()
|
||||
|
||||
fmt.Fprintf(w, "data: event%d\n\n", i)
|
||||
if err := rc.Flush(); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}))
|
||||
defer upstreamServer.Close()
|
||||
|
||||
upstreamAddr := strings.TrimPrefix(upstreamServer.URL, "http://")
|
||||
|
||||
cfg := &config.Config{
|
||||
Routes: map[string]*config.Route{
|
||||
"backpressure-route": {
|
||||
Name: "backpressure-route",
|
||||
Upstream: config.Upstream{
|
||||
Address: upstreamAddr,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 5 * time.Second,
|
||||
MaxBodySize: 1024 * 1024,
|
||||
AuthRequired: false,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
defer handler.Close()
|
||||
|
||||
server := httptest.NewServer(handler)
|
||||
defer server.Close()
|
||||
|
||||
resp, err := http.Get(server.URL + "/backpressure")
|
||||
if err != nil {
|
||||
t.Fatalf("request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// Verify X-Accel-Buffering header is passed through
|
||||
if resp.Header.Get("X-Accel-Buffering") != "no" {
|
||||
t.Errorf("X-Accel-Buffering header not propagated, got: %s", resp.Header.Get("X-Accel-Buffering"))
|
||||
}
|
||||
|
||||
// Read events with simulated slow client (small buffer)
|
||||
reader := bufio.NewReader(resp.Body)
|
||||
readStart := time.Now()
|
||||
eventCount := 0
|
||||
|
||||
for {
|
||||
line, err := reader.ReadString('\n')
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
t.Fatalf("read failed: %v", err)
|
||||
}
|
||||
|
||||
if strings.HasPrefix(strings.TrimSpace(line), "data:") {
|
||||
eventCount++
|
||||
// Simulate slow client by adding delay
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify we got all events
|
||||
if eventCount != 20 {
|
||||
t.Errorf("expected 20 events, got %d", eventCount)
|
||||
}
|
||||
|
||||
// Total read time should be roughly eventCount * readDelay
|
||||
// indicating backpressure was applied (upstream couldn't send all at once)
|
||||
elapsed := time.Since(readStart)
|
||||
expectedMin := time.Duration(20*5) * time.Millisecond
|
||||
if elapsed < expectedMin {
|
||||
t.Logf("backpressure test: elapsed=%.0fms (expected ~%.0fms)", elapsed.Seconds()*1000, expectedMin.Seconds()*1000)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrentSSEStreams verifies that HTTP/2 multiplexing handles multiple concurrent streams.
|
||||
// Issue #32: Multiple LLM requests should not block each other.
|
||||
func TestConcurrentSSEStreams(t *testing.T) {
|
||||
upstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("X-Accel-Buffering", "no")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
rc := http.NewResponseController(w)
|
||||
// Each request sends unique identifier
|
||||
reqID := r.URL.Query().Get("id")
|
||||
for i := 0; i < 5; i++ {
|
||||
fmt.Fprintf(w, "data: [%s] event %d\n\n", reqID, i)
|
||||
if err := rc.Flush(); err != nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}))
|
||||
defer upstreamServer.Close()
|
||||
|
||||
upstreamAddr := strings.TrimPrefix(upstreamServer.URL, "http://")
|
||||
|
||||
cfg := &config.Config{
|
||||
Routes: map[string]*config.Route{
|
||||
"concurrent-route": {
|
||||
Name: "concurrent-route",
|
||||
Upstream: config.Upstream{
|
||||
Address: upstreamAddr,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 5 * time.Second,
|
||||
MaxBodySize: 1024 * 1024,
|
||||
AuthRequired: false,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
defer handler.Close()
|
||||
|
||||
server := httptest.NewServer(handler)
|
||||
defer server.Close()
|
||||
|
||||
// Launch multiple concurrent requests
|
||||
var wg sync.WaitGroup
|
||||
results := make(map[string][]string)
|
||||
resultsMu := sync.Mutex{}
|
||||
|
||||
for id := 0; id < 3; id++ {
|
||||
wg.Add(1)
|
||||
go func(streamID int) {
|
||||
defer wg.Done()
|
||||
|
||||
url := fmt.Sprintf("%s/concurrent?id=stream%d", server.URL, streamID)
|
||||
resp, err := http.Get(url)
|
||||
if err != nil {
|
||||
t.Errorf("request failed: %v", err)
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
reader := bufio.NewReader(resp.Body)
|
||||
var events []string
|
||||
|
||||
for {
|
||||
line, err := reader.ReadString('\n')
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
t.Errorf("read failed: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
line = strings.TrimSpace(line)
|
||||
if strings.HasPrefix(line, "data:") {
|
||||
events = append(events, line)
|
||||
}
|
||||
}
|
||||
|
||||
resultsMu.Lock()
|
||||
results[fmt.Sprintf("stream%d", streamID)] = events
|
||||
resultsMu.Unlock()
|
||||
}(id)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
// Verify all streams got their events
|
||||
for i := 0; i < 3; i++ {
|
||||
key := fmt.Sprintf("stream%d", i)
|
||||
events, ok := results[key]
|
||||
if !ok {
|
||||
t.Errorf("stream%d: no results", i)
|
||||
continue
|
||||
}
|
||||
if len(events) != 5 {
|
||||
t.Errorf("stream%d: expected 5 events, got %d", i, len(events))
|
||||
}
|
||||
|
||||
// Verify all events belong to this stream
|
||||
for _, event := range events {
|
||||
if !strings.Contains(event, key) {
|
||||
t.Errorf("stream%d: event from wrong stream: %s", i, event)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestClientDisconnectCancelsUpstream verifies that when a client closes mid-stream,
|
||||
// the upstream request context is cancelled immediately and no goroutines are leaked.
|
||||
func TestClientDisconnectCancelsUpstream(t *testing.T) {
|
||||
|
||||
@@ -1,229 +0,0 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/logging"
|
||||
)
|
||||
|
||||
// exchangeRequest represents a token exchange request (RFC 8693 subset).
|
||||
type exchangeRequest struct {
|
||||
SubjectToken string `json:"subject_token"`
|
||||
ClientID string `json:"client_id"`
|
||||
ClientSecret string `json:"client_secret"`
|
||||
Scope string `json:"scope,omitempty"`
|
||||
Resource string `json:"resource,omitempty"`
|
||||
}
|
||||
|
||||
// exchangeResponse wraps the service token with subject identity metadata.
|
||||
type exchangeResponse struct {
|
||||
AccessToken string `json:"access_token"`
|
||||
TokenType string `json:"token_type"`
|
||||
ExpiresIn int `json:"expires_in"`
|
||||
IssuedTokenType string `json:"issued_token_type,omitempty"`
|
||||
Subject string `json:"subject,omitempty"`
|
||||
ActingParty string `json:"acting_party,omitempty"`
|
||||
GrantedScope string `json:"scope,omitempty"`
|
||||
}
|
||||
|
||||
// handleAuthExchange implements token exchange: a service presents a user's
|
||||
// JWT and its own credentials to get a scoped service token with the user's
|
||||
// identity attached.
|
||||
//
|
||||
// Flow:
|
||||
// 1. Validate subject_token (user's JWT) — signature, expiry, issuer
|
||||
// 2. Authenticate service via client_credentials against Authentik
|
||||
// 3. Verify requested scope is a subset of service's roles
|
||||
// 4. Return service token + subject identity metadata
|
||||
func (h *Handler) handleAuthExchange(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != "POST" {
|
||||
writeProblemDetail(w, http.StatusMethodNotAllowed,
|
||||
"about:blank#method-not-allowed", "Method Not Allowed",
|
||||
"POST only", nil)
|
||||
return
|
||||
}
|
||||
|
||||
if h.jwtValidator == nil || h.config.Auth.TokenURL == "" {
|
||||
writeProblemDetail(w, http.StatusServiceUnavailable,
|
||||
"about:blank#not-configured", "Not Configured",
|
||||
"token exchange not configured", nil)
|
||||
return
|
||||
}
|
||||
|
||||
var req exchangeRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeProblemDetail(w, http.StatusBadRequest,
|
||||
"about:blank#bad-request", "Bad Request",
|
||||
"invalid JSON body", nil)
|
||||
return
|
||||
}
|
||||
|
||||
if req.SubjectToken == "" || req.ClientID == "" || req.ClientSecret == "" {
|
||||
writeProblemDetail(w, http.StatusBadRequest,
|
||||
"about:blank#bad-request", "Bad Request",
|
||||
"subject_token, client_id, and client_secret are required", nil)
|
||||
return
|
||||
}
|
||||
|
||||
// Step 1: Validate subject token
|
||||
subjectClaims, err := h.jwtValidator.ValidateBearerToken("Bearer " + req.SubjectToken)
|
||||
if err != nil {
|
||||
writeProblemDetail(w, http.StatusForbidden,
|
||||
"about:blank#invalid-subject-token", "Invalid Subject Token",
|
||||
fmt.Sprintf("subject token validation failed: %v", err), nil)
|
||||
logging.Errorf("token exchange: invalid subject", err, map[string]string{
|
||||
"path": "/auth/exchange",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
subject := claimStr(subjectClaims, "sub")
|
||||
|
||||
// Step 2: Authenticate service via client_credentials
|
||||
form := url.Values{
|
||||
"grant_type": {"client_credentials"},
|
||||
"client_id": {req.ClientID},
|
||||
"client_secret": {req.ClientSecret},
|
||||
"scope": {"openid roles"},
|
||||
}
|
||||
|
||||
resp, err := h.authHTTP.PostForm(h.config.Auth.TokenURL, form)
|
||||
if err != nil {
|
||||
writeProblemDetail(w, http.StatusBadGateway,
|
||||
"about:blank#bad-gateway", "Bad Gateway",
|
||||
"identity provider unreachable", nil)
|
||||
logging.Errorf("token exchange: upstream error", err, map[string]string{
|
||||
"path": "/auth/exchange",
|
||||
})
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
writeProblemDetail(w, http.StatusForbidden,
|
||||
"about:blank#invalid-actor", "Invalid Actor Credentials",
|
||||
fmt.Sprintf("service authentication failed (HTTP %d)", resp.StatusCode), nil)
|
||||
return
|
||||
}
|
||||
|
||||
var tokenResp struct {
|
||||
AccessToken string `json:"access_token"`
|
||||
ExpiresIn int `json:"expires_in"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&tokenResp); err != nil {
|
||||
writeProblemDetail(w, http.StatusBadGateway,
|
||||
"about:blank#bad-gateway", "Bad Gateway",
|
||||
"invalid response from identity provider", nil)
|
||||
return
|
||||
}
|
||||
|
||||
// Step 3: Decode service token to check roles
|
||||
serviceRoles, err := extractRolesFromJWT(tokenResp.AccessToken)
|
||||
if err != nil {
|
||||
writeProblemDetail(w, http.StatusBadGateway,
|
||||
"about:blank#bad-gateway", "Bad Gateway",
|
||||
"cannot decode service token", nil)
|
||||
return
|
||||
}
|
||||
|
||||
if req.Scope != "" && !scopeSubset(req.Scope, serviceRoles) {
|
||||
writeProblemDetail(w, http.StatusForbidden,
|
||||
"about:blank#scope-escalation", "Scope Escalation Denied",
|
||||
fmt.Sprintf("requested scope %q exceeds service roles %v", req.Scope, serviceRoles), nil)
|
||||
return
|
||||
}
|
||||
|
||||
grantedScope := req.Scope
|
||||
if grantedScope == "" {
|
||||
grantedScope = strings.Join(serviceRoles, " ")
|
||||
}
|
||||
|
||||
// Step 4: Return service token with subject metadata
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
json.NewEncoder(w).Encode(exchangeResponse{
|
||||
AccessToken: tokenResp.AccessToken,
|
||||
TokenType: "Bearer",
|
||||
ExpiresIn: tokenResp.ExpiresIn,
|
||||
IssuedTokenType: "urn:ietf:params:oauth:token-type:access_token",
|
||||
Subject: subject,
|
||||
ActingParty: req.ClientID,
|
||||
GrantedScope: grantedScope,
|
||||
})
|
||||
}
|
||||
|
||||
// scopeSubset checks that every space-separated scope token is in allowed roles.
|
||||
func scopeSubset(requested string, allowed []string) bool {
|
||||
allowedSet := make(map[string]bool, len(allowed))
|
||||
for _, r := range allowed {
|
||||
allowedSet[r] = true
|
||||
}
|
||||
if allowedSet["*"] {
|
||||
return true
|
||||
}
|
||||
for _, s := range strings.Fields(requested) {
|
||||
if !allowedSet[s] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// extractRolesFromJWT decodes the payload of a JWT without verification
|
||||
// and returns the "roles" claim. Used after the token was already obtained
|
||||
// from a trusted source (Authentik client_credentials response).
|
||||
func extractRolesFromJWT(token string) ([]string, error) {
|
||||
parts := strings.SplitN(token, ".", 3)
|
||||
if len(parts) != 3 {
|
||||
return nil, fmt.Errorf("invalid JWT format")
|
||||
}
|
||||
|
||||
payload, err := base64.RawURLEncoding.DecodeString(parts[1])
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("base64 decode failed: %w", err)
|
||||
}
|
||||
|
||||
var claims map[string]interface{}
|
||||
if err := json.Unmarshal(payload, &claims); err != nil {
|
||||
return nil, fmt.Errorf("JSON decode failed: %w", err)
|
||||
}
|
||||
|
||||
return claimStrSlice(claims, "roles"), nil
|
||||
}
|
||||
|
||||
// claimStr extracts a string claim.
|
||||
func claimStr(claims map[string]interface{}, key string) string {
|
||||
v, ok := claims[key]
|
||||
if !ok || v == nil {
|
||||
return ""
|
||||
}
|
||||
s, ok := v.(string)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// claimStrSlice extracts a string slice from a JSON-deserialized []interface{}.
|
||||
func claimStrSlice(claims map[string]interface{}, key string) []string {
|
||||
v, ok := claims[key]
|
||||
if !ok || v == nil {
|
||||
return nil
|
||||
}
|
||||
raw, ok := v.([]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(raw))
|
||||
for _, item := range raw {
|
||||
if s, ok := item.(string); ok && s != "" {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -1,215 +0,0 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
|
||||
)
|
||||
|
||||
// fakeJWT creates a JWT-shaped string (header.payload.signature) with given claims.
|
||||
// Not cryptographically signed — used only with stubbed validators.
|
||||
func fakeJWT(claims map[string]interface{}) string {
|
||||
header := base64.RawURLEncoding.EncodeToString([]byte(`{"alg":"none","typ":"JWT"}`))
|
||||
payload, _ := json.Marshal(claims)
|
||||
payloadB64 := base64.RawURLEncoding.EncodeToString(payload)
|
||||
return header + "." + payloadB64 + ".fakesig"
|
||||
}
|
||||
|
||||
// stubJWTValidator returns claims from a pre-set map keyed by token.
|
||||
type stubJWTValidator struct {
|
||||
tokens map[string]map[string]interface{}
|
||||
}
|
||||
|
||||
func (s *stubJWTValidator) ValidateBearerToken(authHeader string) (map[string]interface{}, error) {
|
||||
token := strings.TrimPrefix(authHeader, "Bearer ")
|
||||
if claims, ok := s.tokens[token]; ok {
|
||||
return claims, nil
|
||||
}
|
||||
return nil, fmt.Errorf("invalid token")
|
||||
}
|
||||
|
||||
func (s *stubJWTValidator) CheckPermissions(claims map[string]interface{}, required ...string) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// We can't use stubJWTValidator directly because Handler expects *auth.Validator.
|
||||
// Instead, test via the endpoint with a real JWKS server or test the helpers directly.
|
||||
|
||||
func TestScopeSubset(t *testing.T) {
|
||||
tests := []struct {
|
||||
requested string
|
||||
allowed []string
|
||||
want bool
|
||||
}{
|
||||
{"memory:read", []string{"llm:inference", "memory:read"}, true},
|
||||
{"memory:read memory:write", []string{"memory:read", "memory:write"}, true},
|
||||
{"memory:write", []string{"memory:read"}, false},
|
||||
{"admin:*", []string{"memory:read"}, false},
|
||||
{"anything", []string{"*"}, true},
|
||||
{"", []string{"memory:read"}, true},
|
||||
{"memory:read", []string{}, false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
got := scopeSubset(tt.requested, tt.allowed)
|
||||
if got != tt.want {
|
||||
t.Errorf("scopeSubset(%q, %v) = %v, want %v", tt.requested, tt.allowed, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRolesFromJWT(t *testing.T) {
|
||||
token := fakeJWT(map[string]interface{}{
|
||||
"roles": []interface{}{"llm:inference", "memory:read"},
|
||||
"sub": "test-user",
|
||||
})
|
||||
|
||||
roles, err := extractRolesFromJWT(token)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if len(roles) != 2 || roles[0] != "llm:inference" || roles[1] != "memory:read" {
|
||||
t.Errorf("roles = %v, want [llm:inference memory:read]", roles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRolesFromJWT_InvalidFormat(t *testing.T) {
|
||||
_, err := extractRolesFromJWT("not-a-jwt")
|
||||
if err == nil {
|
||||
t.Error("expected error for invalid JWT")
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRolesFromJWT_NoRoles(t *testing.T) {
|
||||
token := fakeJWT(map[string]interface{}{"sub": "user"})
|
||||
roles, err := extractRolesFromJWT(token)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if roles != nil {
|
||||
t.Errorf("expected nil roles, got %v", roles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAuthExchange_WrongMethod(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
h.jwtValidator = auth.NewValidator("", "", "")
|
||||
|
||||
r := httptest.NewRequest("GET", "/auth/exchange", nil)
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthExchange(w, r)
|
||||
|
||||
if w.Code != 405 {
|
||||
t.Errorf("expected 405, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAuthExchange_NotConfigured(t *testing.T) {
|
||||
h := &Handler{
|
||||
config: &config.Config{},
|
||||
routes: make(map[string]*Route),
|
||||
transports: make(map[string]*http.Transport),
|
||||
}
|
||||
|
||||
body := `{"subject_token":"x","client_id":"y","client_secret":"z"}`
|
||||
r := httptest.NewRequest("POST", "/auth/exchange", strings.NewReader(body))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthExchange(w, r)
|
||||
|
||||
if w.Code != 503 {
|
||||
t.Errorf("expected 503, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAuthExchange_MissingFields(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
h.jwtValidator = auth.NewValidator("", "", "")
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
body string
|
||||
}{
|
||||
{"missing subject", `{"client_id":"x","client_secret":"y"}`},
|
||||
{"missing client_id", `{"subject_token":"x","client_secret":"y"}`},
|
||||
{"missing client_secret", `{"subject_token":"x","client_id":"y"}`},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
r := httptest.NewRequest("POST", "/auth/exchange", strings.NewReader(tt.body))
|
||||
w := httptest.NewRecorder()
|
||||
h.handleAuthExchange(w, r)
|
||||
if w.Code != 400 {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAuthExchange_InvalidJSON(t *testing.T) {
|
||||
h := newTestHandler("https://auth.example.com/token/", "api-gw", "secret", &stubAuthClient{})
|
||||
h.jwtValidator = auth.NewValidator("", "", "")
|
||||
|
||||
r := httptest.NewRequest("POST", "/auth/exchange", strings.NewReader("not json"))
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
h.handleAuthExchange(w, r)
|
||||
|
||||
if w.Code != 400 {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAuthExchange_ScopeEscalation(t *testing.T) {
|
||||
// Test scopeSubset directly since full integration needs JWKS
|
||||
if scopeSubset("admin:delete", []string{"memory:read", "memory:write"}) {
|
||||
t.Error("scope escalation should be denied")
|
||||
}
|
||||
if !scopeSubset("memory:read", []string{"memory:read", "memory:write"}) {
|
||||
t.Error("valid scope should be allowed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestClaimStr(t *testing.T) {
|
||||
claims := map[string]interface{}{"sub": "user-1", "num": 42, "nil": nil}
|
||||
if got := claimStr(claims, "sub"); got != "user-1" {
|
||||
t.Errorf("claimStr(sub) = %q, want user-1", got)
|
||||
}
|
||||
if got := claimStr(claims, "num"); got != "" {
|
||||
t.Errorf("claimStr(num) = %q, want empty", got)
|
||||
}
|
||||
if got := claimStr(claims, "nil"); got != "" {
|
||||
t.Errorf("claimStr(nil) = %q, want empty", got)
|
||||
}
|
||||
if got := claimStr(claims, "missing"); got != "" {
|
||||
t.Errorf("claimStr(missing) = %q, want empty", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClaimStrSlice(t *testing.T) {
|
||||
claims := map[string]interface{}{
|
||||
"roles": []interface{}{"a", "b", "", nil, 42},
|
||||
"empty": []interface{}{},
|
||||
"str": "not-a-slice",
|
||||
}
|
||||
if got := claimStrSlice(claims, "roles"); len(got) != 2 || got[0] != "a" || got[1] != "b" {
|
||||
t.Errorf("claimStrSlice(roles) = %v, want [a b]", got)
|
||||
}
|
||||
if got := claimStrSlice(claims, "empty"); len(got) != 0 {
|
||||
t.Errorf("claimStrSlice(empty) = %v, want empty", got)
|
||||
}
|
||||
if got := claimStrSlice(claims, "str"); got != nil {
|
||||
t.Errorf("claimStrSlice(str) = %v, want nil", got)
|
||||
}
|
||||
if got := claimStrSlice(claims, "missing"); got != nil {
|
||||
t.Errorf("claimStrSlice(missing) = %v, want nil", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,484 @@
|
||||
// Package proxy provides request routing and forwarding.
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
// WorkflowRequest represents a workflow execution request
|
||||
type WorkflowRequest struct {
|
||||
// Workflow ID or name
|
||||
Workflow string `json:"workflow"`
|
||||
|
||||
// Input parameters for the workflow
|
||||
Input map[string]interface{} `json:"input"`
|
||||
|
||||
// Optional: timeout in seconds
|
||||
Timeout int `json:"timeout,omitempty"`
|
||||
|
||||
// Optional: wait for result (default: true)
|
||||
Wait *bool `json:"wait,omitempty"`
|
||||
}
|
||||
|
||||
// WorkflowResponse represents the response from workflow execution
|
||||
type WorkflowResponse struct {
|
||||
// Workflow execution ID
|
||||
ID string `json:"id"`
|
||||
|
||||
// Workflow name
|
||||
Workflow string `json:"workflow"`
|
||||
|
||||
// Execution status: pending, running, completed, failed
|
||||
Status string `json:"status"`
|
||||
|
||||
// Output of the workflow
|
||||
Output interface{} `json:"output,omitempty"`
|
||||
|
||||
// Error message if workflow failed
|
||||
Error string `json:"error,omitempty"`
|
||||
|
||||
// Timestamp when workflow was created
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
|
||||
// Timestamp when workflow completed
|
||||
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
||||
}
|
||||
|
||||
// PredefinedWorkflow defines a workflow template that combines multiple API calls
|
||||
type PredefinedWorkflow struct {
|
||||
Name string
|
||||
Description string
|
||||
Handler func(*http.Request, *Handler, map[string]interface{}) (interface{}, error)
|
||||
}
|
||||
|
||||
// handleWorkflow handles the /workflows endpoint
|
||||
// It accepts workflow definitions and orchestrates API calls
|
||||
func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) {
|
||||
// Only POST is supported
|
||||
if r.Method != "POST" {
|
||||
w.Header().Set("Content-Type", "application/problem+json")
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
fmt.Fprintf(w, `{"type":"https://api.example.com/problems/method-not-allowed","title":"Method Not Allowed","status":405,"detail":"Only POST is supported for /workflows"}`)
|
||||
return
|
||||
}
|
||||
|
||||
// Parse request body
|
||||
var workflowReq WorkflowRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&workflowReq); err != nil {
|
||||
writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/invalid-workflow-request", "Invalid Workflow Request", "Failed to parse workflow request: "+err.Error(), nil)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate workflow name
|
||||
if workflowReq.Workflow == "" {
|
||||
writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/missing-workflow", "Missing Workflow", "The 'workflow' field is required", nil)
|
||||
return
|
||||
}
|
||||
|
||||
// Get predefined workflow
|
||||
workflow, ok := h.getWorkflow(workflowReq.Workflow)
|
||||
if !ok {
|
||||
availableWorkflows := h.getAvailableWorkflows()
|
||||
writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/unknown-workflow", "Unknown Workflow", fmt.Sprintf("Workflow %q is not available", workflowReq.Workflow), availableWorkflows)
|
||||
return
|
||||
}
|
||||
|
||||
// Default wait to true
|
||||
wait := true
|
||||
if workflowReq.Wait != nil {
|
||||
wait = *workflowReq.Wait
|
||||
}
|
||||
|
||||
// Set default timeout if not provided
|
||||
timeout := time.Duration(30) * time.Second
|
||||
if workflowReq.Timeout > 0 {
|
||||
timeout = time.Duration(workflowReq.Timeout) * time.Second
|
||||
}
|
||||
|
||||
// Create a context with timeout for workflow execution
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
// Execute workflow
|
||||
output, err := workflow.Handler(r.WithContext(ctx), h, workflowReq.Input)
|
||||
|
||||
// Build response
|
||||
workflowResp := WorkflowResponse{
|
||||
ID: generateWorkflowID(),
|
||||
Workflow: workflowReq.Workflow,
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
workflowResp.Status = "failed"
|
||||
workflowResp.Error = err.Error()
|
||||
} else {
|
||||
if wait {
|
||||
workflowResp.Status = "completed"
|
||||
workflowResp.Output = output
|
||||
now := time.Now()
|
||||
workflowResp.CompletedAt = &now
|
||||
} else {
|
||||
workflowResp.Status = "pending"
|
||||
}
|
||||
}
|
||||
|
||||
// Write response
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
} else {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
json.NewEncoder(w).Encode(workflowResp)
|
||||
}
|
||||
|
||||
// getWorkflow returns a predefined workflow by name
|
||||
func (h *Handler) getWorkflow(name string) (*PredefinedWorkflow, bool) {
|
||||
workflows := h.getPredefinedWorkflows()
|
||||
for _, wf := range workflows {
|
||||
if wf.Name == name {
|
||||
return &wf, true
|
||||
}
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// getPredefinedWorkflows returns all available workflows
|
||||
func (h *Handler) getPredefinedWorkflows() []PredefinedWorkflow {
|
||||
return []PredefinedWorkflow{
|
||||
{
|
||||
Name: "chat-and-embed",
|
||||
Description: "Chat with a model and then embed the response",
|
||||
Handler: h.chatAndEmbedWorkflow,
|
||||
},
|
||||
{
|
||||
Name: "multi-model-chat",
|
||||
Description: "Chat with multiple models sequentially",
|
||||
Handler: h.multiModelChatWorkflow,
|
||||
},
|
||||
{
|
||||
Name: "rag-pipeline",
|
||||
Description: "RAG pipeline: embed query, rerank, then chat with context",
|
||||
Handler: h.ragPipelineWorkflow,
|
||||
},
|
||||
{
|
||||
Name: "batch-embeddings",
|
||||
Description: "Generate embeddings for multiple texts",
|
||||
Handler: h.batchEmbeddingsWorkflow,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// getAvailableWorkflows returns a list of available workflow names
|
||||
func (h *Handler) getAvailableWorkflows() []string {
|
||||
workflows := h.getPredefinedWorkflows()
|
||||
names := make([]string, len(workflows))
|
||||
for i, wf := range workflows {
|
||||
names[i] = wf.Name
|
||||
}
|
||||
return names
|
||||
}
|
||||
|
||||
// Workflow implementations
|
||||
|
||||
// chatAndEmbedWorkflow: Chat with a model, then embed the response
|
||||
func (h *Handler) chatAndEmbedWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
model, ok := input["model"].(string)
|
||||
if !ok || model == "" {
|
||||
return nil, fmt.Errorf("missing required parameter: model")
|
||||
}
|
||||
|
||||
embedModel, ok := input["embed_model"].(string)
|
||||
if !ok {
|
||||
embedModel = "nomic-ai/nomic-embed-text-v2-moe"
|
||||
}
|
||||
|
||||
messages, ok := input["messages"].([]interface{})
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("missing required parameter: messages")
|
||||
}
|
||||
|
||||
// Step 1: Chat
|
||||
chatReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
chatBody, _ := json.Marshal(chatReq)
|
||||
chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody)))
|
||||
chatHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
// Create a response writer to capture the chat response
|
||||
chatResp := &responseCapture{}
|
||||
handler.ServeHTTP(chatResp, chatHTTPReq)
|
||||
|
||||
var chatResult map[string]interface{}
|
||||
if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse chat response: %v", err)
|
||||
}
|
||||
|
||||
// Extract message content
|
||||
var messageContent string
|
||||
if choices, ok := chatResult["choices"].([]interface{}); ok && len(choices) > 0 {
|
||||
if choice, ok := choices[0].(map[string]interface{}); ok {
|
||||
if message, ok := choice["message"].(map[string]interface{}); ok {
|
||||
if content, ok := message["content"].(string); ok {
|
||||
messageContent = content
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Step 2: Embed the response
|
||||
embedReq := map[string]interface{}{
|
||||
"model": embedModel,
|
||||
"input": messageContent,
|
||||
}
|
||||
|
||||
embedBody, _ := json.Marshal(embedReq)
|
||||
embedHTTPReq, _ := http.NewRequest("POST", "/v1/embeddings", io.NopCloser(bytes.NewReader(embedBody)))
|
||||
embedHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
embedResp := &responseCapture{}
|
||||
handler.ServeHTTP(embedResp, embedHTTPReq)
|
||||
|
||||
var embedResult map[string]interface{}
|
||||
if err := json.Unmarshal(embedResp.body.Bytes(), &embedResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse embedding response: %v", err)
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"chat_response": chatResult,
|
||||
"embedding_response": embedResult,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// multiModelChatWorkflow: Chat with multiple models sequentially
|
||||
func (h *Handler) multiModelChatWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
models, ok := input["models"].([]interface{})
|
||||
if !ok || len(models) == 0 {
|
||||
return nil, fmt.Errorf("missing required parameter: models (array)")
|
||||
}
|
||||
|
||||
messages, ok := input["messages"].([]interface{})
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("missing required parameter: messages")
|
||||
}
|
||||
|
||||
results := make([]map[string]interface{}, 0)
|
||||
|
||||
for _, modelInterface := range models {
|
||||
model, ok := modelInterface.(string)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
chatReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
chatBody, _ := json.Marshal(chatReq)
|
||||
chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody)))
|
||||
chatHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
chatResp := &responseCapture{}
|
||||
handler.ServeHTTP(chatResp, chatHTTPReq)
|
||||
|
||||
var chatResult map[string]interface{}
|
||||
if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil {
|
||||
results = append(results, map[string]interface{}{
|
||||
"model": model,
|
||||
"error": err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
results = append(results, map[string]interface{}{
|
||||
"model": model,
|
||||
"result": chatResult,
|
||||
})
|
||||
}
|
||||
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// ragPipelineWorkflow: RAG pipeline - embed query, rerank, chat with context
|
||||
func (h *Handler) ragPipelineWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
query, ok := input["query"].(string)
|
||||
if !ok || query == "" {
|
||||
return nil, fmt.Errorf("missing required parameter: query")
|
||||
}
|
||||
|
||||
documents, ok := input["documents"].([]interface{})
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("missing required parameter: documents")
|
||||
}
|
||||
|
||||
model, ok := input["model"].(string)
|
||||
if !ok {
|
||||
model = "reasoning"
|
||||
}
|
||||
|
||||
rerankModel, ok := input["rerank_model"].(string)
|
||||
if !ok {
|
||||
rerankModel = "BAAI/bge-reranker-base"
|
||||
}
|
||||
|
||||
topK := 3
|
||||
if tk, ok := input["top_k"].(float64); ok {
|
||||
topK = int(tk)
|
||||
}
|
||||
|
||||
// Step 1: Rerank documents based on query
|
||||
rerankReq := map[string]interface{}{
|
||||
"model": rerankModel,
|
||||
"query": query,
|
||||
"texts": documents,
|
||||
"top_k": topK,
|
||||
}
|
||||
|
||||
rerankBody, _ := json.Marshal(rerankReq)
|
||||
rerankHTTPReq, _ := http.NewRequest("POST", "/v1/rerank", io.NopCloser(bytes.NewReader(rerankBody)))
|
||||
rerankHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
rerankResp := &responseCapture{}
|
||||
handler.ServeHTTP(rerankResp, rerankHTTPReq)
|
||||
|
||||
var rerankResult map[string]interface{}
|
||||
if err := json.Unmarshal(rerankResp.body.Bytes(), &rerankResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse rerank response: %v", err)
|
||||
}
|
||||
|
||||
// Extract top documents
|
||||
var topDocs []string
|
||||
if results, ok := rerankResult["results"].([]interface{}); ok {
|
||||
for i, resultInterface := range results {
|
||||
if i >= topK {
|
||||
break
|
||||
}
|
||||
if result, ok := resultInterface.(map[string]interface{}); ok {
|
||||
if text, ok := result["text"].(string); ok {
|
||||
topDocs = append(topDocs, text)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Step 2: Chat with context
|
||||
context := fmt.Sprintf("Context from documents:\n%v\n\nQuery: %s", topDocs, query)
|
||||
|
||||
chatReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{
|
||||
"role": "user",
|
||||
"content": context,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
chatBody, _ := json.Marshal(chatReq)
|
||||
chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody)))
|
||||
chatHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
chatResp := &responseCapture{}
|
||||
handler.ServeHTTP(chatResp, chatHTTPReq)
|
||||
|
||||
var chatResult map[string]interface{}
|
||||
if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse chat response: %v", err)
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"reranked_documents": topDocs,
|
||||
"chat_response": chatResult,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// batchEmbeddingsWorkflow: Generate embeddings for multiple texts
|
||||
func (h *Handler) batchEmbeddingsWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) {
|
||||
texts, ok := input["texts"].([]interface{})
|
||||
if !ok || len(texts) == 0 {
|
||||
return nil, fmt.Errorf("missing required parameter: texts (array)")
|
||||
}
|
||||
|
||||
model, ok := input["model"].(string)
|
||||
if !ok {
|
||||
model = "nomic-ai/nomic-embed-text-v2-moe"
|
||||
}
|
||||
|
||||
// Convert interface{} to []string
|
||||
textStrings := make([]string, 0)
|
||||
for _, t := range texts {
|
||||
if str, ok := t.(string); ok {
|
||||
textStrings = append(textStrings, str)
|
||||
}
|
||||
}
|
||||
|
||||
if len(textStrings) == 0 {
|
||||
return nil, fmt.Errorf("no valid text strings in texts array")
|
||||
}
|
||||
|
||||
embedReq := map[string]interface{}{
|
||||
"model": model,
|
||||
"input": textStrings,
|
||||
}
|
||||
|
||||
embedBody, _ := json.Marshal(embedReq)
|
||||
embedHTTPReq, _ := http.NewRequest("POST", "/v1/embeddings", io.NopCloser(bytes.NewReader(embedBody)))
|
||||
embedHTTPReq.Header.Set("Content-Type", "application/json")
|
||||
|
||||
embedResp := &responseCapture{}
|
||||
handler.ServeHTTP(embedResp, embedHTTPReq)
|
||||
|
||||
var embedResult map[string]interface{}
|
||||
if err := json.Unmarshal(embedResp.body.Bytes(), &embedResult); err != nil {
|
||||
return nil, fmt.Errorf("failed to parse embedding response: %v", err)
|
||||
}
|
||||
|
||||
return embedResult, nil
|
||||
}
|
||||
|
||||
// Utility functions
|
||||
|
||||
// responseCapture captures HTTP response for reuse within workflows
|
||||
type responseCapture struct {
|
||||
status int
|
||||
header http.Header
|
||||
body bytes.Buffer
|
||||
}
|
||||
|
||||
func (w *responseCapture) Header() http.Header {
|
||||
if w.header == nil {
|
||||
w.header = make(http.Header)
|
||||
}
|
||||
return w.header
|
||||
}
|
||||
|
||||
func (w *responseCapture) Write(b []byte) (int, error) {
|
||||
if w.status == 0 {
|
||||
w.status = http.StatusOK
|
||||
}
|
||||
return w.body.Write(b)
|
||||
}
|
||||
|
||||
func (w *responseCapture) WriteHeader(statusCode int) {
|
||||
if w.status == 0 {
|
||||
w.status = statusCode
|
||||
}
|
||||
}
|
||||
|
||||
// generateWorkflowID generates a unique workflow execution ID
|
||||
func generateWorkflowID() string {
|
||||
return fmt.Sprintf("wf_%d", time.Now().UnixNano())
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,258 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/config"
|
||||
)
|
||||
|
||||
func TestWorkflowEndpointNotFound(t *testing.T) {
|
||||
// Create a minimal config
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: map[string]*config.ModelUpstream{
|
||||
"reasoning": {
|
||||
Address: "localhost:8001",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test POST /workflows with unknown workflow
|
||||
body := map[string]interface{}{
|
||||
"workflow": "unknown-workflow",
|
||||
"input": map[string]interface{}{},
|
||||
}
|
||||
|
||||
bodyBytes, _ := json.Marshal(body)
|
||||
req := httptest.NewRequest("POST", "/workflows", bytes.NewReader(bodyBytes))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("Expected 400, got %d", w.Code)
|
||||
}
|
||||
|
||||
var response map[string]interface{}
|
||||
json.Unmarshal(w.Body.Bytes(), &response)
|
||||
|
||||
if response["type"] != "https://api.example.com/problems/unknown-workflow" {
|
||||
t.Errorf("Expected unknown-workflow error, got %v", response["type"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowEndpointMissingWorkflow(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test POST /workflows with missing workflow field
|
||||
body := map[string]interface{}{
|
||||
"input": map[string]interface{}{},
|
||||
}
|
||||
|
||||
bodyBytes, _ := json.Marshal(body)
|
||||
req := httptest.NewRequest("POST", "/workflows", bytes.NewReader(bodyBytes))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("Expected 400, got %d", w.Code)
|
||||
}
|
||||
|
||||
var response map[string]interface{}
|
||||
json.Unmarshal(w.Body.Bytes(), &response)
|
||||
|
||||
if response["type"] != "https://api.example.com/problems/missing-workflow" {
|
||||
t.Errorf("Expected missing-workflow error, got %v", response["type"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowEndpointInvalidMethod(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test GET /workflows (should be 405)
|
||||
req := httptest.NewRequest("GET", "/workflows", nil)
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusMethodNotAllowed {
|
||||
t.Errorf("Expected 405, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowEndpointInvalidJSON(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test POST /workflows with invalid JSON
|
||||
req := httptest.NewRequest("POST", "/workflows", bytes.NewReader([]byte("not json")))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
handler.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("Expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetAvailableWorkflows(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
workflows := handler.getAvailableWorkflows()
|
||||
|
||||
expectedWorkflows := []string{
|
||||
"chat-and-embed",
|
||||
"multi-model-chat",
|
||||
"rag-pipeline",
|
||||
"batch-embeddings",
|
||||
}
|
||||
|
||||
if len(workflows) != len(expectedWorkflows) {
|
||||
t.Errorf("Expected %d workflows, got %d", len(expectedWorkflows), len(workflows))
|
||||
}
|
||||
|
||||
// Check that all expected workflows are present
|
||||
for _, expected := range expectedWorkflows {
|
||||
found := false
|
||||
for _, actual := range workflows {
|
||||
if actual == expected {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("Expected workflow %q not found", expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetWorkflow(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
Routes: make(map[string]*config.Route),
|
||||
Models: make(map[string]*config.ModelUpstream),
|
||||
}
|
||||
|
||||
handler := New(cfg)
|
||||
|
||||
// Test getting a valid workflow
|
||||
workflow, ok := handler.getWorkflow("chat-and-embed")
|
||||
if !ok {
|
||||
t.Error("Expected to find chat-and-embed workflow")
|
||||
}
|
||||
if workflow.Name != "chat-and-embed" {
|
||||
t.Errorf("Expected workflow name chat-and-embed, got %s", workflow.Name)
|
||||
}
|
||||
|
||||
// Test getting an invalid workflow
|
||||
workflow, ok = handler.getWorkflow("invalid-workflow")
|
||||
if ok {
|
||||
t.Error("Expected not to find invalid-workflow")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateWorkflowID(t *testing.T) {
|
||||
id1 := generateWorkflowID()
|
||||
id2 := generateWorkflowID()
|
||||
|
||||
if id1 == id2 {
|
||||
t.Error("Generated workflow IDs should be unique")
|
||||
}
|
||||
|
||||
if !bytes.HasPrefix([]byte(id1), []byte("wf_")) {
|
||||
t.Errorf("Workflow ID should start with 'wf_', got %s", id1)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseCapture(t *testing.T) {
|
||||
rc := &responseCapture{}
|
||||
|
||||
// Test Header
|
||||
rc.Header().Set("X-Test", "value")
|
||||
if rc.Header().Get("X-Test") != "value" {
|
||||
t.Error("Header not set correctly")
|
||||
}
|
||||
|
||||
// Test Write
|
||||
n, err := rc.Write([]byte("test content"))
|
||||
if err != nil {
|
||||
t.Errorf("Unexpected error: %v", err)
|
||||
}
|
||||
if n != 12 {
|
||||
t.Errorf("Expected 12 bytes written, got %d", n)
|
||||
}
|
||||
if rc.body.String() != "test content" {
|
||||
t.Errorf("Expected 'test content', got %s", rc.body.String())
|
||||
}
|
||||
|
||||
// Test WriteHeader
|
||||
rc.WriteHeader(http.StatusOK)
|
||||
if rc.status != http.StatusOK {
|
||||
t.Errorf("Expected status 200, got %d", rc.status)
|
||||
}
|
||||
|
||||
// Test WriteHeader doesn't override
|
||||
rc.WriteHeader(http.StatusInternalServerError)
|
||||
if rc.status != http.StatusOK {
|
||||
t.Error("WriteHeader should not override existing status")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkflowResponseSerialization(t *testing.T) {
|
||||
resp := WorkflowResponse{
|
||||
ID: "wf_123",
|
||||
Workflow: "test-workflow",
|
||||
Status: "completed",
|
||||
Output: map[string]interface{}{
|
||||
"key": "value",
|
||||
},
|
||||
Error: "",
|
||||
}
|
||||
|
||||
data, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
t.Errorf("Failed to marshal response: %v", err)
|
||||
}
|
||||
|
||||
var unmarshaled WorkflowResponse
|
||||
if err := json.Unmarshal(data, &unmarshaled); err != nil {
|
||||
t.Errorf("Failed to unmarshal response: %v", err)
|
||||
}
|
||||
|
||||
if unmarshaled.ID != resp.ID {
|
||||
t.Errorf("Expected ID %s, got %s", resp.ID, unmarshaled.ID)
|
||||
}
|
||||
if unmarshaled.Workflow != resp.Workflow {
|
||||
t.Errorf("Expected Workflow %s, got %s", resp.Workflow, unmarshaled.Workflow)
|
||||
}
|
||||
if unmarshaled.Status != resp.Status {
|
||||
t.Errorf("Expected Status %s, got %s", resp.Status, unmarshaled.Status)
|
||||
}
|
||||
}
|
||||
@@ -33,8 +33,8 @@ func NewRouter(healthChecker *HealthChecker, dispatcher *serviceadapter.Dispatch
|
||||
// ServeHTTP implements http.Handler.
|
||||
// Priority order:
|
||||
// 1. /healthz and /readyz to health handlers
|
||||
// 2. X-Service header to ServiceAdapter dispatcher (phase 8) - PREFERRED routing method
|
||||
// 3. /workflow* to temporal handler - DEPRECATED: use X-Service: workflow instead
|
||||
// 2. X-Service header to ServiceAdapter dispatcher (phase 8)
|
||||
// 3. /workflow* to temporal handler
|
||||
// 4. All other paths to upstream handler (phase 0-7)
|
||||
func (r *Router) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
// Health endpoints first
|
||||
@@ -48,8 +48,6 @@ func (r *Router) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
}
|
||||
|
||||
// X-Service (ServiceAdapter) routing - checked before path-based routing
|
||||
// PREFERRED: All service routing should use X-Service header pattern for consistency,
|
||||
// auth enforcement, and resource-based access control.
|
||||
if req.Header.Get("X-Service") != "" {
|
||||
if r.dispatcher != nil {
|
||||
r.dispatcher.Dispatch(w, req)
|
||||
@@ -58,8 +56,6 @@ func (r *Router) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
}
|
||||
|
||||
// Workflow endpoints
|
||||
// DEPRECATED: Path-based /workflow routing is legacy.
|
||||
// New clients should use X-Service: workflow header instead for consistent auth.
|
||||
switch req.URL.Path {
|
||||
case "/workflow", "/workflow/health", "/workflow/metrics":
|
||||
r.temporalHandler.ServeHTTP(w, req)
|
||||
|
||||
@@ -6,8 +6,6 @@ import (
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/net/http2"
|
||||
)
|
||||
|
||||
// Server wraps an HTTP server with graceful shutdown support.
|
||||
@@ -21,7 +19,8 @@ type Server struct {
|
||||
|
||||
// New creates a new Server with the given configuration.
|
||||
func New(listenAddr string, shutdownTimeout time.Duration, handler http.Handler) *Server {
|
||||
httpServer := &http.Server{
|
||||
return &Server{
|
||||
httpServer: &http.Server{
|
||||
Addr: listenAddr,
|
||||
Handler: handler,
|
||||
// ReadHeaderTimeout (not ReadTimeout) and a long WriteTimeout: both
|
||||
@@ -33,18 +32,7 @@ func New(listenAddr string, shutdownTimeout time.Duration, handler http.Handler)
|
||||
ReadHeaderTimeout: 15 * time.Second,
|
||||
WriteTimeout: 1 * time.Hour,
|
||||
IdleTimeout: 60 * time.Second,
|
||||
}
|
||||
|
||||
// Issue #32: Enable HTTP/2 for multiplexing concurrent streams.
|
||||
// This allows multiple LLM requests over a single connection,
|
||||
// improving throughput and reducing latency for concurrent clients.
|
||||
if err := http2.ConfigureServer(httpServer, nil); err != nil {
|
||||
// Silently fail HTTP/2 config (shouldn't happen, but gracefully degrade)
|
||||
// Server will still work with HTTP/1.1
|
||||
}
|
||||
|
||||
return &Server{
|
||||
httpServer: httpServer,
|
||||
},
|
||||
shutdownTimeout: shutdownTimeout,
|
||||
healthChecker: NewHealthChecker(false, false),
|
||||
}
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
package serviceadapter
|
||||
|
||||
// WorkflowAdapter handles X-Service: workflow requests.
|
||||
type WorkflowAdapter struct{}
|
||||
|
||||
// SQSAdapter handles X-Service: sqs requests.
|
||||
type SQSAdapter struct{}
|
||||
|
||||
|
||||
+100
-123
@@ -10,36 +10,46 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"golang.org/x/net/http2"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/identity"
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/problem"
|
||||
)
|
||||
|
||||
// Dispatcher routes X-Service requests to upstreams.
|
||||
// Auth per service:
|
||||
// SQS: Gateway validates JWT (kmsvc code unverified)
|
||||
// MinIO, Temporal: Native JWT support (dumb pipe pass-through)
|
||||
// Memory, IAM: Services validate JWTs themselves
|
||||
type Dispatcher struct {
|
||||
registry *Registry
|
||||
jwtValidator *auth.Validator
|
||||
sqsJWTAuth *auth.Validator
|
||||
}
|
||||
|
||||
// NewDispatcher creates a dispatcher with a shared multi-issuer JWT validator.
|
||||
// Pass nil to disable auth enforcement (all requests pass through).
|
||||
func NewDispatcher(registry *Registry, jwtValidator *auth.Validator) *Dispatcher {
|
||||
// NewDispatcher creates a new service adapter dispatcher.
|
||||
func NewDispatcher(registry *Registry) *Dispatcher {
|
||||
// Create JWT validator for SQS
|
||||
// Issuer and JWKS URL should match Authentik application config
|
||||
sqsValidator := auth.NewValidator(
|
||||
"https://authentik.riotpiao.com/application/o/sqs/",
|
||||
"sqs",
|
||||
"https://authentik.riotpiao.com/application/o/sqs/jwks/",
|
||||
)
|
||||
|
||||
return &Dispatcher{
|
||||
registry: registry,
|
||||
jwtValidator: jwtValidator,
|
||||
sqsJWTAuth: sqsValidator,
|
||||
}
|
||||
}
|
||||
|
||||
// Matches returns true if the request has an X-Service header.
|
||||
// Matches returns true if the request should be dispatched based on X-Service header.
|
||||
func (d *Dispatcher) Matches(r *http.Request) bool {
|
||||
return r.Header.Get("X-Service") != ""
|
||||
}
|
||||
|
||||
// Dispatch routes a request to the appropriate adapter.
|
||||
// Returns a problem document if the adapter or resource is not found.
|
||||
func (d *Dispatcher) Dispatch(w http.ResponseWriter, r *http.Request) {
|
||||
serviceName := r.Header.Get("X-Service")
|
||||
if serviceName == "" {
|
||||
@@ -47,117 +57,102 @@ func (d *Dispatcher) Dispatch(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Look up service adapter
|
||||
adapter := d.registry.Get(serviceName)
|
||||
if adapter == nil {
|
||||
d.writeError(w, problem.NotFound(fmt.Sprintf("service '%s' not found", serviceName)))
|
||||
p := problem.NotFound(fmt.Sprintf("service '%s' not found", serviceName))
|
||||
_ = p.Write(w)
|
||||
return
|
||||
}
|
||||
|
||||
// Get resource and method from request
|
||||
resourceName := r.Header.Get("X-Resource")
|
||||
if resourceName == "" {
|
||||
d.writeError(w, problem.BadRequest("X-Resource header required"))
|
||||
return
|
||||
}
|
||||
|
||||
resource := findResource(adapter, resourceName)
|
||||
// Find resource
|
||||
var resource *Resource
|
||||
for i := range adapter.Spec.Resources {
|
||||
if adapter.Spec.Resources[i].Name == resourceName {
|
||||
resource = &adapter.Spec.Resources[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if resource == nil {
|
||||
d.writeError(w, problem.NotFound(
|
||||
fmt.Sprintf("resource '%s' not found in service '%s'", resourceName, serviceName)))
|
||||
p := problem.NotFound(fmt.Sprintf("resource '%s' not found in service '%s'", resourceName, serviceName))
|
||||
_ = p.Write(w)
|
||||
return
|
||||
}
|
||||
|
||||
method := findMethod(resource, r.Method)
|
||||
// Find method matching HTTP verb
|
||||
var method *Method
|
||||
for i := range resource.Methods {
|
||||
if resource.Methods[i].Verb == r.Method {
|
||||
method = &resource.Methods[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if method == nil {
|
||||
d.writeError(w, problem.NotFound(
|
||||
fmt.Sprintf("method %s not defined for resource '%s'", r.Method, resourceName)))
|
||||
p := problem.NotFound(fmt.Sprintf("method %s not defined for resource '%s'", r.Method, resourceName))
|
||||
_ = p.Write(w)
|
||||
return
|
||||
}
|
||||
|
||||
// JWT auth enforcement for adapters that require it
|
||||
if adapter.Spec.Auth.Required && d.jwtValidator != nil {
|
||||
if !d.authenticate(w, r, serviceName, method.Verb) {
|
||||
// Gateway-level JWT validation for SQS (code unverified in kmsvc)
|
||||
// MinIO, Temporal, Memory, IAM have native JWT support - pass through
|
||||
if adapter.Spec.Auth.Required && serviceName == "sqs" {
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
if authHeader == "" {
|
||||
p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
|
||||
"Forbidden", "SQS requires Authorization header")
|
||||
_ = p.Write(w)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate JWT signature against Authentik JWKS
|
||||
claims, err := d.sqsJWTAuth.ValidateBearerToken(authHeader)
|
||||
if err != nil {
|
||||
p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
|
||||
"Forbidden", fmt.Sprintf("JWT validation failed: %v", err))
|
||||
_ = p.Write(w)
|
||||
return
|
||||
}
|
||||
|
||||
// Check required permissions (sqs:read or sqs:write or *)
|
||||
hasPermission := d.sqsJWTAuth.CheckPermissions(claims, "sqs:read", "sqs:write", "*")
|
||||
if !hasPermission {
|
||||
p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
|
||||
"Forbidden", "Insufficient permissions for SQS")
|
||||
_ = p.Write(w)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Detect protocol from upstream URL scheme
|
||||
upstreamURL := adapter.Spec.Upstream.URL
|
||||
if strings.HasPrefix(upstreamURL, "grpc://") {
|
||||
// gRPC upstream (Temporal, etc.)
|
||||
d.dispatchGRPC(w, r, upstreamURL, method, adapter)
|
||||
} else {
|
||||
// HTTP upstream (MinIO, Authentik, etc.)
|
||||
d.dispatchHTTP(w, r, upstreamURL, method, adapter)
|
||||
}
|
||||
}
|
||||
|
||||
// authenticate validates the JWT and checks service-level capability.
|
||||
// Returns false (and writes error response) if auth fails.
|
||||
func (d *Dispatcher) authenticate(w http.ResponseWriter, r *http.Request, serviceName, verb string) bool {
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
if authHeader == "" {
|
||||
d.writeError(w, problem.NewProblem(http.StatusUnauthorized,
|
||||
"about:blank#unauthorized", "Unauthorized",
|
||||
fmt.Sprintf("service '%s' requires Authorization header", serviceName)))
|
||||
return false
|
||||
}
|
||||
|
||||
claims, err := d.jwtValidator.ValidateBearerToken(authHeader)
|
||||
if err != nil {
|
||||
d.writeError(w, problem.NewProblem(http.StatusForbidden,
|
||||
"about:blank#forbidden", "Forbidden",
|
||||
fmt.Sprintf("JWT validation failed: %v", err)))
|
||||
return false
|
||||
}
|
||||
|
||||
// Check capability: <service>:read for GET/HEAD, <service>:write for mutating verbs
|
||||
required := capabilityForVerb(serviceName, verb)
|
||||
if !d.jwtValidator.CheckPermissions(claims, required, "*") {
|
||||
d.writeError(w, problem.NewProblem(http.StatusForbidden,
|
||||
"about:blank#insufficient-permissions", "Insufficient Permissions",
|
||||
fmt.Sprintf("required capability: %s", required)))
|
||||
return false
|
||||
}
|
||||
|
||||
// Inject identity headers for downstream
|
||||
identity.Inject(r, claims)
|
||||
return true
|
||||
}
|
||||
|
||||
// capabilityForVerb maps HTTP verbs to <service>:read or <service>:write.
|
||||
func capabilityForVerb(serviceName, verb string) string {
|
||||
switch verb {
|
||||
case "GET", "HEAD", "OPTIONS":
|
||||
return serviceName + ":read"
|
||||
default:
|
||||
return serviceName + ":write"
|
||||
}
|
||||
}
|
||||
|
||||
func findResource(adapter *ServiceAdapter, name string) *Resource {
|
||||
for i := range adapter.Spec.Resources {
|
||||
if adapter.Spec.Resources[i].Name == name {
|
||||
return &adapter.Spec.Resources[i]
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func findMethod(resource *Resource, verb string) *Method {
|
||||
for i := range resource.Methods {
|
||||
if resource.Methods[i].Verb == verb {
|
||||
return &resource.Methods[i]
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// dispatchHTTP forwards HTTP requests to upstream, passing Authorization header through.
|
||||
func (d *Dispatcher) dispatchHTTP(w http.ResponseWriter, r *http.Request, upstreamURL string, method *Method, adapter *ServiceAdapter) {
|
||||
parsedURL, err := url.Parse(upstreamURL)
|
||||
if err != nil {
|
||||
d.writeError(w, problem.NewProblem(http.StatusInternalServerError,
|
||||
"about:blank#server-error", "Internal Server Error",
|
||||
fmt.Sprintf("invalid upstream URL: %v", err)))
|
||||
d.writeError(w, problem.NewProblem(http.StatusInternalServerError, "about:blank#server-error",
|
||||
"Internal Server Error", fmt.Sprintf("invalid upstream URL: %v", err)))
|
||||
return
|
||||
}
|
||||
|
||||
// Create reverse proxy
|
||||
proxy := httputil.NewSingleHostReverseProxy(parsedURL)
|
||||
proxy.Director = func(req *http.Request) {
|
||||
req.URL.Scheme = parsedURL.Scheme
|
||||
@@ -165,12 +160,10 @@ func (d *Dispatcher) dispatchHTTP(w http.ResponseWriter, r *http.Request, upstre
|
||||
req.URL.Path = method.UpstreamPath
|
||||
req.RequestURI = ""
|
||||
req.Host = parsedURL.Host
|
||||
|
||||
// Preserve Authorization header for S3 SigV4 and other auth schemes
|
||||
// Note: httputil.ReverseProxy preserves most headers automatically,
|
||||
// but we need to ensure Authorization isn't lost when overriding Director
|
||||
// Authorization header passes through unchanged
|
||||
}
|
||||
|
||||
// Set timeout
|
||||
timeout := adapter.Spec.Upstream.TimeoutSeconds
|
||||
if timeout <= 0 {
|
||||
timeout = 30
|
||||
@@ -180,25 +173,29 @@ func (d *Dispatcher) dispatchHTTP(w http.ResponseWriter, r *http.Request, upstre
|
||||
TLSHandshakeTimeout: time.Duration(timeout) * time.Second,
|
||||
}
|
||||
|
||||
// Forward the request
|
||||
proxy.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
// dispatchGRPC forwards gRPC requests to upstream.
|
||||
// gRPC URL format: grpc://host:port
|
||||
func (d *Dispatcher) dispatchGRPC(w http.ResponseWriter, r *http.Request, upstreamURL string, method *Method, adapter *ServiceAdapter) {
|
||||
// Extract host:port from grpc://host:port
|
||||
host := strings.TrimPrefix(upstreamURL, "grpc://")
|
||||
if host == upstreamURL {
|
||||
d.writeError(w, problem.NewProblem(http.StatusInternalServerError,
|
||||
"about:blank#server-error", "Internal Server Error",
|
||||
"invalid gRPC URL format"))
|
||||
d.writeError(w, problem.NewProblem(http.StatusInternalServerError, "about:blank#server-error",
|
||||
"Internal Server Error", "invalid gRPC URL format"))
|
||||
return
|
||||
}
|
||||
|
||||
// Validate that this is a gRPC request
|
||||
if !strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") {
|
||||
d.writeError(w, problem.NewProblem(http.StatusBadRequest,
|
||||
"about:blank#bad-request", "Bad Request",
|
||||
"gRPC service requires application/grpc content-type"))
|
||||
d.writeError(w, problem.NewProblem(http.StatusBadRequest, "about:blank#bad-request",
|
||||
"Bad Request", "gRPC service requires application/grpc content-type"))
|
||||
return
|
||||
}
|
||||
|
||||
// Set timeout
|
||||
timeout := adapter.Spec.Upstream.TimeoutSeconds
|
||||
if timeout <= 0 {
|
||||
timeout = 30
|
||||
@@ -207,45 +204,25 @@ func (d *Dispatcher) dispatchGRPC(w http.ResponseWriter, r *http.Request, upstre
|
||||
ctx, cancel := context.WithTimeout(r.Context(), time.Duration(timeout)*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Dial gRPC upstream
|
||||
conn, err := grpc.DialContext(ctx, host,
|
||||
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(100*1024*1024)),
|
||||
grpc.WithDefaultCallOptions(
|
||||
grpc.MaxCallRecvMsgSize(100 * 1024 * 1024), // 100MB
|
||||
),
|
||||
)
|
||||
if err != nil {
|
||||
d.writeError(w, problem.NewProblem(http.StatusBadGateway,
|
||||
"about:blank#bad-gateway", "Bad Gateway",
|
||||
fmt.Sprintf("failed to dial gRPC upstream: %v", err)))
|
||||
d.writeError(w, problem.NewProblem(http.StatusBadGateway, "about:blank#bad-gateway",
|
||||
"Bad Gateway", fmt.Sprintf("failed to dial gRPC upstream: %v", err)))
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
// Create HTTP/2 reverse proxy for gRPC
|
||||
// gRPC uses HTTP/2 protocol, so we need an HTTP/2-capable transport
|
||||
upstreamURLObj := &url.URL{
|
||||
Scheme: "http",
|
||||
Host: host,
|
||||
}
|
||||
|
||||
proxy := httputil.NewSingleHostReverseProxy(upstreamURLObj)
|
||||
proxy.Director = func(req *http.Request) {
|
||||
req.URL.Scheme = "http"
|
||||
req.URL.Host = host
|
||||
req.URL.Path = method.UpstreamPath
|
||||
req.RequestURI = ""
|
||||
req.Host = host
|
||||
}
|
||||
|
||||
// Create HTTP/2 client transport for gRPC calls
|
||||
// gRPC requires HTTP/2 for proper message framing
|
||||
h2transport := &http2.Transport{
|
||||
AllowHTTP: true,
|
||||
}
|
||||
|
||||
// Set the transport on the proxy
|
||||
proxy.Transport = h2transport
|
||||
|
||||
// Serve the request through the proxy
|
||||
proxy.ServeHTTP(w, r)
|
||||
// Forward gRPC request
|
||||
// Note: Full gRPC forwarding requires grpcproxy or custom middleware.
|
||||
// For now, return unimplemented (Temporal support coming in Phase 9)
|
||||
d.writeError(w, problem.NewProblem(http.StatusNotImplemented, "about:blank#not-implemented",
|
||||
"Not Implemented", "gRPC forwarding not yet implemented - use in-cluster gRPC clients directly"))
|
||||
}
|
||||
|
||||
func (d *Dispatcher) writeError(w http.ResponseWriter, p *problem.Problem) {
|
||||
|
||||
@@ -1,265 +0,0 @@
|
||||
package serviceadapter
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/identity"
|
||||
)
|
||||
|
||||
// stubValidator implements the minimum interface for testing auth.
|
||||
// Real auth.Validator needs JWKS — we test the dispatcher logic, not JWT crypto.
|
||||
|
||||
func newTestRegistry(adapters ...ServiceAdapter) *Registry {
|
||||
r := NewRegistry(nil)
|
||||
for i := range adapters {
|
||||
_ = r.Add(&adapters[i])
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func sqsAdapter(authRequired bool) ServiceAdapter {
|
||||
return ServiceAdapter{
|
||||
Name: "sqs",
|
||||
ServiceName: "sqs",
|
||||
Spec: Spec{
|
||||
ServiceName: "sqs",
|
||||
Upstream: Upstream{URL: "http://localhost:9999", TimeoutSeconds: 5},
|
||||
Auth: Auth{Required: authRequired},
|
||||
Resources: []Resource{
|
||||
{
|
||||
Name: "list-queues",
|
||||
Methods: []Method{
|
||||
{Verb: "GET", UpstreamPath: "/sqs/queues"},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "send-message",
|
||||
Methods: []Method{
|
||||
{Verb: "POST", UpstreamPath: "/sqs/send"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func memoryAdapter() ServiceAdapter {
|
||||
return ServiceAdapter{
|
||||
Name: "memory",
|
||||
ServiceName: "memory",
|
||||
Spec: Spec{
|
||||
ServiceName: "memory",
|
||||
Upstream: Upstream{URL: "http://localhost:8888", TimeoutSeconds: 5},
|
||||
Auth: Auth{Required: false},
|
||||
Resources: []Resource{
|
||||
{
|
||||
Name: "skills",
|
||||
Methods: []Method{
|
||||
{Verb: "GET", UpstreamPath: "/memory/skills"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_MissingXService(t *testing.T) {
|
||||
d := NewDispatcher(newTestRegistry(), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_UnknownService(t *testing.T) {
|
||||
d := NewDispatcher(newTestRegistry(), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("X-Service", "nonexistent")
|
||||
r.Header.Set("X-Resource", "foo")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("expected 404, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_MissingXResource(t *testing.T) {
|
||||
d := NewDispatcher(newTestRegistry(memoryAdapter()), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("X-Service", "memory")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("expected 400, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_UnknownResource(t *testing.T) {
|
||||
d := NewDispatcher(newTestRegistry(memoryAdapter()), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("X-Service", "memory")
|
||||
r.Header.Set("X-Resource", "nonexistent")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("expected 404, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_WrongHTTPVerb(t *testing.T) {
|
||||
d := NewDispatcher(newTestRegistry(memoryAdapter()), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("DELETE", "/", nil)
|
||||
r.Header.Set("X-Service", "memory")
|
||||
r.Header.Set("X-Resource", "skills")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("expected 404, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_AuthRequired_NoToken(t *testing.T) {
|
||||
// Use nil validator — auth required but no validator means 401
|
||||
// Actually with nil validator, auth is skipped. Use a real scenario.
|
||||
// We need a mock validator. For now test that auth.Required=false passes through.
|
||||
// The real auth test needs the full JWKS setup which is an integration test.
|
||||
|
||||
// Test: auth required, no validator configured = passes through (defense in depth via NetworkPolicy)
|
||||
d := NewDispatcher(newTestRegistry(sqsAdapter(true)), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("X-Service", "sqs")
|
||||
r.Header.Set("X-Resource", "list-queues")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
// With nil validator, auth check is skipped — request reaches upstream (which will fail since localhost:9999 is down)
|
||||
// The key assertion: it did NOT return 401/403, it tried to proxy
|
||||
if w.Code == http.StatusUnauthorized || w.Code == http.StatusForbidden {
|
||||
t.Errorf("expected proxy attempt (not auth rejection), got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_AuthNotRequired_NoToken(t *testing.T) {
|
||||
// Start a test upstream
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
json.NewEncoder(w).Encode(map[string]string{"path": r.URL.Path})
|
||||
}))
|
||||
defer upstream.Close()
|
||||
|
||||
adapter := memoryAdapter()
|
||||
adapter.Spec.Upstream.URL = upstream.URL
|
||||
|
||||
d := NewDispatcher(newTestRegistry(adapter), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("X-Service", "memory")
|
||||
r.Header.Set("X-Resource", "skills")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Errorf("expected 200, got %d", w.Code)
|
||||
}
|
||||
|
||||
var body map[string]string
|
||||
json.NewDecoder(w.Body).Decode(&body)
|
||||
if body["path"] != "/memory/skills" {
|
||||
t.Errorf("expected upstream path /memory/skills, got %s", body["path"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_PassThroughHeaders(t *testing.T) {
|
||||
var receivedAuth string
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
receivedAuth = r.Header.Get("Authorization")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer upstream.Close()
|
||||
|
||||
adapter := memoryAdapter()
|
||||
adapter.Spec.Upstream.URL = upstream.URL
|
||||
|
||||
d := NewDispatcher(newTestRegistry(adapter), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("X-Service", "memory")
|
||||
r.Header.Set("X-Resource", "skills")
|
||||
r.Header.Set("Authorization", "Bearer some-jwt")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if receivedAuth != "Bearer some-jwt" {
|
||||
t.Errorf("Authorization header not passed through, got %q", receivedAuth)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCapabilityForVerb(t *testing.T) {
|
||||
tests := []struct {
|
||||
service string
|
||||
verb string
|
||||
want string
|
||||
}{
|
||||
{"sqs", "GET", "sqs:read"},
|
||||
{"sqs", "HEAD", "sqs:read"},
|
||||
{"sqs", "OPTIONS", "sqs:read"},
|
||||
{"sqs", "POST", "sqs:write"},
|
||||
{"sqs", "PUT", "sqs:write"},
|
||||
{"sqs", "DELETE", "sqs:write"},
|
||||
{"sqs", "PATCH", "sqs:write"},
|
||||
{"memory", "GET", "memory:read"},
|
||||
{"memory", "POST", "memory:write"},
|
||||
{"s3", "GET", "s3:read"},
|
||||
{"s3", "PUT", "s3:write"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
got := capabilityForVerb(tt.service, tt.verb)
|
||||
if got != tt.want {
|
||||
t.Errorf("capabilityForVerb(%s, %s) = %s, want %s", tt.service, tt.verb, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_IdentityHeadersNotSet_WhenNoAuth(t *testing.T) {
|
||||
var gotUser, gotVerified string
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotUser = r.Header.Get(identity.HeaderUser)
|
||||
gotVerified = r.Header.Get(identity.HeaderAuthVerified)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer upstream.Close()
|
||||
|
||||
adapter := memoryAdapter()
|
||||
adapter.Spec.Upstream.URL = upstream.URL
|
||||
|
||||
d := NewDispatcher(newTestRegistry(adapter), nil)
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("GET", "/", nil)
|
||||
r.Header.Set("X-Service", "memory")
|
||||
r.Header.Set("X-Resource", "skills")
|
||||
|
||||
d.Dispatch(w, r)
|
||||
|
||||
if gotUser != "" {
|
||||
t.Errorf("X-Forwarded-User should not be set without auth, got %q", gotUser)
|
||||
}
|
||||
if gotVerified != "" {
|
||||
t.Errorf("X-Auth-Verified should not be set without auth, got %q", gotVerified)
|
||||
}
|
||||
}
|
||||
@@ -1,288 +0,0 @@
|
||||
package serviceadapter
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
"forgejo.riotpiao.com/rock/homelab-frontend/internal/temporal"
|
||||
)
|
||||
|
||||
// WorkflowAdapter handles X-Service: workflow requests.
|
||||
// It forwards workflow operations to the Temporal gRPC service.
|
||||
// Users can specify namespace via the request payload.
|
||||
type WorkflowAdapter struct {
|
||||
temporalHandler *temporal.Handler
|
||||
}
|
||||
|
||||
// NewWorkflowAdapter creates a new WorkflowAdapter.
|
||||
func NewWorkflowAdapter(handler *temporal.Handler) *WorkflowAdapter {
|
||||
return &WorkflowAdapter{
|
||||
temporalHandler: handler,
|
||||
}
|
||||
}
|
||||
|
||||
// HandleStart handles workflow start requests.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "...", "workflow_type": "...", "task_queue": "...", "input": {...} }
|
||||
func (wa *WorkflowAdapter) HandleStart(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleDescribe handles workflow describe requests.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "..." }
|
||||
func (wa *WorkflowAdapter) HandleDescribe(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleList handles workflow list requests.
|
||||
// Expects payload: { "namespace": "default", "query": "..." (optional) }
|
||||
func (wa *WorkflowAdapter) HandleList(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleHistory handles workflow history requests.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "..." }
|
||||
func (wa *WorkflowAdapter) HandleHistory(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleTerminate handles workflow termination.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "...", "reason": "..." }
|
||||
func (wa *WorkflowAdapter) HandleTerminate(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleCancel handles workflow cancellation.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "..." }
|
||||
func (wa *WorkflowAdapter) HandleCancel(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleSignal handles workflow signal.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "...", "signal_name": "...", "signal_data": {...} }
|
||||
func (wa *WorkflowAdapter) HandleSignal(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleQuery handles workflow query.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "...", "query_type": "...", "query_data": {...} }
|
||||
func (wa *WorkflowAdapter) HandleQuery(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleReset handles workflow reset.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "...", "reset_type": "..." }
|
||||
func (wa *WorkflowAdapter) HandleReset(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(w, r)
|
||||
}
|
||||
|
||||
// HandleUpdate handles workflow update.
|
||||
// Expects payload: { "namespace": "default", "workflow_id": "...", "update_data": {...} }
|
||||
func (wa *WorkflowAdapter) HandleUpdate(w http.ResponseWriter, r *http.Request) {
|
||||
wa.forwardToTemporal(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) {
|
||||
// Read request body
|
||||
body, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
http.Error(w, fmt.Sprintf("failed to read request body: %v", err), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
defer r.Body.Close()
|
||||
|
||||
// Parse JSON to check for namespace
|
||||
var payload map[string]interface{}
|
||||
if err := json.Unmarshal(body, &payload); err != nil {
|
||||
http.Error(w, fmt.Sprintf("invalid JSON payload: %v", err), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Ensure namespace is specified (required for Temporal routing)
|
||||
namespace, ok := payload["namespace"].(string)
|
||||
if !ok || namespace == "" {
|
||||
http.Error(w, `"namespace" field required in payload`, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Forward to temporal handler by calling it with the request
|
||||
// Restore body for temporal handler
|
||||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||||
r.ContentLength = int64(len(body))
|
||||
|
||||
// Call temporal handler
|
||||
wa.temporalHandler.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
// GetSpec returns the ServiceAdapter spec for workflow service.
|
||||
// This defines the available resources and methods.
|
||||
func GetWorkflowSpec() *Spec {
|
||||
return &Spec{
|
||||
ServiceName: "workflow",
|
||||
Upstream: Upstream{
|
||||
URL: "grpc://temporal:7233", // gRPC endpoint
|
||||
TimeoutSeconds: 30,
|
||||
},
|
||||
Auth: Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:execute",
|
||||
},
|
||||
Retryable: true,
|
||||
Resources: []Resource{
|
||||
{
|
||||
Name: "start",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/StartWorkflowExecution",
|
||||
RequestSchema: "workflow_start_request",
|
||||
ResponseSchema: "workflow_start_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:execute",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "describe",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/DescribeWorkflowExecution",
|
||||
RequestSchema: "workflow_describe_request",
|
||||
ResponseSchema: "workflow_describe_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:read",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "list",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/ListWorkflowExecutions",
|
||||
RequestSchema: "workflow_list_request",
|
||||
ResponseSchema: "workflow_list_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:read",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "history",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/GetWorkflowExecutionHistory",
|
||||
RequestSchema: "workflow_history_request",
|
||||
ResponseSchema: "workflow_history_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:read",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "terminate",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/TerminateWorkflowExecution",
|
||||
RequestSchema: "workflow_terminate_request",
|
||||
ResponseSchema: "workflow_terminate_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:execute",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "cancel",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/RequestCancelWorkflowExecution",
|
||||
RequestSchema: "workflow_cancel_request",
|
||||
ResponseSchema: "workflow_cancel_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:execute",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "signal",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/SignalWorkflowExecution",
|
||||
RequestSchema: "workflow_signal_request",
|
||||
ResponseSchema: "workflow_signal_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:signal",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "query",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/QueryWorkflow",
|
||||
RequestSchema: "workflow_query_request",
|
||||
ResponseSchema: "workflow_query_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:query",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "reset",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/ResetWorkflowExecution",
|
||||
RequestSchema: "workflow_reset_request",
|
||||
ResponseSchema: "workflow_reset_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:execute",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "update",
|
||||
Methods: []Method{
|
||||
{
|
||||
Verb: "POST",
|
||||
UpstreamPath: "/temporal.workflowservice.v1.WorkflowService/UpdateWorkflowExecution",
|
||||
RequestSchema: "workflow_update_request",
|
||||
ResponseSchema: "workflow_update_response",
|
||||
Auth: &Auth{
|
||||
Required: true,
|
||||
Capability: "workflow:execute",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
+147
-16
File diff suppressed because one or more lines are too long
@@ -57,12 +57,6 @@ spec:
|
||||
value: "0.0.0.0:8080"
|
||||
- name: CONFIG_PATH
|
||||
value: "/etc/gateway/config.yaml"
|
||||
- name: AUTH_CLIENT_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: api-gw-client-secret
|
||||
key: client-secret
|
||||
optional: true
|
||||
- name: SHUTDOWN_TIMEOUT
|
||||
value: "5m"
|
||||
- name: LOG_LEVEL
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
apiVersion: v1
|
||||
kind: Secret
|
||||
metadata:
|
||||
name: api-gateway-config
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
type: Opaque
|
||||
stringData:
|
||||
config.yaml: |
|
||||
auth:
|
||||
enabled: true
|
||||
issuer: "https://authentik.riotpiao.com/application/o/api-gw/"
|
||||
audience: "api-gw"
|
||||
jwksUrl: "http://authentik-server.iam.svc.cluster.local/application/o/api-gw/jwks/"
|
||||
requiredCapability: "llm:inference"
|
||||
routes: []
|
||||
models:
|
||||
- name: "reasoning"
|
||||
address: "reasoning-predictor.llm-serving:80"
|
||||
path: "/v1/chat/completions"
|
||||
- name: "ornith:35b"
|
||||
address: "ornith-predictor.llm-serving:80"
|
||||
path: "/v1/chat/completions"
|
||||
- name: "qwen2.5:3b-instruct"
|
||||
address: "ornith-predictor.llm-serving:80"
|
||||
path: "/v1/chat/completions"
|
||||
- name: "nomic-ai/nomic-embed-text-v2-moe"
|
||||
address: "embeddings-predictor.llm-serving:80"
|
||||
path: "/v1/embeddings"
|
||||
- name: "BAAI/bge-reranker-base"
|
||||
address: "reranker-predictor.llm-serving:80"
|
||||
path: "/v1/rerank"
|
||||
adapters:
|
||||
- serviceName: sqs
|
||||
upstream:
|
||||
url: http://management-service.sqs.svc.cluster.local:9090
|
||||
timeoutSeconds: 30
|
||||
auth:
|
||||
required: true
|
||||
resources:
|
||||
- name: send-message
|
||||
methods:
|
||||
- verb: POST
|
||||
upstreamPath: /sqs/send
|
||||
- name: receive-message
|
||||
methods:
|
||||
- verb: POST
|
||||
upstreamPath: /sqs/receive
|
||||
- name: list-queues
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /sqs/queues
|
||||
- serviceName: workflow
|
||||
upstream:
|
||||
url: grpc://temporal-frontend.temporal.svc.cluster.local:7233
|
||||
timeoutSeconds: 60
|
||||
auth:
|
||||
required: false
|
||||
resources:
|
||||
- name: execute
|
||||
methods:
|
||||
- verb: POST
|
||||
upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ExecuteWorkflow
|
||||
- name: describe
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/DescribeWorkflowExecution
|
||||
- name: list
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ListWorkflowExecutions
|
||||
- serviceName: memory
|
||||
upstream:
|
||||
url: http://poimen-memory.poimen.svc.cluster.local:8080
|
||||
timeoutSeconds: 30
|
||||
auth:
|
||||
required: false
|
||||
resources:
|
||||
- name: query
|
||||
methods:
|
||||
- verb: POST
|
||||
upstreamPath: /memory/query
|
||||
- name: ingest
|
||||
methods:
|
||||
- verb: POST
|
||||
upstreamPath: /memory/ingest
|
||||
- name: skills
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /memory/skills
|
||||
- serviceName: s3
|
||||
upstream:
|
||||
url: http://minio.storage.svc.cluster.local:9000
|
||||
timeoutSeconds: 30
|
||||
auth:
|
||||
required: false
|
||||
resources:
|
||||
- name: list-objects
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /
|
||||
- name: get-object
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /
|
||||
- name: put-object
|
||||
methods:
|
||||
- verb: PUT
|
||||
upstreamPath: /
|
||||
- serviceName: iam
|
||||
upstream:
|
||||
url: http://authentik-server.iam.svc.cluster.local:80
|
||||
timeoutSeconds: 30
|
||||
auth:
|
||||
required: false
|
||||
resources:
|
||||
- name: list-roles
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /api/v3/roles
|
||||
- name: list-users
|
||||
methods:
|
||||
- verb: GET
|
||||
upstreamPath: /api/v3/users
|
||||
- name: create-role
|
||||
methods:
|
||||
- verb: POST
|
||||
upstreamPath: /api/v3/roles
|
||||
@@ -1,10 +0,0 @@
|
||||
apiVersion: v1
|
||||
kind: Secret
|
||||
metadata:
|
||||
name: api-gateway-config
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
type: Opaque
|
||||
stringData:
|
||||
config.yaml: "# PRODUCTION GATEWAY CONFIGURATION\n# ==========================================\n# All upstream services MUST use Kubernetes internal service DNS names\n# Format: <service>.<namespace>.svc.cluster.local\n# \n# This ensures:\n# - Communication within cluster network only (no external IP exposure)\n# - Pod-to-pod service discovery via internal DNS\n# - Security policy enforcement at network level\n# - Service-level load balancing via kube-proxy\n#\n# Routing Pattern:\n# PREFERRED: X-Service header routing (e.g., X-Service: workflow)\n# Legacy: Path-based routing (e.g., /workflow) - being deprecated\n#\nauth:\n enabled: true\n issuer: \"https://authentik.riotpiao.com/application/o/api-gw/\"\n audience: \"api-gw\"\n jwksUrl: \"http://authentik-server.iam.svc.cluster.local/application/o/api-gw/jwks/\"\n requiredCapability: \"llm:inference\"\n tokenUrl: \"http://authentik-server.iam.svc.cluster.local/application/o/token/\"\n clientId: \"api-gw\"\nroutes: []\nmodels:\n# All model services use internal Kubernetes DNS (llm-serving namespace)\n- name: \"reasoning\"\n address: \"reasoning-predictor.llm-serving.svc.cluster.local:80\"\n path: \"/v1/chat/completions\"\n- name: \"ornith:35b\"\n address: \"ornith-predictor.llm-serving.svc.cluster.local:80\"\n path: \"/v1/chat/completions\"\n- name: \"qwen2.5:3b-instruct\"\n address: \"qwen-cpu.llm-serving.svc.cluster.local:80\"\n path: \"/v1/chat/completions\"\n- name: \"nomic-ai/nomic-embed-text-v2-moe\"\n address: \"embeddings-predictor.llm-serving.svc.cluster.local:80\"\n path: \"/v1/embeddings\"\n- name: \"BAAI/bge-reranker-base\"\n address: \"reranker-predictor.llm-serving.svc.cluster.local:80\"\n path: \"/v1/rerank\"\nadapters:\n- serviceName: sqs\n upstream:\n url: http://management-service.sqs.svc.cluster.local:9090\n timeoutSeconds: 30\n auth:\n required: true\n resources:\n - name: send-message\n methods:\n - verb: POST\n upstreamPath: /sqs/send\n - name: receive-message\n methods:\n - verb: POST\n upstreamPath: /sqs/receive\n - name: list-queues\n methods:\n - verb: GET\n upstreamPath: /sqs/queues\n- serviceName: workflow\n upstream:\n url: grpc://temporal-frontend.temporal.svc.cluster.local:7233\n timeoutSeconds: 60\n auth:\n required: false\n resources:\n - name: execute\n methods:\n - verb: POST\n upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ExecuteWorkflow\n - name: describe\n methods:\n - verb: GET\n upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/DescribeWorkflowExecution\n - name: list\n methods:\n - verb: GET\n upstreamPath: /temporal.api.workflowservice.v1.WorkflowService/ListWorkflowExecutions\n- serviceName: memory\n upstream:\n url: http://poimen-memory.poimen.svc.cluster.local:8080\n timeoutSeconds: 30\n auth:\n required: false\n resources:\n - name: query\n methods:\n - verb: POST\n upstreamPath: /memory/query\n - name: ingest\n methods:\n - verb: POST\n upstreamPath: /memory/ingest\n - name: skills\n methods:\n - verb: GET\n upstreamPath: /memory/skills\n- serviceName: s3\n upstream:\n url: http://minio.storage.svc.cluster.local:80\n timeoutSeconds: 30\n auth:\n required: false\n resources:\n - name: list-objects\n methods:\n - verb: GET\n upstreamPath: /\n - name: get-object\n methods:\n - verb: GET\n upstreamPath: /\n - name: put-object\n methods:\n - verb: PUT\n upstreamPath: /\n- serviceName: iam\n upstream:\n url: http://authentik-server.iam.svc.cluster.local:80\n timeoutSeconds: 30\n auth:\n required: false\n resources:\n - name: list-roles\n methods:\n - verb: GET\n upstreamPath: /api/v3/roles\n - name: list-users\n methods:\n - verb: GET\n upstreamPath: /api/v3/users\n - name: create-role\n methods:\n - verb: POST\n upstreamPath: /api/v3/roles\n"
|
||||
@@ -1,321 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -1,70 +0,0 @@
|
||||
apiVersion: batch/v1
|
||||
kind: Job
|
||||
metadata:
|
||||
name: api-gateway-integration-test
|
||||
namespace: api
|
||||
spec:
|
||||
template:
|
||||
spec:
|
||||
serviceAccountName: api-gateway
|
||||
restartPolicy: Never
|
||||
containers:
|
||||
- name: integration-tester
|
||||
image: golang:1.26-bookworm
|
||||
imagePullPolicy: IfNotPresent
|
||||
workingDir: /workspace
|
||||
command:
|
||||
- /bin/bash
|
||||
- -c
|
||||
- |
|
||||
set -e
|
||||
echo "Starting integration tests..."
|
||||
|
||||
# Clone the repo
|
||||
git clone https://forgejo.riotpiao.com/riotpiao-poimen/homelab-frontend.git .
|
||||
|
||||
# Wait for gateway to be ready
|
||||
echo "Waiting for gateway service to be ready..."
|
||||
for i in {1..30}; do
|
||||
if curl -s http://api-gateway:8080/healthz | grep -q "alive"; then
|
||||
echo "✓ Gateway is ready"
|
||||
break
|
||||
fi
|
||||
echo "Attempting to reach gateway ($i/30)..."
|
||||
sleep 2
|
||||
done
|
||||
|
||||
# Run integration tests
|
||||
echo "Running integration tests..."
|
||||
go test -v -tags=integration -timeout=5m ./internal/integration/...
|
||||
|
||||
echo "✓ Integration tests completed"
|
||||
env:
|
||||
- name: GATEWAY_URL
|
||||
value: "http://api-gateway:8080"
|
||||
resources:
|
||||
requests:
|
||||
cpu: 250m
|
||||
memory: 512Mi
|
||||
limits:
|
||||
cpu: 500m
|
||||
memory: 1Gi
|
||||
securityContext:
|
||||
runAsNonRoot: true
|
||||
runAsUser: 65532
|
||||
allowPrivilegeEscalation: false
|
||||
capabilities:
|
||||
drop:
|
||||
- ALL
|
||||
readOnlyRootFilesystem: true
|
||||
volumeMounts:
|
||||
- name: tmp
|
||||
mountPath: /tmp
|
||||
- name: home
|
||||
mountPath: /home/nonroot
|
||||
volumes:
|
||||
- name: tmp
|
||||
emptyDir: {}
|
||||
- name: home
|
||||
emptyDir: {}
|
||||
backoffLimit: 1
|
||||
@@ -8,7 +8,7 @@ resources:
|
||||
- service.yaml
|
||||
- deployment.yaml
|
||||
- network-policy.yaml
|
||||
- gateway-config-secret.yaml
|
||||
- gateway-config-secret.enc.yaml
|
||||
|
||||
# The deployed image tag lives here and nowhere else. CI publishes
|
||||
# forgejo.riotpiao.com/rock/api-gateway:<commit-sha> and tags it as :latest on main.
|
||||
|
||||
@@ -1,27 +0,0 @@
|
||||
apiVersion: kustomize.config.k8s.io/v1beta1
|
||||
kind: Kustomization
|
||||
|
||||
namespace: api
|
||||
|
||||
resources:
|
||||
- serviceaccount.yaml
|
||||
- service.yaml
|
||||
- deployment.yaml
|
||||
- network-policy.yaml
|
||||
- gateway-config-secret.enc.yaml
|
||||
|
||||
# The deployed image tag lives here and nowhere else. CI publishes
|
||||
# forgejo.riotpiao.com/rock/api-gateway:<commit-sha> and tags it as :latest on main.
|
||||
# ArgoCD auto-syncs when the latest image is available.
|
||||
images:
|
||||
- name: forgejo.riotpiao.com/rock/api-gateway
|
||||
newTag: latest
|
||||
|
||||
commonLabels:
|
||||
app: api-gateway
|
||||
managed-by: argocd
|
||||
|
||||
commonAnnotations:
|
||||
argocd.argoproj.io/sync-wave: "2"
|
||||
# Wave 2 ensures the gateway is ready before anything that depends on it
|
||||
# Kong remains on wave 7 unchanged
|
||||
@@ -46,14 +46,6 @@ spec:
|
||||
ports:
|
||||
- protocol: TCP
|
||||
port: 8080
|
||||
# Allow from paperless namespace (paperless-ai document auto-tagging)
|
||||
- from:
|
||||
- namespaceSelector:
|
||||
matchLabels:
|
||||
kubernetes.io/metadata.name: paperless
|
||||
ports:
|
||||
- protocol: TCP
|
||||
port: 8080
|
||||
egress:
|
||||
# Allow DNS
|
||||
- to:
|
||||
@@ -131,14 +123,10 @@ spec:
|
||||
- protocol: TCP
|
||||
port: 8080
|
||||
# Allow to MinIO (S3-compatible storage)
|
||||
# Service `minio` listens on port 80 (targetPort 9000).
|
||||
# Headless `minio-cluster-hl` is 9000. Allow both.
|
||||
- to:
|
||||
- namespaceSelector:
|
||||
matchLabels:
|
||||
kubernetes.io/metadata.name: storage
|
||||
ports:
|
||||
- protocol: TCP
|
||||
port: 80
|
||||
- protocol: TCP
|
||||
port: 9000
|
||||
|
||||
@@ -1,29 +0,0 @@
|
||||
apiVersion: v1
|
||||
kind: Secret
|
||||
metadata:
|
||||
name: smtp-credentials
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: notification
|
||||
type: Opaque
|
||||
data:
|
||||
host: <base64-encoded SMTP hostname>
|
||||
port: <base64-encoded SMTP port, e.g., "587">
|
||||
from: <base64-encoded sender email>
|
||||
user: <base64-encoded SMTP username>
|
||||
password: <base64-encoded SMTP password>
|
||||
|
||||
# To create from plaintext:
|
||||
# kubectl create secret generic smtp-credentials \
|
||||
# --from-literal=host=mail.example.com \
|
||||
# --from-literal=port=587 \
|
||||
# [email protected] \
|
||||
# --from-literal=user=smtp-user \
|
||||
# --from-literal=password=smtp-pass \
|
||||
# -n api \
|
||||
# -o yaml > smtp-secrets.yaml
|
||||
#
|
||||
# Then encrypt with SOPS:
|
||||
# sops -e smtp-secrets.yaml > smtp-secrets.enc.yaml
|
||||
# rm smtp-secrets.yaml
|
||||
@@ -1,48 +0,0 @@
|
||||
# ServiceAccount and RBAC for CI runner to create/watch Tekton PipelineRuns.
|
||||
# Applied to the `api` namespace where PipelineRuns execute.
|
||||
apiVersion: v1
|
||||
kind: ServiceAccount
|
||||
metadata:
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: ci
|
||||
---
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: Role
|
||||
metadata:
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
rules:
|
||||
- apiGroups: ["tekton.dev"]
|
||||
resources: ["taskruns"]
|
||||
verbs: ["create", "get", "list", "watch", "delete"]
|
||||
- apiGroups: [""]
|
||||
resources: ["pods", "pods/log"]
|
||||
verbs: ["get", "list"]
|
||||
---
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: RoleBinding
|
||||
metadata:
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
subjects:
|
||||
- kind: ServiceAccount
|
||||
name: ci-tekton-trigger
|
||||
namespace: api
|
||||
roleRef:
|
||||
kind: Role
|
||||
name: ci-tekton-trigger
|
||||
apiGroup: rbac.authorization.k8s.io
|
||||
---
|
||||
# Secret to generate a long-lived token for the CI runner.
|
||||
# The runner mounts this as KUBECONFIG_B64 or uses it directly.
|
||||
apiVersion: v1
|
||||
kind: Secret
|
||||
metadata:
|
||||
name: ci-tekton-trigger-token
|
||||
namespace: api
|
||||
annotations:
|
||||
kubernetes.io/service-account.name: ci-tekton-trigger
|
||||
type: kubernetes.io/service-account-token
|
||||
@@ -1,25 +0,0 @@
|
||||
apiVersion: kustomize.config.k8s.io/v1beta1
|
||||
kind: Kustomization
|
||||
|
||||
namespace: api
|
||||
|
||||
resources:
|
||||
- ci-rbac.yaml
|
||||
- task-integration-test.yaml
|
||||
- task-load-test.yaml
|
||||
- task-workflow-visibility.yaml
|
||||
- pipeline-sse-optimization.yaml
|
||||
|
||||
generatorOptions:
|
||||
disableNameSuffixHash: true
|
||||
|
||||
configMapGenerator:
|
||||
- name: integration-test-script
|
||||
files:
|
||||
- scripts/integration-test.sh
|
||||
- name: load-test-script
|
||||
files:
|
||||
- scripts/load-test.sh
|
||||
- name: workflow-visibility-test-script
|
||||
files:
|
||||
- scripts/workflow-visibility-test.sh
|
||||
@@ -1,124 +0,0 @@
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: Pipeline
|
||||
metadata:
|
||||
name: sse-optimization-tests
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: ci-cd
|
||||
spec:
|
||||
description: >
|
||||
Test pipeline for SSE optimization (issues #31, #32, #33).
|
||||
Runs both functional integration tests and performance load tests.
|
||||
|
||||
params:
|
||||
- name: image
|
||||
type: string
|
||||
description: "Container image to test (repo:tag)"
|
||||
- name: gateway-port
|
||||
type: string
|
||||
default: "8080"
|
||||
|
||||
tasks:
|
||||
# Functional integration tests first (quick smoke test)
|
||||
- name: integration-tests
|
||||
taskRef:
|
||||
name: integration-test
|
||||
params:
|
||||
- name: image
|
||||
value: $(params.image)
|
||||
- name: gateway-port
|
||||
value: $(params.gateway-port)
|
||||
|
||||
# Workflow visibility tests (runs after integration tests pass)
|
||||
- name: workflow-visibility-tests
|
||||
runAfter:
|
||||
- integration-tests
|
||||
taskRef:
|
||||
name: workflow-visibility-test
|
||||
params:
|
||||
- name: image
|
||||
value: $(params.image)
|
||||
- name: gateway-port
|
||||
value: $(params.gateway-port)
|
||||
|
||||
# Performance load tests (runs after integration tests pass)
|
||||
- name: load-tests
|
||||
runAfter:
|
||||
- workflow-visibility-tests
|
||||
taskRef:
|
||||
name: load-test-sse-streaming
|
||||
params:
|
||||
- name: image
|
||||
value: $(params.image)
|
||||
- name: gateway-port
|
||||
value: $(params.gateway-port)
|
||||
- name: concurrent-streams
|
||||
value: "10"
|
||||
- name: events-per-stream
|
||||
value: "100"
|
||||
- name: event-interval-ms
|
||||
value: "50"
|
||||
|
||||
# Summary reporter
|
||||
- name: report-results
|
||||
runAfter:
|
||||
- load-tests
|
||||
- workflow-visibility-tests
|
||||
taskSpec:
|
||||
description: "Report combined test results"
|
||||
params:
|
||||
- name: integration-result
|
||||
type: string
|
||||
- name: integration-summary
|
||||
type: string
|
||||
- name: workflow-result
|
||||
type: string
|
||||
- name: workflow-summary
|
||||
type: string
|
||||
- name: load-result
|
||||
type: string
|
||||
- name: load-summary
|
||||
type: string
|
||||
- name: load-metrics
|
||||
type: string
|
||||
steps:
|
||||
- name: print-summary
|
||||
image: busybox
|
||||
script: |
|
||||
#!/bin/sh
|
||||
echo "╔═══════════════════════════════════════════════════════════╗"
|
||||
echo "║ SSE Optimization + Workflow Tests (PR #26) ║"
|
||||
echo "╠═══════════════════════════════════════════════════════════╣"
|
||||
echo "║ ║"
|
||||
echo "║ Integration Tests: ║"
|
||||
echo "║ Status: $(params.integration-result)"
|
||||
echo "║ Summary: $(params.integration-summary)"
|
||||
echo "║ ║"
|
||||
echo "║ Workflow Visibility (namespace pass-down): ║"
|
||||
echo "║ Status: $(params.workflow-result)"
|
||||
echo "║ Summary: $(params.workflow-summary)"
|
||||
echo "║ ║"
|
||||
echo "║ Load Tests (Issues #31, #32, #33): ║"
|
||||
echo "║ Status: $(params.load-result)"
|
||||
echo "║ Summary: $(params.load-summary)"
|
||||
echo "║ ║"
|
||||
echo "║ Performance Metrics: ║"
|
||||
echo "║ $(params.load-metrics)"
|
||||
echo "║ ║"
|
||||
echo "╚═══════════════════════════════════════════════════════════╝"
|
||||
params:
|
||||
- name: integration-result
|
||||
value: $(tasks.integration-tests.results.result)
|
||||
- name: integration-summary
|
||||
value: $(tasks.integration-tests.results.summary)
|
||||
- name: workflow-result
|
||||
value: $(tasks.workflow-visibility-tests.results.result)
|
||||
- name: workflow-summary
|
||||
value: $(tasks.workflow-visibility-tests.results.summary)
|
||||
- name: load-result
|
||||
value: $(tasks.load-tests.results.result)
|
||||
- name: load-summary
|
||||
value: $(tasks.load-tests.results.summary)
|
||||
- name: load-metrics
|
||||
value: $(tasks.load-tests.results.metrics)
|
||||
@@ -1,140 +0,0 @@
|
||||
#!/bin/sh
|
||||
set -e
|
||||
|
||||
# Integration test runner for API gateway.
|
||||
# Tests X-Service + X-Resource header routing against a gateway on localhost.
|
||||
#
|
||||
# Required env:
|
||||
# GW — gateway base URL (e.g. http://localhost:8080)
|
||||
# RESULTS_DIR — directory to write Tekton results
|
||||
|
||||
PASS=0; FAIL=0; TOTAL=0
|
||||
|
||||
assert() {
|
||||
NAME="$1"; EXPECT="$2"
|
||||
shift 2
|
||||
TOTAL=$((TOTAL + 1))
|
||||
CODE=$(curl -s -o /dev/null -w '%{http_code}' "$@" 2>/dev/null || echo "000")
|
||||
|
||||
if [ "$CODE" = "$EXPECT" ]; then
|
||||
echo " ✓ ${NAME} (${CODE})"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ✗ ${NAME} — expected ${EXPECT}, got ${CODE}"
|
||||
FAIL=$((FAIL + 1))
|
||||
fi
|
||||
}
|
||||
|
||||
# ── Wait for sidecar gateway ──
|
||||
echo "⏳ Waiting for gateway sidecar..."
|
||||
READY=false
|
||||
for i in $(seq 1 60); do
|
||||
CODE=$(curl -s -o /dev/null -w '%{http_code}' "${GW}/healthz" 2>/dev/null || echo "000")
|
||||
if [ "$CODE" = "200" ]; then
|
||||
sleep 1
|
||||
C2=$(curl -s -o /dev/null -w '%{http_code}' "${GW}/healthz" 2>/dev/null || echo "000")
|
||||
C3=$(curl -s -o /dev/null -w '%{http_code}' "${GW}/healthz" 2>/dev/null || echo "000")
|
||||
if [ "$C2" = "200" ] && [ "$C3" = "200" ]; then
|
||||
READY=true
|
||||
echo "✓ Gateway ready"
|
||||
break
|
||||
fi
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
|
||||
if [ "$READY" = "false" ]; then
|
||||
echo "✗ Gateway never became ready"
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
echo "0/0 gateway timeout" > "${RESULTS_DIR}/summary"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo ""
|
||||
echo "═══ Integration Tests ═══"
|
||||
echo ""
|
||||
|
||||
# ── Health ──
|
||||
echo "▸ Health"
|
||||
assert "GET /healthz" 200 -X GET "${GW}/healthz"
|
||||
assert "GET /readyz" 200 -X GET "${GW}/readyz"
|
||||
|
||||
# ── Header validation ──
|
||||
echo "▸ Header validation"
|
||||
assert "X-Service without X-Resource → 400" 400 \
|
||||
-X GET -H "X-Service: memory" "${GW}/"
|
||||
assert "unknown service → 404" 404 \
|
||||
-X GET -H "X-Service: nonexistent" -H "X-Resource: foo" "${GW}/"
|
||||
|
||||
# ── S3 (no auth, MinIO rejects → 403) ──
|
||||
echo "▸ S3 service"
|
||||
assert "s3/list-objects" 403 \
|
||||
-X GET -H "X-Service: s3" -H "X-Resource: list-objects" "${GW}/"
|
||||
|
||||
# ── SQS (auth required → 401) ──
|
||||
echo "▸ SQS service"
|
||||
assert "sqs/list-queues" 401 \
|
||||
-X GET -H "X-Service: sqs" -H "X-Resource: list-queues" "${GW}/"
|
||||
|
||||
# ── Workflow visibility (namespace pass-down) ──
|
||||
echo "▸ Workflow service"
|
||||
|
||||
# Test 1: List workflows in poimen-harness namespace (should see 4 terminated workflows)
|
||||
echo " Testing workflow visibility in poimen-harness namespace..."
|
||||
WF_LIST=$(curl -s -X POST \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: list" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"namespace": "poimen-harness"}' \
|
||||
"${GW}/" 2>/dev/null || echo '{}')
|
||||
|
||||
# Check if response contains workflows
|
||||
if echo "$WF_LIST" | grep -q '"executions"'; then
|
||||
echo " ✓ Workflow list returned (poimen-harness namespace)"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ✗ Workflow list failed to return executions"
|
||||
FAIL=$((FAIL + 1))
|
||||
fi
|
||||
TOTAL=$((TOTAL + 1))
|
||||
|
||||
# Test 2: Verify we can query terminated workflows
|
||||
echo " Testing terminated workflow visibility..."
|
||||
if echo "$WF_LIST" | grep -q '"Completed\|"status"'; then
|
||||
echo " ✓ Found completed/terminated workflows in response"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ⚠ No terminated workflows found in response (may be empty namespace)"
|
||||
# Don't fail if namespace is empty - just note it
|
||||
fi
|
||||
TOTAL=$((TOTAL + 1))
|
||||
|
||||
# Test 3: Verify namespace is required (missing namespace → 400)
|
||||
echo " Testing namespace validation..."
|
||||
NO_NS=$(curl -s -w '%{http_code}' -X POST \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: list" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{}' \
|
||||
"${GW}/" 2>/dev/null || echo "000")
|
||||
|
||||
if [ "$NO_NS" = "400" ]; then
|
||||
echo " ✓ Correctly rejected list without namespace (400)"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ✗ Expected 400 for missing namespace, got $NO_NS"
|
||||
FAIL=$((FAIL + 1))
|
||||
fi
|
||||
TOTAL=$((TOTAL + 1))
|
||||
|
||||
echo ""
|
||||
echo "═══ Results: ${PASS}/${TOTAL} passed, ${FAIL} failed ═══"
|
||||
|
||||
if [ "$FAIL" -eq 0 ]; then
|
||||
echo "pass" > "${RESULTS_DIR}/result"
|
||||
else
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
fi
|
||||
echo "${PASS}/${TOTAL} passed, ${FAIL} failed" > "${RESULTS_DIR}/summary"
|
||||
|
||||
[ "$FAIL" -eq 0 ]
|
||||
@@ -1,224 +0,0 @@
|
||||
#!/bin/sh
|
||||
set -e
|
||||
|
||||
# Load test for SSE streaming with concurrent streams.
|
||||
# Measures TTFT, throughput, latency distribution, and backpressure.
|
||||
# Tests issues #31 (TCP backpressure), #32 (HTTP/2 multiplexing), #33 (no buffering).
|
||||
#
|
||||
# Required env:
|
||||
# GW — gateway base URL (e.g. http://localhost:8080)
|
||||
# CONCURRENT_STREAMS — number of concurrent streams (default: 10)
|
||||
# EVENTS_PER_STREAM — events per stream (default: 100)
|
||||
# EVENT_INTERVAL_MS — ms between events (default: 50)
|
||||
# RESULTS_DIR — directory to write Tekton results
|
||||
|
||||
: "${CONCURRENT_STREAMS:=10}"
|
||||
: "${EVENTS_PER_STREAM:=100}"
|
||||
: "${EVENT_INTERVAL_MS:=50}"
|
||||
: "${RESULTS_DIR:=/tekton/results}"
|
||||
|
||||
TEMP_DIR=$(mktemp -d)
|
||||
trap "rm -rf $TEMP_DIR" EXIT
|
||||
|
||||
# ── Wait for gateway ready ──
|
||||
echo "⏳ Waiting for gateway sidecar..."
|
||||
READY=false
|
||||
for i in $(seq 1 60); do
|
||||
if curl -s -f "${GW}/healthz" > /dev/null 2>&1; then
|
||||
echo "✓ Gateway ready"
|
||||
READY=true
|
||||
break
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
|
||||
if [ "$READY" = "false" ]; then
|
||||
echo "✗ Gateway never became ready"
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
echo "gateway timeout" > "${RESULTS_DIR}/summary"
|
||||
echo '{"error":"gateway_timeout"}' > "${RESULTS_DIR}/metrics"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Give gateway a moment to stabilize
|
||||
sleep 2
|
||||
|
||||
echo ""
|
||||
echo "═══ SSE Streaming Load Test ═══"
|
||||
echo "Concurrent streams: $CONCURRENT_STREAMS"
|
||||
echo "Events per stream: $EVENTS_PER_STREAM"
|
||||
echo "Event interval: ${EVENT_INTERVAL_MS}ms"
|
||||
echo ""
|
||||
|
||||
# Create upstream mock that simulates LLM streaming
|
||||
# This is a simple curl request that streams SSE events
|
||||
UPSTREAM_URL="${GW}/healthz"
|
||||
|
||||
# Counter for metrics
|
||||
TOTAL_EVENTS=0
|
||||
TOTAL_TIME_MS=0
|
||||
MIN_TTFT_MS=999999
|
||||
MAX_TTFT_MS=0
|
||||
FAILED_STREAMS=0
|
||||
|
||||
# Launch concurrent streams
|
||||
for stream_id in $(seq 1 "$CONCURRENT_STREAMS"); do
|
||||
(
|
||||
# Each stream makes concurrent requests and measures latency
|
||||
METRICS_FILE="${TEMP_DIR}/stream_${stream_id}_metrics.txt"
|
||||
STREAM_START=$(date +%s%3N)
|
||||
FIRST_BYTE_TIME=""
|
||||
EVENT_COUNT=0
|
||||
|
||||
# Simulate SSE stream with curl (timeout+head to get first byte timing)
|
||||
# In real scenario, this would be /v1/chat/completions with SSE response
|
||||
CURL_START=$(date +%s%N)
|
||||
|
||||
# Use curl to measure time-to-first-byte
|
||||
curl -s -w "\nTTFB:%{time_starttransfer}\nTOTAL:%{time_total}" \
|
||||
"${GW}/healthz" > "${METRICS_FILE}.raw" 2>&1 || true
|
||||
|
||||
CURL_END=$(date +%s%N)
|
||||
CURL_TIME_MS=$(( (CURL_END - CURL_START) / 1000000 ))
|
||||
|
||||
# Extract TTFB from curl output
|
||||
TTFB=$(grep "^TTFB:" "${METRICS_FILE}.raw" | cut -d: -f2 | awk '{print int($1 * 1000)}' || echo "0")
|
||||
TOTAL_TIME=$(grep "^TOTAL:" "${METRICS_FILE}.raw" | cut -d: -f2 | awk '{print int($1 * 1000)}' || echo "0")
|
||||
|
||||
# Store metrics
|
||||
echo "$TTFB" > "${METRICS_FILE}.ttfb"
|
||||
echo "$TOTAL_TIME" > "${METRICS_FILE}.total"
|
||||
|
||||
if [ "$TTFB" -gt 0 ]; then
|
||||
if [ "$TTFB" -lt "$MIN_TTFT_MS" ]; then
|
||||
echo "$TTFB" > "${TEMP_DIR}/min_ttft"
|
||||
fi
|
||||
if [ "$TTFB" -gt "$MAX_TTFT_MS" ]; then
|
||||
echo "$TTFB" > "${TEMP_DIR}/max_ttft"
|
||||
fi
|
||||
fi
|
||||
|
||||
rm -f "${METRICS_FILE}.raw"
|
||||
) &
|
||||
done
|
||||
|
||||
# Wait for all streams to complete
|
||||
wait
|
||||
echo "✓ All concurrent streams completed"
|
||||
|
||||
# Collect metrics from all streams
|
||||
echo ""
|
||||
echo "═══ Metrics Collection ═══"
|
||||
|
||||
TTFB_VALUES=""
|
||||
TOTAL_VALUES=""
|
||||
VALID_STREAMS=0
|
||||
|
||||
for stream_id in $(seq 1 "$CONCURRENT_STREAMS"); do
|
||||
TTFB_FILE="${TEMP_DIR}/stream_${stream_id}_metrics.txt.ttfb"
|
||||
TOTAL_FILE="${TEMP_DIR}/stream_${stream_id}_metrics.txt.total"
|
||||
|
||||
if [ -f "$TTFB_FILE" ] && [ -f "$TOTAL_FILE" ]; then
|
||||
TTFB=$(cat "$TTFB_FILE" 2>/dev/null || echo "0")
|
||||
TOTAL=$(cat "$TOTAL_FILE" 2>/dev/null || echo "0")
|
||||
|
||||
if [ "$TTFB" -gt 0 ]; then
|
||||
TTFB_VALUES="${TTFB_VALUES}${TTFB} "
|
||||
TOTAL_VALUES="${TOTAL_VALUES}${TOTAL} "
|
||||
VALID_STREAMS=$((VALID_STREAMS + 1))
|
||||
fi
|
||||
fi
|
||||
done
|
||||
|
||||
# Calculate statistics (sort and pick percentiles)
|
||||
if [ "$VALID_STREAMS" -gt 0 ]; then
|
||||
# Sort TTFB values
|
||||
SORTED_TTFB=$(echo "$TTFB_VALUES" | tr ' ' '\n' | sort -n | grep -v '^$')
|
||||
|
||||
# Calculate percentiles
|
||||
P50_TTFB=$(echo "$SORTED_TTFB" | awk '{arr[NR]=$0} END {print arr[int(NR*0.5)]}')
|
||||
P99_TTFB=$(echo "$SORTED_TTFB" | awk '{arr[NR]=$0} END {print arr[int(NR*0.99)]}')
|
||||
MIN_TTFB=$(echo "$SORTED_TTFB" | head -1)
|
||||
MAX_TTFB=$(echo "$SORTED_TTFB" | tail -1)
|
||||
|
||||
# Calculate average
|
||||
AVG_TTFB=$(echo "$SORTED_TTFB" | awk '{sum+=$0; n++} END {if(n>0) print int(sum/n); else print 0}')
|
||||
|
||||
# Throughput: events/sec (simplified: using successful streams)
|
||||
THROUGHPUT=$(echo "scale=2; $VALID_STREAMS * 1000 / $MAX_TTFB" | bc 2>/dev/null || echo "0")
|
||||
|
||||
echo "✓ Streams completed: $VALID_STREAMS/$CONCURRENT_STREAMS"
|
||||
echo "✓ TTFB (Time-To-First-Byte):"
|
||||
echo " Min: ${MIN_TTFB}ms"
|
||||
echo " P50: ${P50_TTFB}ms"
|
||||
echo " P99: ${P99_TTFB}ms"
|
||||
echo " Max: ${MAX_TTFB}ms"
|
||||
echo " Avg: ${AVG_TTFB}ms"
|
||||
echo "✓ Throughput: ~${THROUGHPUT} streams/sec"
|
||||
|
||||
# Check pass/fail criteria
|
||||
# TTFB should be < 1000ms for health checks, < 5000ms for SSE streams
|
||||
FAIL=0
|
||||
if [ "$P99_TTFB" -gt 5000 ]; then
|
||||
echo "✗ P99 TTFB exceeds 5000ms threshold"
|
||||
FAIL=1
|
||||
fi
|
||||
|
||||
if [ "$VALID_STREAMS" -lt "$((CONCURRENT_STREAMS / 2))" ]; then
|
||||
echo "✗ Less than 50% of streams completed successfully"
|
||||
FAIL=1
|
||||
fi
|
||||
|
||||
# Write results
|
||||
if [ "$FAIL" -eq 0 ]; then
|
||||
echo "pass" > "${RESULTS_DIR}/result"
|
||||
SUMMARY="${VALID_STREAMS}/${CONCURRENT_STREAMS} streams OK | P50 TTFB: ${P50_TTFB}ms | P99 TTFB: ${P99_TTFB}ms | Throughput: ${THROUGHPUT} streams/sec"
|
||||
else
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
SUMMARY="FAILED: ${VALID_STREAMS}/${CONCURRENT_STREAMS} streams completed | P99 TTFB: ${P99_TTFB}ms (threshold: 5000ms)"
|
||||
fi
|
||||
|
||||
# Write detailed metrics
|
||||
cat > "${RESULTS_DIR}/metrics" <<EOF
|
||||
{
|
||||
"test_type": "sse_streaming_load_test",
|
||||
"timestamp": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
|
||||
"configuration": {
|
||||
"concurrent_streams": $CONCURRENT_STREAMS,
|
||||
"events_per_stream": $EVENTS_PER_STREAM,
|
||||
"event_interval_ms": $EVENT_INTERVAL_MS
|
||||
},
|
||||
"results": {
|
||||
"streams_completed": $VALID_STREAMS,
|
||||
"streams_total": $CONCURRENT_STREAMS,
|
||||
"ttfb_ms": {
|
||||
"min": $MIN_TTFB,
|
||||
"p50": $P50_TTFB,
|
||||
"p99": $P99_TTFB,
|
||||
"max": $MAX_TTFB,
|
||||
"avg": $AVG_TTFB
|
||||
},
|
||||
"throughput_streams_per_sec": $THROUGHPUT
|
||||
},
|
||||
"issues_tested": [
|
||||
"#31: TCP backpressure for streaming LLM responses",
|
||||
"#32: HTTP/2 multiplexing for concurrent streams",
|
||||
"#33: Disable proxy buffering for SSE"
|
||||
]
|
||||
}
|
||||
EOF
|
||||
|
||||
else
|
||||
echo "✗ No valid streams collected"
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
echo "no_valid_streams" > "${RESULTS_DIR}/summary"
|
||||
echo '{"error":"no_valid_streams"}' > "${RESULTS_DIR}/metrics"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo ""
|
||||
echo "═══ Summary ═══"
|
||||
echo "$SUMMARY"
|
||||
echo "$SUMMARY" > "${RESULTS_DIR}/summary"
|
||||
|
||||
exit "$FAIL"
|
||||
@@ -1,154 +0,0 @@
|
||||
#!/bin/sh
|
||||
set -e
|
||||
|
||||
# Workflow visibility test for gateway.
|
||||
# Verifies that the WorkflowAdapter provides visibility into terminated workflows
|
||||
# in the poimen-harness namespace via X-Service: workflow routing.
|
||||
#
|
||||
# Expected: 4 terminated workflows in poimen-harness namespace
|
||||
#
|
||||
# Required env:
|
||||
# GW — gateway base URL (e.g. http://localhost:8080)
|
||||
# RESULTS_DIR — directory to write Tekton results
|
||||
|
||||
: "${RESULTS_DIR:=/tekton/results}"
|
||||
|
||||
PASS=0
|
||||
FAIL=0
|
||||
TOTAL=0
|
||||
|
||||
echo "═══ Workflow Visibility Test ═══"
|
||||
echo ""
|
||||
echo "Testing WorkflowAdapter namespace pass-down"
|
||||
echo "Expected: 4 terminated workflows in poimen-harness namespace"
|
||||
echo ""
|
||||
|
||||
# ── Wait for gateway ──
|
||||
echo "⏳ Waiting for gateway..."
|
||||
READY=false
|
||||
for i in $(seq 1 60); do
|
||||
if curl -s -f "${GW}/healthz" > /dev/null 2>&1; then
|
||||
echo "✓ Gateway ready"
|
||||
READY=true
|
||||
break
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
|
||||
if [ "$READY" = "false" ]; then
|
||||
echo "✗ Gateway timeout"
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
echo "Gateway did not become ready" > "${RESULTS_DIR}/summary"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# ── Test 1: List workflows in poimen-harness ──
|
||||
TOTAL=$((TOTAL + 1))
|
||||
echo "Test 1: List workflows in poimen-harness namespace"
|
||||
|
||||
WF_RESPONSE=$(curl -s -X POST \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: list" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"namespace": "poimen-harness"}' \
|
||||
"${GW}/" 2>/dev/null || echo "")
|
||||
|
||||
if [ -z "$WF_RESPONSE" ]; then
|
||||
echo " ✗ No response from workflow list endpoint"
|
||||
FAIL=$((FAIL + 1))
|
||||
else
|
||||
echo " ✓ Received workflow list response"
|
||||
PASS=$((PASS + 1))
|
||||
|
||||
# Extract workflow count (if available)
|
||||
WF_COUNT=$(echo "$WF_RESPONSE" | grep -o '"execution_time"' | wc -l || echo "0")
|
||||
echo " Found workflows: $WF_COUNT"
|
||||
fi
|
||||
|
||||
# ── Test 2: Verify namespace is required ──
|
||||
TOTAL=$((TOTAL + 1))
|
||||
echo "Test 2: Namespace validation (missing namespace should fail)"
|
||||
|
||||
NO_NS_RESPONSE=$(curl -s -w "\n%{http_code}" -X POST \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: list" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{}' \
|
||||
"${GW}/" 2>/dev/null || echo "")
|
||||
|
||||
NO_NS_CODE=$(echo "$NO_NS_RESPONSE" | tail -1)
|
||||
|
||||
if [ "$NO_NS_CODE" = "400" ]; then
|
||||
echo " ✓ Correctly rejected missing namespace (HTTP 400)"
|
||||
PASS=$((PASS + 1))
|
||||
elif [ "$NO_NS_CODE" = "401" ]; then
|
||||
echo " ⚠ Got 401 (auth required) - namespace validation happens after auth check"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ✗ Expected 400/401, got $NO_NS_CODE"
|
||||
FAIL=$((FAIL + 1))
|
||||
fi
|
||||
|
||||
# ── Test 3: Query specific terminated workflow ──
|
||||
TOTAL=$((TOTAL + 1))
|
||||
echo "Test 3: Describe specific workflow (if available)"
|
||||
|
||||
# Try to describe a workflow - this will fail if no workflows exist, but shows the feature works
|
||||
DESCRIBE_RESPONSE=$(curl -s -X POST \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: describe" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"namespace": "poimen-harness", "workflow_id": "test-workflow"}' \
|
||||
"${GW}/" 2>/dev/null || echo "")
|
||||
|
||||
if [ -n "$DESCRIBE_RESPONSE" ]; then
|
||||
echo " ✓ Describe endpoint responded"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ⚠ Describe endpoint no response (may indicate workflow doesn't exist)"
|
||||
# Not a failure - endpoint exists but workflow may not
|
||||
fi
|
||||
|
||||
# ── Test 4: Verify auth requirement ──
|
||||
TOTAL=$((TOTAL + 1))
|
||||
echo "Test 4: Auth requirement (workflow service requires Authorization)"
|
||||
|
||||
NO_AUTH_CODE=$(curl -s -w '%{http_code}' -o /dev/null -X POST \
|
||||
-H "X-Service: workflow" \
|
||||
-H "X-Resource: list" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"namespace": "poimen-harness"}' \
|
||||
"${GW}/" 2>/dev/null || echo "000")
|
||||
|
||||
if [ "$NO_AUTH_CODE" = "401" ]; then
|
||||
echo " ✓ Correctly requires auth (HTTP 401)"
|
||||
PASS=$((PASS + 1))
|
||||
else
|
||||
echo " ✗ Expected 401, got $NO_AUTH_CODE"
|
||||
echo " (Auth may be disabled in test environment)"
|
||||
FAIL=$((FAIL + 1))
|
||||
fi
|
||||
|
||||
# ── Summary ──
|
||||
echo ""
|
||||
echo "═══ Results ═══"
|
||||
echo "Passed: $PASS/$TOTAL"
|
||||
echo "Failed: $FAIL/$TOTAL"
|
||||
echo ""
|
||||
|
||||
if [ "$FAIL" -eq 0 ]; then
|
||||
echo "pass" > "${RESULTS_DIR}/result"
|
||||
SUMMARY="Workflow visibility test passed. WorkflowAdapter can list/describe workflows in poimen-harness namespace with namespace pass-down support."
|
||||
echo "✓ All tests passed"
|
||||
else
|
||||
echo "fail" > "${RESULTS_DIR}/result"
|
||||
SUMMARY="$FAIL tests failed. Check WorkflowAdapter implementation and namespace validation."
|
||||
echo "✗ Some tests failed"
|
||||
fi
|
||||
|
||||
echo "$SUMMARY" > "${RESULTS_DIR}/summary"
|
||||
echo "" >> "${RESULTS_DIR}/summary"
|
||||
echo "Passed: $PASS/$TOTAL" >> "${RESULTS_DIR}/summary"
|
||||
echo "Failed: $FAIL/$TOTAL" >> "${RESULTS_DIR}/summary"
|
||||
|
||||
[ "$FAIL" -eq 0 ]
|
||||
@@ -1,75 +0,0 @@
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: Task
|
||||
metadata:
|
||||
name: integration-test
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: testing
|
||||
spec:
|
||||
description: >
|
||||
Spin up a gateway pod from the given image as a sidecar,
|
||||
run curl-based integration tests, report pass/fail.
|
||||
params:
|
||||
- name: image
|
||||
type: string
|
||||
description: "Container image to test (repo:tag)"
|
||||
- name: gateway-port
|
||||
type: string
|
||||
default: "8080"
|
||||
results:
|
||||
- name: result
|
||||
type: string
|
||||
- name: summary
|
||||
type: string
|
||||
|
||||
sidecars:
|
||||
- name: gateway
|
||||
image: $(params.image)
|
||||
env:
|
||||
- name: LISTEN_ADDR
|
||||
value: "0.0.0.0:$(params.gateway-port)"
|
||||
- name: CONFIG_PATH
|
||||
value: /etc/gateway/config.yaml
|
||||
- name: LOG_LEVEL
|
||||
value: info
|
||||
- name: AUTH_CLIENT_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: api-gw-client-secret
|
||||
key: client-secret
|
||||
optional: true
|
||||
volumeMounts:
|
||||
- name: gateway-config
|
||||
mountPath: /etc/gateway
|
||||
readOnly: true
|
||||
|
||||
steps:
|
||||
- name: run-tests
|
||||
image: curlimages/curl:8.13.0
|
||||
env:
|
||||
- name: GW
|
||||
value: "http://localhost:$(params.gateway-port)"
|
||||
- name: RESULTS_DIR
|
||||
value: /tekton/results
|
||||
command: ["sh", "/scripts/integration-test.sh"]
|
||||
volumeMounts:
|
||||
- name: test-script
|
||||
mountPath: /scripts
|
||||
readOnly: true
|
||||
computeResources:
|
||||
requests:
|
||||
cpu: 100m
|
||||
memory: 64Mi
|
||||
limits:
|
||||
cpu: 200m
|
||||
memory: 128Mi
|
||||
|
||||
volumes:
|
||||
- name: gateway-config
|
||||
secret:
|
||||
secretName: api-gateway-config
|
||||
- name: test-script
|
||||
configMap:
|
||||
name: integration-test-script
|
||||
defaultMode: 0755
|
||||
@@ -1,101 +0,0 @@
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: Task
|
||||
metadata:
|
||||
name: load-test-sse-streaming
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: performance-testing
|
||||
spec:
|
||||
description: >
|
||||
Load-test SSE streaming with concurrent streams.
|
||||
Measures TTFT (time-to-first-token), throughput, latency distribution,
|
||||
and backpressure handling. Tests issues #31, #32, #33.
|
||||
params:
|
||||
- name: image
|
||||
type: string
|
||||
description: "Container image to test (repo:tag)"
|
||||
- name: gateway-port
|
||||
type: string
|
||||
default: "8080"
|
||||
- name: concurrent-streams
|
||||
type: string
|
||||
default: "10"
|
||||
description: "Number of concurrent SSE streams to generate"
|
||||
- name: events-per-stream
|
||||
type: string
|
||||
default: "100"
|
||||
description: "Number of events each stream should receive"
|
||||
- name: event-interval-ms
|
||||
type: string
|
||||
default: "50"
|
||||
description: "Milliseconds between events from upstream"
|
||||
results:
|
||||
- name: result
|
||||
type: string
|
||||
description: "pass or fail"
|
||||
- name: summary
|
||||
type: string
|
||||
description: "Summary of load test results"
|
||||
- name: metrics
|
||||
type: string
|
||||
description: "Raw metrics JSON (TTFT, throughput, latency percentiles)"
|
||||
|
||||
sidecars:
|
||||
- name: gateway
|
||||
image: $(params.image)
|
||||
env:
|
||||
- name: LISTEN_ADDR
|
||||
value: "0.0.0.0:$(params.gateway-port)"
|
||||
- name: CONFIG_PATH
|
||||
value: /etc/gateway/config.yaml
|
||||
- name: LOG_LEVEL
|
||||
value: info
|
||||
- name: AUTH_CLIENT_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: api-gw-client-secret
|
||||
key: client-secret
|
||||
optional: true
|
||||
volumeMounts:
|
||||
- name: gateway-config
|
||||
mountPath: /etc/gateway
|
||||
readOnly: true
|
||||
|
||||
steps:
|
||||
- name: run-load-test
|
||||
image: curlimages/curl:8.13.0
|
||||
env:
|
||||
- name: GW
|
||||
value: "http://localhost:$(params.gateway-port)"
|
||||
- name: CONCURRENT_STREAMS
|
||||
value: $(params.concurrent-streams)
|
||||
- name: EVENTS_PER_STREAM
|
||||
value: $(params.events-per-stream)
|
||||
- name: EVENT_INTERVAL_MS
|
||||
value: $(params.event-interval-ms)
|
||||
- name: RESULTS_DIR
|
||||
value: /tekton/results
|
||||
command: ["sh", "/scripts/load-test.sh"]
|
||||
volumeMounts:
|
||||
- name: test-script
|
||||
mountPath: /scripts
|
||||
readOnly: true
|
||||
computeResources:
|
||||
requests:
|
||||
cpu: 500m
|
||||
memory: 256Mi
|
||||
limits:
|
||||
cpu: 1000m
|
||||
memory: 512Mi
|
||||
# Load test needs more time than unit tests
|
||||
timeout: 10m
|
||||
|
||||
volumes:
|
||||
- name: gateway-config
|
||||
secret:
|
||||
secretName: api-gateway-config
|
||||
- name: test-script
|
||||
configMap:
|
||||
name: load-test-script
|
||||
defaultMode: 0755
|
||||
@@ -1,84 +0,0 @@
|
||||
apiVersion: tekton.dev/v1
|
||||
kind: Task
|
||||
metadata:
|
||||
name: workflow-visibility-test
|
||||
namespace: api
|
||||
labels:
|
||||
app: api-gateway
|
||||
component: testing
|
||||
spec:
|
||||
description: >
|
||||
Test workflow visibility via WorkflowAdapter.
|
||||
Verifies that the gateway provides visibility into terminated workflows
|
||||
in the poimen-harness namespace via X-Service: workflow routing.
|
||||
This ensures namespace pass-down is working correctly.
|
||||
|
||||
params:
|
||||
- name: image
|
||||
type: string
|
||||
description: "Container image to test (repo:tag)"
|
||||
- name: gateway-port
|
||||
type: string
|
||||
default: "8080"
|
||||
|
||||
results:
|
||||
- name: result
|
||||
type: string
|
||||
description: "pass or fail"
|
||||
- name: summary
|
||||
type: string
|
||||
description: "Test summary"
|
||||
- name: workflow-count
|
||||
type: string
|
||||
description: "Number of workflows found in poimen-harness"
|
||||
|
||||
sidecars:
|
||||
- name: gateway
|
||||
image: $(params.image)
|
||||
env:
|
||||
- name: LISTEN_ADDR
|
||||
value: "0.0.0.0:$(params.gateway-port)"
|
||||
- name: CONFIG_PATH
|
||||
value: /etc/gateway/config.yaml
|
||||
- name: LOG_LEVEL
|
||||
value: info
|
||||
- name: AUTH_CLIENT_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: api-gw-client-secret
|
||||
key: client-secret
|
||||
optional: true
|
||||
volumeMounts:
|
||||
- name: gateway-config
|
||||
mountPath: /etc/gateway
|
||||
readOnly: true
|
||||
|
||||
steps:
|
||||
- name: run-workflow-visibility-test
|
||||
image: curlimages/curl:8.13.0
|
||||
env:
|
||||
- name: GW
|
||||
value: "http://localhost:$(params.gateway-port)"
|
||||
- name: RESULTS_DIR
|
||||
value: /tekton/results
|
||||
command: ["sh", "/scripts/workflow-visibility-test.sh"]
|
||||
volumeMounts:
|
||||
- name: test-script
|
||||
mountPath: /scripts
|
||||
readOnly: true
|
||||
computeResources:
|
||||
requests:
|
||||
cpu: 100m
|
||||
memory: 64Mi
|
||||
limits:
|
||||
cpu: 200m
|
||||
memory: 128Mi
|
||||
|
||||
volumes:
|
||||
- name: gateway-config
|
||||
secret:
|
||||
secretName: api-gateway-config
|
||||
- name: test-script
|
||||
configMap:
|
||||
name: workflow-visibility-test-script
|
||||
defaultMode: 0755
|
||||
Reference in New Issue
Block a user