From a5055cc407b721440685d017865ac982de6a470b Mon Sep 17 00:00:00 2001 From: lukasmetzner Date: Wed, 9 Apr 2025 15:20:42 +0200 Subject: [PATCH] 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". --- .../controllers/route/route_controller.go | 123 ++++++++++++++++-- .../route/route_controller_test.go | 89 +++++++++++++ .../pkg/features/kube_features.go | 18 +++ 3 files changed, 220 insertions(+), 10 deletions(-) diff --git a/staging/src/k8s.io/cloud-provider/controllers/route/route_controller.go b/staging/src/k8s.io/cloud-provider/controllers/route/route_controller.go index fe13c5feda5..f1e98c16dbd 100644 --- a/staging/src/k8s.io/cloud-provider/controllers/route/route_controller.go +++ b/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 { diff --git a/staging/src/k8s.io/cloud-provider/controllers/route/route_controller_test.go b/staging/src/k8s.io/cloud-provider/controllers/route/route_controller_test.go index b103c4b4d55..8c096ba8b90 100644 --- a/staging/src/k8s.io/cloud-provider/controllers/route/route_controller_test.go +++ b/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 diff --git a/staging/src/k8s.io/controller-manager/pkg/features/kube_features.go b/staging/src/k8s.io/controller-manager/pkg/features/kube_features.go index e6a1f0067fa..5423e351ba5 100644 --- a/staging/src/k8s.io/controller-manager/pkg/features/kube_features.go +++ b/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}, },