feat: Introduce the FlowController supervisor

This commit introduces the `FlowController`, a high-throughput, sharded
supervisor that orchestrates a pool of stateful `ShardProcessor`
workers. This new component is the central processing engine of the Flow
Control system, implementing a "supervisor-worker" pattern.

Key features of the `FlowController` include:

- Supervisor-Worker Architecture: Acts as a stateless supervisor,
  managing the lifecycle of stateful `ShardProcessor` workers. It
  includes a reconciliation loop to garbage-collect workers for stale
  shards.
- Flow-Aware Load Balancing: Implements a "Join-Shortest-Queue-by-Bytes"
  (JSQ-Bytes) algorithm to distribute incoming requests to the
  least-loaded worker, promoting emergent fairness.
- Synchronous API: Exposes a blocking `EnqueueAndWait` method, which
  simplifies client integration (e.g., with Envoy `ext_proc`) and
  provides direct backpressure.
- Lazy Worker Initialization: Workers are created on-demand when a shard
  shard first becomes active to conserve resources and reduce contention
  on the hot path.
- Configuration: A new `Config` object allows for tuning parameters like
  TTLs, buffer sizes, and reconciliation intervals.

Author: LukeAVanDrie

pkg/epp/flowcontrol/controller/config.go

@@ -0,0 +1,110 @@
+/*
+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 controller
+
+import (
+	"fmt"
+	"time"
+)
+
+const (
+	// defaultExpiryCleanupInterval is the default frequency for scanning for expired items.
+	defaultExpiryCleanupInterval = 1 * time.Second
+	// defaultProcessorReconciliationInterval is the default frequency for the supervisor loop.
+	defaultProcessorReconciliationInterval = 5 * time.Second
+	// defaultEnqueueChannelBufferSize is the default size of a worker's incoming request buffer.
+	defaultEnqueueChannelBufferSize = 100
+)
+
+// Config holds the configuration for the `FlowController`.
+type Config struct {
+	// DefaultRequestTTL is the default Time-To-Live applied to requests that do not
+	// specify their own TTL hint.
+	// Optional: If zero, no TTL is applied by default and we rely solely on request context cancellation.
+	DefaultRequestTTL time.Duration
+
+	// ExpiryCleanupInterval is the interval at which each shard processor scans its queues for expired items.
+	// Optional: Defaults to `defaultExpiryCleanupInterval` (1 second).
+	ExpiryCleanupInterval time.Duration
+
+	// ProcessorReconciliationInterval is the frequency at which the `FlowController`'s supervisor loop garbage collects
+	// stale workers.
+	// Optional: Defaults to `defaultProcessorReconciliationInterval` (5 seconds).
+	ProcessorReconciliationInterval time.Duration
+
+	// EnqueueChannelBufferSize is the size of the buffered channel that accepts incoming requests for each shard
+	// processor. This buffer acts as a shock absorber, decoupling the high-frequency distributor from the processor's
+	// serial execution loop and allowing the system to handle short bursts of traffic without blocking.
+	// Optional: Defaults to `defaultEnqueueChannelBufferSize` (100).
+	EnqueueChannelBufferSize int
+}
+
+// newConfig performs validation and initialization, returning a guaranteed-valid `Config` object.
+// This is the required constructor for creating a new configuration.
+// It does not mutate the input `cfg`.
+func newConfig(cfg Config) (*Config, error) {
+	newCfg := cfg.deepCopy()
+	if err := newCfg.validateAndApplyDefaults(); err != nil {
+		return nil, err
+	}
+	return newCfg, nil
+}
+
+// validateAndApplyDefaults checks the global configuration for validity and then mutates the receiver to populate any
+// empty fields with system defaults.
+func (c *Config) validateAndApplyDefaults() error {
+	// --- Validation ---
+	if c.DefaultRequestTTL < 0 {
+		return fmt.Errorf("DefaultRequestTTL cannot be negative, but got %v", c.DefaultRequestTTL)
+	}
+	if c.ExpiryCleanupInterval < 0 {
+		return fmt.Errorf("ExpiryCleanupInterval cannot be negative, but got %v", c.ExpiryCleanupInterval)
+	}
+	if c.ProcessorReconciliationInterval < 0 {
+		return fmt.Errorf("ProcessorReconciliationInterval cannot be negative, but got %v",
+			c.ProcessorReconciliationInterval)
+	}
+	if c.EnqueueChannelBufferSize < 0 {
+		return fmt.Errorf("EnqueueChannelBufferSize cannot be negative, but got %d", c.EnqueueChannelBufferSize)
+	}
+
+	// --- Defaulting ---
+	if c.ExpiryCleanupInterval == 0 {
+		c.ExpiryCleanupInterval = defaultExpiryCleanupInterval
+	}
+	if c.ProcessorReconciliationInterval == 0 {
+		c.ProcessorReconciliationInterval = defaultProcessorReconciliationInterval
+	}
+	if c.EnqueueChannelBufferSize == 0 {
+		c.EnqueueChannelBufferSize = defaultEnqueueChannelBufferSize
+	}
+	return nil
+}
+
+// deepCopy creates a deep copy of the `Config` object.
+func (c *Config) deepCopy() *Config {
+	if c == nil {
+		return nil
+	}
+	newCfg := &Config{
+		DefaultRequestTTL:               c.DefaultRequestTTL,
+		ExpiryCleanupInterval:           c.ExpiryCleanupInterval,
+		ProcessorReconciliationInterval: c.ProcessorReconciliationInterval,
+		EnqueueChannelBufferSize:        c.EnqueueChannelBufferSize,
+	}
+	return newCfg
+}

pkg/epp/flowcontrol/controller/config_test.go

@@ -0,0 +1,135 @@
+/*
+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,
+WITHOUTHOUT 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 controller
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewConfig(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name          string
+		input         Config
+		expectErr     bool
+		expectedCfg   Config
+		shouldDefault bool
+	}{
+		{
+			name: "ValidConfig_NoChanges",
+			input: Config{
+				DefaultRequestTTL:               10 * time.Second,
+				ExpiryCleanupInterval:           2 * time.Second,
+				ProcessorReconciliationInterval: 10 * time.Second,
+				EnqueueChannelBufferSize:        200,
+			},
+			expectErr: false,
+			expectedCfg: Config{
+				DefaultRequestTTL:               10 * time.Second,
+				ExpiryCleanupInterval:           2 * time.Second,
+				ProcessorReconciliationInterval: 10 * time.Second,
+				EnqueueChannelBufferSize:        200,
+			},
+		},
+		{
+			name:      "EmptyConfig_ShouldApplyDefaults",
+			input:     Config{},
+			expectErr: false,
+			expectedCfg: Config{
+				DefaultRequestTTL:               0,
+				ExpiryCleanupInterval:           defaultExpiryCleanupInterval,
+				ProcessorReconciliationInterval: defaultProcessorReconciliationInterval,
+				EnqueueChannelBufferSize:        defaultEnqueueChannelBufferSize,
+			},
+			shouldDefault: true,
+		},
+		{
+			name:      "NegativeDefaultRequestTTL_Invalid",
+			input:     Config{DefaultRequestTTL: -1},
+			expectErr: true,
+		},
+		{
+			name:      "NegativeExpiryCleanupInterval_Invalid",
+			input:     Config{ExpiryCleanupInterval: -1},
+			expectErr: true,
+		},
+		{
+			name:      "NegativeProcessorReconciliationInterval_Invalid",
+			input:     Config{ProcessorReconciliationInterval: -1},
+			expectErr: true,
+		},
+		{
+			name:      "NegativeEnqueueChannelBufferSize_Invalid",
+			input:     Config{EnqueueChannelBufferSize: -1},
+			expectErr: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			originalInput := tc.input.deepCopy()
+			validatedCfg, err := newConfig(tc.input)
+
+			if tc.expectErr {
+				require.Error(t, err, "expected an error but got nil")
+				assert.Nil(t, validatedCfg, "validatedCfg should be nil on error")
+			} else {
+				require.NoError(t, err, "expected no error but got: %v", err)
+				require.NotNil(t, validatedCfg, "validatedCfg should not be nil on success")
+				assert.Equal(t, tc.expectedCfg, *validatedCfg, "validatedCfg should match expected config")
+
+				// Ensure the original config is not mutated.
+				assert.Equal(t, *originalInput, tc.input, "input config should not be mutated")
+			}
+		})
+	}
+}
+
+func TestConfig_DeepCopy(t *testing.T) {
+	t.Parallel()
+
+	t.Run("ShouldReturnNil_ForNilReceiver", func(t *testing.T) {
+		t.Parallel()
+		var nilConfig *Config
+		assert.Nil(t, nilConfig.deepCopy(), "Deep copy of a nil config should be nil")
+	})
+
+	t.Run("ShouldCreateIdenticalButSeparateObject", func(t *testing.T) {
+		t.Parallel()
+		original := &Config{
+			DefaultRequestTTL:               1 * time.Second,
+			ExpiryCleanupInterval:           2 * time.Second,
+			ProcessorReconciliationInterval: 3 * time.Second,
+			EnqueueChannelBufferSize:        4,
+		}
+		clone := original.deepCopy()
+
+		require.NotSame(t, original, clone, "Clone should be a new object in memory")
+		assert.Equal(t, *original, *clone, "Cloned object should have identical values")
+
+		// Modify the clone and ensure the original is unchanged.
+		clone.DefaultRequestTTL = 99 * time.Second
+		assert.NotEqual(t, original.DefaultRequestTTL, clone.DefaultRequestTTL,
+			"Original should not be mutated after clone is changed")
+	})
+}