performance: add opt-in debounce for resource updates

api/v1alpha1/envoygateway_helpers.go

@@ -8,6 +8,7 @@ package v1alpha1
 import (
 	"net"
 	"strconv"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
@@ -87,6 +88,9 @@ func (e *EnvoyGateway) SetEnvoyGatewayDefaults() {
 	if e.XDSServer == nil {
 		e.XDSServer = DefaultXDSServer()
 	}
+	if e.Debounce == nil {
+		e.Debounce = DefaultDebounce()
+	}
 }
 
 // GetEnvoyGatewayAdmin returns the EnvoyGatewayAdmin of EnvoyGateway or a default EnvoyGatewayAdmin if unspecified.
@@ -239,6 +243,26 @@ func DefaultXDSServer() *XDSServer {
 	return &XDSServer{}
 }
 
+const (
+	// DefaultDebounceAfter is the default quiet period before a pending batch of
+	// resource changes is flushed.
+	DefaultDebounceAfter = 100 * time.Millisecond
+
+	// DefaultDebounceMax is the default upper bound on how long a resource change
+	// may be held before a flush is forced.
+	DefaultDebounceMax = 10 * time.Second
+)
+
+// DefaultDebounce returns a new Debounce with default configuration parameters.
+// Debouncing is disabled by default.
+func DefaultDebounce() *Debounce {
+	return &Debounce{
+		Enable: new(false),
+		After:  new(gwapiv1.Duration(DefaultDebounceAfter.String())),
+		Max:    new(gwapiv1.Duration(DefaultDebounceMax.String())),
+	}
+}
+
 // DefaultEnvoyGatewayLogging returns a new EnvoyGatewayLogging with default configuration parameters.
 func DefaultEnvoyGatewayLogging() *EnvoyGatewayLogging {
 	return &EnvoyGatewayLogging{

api/v1alpha1/envoygateway_helpers_test.go

@@ -9,6 +9,9 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/utils/ptr"
+	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 )
 
 func TestIsRunningOnKubernetes(t *testing.T) {
@@ -430,3 +433,17 @@ func TestGetKubernetesInfrastructureConfiguration(t *testing.T) {
 		})
 	}
 }
+
+func TestDebounceDefaultsToDisabled(t *testing.T) {
+	eg := &EnvoyGateway{}
+	eg.SetEnvoyGatewayDefaults()
+
+	require.NotNil(t, eg.Debounce)
+	require.False(t, ptr.Deref(eg.Debounce.Enable, true))
+	require.Equal(t, gwapiv1.Duration("100ms"), *eg.Debounce.After)
+	require.Equal(t, gwapiv1.Duration("10s"), *eg.Debounce.Max)
+
+	// The advertised defaults must match the constants the runner falls back to.
+	require.Equal(t, DefaultDebounceAfter.String(), string(*eg.Debounce.After))
+	require.Equal(t, DefaultDebounceMax.String(), string(*eg.Debounce.Max))
+}

api/v1alpha1/envoygateway_types.go

@@ -82,6 +82,14 @@ type EnvoyGatewaySpec struct {
 	// +optional
 	XDSServer *XDSServer `json:"xdsServer,omitempty"`
 
+	// Debounce defines how Envoy Gateway coalesces bursts of resource changes
+	// before retranslating and pushing new configuration to Envoy Proxy.
+	// If unspecified, debouncing is disabled and every change is translated and
+	// pushed on its own.
+	//
+	// +optional
+	Debounce *Debounce `json:"debounce,omitempty"`
+
 	// RateLimit defines the configuration associated with the Rate Limit service
 	// deployed by Envoy Gateway required to implement the Global Rate limiting
 	// functionality. The specific rate limit service used here is the reference
@@ -246,6 +254,46 @@ type XDSServer struct {
 	MaxReceiveMessageSize *resource.Quantity `json:"maxReceiveMessageSize,omitempty"`
 }
 
+// Debounce defines how Envoy Gateway coalesces bursts of resource changes before
+// retranslating and pushing new configuration to Envoy Proxy.
+//
+// Without debouncing, each resource change is translated and pushed on its own, so a
+// burst of changes costs a translation and a push per change even though only the
+// resulting state matters. That spends control plane CPU on work that is immediately
+// superseded, and makes the proxies apply configuration that will be replaced moments
+// later.
+//
+// Debouncing merges changes that arrive close together into a single translation, so
+// the cost of a burst approaches that of a single change. The tradeoff is that
+// propagation of a change may be delayed by up to Max.
+type Debounce struct {
+	// Enable turns on debouncing of resource changes.
+	//
+	// +optional
+	// +kubebuilder:default=false
+	Enable *bool `json:"enable,omitempty"`
+
+	// After is the quiet period. A pending batch of changes is flushed once no new
+	// change has arrived for this duration, so isolated changes still propagate
+	// promptly.
+	//
+	// If unspecified, defaults to 100ms.
+	//
+	// +optional
+	// +kubebuilder:default="100ms"
+	After *gwapiv1.Duration `json:"after,omitempty"`
+
+	// Max bounds how long a change may be held before a flush is forced. Under
+	// sustained churn the quiet period never elapses, so this caps how far behind
+	// the proxies' configuration can fall.
+	//
+	// Must be greater than or equal to After. If unspecified, defaults to 10s.
+	//
+	// +optional
+	// +kubebuilder:default="10s"
+	Max *gwapiv1.Duration `json:"max,omitempty"`
+}
+
 // LeaderElection defines the desired leader election settings.
 type LeaderElection struct {
 	// LeaseDuration defines the time non-leader contenders will wait before attempting to claim leadership.

api/v1alpha1/validation/envoygateway_validate.go

@@ -68,6 +68,10 @@ func ValidateEnvoyGateway(eg *egv1a1.EnvoyGateway) error {
 		return err
 	}
 
+	if err := validateEnvoyGatewayDebounce(eg.Debounce); err != nil {
+		return err
+	}
+
 	if eg.ExtensionAPIs != nil && eg.ExtensionAPIs.DisableLua != nil && *eg.ExtensionAPIs.DisableLua == eg.ExtensionAPIs.EnableLua {
 		return fmt.Errorf("disableLua and enableLua must not have the same value")
 	}
@@ -355,6 +359,42 @@ func validateEnvoyGatewayXDSServer(xdsServer *egv1a1.XDSServer) error {
 	return nil
 }
 
+func validateEnvoyGatewayDebounce(debounce *egv1a1.Debounce) error {
+	if debounce == nil {
+		return nil
+	}
+
+	after := egv1a1.DefaultDebounceAfter
+	if debounce.After != nil {
+		d, err := time.ParseDuration(string(*debounce.After))
+		if err != nil {
+			return fmt.Errorf("invalid debounce.after: %w", err)
+		}
+		if d <= 0 {
+			return fmt.Errorf("debounce.after must be greater than zero")
+		}
+		after = d
+	}
+
+	maxDelay := egv1a1.DefaultDebounceMax
+	if debounce.Max != nil {
+		d, err := time.ParseDuration(string(*debounce.Max))
+		if err != nil {
+			return fmt.Errorf("invalid debounce.max: %w", err)
+		}
+		if d <= 0 {
+			return fmt.Errorf("debounce.max must be greater than zero")
+		}
+		maxDelay = d
+	}
+
+	if maxDelay < after {
+		return fmt.Errorf("debounce.max (%s) must be greater than or equal to debounce.after (%s)", maxDelay, after)
+	}
+
+	return nil
+}
+
 func validateEnvoyGatewayOpenTelemetrySink(sink *egv1a1.EnvoyGatewayOpenTelemetrySink) error {
 	if sink.Protocol != egv1a1.GRPCProtocol && sink.Protocol != egv1a1.HTTPProtocol {
 		return fmt.Errorf("unsupported protocol %s for OpenTelemetry sink, only 'grpc' and 'http' are supported", sink.Protocol)

api/v1alpha1/validation/envoygateway_validate_test.go

@@ -1137,6 +1137,59 @@ func TestValidateEnvoyGatewayXDSServer(t *testing.T) {
 	})
 }
 
+func TestValidateEnvoyGatewayDebounce(t *testing.T) {
+	duration := func(s string) *gwapiv1.Duration {
+		d := gwapiv1.Duration(s)
+		return &d
+	}
+
+	t.Run("valid no overrides", func(t *testing.T) {
+		require.NoError(t, validateEnvoyGatewayDebounce(nil))
+		require.NoError(t, validateEnvoyGatewayDebounce(&egv1a1.Debounce{}))
+	})
+
+	t.Run("valid overrides", func(t *testing.T) {
+		d := &egv1a1.Debounce{After: duration("100ms"), Max: duration("10s")}
+		require.NoError(t, validateEnvoyGatewayDebounce(d))
+	})
+
+	t.Run("valid equal after and max", func(t *testing.T) {
+		d := &egv1a1.Debounce{After: duration("1s"), Max: duration("1s")}
+		require.NoError(t, validateEnvoyGatewayDebounce(d))
+	})
+
+	t.Run("invalid after duration", func(t *testing.T) {
+		d := &egv1a1.Debounce{After: duration("bad")}
+		require.Error(t, validateEnvoyGatewayDebounce(d))
+	})
+
+	t.Run("invalid max duration", func(t *testing.T) {
+		d := &egv1a1.Debounce{Max: duration("bad")}
+		require.Error(t, validateEnvoyGatewayDebounce(d))
+	})
+
+	t.Run("non positive after", func(t *testing.T) {
+		d := &egv1a1.Debounce{After: duration("0s")}
+		require.Error(t, validateEnvoyGatewayDebounce(d))
+	})
+
+	t.Run("non positive max", func(t *testing.T) {
+		d := &egv1a1.Debounce{Max: duration("-1s")}
+		require.Error(t, validateEnvoyGatewayDebounce(d))
+	})
+
+	t.Run("max shorter than after", func(t *testing.T) {
+		d := &egv1a1.Debounce{After: duration("5s"), Max: duration("1s")}
+		require.ErrorContains(t, validateEnvoyGatewayDebounce(d), "must be greater than or equal to")
+	})
+
+	t.Run("max shorter than defaulted after", func(t *testing.T) {
+		// After falls back to its 100ms default, so a 10ms max is invalid.
+		d := &egv1a1.Debounce{Max: duration("10ms")}
+		require.Error(t, validateEnvoyGatewayDebounce(d))
+	})
+}
+
 func TestDefaultEnvoyGatewayLoggingLevel(t *testing.T) {
 	type args struct {
 		component string

internal/gatewayapi/runner/runner.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 
@@ -160,14 +161,23 @@ func (r *Runner) Name() string {
 // Start starts the gateway-api translator runner
 func (r *Runner) Start(ctx context.Context) error {
 	r.Logger = r.Logger.WithName(r.Name()).WithValues("runner", r.Name())
+
+	debounce, err := r.debounceOptions()
+	if err != nil {
+		return err
+	}
+	if debounce != nil {
+		r.Logger.Info("debouncing resource updates", "after", debounce.After, "max", debounce.Max)
+	}
+
 	r.done.Go(func() {
 		r.startWasmCache(ctx)
 	})
 	// Do not call .Subscribe() inside Goroutine since it is supposed to be called from the same
 	// Goroutine where Close() is called.
 	c := r.ProviderResources.GatewayAPIResources.Subscribe(ctx)
 	r.done.Go(func() {
-		r.subscribeAndTranslate(c)
+		r.subscribeAndTranslate(c, debounce)
 	})
 	r.Logger.Info("started")
 	return nil
@@ -204,8 +214,50 @@ func (r *Runner) startWasmCache(ctx context.Context) {
 	r.wasmCache.Start(ctx)
 }
 
-func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *resource.ControllerResourcesContext]) {
-	message.HandleSubscription(
+// debounceOptions returns the debounce settings to apply to the resource
+// subscription, or nil when debouncing is disabled.
+func (r *Runner) debounceOptions() (*message.DebounceOptions, error) {
+	if r.EnvoyGateway == nil || r.EnvoyGateway.Debounce == nil ||
+		!ptr.Deref(r.EnvoyGateway.Debounce.Enable, false) {
+		return nil, nil
+	}
+
+	cfg := r.EnvoyGateway.Debounce
+	opts := &message.DebounceOptions{
+		After: egv1a1.DefaultDebounceAfter,
+		Max:   egv1a1.DefaultDebounceMax,
+	}
+
+	if cfg.After != nil {
+		d, err := time.ParseDuration(string(*cfg.After))
+		if err != nil {
+			return nil, fmt.Errorf("invalid debounce.after: %w", err)
+		}
+		if d <= 0 {
+			return nil, fmt.Errorf("debounce.after must be greater than zero")
+		}
+		opts.After = d
+	}
+
+	if cfg.Max != nil {
+		d, err := time.ParseDuration(string(*cfg.Max))
+		if err != nil {
+			return nil, fmt.Errorf("invalid debounce.max: %w", err)
+		}
+		if d <= 0 {
+			return nil, fmt.Errorf("debounce.max must be greater than zero")
+		}
+		opts.Max = d
+	}
+
+	return opts, nil
+}
+
+func (r *Runner) subscribeAndTranslate(
+	sub <-chan watchable.Snapshot[string, *resource.ControllerResourcesContext],
+	debounce *message.DebounceOptions,
+) {
+	message.HandleSubscriptionWithDebounce(
 		r.Logger,
 		message.Metadata{Runner: r.Name(), Message: message.ProviderResourcesMessageName}, sub,
 		func(update message.Update[string, *resource.ControllerResourcesContext], errChan chan error) {
@@ -636,6 +688,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
 			// Delete keys using mark and sweep
 			r.deleteKeys(keysToDelete)
 		},
+		debounce,
 	)
 	r.Logger.Info("shutting down")
 }