Merge pull request #2675 from sawsa307/crd-client

Create Node Topology CRD client and informer.

cmd/glbc/main.go

@@ -27,6 +27,7 @@ import (
 
 	firewallcrclient "github.com/GoogleCloudPlatform/gke-networking-api/client/gcpfirewall/clientset/versioned"
 	networkclient "github.com/GoogleCloudPlatform/gke-networking-api/client/network/clientset/versioned"
+	nodetopologyclient "github.com/GoogleCloudPlatform/gke-networking-api/client/nodetopology/clientset/versioned"
 	k8scp "github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud"
 	flag "github.com/spf13/pflag"
 	crdclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
@@ -204,6 +205,14 @@ func main() {
 		}
 	}
 
+	var nodeTopologyClient nodetopologyclient.Interface
+	if flags.F.EnableMultiSubnetClusterPhase1 {
+		nodeTopologyClient, err = nodetopologyclient.NewForConfig(kubeConfig)
+		if err != nil {
+			klog.Fatalf("Failed to create Node Topology Client: %v", err)
+		}
+	}
+
 	namer, err := app.NewNamer(kubeClient, flags.F.ClusterName, firewalls.DefaultFirewallName, rootLogger)
 	if err != nil {
 		klog.Fatalf("app.NewNamer(ctx.KubeClient, %q, %q) = %v", flags.F.ClusterName, firewalls.DefaultFirewallName, err)
@@ -256,7 +265,7 @@ func main() {
 		EnableWeightedL4NetLB:         flags.F.EnableWeightedL4NetLB,
 		DisableL4LBFirewall:           flags.F.DisableL4LBFirewall,
 	}
-	ctx := ingctx.NewControllerContext(kubeConfig, kubeClient, backendConfigClient, frontendConfigClient, firewallCRClient, svcNegClient, ingParamsClient, svcAttachmentClient, networkClient, eventRecorderKubeClient, cloud, namer, kubeSystemUID, ctxConfig, rootLogger)
+	ctx := ingctx.NewControllerContext(kubeConfig, kubeClient, backendConfigClient, frontendConfigClient, firewallCRClient, svcNegClient, ingParamsClient, svcAttachmentClient, networkClient, nodeTopologyClient, eventRecorderKubeClient, cloud, namer, kubeSystemUID, ctxConfig, rootLogger)
 	go app.RunHTTPServer(ctx.HealthCheck, rootLogger)
 
 	var once sync.Once

go.mod

@@ -3,7 +3,7 @@ module k8s.io/ingress-gce
 go 1.22.4
 
 require (
-	github.com/GoogleCloudPlatform/gke-networking-api v0.1.0
+	github.com/GoogleCloudPlatform/gke-networking-api v0.1.2-0.20240909212819-4b1bab7c69ea
 	github.com/GoogleCloudPlatform/k8s-cloud-provider v1.32.0
 	github.com/go-logr/logr v1.4.1
 	github.com/golang/protobuf v1.5.4

go.sum

@@ -6,8 +6,8 @@ cloud.google.com/go/auth/oauth2adapt v0.2.2/go.mod h1:wcYjgpZI9+Yu7LyYBg4pqSiaRk
 cloud.google.com/go/compute/metadata v0.3.0 h1:Tz+eQXMEqDIKRsmY3cHTL6FVaynIjX2QxYC4trgAKZc=
 cloud.google.com/go/compute/metadata v0.3.0/go.mod h1:zFmK7XCadkQkj6TtorcaGlCW1hT1fIilQDwofLpJ20k=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/GoogleCloudPlatform/gke-networking-api v0.1.0 h1:HooP7w65j7M/mZ4oeXSrEyZNJZi2aw0sOwuHjh+NDYI=
-github.com/GoogleCloudPlatform/gke-networking-api v0.1.0/go.mod h1:hngDInU0bLls5Nk7vssNbT6noFXS6ZwldjRhHwu85uE=
+github.com/GoogleCloudPlatform/gke-networking-api v0.1.2-0.20240909212819-4b1bab7c69ea h1:vstMt6suj5yqM0ey5znDUd5EVAk27MF3hWeA2namKtM=
+github.com/GoogleCloudPlatform/gke-networking-api v0.1.2-0.20240909212819-4b1bab7c69ea/go.mod h1:YnoYXo/cwpqFmIXKblHOV5jFEpsSL3PZeo0zaR3oGTI=
 github.com/GoogleCloudPlatform/k8s-cloud-provider v1.32.0 h1:mJj6GEg+6v18wbNRbuQompx7UXB9TMhreiYLKv8BADU=
 github.com/GoogleCloudPlatform/k8s-cloud-provider v1.32.0/go.mod h1:o5jNXLZ3JSgVv+mR2PFp7syFS3DGi6DlzMDwi928YJA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=

pkg/common/operator/ingress_test.go

@@ -57,7 +57,7 @@ func TestReferencesSvcNeg(t *testing.T) {
 		HealthCheckPath:               "/",
 		EnableIngressRegionalExternal: true,
 	}
-	ctx := context.NewControllerContext(nil, kubeClient, nil, nil, nil, nil, nil, nil, nil, kubeClient /*kube client to be used for events*/, gceClient, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
+	ctx := context.NewControllerContext(nil, kubeClient, nil, nil, nil, nil, nil, nil, nil, nil, kubeClient /*kube client to be used for events*/, gceClient, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
 
 	if err := addTestService(ctx); err != nil {
 		t.Fatalf("Failed to add test service: %v", err)

pkg/context/context.go

@@ -22,6 +22,8 @@ import (
 	informerfirewall "github.com/GoogleCloudPlatform/gke-networking-api/client/gcpfirewall/informers/externalversions/gcpfirewall/v1"
 	networkclient "github.com/GoogleCloudPlatform/gke-networking-api/client/network/clientset/versioned"
 	informernetwork "github.com/GoogleCloudPlatform/gke-networking-api/client/network/informers/externalversions/network/v1"
+	nodetopologyclient "github.com/GoogleCloudPlatform/gke-networking-api/client/nodetopology/clientset/versioned"
+	informernodetopology "github.com/GoogleCloudPlatform/gke-networking-api/client/nodetopology/informers/externalversions/nodetopology/v1"
 	apiv1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -79,6 +81,7 @@ type ControllerContext struct {
 	SAClient            serviceattachmentclient.Interface
 	FirewallClient      firewallclient.Interface
 	EventRecorderClient kubernetes.Interface
+	NodeTopologyClient  nodetopologyclient.Interface
 
 	Cloud *gce.Cloud
 
@@ -104,6 +107,7 @@ type ControllerContext struct {
 	FirewallInformer         cache.SharedIndexInformer
 	NetworkInformer          cache.SharedIndexInformer
 	GKENetworkParamsInformer cache.SharedIndexInformer
+	NodeTopologyInformer     cache.SharedIndexInformer
 
 	ControllerMetrics *metrics.ControllerMetrics
 
@@ -160,6 +164,7 @@ func NewControllerContext(
 	ingParamsClient ingparamsclient.Interface,
 	saClient serviceattachmentclient.Interface,
 	networkClient networkclient.Interface,
+	nodeTopologyClient nodetopologyclient.Interface,
 	eventRecorderClient kubernetes.Interface,
 	cloud *gce.Cloud,
 	clusterNamer *namer.Namer,
@@ -186,6 +191,7 @@ func NewControllerContext(
 		SvcNegClient:            svcnegClient,
 		SAClient:                saClient,
 		EventRecorderClient:     eventRecorderClient,
+		NodeTopologyClient:      nodeTopologyClient,
 		Cloud:                   cloud,
 		ClusterNamer:            clusterNamer,
 		L4Namer:                 namer.NewL4Namer(string(kubeSystemUID), clusterNamer),
@@ -226,6 +232,10 @@ func NewControllerContext(
 		context.RegionalCluster = true
 	}
 
+	if flags.F.EnableMultiSubnetClusterPhase1 && nodeTopologyClient != nil {
+		context.NodeTopologyInformer = informernodetopology.NewNodeTopologyInformer(nodeTopologyClient, config.ResyncPeriod, utils.NewNamespaceIndexer())
+	}
+
 	// Do not trigger periodic resync on EndpointSlices object.
 	// This aims improve NEG controller performance by avoiding unnecessary NEG sync that triggers for each NEG syncer.
 	// As periodic resync may temporary starve NEG API ratelimit quota.
@@ -333,6 +343,10 @@ func (ctx *ControllerContext) HasSynced() bool {
 		funcs = append(funcs, ctx.FirewallInformer.HasSynced)
 	}
 
+	if ctx.NodeTopologyInformer != nil {
+		funcs = append(funcs, ctx.NodeTopologyInformer.HasSynced)
+	}
+
 	for _, f := range funcs {
 		if !f() {
 			return false

pkg/controller/controller_test.go

@@ -87,7 +87,7 @@ func newLoadBalancerController() *LoadBalancerController {
 		HealthCheckPath:               "/",
 		EnableIngressRegionalExternal: true,
 	}
-	ctx := context.NewControllerContext(nil, kubeClient, backendConfigClient, nil, nil, svcNegClient, nil, nil, nil, kubeClient /*kube client to be used for events*/, fakeGCE, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
+	ctx := context.NewControllerContext(nil, kubeClient, backendConfigClient, nil, nil, svcNegClient, nil, nil, nil, nil, kubeClient /*kube client to be used for events*/, fakeGCE, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
 	lbc := NewLoadBalancerController(ctx, stopCh, klog.TODO())
 	// TODO(rramkumar): Fix this so we don't have to override with our fake
 	lbc.instancePool = instancegroups.NewManager(&instancegroups.ManagerConfig{

pkg/firewalls/controller_test.go

@@ -54,7 +54,7 @@ func newFirewallController() *FirewallController {
 		ResyncPeriod:          1 * time.Minute,
 		DefaultBackendSvcPort: test.DefaultBeSvcPort,
 	}
-	ctx := context.NewControllerContext(nil, kubeClient, backendConfigClient, nil, firewallClient, nil, nil, nil, nil, kubeClient /*kube client to be used for events*/, fakeGCE, defaultNamer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
+	ctx := context.NewControllerContext(nil, kubeClient, backendConfigClient, nil, firewallClient, nil, nil, nil, nil, nil, kubeClient /*kube client to be used for events*/, fakeGCE, defaultNamer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
 	fwc := NewFirewallController(ctx, []string{"30000-32767"}, false, false, true, make(chan struct{}), klog.TODO())
 	fwc.hasSynced = func() bool { return true }
 

pkg/l4lb/l4controller_test.go

@@ -851,7 +851,7 @@ func newServiceController(t *testing.T, fakeGCE *gce.Cloud) *L4Controller {
 		ResyncPeriod: 1 * time.Minute,
 		NumL4Workers: 5,
 	}
-	ctx := context.NewControllerContext(nil, kubeClient, nil, nil, nil, svcNegClient, nil, nil, nil, kubeClient /*kube client to be used for events*/, fakeGCE, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
+	ctx := context.NewControllerContext(nil, kubeClient, nil, nil, nil, svcNegClient, nil, nil, nil, nil, kubeClient /*kube client to be used for events*/, fakeGCE, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
 	// Add some nodes so that NEG linker kicks in during ILB creation.
 	nodes, err := test.CreateAndInsertNodes(ctx.Cloud, []string{"instance-1"}, vals.ZoneName)
 	if err != nil {

pkg/l4lb/l4netlbcontroller_test.go

@@ -310,7 +310,7 @@ func buildContext(vals gce.TestClusterValues) *ingctx.ControllerContext {
 		NumL4NetLBWorkers: 5,
 		MaxIGSize:         1000,
 	}
-	return ingctx.NewControllerContext(nil, kubeClient, nil, nil, nil, svcNegClient, nil, nil, networkClient, kubeClient /*kube client to be used for events*/, fakeGCE, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
+	return ingctx.NewControllerContext(nil, kubeClient, nil, nil, nil, svcNegClient, nil, nil, networkClient, nil, kubeClient /*kube client to be used for events*/, fakeGCE, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
 }
 
 func newL4NetLBServiceController() *L4NetLBController {

pkg/psc/controller_test.go

@@ -1097,7 +1097,7 @@ func newTestController(clusterType string) *Controller {
 
 	flags.F.GKEClusterName = ClusterName
 	flags.F.GKEClusterType = clusterType
-	ctx := context.NewControllerContext(nil, kubeClient, nil, nil, nil, nil, nil, saClient, nil, kubeClient /*kube client to be used for events*/, gceClient, resourceNamer, kubeSystemUID, ctxConfig, klog.TODO())
+	ctx := context.NewControllerContext(nil, kubeClient, nil, nil, nil, nil, nil, saClient, nil, nil, kubeClient /*kube client to be used for events*/, gceClient, resourceNamer, kubeSystemUID, ctxConfig, klog.TODO())
 
 	return NewController(ctx, make(<-chan struct{}), klog.TODO())
 }