Add Capacity Buffer controller logic

Author: abdelrahman882

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    = "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/buffer_controller.go

@@ -0,0 +1,142 @@
+/*
+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"
+	buffers_client "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"
+	buffers_filters "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/filters"
+	buffers_translators "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/translators"
+	buffers_updater "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/updater"
+
+	kube_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
+	buffersStrategyFilter buffers_filters.BuffersFilter
+	buffersStatusFilter   buffers_filters.BuffersFilter
+	buffersTranslator     buffers_translators.BuffersTranslator
+	buffersUpdater        buffers_updater.BuffersStatusUpdater
+	loopInterval          time.Duration
+}
+
+// NewBufferController creates new bufferController object
+func NewBufferController(
+	buffersLister v1.CapacityBufferLister,
+	buffersStrategyFilter buffers_filters.BuffersFilter,
+	buffersStatusFilter buffers_filters.BuffersFilter,
+	buffersTranslator buffers_translators.BuffersTranslator,
+	buffersUpdater buffers_updater.BuffersStatusUpdater,
+	loopInterval time.Duration,
+) (BufferController, error) {
+	return &bufferController{
+		buffersLister:         buffersLister,
+		buffersStrategyFilter: buffersStrategyFilter,
+		buffersStatusFilter:   buffersStatusFilter,
+		buffersTranslator:     buffersTranslator,
+		buffersUpdater:        buffersUpdater,
+		loopInterval:          loopInterval,
+	}, nil
+}
+
+// NewDefaultBufferController creates Updater with default configs
+func NewDefaultBufferController(
+	listerRegistry kubernetes.ListerRegistry,
+	capacityBufferClinet buffers_client.Clientset,
+	nodeBufferListener v1.CapacityBufferLister,
+	kubeClient kube_client.Clientset,
+) BufferController {
+	return &bufferController{
+		buffersLister:         nodeBufferListener,
+		buffersStrategyFilter: buffers_filters.NewDefaultBuffersStrategyFilter([]string{common.ActiveProvisioningStrategy}),
+		buffersStatusFilter: buffers_filters.NewDefaultBuffersStatusFilter(map[string]string{
+			common.ReadyForProvisioningCondition: common.ConditionTrue,
+			common.ProvisioningCondition:         common.ConditionTrue,
+		}),
+		buffersTranslator: buffers_translators.NewCombinedBuffersTranslator(
+			[]buffers_translators.BuffersTranslator{
+				buffers_translators.NewDefaultPodTemplateBufferTranslator(),
+			},
+		),
+		buffersUpdater: *buffers_updater.NewDefaultBuffersStatusUpdater(&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() {
+
+	// 1. 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))
+
+	// 2. Filter the desired provisioning strategy
+	filteredBuffers, _ := c.buffersStrategyFilter.Filter(buffers)
+	klog.V(2).Infof("Capacity buffer controller filtered %v buffers with buffers strategy filter", len(filteredBuffers))
+
+	// 2. Filter the desired status
+	_, toBeTranslatedBuffers := c.buffersStatusFilter.Filter(filteredBuffers)
+	klog.V(2).Infof("Capacity buffer controller filtered %v buffers with buffers status filter", len(filteredBuffers))
+
+	// 3. Extract a) pod specs and b) number of replicase from filtered buffers
+	errors := c.buffersTranslator.Translate(toBeTranslatedBuffers)
+	logErrors(errors)
+
+	// 4. Update buffer status by calling API server
+	errors = c.buffersUpdater.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/buffers_filter.go

@@ -0,0 +1,78 @@
+/*
+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 buffersfilter
+
+import (
+	api_v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1"
+)
+
+// BuffersFilter filters CapacityBuffer based on some criteria.
+type BuffersFilter interface {
+	Filter(buffers []*api_v1.CapacityBuffer) ([]*api_v1.CapacityBuffer, []*api_v1.CapacityBuffer)
+	CleanUp()
+}
+
+// NoOpBuffersFilter is returning buffers lists without filtering any of them.
+type NoOpBuffersFilter struct {
+}
+
+// NewDefaultNoOpBuffersFilter creates an instance of PodListProcessor.
+func NewDefaultNoOpBuffersFilter() *NoOpBuffersFilter {
+	return &NoOpBuffersFilter{}
+}
+
+// Filter do no filteration
+func (f *NoOpBuffersFilter) Filter(buffers []*api_v1.CapacityBuffer) ([]*api_v1.CapacityBuffer, []*api_v1.CapacityBuffer) {
+	return buffers, []*api_v1.CapacityBuffer{}
+}
+
+// CleanUp cleans up the filter's internal structures.
+func (f *NoOpBuffersFilter) CleanUp() {
+}
+
+// CombinedBuffersFilter is a list of BuffersFilter
+type CombinedBuffersFilter struct {
+	filters []BuffersFilter
+}
+
+// NewCombinedBuffersFilter construct CombinedBuffersFilter.
+func NewCombinedBuffersFilter(filters []BuffersFilter) *CombinedBuffersFilter {
+	return &CombinedBuffersFilter{filters}
+}
+
+// AddFilter append a filter to the list.
+func (f *CombinedBuffersFilter) AddFilter(filter BuffersFilter) {
+	f.filters = append(f.filters, filter)
+}
+
+// Filter runs sub-filters sequentially
+func (f *CombinedBuffersFilter) Filter(buffers []*api_v1.CapacityBuffer) ([]*api_v1.CapacityBuffer, []*api_v1.CapacityBuffer) {
+	var totalFilteredOutBuffers []*api_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 *CombinedBuffersFilter) CleanUp() {
+	for _, filter := range f.filters {
+		filter.CleanUp()
+	}
+}

cluster-autoscaler/capacitybuffer/filters/buffers_status_filter.go

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