minor test fixes + equals on policy ir (#10715)

Author: yuval-k

internal/kgateway/controller/start.go

@@ -51,6 +51,10 @@ type SetupOpts struct {
 
 	// static set of global Settings
 	GlobalSettings *settings.Settings
+
+	PprofBindAddress       string
+	HealthProbeBindAddress string
+	MetricsBindAddress     string
 }
 
 var setupLog = ctrl.Log.WithName("setup")
@@ -103,11 +107,11 @@ func NewControllerBuilder(ctx context.Context, cfg StartConfig) (*ControllerBuil
 	mgrOpts := ctrl.Options{
 		BaseContext:      func() context.Context { return ctx },
 		Scheme:           scheme,
-		PprofBindAddress: "127.0.0.1:9099",
+		PprofBindAddress: cfg.SetupOpts.PprofBindAddress,
 		// if you change the port here, also change the port "health" in the helmchart.
-		HealthProbeBindAddress: ":9093",
+		HealthProbeBindAddress: cfg.SetupOpts.HealthProbeBindAddress,
 		Metrics: metricsserver.Options{
-			BindAddress: ":9092",
+			BindAddress: cfg.SetupOpts.MetricsBindAddress,
 		},
 		Controller: config.Controller{
 			// see https://github.com/kubernetes-sigs/controller-runtime/issues/2937

internal/kgateway/ir/iface.go

@@ -131,7 +131,7 @@ func (c PolicyWrapper) Equals(in PolicyWrapper) bool {
 		return false
 	}
 
-	return versionEquals(c.Policy, in.Policy)
+	return versionEquals(c.Policy, in.Policy) && c.PolicyIR.Equals(in.PolicyIR)
 }
 
 var (

internal/kgateway/krtcollections/uniqueclients_test.go

@@ -75,6 +75,84 @@ func TestUniqueClients(t *testing.T) {
 				})),
 			),
 		},
+		{
+			name: "two UCCs",
+			inputs: []any{
+				&corev1.Pod{
+					TypeMeta: metav1.TypeMeta{},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "podname",
+						Namespace: "ns",
+						Labels:    map[string]string{"a": "b"},
+					},
+					Spec: corev1.PodSpec{
+						NodeName: "node",
+					},
+				},
+				&corev1.Node{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "node",
+						Labels: map[string]string{
+							corev1.LabelTopologyRegion: "region",
+							corev1.LabelTopologyZone:   "zone",
+						},
+					},
+				},
+				&corev1.Pod{
+					TypeMeta: metav1.TypeMeta{},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "podname2",
+						Namespace: "ns",
+						Labels:    map[string]string{"a": "b"},
+					},
+					Spec: corev1.PodSpec{
+						NodeName: "node2",
+					},
+				},
+				&corev1.Node{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "node2",
+						Labels: map[string]string{
+							corev1.LabelTopologyRegion: "region2",
+							corev1.LabelTopologyZone:   "zone2",
+						},
+					},
+				},
+			},
+			requests: []*envoy_service_discovery_v3.DiscoveryRequest{
+				{
+					Node: &corev3.Node{
+						Id: "podname.ns",
+						Metadata: &structpb.Struct{
+							Fields: map[string]*structpb.Value{
+								xds.RoleKey: structpb.NewStringValue(wellknown.GatewayApiProxyValue + "~best-proxy-role"),
+							},
+						},
+					},
+				},
+				{
+					Node: &corev3.Node{
+						Id: "podname2.ns",
+						Metadata: &structpb.Struct{
+							Fields: map[string]*structpb.Value{
+								xds.RoleKey: structpb.NewStringValue(wellknown.GatewayApiProxyValue + "~best-proxy-role"),
+							},
+						},
+					},
+				},
+			},
+			result: sets.New(
+				fmt.Sprintf("kgateway-kube-gateway-api~best-proxy-role~%d~ns", utils.HashLabels(map[string]string{
+					corev1.LabelTopologyRegion: "region",
+					corev1.LabelTopologyZone:   "zone",
+					"a":                        "b",
+				})), fmt.Sprintf("kgateway-kube-gateway-api~best-proxy-role~%d~ns", utils.HashLabels(map[string]string{
+					corev1.LabelTopologyRegion: "region2",
+					corev1.LabelTopologyZone:   "zone2",
+					"a":                        "b",
+				})),
+			),
+		},
 		{
 			name:   "no-pods",
 			inputs: nil,
@@ -140,16 +218,17 @@ func TestUniqueClients(t *testing.T) {
 			g.Expect(names).To(Equal(tc.result))
 
 			for i := range tc.requests {
-				for j := 0; j < 10; j++ {
-					g.Expect(ucc.List()).To(HaveLen(len(allUcc) - i))
+				for j := 0; j < 9; j++ {
 					cb.OnStreamClosed(int64(i*10+j), nil)
 				}
-				// FIXME: this assertion will be reworked; the goal here is to test jitter
-				// that as clients disconnect we don't impact other UCCs etc.
-				// g.Eventually(func() []ir.UniqlyConnectedClient {
-				// 	allUcc = ucc.List()
-				// 	return allUcc
-				// }, "1s").Should(HaveLen(len(tc.result)))
+			}
+
+			g.Expect(ucc.List()).Should(HaveLen(len(tc.result)))
+
+			for i := range tc.requests {
+				j := 9
+				g.Eventually(ucc.List).Should(HaveLen(len(allUcc) - i))
+				cb.OnStreamClosed(int64(i*10+j), nil)
 			}
 
 			// as events happens async, eventually after all clients disconnect all UCCs should be removed

internal/kgateway/setup/controlplane.go

@@ -29,12 +29,17 @@ func NewControlPlane(
 	if err != nil {
 		return nil, err
 	}
-	return NewControlPlaneWithListener(ctx, lis, callbacks)
+	snapshotCache, grpcServer := NewControlPlaneWithListener(ctx, lis, callbacks)
+	go func() {
+		<-ctx.Done()
+		grpcServer.GracefulStop()
+	}()
+	return snapshotCache, err
 }
 
 func NewControlPlaneWithListener(ctx context.Context,
 	lis net.Listener,
-	callbacks xdsserver.Callbacks) (envoycache.SnapshotCache, error) {
+	callbacks xdsserver.Callbacks) (envoycache.SnapshotCache, *grpc.Server) {
 	logger := contextutils.LoggerFrom(ctx).Desugar()
 	serverOpts := []grpc.ServerOption{
 		grpc.StreamInterceptor(
@@ -61,10 +66,6 @@ func NewControlPlaneWithListener(ctx context.Context,
 	envoy_service_discovery_v3.RegisterAggregatedDiscoveryServiceServer(grpcServer, xdsServer)
 
 	go grpcServer.Serve(lis)
-	go func() {
-		<-ctx.Done()
-		grpcServer.GracefulStop()
-	}()
 
-	return snapshotCache, nil
+	return snapshotCache, grpcServer
 }

internal/kgateway/setup/ggv2setup.go

@@ -73,10 +73,13 @@ func StartGGv2(
 	}
 
 	setupOpts := &controller.SetupOpts{
-		Cache:               cache,
-		KrtDebugger:         new(krt.DebugHandler),
-		ExtraGatewayClasses: extraGwClasses,
-		GlobalSettings:      st,
+		Cache:                  cache,
+		KrtDebugger:            new(krt.DebugHandler),
+		ExtraGatewayClasses:    extraGwClasses,
+		GlobalSettings:         st,
+		PprofBindAddress:       "127.0.0.1:9099",
+		HealthProbeBindAddress: ":9093",
+		MetricsBindAddress:     ":9092",
 	}
 
 	restConfig := ctrl.GetConfigOrDie()

internal/kgateway/setup/ggv2setup_test.go

@@ -178,10 +178,8 @@ func TestScenarios(t *testing.T) {
 		t.Fatalf("can't listen %v", err)
 	}
 	xdsPort := lis.Addr().(*net.TCPAddr).Port
-	snapCache, err := ggv2setup.NewControlPlaneWithListener(ctx, lis, uniqueClientCallbacks)
-	if err != nil {
-		t.Fatalf("can't listen %v", err)
-	}
+	snapCache, grpcServer := ggv2setup.NewControlPlaneWithListener(ctx, lis, uniqueClientCallbacks)
+	t.Cleanup(func() { grpcServer.Stop() })
 
 	st, err := settings.BuildSettings()
 	if err != nil {

internal/kgateway/setup/testdata/setupyaml/setup.yaml

@@ -15,4 +15,29 @@ apiVersion: gateway.kgateway.dev/v1alpha1
 metadata:
   name: kgateway
 spec:
-  kube: {}
+  kube:
+    deployment:
+      replicas: 1
+    envoyContainer:
+      image:
+        registry: ghcr.io/kgateway-dev
+        repository: envoy-wrapper
+        tag: v0.0.1
+        pullPolicy: IfNotPresent
+      securityContext:
+        allowPrivilegeEscalation: false
+        capabilities:
+          add:
+          - NET_BIND_SERVICE
+          drop:
+          - ALL
+        readOnlyRootFilesystem: true
+        runAsNonRoot: true
+        runAsUser: 10101
+    service:
+      type: LoadBalancer
+    stats:
+      enableStatsRoute: true
+      enabled: true
+      routePrefixRewrite: /stats/prometheus
+      statsRoutePrefixRewrite: /stats