add feature flag enable-multi-subnet-cluster for GKE Multi-Subnet Cluster feature (#848)

* add feature flag

* minor fix

* fix test

* tidy nil check

* dedup queue item for update with same label, fix typo

* Revert "dedup queue item for update with same label, fix typo"

This reverts commit ebadd53.

* minor fix typo

* fix typo

* fix comment

Author: YifeiZhuang

cmd/cloud-controller-manager/nodeipamcontroller.go

@@ -149,6 +149,7 @@ func startNodeIpamController(ccmConfig *cloudcontrollerconfig.CompletedConfig, n
 		nwInformer,
 		gnpInformer,
 		nodeTopologyClient,
+		nodeIPAMConfig.EnableMultiSubnetCluster,
 		clusterCIDRs,
 		serviceCIDR,
 		secondaryServiceCIDR,

cmd/cloud-controller-manager/options/nodeipamcontroller.go

@@ -39,6 +39,7 @@ func (o *NodeIPAMControllerOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.Int32Var(&o.NodeCIDRMaskSize, "node-cidr-mask-size", o.NodeCIDRMaskSize, "Mask size for node cidr in cluster. Default is 24 for IPv4 and 64 for IPv6.")
 	fs.Int32Var(&o.NodeCIDRMaskSizeIPv4, "node-cidr-mask-size-ipv4", o.NodeCIDRMaskSizeIPv4, "Mask size for IPv4 node cidr in dual-stack cluster. Default is 24.")
 	fs.Int32Var(&o.NodeCIDRMaskSizeIPv6, "node-cidr-mask-size-ipv6", o.NodeCIDRMaskSizeIPv6, "Mask size for IPv6 node cidr in dual-stack cluster. Default is 64.")
+	fs.BoolVar(&o.EnableMultiSubnetCluster, "enable-multi-subnet-cluster", o.EnableMultiSubnetCluster, "Enabled multi-subnet cluster feature. This enables generating updated nodeTopology custom resource. ")
 }
 
 // ApplyTo fills up NodeIpamController config with options.
@@ -59,6 +60,7 @@ func (o *NodeIPAMControllerOptions) ApplyTo(cfg *nodeipamconfig.NodeIPAMControll
 	cfg.NodeCIDRMaskSize = o.NodeCIDRMaskSize
 	cfg.NodeCIDRMaskSizeIPv4 = o.NodeCIDRMaskSizeIPv4
 	cfg.NodeCIDRMaskSizeIPv6 = o.NodeCIDRMaskSizeIPv6
+	cfg.EnableMultiSubnetCluster = o.EnableMultiSubnetCluster
 
 	return nil
 }

pkg/controller/nodeipam/config/types.go

@@ -31,4 +31,9 @@ type NodeIPAMControllerConfiguration struct {
 	// NodeCIDRMaskSizeIPv6 is the mask size for IPv6 node cidr in dual-stack cluster.
 	// This can be used only with dual stack clusters and is incompatible with single stack clusters.
 	NodeCIDRMaskSizeIPv6 int32
+	// EnableMultiSubnetCluster is bound to a command-line flag. When true, it enables
+	// generating nodeTopology custom resource based on node's subnetwork configuration,
+	// which is represented by a node label. Enabling this feature also assumes that a
+	// nodeTopology CR named 'default' is already installed.
+	EnableMultiSubnetCluster bool
 }

pkg/controller/nodeipam/ipam/cidr_allocator.go

@@ -115,7 +115,7 @@ type CIDRAllocatorParams struct {
 }
 
 // New creates a new CIDR range allocator.
-func New(kubeClient clientset.Interface, cloud cloudprovider.Interface, nodeInformer informers.NodeInformer, nwInformer networkinformer.NetworkInformer, gnpInformer networkinformer.GKENetworkParamSetInformer, nodeTopologyClient nodetopologyclientset.Interface, allocatorType CIDRAllocatorType, allocatorParams CIDRAllocatorParams) (CIDRAllocator, error) {
+func New(kubeClient clientset.Interface, cloud cloudprovider.Interface, nodeInformer informers.NodeInformer, nwInformer networkinformer.NetworkInformer, gnpInformer networkinformer.GKENetworkParamSetInformer, nodeTopologyClient nodetopologyclientset.Interface, enableMultiSubnetCluster bool, allocatorType CIDRAllocatorType, allocatorParams CIDRAllocatorParams) (CIDRAllocator, error) {
 	nodeList, err := listNodes(kubeClient)
 	if err != nil {
 		return nil, err
@@ -125,7 +125,7 @@ func New(kubeClient clientset.Interface, cloud cloudprovider.Interface, nodeInfo
 	case RangeAllocatorType:
 		return NewCIDRRangeAllocator(kubeClient, nodeInformer, allocatorParams, nodeList)
 	case CloudAllocatorType:
-		return NewCloudCIDRAllocator(kubeClient, cloud, nwInformer, gnpInformer, nodeTopologyClient, nodeInformer, allocatorParams)
+		return NewCloudCIDRAllocator(kubeClient, cloud, nwInformer, gnpInformer, nodeTopologyClient, enableMultiSubnetCluster, nodeInformer, allocatorParams)
 	default:
 		return nil, fmt.Errorf("invalid CIDR allocator type: %v", allocatorType)
 	}

pkg/controller/nodeipam/ipam/cloud_cidr_allocator.go

@@ -71,6 +71,12 @@ const (
 	stackIPv6     clusterStackType = "IPv6"
 )
 
+// enableNodeTopology is bound to a command-line flag. When true, it enables
+// generating nodeTopology custom resource based on node's subnetwork configuration,
+// which is represented by a node label. Enabling this feature also assumes that a
+// nodeTopology CR named 'default' is already installed.
+var enableNodeTopology bool
+
 // cloudCIDRAllocator allocates node CIDRs according to IP address aliases
 // assigned by the cloud provider. In this case, the allocation and
 // deallocation is delegated to the external provider, and the controller
@@ -100,7 +106,7 @@ type cloudCIDRAllocator struct {
 var _ CIDRAllocator = (*cloudCIDRAllocator)(nil)
 
 // NewCloudCIDRAllocator creates a new cloud CIDR allocator.
-func NewCloudCIDRAllocator(client clientset.Interface, cloud cloudprovider.Interface, nwInformer networkinformer.NetworkInformer, gnpInformer networkinformer.GKENetworkParamSetInformer, nodeTopologyClient nodetopologyclientset.Interface, nodeInformer informers.NodeInformer, allocatorParams CIDRAllocatorParams) (CIDRAllocator, error) {
+func NewCloudCIDRAllocator(client clientset.Interface, cloud cloudprovider.Interface, nwInformer networkinformer.NetworkInformer, gnpInformer networkinformer.GKENetworkParamSetInformer, nodeTopologyClient nodetopologyclientset.Interface, enableMultiSubnetCluster bool, nodeInformer informers.NodeInformer, allocatorParams CIDRAllocatorParams) (CIDRAllocator, error) {
 	if client == nil {
 		klog.Fatalf("kubeClient is nil when starting NodeController")
 	}
@@ -131,24 +137,16 @@ func NewCloudCIDRAllocator(client clientset.Interface, cloud cloudprovider.Inter
 		stackType = stackIPv6
 	}
 
-	nodeTopologySyncer := &NodeTopologySyncer{
-		nodeTopologyClient: nodeTopologyClient,
-		cloud:              gceCloud,
-		nodeLister:         nodeInformer.Lister(),
-	}
-	nodetopologyQueue := NewTaskQueue("nodetopologgTaskQueue", "nodetopologyCRD", nodeTopologyWorkers, nodeTopologyKeyFun, nodeTopologySyncer.sync)
-
 	ca := &cloudCIDRAllocator{
-		client:            client,
-		cloud:             gceCloud,
-		networksLister:    nwInformer.Lister(),
-		gnpLister:         gnpInformer.Lister(),
-		nodeLister:        nodeInformer.Lister(),
-		nodesSynced:       nodeInformer.Informer().HasSynced,
-		recorder:          recorder,
-		queue:             workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workqueue.RateLimitingQueueConfig{Name: workqueueName}),
-		nodeTopologyQueue: nodetopologyQueue,
-		stackType:         stackType,
+		client:         client,
+		cloud:          gceCloud,
+		networksLister: nwInformer.Lister(),
+		gnpLister:      gnpInformer.Lister(),
+		nodeLister:     nodeInformer.Lister(),
+		nodesSynced:    nodeInformer.Informer().HasSynced,
+		recorder:       recorder,
+		queue:          workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workqueue.RateLimitingQueueConfig{Name: workqueueName}),
+		stackType:      stackType,
 	}
 
 	nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -174,26 +172,32 @@ func NewCloudCIDRAllocator(client clientset.Interface, cloud cloudprovider.Inter
 		}),
 		DeleteFunc: nodeutil.CreateDeleteNodeHandler(ca.ReleaseCIDR),
 	})
-	nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: nodeutil.CreateAddNodeHandler(func(node *v1.Node) error {
-			if ca.nodeTopologyQueue != nil {
+
+	enableNodeTopology = enableMultiSubnetCluster
+	if enableNodeTopology {
+		nodeTopologySyncer := &NodeTopologySyncer{
+			nodeTopologyClient: nodeTopologyClient,
+			cloud:              gceCloud,
+			nodeLister:         nodeInformer.Lister(),
+		}
+		nodetopologyQueue := NewTaskQueue("nodetopologyTaskQueue", "nodetopologyCRD", nodeTopologyWorkers, nodeTopologyKeyFun, nodeTopologySyncer.sync)
+		ca.nodeTopologyQueue = nodetopologyQueue
+
+		nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc: nodeutil.CreateAddNodeHandler(func(node *v1.Node) error {
 				ca.nodeTopologyQueue.Enqueue(node)
-			}
-			return nil
-		}),
-		UpdateFunc: nodeutil.CreateUpdateNodeHandler(func(oldNode, newNode *v1.Node) error {
-			if ca.nodeTopologyQueue != nil {
+				return nil
+			}),
+			UpdateFunc: nodeutil.CreateUpdateNodeHandler(func(oldNode, newNode *v1.Node) error {
 				nodetopologyQueue.Enqueue(newNode)
-			}
-			return nil
-		}),
-		DeleteFunc: nodeutil.CreateDeleteNodeHandler(func(node *v1.Node) error {
-			if ca.nodeTopologyQueue != nil {
+				return nil
+			}),
+			DeleteFunc: nodeutil.CreateDeleteNodeHandler(func(node *v1.Node) error {
 				nodetopologyQueue.Enqueue(node)
-			}
-			return nil
-		}),
-	})
+				return nil
+			}),
+		})
+	}
 
 	nwInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(originalObj interface{}) {
@@ -270,8 +274,6 @@ func (ca *cloudCIDRAllocator) Run(stopCh <-chan struct{}) {
 	ctx, cancelFn := context.WithCancel(context.Background())
 	defer cancelFn()
 	defer ca.queue.ShutDown()
-	defer ca.nodeTopologyQueue.Shutdown()
-
 	klog.Infof("Starting cloud CIDR allocator")
 	defer klog.Infof("Shutting down cloud CIDR allocator")
 
@@ -282,18 +284,26 @@ func (ca *cloudCIDRAllocator) Run(stopCh <-chan struct{}) {
 	for i := 0; i < cidrUpdateWorkers; i++ {
 		go wait.UntilWithContext(ctx, ca.runWorker, time.Second)
 	}
-	if ca.nodeTopologyQueue != nil {
-		ca.nodeTopologyQueue.Run()
-	}
 
-	go func() {
-		time.Sleep(nodeTopologyReconcileInterval)
-		wait.Until(
-			func() {
-				ca.nodeTopologyQueue.Enqueue(nodeTopologyReconcileFakeNode)
-			},
-			nodeTopologyReconcileInterval, stopCh)
-	}()
+	if enableNodeTopology {
+		if ca.nodeTopologyQueue != nil {
+			defer ca.nodeTopologyQueue.Shutdown()
+			ca.nodeTopologyQueue.Run()
+		}
+		go func() {
+			time.Sleep(nodeTopologyReconcileInterval)
+			wait.Until(
+				func() {
+					if ca.nodeTopologyQueue != nil {
+						// nodeTopologyReconcileFakeNode triggers reconciliation. Node_topology_syncer
+						// will not find the fake node in nodeInformer cache, forcing a full reconciliation
+						// of the nodeTopology custom resource.
+						ca.nodeTopologyQueue.Enqueue(nodeTopologyReconcileFakeNode)
+					}
+				},
+				nodeTopologyReconcileInterval, stopCh)
+		}()
+	}
 
 	<-stopCh
 }

pkg/controller/nodeipam/ipam/cloud_cidr_allocator_test.go

@@ -143,7 +143,7 @@ func TestNodeTopologyQueuePeriodicSync(t *testing.T) {
 	allocatorParams := CIDRAllocatorParams{}
 
 	KuberntesClientSet := fake.NewSimpleClientset()
-	ca, _ := NewCloudCIDRAllocator(KuberntesClientSet, fakeGCE, nwInformer, gnpInformer, nodeTopologyClient, fakeNodeInformer, allocatorParams)
+	ca, _ := NewCloudCIDRAllocator(KuberntesClientSet, fakeGCE, nwInformer, gnpInformer, nodeTopologyClient, true, fakeNodeInformer, allocatorParams)
 	cloudAllocator, _ := ca.(*cloudCIDRAllocator)
 	cloudAllocator.nodeTopologyQueue.Run()
 
@@ -223,7 +223,7 @@ func TestNodeTopologyQueue_AddOrUpdate(t *testing.T) {
 	allocatorParams := CIDRAllocatorParams{}
 
 	KuberntesClientSet := fake.NewSimpleClientset()
-	ca, _ := NewCloudCIDRAllocator(KuberntesClientSet, fakeGCE, nwInformer, gnpInformer, nodeTopologyClient, fakeNodeInformer, allocatorParams)
+	ca, _ := NewCloudCIDRAllocator(KuberntesClientSet, fakeGCE, nwInformer, gnpInformer, nodeTopologyClient, true, fakeNodeInformer, allocatorParams)
 	cloudAllocator, _ := ca.(*cloudCIDRAllocator)
 
 	fakeInformerFactory.Start(wait.NeverStop)
@@ -300,7 +300,7 @@ func TestNodeTopologyCR_DELETION(t *testing.T) {
 	allocatorParams := CIDRAllocatorParams{}
 
 	KuberntesClientSet := fake.NewSimpleClientset()
-	ca, _ := NewCloudCIDRAllocator(KuberntesClientSet, fakeGCE, nwInformer, gnpInformer, nodeTopologyClient, fakeNodeInformer, allocatorParams)
+	ca, _ := NewCloudCIDRAllocator(KuberntesClientSet, fakeGCE, nwInformer, gnpInformer, nodeTopologyClient, true, fakeNodeInformer, allocatorParams)
 	cloudAllocator, _ := ca.(*cloudCIDRAllocator)
 
 	fakeInformerFactory.Start(wait.NeverStop)

pkg/controller/nodeipam/node_ipam_controller.go

@@ -79,6 +79,7 @@ func NewNodeIpamController(
 	nwInformer networkinformer.NetworkInformer,
 	gnpInformer networkinformer.GKENetworkParamSetInformer,
 	nodeTopologyClient nodetopologyclientset.Interface,
+	enableMultiSubnetCluster bool,
 	clusterCIDRs []*net.IPNet,
 	serviceCIDR *net.IPNet,
 	secondaryServiceCIDR *net.IPNet,
@@ -135,7 +136,7 @@ func NewNodeIpamController(
 			NodeCIDRMaskSizes:    nodeCIDRMaskSizes,
 		}
 
-		ic.cidrAllocator, err = ipam.New(kubeClient, cloud, nodeInformer, nwInformer, gnpInformer, nodeTopologyClient, ic.allocatorType, allocatorParams)
+		ic.cidrAllocator, err = ipam.New(kubeClient, cloud, nodeInformer, nwInformer, gnpInformer, nodeTopologyClient, enableMultiSubnetCluster, ic.allocatorType, allocatorParams)
 		if err != nil {
 			return nil, err
 		}

pkg/controller/nodeipam/node_ipam_controller_test.go

@@ -63,7 +63,7 @@ func newTestNodeIpamController(clusterCIDR []*net.IPNet, serviceCIDR *net.IPNet,
 	fakeGCE := gce.NewFakeGCECloud(gce.DefaultTestClusterValues())
 	return NewNodeIpamController(
 		fakeNodeInformer, fakeGCE, clientSet, fakeNwInformer, fakeGNPInformer, nodeTopologyFakeClient,
-		clusterCIDR, serviceCIDR, secondaryServiceCIDR, nodeCIDRMaskSizes, allocatorType,
+		true, clusterCIDR, serviceCIDR, secondaryServiceCIDR, nodeCIDRMaskSizes, allocatorType,
 	)
 }