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: "<quota> <period>" or "max <period>".
+	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
+}

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),
+		requestTotal:            make(map[string]int64),
-		requestDuration:        make(map[string]int64),
+		requestDuration:         make(map[string]int64),
-		requestDurationBuckets: make(map[string]map[float64]int64),
+		requestDurationBuckets:  make(map[string]map[float64]int64),
-		bytesIn:                make(map[string]int64),
+		bytesIn:                 make(map[string]int64),
-		bytesOut:               make(map[string]int64),
+		bytesOut:                make(map[string]int64),
-		upstreamHealth:         make(map[string]int),
+		upstreamHealth:          make(map[string]int),
 		streamingResponsesTotal: make(map[string]int64),
-		streamingByteCount:     make(map[string]int64),
+		streamingByteCount:      make(map[string]int64),
-		ttftMs:                 make(map[string][]int64),
+		ttftMs:                  make(map[string][]int64),
-		itlMs:                  make(map[string][]int64),
+		itlMs:                   make(map[string][]int64),
-		tokenCount:             make(map[string]int64),
+		tokenCount:              make(map[string]int64),
+		admissionRejected:       make(map[string]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

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"
 }
 

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
+}

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 {
+		if len(addrs) == 0 {
-			h.schedulers[modelName] = NewScheduler(addrs, maxSlots, maxQueue)
+			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.
+// to Release (caller must defer), the (possibly patched) route, whether SSE
-// If the slot cannot be acquired it writes a 503 and returns (nil, nil, false).
+// headers were already written while parking, and ok=true.
-func (h *Handler) acquireScheduledBackend(ctx context.Context, w http.ResponseWriter, r *http.Request, sched *Scheduler, route *Route, cacheKey string) (*Backend, *Route, bool) {
+// 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")
+			if !prewritten {
-			w.WriteHeader(http.StatusServiceUnavailable)
+				w.Header().Set("Retry-After", "5")
-			fmt.Fprintf(w, "queue full")
+			}
+			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)
+			reject("upstream busy")
-			fmt.Fprintf(w, "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{

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
+	// Extract only the top-level "model" value with a token walk: the common
-	var payload map[string]interface{}
+	// path must not materialise the whole body (messages and all) as a map.
-	if err := json.Unmarshal(bodyBytes, &payload); err != nil {
+	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)
+	var modelName string
-	if !ok {
+	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
+		var fields map[string]json.RawMessage
-		newBody, err := json.Marshal(payload)
+		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
+}

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
+	Address     string
-	queue   chan struct{}
+	queue       chan struct{}
-	slots   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,

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")
 

k8s/deployment.yaml

@@ -21,9 +21,10 @@ spec:
       labels:
         app: api-gateway
         component: gateway
-        # Required by the llm-serving-default-deny NetworkPolicy, which admits
+        # The llm-serving-ingress NetworkPolicy in the llm-serving namespace now
-        # only pods labelled llm-client=true (from any namespace) on port 8080.
+        # selects by namespace (kubernetes.io/metadata.name: api), not by this
-        # Without it every upstream dial times out and dispatch returns 502.
+        # 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"