Merge pull request #8521 from abdelrahman882/capacity-buffer-controller

Add Capacity Buffer controller logic

Author: k8s-ci-robot

cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1/types.go

@@ -97,9 +97,9 @@ type ResourceList map[ResourceName]resource.Quantity
 // CapacityBufferSpec defines the desired state of CapacityBuffer.
 type CapacityBufferSpec struct {
 	// ProvisioningStrategy defines how the buffer is utilized.
-	// "active-capacity" is the default strategy, where the buffer actively scales up the cluster by creating placeholder pods.
-	// +kubebuilder:validation:Enum=active-capacity
-	// +kubebuilder:default="active-capacity"
+	// "buffer.x-k8s.io/active-capacity" is the default strategy, where the buffer actively scales up the cluster by creating placeholder pods.
+	// +kubebuilder:validation:Enum=buffer.x-k8s.io/active-capacity
+	// +kubebuilder:default="buffer.x-k8s.io/active-capacity"
 	// +optional
 	ProvisioningStrategy *string `json:"provisioningStrategy,omitempty" protobuf:"bytes,1,opt,name=provisioningStrategy"`
 
@@ -123,24 +123,18 @@ type CapacityBufferSpec struct {
 	// If neither `replicas` nor `percentage` is set, as many chunks as fit within
 	// defined resource limits (if any) will be created. If both are set, the maximum
 	// of the two will be used.
-	// This field is mutually exclusive with `percentage` when `scalableRef` is set.
 	// +optional
 	// +kubebuilder:validation:Minimum=0
 	// +kubebuilder:validation:ExclusiveMinimum=false
-	// +kubebuilder:validation:Xor=replicas,percentage
 	Replicas *int32 `json:"replicas,omitempty" protobuf:"varint,4,opt,name=replicas"`
 
 	// Percentage defines the desired buffer capacity as a percentage of the
 	// `scalableRef`'s current replicas. This is only applicable if `scalableRef` is set.
 	// The absolute number of replicas is calculated from the percentage by rounding up to a minimum of 1.
 	// For example, if `scalableRef` has 10 replicas and `percentage` is 20, 2 buffer chunks will be created.
-	// This field is mutually exclusive with `replicas`.
 	// +optional
 	// +kubebuilder:validation:Minimum=0
-	// +kubebuilder:validation:Maximum=100
-	// +kubebuilder:validation:ExclusiveMaximum=false
 	// +kubebuilder:validation:ExclusiveMinimum=false
-	// +kubebuilder:validation:Xor=replicas,percentage
 	Percentage *int32 `json:"percentage,omitempty" protobuf:"varint,5,opt,name=percentage"`
 
 	// Limits, if specified, will limit the number of chunks created for this buffer

cluster-autoscaler/apis/config/crd/autoscaling.x-k8s.io_capacitybuffers.yaml

@@ -100,12 +100,12 @@ spec:
                 - name
                 type: object
               provisioningStrategy:
-                default: active-capacity
+                default: buffer.x-k8s.io/active-capacity
                 description: |-
                   ProvisioningStrategy defines how the buffer is utilized.
-                  "active-capacity" is the default strategy, where the buffer actively scales up the cluster by creating placeholder pods.
+                  "buffer.x-k8s.io/active-capacity" is the default strategy, where the buffer actively scales up the cluster by creating placeholder pods.
                 enum:
-                - active-capacity
+                - buffer.x-k8s.io/active-capacity
                 type: string
               replicas:
                 description: |-

cluster-autoscaler/capacitybuffer/common/common.go

@@ -0,0 +1,49 @@
+/*
+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 common
+
+import (
+	"context"
+
+	v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1"
+	client "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/client/clientset/versioned"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// Constants to use in Capacity Buffers objects
+const (
+	ActiveProvisioningStrategy    = "buffer.x-k8s.io/active-capacity"
+	ReadyForProvisioningCondition = "ReadyForProvisioning"
+	ProvisioningCondition         = "Provisioning"
+	ConditionTrue                 = "True"
+	ConditionFalse                = "False"
+	DefaultNamespace              = "default"
+)
+
+// CreatePodTemplate creates a pod template object by calling API server
+func CreatePodTemplate(client *kubernetes.Clientset, podTemplate *corev1.PodTemplate) (*corev1.PodTemplate, error) {
+	return client.CoreV1().PodTemplates(DefaultNamespace).Create(context.TODO(), podTemplate, metav1.CreateOptions{})
+}
+
+// UpdateBufferStatus updates the passed buffer object with its defined status
+func UpdateBufferStatus(buffersClient client.Interface, buffer *v1.CapacityBuffer) error {
+	_, err := buffersClient.AutoscalingV1().CapacityBuffers(DefaultNamespace).UpdateStatus(context.TODO(), buffer, metav1.UpdateOptions{})
+	return err
+}

cluster-autoscaler/capacitybuffer/controller/controller.go

@@ -0,0 +1,143 @@
+/*
+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 controller
+
+import (
+	"time"
+
+	"k8s.io/klog/v2"
+
+	"k8s.io/apimachinery/pkg/labels"
+	buffersclient "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/client/clientset/versioned"
+
+	v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/client/listers/autoscaling.x-k8s.io/v1"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/kubernetes"
+
+	common "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/common"
+	filters "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/filters"
+	translators "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/translators"
+	updater "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/updater"
+
+	client "k8s.io/client-go/kubernetes"
+)
+
+const loopInterval = time.Second * 5
+
+// BufferController performs updates on Buffers and convert them to pods to be injected
+type BufferController interface {
+	// Run to run the reconciliation loop frequently every x seconds
+	Run(stopCh <-chan struct{})
+}
+
+type bufferController struct {
+	buffersLister  v1.CapacityBufferLister
+	strategyFilter filters.Filter
+	statusFilter   filters.Filter
+	translator     translators.Translator
+	updater        updater.StatusUpdater
+	loopInterval   time.Duration
+}
+
+// NewBufferController creates new bufferController object
+func NewBufferController(
+	buffersLister v1.CapacityBufferLister,
+	strategyFilter filters.Filter,
+	statusFilter filters.Filter,
+	translator translators.Translator,
+	updater updater.StatusUpdater,
+	loopInterval time.Duration,
+) BufferController {
+	return &bufferController{
+		buffersLister:  buffersLister,
+		strategyFilter: strategyFilter,
+		statusFilter:   statusFilter,
+		translator:     translator,
+		updater:        updater,
+		loopInterval:   loopInterval,
+	}
+}
+
+// NewDefaultBufferController creates bufferController with default configs
+func NewDefaultBufferController(
+	listerRegistry kubernetes.ListerRegistry,
+	capacityBufferClinet buffersclient.Clientset,
+	nodeBufferListener v1.CapacityBufferLister,
+	kubeClient client.Clientset,
+) BufferController {
+	return &bufferController{
+		buffersLister: nodeBufferListener,
+		// Accepting empty string as it represents nil value for ProvisioningStrategy
+		strategyFilter: filters.NewStrategyFilter([]string{common.ActiveProvisioningStrategy, ""}),
+		statusFilter: filters.NewStatusFilter(map[string]string{
+			common.ReadyForProvisioningCondition: common.ConditionTrue,
+			common.ProvisioningCondition:         common.ConditionTrue,
+		}),
+		translator: translators.NewCombinedTranslator(
+			[]translators.Translator{
+				translators.NewPodTemplateBufferTranslator(),
+			},
+		),
+		updater:      *updater.NewStatusUpdater(&capacityBufferClinet),
+		loopInterval: loopInterval,
+	}
+}
+
+// Run to run the controller reconcile loop
+func (c *bufferController) Run(stopCh <-chan struct{}) {
+	for {
+		select {
+		case <-stopCh:
+			return
+		case <-time.After(c.loopInterval):
+			c.reconcile()
+		}
+	}
+}
+
+// Reconcile represents single iteration in the main-loop of Updater
+func (c *bufferController) reconcile() {
+
+	// List all capacity buffers objects
+	buffers, err := c.buffersLister.List(labels.Everything())
+	if err != nil {
+		klog.Errorf("Capacity buffer controller failed to list buffers with error: %v", err.Error())
+		return
+	}
+	klog.V(2).Infof("Capacity buffer controller listed [%v] buffers", len(buffers))
+
+	// Filter the desired provisioning strategy
+	filteredBuffers, _ := c.strategyFilter.Filter(buffers)
+	klog.V(2).Infof("Capacity buffer controller filtered %v buffers with buffers strategy filter", len(filteredBuffers))
+
+	// Filter the desired status
+	toBeTranslatedBuffers, _ := c.statusFilter.Filter(filteredBuffers)
+	klog.V(2).Infof("Capacity buffer controller filtered %v buffers with buffers status filter", len(filteredBuffers))
+
+	// Extract pod specs and number of replicas from filtered buffers
+	errors := c.translator.Translate(toBeTranslatedBuffers)
+	logErrors(errors)
+
+	// Update buffer status by calling API server
+	errors = c.updater.Update(toBeTranslatedBuffers)
+	logErrors(errors)
+}
+
+func logErrors(errors []error) {
+	for _, error := range errors {
+		klog.Errorf("Capacity buffer controller error: %v", error.Error())
+	}
+}

cluster-autoscaler/capacitybuffer/filters/filter.go

@@ -0,0 +1,60 @@
+/*
+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 filter
+
+import (
+	v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1"
+)
+
+// Filter filters CapacityBuffer based on some criteria.
+type Filter interface {
+	Filter(buffers []*v1.CapacityBuffer) ([]*v1.CapacityBuffer, []*v1.CapacityBuffer)
+	CleanUp()
+}
+
+// combinedFilter is a list of Filter
+type combinedFilter struct {
+	filters []Filter
+}
+
+// NewCombinedFilter construct combinedFilter.
+func NewCombinedFilter(filters []Filter) *combinedFilter {
+	return &combinedFilter{filters}
+}
+
+// AddFilter append a filter to the list.
+func (f *combinedFilter) AddFilter(filter Filter) {
+	f.filters = append(f.filters, filter)
+}
+
+// Filter runs sub-filters sequentially
+func (f *combinedFilter) Filter(buffers []*v1.CapacityBuffer) ([]*v1.CapacityBuffer, []*v1.CapacityBuffer) {
+	var totalFilteredOutBuffers []*v1.CapacityBuffer
+	for _, buffersFilter := range f.filters {
+		updatedBuffersList, filteredOutBuffers := buffersFilter.Filter(buffers)
+		buffers = updatedBuffersList
+		totalFilteredOutBuffers = append(totalFilteredOutBuffers, filteredOutBuffers...)
+	}
+	return buffers, totalFilteredOutBuffers
+}
+
+// CleanUp cleans up the filter's internal structures.
+func (f *combinedFilter) CleanUp() {
+	for _, filter := range f.filters {
+		filter.CleanUp()
+	}
+}

cluster-autoscaler/capacitybuffer/filters/status_filter.go

@@ -0,0 +1,62 @@
+/*
+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 filter
+
+import (
+	v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1"
+)
+
+// statusFilter filters out buffers with the defined conditions
+type statusFilter struct {
+	conditions map[string]string
+}
+
+// NewStatusFilter creates an instance of statusFilter that filters out the buffers with condition in passed conditions.
+func NewStatusFilter(conditions map[string]string) *statusFilter {
+	return &statusFilter{
+		conditions: conditions,
+	}
+}
+
+// Filter filters the passed buffers based on buffer status conditions
+func (f *statusFilter) Filter(buffersToFilter []*v1.CapacityBuffer) ([]*v1.CapacityBuffer, []*v1.CapacityBuffer) {
+	var buffers []*v1.CapacityBuffer
+	var filteredOutBuffers []*v1.CapacityBuffer
+
+	for _, buffer := range buffersToFilter {
+		if !f.hasCondition(buffer) {
+			buffers = append(buffers, buffer)
+		} else {
+			filteredOutBuffers = append(filteredOutBuffers, buffer)
+		}
+	}
+	return buffers, filteredOutBuffers
+}
+
+func (f *statusFilter) hasCondition(buffer *v1.CapacityBuffer) bool {
+	bufferConditions := buffer.Status.Conditions
+	for _, condition := range bufferConditions {
+		if val, found := f.conditions[condition.Type]; found && val == string(condition.Status) {
+			return true
+		}
+	}
+	return false
+}
+
+// CleanUp cleans up the filter's internal structures.
+func (f *statusFilter) CleanUp() {
+}