Author SHA1 Message Date
Admin Bot 69c506c9f9 feat(audit): log successful JWT auth with subject, acting_party, roles
- Add 'auth ok' info log after JWT validation passes
- Export ClaimString/ClaimStringSlice from identity package
- Log fields: subject, acting_party, roles/permissions, path, method
- Complements existing 'auth failed' error logs for full audit trail
2026-09-10 04:33:56 +09:00
Admin Botandpoimen dba718e87b feat: route qwen2.5:3b-instruct to CPU service on cp-2
Model moved from ornith-predictor GPU to dedicated qwen-cpu
Deployment on talos-cp-2 (CPU). Frees 1 GPU for ComfyUI.

Co-authored-by: poimen <[email protected]>
2026-09-08 18:26:32 -07:00
19 changed files with 325 additions and 879 deletions
+8 -79
View File
@@ -17,13 +17,10 @@ jobs:
name: CI name: CI
runs-on: golang runs-on: golang
steps: steps:
- name: Install dependencies - name: Install Node.js and Docker
run: | run: |
apt-get update apt-get update
apt-get install -y docker.io curl apt-get install -y nodejs docker.io
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: Checkout code - name: Checkout code
uses: actions/checkout@v4 uses: actions/checkout@v4
@@ -50,82 +47,14 @@ jobs:
run: | run: |
docker build --no-cache \ docker build --no-cache \
-t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \ -t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \
-t "${IMAGE}:latest" \
-f Dockerfile . -f Dockerfile .
- name: Push image (SHA tag) - name: Push Docker image
run: docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
# ── Tekton integration tests ─────────────────────────────
- name: Setup kubeconfig
run: | run: |
mkdir -p ~/.kube docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
echo "${KUBECONFIG_B64}" | base64 -d > ~/.kube/config
kubectl cluster-info
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 pipelinerun "${RUN_NAME}" -n api --ignore-not-found
# Create PipelineRun — spins up gateway sidecar + curl tests
cat <<YAML | kubectl create -f -
apiVersion: tekton.dev/v1
kind: PipelineRun
metadata:
name: ${RUN_NAME}
namespace: api
labels:
commit-sha: "${SHA}"
spec:
pipelineRef:
name: integration-test-pipeline
params:
- name: image
value: "${IMAGE}:${SHA}"
YAML
echo "✓ PipelineRun created: ${RUN_NAME}"
# Wait for completion (Succeeded or Failed)
echo "Waiting for tests (timeout 5m)..."
if kubectl wait pipelinerun/"${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 ==="
kubectl logs -n api "pipelinerun/${RUN_NAME}" --all-containers 2>/dev/null || true
echo ""
REASON=$(kubectl get pipelinerun "${RUN_NAME}" -n api \
-o jsonpath='{.status.conditions[0].reason}')
SUMMARY=$(kubectl get pipelinerun "${RUN_NAME}" -n api \
-o jsonpath='{.status.results[?(@.name=="test-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}:latest" docker push "${IMAGE}:latest"
echo "✓ Promoted to latest" echo "✓ Pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
- name: Cleanup - name: Prune unused images
if: always() run: docker image prune -a --force 2>&1 | tail -3 || true
run: docker image prune -af 2>&1 | tail -3 || true
-34
View File
@@ -1,34 +0,0 @@
# 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)
-17
View File
@@ -24,8 +24,6 @@ type Config struct {
Adapters []*serviceadapter.ServiceAdapter Adapters []*serviceadapter.ServiceAdapter
// Auth holds JWT authentication configuration for /v1/* endpoints. // Auth holds JWT authentication configuration for /v1/* endpoints.
Auth AuthConfig Auth AuthConfig
// Temporal holds Temporal server configuration.
Temporal TemporalConfig
} }
// ModelUpstream holds upstream configuration for a specific model. // ModelUpstream holds upstream configuration for a specific model.
@@ -40,12 +38,6 @@ type ModelUpstream struct {
AuthRequired bool 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. // AuthConfig holds JWT authentication configuration.
type AuthConfig struct { type AuthConfig struct {
// Enabled globally enables/disables auth for /v1/* endpoints. // Enabled globally enables/disables auth for /v1/* endpoints.
@@ -146,12 +138,6 @@ func Load() (*Config, error) {
authConfig = loadedAuth authConfig = loadedAuth
} }
temporalHostPort := "localhost:7233"
// Allow override via environment variable
if hostPort, ok := os.LookupEnv("TEMPORAL_HOST_PORT"); ok {
temporalHostPort = hostPort
}
return &Config{ return &Config{
ListenAddr: listenAddr, ListenAddr: listenAddr,
ShutdownTimeout: shutdownTimeout, ShutdownTimeout: shutdownTimeout,
@@ -159,8 +145,5 @@ func Load() (*Config, error) {
Models: models, Models: models,
Adapters: adapters, Adapters: adapters,
Auth: authConfig, Auth: authConfig,
Temporal: TemporalConfig{
HostPort: temporalHostPort,
},
}, nil }, nil
} }
+9 -9
View File
@@ -52,18 +52,18 @@ func StripIncoming(r *http.Request) {
func Inject(r *http.Request, claims jwt.MapClaims) { func Inject(r *http.Request, claims jwt.MapClaims) {
r.Header.Set(HeaderAuthVerified, "true") r.Header.Set(HeaderAuthVerified, "true")
if sub := claimString(claims, "sub"); sub != "" { if sub := ClaimString(claims, "sub"); sub != "" {
r.Header.Set(HeaderUser, sub) r.Header.Set(HeaderUser, sub)
} }
if roles := claimStringSlice(claims, "roles"); len(roles) > 0 { if roles := ClaimStringSlice(claims, "roles"); len(roles) > 0 {
r.Header.Set(HeaderRoles, strings.Join(roles, ",")) r.Header.Set(HeaderRoles, strings.Join(roles, ","))
} else if perms := claimStringSlice(claims, "permissions"); len(perms) > 0 { } else if perms := ClaimStringSlice(claims, "permissions"); len(perms) > 0 {
r.Header.Set(HeaderRoles, strings.Join(perms, ",")) r.Header.Set(HeaderRoles, strings.Join(perms, ","))
} }
if azp := claimString(claims, "azp"); azp != "" { if azp := ClaimString(claims, "azp"); azp != "" {
sub := claimString(claims, "sub") sub := ClaimString(claims, "sub")
// Only set acting-service when azp differs from sub // Only set acting-service when azp differs from sub
// (i.e., a service account acting, not the user themselves) // (i.e., a service account acting, not the user themselves)
if azp != sub { if azp != sub {
@@ -72,9 +72,9 @@ func Inject(r *http.Request, claims jwt.MapClaims) {
} }
} }
// claimString extracts a string value from claims, returning "" if // ClaimString extracts a string value from claims, returning "" if
// the key is missing or not a string. // the key is missing or not a string.
func claimString(claims jwt.MapClaims, key string) string { func ClaimString(claims jwt.MapClaims, key string) string {
val, ok := claims[key] val, ok := claims[key]
if !ok || val == nil { if !ok || val == nil {
return "" return ""
@@ -86,10 +86,10 @@ func claimString(claims jwt.MapClaims, key string) string {
return s return s
} }
// claimStringSlice extracts a []string from claims. JWT libraries // ClaimStringSlice extracts a []string from claims. JWT libraries
// deserialize JSON arrays as []interface{}, so each element is // deserialize JSON arrays as []interface{}, so each element is
// type-asserted individually. Non-string elements are skipped. // type-asserted individually. Non-string elements are skipped.
func claimStringSlice(claims jwt.MapClaims, key string) []string { func ClaimStringSlice(claims jwt.MapClaims, key string) []string {
val, ok := claims[key] val, ok := claims[key]
if !ok || val == nil { if !ok || val == nil {
return nil return nil
-299
View File
@@ -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"])
})
}
}
+18
View File
@@ -359,6 +359,24 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Inject identity headers for downstream services // Inject identity headers for downstream services
identity.Inject(r, claims) identity.Inject(r, claims)
// Audit trail: log successful JWT authentication
auditFields := map[string]string{
"path": r.URL.Path,
"method": r.Method,
}
if sub := identity.ClaimString(claims, "sub"); sub != "" {
auditFields["subject"] = sub
}
if azp := identity.ClaimString(claims, "azp"); azp != "" {
auditFields["acting_party"] = azp
}
if roles := identity.ClaimStringSlice(claims, "roles"); len(roles) > 0 {
auditFields["roles"] = strings.Join(roles, ",")
} else if perms := identity.ClaimStringSlice(claims, "permissions"); len(perms) > 0 {
auditFields["permissions"] = strings.Join(perms, ",")
}
logging.Infof("auth ok", auditFields)
// Check required capability if configured // Check required capability if configured
if h.config.Auth.RequiredCapability != "" { if h.config.Auth.RequiredCapability != "" {
if !h.jwtValidator.CheckPermissions(claims, h.config.Auth.RequiredCapability, "*") { if !h.jwtValidator.CheckPermissions(claims, h.config.Auth.RequiredCapability, "*") {
+2 -6
View File
@@ -33,8 +33,8 @@ func NewRouter(healthChecker *HealthChecker, dispatcher *serviceadapter.Dispatch
// ServeHTTP implements http.Handler. // ServeHTTP implements http.Handler.
// Priority order: // Priority order:
// 1. /healthz and /readyz to health handlers // 1. /healthz and /readyz to health handlers
// 2. X-Service header to ServiceAdapter dispatcher (phase 8) - PREFERRED routing method // 2. X-Service header to ServiceAdapter dispatcher (phase 8)
// 3. /workflow* to temporal handler - DEPRECATED: use X-Service: workflow instead // 3. /workflow* to temporal handler
// 4. All other paths to upstream handler (phase 0-7) // 4. All other paths to upstream handler (phase 0-7)
func (r *Router) ServeHTTP(w http.ResponseWriter, req *http.Request) { func (r *Router) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// Health endpoints first // 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 // 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 req.Header.Get("X-Service") != "" {
if r.dispatcher != nil { if r.dispatcher != nil {
r.dispatcher.Dispatch(w, req) r.dispatcher.Dispatch(w, req)
@@ -58,8 +56,6 @@ func (r *Router) ServeHTTP(w http.ResponseWriter, req *http.Request) {
} }
// Workflow endpoints // 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 { switch req.URL.Path {
case "/workflow", "/workflow/health", "/workflow/metrics": case "/workflow", "/workflow/health", "/workflow/metrics":
r.temporalHandler.ServeHTTP(w, req) r.temporalHandler.ServeHTTP(w, req)
+3 -32
View File
@@ -10,7 +10,6 @@ import (
"strings" "strings"
"time" "time"
"golang.org/x/net/http2"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/credentials/insecure"
@@ -165,10 +164,6 @@ func (d *Dispatcher) dispatchHTTP(w http.ResponseWriter, r *http.Request, upstre
req.URL.Path = method.UpstreamPath req.URL.Path = method.UpstreamPath
req.RequestURI = "" req.RequestURI = ""
req.Host = parsedURL.Host 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
} }
timeout := adapter.Spec.Upstream.TimeoutSeconds timeout := adapter.Spec.Upstream.TimeoutSeconds
@@ -219,33 +214,9 @@ func (d *Dispatcher) dispatchGRPC(w http.ResponseWriter, r *http.Request, upstre
} }
defer conn.Close() defer conn.Close()
// Create HTTP/2 reverse proxy for gRPC d.writeError(w, problem.NewProblem(http.StatusNotImplemented,
// gRPC uses HTTP/2 protocol, so we need an HTTP/2-capable transport "about:blank#not-implemented", "Not Implemented",
upstreamURLObj := &url.URL{ "gRPC forwarding not yet implemented"))
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)
} }
func (d *Dispatcher) writeError(w http.ResponseWriter, p *problem.Problem) { func (d *Dispatcher) writeError(w http.ResponseWriter, p *problem.Problem) {
+153 -20
View File
File diff suppressed because one or more lines are too long
+131
View File
@@ -0,0 +1,131 @@
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"
tokenUrl: "http://authentik-server.iam.svc.cluster.local/application/o/token/"
clientId: "api-gw"
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: "qwen-cpu.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:80
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
-10
View File
@@ -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"
-70
View File
@@ -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
+1 -1
View File
@@ -8,7 +8,7 @@ resources:
- service.yaml - service.yaml
- deployment.yaml - deployment.yaml
- network-policy.yaml - network-policy.yaml
- gateway-config-secret.yaml - gateway-config-secret.enc.yaml
# The deployed image tag lives here and nowhere else. CI publishes # 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. # forgejo.riotpiao.com/rock/api-gateway:<commit-sha> and tags it as :latest on main.
-8
View File
@@ -46,14 +46,6 @@ spec:
ports: ports:
- protocol: TCP - protocol: TCP
port: 8080 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: egress:
# Allow DNS # Allow DNS
- to: - to:
-58
View File
@@ -1,58 +0,0 @@
# Tekton Integration Tests
Curl-based integration tests for the API gateway, orchestrated by Tekton.
## How It Works
```
CI pushes image:sha → creates PipelineRun → Tekton spins up gateway sidecar
→ runs curl tests → reports pass/fail → CI promotes to :latest if pass
```
The Task runs the gateway image as a **sidecar** (same pod, localhost),
then executes `scripts/integration-test.sh` which tests every adapter
via `X-Service` + `X-Resource` header routing.
## Files
| File | Purpose |
|------|---------|
| `task-integration-test.yaml` | Task: sidecar gateway + curl test step |
| `pipeline-integration-test.yaml` | Pipeline: wraps the Task |
| `scripts/integration-test.sh` | Test script (mounted as ConfigMap) |
| `kustomization.yaml` | Generates ConfigMap from script |
## Manual Run
```bash
kubectl apply -k k8s/tekton/
kubectl create -f - <<'EOF'
apiVersion: tekton.dev/v1
kind: PipelineRun
metadata:
name: integration-test-manual
namespace: api
spec:
pipelineRef:
name: integration-test-pipeline
params:
- name: image
value: forgejo.riotpiao.com/rock/api-gateway:latest
EOF
# Watch
kubectl logs -f -n api pipelinerun/integration-test-manual -c step-run-tests
```
## Updating Tests
Edit `scripts/integration-test.sh`, then:
```bash
kubectl apply -k k8s/tekton/ # recreates ConfigMap
```
## Tekton Infrastructure
Tekton Pipelines is installed in `~/workplace/homelab` via ArgoCD
(`k8s/argocd/apps/06-ci-cd.yaml` → vendored `k8s/infra/tekton/release.yaml`).
-16
View File
@@ -1,16 +0,0 @@
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: api
resources:
- task-integration-test.yaml
- pipeline-integration-test.yaml
generatorOptions:
disableNameSuffixHash: true
configMapGenerator:
- name: integration-test-script
files:
- scripts/integration-test.sh
-30
View File
@@ -1,30 +0,0 @@
apiVersion: tekton.dev/v1
kind: Pipeline
metadata:
name: integration-test-pipeline
namespace: api
labels:
app: api-gateway
component: testing
spec:
description: >
Run integration tests against a gateway image.
Spins up the image as a sidecar, tests via curl, reports pass/fail.
params:
- name: image
type: string
description: "Container image to test (repo:sha)"
results:
- name: test-result
description: "pass or fail"
value: $(tasks.integration-test.results.result)
- name: test-summary
description: "e.g. 8/8 passed"
value: $(tasks.integration-test.results.summary)
tasks:
- name: integration-test
taskRef:
name: integration-test
params:
- name: image
value: $(params.image)
-115
View File
@@ -1,115 +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
# remaining args are the full curl flags
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 to be fully ready ──
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 (stable after 3 checks)"
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 (path-based, no headers) ──
echo "▸ Health"
assert "GET /healthz" 200 \
-X GET "${GW}/healthz"
assert "GET /readyz" 200 \
-X GET "${GW}/readyz"
# ── Routing: missing headers → 400 ──
echo "▸ Header validation"
assert "no X-Service → 400" 400 \
-X GET "${GW}/"
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}/"
# ── Memory service (POST, no auth) ──
echo "▸ Memory service"
assert "memory/query (POST)" 200 \
-X POST -H "X-Service: memory" -H "X-Resource: query" \
-H "Content-Type: application/json" -d '{"query":"test"}' "${GW}/"
assert "memory/ingest (POST)" 200 \
-X POST -H "X-Service: memory" -H "X-Resource: ingest" \
-H "Content-Type: application/json" \
-d '{"content":"integration test","metadata":{"source":"tekton"}}' "${GW}/"
# ── S3 service (GET, no auth → MinIO 403) ──
echo "▸ S3 service"
assert "s3/list-objects (GET → 403)" 403 \
-X GET -H "X-Service: s3" -H "X-Resource: list-objects" "${GW}/"
# ── SQS service (GET, auth required → 401) ──
echo "▸ SQS service"
assert "sqs/list-queues (GET → 401)" 401 \
-X GET -H "X-Service: sqs" -H "X-Resource: list-queues" "${GW}/"
# ── Workflow service (gRPC, GET) ──
echo "▸ Workflow service"
assert "workflow/list (GET → upstream err)" 502 \
-X GET -H "X-Service: workflow" -H "X-Resource: list" "${GW}/"
# ── IAM service (GET, Authentik) ──
echo "▸ IAM service"
assert "iam/list-users (GET → Authentik redirect)" 302 \
-X GET -H "X-Service: iam" -H "X-Resource: list-users" "${GW}/"
echo ""
echo "═══ Results: ${PASS}/${TOTAL} passed, ${FAIL} failed ═══"
# Write Tekton results
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 ]
-75
View File
@@ -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