feat: Enable Scale-from-Zero with Flow Control enabled (#1952)

* contracts: add PodLocator for candidate resolution

This defines the contract for resolving candidate pods based on request
metadata, decoupling the resolution logic from the storage layer.

* director: delegate candidate resolution

Refactors the Director to use the injected PodLocator interface instead
of the private getCandidatePodsForScheduling method. This prepares the
Director for lazy resolution without changing current behavior.

* flowcontrol: add metadata support to request type

Updates the FlowControlRequest interface to carry request metadata
instead of a pre-resolved list of candidate pods. This prepares the
system for lazy pod resolution.

- Adds GetMetadata() to FlowControlRequest.
- Removes CandidatePodsForScheduling() from FlowControlRequest.
- Updates mocks in flowcontrol/types and contracts.

* 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.

* resolve merge conflicts after rebase

* Address reviewer feedback

Author: LukeAVanDrie

cmd/epp/runner/runner.go

@@ -60,6 +60,7 @@ import (
 	dlmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	fccontroller "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/controller"
 	fcregistry "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/registry"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
@@ -330,7 +331,10 @@ func (r *Runner) Run(ctx context.Context) error {
 
 	// --- Admission Control Initialization ---
 	var admissionController requestcontrol.AdmissionController
+	var locator contracts.PodLocator
+	locator = requestcontrol.NewDatastorePodLocator(ds)
 	if r.featureGates[flowcontrol.FeatureGate] {
+		locator = requestcontrol.NewCachedPodLocator(ctx, locator, time.Millisecond*50)
 		setupLog.Info("Initializing experimental Flow Control layer")
 		fcCfg, err := flowControlConfig.ValidateAndApplyDefaults()
 		if err != nil {
@@ -342,24 +346,28 @@ 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,
+			locator,
+			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, locator)
 	}
 
-	locator := requestcontrol.NewDatastorePodLocator(ds)
-	cachedLocator := requestcontrol.NewCachedPodLocator(ctx, locator, time.Millisecond*50)
 	director := requestcontrol.NewDirectorWithConfig(
 		ds,
 		scheduler,
 		admissionController,
-		cachedLocator,
+		locator,
 		r.requestControlConfig)
 
 	// --- Setup ExtProc Server Runner ---

pkg/epp/flowcontrol/contracts/mocks/mocks.go

@@ -41,6 +41,8 @@ import (
 	typesmocks "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types/mocks"
 )
 
+// --- RegistryShard Mocks ---
+
 // MockRegistryShard is a simple "stub-style" mock for testing.
 // Its methods are implemented as function fields (e.g., `IDFunc`). A test can inject behavior by setting the desired
 // function field in the test setup. If a func is nil, the method will return a zero value.
@@ -111,6 +113,8 @@ func (m *MockRegistryShard) Stats() contracts.ShardStats {
 	return contracts.ShardStats{}
 }
 
+// --- Dependency Mocks ---
+
 // MockSaturationDetector is a simple "stub-style" mock for testing.
 type MockSaturationDetector struct {
 	IsSaturatedFunc func(ctx context.Context, candidatePods []metrics.PodMetrics) bool
@@ -123,6 +127,30 @@ func (m *MockSaturationDetector) IsSaturated(ctx context.Context, candidatePods
 	return false
 }
 
+// MockPodLocator provides a mock implementation of the contracts.PodLocator interface.
+// It allows tests to control the exact set of pods returned for a given request.
+type MockPodLocator struct {
+	// LocateFunc allows injecting custom logic.
+	LocateFunc func(ctx context.Context, requestMetadata map[string]any) []metrics.PodMetrics
+	// Pods is a static return value used if LocateFunc is nil.
+	Pods []metrics.PodMetrics
+}
+
+func (m *MockPodLocator) Locate(ctx context.Context, requestMetadata map[string]any) []metrics.PodMetrics {
+	if m.LocateFunc != nil {
+		return m.LocateFunc(ctx, requestMetadata)
+	}
+	// Return copy to be safe
+	if m.Pods == nil {
+		return nil
+	}
+	result := make([]metrics.PodMetrics, len(m.Pods))
+	copy(result, m.Pods)
+	return result
+}
+
+// --- ManagedQueue Mock ---
+
 // MockManagedQueue is a high-fidelity, thread-safe mock of the `contracts.ManagedQueue` interface, designed
 // specifically for testing the concurrent `controller/internal.ShardProcessor`.
 //

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/flowcontrol/types/mocks/mocks.go

@@ -21,20 +21,19 @@ package mocks
 import (
 	"time"
 
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 )
 
-// MockFlowControlRequest provides a mock implementation of the `types.FlowControlRequest` interface.
+// MockFlowControlRequest provides a mock implementation of the types.FlowControlRequest interface.
 type MockFlowControlRequest struct {
-	FlowKeyV                    types.FlowKey
-	ByteSizeV                   uint64
-	InitialEffectiveTTLV        time.Duration
-	IDV                         string
-	CandidatePodsForSchedulingV []*metrics.FakePodMetrics
+	FlowKeyV             types.FlowKey
+	ByteSizeV            uint64
+	InitialEffectiveTTLV time.Duration
+	IDV                  string
+	MetadataV            map[string]any
 }
 
-// NewMockFlowControlRequest creates a new `MockFlowControlRequest` instance.
+// NewMockFlowControlRequest creates a new MockFlowControlRequest instance.
 func NewMockFlowControlRequest(
 	byteSize uint64,
 	id string,
@@ -44,21 +43,15 @@ func NewMockFlowControlRequest(
 		ByteSizeV: byteSize,
 		IDV:       id,
 		FlowKeyV:  key,
+		MetadataV: make(map[string]any),
 	}
 }
 
 func (m *MockFlowControlRequest) FlowKey() types.FlowKey             { return m.FlowKeyV }
 func (m *MockFlowControlRequest) ByteSize() uint64                   { return m.ByteSizeV }
 func (m *MockFlowControlRequest) InitialEffectiveTTL() time.Duration { return m.InitialEffectiveTTLV }
 func (m *MockFlowControlRequest) ID() string                         { return m.IDV }
-
-func (m *MockFlowControlRequest) CandidatePodsForScheduling() []metrics.PodMetrics {
-	pods := make([]metrics.PodMetrics, 0, len(m.CandidatePodsForSchedulingV))
-	for i, pod := range m.CandidatePodsForSchedulingV {
-		pods[i] = pod
-	}
-	return pods
-}
+func (m *MockFlowControlRequest) GetMetadata() map[string]any        { return m.MetadataV }
 
 var _ types.FlowControlRequest = &MockFlowControlRequest{}
 

pkg/epp/flowcontrol/types/request.go

@@ -18,8 +18,6 @@ package types
 
 import (
 	"time"
-
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 )
 
 // FlowControlRequest is the contract for an incoming request submitted to the `controller.FlowController`. It
@@ -45,15 +43,15 @@ type FlowControlRequest interface {
 	// applied.
 	InitialEffectiveTTL() time.Duration
 
-	// CandidatePodsForScheduling passes through a set of candidate pods a request may be admitted to.
-	// This is necessary for invoking `contracts.SaturationDetector.IsSaturated`, but it is otherwise unused in the Flow
-	// Control system.
-	CandidatePodsForScheduling() []metrics.PodMetrics
-
 	// ID returns an optional, user-facing unique identifier for this specific request. It is intended for logging,
 	// tracing, and observability. The `controller.FlowController` does not use this ID for dispatching decisions; it uses
 	// the internal, opaque `QueueItemHandle`.
 	ID() string
+
+	// GetMetadata returns the opaque metadata associated with the request (e.g., header-derived context, subset filters).
+	// This data is passed transparently to components like the contracts.PodLocator to resolve resources (candidate pods)
+	// lazily during the dispatch cycle.
+	GetMetadata() map[string]any
 }
 
 // QueueItemHandle is an opaque handle to an item that has been successfully added to a `framework.SafeQueue`. It acts