refactor: relocate SaturationDetector interface and its mock to the framework interface package.

LukeAVanDrie · LukeAVanDrie · commit 3970c2af9862 · 2026-03-17T16:56:05.000Z
diff --git a/apix/config/v1alpha1/endpointpickerconfig_types.go b/apix/config/v1alpha1/endpointpickerconfig_types.go
@@ -163,7 +163,6 @@ func (fg FeatureGates) String() string {
 	return "{" + result + "}"
 }
 
-
 // DataLayerConfig contains the configuration of the DataLayer feature
 type DataLayerConfig struct {
 	// +required
diff --git a/apix/config/v1alpha1/zz_generated.deepcopy.go b/apix/config/v1alpha1/zz_generated.deepcopy.go
diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go
@@ -93,7 +93,6 @@ const (
 	// control layer.
 	// DEPRECATION NOTICE - this env var will be removed in the next version as we switch to configuring the EPP using FeatureGates in the config file.
 	enableExperimentalFlowControlLayer = "ENABLE_EXPERIMENTAL_FLOW_CONTROL_LAYER"
-
 )
 
 var (
@@ -550,7 +549,6 @@ func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *conf
 	// The plugins will be executed in topologically sorted order to ensure that data is produced before it is consumed.
 	r.requestControlConfig.OrderPrepareDataPlugins(dag)
 
-
 	r.parser = handlers.NewParser(cfg.ParserConfig)
 	logger.Info("loaded configuration from file/text successfully")
 
@@ -569,7 +567,6 @@ func applyDeprecatedEnvFeatureGate(envVar, featureName, featureGate string, rawC
 	}
 }
 
-
 func (r *Runner) setupDataLayer(enableNewMetrics bool, cfg *datalayer.Config,
 	epf datalayer.EndpointFactory, mgr ctrl.Manager) error {
 	disallowedMetricsExtractor := ""
diff --git a/pkg/epp/config/loader/configloader_test.go b/pkg/epp/config/loader/configloader_test.go
@@ -526,7 +526,6 @@ func TestInstantiateAndConfigure(t *testing.T) {
 	}
 }
 
-
 // --- Helpers & Mocks ---
 
 func hasPluginType(handle fwkplugin.Handle, typeName string) bool {
diff --git a/pkg/epp/flowcontrol/benchmark/benchmark.go b/pkg/epp/flowcontrol/benchmark/benchmark.go
@@ -120,12 +120,13 @@ func (m benchMatrix) name() string {
 
 // testDetector exposes an API to manually release downstream capacity during a test run.
 type testDetector interface {
-	contracts.SaturationDetector
+	flowcontrol.SaturationDetector
 	Release()
 }
 
 // benchDetector models target saturation based strictly on active request counts.
 type benchDetector struct {
+	plugin.Plugin
 	concurrencyLimit atomic.Int64
 	// _ prevents false sharing between atomic counters on multicore CPU cache lines.
 	_        [56]byte
@@ -159,7 +160,9 @@ func (d *benchDetector) Saturation(ctx context.Context, candidates []metrics.Pod
 
 // alwaysSaturatedDetector simulates a permanently saturated downstream pool.
 // It is strictly used to evaluate garbage collection pathways and client abandonment scenarios.
-type alwaysSaturatedDetector struct{}
+type alwaysSaturatedDetector struct {
+	plugin.Plugin
+}
 
 // Release is a no-op for the permanently saturated mock.
 func (d *alwaysSaturatedDetector) Release() {}
diff --git a/pkg/epp/flowcontrol/contracts/dependencies.go b/pkg/epp/flowcontrol/contracts/dependencies.go
@@ -31,15 +31,3 @@ type PodLocator interface {
 	// Locate returns a list of pod metrics that match the criteria defined in the request metadata.
 	Locate(ctx context.Context, requestMetadata map[string]any) []fwkdl.Endpoint
 }
-
-// SaturationDetector defines the contract for a component that provides real-time load signals to the FlowController.
-type SaturationDetector interface {
-	// Saturation returns the saturation level of the pool
-	// - A value >= 1.0 indicates that the system is fully saturated.
-	// - A value < 1.0 indicates the ratio of used capacity to total capacity.
-	//
-	// FlowController consumes this signal to make dispatch decisions:
-	// - If Saturation() >= 1.0: Stop dispatching (enforce HoL blocking).
-	// - If Saturation() < 1.0: Continue dispatching.
-	Saturation(ctx context.Context, candidatePods []fwkdl.Endpoint) float64
-}
diff --git a/pkg/epp/flowcontrol/contracts/mocks/mocks.go b/pkg/epp/flowcontrol/contracts/mocks/mocks.go
@@ -108,18 +108,6 @@ var _ contracts.RegistryShard = &MockRegistryShard{}
 
 // --- Dependency Mocks ---
 
-// MockSaturationDetector is a simple "stub-style" mock for testing.
-type MockSaturationDetector struct {
-	SaturationFunc func(ctx context.Context, candidatePods []metrics.PodMetrics) float64
-}
-
-func (m *MockSaturationDetector) Saturation(ctx context.Context, candidatePods []metrics.PodMetrics) float64 {
-	if m.SaturationFunc != nil {
-		return m.SaturationFunc(ctx, candidatePods)
-	}
-	return 0.0
-}
-
 // 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 {
diff --git a/pkg/epp/flowcontrol/controller/controller.go b/pkg/epp/flowcontrol/controller/controller.go
@@ -62,7 +62,7 @@ type shardProcessor interface {
 type shardProcessorFactory func(
 	ctx context.Context,
 	shard contracts.RegistryShard,
-	saturationDetector contracts.SaturationDetector,
+	saturationDetector flowcontrol.SaturationDetector,
 	podLocator contracts.PodLocator,
 	clock clock.WithTicker,
 	cleanupSweepInterval time.Duration,
@@ -98,7 +98,7 @@ type FlowController struct {
 
 	config                *Config
 	registry              registryClient
-	saturationDetector    contracts.SaturationDetector
+	saturationDetector    flowcontrol.SaturationDetector
 	podLocator            contracts.PodLocator
 	clock                 clock.WithTicker
 	logger                logr.Logger
@@ -131,7 +131,7 @@ func NewFlowController(
 	poolName string,
 	config *Config,
 	registry contracts.FlowRegistry,
-	sd contracts.SaturationDetector,
+	sd flowcontrol.SaturationDetector,
 	podLocator contracts.PodLocator,
 	opts ...flowControllerOption,
 ) (*FlowController, error) {
@@ -148,7 +148,7 @@ func NewFlowController(
 	fc.shardProcessorFactory = func(
 		ctx context.Context,
 		shard contracts.RegistryShard,
-		saturationDetector contracts.SaturationDetector,
+		saturationDetector flowcontrol.SaturationDetector,
 		podLocator contracts.PodLocator,
 		clock clock.WithTicker,
 		cleanupSweepInterval time.Duration,
diff --git a/pkg/epp/flowcontrol/controller/controller_test.go b/pkg/epp/flowcontrol/controller/controller_test.go
@@ -115,7 +115,7 @@ func newUnitHarness(
 		opt(harnessOpts)
 	}
 
-	mockDetector := &mocks.MockSaturationDetector{}
+	mockDetector := &frameworkmocks.MockSaturationDetector{}
 	mockPodLocator := &mocks.MockPodLocator{}
 
 	mockProcessorFactory := &mockShardProcessorFactory{
@@ -154,7 +154,7 @@ func newUnitHarness(
 // validating the controller-processor interaction.
 func newIntegrationHarness(t *testing.T, ctx context.Context, cfg *Config, registry *mockRegistryClient) *testHarness {
 	t.Helper()
-	mockDetector := &mocks.MockSaturationDetector{}
+	mockDetector := &frameworkmocks.MockSaturationDetector{}
 	mockPodLocator := &mocks.MockPodLocator{}
 
 	// Align FakeClock with system time. See explanation in newUnitHarness.
@@ -276,7 +276,7 @@ type mockShardProcessorFactory struct {
 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,
+	_ flowcontrol.SaturationDetector,
 	_ contracts.PodLocator,
 	_ clock.WithTicker,
 	_ time.Duration,
@@ -1140,7 +1140,7 @@ func TestFlowController_WorkerManagement(t *testing.T) {
 		h.fc.shardProcessorFactory = func(
 			ctx context.Context, // The context created by getOrStartWorker for the potential new processor.
 			shard contracts.RegistryShard,
-			_ contracts.SaturationDetector,
+			_ flowcontrol.SaturationDetector,
 			_ contracts.PodLocator,
 			_ clock.WithTicker,
 			_ time.Duration,
diff --git a/pkg/epp/flowcontrol/controller/internal/processor.go b/pkg/epp/flowcontrol/controller/internal/processor.go
@@ -67,7 +67,7 @@ var ErrProcessorBusy = errors.New("shard processor is busy")
 type ShardProcessor struct {
 	poolName             string
 	shard                contracts.RegistryShard
-	saturationDetector   contracts.SaturationDetector
+	saturationDetector   flowcontrol.SaturationDetector
 	podLocator           contracts.PodLocator
 	clock                clock.WithTicker
 	cleanupSweepInterval time.Duration
@@ -90,7 +90,7 @@ func NewShardProcessor(
 	ctx context.Context,
 	poolName string,
 	shard contracts.RegistryShard,
-	saturationDetector contracts.SaturationDetector,
+	saturationDetector flowcontrol.SaturationDetector,
 	podLocator contracts.PodLocator,
 	clock clock.WithTicker,
 	cleanupSweepInterval time.Duration,
diff --git a/pkg/epp/flowcontrol/controller/internal/processor_test.go b/pkg/epp/flowcontrol/controller/internal/processor_test.go
@@ -73,7 +73,7 @@ type testHarness struct {
 	processor          *ShardProcessor
 	clock              *testclock.FakeClock
 	logger             logr.Logger
-	saturationDetector *mocks.MockSaturationDetector
+	saturationDetector *fwmocks.MockSaturationDetector
 	podLocator         *mocks.MockPodLocator
 
 	// --- Centralized Mock State ---
@@ -94,7 +94,7 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn
 		MockRegistryShard:  &mocks.MockRegistryShard{},
 		clock:              testclock.NewFakeClock(time.Now()),
 		logger:             logr.Discard(),
-		saturationDetector: &mocks.MockSaturationDetector{},
+		saturationDetector: &fwmocks.MockSaturationDetector{},
 		podLocator:         &mocks.MockPodLocator{Pods: []metrics.PodMetrics{&metrics.FakePodMetrics{}}},
 		startSignal:        make(chan struct{}),
 		queues:             make(map[flowcontrol.FlowKey]*mocks.MockManagedQueue),
diff --git a/pkg/epp/framework/interface/flowcontrol/mocks/mocks.go b/pkg/epp/framework/interface/flowcontrol/mocks/mocks.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"time"
 
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/datalayer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/scheduling"
@@ -277,3 +278,30 @@ func (m *MockFairnessPolicy) Pick(ctx context.Context, flowGroup flowcontrol.Pri
 }
 
 var _ flowcontrol.FairnessPolicy = &MockFairnessPolicy{}
+
+// MockSaturationDetector is a simple "stub-style" mock for testing.
+type MockSaturationDetector struct {
+	TypedNameV     plugin.TypedName
+	InitFunc       func(ctx context.Context, config any) error
+	SaturationFunc func(ctx context.Context, candidatePods []datalayer.Endpoint) float64
+}
+
+func (m *MockSaturationDetector) TypedName() plugin.TypedName {
+	return m.TypedNameV
+}
+
+func (m *MockSaturationDetector) Init(ctx context.Context, config any) error {
+	if m.InitFunc != nil {
+		return m.InitFunc(ctx, config)
+	}
+	return nil
+}
+
+func (m *MockSaturationDetector) Saturation(ctx context.Context, candidatePods []datalayer.Endpoint) float64 {
+	if m.SaturationFunc != nil {
+		return m.SaturationFunc(ctx, candidatePods)
+	}
+	return 0.0
+}
+
+var _ flowcontrol.SaturationDetector = &MockSaturationDetector{}
diff --git a/pkg/epp/framework/interface/flowcontrol/plugins.go b/pkg/epp/framework/interface/flowcontrol/plugins.go
@@ -41,7 +41,6 @@ type SaturationDetector interface {
 	Saturation(ctx context.Context, candidatePods []datalayer.Endpoint) float64
 }
 
-
 var (
 	// ErrIncompatiblePriorityType indicates that a FairnessPolicy attempted to compare items from two different flow
 	// queues whose ItemComparators have different ScoreType values, making a meaningful comparison impossible.
diff --git a/pkg/epp/requestcontrol/admission.go b/pkg/epp/requestcontrol/admission.go
@@ -64,7 +64,7 @@ type flowController interface {
 // rejectIfSheddableAndSaturated checks if a request should be immediately rejected.
 func rejectIfSheddableAndSaturated(
 	ctx context.Context,
-	sd contracts.SaturationDetector,
+	sd flowcontrol.SaturationDetector,
 	locator contracts.PodLocator,
 	reqCtx *handlers.RequestContext,
 	priority int,
@@ -89,13 +89,13 @@ func rejectIfSheddableAndSaturated(
 // It rejects sheddable requests (priority < 0) if the saturationDetector indicates that the system is currently
 // saturated. Non-sheddable requests always bypass the saturation check.
 type LegacyAdmissionController struct {
-	saturationDetector contracts.SaturationDetector
+	saturationDetector flowcontrol.SaturationDetector
 	podLocator         contracts.PodLocator
 }
 
 // NewLegacyAdmissionController creates a new LegacyAdmissionController.
 func NewLegacyAdmissionController(
-	sd contracts.SaturationDetector,
+	sd flowcontrol.SaturationDetector,
 	pl contracts.PodLocator,
 ) *LegacyAdmissionController {
 	return &LegacyAdmissionController{
diff --git a/pkg/epp/requestcontrol/admission_test.go b/pkg/epp/requestcontrol/admission_test.go
@@ -29,7 +29,8 @@ import (
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts/mocks"
 	fctypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
+	flowcontrol "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
+	fwmocks "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol/mocks"
 	schedulingtypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/scheduling"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
 )
@@ -110,7 +111,7 @@ func TestLegacyAdmissionController_Admit(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
-			mockDetector := &mocks.MockSaturationDetector{
+			mockDetector := &fwmocks.MockSaturationDetector{
 				SaturationFunc: func(context.Context, []backendmetrics.PodMetrics) float64 {
 					if tc.isSaturated {
 						return 1.0
diff --git a/pkg/epp/server/runserver.go b/pkg/epp/server/runserver.go
@@ -40,10 +40,10 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/controller"
 	datalayerlogger "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer/logger"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+	fwkfc "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
 	fwkrh "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/requesthandling"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/requestcontrol"
-	fwkfc "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
 )
 
 // ExtProcServerRunner provides methods to manage an external process server.

Original file line number	Diff line number	Diff line change
`@@ -163,7 +163,6 @@ func (fg FeatureGates) String() string {`
`163`	`163`	`return "{" + result + "}"`
`164`	`164`	`}`
`165`	`165`
`166`		`-`
`167`	`166`	`// DataLayerConfig contains the configuration of the DataLayer feature`
`168`	`167`	`type DataLayerConfig struct {`
`169`	`168`	`// +required`
Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,6 @@ const (`
`93`	`93`	`// control layer.`
`94`	`94`	`// DEPRECATION NOTICE - this env var will be removed in the next version as we switch to configuring the EPP using FeatureGates in the config file.`
`95`	`95`	`enableExperimentalFlowControlLayer = "ENABLE_EXPERIMENTAL_FLOW_CONTROL_LAYER"`
`96`		`-`
`97`	`96`	`)`
`98`	`97`
`99`	`98`	`var (`
`@@ -550,7 +549,6 @@ func (r Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig conf`
`550`	`549`	`// The plugins will be executed in topologically sorted order to ensure that data is produced before it is consumed.`
`551`	`550`	`r.requestControlConfig.OrderPrepareDataPlugins(dag)`
`552`	`551`
`553`		`-`
`554`	`552`	`r.parser = handlers.NewParser(cfg.ParserConfig)`
`555`	`553`	`logger.Info("loaded configuration from file/text successfully")`
`556`	`554`
`@@ -569,7 +567,6 @@ func applyDeprecatedEnvFeatureGate(envVar, featureName, featureGate string, rawC`
`569`	`567`	`}`
`570`	`568`	`}`
`571`	`569`
`572`		`-`
`573`	`570`	`func (r Runner) setupDataLayer(enableNewMetrics bool, cfg datalayer.Config,`
`574`	`571`	`epf datalayer.EndpointFactory, mgr ctrl.Manager) error {`
`575`	`572`	`disallowedMetricsExtractor := ""`
Original file line number	Diff line number	Diff line change
`@@ -526,7 +526,6 @@ func TestInstantiateAndConfigure(t *testing.T) {`
`526`	`526`	`}`
`527`	`527`	`}`
`528`	`528`
`529`		`-`
`530`	`529`	`// --- Helpers & Mocks ---`
`531`	`530`
`532`	`531`	`func hasPluginType(handle fwkplugin.Handle, typeName string) bool {`