Introduce SaturationDetector component

This commit adds a new `SaturationDetector` component responsible for
determining if backend model servers are saturated. It bases its
decision on observed metrics like queue depth and KV cache utilization,
using configurable thresholds.

The detector is designed to be a self-contained unit that can be
leveraged by other components for admission control and capacity
assessment. This is the first step in a larger refactoring to
externalize and centralize saturation detection logic.

Author: LukeAVanDrie

pkg/epp/saturationdetector/config.go

@@ -0,0 +1,61 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package saturationdetector
+
+import (
+	"fmt"
+	"time"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	schedcfg "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/config"
+	envutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/env"
+)
+
+// Default configuration values
+const (
+	DefaultQueueDepthThreshold  = schedcfg.DefaultQueueThresholdCritical
+	DefaultKVCacheUtilThreshold = schedcfg.DefaultKVCacheThreshold
+	// DefaultMetricsStalenessThreshold defines how old metrics can be before they
+	// are considered stale.
+	// Given the pod metrics refresh interval is 50ms, a threshold slightly above
+	// that should be fine.
+	DefaultMetricsStalenessThreshold = 200 * time.Millisecond
+)
+
+// Environment variable names for SaturationDetector configuration
+const (
+	EnvSdQueueDepthThreshold       = "SD_QUEUE_DEPTH_THRESHOLD"
+	EnvSdKVCacheUtilThreshold      = "SD_KV_CACHE_UTIL_THRESHOLD"
+	EnvSdMetricsStalenessThreshold = "SD_METRICS_STALENESS_THRESHOLD"
+)
+
+// LoadConfigFromEnv loads SaturationDetector Config from environment
+// variables.
+func LoadConfigFromEnv() *Config {
+	// Use a default logger for initial configuration loading.
+	logger := log.Log.WithName("saturation-detector-config")
+
+	cfg := &Config{}
+
+	cfg.QueueDepthThreshold = envutil.GetEnvInt(EnvSdQueueDepthThreshold, DefaultQueueDepthThreshold, logger)
+	cfg.KVCacheUtilThreshold = envutil.GetEnvFloat(EnvSdKVCacheUtilThreshold, DefaultKVCacheUtilThreshold, logger)
+	cfg.MetricsStalenessThreshold = envutil.GetEnvDuration(EnvSdMetricsStalenessThreshold, DefaultMetricsStalenessThreshold, logger)
+
+	// NewDetector validates the config and assigns defaults.
+	logger.Info("SaturationDetector configuration loaded from env",
+		"config", fmt.Sprintf("%+v", cfg))
+	return cfg
+}

pkg/epp/saturationdetector/saturationdetector.go

@@ -0,0 +1,175 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package saturationdetector implements a mechanism to determine if the
+// backend model servers are considered saturated based on observed metrics.
+//
+// The current implementation provides a global saturation signal (IsSaturated)
+// primarily based on backend queue depths and KV cache utilization, reflecting
+// the saturation signals previously used by the Scheduler before the
+// introduction of the FlowController. It fetches live metrics from the
+// provided Datastore.
+//
+// TODO: Explore more advanced saturation signals in the future, such as:
+//   - Latency-objective-based saturation.
+//   - Predictive saturation based on trends.
+//   - Hysteresis bands or other smoothing techniques to prevent rapid
+//     oscillations of the saturation signal.
+package saturationdetector
+
+import (
+	"errors"
+	"time"
+
+	"github.com/go-logr/logr"
+	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+// clock allows mocking time in tests.
+type clock interface {
+	now() time.Time
+}
+
+// realClock provides the real time.
+type realClock struct{}
+
+func (c realClock) now() time.Time { return time.Now() }
+
+// ErrNilDatastore indicates NewSaturationDetector was called with a nil
+// datastore.
+var ErrNilDatastore = errors.New("datastore cannot be nil")
+
+// Config holds the configuration for the SaturationDetector.
+type Config struct {
+	// QueueDepthThreshold defines the backend waiting queue size above which a
+	// pod is considered to have insufficient capacity for new requests.
+	QueueDepthThreshold int
+	// KVCacheUtilThreshold defines the KV cache utilization (0.0 to 1.0) above
+	// which a pod is considered to have insufficient capacity.
+	KVCacheUtilThreshold float64
+	// MetricsStalenessThreshold defines how old a pod's metrics can be.
+	// If a pod's metrics are older than this, it might be excluded from
+	// "good capacity" considerations or treated as having no capacity for
+	// safety.
+	MetricsStalenessThreshold time.Duration
+}
+
+// Datastore provides an interface to access backend pod metrics.
+type Datastore interface {
+	PodGetAll() []backendmetrics.PodMetrics
+}
+
+// Detector determines system saturation based on metrics from the Datastore.
+type Detector struct {
+	datastore Datastore
+	config    Config
+	logger    logr.Logger
+	clock     clock
+}
+
+// NewDetector creates a new SaturationDetector.
+// The datastore is expected to provide access to live/recently-updated pod
+// metrics.
+// The config provides the thresholds for determining saturation.
+func NewDetector(config Config, datastore Datastore, logger logr.Logger) (*Detector, error) {
+	if datastore == nil {
+		return nil, ErrNilDatastore
+	}
+	if config.MetricsStalenessThreshold <= 0 {
+		config.MetricsStalenessThreshold = DefaultMetricsStalenessThreshold
+	}
+	logger.V(logutil.DEFAULT).Info("Creating new SaturationDetector",
+		"queueDepthThreshold", config.QueueDepthThreshold,
+		"kvCacheUtilThreshold", config.KVCacheUtilThreshold,
+		"metricsStalenessThreshold", config.MetricsStalenessThreshold.String())
+	return &Detector{
+		datastore: datastore,
+		config:    config,
+		logger:    logger.WithName("SaturationDetector"),
+		clock:     realClock{},
+	}, nil
+}
+
+// IsSaturated checks if the system is currently considered saturated.
+// The system is saturated if NO pod currently has "good capacity".
+// "Good capacity" means:
+//  1. Metrics are fresh (not stale).
+//  2. WaitingQueueSize <= QueueDepthThreshold.
+//  3. KVCacheUsagePercent <= KVCacheUtilThreshold.
+//
+// If no pods are found in the datastore, the system is considered saturated
+// (no capacity).
+func (d *Detector) IsSaturated() bool {
+	allPodsMetrics := d.datastore.PodGetAll()
+	if len(allPodsMetrics) == 0 {
+		d.logger.V(logutil.VERBOSE).Info("No pods found in datastore; system is considered SATURATED (no capacity).")
+		// If there are no pods, there is no capacity to serve requests.
+		// Treat this as a saturated state to enable FlowController queuing.
+		return true
+	}
+
+	now := d.clock.now()
+	foundPodWithGoodCapacity := false
+	for _, podMetric := range allPodsMetrics {
+		metrics := podMetric.GetMetrics()
+		podNn := "unknown-pod"
+		if podMetric.GetPod() != nil {
+			podNn = podMetric.GetPod().NamespacedName.String()
+		}
+
+		if metrics == nil {
+			d.logger.V(logutil.VERBOSE).Info("Pod has nil metrics, skipping for saturation check",
+				"pod", podNn)
+			continue
+		}
+
+		// 1. Check for metric staleness
+		if now.Sub(metrics.UpdateTime) > d.config.MetricsStalenessThreshold {
+			d.logger.V(logutil.VERBOSE).Info("Pod metrics are stale, considered as not having good capacity",
+				"pod", podNn,
+				"updateTime", metrics.UpdateTime,
+				"stalenessThreshold", d.config.MetricsStalenessThreshold)
+			continue
+		}
+
+		// 2. Check queue depth
+		isQueueGood := metrics.WaitingQueueSize <= d.config.QueueDepthThreshold
+
+		// 3. Check KV cache utilization
+		isKVCacheGood := metrics.KVCacheUsagePercent <= d.config.KVCacheUtilThreshold
+
+		if isQueueGood && isKVCacheGood {
+			d.logger.V(logutil.VERBOSE).Info("Found pod with good capacity",
+				"pod", podNn,
+				"waitingQueue", metrics.WaitingQueueSize,
+				"queueThreshold", d.config.QueueDepthThreshold,
+				"kvCacheUtil", metrics.KVCacheUsagePercent,
+				"kvCacheThreshold", d.config.KVCacheUtilThreshold)
+			foundPodWithGoodCapacity = true
+			// Found at least one pod with good capacity, so system is NOT saturated.
+			break
+		}
+	}
+
+	if !foundPodWithGoodCapacity {
+		d.logger.V(logutil.VERBOSE).Info("No pods found with good capacity; system is considered SATURATED.")
+		return true
+	}
+
+	d.logger.V(logutil.VERBOSE).Info("System is considered NOT saturated (at least one pod has good capacity).")
+	return false
+}