feat(proxy): engine-wide LLM admission control via vLLM gauges (#64)

The per-pod semaphore was not a cap: 3 replicas x 4 slots advertised 12
concurrent requests while the engine runs 4, and nothing coordinated the
replicas. Admission is now gated on the engine's own published gauges,
which is global by construction because every replica contends for the
same engine.

- new internal/proxy/engine_state.go: 2s scrape of the backend /metrics,
  hand-rolled exposition parse (no new dependency), unhealthy after 3
  consecutive failed polls, and a startup self-check that raises
  observer_degraded rather than falling back to fail-open silently
- gate: park while kv_cache_usage_perc >= 0.90 or running + waiting >=
  ADMISSION_MAX_INFLIGHT, bounded by ADMISSION_MAX_PARK_SECONDS, then
  503 + Retry-After: 5. Unknown state fails open on purpose
- streaming requests receive SSE keep-alive heartbeats while parked
- router: extract only the top-level model field instead of unmarshalling
  the whole body into a map; body cap from MAX_BODY_BYTES (1 MiB, the
  model path previously inherited the 100 MB default)
- GOMAXPROCS from the cgroup CPU quota: the container is limited to 500m
  while Go defaulted to the host core count and was CFS-throttled
- VLLM_MAX_QUEUE_DEPTH default 20 -> 8
- delete k8s/configmap.yaml: absent from kustomization.yaml, and the live
  ConfigMap of that name carries an unrelated key
- correct the llm-client NetworkPolicy comment; llm-serving-ingress now
  selects by namespace, not by that label

Observers run only for models named in ADMISSION_OBSERVE_MODELS.
reasoning is the only vLLM engine here; ornith and qwen-cpu are Ollama
and embeddings/rerank are TEI, none of which publish vllm:* series, so
observing them would hold observer_degraded at 1 permanently.

Companion change (engine flags + dashboard panels) is
riotpiao-poimen/homelab#51.

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