feat(ccm): watch based route controller

Introduces a new watch based route controller for the cloud-controller-manager. This prevents the exhaustion of API rate limits through an overagressive routes controller. This is further described in #60646. The corresponding feature gate is called "CloudControllerManagerWatchBasedRoutesReconciliation".

staging/src/k8s.io/cloud-provider/controllers/route/route_controller.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -41,9 +42,12 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	clientretry "k8s.io/client-go/util/retry"
+	"k8s.io/client-go/util/workqueue"
 	cloudprovider "k8s.io/cloud-provider"
 	controllersmetrics "k8s.io/component-base/metrics/prometheus/controllers"
 	nodeutil "k8s.io/component-helpers/node/util"
+	"k8s.io/controller-manager/pkg/features"
+	_ "k8s.io/controller-manager/pkg/features/register"
 )
 
 const (
@@ -63,10 +67,12 @@ type RouteController struct {
 	kubeClient       clientset.Interface
 	clusterName      string
 	clusterCIDRs     []*net.IPNet
+	nodeInformer     coreinformers.NodeInformer
 	nodeLister       corelisters.NodeLister
 	nodeListerSynced cache.InformerSynced
 	broadcaster      record.EventBroadcaster
 	recorder         record.EventRecorder
+	workqueue        workqueue.TypedRateLimitingInterface[string]
 }
 
 func New(routes cloudprovider.Routes, kubeClient clientset.Interface, nodeInformer coreinformers.NodeInformer, clusterName string, clusterCIDRs []*net.IPNet) *RouteController {
@@ -79,16 +85,66 @@ func New(routes cloudprovider.Routes, kubeClient clientset.Interface, nodeInform
 		kubeClient:       kubeClient,
 		clusterName:      clusterName,
 		clusterCIDRs:     clusterCIDRs,
+		nodeInformer:     nodeInformer,
 		nodeLister:       nodeInformer.Lister(),
 		nodeListerSynced: nodeInformer.Informer().HasSynced,
 	}
 
+	if utilfeature.DefaultFeatureGate.Enabled(features.CloudControllerManagerWatchBasedRoutesReconciliation) {
+		rc.workqueue = workqueue.NewTypedRateLimitingQueueWithConfig(
+			workqueue.DefaultTypedControllerRateLimiter[string](),
+			workqueue.TypedRateLimitingQueueConfig[string]{Name: "Routes"},
+		)
+
+		rc.nodeInformer.Informer().AddEventHandler(
+			// It is only necessary to reconcile the routes for any events that have the potential to impact them:
+			// - Node is added
+			// - Node is removed
+			// - Node .Status.Addresses is changed
+			//	 TODO: is this a reasonable assumption or are there other values that can impact the Routes?
+			// - Node .Spec.PodCIDRs is changed
+			cache.ResourceEventHandlerFuncs{
+				AddFunc:    rc.enqueueReconcile,
+				UpdateFunc: rc.handleNodeUpdate,
+				DeleteFunc: rc.enqueueReconcile,
+			},
+		)
+	}
+
 	return rc
 }
 
+func (rc *RouteController) enqueueReconcile(_ interface{}) {
+	// reconcileNodeRoutes always reconciles the full cluster. It does not make sense to have
+	// separate entries in the workqueue per node, but a single one for the whole cluster is enough.
+	rc.workqueue.Add("routes")
+}
+
+func (rc *RouteController) handleNodeUpdate(oldObj, newObj interface{}) {
+	// Resync sends an update with old==new, so these events are ignored here.
+
+	oldNode, oldOk := oldObj.(*v1.Node)
+	newNode, newOk := newObj.(*v1.Node)
+
+	if !oldOk || !newOk {
+		return
+	}
+
+	diffInPodCIDR := !reflect.DeepEqual(oldNode.Spec.PodCIDRs, newNode.Spec.PodCIDRs)
+	diffInNodeAddresses := !reflect.DeepEqual(oldNode.Status.Addresses, newNode.Status.Addresses)
+
+	if diffInPodCIDR || diffInNodeAddresses {
+		rc.enqueueReconcile(newObj)
+	}
+}
+
 func (rc *RouteController) Run(ctx context.Context, syncPeriod time.Duration, controllerManagerMetrics *controllersmetrics.ControllerManagerMetrics) {
 	defer utilruntime.HandleCrash()
 
+	if utilfeature.DefaultFeatureGate.Enabled(features.CloudControllerManagerWatchBasedRoutesReconciliation) {
+		defer rc.workqueue.ShutDown()
+	}
+
 	rc.broadcaster = record.NewBroadcaster(record.WithContext(ctx))
 	rc.recorder = rc.broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "route_controller"})
 
@@ -108,20 +164,67 @@ func (rc *RouteController) Run(ctx context.Context, syncPeriod time.Duration, co
 		return
 	}
 
-	// TODO: If we do just the full Resync every 5 minutes (default value)
-	// that means that we may wait up to 5 minutes before even starting
-	// creating a route for it. This is bad.
-	// We should have a watch on node and if we observe a new node (with CIDR?)
-	// trigger reconciliation for that node.
-	go wait.NonSlidingUntil(func() {
-		if err := rc.reconcileNodeRoutes(ctx); err != nil {
-			klog.Errorf("Couldn't reconcile node routes: %v", err)
-		}
-	}, syncPeriod, ctx.Done())
+	if utilfeature.DefaultFeatureGate.Enabled(features.CloudControllerManagerWatchBasedRoutesReconciliation) {
+		// Process any node events that may change the routes.
+		// It does not make sense to add concurrency here, as the routes
+		// controller always processes the whole cluster.
+		go wait.UntilWithContext(ctx, rc.runWorker, time.Second)
+
+		// We should still regularly run the reconcile, even if no events come in. Usually controllers use the
+		// `syncPeriod` for this, but for the route controller the default period is very low (5s) and not useful for
+		// watch-based reconciliation.
+		// Because of our event filters the usual resync period from the informer does not trigger reconciles.
+		// TODO: What time should we use?
+	} else {
+		go wait.NonSlidingUntil(func() {
+			if err := rc.reconcileNodeRoutes(ctx); err != nil {
+				klog.Errorf("Couldn't reconcile node routes: %v", err)
+			}
+		}, syncPeriod, ctx.Done())
+	}
 
 	<-ctx.Done()
 }
 
+func (rc *RouteController) runWorker(ctx context.Context) {
+	for rc.processNextWorkItem(ctx) {
+	}
+}
+
+// processNextWorkItem will read a single work item off the workqueue and
+// attempt to process it, by calling reconcileNodeRoutes.
+func (rc *RouteController) processNextWorkItem(ctx context.Context) bool {
+	obj, shutdown := rc.workqueue.Get()
+	if shutdown {
+		return false
+	}
+
+	// We wrap this block in a func so we can defer rc.workqueue.Done.
+	err := func(key string) error {
+		defer rc.workqueue.Done(key)
+
+		// Run the route reconciliation
+		if err := rc.reconcileNodeRoutes(ctx); err != nil {
+			// Put the item back on the workqueue to handle any transient errors.
+			rc.workqueue.AddRateLimited(key)
+			klog.Infof("Couldn't reconcile node routes: %v, requeuing", err)
+			return fmt.Errorf("couldn't reconcile node routes: %w, requeuing", err)
+		}
+
+		// Finally, if no error occurs we Forget this item so it does not
+		// get queued again until another change happens.
+		rc.workqueue.Forget(key)
+		return nil
+	}(obj)
+
+	if err != nil {
+		utilruntime.HandleError(err)
+		return true
+	}
+
+	return true
+}
+
 func (rc *RouteController) reconcileNodeRoutes(ctx context.Context) error {
 	routeList, err := rc.routes.ListRoutes(ctx, rc.clusterName)
 	if err != nil {

staging/src/k8s.io/cloud-provider/controllers/route/route_controller_test.go

@@ -22,6 +22,10 @@ import (
 	"testing"
 	"time"
 
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/controller-manager/pkg/features"
+
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -513,6 +517,91 @@ func TestReconcile(t *testing.T) {
 	}
 }
 
+func TestHandleNodeUpdate(t *testing.T) {
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CloudControllerManagerWatchBasedRoutesReconciliation, true)
+
+	cluster := "my-k8s"
+	node := v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-1", UID: "01"}, Spec: v1.NodeSpec{PodCIDR: "10.120.0.0/24", PodCIDRs: []string{"10.120.0.0/24"}}, Status: v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "10.0.1.1"}}}}
+
+	testCases := []struct {
+		description           string
+		clientset             *fake.Clientset
+		updatedNode           v1.Node
+		expectedWorkqueueItem string
+	}{
+		{
+			description:           "internal IP updated",
+			clientset:             fake.NewClientset(&v1.NodeList{Items: []v1.Node{node}}),
+			updatedNode:           v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-1", UID: "01"}, Spec: v1.NodeSpec{PodCIDR: "10.120.0.0/24", PodCIDRs: []string{"10.120.0.0/24"}}, Status: v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "10.0.1.2"}}}},
+			expectedWorkqueueItem: "routes",
+		},
+		{
+			description:           "pod CIDR updated",
+			clientset:             fake.NewClientset(&v1.NodeList{Items: []v1.Node{node}}),
+			updatedNode:           v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-1", UID: "01"}, Spec: v1.NodeSpec{PodCIDR: "10.121.0.0/24", PodCIDRs: []string{"10.121.0.0/24"}}, Status: v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "10.0.1.1"}}}},
+			expectedWorkqueueItem: "routes",
+		},
+		{
+			description: "node object not updated",
+			clientset:   fake.NewClientset(&v1.NodeList{Items: []v1.Node{node}}),
+			updatedNode: node,
+		},
+		{
+			description: "unrelated node update",
+			clientset:   fake.NewClientset(&v1.NodeList{Items: []v1.Node{node}}),
+			updatedNode: v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "node-1",
+					UID:  "01",
+				},
+				Spec: v1.NodeSpec{
+					PodCIDR:  "10.120.0.0/24",
+					PodCIDRs: []string{"10.120.0.0/24"},
+				},
+				Status: v1.NodeStatus{
+					Addresses: []v1.NodeAddress{
+						{
+							Type:    v1.NodeInternalIP,
+							Address: "10.0.1.1",
+						},
+					},
+					Images: []v1.ContainerImage{
+						{
+							Names: []string{
+								"registry.k8s.io/pause:latest",
+							},
+							SizeBytes: 239840,
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.description, func(t *testing.T) {
+			cloud := &fakecloud.Cloud{RouteMap: make(map[string]*fakecloud.Route)}
+			routes, ok := cloud.Routes()
+			assert.True(t, ok, "fakecloud failed to run Routes()")
+
+			cidrs := make([]*net.IPNet, 0)
+			_, cidr, _ := netutils.ParseCIDRSloppy("10.120.0.0/16")
+			cidrs = append(cidrs, cidr)
+
+			informerFactory := informers.NewSharedInformerFactory(testCase.clientset, 0)
+			rc := New(routes, testCase.clientset, informerFactory.Core().V1().Nodes(), cluster, cidrs)
+			require.NotNil(t, rc.workqueue)
+
+			rc.handleNodeUpdate(&node, &testCase.updatedNode)
+
+			if testCase.expectedWorkqueueItem != "" {
+				item, shutdown := rc.workqueue.Get()
+				require.False(t, shutdown, "workqueue is shutdown")
+				assert.Equal(t, testCase.expectedWorkqueueItem, item, "unexpected item from workqueue")
+			}
+		})
+	}
+}
+
 func routeListEqual(list1, list2 []*cloudprovider.Route) bool {
 	if len(list1) != len(list2) {
 		return false

staging/src/k8s.io/controller-manager/pkg/features/kube_features.go

@@ -31,6 +31,20 @@ import (
 // of code conflicts because changes are more likely to be scattered
 // across the file.
 const (
+	// Every feature gate should add method here following this template:
+	//
+	// // owner: @username
+	// MyFeature featuregate.Feature = "MyFeature"
+	//
+	// Feature gates should be listed in alphabetical, case-sensitive
+	// (upper before any lower case character) order. This reduces the risk
+	// of code conflicts because changes are more likely to be scattered
+	// across the file.
+
+	// owner: @lukasmetzner
+	// Use watch based route controller reconcilation instead of frequent periodic reconciliation.
+	CloudControllerManagerWatchBasedRoutesReconciliation featuregate.Feature = "CloudControllerManagerWatchBasedRoutesReconciliation"
+
 	// owner: @nckturner
 	// kep:  http://kep.k8s.io/2699
 	// Enable webhook in cloud controller manager
@@ -44,6 +58,10 @@ func SetupCurrentKubernetesSpecificFeatureGates(featuregates featuregate.Mutable
 // versionedCloudPublicFeatureGates consists of versioned cloud-specific feature keys.
 // To add a new feature, define a key for it above and add it here.
 var versionedCloudPublicFeatureGates = map[featuregate.Feature]featuregate.VersionedSpecs{
+	CloudControllerManagerWatchBasedRoutesReconciliation: {
+		// TODO: Update to 1.34 once the Kubernetes version is changed
+		{Version: version.MustParse("1.33"), Default: false, PreRelease: featuregate.Alpha},
+	},
 	CloudControllerManagerWebhook: {
 		{Version: version.MustParse("1.27"), Default: false, PreRelease: featuregate.Alpha},
 	},