Add node maintenance window controller

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())

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,

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"
 )

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)
+}

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")
+	}
+}