refactor: rename PodLocator to EndpointCandidates

Rename PodLocator interface and related types to EndpointCandidates to better reflect that the component resolves endpoint candidates rather than pods specifically. Update all implementations, mocks, tests, and documentation accordingly.

Author: cyclinder

cmd/epp/runner/runner.go

@@ -329,10 +329,10 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 
 	// --- Admission Control Initialization ---
 	var admissionController requestcontrol.AdmissionController
-	var locator contracts.PodLocator
-	locator = requestcontrol.NewDatastorePodLocator(ds, requestcontrol.WithDisableEndpointSubsetFilter(opts.DisableEndpointSubsetFilter))
+	var endpointCandidates contracts.EndpointCandidates
+	endpointCandidates = requestcontrol.NewDatastoreEndpointCandidates(ds, requestcontrol.WithDisableEndpointSubsetFilter(opts.DisableEndpointSubsetFilter))
 	if r.featureGates[flowcontrol.FeatureGate] {
-		locator = requestcontrol.NewCachedPodLocator(ctx, locator, time.Millisecond*50)
+		endpointCandidates = requestcontrol.NewCachedEndpointCandidates(ctx, endpointCandidates, time.Millisecond*50)
 		setupLog.Info("Initializing experimental Flow Control layer")
 		registry, err := fcregistry.NewFlowRegistry(eppConfig.FlowControlConfig.Registry, setupLog)
 		if err != nil {
@@ -343,7 +343,7 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 			opts.PoolName,
 			eppConfig.FlowControlConfig.Controller,
 			registry, eppConfig.SaturationDetector,
-			locator, eppConfig.FlowControlConfig.UsageLimitPolicy,
+			endpointCandidates, eppConfig.FlowControlConfig.UsageLimitPolicy,
 		)
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to initialize Flow Controller: %w", err)
@@ -352,12 +352,11 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 		admissionController = requestcontrol.NewFlowControlAdmissionController(fc, opts.PoolName)
 	} else {
 		setupLog.Info("Experimental Flow Control layer is disabled, using legacy admission control")
-		admissionController = requestcontrol.NewLegacyAdmissionController(eppConfig.SaturationDetector, locator)
+		admissionController = requestcontrol.NewLegacyAdmissionController(eppConfig.SaturationDetector, endpointCandidates)
 	}
 
-	director := requestcontrol.NewDirectorWithConfig(ds, scheduler, admissionController, r.parser, locator, r.requestControlConfig)
+	director := requestcontrol.NewDirectorWithConfig(ds, scheduler, admissionController, r.parser, endpointCandidates, r.requestControlConfig)
 
-	// --- Setup ExtProc Server Runner ---
 	serverRunner := &runserver.ExtProcServerRunner{
 		GrpcPort:                         opts.GRPCPort,
 		GKNN:                             *gknn,

pkg/epp/flowcontrol/benchmark/benchmark.go

@@ -276,7 +276,7 @@ func setupBenchmarkHarness(
 		}
 	}
 
-	fc, err := controller.NewFlowController(ctx, "benchmark", cfg, reg, detector, &mocks.MockPodLocator{}, usagelimits.DefaultPolicy())
+	fc, err := controller.NewFlowController(ctx, "benchmark", cfg, reg, detector, &mocks.MockEndpointCandidates{}, usagelimits.DefaultPolicy())
 	if err != nil {
 		b.Fatalf("Failed to init FlowController: %v", err)
 	}

pkg/epp/flowcontrol/contracts/dependencies.go

@@ -22,12 +22,12 @@ import (
 	fwkdl "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/datalayer"
 )
 
-// PodLocator defines the contract for a component that resolves the set of candidate pods for a request based on its
-// metadata (e.g., subsetting).
+// EndpointCandidates defines the contract for a component that resolves the set of candidate endpoints for a request
+// based on its metadata (e.g., subsetting).
 //
-// This interface allows the Flow Controller to fetch a fresh list of pods dynamically during the dispatch cycle,
-// enabling support for "Scale-from-Zero" scenarios where pods may not exist when the request is first enqueued.
-type PodLocator interface {
-	// Locate returns a list of pod metrics that match the criteria defined in the request metadata.
+// This interface allows the Flow Controller to fetch a fresh list of candidates dynamically during the dispatch cycle,
+// enabling support for "Scale-from-Zero" scenarios where endpoints may not exist when the request is first enqueued.
+type EndpointCandidates interface {
+	// Locate returns a list of endpoint candidate metrics that match the criteria defined in the request metadata.
 	Locate(ctx context.Context, requestMetadata map[string]any) []fwkdl.Endpoint
 }

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

@@ -108,25 +108,38 @@ var _ contracts.RegistryShard = &MockRegistryShard{}
 
 // --- Dependency Mocks ---
 
-// 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 {
+// MockSaturationDetector is a simple "stub-style" mock for testing.
+type MockSaturationDetector struct {
+	SaturationFunc func(ctx context.Context, candidatePods []fwkdl.Endpoint) float64
+}
+
+func (m *MockSaturationDetector) Saturation(ctx context.Context, candidatePods []fwkdl.Endpoint) float64 {
+	if m.SaturationFunc != nil {
+		return m.SaturationFunc(ctx, candidatePods)
+	}
+	return 0.0
+}
+
+// MockEndpointCandidates provides a mock implementation of the contracts.EndpointCandidates interface.
+// It allows tests to control the exact set of endpoint candidates returned for a given request.
+type MockEndpointCandidates struct {
 	// LocateFunc allows injecting custom logic.
 	LocateFunc func(ctx context.Context, requestMetadata map[string]any) []fwkdl.Endpoint
-	// Pods is a static return value used if LocateFunc is nil.
-	Pods []fwkdl.Endpoint
+	// Candidates is a static return value used if LocateFunc is nil.
+	Candidates []fwkdl.Endpoint
 }
 
-func (m *MockPodLocator) Locate(ctx context.Context, requestMetadata map[string]any) []fwkdl.Endpoint {
+func (m *MockEndpointCandidates) Locate(ctx context.Context, requestMetadata map[string]any) []fwkdl.Endpoint {
 	if m.LocateFunc != nil {
 		return m.LocateFunc(ctx, requestMetadata)
 	}
 	// Return copy to be safe
-	if m.Pods == nil {
+	if m.Candidates == nil {
 		return nil
 	}
-	result := make([]fwkdl.Endpoint, len(m.Pods))
-	copy(result, m.Pods)
+
+	result := make([]fwkdl.Endpoint, len(m.Candidates))
+	copy(result, m.Candidates)
 	return result
 }
 

pkg/epp/flowcontrol/controller/controller.go

@@ -63,7 +63,7 @@ type shardProcessorFactory func(
 	ctx context.Context,
 	shard contracts.RegistryShard,
 	saturationDetector flowcontrol.SaturationDetector,
-	podLocator contracts.PodLocator,
+	endpointCandidates contracts.EndpointCandidates,
 	usageLimitPolicy flowcontrol.UsageLimitPolicy,
 	clock clock.WithTicker,
 	cleanupSweepInterval time.Duration,
@@ -100,7 +100,7 @@ type FlowController struct {
 	config                *Config
 	registry              registryClient
 	saturationDetector    flowcontrol.SaturationDetector
-	podLocator            contracts.PodLocator
+	endpointCandidates    contracts.EndpointCandidates
 	usageLimitPolicy      flowcontrol.UsageLimitPolicy
 	clock                 clock.WithTicker
 	logger                logr.Logger
@@ -134,15 +134,15 @@ func NewFlowController(
 	config *Config,
 	registry contracts.FlowRegistry,
 	sd flowcontrol.SaturationDetector,
-	podLocator contracts.PodLocator,
+	endpointCandidates contracts.EndpointCandidates,
 	usageLimitPolicy flowcontrol.UsageLimitPolicy,
 	opts ...flowControllerOption,
 ) (*FlowController, error) {
 	fc := &FlowController{
 		config:             config,
 		registry:           registry,
 		saturationDetector: sd,
-		podLocator:         podLocator,
+		endpointCandidates: endpointCandidates,
 		usageLimitPolicy:   usageLimitPolicy,
 		clock:              clock.RealClock{},
 		logger:             log.FromContext(ctx).WithName("flow-controller"),
@@ -153,7 +153,7 @@ func NewFlowController(
 		ctx context.Context,
 		shard contracts.RegistryShard,
 		saturationDetector flowcontrol.SaturationDetector,
-		podLocator contracts.PodLocator,
+		endpointCandidates contracts.EndpointCandidates,
 		usageLimitPolicy flowcontrol.UsageLimitPolicy,
 		clock clock.WithTicker,
 		cleanupSweepInterval time.Duration,
@@ -165,12 +165,13 @@ func NewFlowController(
 			poolName,
 			shard,
 			saturationDetector,
-			podLocator,
+			endpointCandidates,
 			usageLimitPolicy,
 			clock,
 			cleanupSweepInterval,
 			enqueueChannelBufferSize,
-			logger)
+			logger,
+		)
 	}
 
 	for _, opt := range opts {
@@ -492,7 +493,7 @@ func (fc *FlowController) getOrStartWorker(shard contracts.RegistryShard) *manag
 		processorCtx,
 		shard,
 		fc.saturationDetector,
-		fc.podLocator,
+		fc.endpointCandidates,
 		fc.usageLimitPolicy,
 		fc.clock,
 		fc.config.ExpiryCleanupInterval,

pkg/epp/flowcontrol/controller/controller_test.go

@@ -126,7 +126,7 @@ func newUnitHarness(
 	}
 
 	mockDetector := &mockSaturationDetector{}
-	mockPodLocator := &mocks.MockPodLocator{}
+	mockEndpointCandidates := &mocks.MockEndpointCandidates{}
 
 	mockProcessorFactory := &mockShardProcessorFactory{
 		processors: make(map[string]*mockShardProcessor),
@@ -144,7 +144,7 @@ func newUnitHarness(
 		withClock(harnessOpts.clock),
 		withShardProcessorFactory(mockProcessorFactory.new),
 	}
-	fc, err := NewFlowController(ctx, "test-pool", cfg, registry, mockDetector, mockPodLocator, usageLimitPolicy, fcOpts...)
+	fc, err := NewFlowController(ctx, "test-pool", cfg, registry, mockDetector, mockEndpointCandidates, usageLimitPolicy, fcOpts...)
 	require.NoError(t, err, "failed to create FlowController for unit test harness")
 
 	h := &testHarness{
@@ -167,7 +167,7 @@ func newUnitHarness(
 func newIntegrationHarness(t *testing.T, ctx context.Context, cfg *Config, registry *mockRegistryClient) *testHarness {
 	t.Helper()
 	mockDetector := &mockSaturationDetector{}
-	mockPodLocator := &mocks.MockPodLocator{}
+	mockEndpointCandidates := &mocks.MockEndpointCandidates{}
 	usageLimitPolicy := usagelimits.DefaultPolicy()
 
 	// Align FakeClock with system time. See explanation in newUnitHarness.
@@ -180,7 +180,7 @@ func newIntegrationHarness(t *testing.T, ctx context.Context, cfg *Config, regis
 		withRegistryClient(registry),
 		withClock(mockClock),
 	}
-	fc, err := NewFlowController(ctx, "test-pool", cfg, registry, mockDetector, mockPodLocator, usageLimitPolicy, opts...)
+	fc, err := NewFlowController(ctx, "test-pool", cfg, registry, mockDetector, mockEndpointCandidates, opts...)
 	require.NoError(t, err, "failed to create FlowController for integration test harness")
 
 	h := &testHarness{
@@ -290,7 +290,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,
 	_ flowcontrol.SaturationDetector,
-	_ contracts.PodLocator,
+	_ contracts.EndpointCandidates,
 	_ flowcontrol.UsageLimitPolicy,
 	_ clock.WithTicker,
 	_ time.Duration,
@@ -1155,7 +1155,7 @@ func TestFlowController_WorkerManagement(t *testing.T) {
 			ctx context.Context, // The context created by getOrStartWorker for the potential new processor.
 			shard contracts.RegistryShard,
 			_ flowcontrol.SaturationDetector,
-			_ contracts.PodLocator,
+			_ contracts.EndpointCandidates,
 			_ flowcontrol.UsageLimitPolicy,
 			_ clock.WithTicker,
 			_ time.Duration,

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

@@ -68,7 +68,7 @@ type ShardProcessor struct {
 	poolName             string
 	shard                contracts.RegistryShard
 	saturationDetector   flowcontrol.SaturationDetector
-	podLocator           contracts.PodLocator
+	endpointCandidates   contracts.EndpointCandidates
 	usageLimitPolicy     flowcontrol.UsageLimitPolicy
 	clock                clock.WithTicker
 	cleanupSweepInterval time.Duration
@@ -92,7 +92,7 @@ func NewShardProcessor(
 	poolName string,
 	shard contracts.RegistryShard,
 	saturationDetector flowcontrol.SaturationDetector,
-	podLocator contracts.PodLocator,
+	endpointCandidates contracts.EndpointCandidates,
 	usageLimitPolicy flowcontrol.UsageLimitPolicy,
 	clock clock.WithTicker,
 	cleanupSweepInterval time.Duration,
@@ -103,7 +103,7 @@ func NewShardProcessor(
 		shard:                shard,
 		poolName:             poolName,
 		saturationDetector:   saturationDetector,
-		podLocator:           podLocator,
+		endpointCandidates:   endpointCandidates,
 		usageLimitPolicy:     usageLimitPolicy,
 		clock:                clock,
 		cleanupSweepInterval: cleanupSweepInterval,
@@ -317,7 +317,7 @@ func (sp *ShardProcessor) dispatchCycle(ctx context.Context) bool {
 		metrics.RecordFlowControlDispatchCycleDuration(time.Since(dispatchCycleStart))
 	}()
 
-	pool := sp.podLocator.Locate(ctx, nil)
+	pool := sp.endpointCandidates.Locate(ctx, nil)
 	saturation := sp.saturationDetector.Saturation(ctx, pool)
 
 	// Record pool saturation metric

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

@@ -88,7 +88,7 @@ type testHarness struct {
 	clock              *testclock.FakeClock
 	logger             logr.Logger
 	saturationDetector *mockSaturationDetector
-	podLocator         *mocks.MockPodLocator
+	endpointCandidates *mocks.MockEndpointCandidates
 
 	// --- Centralized Mock State ---
 	// The harness's mutex protects the single source of truth for all mock state.
@@ -109,7 +109,7 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn
 		clock:              testclock.NewFakeClock(time.Now()),
 		logger:             logr.Discard(),
 		saturationDetector: &mockSaturationDetector{},
-		podLocator:         &mocks.MockPodLocator{Pods: []fwkdl.Endpoint{&metrics.FakePodMetrics{}}},
+		endpointCandidates: &mocks.MockEndpointCandidates{Candidates: []fwkdl.Endpoint{&metrics.FakePodMetrics{}}},
 		startSignal:        make(chan struct{}),
 		queues:             make(map[flowcontrol.FlowKey]*mocks.MockManagedQueue),
 		priorityFlows:      make(map[int][]flowcontrol.FlowKey),
@@ -137,7 +137,7 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn
 		"test-pool",
 		h,
 		h.saturationDetector,
-		h.podLocator,
+		h.endpointCandidates,
 		usagelimits.DefaultPolicy(),
 		h.clock,
 		expiryCleanupInterval,

pkg/epp/framework/interface/flowcontrol/request.go

@@ -57,7 +57,7 @@ type FlowControlRequest interface {
 	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)
+	// This data is passed transparently to components like contracts.EndpointCandidates to resolve resources (endpoint candidates)
 	// lazily during the dispatch cycle.
 	GetMetadata() map[string]any
 

pkg/epp/framework/plugins/requesthandling/parsers/openai/openai.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 
 	fwkplugin "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
@@ -90,7 +91,7 @@ func (p *OpenAIParser) WithName(name string) *OpenAIParser {
 func (p *OpenAIParser) ParseRequest(ctx context.Context, body []byte, headers map[string]string) (*scheduling.LLMRequestBody, error) {
 	bodyMap := make(map[string]any)
 	if err := json.Unmarshal(body, &bodyMap); err != nil {
-		return nil, errors.New("error unmarshaling request bodyMap")
+		return nil, fmt.Errorf("error unmarshaling request bodyMap: %w", err)
 	}
 	extractedBody, err := extractRequestBody(body, headers)
 	if err != nil {