requestcontrol: enable scale-from-zero

Refactors the request processing flow to support queuing when no
backends are available.

- Inverts Director flow: Admission is now called before Pod Resolution.
- Updates AdmissionController interface to remove eager pod list.
- LegacyAdmissionController now resolves pods internally via PodLocator.
- ShardProcessor (Flow Control) now resolves pods lazily via PodLocator
  during the dispatch cycle.
- Updates Runner wiring to inject PodLocator where needed.

Author: LukeAVanDrie

cmd/epp/runner/runner.go

@@ -329,6 +329,8 @@ func (r *Runner) Run(ctx context.Context) error {
 	saturationDetector := saturationdetector.NewDetector(eppConfig.SaturationDetectorConfig, setupLog)
 
 	// --- Admission Control Initialization ---
+	locator := requestcontrol.NewDatastorePodLocator(ds)
+	cachedLocator := requestcontrol.NewCachedPodLocator(ctx, locator, time.Millisecond*50)
 	var admissionController requestcontrol.AdmissionController
 	if r.featureGates[flowcontrol.FeatureGate] {
 		setupLog.Info("Initializing experimental Flow Control layer")
@@ -342,19 +344,23 @@ func (r *Runner) Run(ctx context.Context) error {
 		if err != nil {
 			return fmt.Errorf("failed to initialize Flow Registry: %w", err)
 		}
-		fc, err := fccontroller.NewFlowController(ctx, fcCfg.Controller, registry, saturationDetector, setupLog)
+		fc, err := fccontroller.NewFlowController(
+			ctx,
+			fcCfg.Controller,
+			registry, saturationDetector,
+			cachedLocator,
+			setupLog,
+		)
 		if err != nil {
 			return fmt.Errorf("failed to initialize Flow Controller: %w", err)
 		}
 		go registry.Run(ctx)
-		admissionController = requestcontrol.NewFlowControlAdmissionController(saturationDetector, fc)
+		admissionController = requestcontrol.NewFlowControlAdmissionController(fc)
 	} else {
 		setupLog.Info("Experimental Flow Control layer is disabled, using legacy admission control")
-		admissionController = requestcontrol.NewLegacyAdmissionController(saturationDetector)
+		admissionController = requestcontrol.NewLegacyAdmissionController(saturationDetector, cachedLocator)
 	}
 
-	locator := requestcontrol.NewDatastorePodLocator(ds)
-	cachedLocator := requestcontrol.NewCachedPodLocator(ctx, locator, time.Millisecond*50)
 	director := requestcontrol.NewDirectorWithConfig(
 		ds,
 		scheduler,

pkg/epp/flowcontrol/controller/controller.go

@@ -60,6 +60,7 @@ type shardProcessorFactory func(
 	ctx context.Context,
 	shard contracts.RegistryShard,
 	saturationDetector contracts.SaturationDetector,
+	podLocator contracts.PodLocator,
 	clock clock.WithTicker,
 	cleanupSweepInterval time.Duration,
 	enqueueChannelBufferSize int,
@@ -95,6 +96,7 @@ type FlowController struct {
 	config                Config
 	registry              registryClient
 	saturationDetector    contracts.SaturationDetector
+	podLocator            contracts.PodLocator
 	clock                 clock.WithTicker
 	logger                logr.Logger
 	shardProcessorFactory shardProcessorFactory
@@ -126,13 +128,15 @@ func NewFlowController(
 	config Config,
 	registry contracts.FlowRegistry,
 	sd contracts.SaturationDetector,
+	podLocator contracts.PodLocator,
 	logger logr.Logger,
 	opts ...flowControllerOption,
 ) (*FlowController, error) {
 	fc := &FlowController{
 		config:             config,
 		registry:           registry,
 		saturationDetector: sd,
+		podLocator:         podLocator,
 		clock:              clock.RealClock{},
 		logger:             logger.WithName("flow-controller"),
 		parentCtx:          ctx,
@@ -142,6 +146,7 @@ func NewFlowController(
 		ctx context.Context,
 		shard contracts.RegistryShard,
 		saturationDetector contracts.SaturationDetector,
+		podLocator contracts.PodLocator,
 		clock clock.WithTicker,
 		cleanupSweepInterval time.Duration,
 		enqueueChannelBufferSize int,
@@ -151,6 +156,7 @@ func NewFlowController(
 			ctx,
 			shard,
 			saturationDetector,
+			podLocator,
 			clock,
 			cleanupSweepInterval,
 			enqueueChannelBufferSize,
@@ -448,6 +454,7 @@ func (fc *FlowController) getOrStartWorker(shard contracts.RegistryShard) *manag
 		processorCtx,
 		shard,
 		fc.saturationDetector,
+		fc.podLocator,
 		fc.clock,
 		fc.config.ExpiryCleanupInterval,
 		fc.config.EnqueueChannelBufferSize,

pkg/epp/flowcontrol/controller/controller_test.go

@@ -79,7 +79,6 @@ type testHarness struct {
 	// clock is the clock interface used by the controller.
 	clock        clock.WithTicker
 	mockRegistry *mockRegistryClient
-	mockDetector *mocks.MockSaturationDetector
 	// mockClock provides access to FakeClock methods (Step, HasWaiters) if and only if the underlying clock is a
 	// FakeClock.
 	mockClock            *testclock.FakeClock
@@ -91,6 +90,7 @@ type testHarness struct {
 func newUnitHarness(t *testing.T, ctx context.Context, cfg Config, registry *mockRegistryClient) *testHarness {
 	t.Helper()
 	mockDetector := &mocks.MockSaturationDetector{}
+	mockPodLocator := &mocks.MockPodLocator{}
 
 	// Initialize the FakeClock with the current system time.
 	// The controller implementation uses the injected clock to calculate the deadline timestamp,vbut uses the standard
@@ -113,15 +113,14 @@ func newUnitHarness(t *testing.T, ctx context.Context, cfg Config, registry *moc
 		withClock(mockClock),
 		withShardProcessorFactory(mockProcessorFactory.new),
 	}
-	fc, err := NewFlowController(ctx, cfg, registry, mockDetector, logr.Discard(), opts...)
+	fc, err := NewFlowController(ctx, cfg, registry, mockDetector, mockPodLocator, logr.Discard(), opts...)
 	require.NoError(t, err, "failed to create FlowController for unit test harness")
 
 	h := &testHarness{
 		fc:                   fc,
 		cfg:                  cfg,
 		clock:                mockClock,
 		mockRegistry:         registry,
-		mockDetector:         mockDetector,
 		mockClock:            mockClock,
 		mockProcessorFactory: mockProcessorFactory,
 	}
@@ -133,8 +132,9 @@ func newUnitHarness(t *testing.T, ctx context.Context, cfg Config, registry *moc
 func newIntegrationHarness(t *testing.T, ctx context.Context, cfg Config, registry *mockRegistryClient) *testHarness {
 	t.Helper()
 	mockDetector := &mocks.MockSaturationDetector{}
-	// Align FakeClock with system time. See explanation in newUnitHarness.
+	mockPodLocator := &mocks.MockPodLocator{}
 
+	// Align FakeClock with system time. See explanation in newUnitHarness.
 	mockClock := testclock.NewFakeClock(time.Now())
 	if registry == nil {
 		registry = &mockRegistryClient{}
@@ -144,15 +144,14 @@ func newIntegrationHarness(t *testing.T, ctx context.Context, cfg Config, regist
 		withRegistryClient(registry),
 		withClock(mockClock),
 	}
-	fc, err := NewFlowController(ctx, cfg, registry, mockDetector, logr.Discard(), opts...)
+	fc, err := NewFlowController(ctx, cfg, registry, mockDetector, mockPodLocator, logr.Discard(), opts...)
 	require.NoError(t, err, "failed to create FlowController for integration test harness")
 
 	h := &testHarness{
 		fc:           fc,
 		cfg:          cfg,
 		clock:        mockClock,
 		mockRegistry: registry,
-		mockDetector: mockDetector,
 		mockClock:    mockClock,
 	}
 	return h
@@ -247,6 +246,7 @@ func (f *mockShardProcessorFactory) new(
 	_ context.Context, // The factory does not use the lifecycle context; it's passed to the processor's Run method later.
 	shard contracts.RegistryShard,
 	_ contracts.SaturationDetector,
+	_ contracts.PodLocator,
 	_ clock.WithTicker,
 	_ time.Duration,
 	_ int,
@@ -1001,6 +1001,7 @@ func TestFlowController_WorkerManagement(t *testing.T) {
 			ctx context.Context, // The context created by getOrStartWorker for the potential new processor.
 			shard contracts.RegistryShard,
 			_ contracts.SaturationDetector,
+			_ contracts.PodLocator,
 			_ clock.WithTicker,
 			_ time.Duration,
 			_ int,

pkg/epp/flowcontrol/controller/internal/processor.go

@@ -65,6 +65,7 @@ var ErrProcessorBusy = errors.New("shard processor is busy")
 type ShardProcessor struct {
 	shard                contracts.RegistryShard
 	saturationDetector   contracts.SaturationDetector
+	podLocator           contracts.PodLocator
 	clock                clock.WithTicker
 	cleanupSweepInterval time.Duration
 	logger               logr.Logger
@@ -86,6 +87,7 @@ func NewShardProcessor(
 	ctx context.Context,
 	shard contracts.RegistryShard,
 	saturationDetector contracts.SaturationDetector,
+	podLocator contracts.PodLocator,
 	clock clock.WithTicker,
 	cleanupSweepInterval time.Duration,
 	enqueueChannelBufferSize int,
@@ -94,6 +96,7 @@ func NewShardProcessor(
 	return &ShardProcessor{
 		shard:                shard,
 		saturationDetector:   saturationDetector,
+		podLocator:           podLocator,
 		clock:                clock,
 		cleanupSweepInterval: cleanupSweepInterval,
 		logger:               logger,
@@ -307,8 +310,8 @@ func (sp *ShardProcessor) dispatchCycle(ctx context.Context) bool {
 
 		// --- Viability Check (Saturation/HoL Blocking) ---
 		req := item.OriginalRequest()
-		candidatePods := req.CandidatePodsForScheduling()
-		if sp.saturationDetector.IsSaturated(ctx, candidatePods) {
+		candidates := sp.podLocator.Locate(ctx, req.GetMetadata())
+		if sp.saturationDetector.IsSaturated(ctx, candidates) {
 			sp.logger.V(logutil.DEBUG).Info("Policy's chosen item is saturated; enforcing HoL blocking.",
 				"flowKey", req.FlowKey(), "reqID", req.ID(), "priorityName", originalBand.PriorityName())
 			// Stop the dispatch cycle entirely to respect strict policy decision and prevent priority inversion where

pkg/epp/flowcontrol/controller/internal/processor_test.go

@@ -75,6 +75,7 @@ type testHarness struct {
 	clock              *testclock.FakeClock
 	logger             logr.Logger
 	saturationDetector *mocks.MockSaturationDetector
+	podLocator         *mocks.MockPodLocator
 
 	// --- Centralized Mock State ---
 	// The harness's mutex protects the single source of truth for all mock state.
@@ -96,6 +97,7 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn
 		clock:              testclock.NewFakeClock(time.Now()),
 		logger:             logr.Discard(),
 		saturationDetector: &mocks.MockSaturationDetector{},
+		podLocator:         &mocks.MockPodLocator{Pods: []metrics.PodMetrics{&metrics.FakePodMetrics{}}},
 		startSignal:        make(chan struct{}),
 		queues:             make(map[types.FlowKey]*mocks.MockManagedQueue),
 		priorityFlows:      make(map[int][]types.FlowKey),
@@ -123,6 +125,7 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn
 		h.ctx,
 		h,
 		h.saturationDetector,
+		h.podLocator,
 		h.clock,
 		expiryCleanupInterval,
 		100,

pkg/epp/requestcontrol/admission.go

@@ -20,9 +20,11 @@ import (
 	"context"
 	"time"
 
+	"github.com/go-logr/logr"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
@@ -40,7 +42,6 @@ type AdmissionController interface {
 	// Args:
 	//   ctx: The request context, carrying deadlines, cancellation signals, and logger.
 	//   reqCtx: The handlers.RequestContext containing details about the incoming request.
-	//   candidatePods: A list of potential backend pods that can serve the request.
 	//   priority: The priority level of the request, as determined by the InferenceObjective.
 	//
 	// Returns:
@@ -49,7 +50,6 @@ type AdmissionController interface {
 	Admit(
 		ctx context.Context,
 		reqCtx *handlers.RequestContext,
-		candidatePods []backendmetrics.PodMetrics,
 		priority int,
 	) error
 }
@@ -65,18 +65,17 @@ type flowController interface {
 	EnqueueAndWait(ctx context.Context, req types.FlowControlRequest) (types.QueueOutcome, error)
 }
 
-// rejectIfSheddableAndSaturated checks if a request should be immediately rejected because it's sheddable
-// (priority < 0) and the system is saturated.
+// rejectIfSheddableAndSaturated checks if a request should be immediately rejected.
 func rejectIfSheddableAndSaturated(
 	ctx context.Context,
 	sd saturationDetector,
+	locator contracts.PodLocator,
 	reqCtx *handlers.RequestContext,
-	candidatePods []backendmetrics.PodMetrics,
 	priority int,
+	logger logr.Logger,
 ) error {
 	if requtil.IsSheddable(priority) {
-		logger := log.FromContext(ctx)
-		if sd.IsSaturated(ctx, candidatePods) {
+		if sd.IsSaturated(ctx, locator.Locate(ctx, reqCtx.Request.Metadata)) {
 			logger.V(logutil.TRACE).Info("Request rejected: system saturated and request is sheddable",
 				"requestID", reqCtx.SchedulingRequest.RequestId)
 			return errutil.Error{
@@ -95,25 +94,37 @@ func rejectIfSheddableAndSaturated(
 // saturated. Non-sheddable requests always bypass the saturation check.
 type LegacyAdmissionController struct {
 	saturationDetector saturationDetector
+	podLocator         contracts.PodLocator
 }
 
 // NewLegacyAdmissionController creates a new LegacyAdmissionController.
-func NewLegacyAdmissionController(sd saturationDetector) *LegacyAdmissionController {
-	return &LegacyAdmissionController{saturationDetector: sd}
+func NewLegacyAdmissionController(
+	sd saturationDetector,
+	pl contracts.PodLocator,
+) *LegacyAdmissionController {
+	return &LegacyAdmissionController{
+		saturationDetector: sd,
+		podLocator:         pl,
+	}
 }
 
 // Admit implements the AdmissionController interface for the legacy strategy.
 // It checks for saturation only for requests with priority < 0.
 func (lac *LegacyAdmissionController) Admit(
 	ctx context.Context,
 	reqCtx *handlers.RequestContext,
-	candidatePods []backendmetrics.PodMetrics,
 	priority int,
 ) error {
 	logger := log.FromContext(ctx)
 	logger.V(logutil.TRACE).Info("Executing LegacyAdmissionController",
 		"priority", priority, "fairnessID", reqCtx.FairnessID)
-	if err := rejectIfSheddableAndSaturated(ctx, lac.saturationDetector, reqCtx, candidatePods, priority); err != nil {
+	if err := rejectIfSheddableAndSaturated(
+		ctx,
+		lac.saturationDetector,
+		lac.podLocator,
+		reqCtx, priority,
+		logger,
+	); err != nil {
 		return err
 	}
 	logger.V(logutil.TRACE).Info("Request admitted", "requestID", reqCtx.SchedulingRequest.RequestId)
@@ -123,19 +134,15 @@ func (lac *LegacyAdmissionController) Admit(
 // --- FlowControlAdmissionController ---
 
 // FlowControlAdmissionController delegates admission decisions to the Flow Control layer.
-// It first checks if the request is sheddable and the system is saturated, rejecting immediately if both conditions are
-// true. Otherwise, it uses the provided flowController to enqueue the request and await an outcome.
+// It uses the provided Flow Controller to enqueue the request and await an outcome.
 type FlowControlAdmissionController struct {
-	saturationDetector saturationDetector
-	flowController     flowController
+	flowController flowController
 }
 
 // NewFlowControlAdmissionController creates a new FlowControlAdmissionController.
-// It requires a SaturationDetector and a flowController instance.
-func NewFlowControlAdmissionController(sd saturationDetector, fc flowController) *FlowControlAdmissionController {
+func NewFlowControlAdmissionController(fc flowController) *FlowControlAdmissionController {
 	return &FlowControlAdmissionController{
-		saturationDetector: sd,
-		flowController:     fc,
+		flowController: fc,
 	}
 }
 
@@ -144,24 +151,18 @@ func NewFlowControlAdmissionController(sd saturationDetector, fc flowController)
 func (fcac *FlowControlAdmissionController) Admit(
 	ctx context.Context,
 	reqCtx *handlers.RequestContext,
-	candidatePods []backendmetrics.PodMetrics,
 	priority int,
 ) error {
 	logger := log.FromContext(ctx)
 	logger.V(logutil.TRACE).Info("Executing FlowControlAdmissionController",
 		"requestID", reqCtx.SchedulingRequest.RequestId, "priority", priority, "fairnessID", reqCtx.FairnessID)
-	if err := rejectIfSheddableAndSaturated(ctx, fcac.saturationDetector, reqCtx, candidatePods, priority); err != nil {
-		return err
-	}
-
-	logger.V(logutil.TRACE).Info("Request proceeding to flow control", "requestID", reqCtx.SchedulingRequest.RequestId)
 
 	fcReq := &flowControlRequest{
 		requestID:       reqCtx.SchedulingRequest.RequestId,
 		fairnessID:      reqCtx.FairnessID,
 		priority:        priority,
 		requestByteSize: uint64(reqCtx.RequestSize),
-		candidatePods:   candidatePods,
+		reqMetadata:     reqCtx.Request.Metadata,
 	}
 
 	outcome, err := fcac.flowController.EnqueueAndWait(ctx, fcReq)
@@ -176,20 +177,20 @@ type flowControlRequest struct {
 	fairnessID      string
 	priority        int
 	requestByteSize uint64
-	candidatePods   []backendmetrics.PodMetrics
+	reqMetadata     map[string]any
 }
 
 var _ types.FlowControlRequest = &flowControlRequest{}
 
 func (r *flowControlRequest) ID() string                         { return r.requestID }
 func (r *flowControlRequest) InitialEffectiveTTL() time.Duration { return 0 } // Use controller default.
 func (r *flowControlRequest) ByteSize() uint64                   { return r.requestByteSize }
-func (r *flowControlRequest) CandidatePodsForScheduling() []backendmetrics.PodMetrics {
-	return r.candidatePods
-}
 func (r *flowControlRequest) FlowKey() types.FlowKey {
 	return types.FlowKey{ID: r.fairnessID, Priority: r.priority}
 }
+func (r *flowControlRequest) GetMetadata() map[string]any {
+	return r.reqMetadata
+}
 
 // translateFlowControlOutcome maps the context-rich outcome of the Flow Control layer to the public errutil.Error
 // contract used by the Director.