diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go index 1ecbfbf..1adbb4e 100644 --- a/cmd/gateway/main.go +++ b/cmd/gateway/main.go @@ -4,9 +4,13 @@ import ( "context" "fmt" "log" + "math" "net/http" "os" "os/signal" + "runtime" + "strconv" + "strings" "syscall" "forgejo.riotpiao.com/rock/homelab-frontend/internal/auth" @@ -21,6 +25,10 @@ import ( ) func main() { + // Size the Go scheduler to the container's CPU quota before any server + // goroutine starts. + configureGOMAXPROCS() + ctx := context.Background() // Initialize OpenTelemetry tracing @@ -177,3 +185,53 @@ func main() { log.Printf("gateway shutdown complete") os.Exit(0) } + +// configureGOMAXPROCS sets GOMAXPROCS from the cgroup CPU quota so the runtime +// sizes its scheduler to the container limit rather than the host CPU count. +// The default is left alone when no quota is configured. +func configureGOMAXPROCS() { + quota, period := readCPUQuota() + if quota <= 0 || period <= 0 { + return + } + n := int(math.Ceil(float64(quota) / float64(period))) + if n < 1 { + n = 1 + } + runtime.GOMAXPROCS(n) + log.Printf("GOMAXPROCS set to %d (cgroup cpu quota %d/%d)", n, quota, period) +} + +// readCPUQuota returns the CPU quota and period from cgroup v2, falling back to +// cgroup v1. A quota of 0 means no limit. +func readCPUQuota() (int64, int64) { + // cgroup v2: " " or "max ". + if data, err := os.ReadFile("/sys/fs/cgroup/cpu.max"); err == nil { + fields := strings.Fields(string(data)) + if len(fields) >= 2 { + period, perr := strconv.ParseInt(fields[1], 10, 64) + if perr == nil && period > 0 { + if fields[0] == "max" { + return 0, period + } + if quota, qerr := strconv.ParseInt(fields[0], 10, 64); qerr == nil { + return quota, period + } + } + } + } + + // cgroup v1: cpu.cfs_quota_us is -1 when unlimited. + period := int64(100000) + if data, err := os.ReadFile("/sys/fs/cgroup/cpu/cpu.cfs_period_us"); err == nil { + if v, perr := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64); perr == nil && v > 0 { + period = v + } + } + if data, err := os.ReadFile("/sys/fs/cgroup/cpu/cpu.cfs_quota_us"); err == nil { + if v, qerr := strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64); qerr == nil { + return v, period + } + } + return 0, 0 +} diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index f78dfa2..fc01ca4 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -15,7 +15,7 @@ type Metrics struct { // Request latencies: request_duration_seconds (histogram) // Stored as cumulative buckets for Prometheus text format - requestDuration map[string]int64 // stores duration samples in milliseconds + requestDuration map[string]int64 // stores duration samples in milliseconds requestDurationBuckets map[string]map[float64]int64 // histogram buckets // Bytes counters: gateway_bytes{direction, route, upstream} @@ -41,22 +41,33 @@ type Metrics struct { warmupRequestsDelayed int64 prefixCacheHitTokens int64 queueFullRequests int64 + + // Admission control metrics (#64) + admissionParked int64 + admissionRejected map[string]int64 + engineStateStale int64 + + // Engine gauge readings mirrored from vLLM (#64) + vllmRequestsRunning int64 + vllmKVUsagePerc float64 + observerDegraded int } // NewMetrics creates a new Metrics instance. func NewMetrics() *Metrics { return &Metrics{ - requestTotal: make(map[string]int64), - requestDuration: make(map[string]int64), - requestDurationBuckets: make(map[string]map[float64]int64), - bytesIn: make(map[string]int64), - bytesOut: make(map[string]int64), - upstreamHealth: make(map[string]int), + requestTotal: make(map[string]int64), + requestDuration: make(map[string]int64), + requestDurationBuckets: make(map[string]map[float64]int64), + bytesIn: make(map[string]int64), + bytesOut: make(map[string]int64), + 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), + streamingByteCount: make(map[string]int64), + ttftMs: make(map[string][]int64), + itlMs: make(map[string][]int64), + tokenCount: make(map[string]int64), + admissionRejected: make(map[string]int64), } } @@ -133,7 +144,7 @@ func (m *Metrics) RecordStreamingResponse(route, upstream string, totalBytes int func (m *Metrics) recordDuration(key string, duration time.Duration) { m.requestDuration[key] += int64(duration.Milliseconds()) - + if _, ok := m.requestDurationBuckets[key]; !ok { m.requestDurationBuckets[key] = make(map[float64]int64) } @@ -288,6 +299,12 @@ func (m *Metrics) Reset() { m.warmupRequestsDelayed = 0 m.prefixCacheHitTokens = 0 m.queueFullRequests = 0 + m.admissionParked = 0 + m.admissionRejected = make(map[string]int64) + m.engineStateStale = 0 + m.vllmRequestsRunning = 0 + m.vllmKVUsagePerc = 0 + m.observerDegraded = 0 } // IncrQueueFull increments queue_full_requests_total. @@ -332,6 +349,98 @@ func (m *Metrics) GetPrefixCacheHitTokens() int64 { return m.prefixCacheHitTokens } +// IncrAdmissionParked increments admission_parked_total. +func (m *Metrics) IncrAdmissionParked() { + m.mu.Lock() + defer m.mu.Unlock() + m.admissionParked++ +} + +// GetAdmissionParked returns admission_parked_total. +func (m *Metrics) GetAdmissionParked() int64 { + m.mu.RLock() + defer m.mu.RUnlock() + return m.admissionParked +} + +// IncrAdmissionRejected increments admission_rejected_total{reason=reason}. +func (m *Metrics) IncrAdmissionRejected(reason string) { + m.mu.Lock() + defer m.mu.Unlock() + m.admissionRejected[reason]++ +} + +// GetAdmissionRejected returns a copy of admission_rejected_total by reason. +func (m *Metrics) GetAdmissionRejected() map[string]int64 { + m.mu.RLock() + defer m.mu.RUnlock() + out := make(map[string]int64, len(m.admissionRejected)) + for reason, count := range m.admissionRejected { + out[reason] = count + } + return out +} + +// IncrEngineStateStale increments engine_state_stale_total. +func (m *Metrics) IncrEngineStateStale() { + m.mu.Lock() + defer m.mu.Unlock() + m.engineStateStale++ +} + +// GetEngineStateStale returns engine_state_stale_total. +func (m *Metrics) GetEngineStateStale() int64 { + m.mu.RLock() + defer m.mu.RUnlock() + return m.engineStateStale +} + +// SetVLLMRequestsRunning sets vllm_requests_running from the gateway's own reading. +func (m *Metrics) SetVLLMRequestsRunning(n int64) { + m.mu.Lock() + defer m.mu.Unlock() + m.vllmRequestsRunning = n +} + +// GetVLLMRequestsRunning returns vllm_requests_running. +func (m *Metrics) GetVLLMRequestsRunning() int64 { + m.mu.RLock() + defer m.mu.RUnlock() + return m.vllmRequestsRunning +} + +// SetVLLMKVUsagePerc sets vllm_kv_usage_perc from the gateway's own reading. +func (m *Metrics) SetVLLMKVUsagePerc(v float64) { + m.mu.Lock() + defer m.mu.Unlock() + m.vllmKVUsagePerc = v +} + +// GetVLLMKVUsagePerc returns vllm_kv_usage_perc. +func (m *Metrics) GetVLLMKVUsagePerc() float64 { + m.mu.RLock() + defer m.mu.RUnlock() + return m.vllmKVUsagePerc +} + +// SetObserverDegraded sets observer_degraded (1 = startup self-check failed). +func (m *Metrics) SetObserverDegraded(degraded bool) { + m.mu.Lock() + defer m.mu.Unlock() + if degraded { + m.observerDegraded = 1 + } else { + m.observerDegraded = 0 + } +} + +// GetObserverDegraded returns observer_degraded. +func (m *Metrics) GetObserverDegraded() int { + m.mu.RLock() + defer m.mu.RUnlock() + return m.observerDegraded +} + // globalMetrics is the singleton Metrics instance for gateway-wide telemetry. var ( globalMetrics *Metrics diff --git a/internal/observability/prometheus.go b/internal/observability/prometheus.go index 46c7bc5..69d494c 100644 --- a/internal/observability/prometheus.go +++ b/internal/observability/prometheus.go @@ -48,6 +48,38 @@ func (p *PrometheusExporter) Export() string { lines = append(lines, "# TYPE vllm_prefix_cache_hit_tokens_total counter") lines = append(lines, fmt.Sprintf("vllm_prefix_cache_hit_tokens_total %d", p.metrics.GetPrefixCacheHitTokens())) + lines = append(lines, "# HELP admission_parked_total Requests parked waiting for engine admission") + lines = append(lines, "# TYPE admission_parked_total counter") + lines = append(lines, fmt.Sprintf("admission_parked_total %d", p.metrics.GetAdmissionParked())) + + lines = append(lines, "# HELP admission_rejected_total Requests rejected by engine admission control") + lines = append(lines, "# TYPE admission_rejected_total counter") + rejected := p.metrics.GetAdmissionRejected() + reasons := make([]string, 0, len(rejected)) + for reason := range rejected { + reasons = append(reasons, reason) + } + sort.Strings(reasons) + for _, reason := range reasons { + lines = append(lines, fmt.Sprintf(`admission_rejected_total{reason="%s"} %d`, reason, rejected[reason])) + } + + lines = append(lines, "# HELP engine_state_stale_total Times the engine metrics feed went stale") + lines = append(lines, "# TYPE engine_state_stale_total counter") + lines = append(lines, fmt.Sprintf("engine_state_stale_total %d", p.metrics.GetEngineStateStale())) + + lines = append(lines, "# HELP vllm_requests_running vLLM requests running as read by the gateway") + lines = append(lines, "# TYPE vllm_requests_running gauge") + lines = append(lines, fmt.Sprintf("vllm_requests_running %d", p.metrics.GetVLLMRequestsRunning())) + + lines = append(lines, "# HELP vllm_kv_usage_perc vLLM KV cache usage fraction as read by the gateway") + lines = append(lines, "# TYPE vllm_kv_usage_perc gauge") + lines = append(lines, fmt.Sprintf("vllm_kv_usage_perc %g", p.metrics.GetVLLMKVUsagePerc())) + + lines = append(lines, "# HELP observer_degraded Engine metrics observer startup self-check failed (1) or healthy (0)") + lines = append(lines, "# TYPE observer_degraded gauge") + lines = append(lines, fmt.Sprintf("observer_degraded %d", p.metrics.GetObserverDegraded())) + return strings.Join(lines, "\n") + "\n" } diff --git a/internal/proxy/engine_state.go b/internal/proxy/engine_state.go new file mode 100644 index 0000000..508b9b4 --- /dev/null +++ b/internal/proxy/engine_state.go @@ -0,0 +1,310 @@ +package proxy + +import ( + "bufio" + "context" + "fmt" + "io" + "math" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "forgejo.riotpiao.com/rock/homelab-frontend/internal/logging" + "forgejo.riotpiao.com/rock/homelab-frontend/internal/observability" +) + +// Metric names published by the vLLM engine. Admission is gated on these +// rather than on a per-pod semaphore: every gateway replica reads the same +// engine, so the limit is global by construction (#64). +const ( + metricRequestsRunning = "vllm:num_requests_running" + metricRequestsWaiting = "vllm:num_requests_waiting" + metricKVCacheUsage = "vllm:kv_cache_usage_perc" + + // enginePollInterval is how often each backend's /metrics is scraped. + enginePollInterval = 2 * time.Second + + // engineStaleThreshold is the number of consecutive failed polls after + // which the state is marked unhealthy and admission fails open. + engineStaleThreshold = 3 + + // engineMetricsTimeout bounds a single scrape. + engineMetricsTimeout = 2 * time.Second +) + +// EngineState is the gateway's view of one vLLM engine, mirrored from its +// own metrics endpoint. +type EngineState struct { + Running int64 + Waiting int64 + KVUsage float64 + UpdatedAt time.Time + Healthy bool +} + +// InFlight returns the number of requests the engine currently holds: +// running plus waiting. +func (s EngineState) InFlight() int64 { + return s.Running + s.Waiting +} + +// EngineObserver polls one vLLM backend's /metrics and publishes the gauges +// admission is gated on. Healthy is false until the first successful poll and +// after engineStaleThreshold consecutive failures. +type EngineObserver struct { + addr string + interval time.Duration + client *http.Client + + mu sync.Mutex + state EngineState + failures int + degraded bool + tick chan struct{} +} + +// NewEngineObserver creates an observer for the backend at addr. +func NewEngineObserver(addr string, interval time.Duration) *EngineObserver { + return &EngineObserver{ + addr: addr, + interval: interval, + // Own client: the proxy transports carry socket options and connect + // timeouts tuned for inference, not for a short metrics scrape. + client: &http.Client{Timeout: engineMetricsTimeout}, + tick: make(chan struct{}), + } +} + +// Start launches the polling goroutine. It returns immediately; the goroutine +// exits when ctx is cancelled. +func (o *EngineObserver) Start(ctx context.Context) { + go func() { + ticker := time.NewTicker(o.interval) + defer ticker.Stop() + // The first scrape waits one interval rather than firing at Start(). + // Scraping immediately makes the gateway contact every observed backend + // before any request needs one, which is wasted work and a surprising + // side effect for anything asserting on upstream traffic. State is + // unknown until the first poll, and unknown fails open, so the deferral + // costs at most one interval of ungated admission. The startup + // self-check (decision Q21) still runs, on that first tick. + startup := true + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + o.poll(ctx, startup) + startup = false + } + } + }() +} + +// Get returns the most recent engine state. +func (o *EngineObserver) Get() EngineState { + o.mu.Lock() + defer o.mu.Unlock() + return o.state +} + +// Tick returns a channel that is closed and replaced after every successful +// poll, waking requests parked on engine admission. +func (o *EngineObserver) Tick() <-chan struct{} { + o.mu.Lock() + defer o.mu.Unlock() + return o.tick +} + +// Degraded reports whether the startup self-check found the metrics feed +// unusable (parse failure or missing series). +func (o *EngineObserver) Degraded() bool { + o.mu.Lock() + defer o.mu.Unlock() + return o.degraded +} + +// poll scrapes /metrics once. startup marks the first poll, whose failure is a +// self-diagnosis signal rather than a transient blip. +func (o *EngineObserver) poll(ctx context.Context, startup bool) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+o.addr+"/metrics", nil) + if err != nil { + o.recordFailure() + logging.Warnf("engine metrics request build failed", map[string]string{ + "address": o.addr, + "err": err.Error(), + }) + return + } + resp, err := o.client.Do(req) + if err != nil { + o.recordFailure() + logging.Warnf("engine metrics scrape failed", map[string]string{ + "address": o.addr, + "err": err.Error(), + }) + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + o.recordFailure() + logging.Warnf("engine metrics scrape returned non-200", map[string]string{ + "address": o.addr, + "status": strconv.Itoa(resp.StatusCode), + }) + return + } + + running, waiting, kvUsage, err := ParseEngineMetrics(resp.Body) + if err != nil { + if startup { + // Startup self-check (decision Q21): metric-name drift must be + // loud, never a silent fall-back to fail-open admission. + o.mu.Lock() + o.degraded = true + o.mu.Unlock() + observability.Global().SetObserverDegraded(true) + logging.Errorf("engine metrics unusable at startup", err, map[string]string{ + "address": o.addr, + }) + } else { + logging.Warnf("engine metrics parse failed", map[string]string{ + "address": o.addr, + "err": err.Error(), + }) + } + o.recordFailure() + return + } + + o.recordSuccess(running, waiting, kvUsage) +} + +// recordSuccess stores a fresh reading and wakes parked waiters. +func (o *EngineObserver) recordSuccess(running, waiting int64, kvUsage float64) { + o.mu.Lock() + o.state = EngineState{ + Running: running, + Waiting: waiting, + KVUsage: kvUsage, + UpdatedAt: time.Now(), + Healthy: true, + } + o.failures = 0 + close(o.tick) + o.tick = make(chan struct{}) + o.mu.Unlock() + + m := observability.Global() + m.SetVLLMRequestsRunning(running) + m.SetVLLMKVUsagePerc(kvUsage) +} + +// recordFailure counts a failed poll and, on the third consecutive failure, +// marks the state unhealthy and counts the staleness transition once — not +// once per failed poll. Last known values are kept. +func (o *EngineObserver) recordFailure() { + o.mu.Lock() + o.failures++ + stale := o.failures == engineStaleThreshold + if o.failures >= engineStaleThreshold { + o.state.Healthy = false + } + failures := o.failures + o.mu.Unlock() + + if stale { + observability.Global().IncrEngineStateStale() + logging.Warnf("engine metrics stale; admission is failing open", map[string]string{ + "address": o.addr, + "failures": strconv.Itoa(failures), + }) + } +} + +// ParseEngineMetrics extracts the running, waiting and KV-cache-usage gauges +// from a vLLM Prometheus exposition body. Comment lines, unrelated series and +// non-finite samples are ignored; values are summed across repeated label sets. +// It returns an error naming any of the three series that is absent. +func ParseEngineMetrics(r io.Reader) (running int64, waiting int64, kvUsage float64, err error) { + scanner := bufio.NewScanner(r) + // The real exposition body is large; give the scanner room per line. + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + var hasRunning, hasWaiting, hasKV bool + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + name, value, ok := splitMetricSample(line) + if !ok { + continue + } + switch name { + case metricRequestsRunning, metricRequestsWaiting, metricKVCacheUsage: + default: + continue + } + f, err := strconv.ParseFloat(value, 64) + if err != nil || math.IsNaN(f) || math.IsInf(f, 0) { + continue + } + switch name { + case metricRequestsRunning: + running += int64(f) + hasRunning = true + case metricRequestsWaiting: + waiting += int64(f) + hasWaiting = true + case metricKVCacheUsage: + kvUsage += f + hasKV = true + } + } + if err := scanner.Err(); err != nil { + return 0, 0, 0, fmt.Errorf("read engine metrics: %w", err) + } + + var missing []string + if !hasRunning { + missing = append(missing, metricRequestsRunning) + } + if !hasWaiting { + missing = append(missing, metricRequestsWaiting) + } + if !hasKV { + missing = append(missing, metricKVCacheUsage) + } + if len(missing) > 0 { + return 0, 0, 0, fmt.Errorf("engine metrics missing series: %s", strings.Join(missing, ", ")) + } + return running, waiting, kvUsage, nil +} + +// splitMetricSample splits one Prometheus exposition line into its metric name +// and its first sample value. Lines without a value are reported as not ok. +func splitMetricSample(line string) (name, value string, ok bool) { + end := strings.IndexAny(line, "{ \t") + if end <= 0 { + return "", "", false + } + name = line[:end] + rest := line[end:] + if rest[0] == '{' { + closeBrace := strings.LastIndex(rest, "}") + if closeBrace < 0 { + return "", "", false + } + rest = rest[closeBrace+1:] + } + fields := strings.Fields(rest) + if len(fields) == 0 { + return "", "", false + } + return name, fields[0], true +} diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index a2de895..7c2e05e 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -2,6 +2,7 @@ package proxy import ( + "bytes" "context" "encoding/json" "errors" @@ -47,6 +48,10 @@ type Handler struct { schedulers map[string]*Scheduler // CAG: delay applied to warmup requests before queue entry (#45) warmupDelay time.Duration + // engineObservers polls each model backend's /metrics for admission (#64) + engineObservers map[string]*EngineObserver + // maxBodyBytes caps synthesized model-route request bodies (MAX_BODY_BYTES) + maxBodyBytes int64 // transportsMu guards h.transports for goroutine-safe reads/writes. transportsMu sync.RWMutex } @@ -122,18 +127,80 @@ func New(cfg *config.Config) *Handler { maxSlots = n } } - maxQueue := 20 + maxQueue := 8 if v := os.Getenv("VLLM_MAX_QUEUE_DEPTH"); v != "" { if n, err := strconv.Atoi(v); err == nil && n > 0 { maxQueue = n } } + + // Issue #64: engine-wide admission thresholds. MaxInFlight defaults to the + // same value as VLLM_MAX_NUM_SEQS; the observers below read the engine's + // own gauges, which every replica shares. + admissionMaxInFlight := maxSlots + if v := os.Getenv("ADMISSION_MAX_INFLIGHT"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + admissionMaxInFlight = n + } + } + admissionMaxKVUsage := 0.90 + if v := os.Getenv("ADMISSION_MAX_KV_USAGE"); v != "" { + if f, err := strconv.ParseFloat(v, 64); err == nil && f > 0 { + admissionMaxKVUsage = f + } + } + maxParkSeconds := 120 + if v := os.Getenv("ADMISSION_MAX_PARK_SECONDS"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + maxParkSeconds = n + } + } + h.maxBodyBytes = 1048576 + if v := os.Getenv("MAX_BODY_BYTES"); v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 { + h.maxBodyBytes = n + } + } + + limits := AdmissionLimits{ + MaxInFlight: admissionMaxInFlight, + MaxKVUsage: admissionMaxKVUsage, + MaxPark: time.Duration(maxParkSeconds) * time.Second, + } + + // Issue #64: only backends that actually publish the vLLM scheduler gauges + // may be observed. `reasoning` is the only vLLM engine here; ornith and + // qwen-cpu are Ollama and embeddings/rerank are TEI, none of which export + // vllm:* series. Observing them would trip the startup self-check and hold + // observer_degraded at 1 forever. Unset (the default) observes nothing and + // leaves admission exactly as it was before this change. + observeModels := make(map[string]bool) + for _, name := range strings.Split(os.Getenv("ADMISSION_OBSERVE_MODELS"), ",") { + if name = strings.TrimSpace(name); name != "" { + observeModels[name] = true + } + } + + h.engineObservers = make(map[string]*EngineObserver) h.schedulers = make(map[string]*Scheduler) for modelName, model := range cfg.Models { addrs := model.AllAddresses() - if len(addrs) > 0 { - h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue) + if len(addrs) == 0 { + continue } + if observeModels[modelName] { + for _, addr := range addrs { + if _, ok := h.engineObservers[addr]; !ok { + h.engineObservers[addr] = NewEngineObserver(addr, enginePollInterval) + } + } + } + h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue, h.engineObservers, limits) + } + // The handler has no request-path lifecycle to hang cancellation on, so the + // observers poll for the lifetime of the process. + for _, o := range h.engineObservers { + o.Start(context.Background()) } // Pre-populate transports for all scheduler backends at startup (single-threaded). // getOrCreateTransport is not goroutine-safe; pre-populating avoids concurrent writes. @@ -151,7 +218,7 @@ func New(cfg *config.Config) *Handler { // Create a transport per unique upstream address for connection reuse transport := h.getOrCreateTransport(route.Upstream.Address, &route.Upstream) - upstreamURL, _ := url.Parse("http://" + route.Upstream.Address) + upstreamURL := &url.URL{Scheme: "http", Host: route.Upstream.Address} r := &Route{ Name: name, @@ -395,27 +462,104 @@ func (h *Handler) applyWarmup(ctx context.Context, r *http.Request) bool { // acquireScheduledBackend selects a backend from sched, applies warmup // deprioritisation, acquires a concurrency slot, and patches route when the // selected backend differs from the config default. It returns the backend -// to Release (caller must defer), the (possibly patched) route, and ok=true. -// If the slot cannot be acquired it writes a 503 and returns (nil, nil, false). -func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool) { +// to Release (caller must defer), the (possibly patched) route, whether SSE +// headers were already written while parking, and ok=true. +// If the slot cannot be acquired it writes a 503 and returns ok=false. +func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool, bool) { backend := sched.Select(cacheKey) if !h.applyWarmup(ctx, r) { - return nil, nil, false + return nil, nil, false, false } if backend == nil { - return nil, route, true + return nil, route, false, true } - if err := backend.Acquire(ctx); err != nil { - if errors.Is(err, ErrQueueFull) { + + // Decision Q22: a request that actually parks on the engine gate and wants + // an SSE stream gets its headers pre-written plus a 15 s keep-alive, so no + // intermediary times the connection out before the engine admits it. + var ( + prewritten bool + hbStop chan struct{} + hbDone chan struct{} + ) + onPark := func() { + if !requestWantsStream(r) { + return + } + flusher, ok := w.(http.Flusher) + if !ok { + return + } + prewritten = true + hbStop = make(chan struct{}) + hbDone = make(chan struct{}) + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, ": keep-alive\n\n") + flusher.Flush() + go func() { + defer close(hbDone) + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + for { + select { + case <-hbStop: + return + case <-ctx.Done(): + return + case <-ticker.C: + fmt.Fprint(w, ": keep-alive\n\n") + flusher.Flush() + } + } + }() + } + + err := backend.acquire(ctx, onPark) + if prewritten { + close(hbStop) + <-hbDone + } + if err != nil { + // Once the SSE status line is spent, report the rejection as an event + // instead of a second WriteHeader. + reject := func(body string) { + if prewritten { + fmt.Fprintf(w, "event: error\ndata: {\"error\":%q}\n\n", body) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + return + } + w.WriteHeader(http.StatusServiceUnavailable) + fmt.Fprintf(w, "%s", body) + } + if errors.Is(err, ErrAdmissionTimeout) { + observability.Global().IncrAdmissionRejected("park_timeout") + logging.Warnf("admission timeout", map[string]string{ + "backend": backend.Address, + "path": r.URL.Path, + "model": route.ModelName, + }) + if !prewritten { + w.Header().Set("Retry-After", "5") + } + reject("admission timeout") + } else if errors.Is(err, ErrQueueFull) { observability.Global().IncrQueueFull() + observability.Global().IncrAdmissionRejected("queue_full") logging.Warnf("backend queue full", map[string]string{ "backend": backend.Address, "path": r.URL.Path, "model": route.ModelName, }) - w.Header().Set("Retry-After", "5") - w.WriteHeader(http.StatusServiceUnavailable) - fmt.Fprintf(w, "queue full") + if !prewritten { + w.Header().Set("Retry-After", "5") + } + reject("queue full") } else { logging.Warnf("backend acquire failed", map[string]string{ "backend": backend.Address, @@ -423,15 +567,14 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr "model": route.ModelName, "err": err.Error(), }) - w.WriteHeader(http.StatusServiceUnavailable) - fmt.Fprintf(w, "upstream busy") + reject("upstream busy") } - return nil, nil, false + return nil, nil, prewritten, false } if backend.Address != route.Upstream.Address { patchedCfg := *route.Upstream patchedCfg.Address = backend.Address - patchedURL, _ := url.Parse("http://" + backend.Address) + patchedURL := &url.URL{Scheme: "http", Host: backend.Address} route = &Route{ Name: route.Name, ModelName: route.ModelName, @@ -442,7 +585,27 @@ func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWr }, } } - return backend, route, true + return backend, route, prewritten, true +} + +// requestWantsStream reports whether the JSON request body asks for an SSE +// stream ("stream": true). The body is restored for the upstream. +func requestWantsStream(r *http.Request) bool { + if r.Body == nil { + return false + } + body, err := io.ReadAll(r.Body) + if err != nil { + return false + } + r.Body = io.NopCloser(bytes.NewReader(body)) + var probe struct { + Stream bool `json:"stream"` + } + if err := json.Unmarshal(body, &probe); err != nil { + return false + } + return probe.Stream } // handleAuthEndpoint routes /auth/* requests and returns true if handled. @@ -605,9 +768,10 @@ func (h *Handler) enforceBodyLimit(w http.ResponseWriter, r *http.Request, route // Returns (nil, nil, false) when the backend slot cannot be acquired (response written). func (h *Handler) setupCAGProxy(ctx context.Context, w http.ResponseWriter, r *http.Request, route *Route, sched *Scheduler, cacheKey string) (*httputil.ReverseProxy, *Backend, bool) { var selectedBackend *Backend + var headersPrewritten bool if sched != nil { var ok bool - selectedBackend, route, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey) + selectedBackend, route, headersPrewritten, ok = h.acquireScheduledBackend(ctx, w, r, sched, route, cacheKey) if !ok { return nil, nil, false } @@ -628,6 +792,20 @@ func (h *Handler) setupCAGProxy(ctx context.Context, w http.ResponseWriter, r *h s.MarkFailed(capturedBackend.Address) } } + if headersPrewritten { + // SSE headers and a 200 are already on the wire: a 502 here would be + // an ignored WriteHeader and would tear the event stream. Report the + // failure as an event instead. + fmt.Fprint(w, "event: error\ndata: {\"error\":\"upstream error\"}\n\n") + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + logging.Errorf("upstream error", err, map[string]string{ + "upstream": capturedRoute.Upstream.Address, + "path": r.URL.Path, + }) + return + } w.WriteHeader(http.StatusBadGateway) fmt.Fprintf(w, "upstream error") logging.Errorf("upstream error", err, map[string]string{ diff --git a/internal/proxy/router.go b/internal/proxy/router.go index 544ff3a..493c7d4 100644 --- a/internal/proxy/router.go +++ b/internal/proxy/router.go @@ -69,9 +69,10 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { // Restore the body so it can be read again by the upstream r.Body = io.NopCloser(bytes.NewReader(bodyBytes)) - // Parse the JSON to find the model field - var payload map[string]interface{} - if err := json.Unmarshal(bodyBytes, &payload); err != nil { + // Extract only the top-level "model" value with a token walk: the common + // path must not materialise the whole body (messages and all) as a map. + modelRaw, hasModel, err := extractTopLevelModel(bodyBytes) + if err != nil { return nil, &modelValidationError{ Kind: "invalid_json", Message: "request body is not valid JSON", @@ -79,7 +80,6 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { } // Extract the model name - modelVal, hasModel := payload["model"] if !hasModel { return nil, &modelValidationError{ Kind: "missing_model", @@ -88,7 +88,7 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { } // Handle null model - if modelVal == nil { + if bytes.Equal(bytes.TrimSpace(modelRaw), []byte("null")) { return nil, &modelValidationError{ Kind: "null_model", Message: "'model' field is null", @@ -96,8 +96,8 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { } // Extract as string - modelName, ok := modelVal.(string) - if !ok { + var modelName string + if err := json.Unmarshal(modelRaw, &modelName); err != nil { return nil, &modelValidationError{ Kind: "missing_model", Message: "'model' field must be a string", @@ -122,10 +122,26 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { } } - // If upstream expects a different model name, rewrite the body + // If upstream expects a different model name, rewrite the body. This path is + // uncommon, so it may decode the body as raw fields; the common path above + // does not build a full map. if modelUpstream.UpstreamModel != "" && modelUpstream.UpstreamModel != modelName { - payload["model"] = modelUpstream.UpstreamModel - newBody, err := json.Marshal(payload) + var fields map[string]json.RawMessage + if err := json.Unmarshal(bodyBytes, &fields); err != nil { + return nil, &modelValidationError{ + Kind: "invalid_request", + Message: fmt.Sprintf("failed to rewrite model name: %v", err), + } + } + newModel, err := json.Marshal(modelUpstream.UpstreamModel) + if err != nil { + return nil, &modelValidationError{ + Kind: "invalid_request", + Message: fmt.Sprintf("failed to rewrite model name: %v", err), + } + } + fields["model"] = newModel + newBody, err := json.Marshal(fields) if err != nil { return nil, &modelValidationError{ Kind: "invalid_request", @@ -151,11 +167,11 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { ConnectTimeout: h.defaultConnectTimeout, ReadTimeout: h.defaultReadTimeout, WriteTimeout: h.defaultWriteTimeout, - MaxBodySize: h.defaultMaxBodySize, + MaxBodySize: h.maxBodyBytes, AuthRequired: false, } - targetURL, _ := url.Parse("http://" + modelUpstream.Address) + targetURL := &url.URL{Scheme: "http", Host: modelUpstream.Address} route := &Route{ Name: "v1-chat-" + modelName, @@ -169,3 +185,87 @@ func (h *Handler) routeByModel(r *http.Request, path string) (*Route, error) { return route, nil } + +// extractTopLevelModel walks a JSON object body with a streaming decoder and +// returns the raw value of the top-level "model" field. It never builds the +// full document as a map, so a large "messages" array is skipped by token. +// hasModel is false when the key is absent or the document is a bare null. +func extractTopLevelModel(body []byte) (model json.RawMessage, hasModel bool, err error) { + dec := json.NewDecoder(bytes.NewReader(body)) + tok, err := dec.Token() + if err != nil { + return nil, false, err + } + if tok == nil { + // A bare `null` body decoded to a nil map before, i.e. no model field. + if _, err := dec.Token(); err != io.EOF { + if err == nil { + return nil, false, fmt.Errorf("unexpected data after JSON null") + } + return nil, false, err + } + return nil, false, nil + } + if delim, ok := tok.(json.Delim); !ok || delim != '{' { + return nil, false, fmt.Errorf("request body is not a JSON object") + } + for dec.More() { + keyTok, err := dec.Token() + if err != nil { + return nil, false, err + } + key, _ := keyTok.(string) + if key == "model" { + if err := dec.Decode(&model); err != nil { + return nil, false, err + } + hasModel = true + continue + } + if err := skipJSONValue(dec); err != nil { + return nil, false, err + } + } + // Consume the closing brace and reject trailing data, preserving the + // invalid_json answer the old whole-body decode produced. + if _, err := dec.Token(); err != nil { + return nil, false, err + } + if _, err := dec.Token(); err != io.EOF { + if err == nil { + return nil, false, fmt.Errorf("unexpected data after JSON object") + } + return nil, false, err + } + return model, hasModel, nil +} + +// skipJSONValue advances dec past the next JSON value without materialising it. +func skipJSONValue(dec *json.Decoder) error { + tok, err := dec.Token() + if err != nil { + return err + } + delim, ok := tok.(json.Delim) + if !ok { + return nil // scalar + } + if delim != '{' && delim != '[' { + return fmt.Errorf("unexpected delimiter %v", delim) + } + depth := 1 + for depth > 0 { + tok, err := dec.Token() + if err != nil { + return err + } + if d, ok := tok.(json.Delim); ok { + if d == '{' || d == '[' { + depth++ + } else { + depth-- + } + } + } + return nil +} diff --git a/internal/proxy/scheduler.go b/internal/proxy/scheduler.go index 51f495f..bc0e71e 100644 --- a/internal/proxy/scheduler.go +++ b/internal/proxy/scheduler.go @@ -8,20 +8,41 @@ import ( "sync" "sync/atomic" "time" + + "forgejo.riotpiao.com/rock/homelab-frontend/internal/observability" ) // ErrQueueFull is returned when the in-memory queue is at capacity. var ErrQueueFull = errors.New("backend queue full") +// ErrAdmissionTimeout is returned when a request parked on engine admission +// exceeds the configured maximum park duration. +var ErrAdmissionTimeout = errors.New("admission timeout") + +// AdmissionLimits are the engine-wide admission thresholds applied to every +// backend of a scheduler (#64). +type AdmissionLimits struct { + MaxInFlight int + MaxKVUsage float64 + MaxPark time.Duration +} + // Backend represents a single vLLM upstream with a bounded in-memory queue. // Two channels model the queue: // // queue — waiting requests (bounded by VLLM_MAX_QUEUE_DEPTH) // slots — concurrent in-service requests (bounded by VLLM_MAX_NUM_SEQS) +// +// The engine gate is applied before the queue: admission reads the engine's +// own published gauges via observer, which is global across replicas. type Backend struct { - Address string - queue chan struct{} - slots chan struct{} + Address string + queue chan struct{} + slots chan struct{} + observer *EngineObserver + maxInFlight int + maxKVUsage float64 + maxPark time.Duration } // newBackend creates a Backend with given queue depth and concurrency limit. @@ -40,7 +61,46 @@ func newBackend(addr string, maxSlots, maxQueue int) *Backend { // Acquire enters the in-memory queue then waits for a vLLM concurrency slot. // Returns ErrQueueFull immediately if the queue channel is at capacity (no goroutine queued). // Returns ctx.Err() if the context is cancelled while waiting for a slot. +// Returns ErrAdmissionTimeout if the engine gate parks the request past maxPark. func (b *Backend) Acquire(ctx context.Context) error { + return b.acquire(ctx, nil) +} + +// acquire is Acquire with an optional onPark callback. onPark is invoked once, +// on the calling goroutine, the first time the request has to wait for engine +// admission; the parked streaming path uses it to start an SSE heartbeat. +func (b *Backend) acquire(ctx context.Context, onPark func()) error { + // Engine gate (#64): admit on the engine's own gauges, not on this pod's + // counters. Failing OPEN on !Healthy is deliberate — a metrics outage must + // not become an inference outage; the three-replica gate is only a fairness + // improvement on top of the engine's own queueing. + if b.observer != nil { + deadline := time.Now().Add(b.maxPark) + timer := time.NewTimer(time.Until(deadline)) + defer timer.Stop() + parked := false + for { + st := b.observer.Get() + if !st.Healthy || (st.InFlight() < int64(b.maxInFlight) && st.KVUsage < b.maxKVUsage) { + break + } + if !parked { + observability.Global().IncrAdmissionParked() + parked = true + if onPark != nil { + onPark() + } + } + select { + case <-b.observer.Tick(): + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return ErrAdmissionTimeout + } + } + } + // Non-blocking queue entry: reject immediately when queue is full. select { case b.queue <- struct{}{}: @@ -76,10 +136,15 @@ type Scheduler struct { // NewScheduler creates a Scheduler for the given addresses. // maxSlots = VLLM_MAX_NUM_SEQS, maxQueue = VLLM_MAX_QUEUE_DEPTH. -func NewScheduler(addresses []string, maxSlots, maxQueue int) *Scheduler { +// observers holds one engine observer per backend address (may be nil). +func NewScheduler(addresses []string, maxSlots, maxQueue int, observers map[string]*EngineObserver, limits AdmissionLimits) *Scheduler { backends := make([]*Backend, len(addresses)) for i, addr := range addresses { backends[i] = newBackend(addr, maxSlots, maxQueue) + backends[i].observer = observers[addr] + backends[i].maxInFlight = limits.MaxInFlight + backends[i].maxKVUsage = limits.MaxKVUsage + backends[i].maxPark = limits.MaxPark } return &Scheduler{ backends: backends, diff --git a/internal/proxy/scheduler_test.go b/internal/proxy/scheduler_test.go index e0631b3..e6bc250 100644 --- a/internal/proxy/scheduler_test.go +++ b/internal/proxy/scheduler_test.go @@ -67,7 +67,7 @@ func TestBackendAcquireReturnsErrQueueFullWhenQueueFull(t *testing.T) { // always selects the same backend (rendezvous / highest-random-weight hashing). func TestSchedulerSelectRendezvousHashDeterministic(t *testing.T) { addrs := []string{"a:80", "b:80", "c:80"} - s := NewScheduler(addrs, 10, 10) + s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{}) first := s.Select("same-cache-key") if first == nil { @@ -85,7 +85,7 @@ func TestSchedulerSelectRendezvousHashDeterministic(t *testing.T) { // requests evenly across all backends when no cache key is given. func TestSchedulerSelectNoKeyRoundRobin(t *testing.T) { addrs := []string{"a:80", "b:80", "c:80"} - s := NewScheduler(addrs, 10, 10) + s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{}) counts := make(map[string]int) for i := 0; i < 9; i++ { @@ -160,7 +160,7 @@ func TestBackendAcquireCancelledContext(t *testing.T) { // selected within the fail window (30 s default). func TestSchedulerMarkFailedExcludesBackend(t *testing.T) { addrs := []string{"a:80", "b:80"} - s := NewScheduler(addrs, 10, 10) + s := NewScheduler(addrs, 10, 10, nil, AdmissionLimits{}) s.MarkFailed("a:80") diff --git a/k8s/configmap.yaml b/k8s/configmap.yaml deleted file mode 100644 index 635630e..0000000 --- a/k8s/configmap.yaml +++ /dev/null @@ -1,24 +0,0 @@ -apiVersion: v1 -kind: ConfigMap -metadata: - name: api-gateway-config - namespace: api - labels: - app: api-gateway -data: - config.yaml: ENC[AES256_GCM,data:AQHxAzklWEksAw1faR8Wel1DzrGaR6hZVTh67JZ2muftm7Rc8CEpG7NAQyEn/2hm38LytuEKHDmZbhguEx2KUoBoSG+IYxqfl4uaLw3CllDQeR1VFkYqrtXZmb0K1rHbKYAzaSS74l5ZIEboHse7E39lINAvQc4ST4Fi/4tg4RoBsqlJoUIkI6/EkX0ls5NsXrq/6G5fCzIQv3WLx2pW7z3wwZpoJ5vgSvE9w7nI92aq8BNShFHj7NaELH25F8VvRCvsqSJfSHhJXLVlMaUDEA9Hkyl494sLg/G4jubfCwhvBHXobnPtc1I8Qm/UTzxg8UYWTr6XwmN7Gdxj+BtzOTLSY+wVyOzmdkGOS2dLOYXqOaRH2XDUkJNsemI5Diow8h8hlmpvmhWJ9AsG/jXjnqfuhPBavjmTWRFMT8B62ph+T3+Sx0g+g/JP+Za4Bncy6wZ1iNa0QvVTh+X0s6uJO/u5GOcpzipT3ctuthluwdnMHsWd7MuY5qH0yppUtRfIIa7Tcqmwc/fcPhWy6W+9b1RRvecAc+WTF58ziAq4EysacM/fAaecD44OrH5pM79qpvF6nn8N/oeyHdpHlojiZatrzfmDA2hCx2JGgdrJGzQ7oKlV6acZHGRPEzh+ziVhpZzbKF35QSOIilvSsGAiQGCOobTPPXBAUmMXozWEsK/is2mBSf+Av/McSPH5Xlkg9fC8HijVZmB0M2DCpTiOwr8mATYT81emqH9kCGePwbjNQxAc08n2fugVaW1kITm84tAjCQLv67HvtZMITtabg6KVbK6weW3bULyoqGddN31IaVsBxNuAGJNeQFAtaGOHTlIQu7hH79USPW7DIdBWQMWKogyghKTgv9hZDxQz8SYMCk4rknWpPZ+CJRQaXVKH9i5IubrecoIS4ZD5iHTrjDddL5diYoMaLmaRhxUUMqxTxbXQZHA8lYbhVqnqfoy9fPb9yV9FfNjpTRAtTav03iD2JH9hYOBFpKfv1CNwpRYUjduYsp8ifQffVVHo/Ru2SHRuBxpgGVHf6BqB3xlUsptMrTVJMGYMaKNFfGLm1wuJGsNqx3ZX2/WX93N++j7A25dfV+Jx7k/g6cR7miicPAVuC4y9dYYStlS6pXWg2aQcwFPSCmOaoPP4Ol1yc+5DhO8+j7KDjfzzak04DjDpWoeYECTwXORL3TNNcCi7yWp6h+dUwjrcDMnBBpPnEeSWqyk6usdijUNZiZgupi12szHDKvebvXFKdsoxjywH0nhRnA8+nBQPYxqivSxnaqkSVTgPMLCG2rhxAt9rD2KOBU4/zBMnivaKYOhhqRKivIwj/V8Ybfyok5W/mGfkXvXU3HPYNvCLLuH8pmaG8Y3iQiCoUB2vLOOZVFMk3n3G97GF036lpsTFv8rICHrBoNhFsRvfz+1u4OVEF00nZuaNwV1przu7ePEQEt/wLnr7RzuTU4csMHlRTjn0bPH3tnRoqhpxBXj3cZDdu7ukNerjAb5ytt8OLOtgTPxPe45rxJKAIet5lFBed1Te43kR4zXxf4fdiUXsbsYVfcl/6g/Mi79sD7xGfCGcppXEYwG0umbSTwUhi2nbGQIR2KDMOnGkCveMBOuY8dUI8JpOxPSibHxKzikcHn58+K5BrV3SZE5YY2nBLu9i1RiKRPHl6gnJMUUgb6zVFQhlP5dzQCRNsjvZ7gWKzJKehZUbmuIQWFuA6QM2OWpS6v93XcC8nFENIkkSvIMZiP/shzzv3L1fMR7dTsRqBLhPsxcPRItSBaVfYfrfDVOeS4eGroqkLoIHC39bKxY1pC4l5fxa536H1DKkNiS2bHpDCAcH04uVeOUKmPGLDEqcSiYBpvYXCIDOE/4BFE0wM0BXDNzkYcxvovu02z5Nr3hKbCot0gmeQGOzXWdlOTPAGdW20PU676Fs/G8rqNJhKnkrJEsZ1VsOJuDS+4GAVybAG++NPHrtqmNeN5IthC7Et1qR9dPQJ6idPLxLy0LbhbLJKAAAcpi4ZHrqbsSvu4GfeXp/5YdBKqY0RF9RjL0ptJMRRm+/kg2BMrxBGjNhZMDLgW/9VapITCH1rFBYOz1Cu53+90Asc3Uo66XdqSGQw1yBumIwu66SJArfwqv5Pz9RyabbpDDbXYmDP0AtL+rtVfGnCNCSU3h9og8H18pEBFdmJY7xkkoUu+ncAoL4qp0irdnb1YUGjMUI8f5TuBgLVg0gCZeDIY8Qf3Nty+jdpBmvhYshTdlhQe42n9NmGUtBai8m1tTOMzEo+RDUiIY3B4tczyNm5G2pVXT7dcBtgPrWoMV46+ieUQFueyuVqOL4Xyae3OhothatmGv012zZu0xPWreEP8OPhINy++uTtm4fkVM/ieuglTWfu5DmRV6Xeal/LgVCPgVpQ1WxB613J/7h1zFZg8OLP4X4sQ1dvigUtcWkMzuRQZa1MRkyKp5MOajUDTcNv6BshkIS6yyIiuzu3XJS7MZGh5pFzvcMhFaJqDMsKQ0Celd5MX3/2vutK2EoNIL+6yFOsLFSQ1VthfeLwkWXYS5d3u1uSjsKngLIoZYLuW7WZGaaJVKAcoPzLW7HU8IW3duM4iflkG3KvLLq/aVSE+DN72ga4TnDNtjOlyDJtP/9lvG1IOFLWbhWslrZOYMOqiay3rvd3QT1e6qMPXAlbGWomxMyY7UQqEqbZDCJvL3yZqtI6dhzkyL/CSzSXiU/gaO3qFGoQHNtLwwXqBBsgtUcIZwAW/uzPGgvacLyin6p1ezW4H40UzkhzHBwiMKhRumMPyGCw8uo65nun0KScWqsu0hjDMGHYRBCoHLavOJ+lgXquvcg9ct6w16OXVBH8HiW1cg5UBIywP7oSTzePslxcVHY2dlOBUK5/Exdvg8ss8C6jUdkuSZOlbHBH/RWlYR89VnfoezAKuVgcn2hsGyPmLP+e6IiEgGGzc5z+KPamGAoyqPj77Q8vzBzAXd8PEf80NRz9j9K4EDE4MxadcoLFFW9HkrCYUBWC3JBsMUmuEUIoZIQTAJngvNS3kbx5+GIte+wykugztqAQm37UEsVKxlC1Jns6Xr638oKvmd4uGYmMbFq9ARzhx+NyNMpZWxJubln+LskFGRocQ/HNWr6Fjkb96AGV3gCMMJ3R6XteRJvssqI5q2k9dmfRaoiw/T+bBWlWNfqR7FgVrFCW+KGleBBmZQi8YAfn6pqRIe7tikS+VUaKHOrG1AEnoQMuOQaTy5sWEpm1NG2Ugqe0I3dTZo4HJHdtVwJzYNNMWToGjbGtThpsdvn/kQLMixj0F219ILDBt6YtP7VogMZ9dSUVHIm6MJnnyE2w9OtR5xbwVhYAIUwSEyvPUlNoN9+sQ6Hsevyi5hGqULfh9iXsGOIktHY4ALXpVdsI1O5fIvxWrPkYlHJTBwgfb4RGL65slvHUKCQZAX8ec5qAcJoxBdyZn8V6gVjb6OPNxVhRETDQTYscwfyzUCWNuLtxCnskcvOLm/iFm/2AqKclocf1/YEIpDm+kRElG1CsUa/CgI2xCj6UKDZVVaTwoRCN/PZ1goHkb35n1QhmmZ4kbi9MQHVdFOQnM3AN7c2tdVJ/2vYWfISSD1b+ASG5HP2cZG4HUS5ae4SJd/TED2BMC22lxkHUVeLlztCShyTHak40Au8brJSykJNiTloGY7XHRFIXBdzfBluLpzzcyIGL7pxjJp/tc0DdztxdNhkX+7pKAvOb0DM5HhuyeH3IFxYx8+mJq9bSrD3bjlZXSrPLMgzk/ht102QhrpmSUIoYdJLFCywBmGI9nJd+5CyLps0UlF6OV8t3ndm/NMRF7kcLyhxmYNotddn9tuc2hBM/3zQHs4vwJobGuvr/9gKfgTOm6a3Gvlb0O25wOxJdcMDWxwvIAOBl2LLT+aIqP3fDaKhTeti8p57gWpAOm1OxG8/S5ONsNCjMucy3TT+Jft0JMlZIHLTYcz/GmZzA3GrRc3kbngIyBUjrUkQtwdGGxOgvvv2QCSD7/kcK9lz84UOl4kQVTwVgjoJZk++FUyfVrsCH7BBLct4EGLFMvmvXxVcM+Glxe5eUgepNy5lmF8f7Mjwf9sB/d46oC5qP+N7lUjs2JO0JNcISbchwyYssqS7LuTDaLAsEO0Z8RszhQXu+EW7o6FP/MGytzavRE0hYbDKHMFsqZgV0Z/tiKnhKbPkdIHALFYyh5fUmUntT/K2xYbq2dUXjKBKDUeNN8o9ZhMLjBHollVIjo7LR3/sZNIS0rgHqLGW/5mCr3jLUzfDrciyVgRWHic52F+DalsA0rIS1+b95jhuH1bI+DBihZY6EJtKdX52kjtTL+yvwbkHzxIYBZKhV7VuTTmal1HWgY/blx6nHrCPoz11kCJELxC8IVznE4ZyPTX1sLWgOr1xEG9wZQ4Wbxq3CiBz4uFG4E4guaSJIpKWc0/Xsj11viMmh415AEdW4mWHn7GKhYVj+c9MrKvuJ+ZhGuIt7kSLRJGwCmSbT02CzOpTUFB4H0nUYSDE0tTVLAzJ123sz2Aa3xH8MrCevIGd5uTGsd0c7Gig2ixgzB5VYtFINc1d1Ob4JDWnW5x0Sc2JgXcat0RHPhP18b2INv9q2SV0DEmI6+h+OfZ1WEthim/evofQwD//px9Nkf2lj8TbAGVyCWNck6BlvrFWtP73sPmU4lnoNr6LesrLWMQXY5JiFa8C4M2qqUeHHI2j0MB08zG92TU29O+ovElhrt0d9SpSXTIPT/jWaJNYy6qTUlUcwp8/Pipo1eIRQgjhaIWpzXM1oCd19q8OVlC/QHUbDLTcWn43xNEkiBCL/m+7yLwF7w4rnDpOzWBnZkHklYHAWhs4qRoqgukMWITbXowj7ATXu9SgEqke2KncUVXqWQbTHHIcPwOp/EEC0g8wcxHJZHCx2r8oEi7u6FfJs+mt8yq4gY2IGIr6wMMJn4UU1rloQVt4fhZRYVBEddn37DkWjrZ4oYi54xWLVS3ms6uyZuk3ve2SiHEqQSWMzheNkDMEHRw32GMFe82YUO55+pZ3UEVL/nNLujCCtS19LyM/JarlnZQ2b4yqx7P9P094M3/SixPd22ad4b/xdlcVhih6PBYibw4+heiY5bHn6bncXY5emeaQhm1h314oeq3gTea5zNso/WON80gQnGB4yU4wZsmmP09So1XPsD2YMNNVfi/gK4J8uvocNpwRqP3UbXUOUH+m7vsY+PUbedd6gS4kqb5OjUVpF7PMrguO9A==,iv:ApldFxLlG/wUrrLXZVyEQrZsv25VqZfrtxMOR0b3A4g=,tag:oO98FOrBN7Wu3JUfkNsL+g==,type:str] -sops: - age: - - enc: | - -----BEGIN AGE ENCRYPTED FILE----- - YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSAxNFk4M3VVRHZvM2N5Qm1P - SDZKeTNPYjh1QytMWjVaUFk1WGFkdmo3T21VCnp3MGt2dTNXNHhSdTVEMjhjeUs4 - cFcxWFJRWE9EMDlrMGR6M1BkWXRwNXcKLS0tIHFvaEQ4YWtDU05UTHplUDlRVmFq - SXB2b2wrZlVCeUhXaG5QQkxWQXJ6aDgKqV2V1jyeKQHr1iDaWR36WZWZdkZeSUcw - BcA8vT73qdIb3Rnt6zYIBfgFYuVgQp4m+2pU3J4xgyGKPwQEtLz2XA== - -----END AGE ENCRYPTED FILE----- - recipient: age1e5fq3hwxy78psus2nfvmtmua36g0u3suk78ephw6246l974d2utsvn0hla - encrypted_regex: ^data - lastmodified: "2026-09-12T23:58:01Z" - mac: ENC[AES256_GCM,data:4zterh5GOq6fibw+npmmIB//+7nYoKoQDXoinNCbTEAiQPuxqcpYJn1Mxmt37ZobG8Y7rqQYEpoRLCangFZA2b7mbkEWeEf9qXhP+cufqWB+Aagr7b8XmbL6tNSRxamxr+4WyjJzSrY+5w6d/4usDGRJhs+AB2F/Vmq7peDhZiU=,iv:G+ZbPJ7YiQlrhwt+7EUxWq8pNelw/bUz6thYufh3+rA=,tag:uaCoQ7wg7Rg9xKkq5ZT6NA==,type:str] - version: 3.13.2 diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index c3fd2dd..6a7f02d 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -21,9 +21,10 @@ spec: labels: app: api-gateway component: gateway - # Required by the llm-serving-default-deny NetworkPolicy, which admits - # only pods labelled llm-client=true (from any namespace) on port 8080. - # Without it every upstream dial times out and dispatch returns 502. + # The llm-serving-ingress NetworkPolicy in the llm-serving namespace now + # selects by namespace (kubernetes.io/metadata.name: api), not by this + # label. Kept for older NetworkPolicy revisions that still admit on + # llm-client=true. llm-client: "true" annotations: reloader.stakater.com/auto: "true" @@ -67,6 +68,18 @@ spec: value: "5m" - name: LOG_LEVEL value: "info" + - name: ADMISSION_MAX_PARK_SECONDS + value: "120" + # Only backends that publish the vLLM scheduler gauges may be observed. + # `reasoning` is the only vLLM engine; ornith/qwen-cpu are Ollama and + # embeddings/rerank are TEI, so observing them would hold + # observer_degraded at 1 forever. + - name: ADMISSION_OBSERVE_MODELS + value: "reasoning" + - name: VLLM_MAX_QUEUE_DEPTH + value: "8" + - name: MAX_BODY_BYTES + value: "1048576" # OpenTelemetry tracing configuration - name: OTEL_EXPORTER_OTLP_ENDPOINT value: "otel-collector.tracing.svc.cluster.local:4317"