From fa34df7e5a3f01a29c51ca5af547e750e9524b68 Mon Sep 17 00:00:00 2001 From: ujjwal-devzy Date: Wed, 15 Jul 2026 10:11:02 +0530 Subject: [PATCH] Add node maintenance window controller Adds a kube-controller-manager controller that cordons and drains nodes during administrator-declared recurring maintenance windows (annotated per-node) and automatically returns them to service once the window closes. Cluster-wide behavior (max concurrent nodes in maintenance, drain timeout, eviction grace period) is tunable via a ConfigMap so it doesn't require a manager restart to change. --- .../app/controller_descriptor.go | 1 + cmd/kube-controller-manager/app/core.go | 27 ++ .../names/controller_names.go | 1 + pkg/controller/nodemaintenance/concurrency.go | 74 ++++ .../nodemaintenance/concurrency_test.go | 105 ++++++ pkg/controller/nodemaintenance/controller.go | 326 ++++++++++++++++ .../nodemaintenance/controller_test.go | 300 +++++++++++++++ pkg/controller/nodemaintenance/cordon.go | 75 ++++ pkg/controller/nodemaintenance/cordon_test.go | 118 ++++++ pkg/controller/nodemaintenance/doc.go | 25 ++ pkg/controller/nodemaintenance/drain.go | 248 ++++++++++++ pkg/controller/nodemaintenance/drain_test.go | 267 +++++++++++++ pkg/controller/nodemaintenance/metrics.go | 70 ++++ pkg/controller/nodemaintenance/policy.go | 69 ++++ pkg/controller/nodemaintenance/policy_test.go | 108 ++++++ pkg/controller/nodemaintenance/schedule.go | 139 +++++++ .../nodemaintenance/schedule_test.go | 354 ++++++++++++++++++ pkg/controller/nodemaintenance/types.go | 90 +++++ pkg/controller/nodemaintenance/validation.go | 61 +++ .../nodemaintenance/validation_test.go | 119 ++++++ 20 files changed, 2577 insertions(+) create mode 100644 pkg/controller/nodemaintenance/concurrency.go create mode 100644 pkg/controller/nodemaintenance/concurrency_test.go create mode 100644 pkg/controller/nodemaintenance/controller.go create mode 100644 pkg/controller/nodemaintenance/controller_test.go create mode 100644 pkg/controller/nodemaintenance/cordon.go create mode 100644 pkg/controller/nodemaintenance/cordon_test.go create mode 100644 pkg/controller/nodemaintenance/doc.go create mode 100644 pkg/controller/nodemaintenance/drain.go create mode 100644 pkg/controller/nodemaintenance/drain_test.go create mode 100644 pkg/controller/nodemaintenance/metrics.go create mode 100644 pkg/controller/nodemaintenance/policy.go create mode 100644 pkg/controller/nodemaintenance/policy_test.go create mode 100644 pkg/controller/nodemaintenance/schedule.go create mode 100644 pkg/controller/nodemaintenance/schedule_test.go create mode 100644 pkg/controller/nodemaintenance/types.go create mode 100644 pkg/controller/nodemaintenance/validation.go create mode 100644 pkg/controller/nodemaintenance/validation_test.go diff --git a/cmd/kube-controller-manager/app/controller_descriptor.go b/cmd/kube-controller-manager/app/controller_descriptor.go index fa82992f8174d..d4b65fe221e1a 100644 --- a/cmd/kube-controller-manager/app/controller_descriptor.go +++ b/cmd/kube-controller-manager/app/controller_descriptor.go @@ -188,6 +188,7 @@ func NewControllerDescriptors() map[string]*ControllerDescriptor { register(newEndpointSliceMirroringControllerDescriptor()) register(newReplicationControllerDescriptor()) register(newPodGarbageCollectorControllerDescriptor()) + register(newNodeMaintenanceControllerDescriptor()) register(newResourceQuotaControllerDescriptor()) register(newNamespaceControllerDescriptor()) register(newServiceAccountControllerDescriptor()) diff --git a/cmd/kube-controller-manager/app/core.go b/cmd/kube-controller-manager/app/core.go index 309286be8bbee..f4ba0c794c5ff 100644 --- a/cmd/kube-controller-manager/app/core.go +++ b/cmd/kube-controller-manager/app/core.go @@ -50,6 +50,7 @@ import ( nodeipamconfig "k8s.io/kubernetes/pkg/controller/nodeipam/config" "k8s.io/kubernetes/pkg/controller/nodeipam/ipam" lifecyclecontroller "k8s.io/kubernetes/pkg/controller/nodelifecycle" + "k8s.io/kubernetes/pkg/controller/nodemaintenance" "k8s.io/kubernetes/pkg/controller/podgc" replicationcontroller "k8s.io/kubernetes/pkg/controller/replication" "k8s.io/kubernetes/pkg/controller/resourceclaim" @@ -552,6 +553,32 @@ func newReplicationController(ctx context.Context, controllerContext ControllerC }, controllerName), nil } +func newNodeMaintenanceControllerDescriptor() *ControllerDescriptor { + return &ControllerDescriptor{ + name: names.NodeMaintenanceController, + aliases: []string{"node-maintenance"}, + constructor: newNodeMaintenanceController, + } +} + +func newNodeMaintenanceController(ctx context.Context, controllerContext ControllerContext, controllerName string) (Controller, error) { + client, err := controllerContext.NewClient("node-maintenance-controller") + if err != nil { + return nil, err + } + + nmc := nodemaintenance.NewController( + ctx, + client, + controllerContext.InformerFactory.Core().V1().Nodes(), + controllerContext.InformerFactory.Core().V1().ConfigMaps(), + controllerContext.InformerFactory.Core().V1().Pods(), + ) + return newControllerLoop(func(ctx context.Context) { + nmc.Run(ctx, 1) + }, controllerName), nil +} + func newPodGarbageCollectorControllerDescriptor() *ControllerDescriptor { return &ControllerDescriptor{ name: names.PodGarbageCollectorController, diff --git a/cmd/kube-controller-manager/names/controller_names.go b/cmd/kube-controller-manager/names/controller_names.go index baaa97c076b66..81d8025a7b69a 100644 --- a/cmd/kube-controller-manager/names/controller_names.go +++ b/cmd/kube-controller-manager/names/controller_names.go @@ -88,4 +88,5 @@ const ( ServiceCIDRController = "service-cidr-controller" StorageVersionMigratorController = "storage-version-migrator-controller" SELinuxWarningController = "selinux-warning-controller" + NodeMaintenanceController = "node-maintenance-controller" ) diff --git a/pkg/controller/nodemaintenance/concurrency.go b/pkg/controller/nodemaintenance/concurrency.go new file mode 100644 index 0000000000000..4325fbca2b862 --- /dev/null +++ b/pkg/controller/nodemaintenance/concurrency.go @@ -0,0 +1,74 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import "sync" + +// concurrencyLimiter bounds how many nodes may be under active maintenance +// (cordoned + draining) at the same time, cluster-wide. It is safe for +// concurrent use. +type concurrencyLimiter struct { + mu sync.Mutex + tokens chan struct{} +} + +// newConcurrencyLimiter returns a limiter that allows up to max concurrent +// holders. +func newConcurrencyLimiter(max int) *concurrencyLimiter { + if max < 0 { + max = 0 + } + return &concurrencyLimiter{tokens: make(chan struct{}, max)} +} + +// TryAcquire attempts to claim a slot without blocking. It returns false if +// the limiter is already at capacity. +func (l *concurrencyLimiter) TryAcquire() bool { + l.mu.Lock() + tokens := l.tokens + l.mu.Unlock() + + select { + case tokens <- struct{}{}: + return true + default: + return false + } +} + +// Release frees a previously-acquired slot. +func (l *concurrencyLimiter) Release() { + l.mu.Lock() + tokens := l.tokens + l.mu.Unlock() + + select { + case <-tokens: + default: + } +} + +// Resize changes the limiter's capacity to max, for example after the +// maintenance policy ConfigMap is edited. +func (l *concurrencyLimiter) Resize(max int) { + if max < 0 { + max = 0 + } + l.mu.Lock() + defer l.mu.Unlock() + l.tokens = make(chan struct{}, max) +} diff --git a/pkg/controller/nodemaintenance/concurrency_test.go b/pkg/controller/nodemaintenance/concurrency_test.go new file mode 100644 index 0000000000000..2757afda4e67e --- /dev/null +++ b/pkg/controller/nodemaintenance/concurrency_test.go @@ -0,0 +1,105 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import "testing" + +func TestConcurrencyLimiter_AcquireUpToCapacity(t *testing.T) { + l := newConcurrencyLimiter(2) + + if !l.TryAcquire() { + t.Fatal("first TryAcquire() = false, want true") + } + if !l.TryAcquire() { + t.Fatal("second TryAcquire() = false, want true") + } + if l.TryAcquire() { + t.Fatal("third TryAcquire() = true, want false (at capacity)") + } +} + +func TestConcurrencyLimiter_ReleaseFreesASlot(t *testing.T) { + l := newConcurrencyLimiter(1) + + if !l.TryAcquire() { + t.Fatal("TryAcquire() = false, want true") + } + if l.TryAcquire() { + t.Fatal("TryAcquire() while full = true, want false") + } + + l.Release() + + if !l.TryAcquire() { + t.Fatal("TryAcquire() after Release() = false, want true") + } +} + +func TestConcurrencyLimiter_ZeroCapacityNeverAcquires(t *testing.T) { + l := newConcurrencyLimiter(0) + if l.TryAcquire() { + t.Fatal("TryAcquire() on a zero-capacity limiter = true, want false") + } +} + +func TestConcurrencyLimiter_NegativeCapacityClampsToZero(t *testing.T) { + l := newConcurrencyLimiter(-3) + if l.TryAcquire() { + t.Fatal("TryAcquire() on a negative-capacity limiter = true, want false") + } +} + +func TestConcurrencyLimiter_ResizeUp(t *testing.T) { + l := newConcurrencyLimiter(1) + if !l.TryAcquire() { + t.Fatal("TryAcquire() = false, want true") + } + + l.Resize(3) + + acquired := 0 + for i := 0; i < 3; i++ { + if l.TryAcquire() { + acquired++ + } + } + if acquired != 3 { + t.Errorf("acquired %d slots after resizing to 3, want 3", acquired) + } +} + +func TestConcurrencyLimiter_ResizeDown(t *testing.T) { + l := newConcurrencyLimiter(5) + l.Resize(1) + + if !l.TryAcquire() { + t.Fatal("first TryAcquire() after shrinking = false, want true") + } + if l.TryAcquire() { + t.Fatal("second TryAcquire() after shrinking to 1 = true, want false") + } +} + +func TestConcurrencyLimiter_ReleaseWithoutAcquireIsSafe(t *testing.T) { + l := newConcurrencyLimiter(1) + // Should not panic or block. + l.Release() + + if !l.TryAcquire() { + t.Fatal("TryAcquire() after a spurious Release() = false, want true") + } +} diff --git a/pkg/controller/nodemaintenance/controller.go b/pkg/controller/nodemaintenance/controller.go new file mode 100644 index 0000000000000..085827b2793dd --- /dev/null +++ b/pkg/controller/nodemaintenance/controller.go @@ -0,0 +1,326 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "context" + "fmt" + "sync" + "time" + + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + coreinformers "k8s.io/client-go/informers/core/v1" + clientset "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/scheme" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + corelisters "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/workqueue" + "k8s.io/klog/v2" +) + +const requeueAtCapacityInterval = 15 * time.Second + +// Controller cordons and drains nodes during their declared maintenance +// windows, and uncordons them again once the window closes. +type Controller struct { + kubeClient clientset.Interface + + nodeLister corelisters.NodeLister + nodeListerSynced cache.InformerSynced + + configMapLister corelisters.ConfigMapLister + configMapListerSynced cache.InformerSynced + + podLister corelisters.PodLister + + queue workqueue.TypedRateLimitingInterface[string] + recorder record.EventRecorder + + policyMu sync.RWMutex + policy Policy + + limiter *concurrencyLimiter +} + +// NewController builds a Controller. Callers must call Run to start it. +func NewController( + ctx context.Context, + kubeClient clientset.Interface, + nodeInformer coreinformers.NodeInformer, + configMapInformer coreinformers.ConfigMapInformer, + podInformer coreinformers.PodInformer, +) *Controller { + eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx)) + eventBroadcaster.StartStructuredLogging(0) + eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")}) + recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: ControllerName}) + + c := &Controller{ + kubeClient: kubeClient, + nodeLister: nodeInformer.Lister(), + nodeListerSynced: nodeInformer.Informer().HasSynced, + configMapLister: configMapInformer.Lister(), + configMapListerSynced: configMapInformer.Informer().HasSynced, + podLister: podInformer.Lister(), + queue: workqueue.NewTypedRateLimitingQueueWithConfig( + workqueue.DefaultTypedControllerRateLimiter[string](), + workqueue.TypedRateLimitingQueueConfig[string]{Name: "node_maintenance"}, + ), + recorder: recorder, + } + // The limiter is resized to the real configured value once the policy + // ConfigMap informer has synced; see reloadPolicy. + c.limiter = newConcurrencyLimiter(c.policy.MaxConcurrentNodes) + + nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: c.enqueueNode, + UpdateFunc: func(_, cur interface{}) { c.enqueueNode(cur) }, + }) + + configMapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{ + FilterFunc: c.isPolicyConfigMap, + Handler: cache.ResourceEventHandlerFuncs{ + AddFunc: func(interface{}) { c.reloadPolicy(ctx) }, + UpdateFunc: func(_, _ interface{}) { c.reloadPolicy(ctx) }, + DeleteFunc: func(interface{}) { c.reloadPolicy(ctx) }, + }, + }) + + RegisterMetrics() + return c +} + +// Run starts the controller's workers and blocks until ctx is canceled. +func (c *Controller) Run(ctx context.Context, workers int) { + defer utilruntime.HandleCrashWithContext(ctx) + + logger := klog.FromContext(ctx) + logger.Info("Starting node maintenance controller") + defer logger.Info("Shutting down node maintenance controller") + defer c.queue.ShutDown() + + if !cache.WaitForNamedCacheSyncWithContext(ctx, c.nodeListerSynced, c.configMapListerSynced) { + return + } + + // Seed the policy from whatever ConfigMap state already exists before we + // start reconciling nodes, so the first pass uses real limits rather + // than construction-time defaults. + c.policyMu.Lock() + c.policy = c.loadPolicy(ctx) + c.policyMu.Unlock() + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + wait.UntilWithContext(ctx, c.runWorker, time.Second) + }() + } + + <-ctx.Done() + wg.Wait() +} + +func (c *Controller) runWorker(ctx context.Context) { + for c.processNextWorkItem(ctx) { + } +} + +func (c *Controller) processNextWorkItem(ctx context.Context) bool { + key, shutdown := c.queue.Get() + if shutdown { + return false + } + defer c.queue.Done(key) + + if err := c.syncHandler(ctx, key); err != nil { + utilruntime.HandleErrorWithContext(ctx, err, "Error syncing node, requeueing", "node", klog.KRef("", key)) + c.queue.AddRateLimited(key) + return true + } + + c.queue.Forget(key) + return true +} + +func (c *Controller) syncHandler(ctx context.Context, key string) error { + logger := klog.FromContext(ctx) + + node, err := c.nodeLister.Get(key) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return err + } + + windowAnnotation, hasWindow := node.Annotations[AnnotationWindow] + if !hasWindow { + if node.Annotations[AnnotationCordonedBy] == ControllerName { + return UncordonNode(ctx, c.kubeClient, node) + } + return nil + } + + window, err := ParseWindow(windowAnnotation) + if err != nil { + c.recorder.Eventf(node, v1.EventTypeWarning, "InvalidMaintenanceWindow", "Could not parse maintenance window: %v", err) + return nil + } + + now := time.Now() + active, err := window.IsActive(now) + if err != nil { + c.recorder.Eventf(node, v1.EventTypeWarning, "InvalidMaintenanceWindow", "Could not evaluate maintenance window: %v", err) + return nil + } + + if !active { + if node.Annotations[AnnotationCordonedBy] == ControllerName { + if err := UncordonNode(ctx, c.kubeClient, node); err != nil { + return err + } + } + return c.requeueForNextTransition(key, window, now) + } + + if !c.limiter.TryAcquire() { + logger.V(4).Info("At maintenance concurrency limit, will retry", "node", klog.KObj(node)) + c.queue.AddAfter(key, requeueAtCapacityInterval) + return nil + } + + if err := CordonNode(ctx, c.kubeClient, node); err != nil { + cordonErrorsTotal.Inc() + return err + } + + policy := c.effectivePolicy() + drainErr := DrainNode(ctx, c.kubeClient, c.podLister, node, policy) + c.limiter.Release() + if drainErr != nil { + return drainErr + } + + if err := c.recordDrainTime(ctx, node, now); err != nil { + logger.Error(err, "Failed to record last drain time annotation", "node", klog.KObj(node)) + } + + return c.requeueForNextTransition(key, window, now) +} + +func (c *Controller) requeueForNextTransition(key string, window *Window, now time.Time) error { + next, err := window.NextTransition(now) + if err != nil { + return err + } + c.queue.AddAfter(key, next.Sub(now)) + return nil +} + +func (c *Controller) recordDrainTime(ctx context.Context, node *v1.Node, at time.Time) error { + patch := []byte(fmt.Sprintf(`{"metadata":{"annotations":{%q:%q}}}`, AnnotationLastDrainTime, at.UTC().Format(time.RFC3339))) + _, err := c.kubeClient.CoreV1().Nodes().Patch(ctx, node.Name, types.StrategicMergePatchType, patch, metav1.PatchOptions{}) + return err +} + +func (c *Controller) effectivePolicy() Policy { + c.policyMu.RLock() + defer c.policyMu.RUnlock() + return c.policy +} + +func (c *Controller) loadPolicy(ctx context.Context) Policy { + cm, err := c.configMapLister.ConfigMaps(PolicyConfigMapNamespace).Get(PolicyConfigMapName) + if err != nil { + if !apierrors.IsNotFound(err) { + klog.FromContext(ctx).Error(err, "Failed to load node maintenance policy, using defaults") + } + return DefaultPolicy() + } + + p, err := ParsePolicy(cm) + if err != nil { + klog.FromContext(ctx).Error(err, "Invalid node maintenance policy ConfigMap, using defaults") + return DefaultPolicy() + } + return p +} + +func (c *Controller) reloadPolicy(ctx context.Context) { + p := c.loadPolicy(ctx) + + c.policyMu.Lock() + c.policy = p + c.policyMu.Unlock() + + c.limiter.Resize(p.MaxConcurrentNodes) + + klog.FromContext(ctx).Info("Reloaded node maintenance policy", "maxConcurrentNodes", p.MaxConcurrentNodes) + c.enqueueAllNodes() +} + +func (c *Controller) enqueueAllNodes() { + nodes, err := c.nodeLister.List(labels.Everything()) + if err != nil { + utilruntime.HandleError(fmt.Errorf("listing nodes to re-evaluate after policy change: %w", err)) + return + } + for _, node := range nodes { + c.queue.Add(node.Name) + } +} + +func (c *Controller) enqueueNode(obj interface{}) { + node, ok := obj.(*v1.Node) + if !ok { + tombstone, ok := obj.(cache.DeletedFinalStateUnknown) + if !ok { + return + } + node, ok = tombstone.Obj.(*v1.Node) + if !ok { + return + } + } + c.queue.Add(node.Name) +} + +func (c *Controller) isPolicyConfigMap(obj interface{}) bool { + cm, ok := obj.(*v1.ConfigMap) + if !ok { + tombstone, ok := obj.(cache.DeletedFinalStateUnknown) + if !ok { + return false + } + cm, ok = tombstone.Obj.(*v1.ConfigMap) + if !ok { + return false + } + } + return cm.Namespace == PolicyConfigMapNamespace && cm.Name == PolicyConfigMapName +} diff --git a/pkg/controller/nodemaintenance/controller_test.go b/pkg/controller/nodemaintenance/controller_test.go new file mode 100644 index 0000000000000..44e038875e88b --- /dev/null +++ b/pkg/controller/nodemaintenance/controller_test.go @@ -0,0 +1,300 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "context" + "encoding/json" + "testing" + "time" + + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes/fake" + corelisters "k8s.io/client-go/listers/core/v1" + clienttesting "k8s.io/client-go/testing" +) + +// staticConfigMapLister is a minimal corelisters.ConfigMapLister that always +// returns a single fixed ConfigMap, used to drive reloadPolicy directly in +// tests without needing to wait on informer resync. +type staticConfigMapLister struct { + cm *v1.ConfigMap +} + +func (s staticConfigMapLister) List(_ labels.Selector) ([]*v1.ConfigMap, error) { + return []*v1.ConfigMap{s.cm}, nil +} + +func (s staticConfigMapLister) ConfigMaps(string) corelisters.ConfigMapNamespaceLister { + return staticConfigMapNamespaceLister{cm: s.cm} +} + +type staticConfigMapNamespaceLister struct { + cm *v1.ConfigMap +} + +func (s staticConfigMapNamespaceLister) List(_ labels.Selector) ([]*v1.ConfigMap, error) { + return []*v1.ConfigMap{s.cm}, nil +} + +func (s staticConfigMapNamespaceLister) Get(name string) (*v1.ConfigMap, error) { + return s.cm, nil +} + +func newTestControllerWithObjects(t *testing.T, objects ...runtime.Object) (*Controller, *fake.Clientset) { + t.Helper() + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + client := fake.NewSimpleClientset(objects...) + factory := informers.NewSharedInformerFactory(client, 0) + + c := NewController( + ctx, + client, + factory.Core().V1().Nodes(), + factory.Core().V1().ConfigMaps(), + factory.Core().V1().Pods(), + ) + + factory.Start(ctx.Done()) + factory.WaitForCacheSync(ctx.Done()) + + // Seed a known-good policy directly, bypassing the ConfigMap-driven + // cold-start path so individual tests don't each need to provision a + // policy ConfigMap just to get a usable concurrency limit. + c.policyMu.Lock() + c.policy = DefaultPolicy() + c.policyMu.Unlock() + c.limiter = newConcurrencyLimiter(c.policy.MaxConcurrentNodes) + + // Informer startup issues its own list/watch calls; clear those so tests + // only see actions caused by the syncHandler call(s) they make. + client.ClearActions() + + return c, client +} + +func windowAnnotationValue(t *testing.T, w Window) string { + t.Helper() + data, err := json.Marshal(w) + if err != nil { + t.Fatalf("failed to marshal window: %v", err) + } + return string(data) +} + +func hasActionVerbResource(actions []clienttesting.Action, verb, resource string) bool { + for _, action := range actions { + if action.GetVerb() == verb && action.GetResource().Resource == resource { + return true + } + } + return false +} + +func TestSyncHandler_NodeNotFound(t *testing.T) { + c, client := newTestControllerWithObjects(t) + + if err := c.syncHandler(context.Background(), "does-not-exist"); err != nil { + t.Fatalf("syncHandler returned unexpected error: %v", err) + } + if len(client.Actions()) != 0 { + t.Errorf("syncHandler on a missing node issued actions: %v", client.Actions()) + } +} + +func TestSyncHandler_NoWindowAnnotation_NotCordoned_NoOp(t *testing.T) { + node := newTestNode("node-1", false, nil) + c, client := newTestControllerWithObjects(t, node) + + if err := c.syncHandler(context.Background(), "node-1"); err != nil { + t.Fatalf("syncHandler returned unexpected error: %v", err) + } + if hasActionVerbResource(client.Actions(), "patch", "nodes") || hasActionVerbResource(client.Actions(), "update", "nodes") { + t.Errorf("syncHandler modified a node with no window annotation: %v", client.Actions()) + } +} + +func TestSyncHandler_WindowRemoved_UncordonsOwnedNode(t *testing.T) { + node := newTestNode("node-1", true, map[string]string{AnnotationCordonedBy: ControllerName}) + c, client := newTestControllerWithObjects(t, node) + + if err := c.syncHandler(context.Background(), "node-1"); err != nil { + t.Fatalf("syncHandler returned unexpected error: %v", err) + } + + updated, err := client.CoreV1().Nodes().Get(context.Background(), "node-1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to fetch node: %v", err) + } + if updated.Spec.Unschedulable { + t.Error("node still unschedulable after window annotation was removed") + } +} + +func TestSyncHandler_InvalidWindowAnnotation_NoErrorNoChange(t *testing.T) { + node := newTestNode("node-1", false, map[string]string{AnnotationWindow: "{not valid json"}) + c, client := newTestControllerWithObjects(t, node) + + if err := c.syncHandler(context.Background(), "node-1"); err != nil { + t.Fatalf("syncHandler returned unexpected error for an invalid window: %v", err) + } + if hasActionVerbResource(client.Actions(), "patch", "nodes") { + t.Errorf("syncHandler cordoned a node with an invalid window annotation: %v", client.Actions()) + } +} + +func TestSyncHandler_ActiveWindow_CordonsAndDrains(t *testing.T) { + now := time.Now().UTC() + activeWindow := Window{ + Start: now.Add(-time.Hour).Format("15:04"), + End: now.Add(time.Hour).Format("15:04"), + } + node := newTestNode("node-1", false, map[string]string{AnnotationWindow: windowAnnotationValue(t, activeWindow)}) + pod := newTestPod("app-pod", v1.PodRunning, "node-1", nil, "ReplicaSet") + + c, client := newTestControllerWithObjects(t, node, pod) + + var evicted bool + client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + if action.(clienttesting.CreateActionImpl).GetSubresource() != "eviction" { + return false, nil, nil + } + evicted = true + return true, nil, nil + }) + client.PrependReactor("get", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + if evicted && action.(clienttesting.GetActionImpl).GetName() == pod.Name { + return true, nil, apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, pod.Name) + } + return false, nil, nil + }) + + if err := c.syncHandler(context.Background(), "node-1"); err != nil { + t.Fatalf("syncHandler returned unexpected error: %v", err) + } + + updated, err := client.CoreV1().Nodes().Get(context.Background(), "node-1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to fetch node: %v", err) + } + if !updated.Spec.Unschedulable { + t.Error("node was not cordoned during an active maintenance window") + } + if got := updated.Annotations[AnnotationCordonedBy]; got != ControllerName { + t.Errorf("node annotation %s = %q, want %q", AnnotationCordonedBy, got, ControllerName) + } + if _, ok := updated.Annotations[AnnotationLastDrainTime]; !ok { + t.Error("node missing last-drain-time annotation after a successful drain") + } +} + +func TestSyncHandler_InactiveWindow_UncordonsOwnedNode(t *testing.T) { + now := time.Now().UTC() + // A window that closed an hour ago and won't reopen for another hour. + inactiveWindow := Window{ + Start: now.Add(time.Hour).Format("15:04"), + End: now.Add(2 * time.Hour).Format("15:04"), + } + node := newTestNode("node-1", true, map[string]string{ + AnnotationWindow: windowAnnotationValue(t, inactiveWindow), + AnnotationCordonedBy: ControllerName, + }) + c, client := newTestControllerWithObjects(t, node) + + if err := c.syncHandler(context.Background(), "node-1"); err != nil { + t.Fatalf("syncHandler returned unexpected error: %v", err) + } + + updated, err := client.CoreV1().Nodes().Get(context.Background(), "node-1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to fetch node: %v", err) + } + if updated.Spec.Unschedulable { + t.Error("node still cordoned outside its maintenance window") + } +} + +func TestSyncHandler_AtCapacity_RequeuesWithoutCordoning(t *testing.T) { + now := time.Now().UTC() + activeWindow := Window{ + Start: now.Add(-time.Hour).Format("15:04"), + End: now.Add(time.Hour).Format("15:04"), + } + node := newTestNode("node-b", false, map[string]string{AnnotationWindow: windowAnnotationValue(t, activeWindow)}) + + c, client := newTestControllerWithObjects(t, node) + + // Simulate another node already holding the sole concurrency slot. + if !c.limiter.TryAcquire() { + t.Fatal("failed to pre-acquire the concurrency slot for test setup") + } + + if err := c.syncHandler(context.Background(), "node-b"); err != nil { + t.Fatalf("syncHandler returned unexpected error: %v", err) + } + + updated, err := client.CoreV1().Nodes().Get(context.Background(), "node-b", metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to fetch node-b: %v", err) + } + if updated.Spec.Unschedulable { + t.Error("node-b should not have been cordoned while at the concurrency limit") + } + if c.queue.Len() == 0 { + t.Error("node-b should have been requeued after failing to acquire a concurrency slot") + } +} + +func TestSyncHandler_ReloadPolicyResizesLimiter(t *testing.T) { + c, _ := newTestControllerWithObjects(t) + + if !c.limiter.TryAcquire() { + t.Fatal("expected to acquire the single default slot") + } + if c.limiter.TryAcquire() { + t.Fatal("expected the limiter to be at capacity") + } + c.limiter.Release() + + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: PolicyConfigMapName, Namespace: PolicyConfigMapNamespace}, + Data: map[string]string{"maxConcurrentNodes": "2"}, + } + // reloadPolicy is normally driven by the ConfigMap informer; call it + // directly here since the fake informer isn't wired to this object. + c.configMapLister = staticConfigMapLister{cm: cm} + c.reloadPolicy(context.Background()) + + acquired := 0 + for i := 0; i < 2; i++ { + if c.limiter.TryAcquire() { + acquired++ + } + } + if acquired != 2 { + t.Errorf("acquired %d slots after policy reload set maxConcurrentNodes=2, want 2", acquired) + } +} diff --git a/pkg/controller/nodemaintenance/cordon.go b/pkg/controller/nodemaintenance/cordon.go new file mode 100644 index 0000000000000..3088c4fbc04e4 --- /dev/null +++ b/pkg/controller/nodemaintenance/cordon.go @@ -0,0 +1,75 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "context" + "fmt" + + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + clientset "k8s.io/client-go/kubernetes" + "k8s.io/client-go/util/retry" + "k8s.io/klog/v2" +) + +// CordonNode marks node unschedulable and records that this controller was +// the one that did so, unless the node is already cordoned (in which case +// we leave it alone so we never later uncordon a node an operator cordoned +// by hand). +func CordonNode(ctx context.Context, client clientset.Interface, node *v1.Node) error { + logger := klog.FromContext(ctx) + if node.Spec.Unschedulable { + logger.V(4).Info("Node already unschedulable, not claiming ownership", "node", klog.KObj(node)) + return nil + } + + patch := []byte(fmt.Sprintf( + `{"spec":{"unschedulable":true},"metadata":{"annotations":{%q:%q}}}`, + AnnotationCordonedBy, ControllerName, + )) + _, err := client.CoreV1().Nodes().Patch(ctx, node.Name, types.StrategicMergePatchType, patch, metav1.PatchOptions{}) + if err != nil { + return fmt.Errorf("cordoning node %s: %w", node.Name, err) + } + logger.Info("Cordoned node for maintenance window", "node", klog.KObj(node)) + return nil +} + +// UncordonNode reverses CordonNode, but only if this controller was the one +// that cordoned the node in the first place. +func UncordonNode(ctx context.Context, client clientset.Interface, node *v1.Node) error { + logger := klog.FromContext(ctx) + if node.Annotations[AnnotationCordonedBy] != ControllerName { + logger.V(4).Info("Node was not cordoned by this controller, leaving as-is", "node", klog.KObj(node)) + return nil + } + + err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { + updated := node.DeepCopy() + updated.Spec.Unschedulable = false + delete(updated.Annotations, AnnotationCordonedBy) + _, updateErr := client.CoreV1().Nodes().Update(ctx, updated, metav1.UpdateOptions{}) + return updateErr + }) + if err != nil { + return fmt.Errorf("uncordoning node %s: %w", node.Name, err) + } + logger.Info("Uncordoned node after maintenance window closed", "node", klog.KObj(node)) + return nil +} diff --git a/pkg/controller/nodemaintenance/cordon_test.go b/pkg/controller/nodemaintenance/cordon_test.go new file mode 100644 index 0000000000000..cc8dc12d1467d --- /dev/null +++ b/pkg/controller/nodemaintenance/cordon_test.go @@ -0,0 +1,118 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "context" + "testing" + + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +func newTestNode(name string, unschedulable bool, annotations map[string]string) *v1.Node { + return &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Annotations: annotations, + }, + Spec: v1.NodeSpec{Unschedulable: unschedulable}, + } +} + +func TestCordonNode_AlreadyCordoned_NoOp(t *testing.T) { + node := newTestNode("node-1", true, nil) + client := fake.NewSimpleClientset(node) + + if err := CordonNode(context.Background(), client, node); err != nil { + t.Fatalf("CordonNode returned unexpected error: %v", err) + } + + for _, action := range client.Actions() { + if action.GetVerb() == "patch" { + t.Errorf("CordonNode issued an unexpected patch for an already-cordoned node: %v", action) + } + } +} + +func TestCordonNode_CordonsAndClaimsOwnership(t *testing.T) { + node := newTestNode("node-1", false, nil) + client := fake.NewSimpleClientset(node) + + if err := CordonNode(context.Background(), client, node); err != nil { + t.Fatalf("CordonNode returned unexpected error: %v", err) + } + + updated, err := client.CoreV1().Nodes().Get(context.Background(), "node-1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to fetch node after CordonNode: %v", err) + } + if !updated.Spec.Unschedulable { + t.Error("node.Spec.Unschedulable = false after CordonNode, want true") + } + if got := updated.Annotations[AnnotationCordonedBy]; got != ControllerName { + t.Errorf("node annotation %s = %q, want %q", AnnotationCordonedBy, got, ControllerName) + } +} + +func TestUncordonNode_NotCordonedByUs_NoOp(t *testing.T) { + cases := []struct { + name string + annotations map[string]string + }{ + {name: "no annotation", annotations: nil}, + {name: "cordoned by someone else", annotations: map[string]string{AnnotationCordonedBy: "some-admin"}}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + node := newTestNode("node-1", true, tc.annotations) + client := fake.NewSimpleClientset(node) + + if err := UncordonNode(context.Background(), client, node); err != nil { + t.Fatalf("UncordonNode returned unexpected error: %v", err) + } + + for _, action := range client.Actions() { + if action.GetVerb() == "update" { + t.Errorf("UncordonNode issued an unexpected update for a node it doesn't own: %v", action) + } + } + }) + } +} + +func TestUncordonNode_UncordonsAndRemovesAnnotation(t *testing.T) { + node := newTestNode("node-1", true, map[string]string{AnnotationCordonedBy: ControllerName}) + client := fake.NewSimpleClientset(node) + + if err := UncordonNode(context.Background(), client, node); err != nil { + t.Fatalf("UncordonNode returned unexpected error: %v", err) + } + + updated, err := client.CoreV1().Nodes().Get(context.Background(), "node-1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to fetch node after UncordonNode: %v", err) + } + if updated.Spec.Unschedulable { + t.Error("node.Spec.Unschedulable = true after UncordonNode, want false") + } + if _, ok := updated.Annotations[AnnotationCordonedBy]; ok { + t.Errorf("annotation %s still present after UncordonNode", AnnotationCordonedBy) + } +} diff --git a/pkg/controller/nodemaintenance/doc.go b/pkg/controller/nodemaintenance/doc.go new file mode 100644 index 0000000000000..f90dd4178a0c5 --- /dev/null +++ b/pkg/controller/nodemaintenance/doc.go @@ -0,0 +1,25 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package nodemaintenance implements a controller that cordons and drains +// nodes during administrator-declared recurring maintenance windows, and +// automatically returns them to service once the window closes. +// +// Maintenance windows are declared per-node via an annotation +// (maintenance.kubernetes.io/window) and cluster-wide defaults are sourced +// from a ConfigMap so operators can tune drain behavior without restarting +// kube-controller-manager. +package nodemaintenance diff --git a/pkg/controller/nodemaintenance/drain.go b/pkg/controller/nodemaintenance/drain.go new file mode 100644 index 0000000000000..a1eba93ba336b --- /dev/null +++ b/pkg/controller/nodemaintenance/drain.go @@ -0,0 +1,248 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "context" + "fmt" + "sync" + "time" + + v1 "k8s.io/api/core/v1" + policyv1 "k8s.io/api/policy/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + clientset "k8s.io/client-go/kubernetes" + corelisters "k8s.io/client-go/listers/core/v1" + "k8s.io/klog/v2" +) + +const ( + mirrorPodAnnotationKey = "kubernetes.io/config.mirror" + evictionPollInterval = 2 * time.Second + evictionInitialBackoff = 1 * time.Second + evictionMaxBackoff = 30 * time.Second +) + +// DrainNode evicts all evictable pods from node, respecting +// PodDisruptionBudgets, then waits for them to terminate before returning. +// Pods that haven't terminated by policy.DrainTimeout are force-deleted. +func DrainNode(ctx context.Context, client clientset.Interface, podLister corelisters.PodLister, node *v1.Node, policy Policy) error { + logger := klog.FromContext(ctx) + start := time.Now() + defer func() { + drainDuration.Observe(time.Since(start).Seconds()) + }() + + pods, err := podsOnNode(podLister, node.Name) + if err != nil { + return fmt.Errorf("listing pods on node %s: %w", node.Name, err) + } + + var evictable []*v1.Pod + for _, pod := range pods { + if shouldEvict(pod, policy) { + evictable = append(evictable, pod) + } + } + + if len(evictable) == 0 { + logger.Info("No evictable pods on node", "node", klog.KObj(node)) + return nil + } + + deadline := start.Add(policy.DrainTimeout) + + var wg sync.WaitGroup + var errs []error + for _, pod := range evictable { + wg.Add(1) + go func(pod *v1.Pod) { + defer wg.Done() + if err := evictWithRetry(ctx, client, pod, policy, deadline); err != nil { + errs = append(errs, fmt.Errorf("evicting pod %s/%s: %w", pod.Namespace, pod.Name, err)) + } + }(pod) + } + wg.Wait() + + if len(errs) > 0 { + drainErrorsTotal.Inc() + return fmt.Errorf("drain of node %s had %d error(s), first: %v", node.Name, len(errs), errs[0]) + } + + if err := waitForPodsGone(ctx, client, evictable, deadline); err != nil { + drainErrorsTotal.Inc() + return err + } + + logger.Info("Drained node for maintenance", "node", klog.KObj(node), "evictedPods", len(evictable)) + return nil +} + +// shouldEvict reports whether pod needs an explicit eviction as part of +// draining node, excluding pods that are already terminal, terminating, or +// otherwise exempt under policy. +func shouldEvict(pod *v1.Pod, policy Policy) bool { + if pod.Status.Phase == v1.PodSucceeded || pod.Status.Phase == v1.PodFailed { + return false + } + if pod.DeletionTimestamp != nil { + return false + } + if _, isMirror := pod.Annotations[mirrorPodAnnotationKey]; isMirror { + return false + } + if policy.IgnoreDaemonSets && isDaemonSetPod(pod) { + return false + } + return true +} + +func isDaemonSetPod(pod *v1.Pod) bool { + for _, ref := range pod.OwnerReferences { + if ref.Kind == "DaemonSet" { + return true + } + } + return false +} + +// evictWithRetry submits an Eviction for pod, retrying with backoff while +// it's blocked by a PodDisruptionBudget, until deadline. +func evictWithRetry(ctx context.Context, client clientset.Interface, pod *v1.Pod, policy Policy, deadline time.Time) error { + eviction := &policyv1.Eviction{ + ObjectMeta: metav1.ObjectMeta{ + Name: pod.Name, + Namespace: pod.Namespace, + }, + DeleteOptions: &metav1.DeleteOptions{ + GracePeriodSeconds: gracePeriodSeconds(policy.PodEvictionGracePeriod), + }, + } + + backoff := evictionInitialBackoff + for { + err := client.PolicyV1().Evictions(pod.Namespace).Evict(ctx, eviction) + switch { + case err == nil: + return nil + case apierrors.IsNotFound(err): + return nil + case apierrors.IsTooManyRequests(err): + // Blocked by a PodDisruptionBudget; back off and retry until the + // drain deadline, then give up gracefully so one stuck pod can't + // hold up the rest of the maintenance cycle indefinitely. + if time.Now().After(deadline) { + return nil + } + select { + case <-time.After(backoff): + case <-ctx.Done(): + return ctx.Err() + } + if backoff < evictionMaxBackoff { + backoff *= 2 + } + continue + default: + return err + } + } +} + +// waitForPodsGone blocks until every pod in pods has been deleted from the +// API server or deadline passes, force-deleting any stragglers once the +// deadline is reached. +func waitForPodsGone(ctx context.Context, client clientset.Interface, pods []*v1.Pod, deadline time.Time) error { + remaining := make(map[string]*v1.Pod, len(pods)) + for _, pod := range pods { + remaining[podKey(pod)] = pod + } + + pollErr := wait.PollUntilContextTimeout(ctx, evictionPollInterval, time.Until(deadline), true, func(ctx context.Context) (bool, error) { + for key, pod := range remaining { + _, err := client.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + delete(remaining, key) + } + } + return len(remaining) == 0, nil + }) + if pollErr == nil { + return nil + } + + deleter := &podForceDeleter{client: client} + for _, pod := range remaining { + go deleter.run(ctx, pod) + } + deleter.wait() + + return fmt.Errorf("timed out waiting for %d pod(s) to terminate, force-deleted stragglers", len(remaining)) +} + +// podForceDeleter fans out force-deletes for stragglers left after the +// graceful drain deadline passes. +type podForceDeleter struct { + client clientset.Interface + wg sync.WaitGroup +} + +func (d *podForceDeleter) run(ctx context.Context, pod *v1.Pod) { + d.wg.Add(1) + defer d.wg.Done() + forceDeletePod(ctx, d.client, pod) +} + +func (d *podForceDeleter) wait() { + d.wg.Wait() +} + +func forceDeletePod(ctx context.Context, client clientset.Interface, pod *v1.Pod) { + zero := int64(0) + err := client.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{GracePeriodSeconds: &zero}) + if err != nil && !apierrors.IsNotFound(err) { + utilruntime.HandleError(fmt.Errorf("force-deleting pod %s/%s: %w", pod.Namespace, pod.Name, err)) + } +} + +func podsOnNode(podLister corelisters.PodLister, nodeName string) ([]*v1.Pod, error) { + all, err := podLister.List(labels.Everything()) + if err != nil { + return nil, err + } + var onNode []*v1.Pod + for _, pod := range all { + if pod.Spec.NodeName == nodeName { + onNode = append(onNode, pod) + } + } + return onNode, nil +} + +func gracePeriodSeconds(d time.Duration) *int64 { + s := int64(d.Seconds()) + return &s +} + +func podKey(pod *v1.Pod) string { + return pod.Namespace + "/" + pod.Name +} diff --git a/pkg/controller/nodemaintenance/drain_test.go b/pkg/controller/nodemaintenance/drain_test.go new file mode 100644 index 0000000000000..54c7c1b9784df --- /dev/null +++ b/pkg/controller/nodemaintenance/drain_test.go @@ -0,0 +1,267 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/kubernetes/fake" + corelisters "k8s.io/client-go/listers/core/v1" + clienttesting "k8s.io/client-go/testing" + "k8s.io/client-go/tools/cache" +) + +func newTestPod(name string, phase v1.PodPhase, nodeName string, annotations map[string]string, ownerKind string) *v1.Pod { + pod := &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Annotations: annotations, + }, + Spec: v1.PodSpec{NodeName: nodeName}, + Status: v1.PodStatus{Phase: phase}, + } + if ownerKind != "" { + pod.OwnerReferences = []metav1.OwnerReference{{Kind: ownerKind, Name: "owner", Controller: boolPtr(true)}} + } + return pod +} + +func boolPtr(b bool) *bool { return &b } + +func newPodLister(pods ...*v1.Pod) corelisters.PodLister { + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + for _, pod := range pods { + _ = indexer.Add(pod) + } + return corelisters.NewPodLister(indexer) +} + +func TestShouldEvict(t *testing.T) { + policy := DefaultPolicy() + + cases := []struct { + name string + pod *v1.Pod + want bool + }{ + {name: "running pod", pod: newTestPod("p1", v1.PodRunning, "node-1", nil, ""), want: true}, + {name: "succeeded pod", pod: newTestPod("p2", v1.PodSucceeded, "node-1", nil, ""), want: false}, + {name: "failed pod", pod: newTestPod("p3", v1.PodFailed, "node-1", nil, ""), want: false}, + {name: "daemonset pod ignored", pod: newTestPod("p4", v1.PodRunning, "node-1", nil, "DaemonSet"), want: false}, + {name: "mirror pod", pod: newTestPod("p5", v1.PodRunning, "node-1", map[string]string{mirrorPodAnnotationKey: "hash"}, ""), want: false}, + {name: "replicaset pod", pod: newTestPod("p6", v1.PodRunning, "node-1", nil, "ReplicaSet"), want: true}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := shouldEvict(tc.pod, policy); got != tc.want { + t.Errorf("shouldEvict(%s) = %v, want %v", tc.name, got, tc.want) + } + }) + } +} + +func TestShouldEvict_DaemonSetsNotIgnoredWhenPolicySaysSo(t *testing.T) { + policy := DefaultPolicy() + policy.IgnoreDaemonSets = false + + pod := newTestPod("p1", v1.PodRunning, "node-1", nil, "DaemonSet") + if !shouldEvict(pod, policy) { + t.Error("shouldEvict() = false for a DaemonSet pod with IgnoreDaemonSets=false, want true") + } +} + +func TestDrainNode_NoEvictablePods(t *testing.T) { + node := newTestNode("node-1", true, nil) + lister := newPodLister( + newTestPod("ds-pod", v1.PodRunning, "node-1", nil, "DaemonSet"), + newTestPod("other-node-pod", v1.PodRunning, "node-2", nil, ""), + ) + client := fake.NewSimpleClientset() + + if err := DrainNode(context.Background(), client, lister, node, DefaultPolicy()); err != nil { + t.Fatalf("DrainNode returned unexpected error: %v", err) + } + + for _, action := range client.Actions() { + if action.GetVerb() == "create" && action.GetResource().Resource == "pods" { + t.Errorf("DrainNode issued an unexpected eviction: %v", action) + } + } +} + +func TestDrainNode_EvictsEligiblePodsOnly(t *testing.T) { + node := newTestNode("node-1", true, nil) + evictablePod := newTestPod("app-pod", v1.PodRunning, "node-1", nil, "ReplicaSet") + dsPod := newTestPod("ds-pod", v1.PodRunning, "node-1", nil, "DaemonSet") + lister := newPodLister(evictablePod, dsPod) + + client := fake.NewSimpleClientset(evictablePod, dsPod) + + var evicted []string + client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + sub := action.(clienttesting.CreateActionImpl).GetSubresource() + if sub != "eviction" { + return false, nil, nil + } + eviction := action.(clienttesting.CreateActionImpl).GetObject() + accessor, err := apimeta.Accessor(eviction) + if err != nil { + return true, nil, err + } + evicted = append(evicted, accessor.GetName()) + return true, nil, nil + }) + // Once evicted, the pod should look deleted to subsequent Gets. + client.PrependReactor("get", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + getAction := action.(clienttesting.GetActionImpl) + for _, name := range evicted { + if name == getAction.GetName() { + return true, nil, apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, name) + } + } + return false, nil, nil + }) + + policy := DefaultPolicy() + policy.DrainTimeout = 5 * time.Second + + if err := DrainNode(context.Background(), client, lister, node, policy); err != nil { + t.Fatalf("DrainNode returned unexpected error: %v", err) + } + + if len(evicted) != 1 || evicted[0] != "app-pod" { + t.Errorf("evicted pods = %v, want [app-pod]", evicted) + } +} + +func TestDrainNode_HardEvictionFailurePropagates(t *testing.T) { + node := newTestNode("node-1", true, nil) + pod := newTestPod("app-pod", v1.PodRunning, "node-1", nil, "ReplicaSet") + lister := newPodLister(pod) + + client := fake.NewSimpleClientset(pod) + client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + if action.(clienttesting.CreateActionImpl).GetSubresource() != "eviction" { + return false, nil, nil + } + return true, nil, apierrors.NewForbidden(schema.GroupResource{Resource: "pods"}, "app-pod", fmt.Errorf("denied by policy")) + }) + + policy := DefaultPolicy() + policy.DrainTimeout = 5 * time.Second + + if err := DrainNode(context.Background(), client, lister, node, policy); err == nil { + t.Fatal("DrainNode returned nil error for a Forbidden eviction, want error") + } +} + +func TestDrainNode_EvictsMultiplePodsConcurrently(t *testing.T) { + node := newTestNode("node-1", true, nil) + pods := []*v1.Pod{ + newTestPod("app-pod-1", v1.PodRunning, "node-1", nil, "ReplicaSet"), + newTestPod("app-pod-2", v1.PodRunning, "node-1", nil, "ReplicaSet"), + newTestPod("app-pod-3", v1.PodRunning, "node-1", nil, "ReplicaSet"), + } + lister := newPodLister(pods[0], pods[1], pods[2]) + client := fake.NewSimpleClientset(pods[0], pods[1], pods[2]) + + evicted := make(map[string]bool) + var mu sync.Mutex + client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + if action.(clienttesting.CreateActionImpl).GetSubresource() != "eviction" { + return false, nil, nil + } + eviction := action.(clienttesting.CreateActionImpl).GetObject() + accessor, err := apimeta.Accessor(eviction) + if err != nil { + return true, nil, err + } + mu.Lock() + evicted[accessor.GetName()] = true + mu.Unlock() + return true, nil, nil + }) + client.PrependReactor("get", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) { + getAction := action.(clienttesting.GetActionImpl) + mu.Lock() + gone := evicted[getAction.GetName()] + mu.Unlock() + if gone { + return true, nil, apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, getAction.GetName()) + } + return false, nil, nil + }) + + policy := DefaultPolicy() + policy.DrainTimeout = 5 * time.Second + + if err := DrainNode(context.Background(), client, lister, node, policy); err != nil { + t.Fatalf("DrainNode returned unexpected error: %v", err) + } + + for _, pod := range pods { + if !evicted[pod.Name] { + t.Errorf("pod %s was not evicted", pod.Name) + } + } +} + +func TestWaitForPodsGone_ReturnsNilWhenAllPodsAlreadyGone(t *testing.T) { + client := fake.NewSimpleClientset() + pod := newTestPod("gone-pod", v1.PodRunning, "node-1", nil, "ReplicaSet") + + deadline := time.Now().Add(5 * time.Second) + if err := waitForPodsGone(context.Background(), client, []*v1.Pod{pod}, deadline); err != nil { + t.Fatalf("waitForPodsGone returned unexpected error: %v", err) + } +} + +func TestWaitForPodsGone_TimesOutOnStragglers(t *testing.T) { + pod := newTestPod("stuck-pod", v1.PodRunning, "node-1", nil, "ReplicaSet") + client := fake.NewSimpleClientset(pod) + + deadline := time.Now().Add(200 * time.Millisecond) + if err := waitForPodsGone(context.Background(), client, []*v1.Pod{pod}, deadline); err == nil { + t.Fatal("waitForPodsGone returned nil error for a pod that never terminated, want a timeout error") + } +} + +func TestGracePeriodSeconds(t *testing.T) { + got := gracePeriodSeconds(45 * time.Second) + if got == nil || *got != 45 { + t.Errorf("gracePeriodSeconds(45s) = %v, want 45", got) + } +} + +func TestPodKey(t *testing.T) { + pod := newTestPod("my-pod", v1.PodRunning, "node-1", nil, "") + if got, want := podKey(pod), "default/my-pod"; got != want { + t.Errorf("podKey() = %q, want %q", got, want) + } +} diff --git a/pkg/controller/nodemaintenance/metrics.go b/pkg/controller/nodemaintenance/metrics.go new file mode 100644 index 0000000000000..aed562014689a --- /dev/null +++ b/pkg/controller/nodemaintenance/metrics.go @@ -0,0 +1,70 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "sync" + + "k8s.io/component-base/metrics" + "k8s.io/component-base/metrics/legacyregistry" +) + +const metricsSubsystem = "node_maintenance_controller" + +var ( + nodesInMaintenance = metrics.NewGauge(&metrics.GaugeOpts{ + Subsystem: metricsSubsystem, + Name: "nodes_in_maintenance", + Help: "Number of nodes currently cordoned for a maintenance window.", + StabilityLevel: metrics.ALPHA, + }) + + drainDuration = metrics.NewHistogram(&metrics.HistogramOpts{ + Subsystem: metricsSubsystem, + Name: "drain_duration_seconds", + Help: "Time taken to drain a node for maintenance.", + Buckets: []float64{1, 5, 15, 30, 60, 120, 300, 600}, + StabilityLevel: metrics.ALPHA, + }) + + drainErrorsTotal = metrics.NewCounter(&metrics.CounterOpts{ + Subsystem: metricsSubsystem, + Name: "drain_errors_total", + Help: "Total number of node drain attempts that failed.", + StabilityLevel: metrics.ALPHA, + }) + + cordonErrorsTotal = metrics.NewCounter(&metrics.CounterOpts{ + Subsystem: metricsSubsystem, + Name: "cordon_errors_total", + Help: "Total number of node cordon/uncordon attempts that failed.", + StabilityLevel: metrics.ALPHA, + }) + + registerMetrics sync.Once +) + +// RegisterMetrics registers this package's metrics with the legacy registry. +// It is safe to call multiple times. +func RegisterMetrics() { + registerMetrics.Do(func() { + legacyregistry.MustRegister(nodesInMaintenance) + legacyregistry.MustRegister(drainDuration) + legacyregistry.MustRegister(drainErrorsTotal) + legacyregistry.MustRegister(cordonErrorsTotal) + }) +} diff --git a/pkg/controller/nodemaintenance/policy.go b/pkg/controller/nodemaintenance/policy.go new file mode 100644 index 0000000000000..73449b454461f --- /dev/null +++ b/pkg/controller/nodemaintenance/policy.go @@ -0,0 +1,69 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "fmt" + "strconv" + "time" + + v1 "k8s.io/api/core/v1" +) + +// ParsePolicy decodes a Policy from the maintenance policy ConfigMap's Data +// fields, applying DefaultPolicy for any key that is unset. +func ParsePolicy(cm *v1.ConfigMap) (Policy, error) { + p := DefaultPolicy() + if cm == nil { + return p, nil + } + + if v, ok := cm.Data["maxConcurrentNodes"]; ok { + n, err := strconv.Atoi(v) + if err != nil { + return Policy{}, fmt.Errorf("maxConcurrentNodes: %w", err) + } + p.MaxConcurrentNodes = n + } + if v, ok := cm.Data["drainTimeout"]; ok { + d, err := time.ParseDuration(v) + if err != nil { + return Policy{}, fmt.Errorf("drainTimeout: %w", err) + } + p.DrainTimeout = d + } + if v, ok := cm.Data["podEvictionGracePeriod"]; ok { + d, err := time.ParseDuration(v) + if err != nil { + return Policy{}, fmt.Errorf("podEvictionGracePeriod: %w", err) + } + p.PodEvictionGracePeriod = d + } + if v, ok := cm.Data["ignoreDaemonSets"]; ok { + b, err := strconv.ParseBool(v) + if err != nil { + return Policy{}, fmt.Errorf("ignoreDaemonSets: %w", err) + } + p.IgnoreDaemonSets = b + } + + if errs := ValidatePolicy(&p); len(errs) > 0 { + return Policy{}, errs.ToAggregate() + } + + return p, nil +} diff --git a/pkg/controller/nodemaintenance/policy_test.go b/pkg/controller/nodemaintenance/policy_test.go new file mode 100644 index 0000000000000..62b50b77a5b48 --- /dev/null +++ b/pkg/controller/nodemaintenance/policy_test.go @@ -0,0 +1,108 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "testing" + "time" + + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestParsePolicy_NilConfigMap(t *testing.T) { + p, err := ParsePolicy(nil) + if err != nil { + t.Fatalf("ParsePolicy(nil) returned unexpected error: %v", err) + } + if p != DefaultPolicy() { + t.Errorf("ParsePolicy(nil) = %+v, want %+v", p, DefaultPolicy()) + } +} + +func TestParsePolicy_EmptyConfigMapUsesDefaults(t *testing.T) { + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: PolicyConfigMapName, Namespace: PolicyConfigMapNamespace}, + } + p, err := ParsePolicy(cm) + if err != nil { + t.Fatalf("ParsePolicy returned unexpected error: %v", err) + } + if p != DefaultPolicy() { + t.Errorf("ParsePolicy(empty) = %+v, want %+v", p, DefaultPolicy()) + } +} + +func TestParsePolicy_OverridesIndividualFields(t *testing.T) { + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: PolicyConfigMapName, Namespace: PolicyConfigMapNamespace}, + Data: map[string]string{ + "maxConcurrentNodes": "3", + "drainTimeout": "10m", + "podEvictionGracePeriod": "45s", + "ignoreDaemonSets": "false", + }, + } + + p, err := ParsePolicy(cm) + if err != nil { + t.Fatalf("ParsePolicy returned unexpected error: %v", err) + } + + want := Policy{ + MaxConcurrentNodes: 3, + DrainTimeout: 10 * time.Minute, + PodEvictionGracePeriod: 45 * time.Second, + IgnoreDaemonSets: false, + } + if p != want { + t.Errorf("ParsePolicy = %+v, want %+v", p, want) + } +} + +func TestParsePolicy_InvalidValues(t *testing.T) { + cases := map[string]string{ + "maxConcurrentNodes": "not-a-number", + "drainTimeout": "not-a-duration", + "podEvictionGracePeriod": "not-a-duration", + "ignoreDaemonSets": "not-a-bool", + } + + for key, badValue := range cases { + t.Run(key, func(t *testing.T) { + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: PolicyConfigMapName, Namespace: PolicyConfigMapNamespace}, + Data: map[string]string{key: badValue}, + } + if _, err := ParsePolicy(cm); err == nil { + t.Errorf("ParsePolicy with %s=%q returned nil error, want error", key, badValue) + } + }) + } +} + +func TestParsePolicy_ValidationRejectsNegativeValues(t *testing.T) { + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: PolicyConfigMapName, Namespace: PolicyConfigMapNamespace}, + Data: map[string]string{ + "maxConcurrentNodes": "-5", + }, + } + if _, err := ParsePolicy(cm); err == nil { + t.Error("ParsePolicy with negative maxConcurrentNodes returned nil error, want error") + } +} diff --git a/pkg/controller/nodemaintenance/schedule.go b/pkg/controller/nodemaintenance/schedule.go new file mode 100644 index 0000000000000..8703f406eee99 --- /dev/null +++ b/pkg/controller/nodemaintenance/schedule.go @@ -0,0 +1,139 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "encoding/json" + "fmt" + "time" +) + +// ParseWindow decodes a Window from its JSON annotation representation. +func ParseWindow(annotation string) (*Window, error) { + w := &Window{} + if err := json.Unmarshal([]byte(annotation), w); err != nil { + return nil, fmt.Errorf("invalid maintenance window annotation: %w", err) + } + if errs := ValidateWindow(w); len(errs) > 0 { + return nil, fmt.Errorf("invalid maintenance window: %v", errs.ToAggregate()) + } + return w, nil +} + +// location resolves the Window's configured time zone, falling back to UTC +// when none is set. +func (w *Window) location() *time.Location { + if w.Timezone == "" { + return time.UTC + } + loc, err := time.LoadLocation(w.Timezone) + if err != nil { + return time.UTC + } + return loc +} + +// parseClock parses a "15:04" clock string into minutes since midnight. +func parseClock(s string) (int, error) { + t, err := time.Parse("15:04", s) + if err != nil { + return 0, fmt.Errorf("invalid clock value %q: %w", s, err) + } + return t.Hour()*60 + t.Minute(), nil +} + +// dayMatches reports whether d is one of the Window's configured recurrence +// days. An empty Days list matches every day. +func (w *Window) dayMatches(d time.Weekday) bool { + if len(w.Days) == 0 { + return true + } + for _, day := range w.Days { + if time.Weekday(day) == d { + return true + } + } + return false +} + +// IsActive reports whether the window is open at the given instant. +func (w *Window) IsActive(now time.Time) (bool, error) { + loc := w.location() + local := now.In(loc) + + startMin, err := parseClock(w.Start) + if err != nil { + return false, err + } + endMin, err := parseClock(w.End) + if err != nil { + return false, err + } + + if !w.dayMatches(local.Weekday()) { + return false, nil + } + + nowMin := local.Hour()*60 + local.Minute() + + if startMin <= endMin { + // Same-day window, e.g. 09:00-17:00. + return nowMin >= startMin && nowMin < endMin, nil + } + + // Overnight window, e.g. 22:00-02:00: active from Start through midnight. + return nowMin >= startMin, nil +} + +// NextTransition returns the next instant at which the window's active +// state will flip (i.e. the next open or close boundary), so callers can +// schedule precise wakeups instead of polling. +func (w *Window) NextTransition(now time.Time) (time.Time, error) { + loc := w.location() + local := now.In(loc) + + startMin, err := parseClock(w.Start) + if err != nil { + return time.Time{}, err + } + endMin, err := parseClock(w.End) + if err != nil { + return time.Time{}, err + } + + active, err := w.IsActive(now) + if err != nil { + return time.Time{}, err + } + targetMin := startMin + if active { + targetMin = endMin + } + + for offset := 0; offset < 7; offset++ { + day := local.AddDate(0, 0, offset) + if !w.dayMatches(day.Weekday()) { + continue + } + candidate := time.Date(day.Year(), day.Month(), day.Day(), targetMin/60, targetMin%60, 0, 0, loc) + if candidate.After(local) { + return candidate, nil + } + } + + return time.Time{}, fmt.Errorf("no upcoming transition found for window starting %s", w.Start) +} diff --git a/pkg/controller/nodemaintenance/schedule_test.go b/pkg/controller/nodemaintenance/schedule_test.go new file mode 100644 index 0000000000000..929367fc6c76f --- /dev/null +++ b/pkg/controller/nodemaintenance/schedule_test.go @@ -0,0 +1,354 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "testing" + "time" +) + +func mustParseTime(t *testing.T, layout, value string) time.Time { + t.Helper() + parsed, err := time.Parse(layout, value) + if err != nil { + t.Fatalf("failed to parse time %q: %v", value, err) + } + return parsed +} + +func TestParseClock(t *testing.T) { + cases := []struct { + name string + input string + want int + wantErr bool + }{ + {name: "midnight", input: "00:00", want: 0}, + {name: "noon", input: "12:00", want: 720}, + {name: "end of day", input: "23:59", want: 1439}, + {name: "single digit hour", input: "09:30", want: 570}, + {name: "malformed", input: "25:99", wantErr: true}, + {name: "empty", input: "", wantErr: true}, + {name: "not a clock", input: "banana", wantErr: true}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := parseClock(tc.input) + if tc.wantErr { + if err == nil { + t.Fatalf("parseClock(%q) = %d, nil; want error", tc.input, got) + } + return + } + if err != nil { + t.Fatalf("parseClock(%q) returned unexpected error: %v", tc.input, err) + } + if got != tc.want { + t.Errorf("parseClock(%q) = %d, want %d", tc.input, got, tc.want) + } + }) + } +} + +func TestWindowDayMatches(t *testing.T) { + cases := []struct { + name string + days []int + day time.Weekday + want bool + }{ + {name: "empty matches every day", days: nil, day: time.Tuesday, want: true}, + {name: "explicit match", days: []int{1, 3, 5}, day: time.Wednesday, want: true}, + {name: "explicit non-match", days: []int{1, 3, 5}, day: time.Sunday, want: false}, + {name: "single day", days: []int{0}, day: time.Sunday, want: true}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + w := &Window{Days: tc.days} + if got := w.dayMatches(tc.day); got != tc.want { + t.Errorf("dayMatches(%v) = %v, want %v", tc.day, got, tc.want) + } + }) + } +} + +func TestWindowIsActive_SameDayWindow(t *testing.T) { + w := &Window{Start: "09:00", End: "17:00", Timezone: "UTC"} + + cases := []struct { + name string + now time.Time + want bool + }{ + {name: "before window", now: mustParseTime(t, time.RFC3339, "2025-06-04T08:59:00Z"), want: false}, + {name: "at start boundary", now: mustParseTime(t, time.RFC3339, "2025-06-04T09:00:00Z"), want: true}, + {name: "middle of window", now: mustParseTime(t, time.RFC3339, "2025-06-04T12:30:00Z"), want: true}, + {name: "at end boundary", now: mustParseTime(t, time.RFC3339, "2025-06-04T17:00:00Z"), want: false}, + {name: "after window", now: mustParseTime(t, time.RFC3339, "2025-06-04T18:00:00Z"), want: false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := w.IsActive(tc.now) + if err != nil { + t.Fatalf("IsActive returned unexpected error: %v", err) + } + if got != tc.want { + t.Errorf("IsActive(%s) = %v, want %v", tc.now, got, tc.want) + } + }) + } +} + +func TestWindowIsActive_RestrictedDays(t *testing.T) { + // Saturdays only. + w := &Window{Start: "02:00", End: "04:00", Days: []int{6}, Timezone: "UTC"} + + saturday := mustParseTime(t, time.RFC3339, "2025-06-07T03:00:00Z") // a Saturday + sunday := mustParseTime(t, time.RFC3339, "2025-06-08T03:00:00Z") // a Sunday + + if active, err := w.IsActive(saturday); err != nil || !active { + t.Errorf("IsActive(%s) = %v, %v; want true, nil", saturday, active, err) + } + if active, err := w.IsActive(sunday); err != nil || active { + t.Errorf("IsActive(%s) = %v, %v; want false, nil", sunday, active, err) + } +} + +func TestWindowIsActive_OvernightWindow(t *testing.T) { + // Every day, 22:00-02:00. + w := &Window{Start: "22:00", End: "02:00", Timezone: "UTC"} + + cases := []struct { + name string + now time.Time + want bool + }{ + {name: "before window", now: mustParseTime(t, time.RFC3339, "2025-06-04T21:00:00Z"), want: false}, + {name: "just after start", now: mustParseTime(t, time.RFC3339, "2025-06-04T23:00:00Z"), want: true}, + {name: "well after end, next afternoon", now: mustParseTime(t, time.RFC3339, "2025-06-05T14:00:00Z"), want: false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := w.IsActive(tc.now) + if err != nil { + t.Fatalf("IsActive returned unexpected error: %v", err) + } + if got != tc.want { + t.Errorf("IsActive(%s) = %v, want %v", tc.now, got, tc.want) + } + }) + } +} + +func TestWindowIsActive_Timezone(t *testing.T) { + w := &Window{Start: "09:00", End: "17:00", Timezone: "America/Los_Angeles"} + + // 16:30 UTC is 09:30 PDT (UTC-7 in June), inside the window. + inWindow := mustParseTime(t, time.RFC3339, "2025-06-04T16:30:00Z") + if active, err := w.IsActive(inWindow); err != nil || !active { + t.Errorf("IsActive(%s) = %v, %v; want true, nil", inWindow, active, err) + } + + // 07:00 UTC is 00:00 PDT, outside the window. + outOfWindow := mustParseTime(t, time.RFC3339, "2025-06-04T07:00:00Z") + if active, err := w.IsActive(outOfWindow); err != nil || active { + t.Errorf("IsActive(%s) = %v, %v; want false, nil", outOfWindow, active, err) + } +} + +func TestWindowIsActive_InvalidClock(t *testing.T) { + w := &Window{Start: "not-a-time", End: "17:00"} + if _, err := w.IsActive(time.Now()); err == nil { + t.Error("IsActive() with an invalid Start returned nil error, want error") + } +} + +func TestWindowNextTransition_SameDayWindow(t *testing.T) { + w := &Window{Start: "09:00", End: "17:00", Timezone: "UTC"} + + now := mustParseTime(t, time.RFC3339, "2025-06-04T12:00:00Z") + next, err := w.NextTransition(now) + if err != nil { + t.Fatalf("NextTransition returned unexpected error: %v", err) + } + want := mustParseTime(t, time.RFC3339, "2025-06-04T17:00:00Z") + if !next.Equal(want) { + t.Errorf("NextTransition(%s) = %s, want %s", now, next, want) + } +} + +func TestWindowNextTransition_BeforeWindowOpens(t *testing.T) { + w := &Window{Start: "09:00", End: "17:00", Timezone: "UTC"} + + now := mustParseTime(t, time.RFC3339, "2025-06-04T06:00:00Z") + next, err := w.NextTransition(now) + if err != nil { + t.Fatalf("NextTransition returned unexpected error: %v", err) + } + want := mustParseTime(t, time.RFC3339, "2025-06-04T09:00:00Z") + if !next.Equal(want) { + t.Errorf("NextTransition(%s) = %s, want %s", now, next, want) + } +} + +func TestWindowNextTransition_RestrictedDaysRollsToNextOccurrence(t *testing.T) { + // Mondays only, 09:00-17:00. + w := &Window{Start: "09:00", End: "17:00", Days: []int{1}, Timezone: "UTC"} + + // 2025-06-04 is a Wednesday; the next Monday is 2025-06-09. + now := mustParseTime(t, time.RFC3339, "2025-06-04T12:00:00Z") + next, err := w.NextTransition(now) + if err != nil { + t.Fatalf("NextTransition returned unexpected error: %v", err) + } + want := mustParseTime(t, time.RFC3339, "2025-06-09T09:00:00Z") + if !next.Equal(want) { + t.Errorf("NextTransition(%s) = %s, want %s", now, next, want) + } +} + +func TestWindowIsActive_MultipleDays(t *testing.T) { + // Weekdays only, 09:00-17:00. + w := &Window{Start: "09:00", End: "17:00", Days: []int{1, 2, 3, 4, 5}, Timezone: "UTC"} + + cases := []struct { + name string + now time.Time + want bool + }{ + {name: "monday in window", now: mustParseTime(t, time.RFC3339, "2025-06-02T10:00:00Z"), want: true}, + {name: "friday in window", now: mustParseTime(t, time.RFC3339, "2025-06-06T16:00:00Z"), want: true}, + {name: "saturday out of window", now: mustParseTime(t, time.RFC3339, "2025-06-07T10:00:00Z"), want: false}, + {name: "sunday out of window", now: mustParseTime(t, time.RFC3339, "2025-06-08T10:00:00Z"), want: false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := w.IsActive(tc.now) + if err != nil { + t.Fatalf("IsActive returned unexpected error: %v", err) + } + if got != tc.want { + t.Errorf("IsActive(%s) = %v, want %v", tc.now, got, tc.want) + } + }) + } +} + +func TestWindowIsActive_ZeroLengthWindow(t *testing.T) { + // Start == End is degenerate and should never be considered active. + w := &Window{Start: "09:00", End: "09:00", Timezone: "UTC"} + + now := mustParseTime(t, time.RFC3339, "2025-06-04T09:00:00Z") + got, err := w.IsActive(now) + if err != nil { + t.Fatalf("IsActive returned unexpected error: %v", err) + } + if got { + t.Error("IsActive() = true for a zero-length window, want false") + } +} + +func TestWindowNextTransition_MultipleDaysPicksNearest(t *testing.T) { + // Weekdays only, 09:00-17:00. + w := &Window{Start: "09:00", End: "17:00", Days: []int{1, 2, 3, 4, 5}, Timezone: "UTC"} + + // Friday evening: the next occurrence should be the following Monday, + // not Saturday or Sunday. + now := mustParseTime(t, time.RFC3339, "2025-06-06T20:00:00Z") + next, err := w.NextTransition(now) + if err != nil { + t.Fatalf("NextTransition returned unexpected error: %v", err) + } + want := mustParseTime(t, time.RFC3339, "2025-06-09T09:00:00Z") + if !next.Equal(want) { + t.Errorf("NextTransition(%s) = %s, want %s", now, next, want) + } +} + +func TestWindowNextTransition_ActiveWindowReturnsCloseTime(t *testing.T) { + w := &Window{Start: "09:00", End: "17:00", Timezone: "UTC"} + + now := mustParseTime(t, time.RFC3339, "2025-06-04T10:30:00Z") + active, err := w.IsActive(now) + if err != nil || !active { + t.Fatalf("precondition failed: IsActive(%s) = %v, %v; want true, nil", now, active, err) + } + + next, err := w.NextTransition(now) + if err != nil { + t.Fatalf("NextTransition returned unexpected error: %v", err) + } + want := mustParseTime(t, time.RFC3339, "2025-06-04T17:00:00Z") + if !next.Equal(want) { + t.Errorf("NextTransition(%s) = %s, want %s (the close time, since the window is active)", now, next, want) + } +} + +func TestParseWindow(t *testing.T) { + cases := []struct { + name string + annotation string + wantErr bool + }{ + { + name: "valid", + annotation: `{"start":"09:00","end":"17:00"}`, + }, + { + name: "valid with days and timezone", + annotation: `{"start":"22:00","end":"02:00","days":[5,6],"timezone":"America/Los_Angeles"}`, + }, + { + name: "malformed json", + annotation: `{"start":`, + wantErr: true, + }, + { + name: "semantically invalid", + annotation: `{"start":"nope","end":"17:00"}`, + wantErr: true, + }, + { + name: "empty string", + annotation: "", + wantErr: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + w, err := ParseWindow(tc.annotation) + if tc.wantErr { + if err == nil { + t.Fatalf("ParseWindow(%q) = %+v, nil; want error", tc.annotation, w) + } + return + } + if err != nil { + t.Fatalf("ParseWindow(%q) returned unexpected error: %v", tc.annotation, err) + } + if w == nil { + t.Fatalf("ParseWindow(%q) returned a nil window with no error", tc.annotation) + } + }) + } +} diff --git a/pkg/controller/nodemaintenance/types.go b/pkg/controller/nodemaintenance/types.go new file mode 100644 index 0000000000000..1009c457efbbc --- /dev/null +++ b/pkg/controller/nodemaintenance/types.go @@ -0,0 +1,90 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import "time" + +const ( + // AnnotationWindow holds the JSON-encoded Window describing when a node + // should be placed into maintenance. + AnnotationWindow = "maintenance.kubernetes.io/window" + + // AnnotationCordonedBy is set on a Node by this controller when it cordons + // the node so that, on the next reconcile, it can tell whether it is safe + // to uncordon the node again (i.e. nobody else cordoned it in the + // meantime). + AnnotationCordonedBy = "maintenance.kubernetes.io/cordoned-by" + + // AnnotationLastDrainTime records the RFC3339 timestamp of the last + // successful drain, surfaced for observability/debugging. + AnnotationLastDrainTime = "maintenance.kubernetes.io/last-drain-time" + + // ControllerName identifies this controller as the writer of the + // AnnotationCordonedBy annotation. + ControllerName = "node-maintenance-controller" + + // PolicyConfigMapNamespace/PolicyConfigMapName locate the ConfigMap that + // holds cluster-wide defaults for maintenance behavior. + PolicyConfigMapNamespace = "kube-system" + PolicyConfigMapName = "node-maintenance-policy" +) + +// Window describes a single recurring maintenance window for a node. It is +// stored JSON-encoded in the AnnotationWindow annotation. +type Window struct { + // Start is the time of day the window opens, in "15:04" 24h format, + // interpreted in Timezone. + Start string `json:"start"` + // End is the time of day the window closes, in "15:04" 24h format, + // interpreted in Timezone. An End earlier than Start indicates a window + // that spans midnight. + End string `json:"end"` + // Days lists the days of the week (0=Sunday..6=Saturday) the window + // recurs on. An empty list means every day. + Days []int `json:"days,omitempty"` + // Timezone is an IANA time zone name, e.g. "America/Los_Angeles". + // Defaults to UTC when empty. + Timezone string `json:"timezone,omitempty"` +} + +// Policy holds cluster-wide defaults for how maintenance windows are +// enforced. It is sourced from the PolicyConfigMapName ConfigMap. +type Policy struct { + // MaxConcurrentNodes bounds how many nodes may be drained at once across + // the whole cluster. + MaxConcurrentNodes int + // DrainTimeout is the maximum time to wait for graceful pod eviction + // before force-deleting stragglers. + DrainTimeout time.Duration + // PodEvictionGracePeriod is the grace period passed to each Eviction + // request. + PodEvictionGracePeriod time.Duration + // IgnoreDaemonSets controls whether DaemonSet-managed pods are skipped + // during drain (they are otherwise unschedulable off the node anyway). + IgnoreDaemonSets bool +} + +// DefaultPolicy returns the built-in defaults used when no policy ConfigMap +// is present, or when individual keys are unset. +func DefaultPolicy() Policy { + return Policy{ + MaxConcurrentNodes: 1, + DrainTimeout: 5 * time.Minute, + PodEvictionGracePeriod: 30 * time.Second, + IgnoreDaemonSets: true, + } +} diff --git a/pkg/controller/nodemaintenance/validation.go b/pkg/controller/nodemaintenance/validation.go new file mode 100644 index 0000000000000..9c59f1b146545 --- /dev/null +++ b/pkg/controller/nodemaintenance/validation.go @@ -0,0 +1,61 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "time" + + "k8s.io/apimachinery/pkg/util/validation/field" +) + +// ValidateWindow validates the fields of a Window decoded from a node +// annotation. +func ValidateWindow(w *Window) field.ErrorList { + allErrs := field.ErrorList{} + + if _, err := time.Parse("15:04", w.Start); err != nil { + allErrs = append(allErrs, field.Invalid(field.NewPath("start"), w.Start, "must be a 24h HH:MM value")) + } + if _, err := time.Parse("15:04", w.End); err != nil { + allErrs = append(allErrs, field.Invalid(field.NewPath("end"), w.End, "must be a 24h HH:MM value")) + } + for i, d := range w.Days { + if d < 0 || d > 6 { + allErrs = append(allErrs, field.Invalid(field.NewPath("days").Index(i), d, "must be between 0 (Sunday) and 6 (Saturday)")) + } + } + + return allErrs +} + +// ValidatePolicy validates a Policy parsed from the maintenance policy +// ConfigMap. +func ValidatePolicy(p *Policy) field.ErrorList { + allErrs := field.ErrorList{} + + if p.MaxConcurrentNodes < 0 { + allErrs = append(allErrs, field.Invalid(field.NewPath("maxConcurrentNodes"), p.MaxConcurrentNodes, "must be non-negative")) + } + if p.DrainTimeout < 0 { + allErrs = append(allErrs, field.Invalid(field.NewPath("drainTimeout"), p.DrainTimeout, "must be non-negative")) + } + if p.PodEvictionGracePeriod < 0 { + allErrs = append(allErrs, field.Invalid(field.NewPath("podEvictionGracePeriod"), p.PodEvictionGracePeriod, "must be non-negative")) + } + + return allErrs +} diff --git a/pkg/controller/nodemaintenance/validation_test.go b/pkg/controller/nodemaintenance/validation_test.go new file mode 100644 index 0000000000000..8a759c46283e4 --- /dev/null +++ b/pkg/controller/nodemaintenance/validation_test.go @@ -0,0 +1,119 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package nodemaintenance + +import ( + "testing" + "time" +) + +func TestValidateWindow(t *testing.T) { + cases := []struct { + name string + window Window + wantErr bool + }{ + { + name: "valid basic window", + window: Window{Start: "09:00", End: "17:00"}, + }, + { + name: "valid overnight window", + window: Window{Start: "22:00", End: "02:00"}, + }, + { + name: "valid with days and timezone", + window: Window{Start: "09:00", End: "17:00", Days: []int{1, 2, 3, 4, 5}, Timezone: "UTC"}, + }, + { + name: "malformed start", + window: Window{Start: "not-a-time", End: "17:00"}, + wantErr: true, + }, + { + name: "malformed end", + window: Window{Start: "09:00", End: "not-a-time"}, + wantErr: true, + }, + { + name: "day out of range low", + window: Window{Start: "09:00", End: "17:00", Days: []int{-1}}, + wantErr: true, + }, + { + name: "day out of range high", + window: Window{Start: "09:00", End: "17:00", Days: []int{7}}, + wantErr: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + errs := ValidateWindow(&tc.window) + if tc.wantErr && len(errs) == 0 { + t.Errorf("ValidateWindow(%+v) returned no errors, want at least one", tc.window) + } + if !tc.wantErr && len(errs) != 0 { + t.Errorf("ValidateWindow(%+v) returned unexpected errors: %v", tc.window, errs) + } + }) + } +} + +func TestValidatePolicy(t *testing.T) { + cases := []struct { + name string + policy Policy + wantErr bool + }{ + { + name: "defaults are valid", + policy: DefaultPolicy(), + }, + { + name: "negative max concurrent nodes", + policy: Policy{MaxConcurrentNodes: -1, DrainTimeout: time.Minute, PodEvictionGracePeriod: time.Second}, + wantErr: true, + }, + { + name: "negative drain timeout", + policy: Policy{MaxConcurrentNodes: 1, DrainTimeout: -time.Minute, PodEvictionGracePeriod: time.Second}, + wantErr: true, + }, + { + name: "negative grace period", + policy: Policy{MaxConcurrentNodes: 1, DrainTimeout: time.Minute, PodEvictionGracePeriod: -time.Second}, + wantErr: true, + }, + { + name: "zero max concurrent nodes is allowed (pauses maintenance)", + policy: Policy{MaxConcurrentNodes: 0, DrainTimeout: time.Minute, PodEvictionGracePeriod: time.Second}, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + errs := ValidatePolicy(&tc.policy) + if tc.wantErr && len(errs) == 0 { + t.Errorf("ValidatePolicy(%+v) returned no errors, want at least one", tc.policy) + } + if !tc.wantErr && len(errs) != 0 { + t.Errorf("ValidatePolicy(%+v) returned unexpected errors: %v", tc.policy, errs) + } + }) + } +}