[WIP] fix: stop endpoint events from rebuilding Gateway proxies

changelog/v1.22.2/stop-endpoint-proxy-recompute.yaml

@@ -0,0 +1,9 @@
+changelog:
+  - type: FIX
+    issueLink: https://github.com/solo-io/solo-projects/issues/8013
+    resolvesIssue: false
+    description: >-
+      Prevent Pod and EndpointSlice events from triggering full Kubernetes
+      Gateway proxy retranslation. Endpoint changes continue to update xDS
+      through the KRT endpoint pipeline, reducing unnecessary CPU and memory
+      churn in event-heavy clusters.

projects/gateway2/controller/controller.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
-	discoveryv1 "k8s.io/api/discovery/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -93,8 +92,8 @@ func NewBaseGatewayController(ctx context.Context, cfg GatewayConfig) error {
 		controllerBuilder.watchVirtualHostOptions,
 		controllerBuilder.watchUpstreams,
 		controllerBuilder.watchServices,
-		controllerBuilder.watchEndpointSlices,
-		controllerBuilder.watchPods,
+		// Pod and EndpointSlice updates flow through the KRT endpoint collections.
+		// Watching them here would unnecessarily rebuild every Gateway proxy.
 		controllerBuilder.watchSecrets,
 		controllerBuilder.addIndexes,
 		controllerBuilder.addHttpLisOptIndexes,
@@ -472,18 +471,6 @@ func (c *controllerBuilder) watchDirectResponses(_ context.Context) error {
 		Complete(reconcile.Func(c.reconciler.ReconcileDirectResponses))
 }
 
-func (c *controllerBuilder) watchPods(ctx context.Context) error {
-	return ctrl.NewControllerManagedBy(c.cfg.Mgr).
-		For(&corev1.Pod{}).
-		Complete(reconcile.Func(c.reconciler.ReconcilePods))
-}
-
-func (c *controllerBuilder) watchEndpointSlices(ctx context.Context) error {
-	return ctrl.NewControllerManagedBy(c.cfg.Mgr).
-		For(&discoveryv1.EndpointSlice{}).
-		Complete(reconcile.Func(c.reconciler.ReconcileEndpointSlices))
-}
-
 func (c *controllerBuilder) watchSecrets(ctx context.Context) error {
 	return ctrl.NewControllerManagedBy(c.cfg.Mgr).
 		For(&corev1.Secret{}).
@@ -545,18 +532,6 @@ func (r *controllerReconciler) ReconcileServices(ctx context.Context, req ctrl.R
 	return ctrl.Result{}, nil
 }
 
-func (r *controllerReconciler) ReconcilePods(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	// eventually reconcile only effected listeners etc
-	r.kick(ctx)
-	return ctrl.Result{}, nil
-}
-
-func (r *controllerReconciler) ReconcileEndpointSlices(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	// eventually reconcile only effected listeners etc
-	r.kick(ctx)
-	return ctrl.Result{}, nil
-}
-
 func (r *controllerReconciler) ReconcileSecrets(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	// eventually reconcile only effected listeners etc
 	r.kick(ctx)

projects/gateway2/controller/controller_suite_test.go

@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
 
 	"sigs.k8s.io/controller-runtime/pkg/config"
@@ -45,6 +46,7 @@ var (
 	cancel    context.CancelFunc
 
 	kubeconfig string
+	kickCount  atomic.Int64
 
 	gwClasses = sets.New(gatewayClassName, altGatewayClassName)
 )
@@ -124,7 +126,7 @@ var _ = BeforeSuite(func() {
 		ControllerName: gatewayControllerName,
 		GWClasses:      gwClasses,
 		AutoProvision:  true,
-		Kick:           func(ctx context.Context) { return },
+		Kick:           func(ctx context.Context) { kickCount.Add(1) },
 		Extensions:     exts,
 	}
 	err = controller.NewBaseGatewayController(ctx, cfg)

projects/gateway2/controller/recompute_test.go

@@ -0,0 +1,71 @@
+package controller_test
+
+import (
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+)
+
+var _ = Describe("Proxy recomputation", Ordered, func() {
+	const timeout = 10 * time.Second
+
+	It("does not kick proxy translation for endpoint-only events", func() {
+		suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+		serviceName := "kick-control-" + suffix
+
+		beforeService := kickCount.Load()
+		Expect(k8sClient.Create(ctx, &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      serviceName,
+				Namespace: "default",
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{{
+					Name: "http",
+					Port: 8080,
+				}},
+			},
+		})).To(Succeed())
+		Eventually(kickCount.Load, timeout).Should(BeNumerically(">", beforeService),
+			"the positive control should prove that Kick is wired to the controller")
+
+		baseline := kickCount.Load()
+		Expect(k8sClient.Create(ctx, &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "no-kick-pod-" + suffix,
+				Namespace: "default",
+			},
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{{
+					Name:  "test",
+					Image: "test",
+				}},
+			},
+		})).To(Succeed())
+		Expect(k8sClient.Create(ctx, &discoveryv1.EndpointSlice{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "no-kick-slice-" + suffix,
+				Namespace: "default",
+				Labels: map[string]string{
+					discoveryv1.LabelServiceName: serviceName,
+				},
+			},
+			AddressType: discoveryv1.AddressTypeIPv4,
+			Endpoints: []discoveryv1.Endpoint{{
+				Addresses: []string{"10.0.0.1"},
+			}},
+			Ports: []discoveryv1.EndpointPort{{
+				Name: ptr.To("http"),
+				Port: ptr.To(int32(8080)),
+			}},
+		})).To(Succeed())
+
+		Consistently(kickCount.Load, time.Second).Should(Equal(baseline))
+	})
+})

projects/gateway2/krtcollections/endpoints_test.go

@@ -3,6 +3,7 @@ package krtcollections
 import (
 	"context"
 	"testing"
+	"time"
 
 	envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
@@ -21,6 +22,107 @@ import (
 	"k8s.io/utils/ptr"
 )
 
+func TestEndpointInputsRecomputeEndpoints(t *testing.T) {
+	g := gomega.NewWithT(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	upstream := UpstreamWrapper{
+		Inner: &gloov1.Upstream{
+			Metadata: &core.Metadata{Name: "upstream", Namespace: "ns"},
+			UpstreamType: &gloov1.Upstream_Kube{
+				Kube: &kubernetes.UpstreamSpec{
+					ServiceName:      "svc",
+					ServiceNamespace: "ns",
+					ServicePort:      8080,
+				},
+			},
+		},
+	}
+	service := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: "svc", Namespace: "ns"},
+		Spec: corev1.ServiceSpec{
+			Ports: []corev1.ServicePort{{
+				Name: "http",
+				Port: 8080,
+			}},
+		},
+	}
+	endpointSlice := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "svc-abcde",
+			Namespace: "ns",
+			Labels: map[string]string{
+				discoveryv1.LabelServiceName: "svc",
+			},
+		},
+		AddressType: discoveryv1.AddressTypeIPv4,
+		Endpoints: []discoveryv1.Endpoint{{
+			Addresses: []string{"10.0.0.1"},
+			TargetRef: &corev1.ObjectReference{
+				Kind:      "Pod",
+				Name:      "pod",
+				Namespace: "ns",
+			},
+		}},
+		Ports: []discoveryv1.EndpointPort{{
+			Name: ptr.To("http"),
+			Port: ptr.To(int32(8080)),
+		}},
+	}
+	pod := LocalityPod{
+		Named:           krt.Named{Name: "pod", Namespace: "ns"},
+		AugmentedLabels: map[string]string{"version": "v1"},
+		Addresses:       []string{"10.0.0.1"},
+	}
+
+	upstreams := krt.NewStaticCollection(nil, []UpstreamWrapper{upstream})
+	services := krt.NewStaticCollection(nil, []*corev1.Service{service})
+	endpointSlices := krt.NewStaticCollection(nil, []*discoveryv1.EndpointSlice{endpointSlice})
+	pods := krt.NewStaticCollection(nil, []LocalityPod{pod})
+	endpointSettings := krt.NewStatic(&EndpointsSettings{}, true)
+	endpointSlicesByService := krt.NewIndex(endpointSlices, "TestRecomputeEndpointSlicesByService", func(es *discoveryv1.EndpointSlice) []types.NamespacedName {
+		return []types.NamespacedName{{
+			Namespace: es.Namespace,
+			Name:      es.Labels[discoveryv1.LabelServiceName],
+		}}
+	})
+
+	endpoints := NewGlooK8sEndpoints(ctx, EndpointsInputs{
+		Upstreams:               upstreams,
+		EndpointSlices:          endpointSlices,
+		EndpointSlicesByService: endpointSlicesByService,
+		Pods:                    pods,
+		EndpointsSettings:       endpointSettings,
+		Services:                services,
+	})
+	g.Eventually(endpoints.List, time.Second).Should(HaveLen(1))
+	initialHash := endpoints.List()[0].LbEpsEqualityHash
+
+	updatedEndpointSlice := endpointSlice.DeepCopy()
+	updatedEndpointSlice.Endpoints[0].Addresses = []string{"10.0.0.2"}
+	endpointSlices.Reset([]*discoveryv1.EndpointSlice{updatedEndpointSlice})
+	g.Eventually(func() []uint64 {
+		current := endpoints.List()
+		if len(current) != 1 {
+			return nil
+		}
+		return []uint64{current[0].LbEpsEqualityHash}
+	}, time.Second).Should(SatisfyAll(HaveLen(1), Not(ContainElement(initialHash))))
+	endpointSliceHash := endpoints.List()[0].LbEpsEqualityHash
+
+	updatedPod := pod
+	updatedPod.AugmentedLabels = map[string]string{"version": "v2"}
+	pods.Reset([]LocalityPod{updatedPod})
+	g.Eventually(func() []uint64 {
+		current := endpoints.List()
+		if len(current) != 1 {
+			return nil
+		}
+		return []uint64{current[0].LbEpsEqualityHash}
+	}, time.Second).Should(SatisfyAll(HaveLen(1), Not(ContainElement(endpointSliceHash))))
+}
+
 func TestEndpointsForUpstreamOrderDoesntMatter(t *testing.T) {
 	g := gomega.NewWithT(t)