Add Capacity buffers updater interface to allow wrapping updating logic

Author: abdelrahman882

cluster-autoscaler/capacitybuffer/controller/controller.go

@@ -51,7 +51,7 @@ type bufferController struct {
 	strategyFilter filters.Filter
 	translator     translators.Translator
 	quotaAllocator *resourceQuotaAllocator
-	updater        updater.StatusUpdater
+	updater        updater.Updater
 	queue          workqueue.TypedRateLimitingInterface[string]
 }
 
@@ -60,7 +60,7 @@ func NewBufferController(
 	client *cbclient.CapacityBufferClient,
 	strategyFilter filters.Filter,
 	translator translators.Translator,
-	updater updater.StatusUpdater,
+	updater updater.Updater,
 ) BufferController {
 	bc := &bufferController{
 		client:         client,
@@ -93,7 +93,7 @@ func NewDefaultBufferController(
 			},
 		),
 		quotaAllocator: newResourceQuotaAllocator(client),
-		updater:        *updater.NewStatusUpdater(client),
+		updater:        updater.NewStatusUpdater(client),
 		queue: workqueue.NewTypedRateLimitingQueueWithConfig(
 			workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{Name: "CapacityBuffers"},
 		),
@@ -307,7 +307,7 @@ func (c *bufferController) reconcileNamespace(namespace string) error {
 	}
 
 	// Update buffer status by calling API server
-	updateErrors := c.updater.Update(filteredBuffers)
+	_, updateErrors := c.updater.Update(filteredBuffers)
 	for _, err := range updateErrors {
 		runtime.HandleError(fmt.Errorf("capacity buffer controller error: %w", err))
 	}

cluster-autoscaler/capacitybuffer/updater/status_updater.go

@@ -33,16 +33,19 @@ func NewStatusUpdater(client *cbclient.CapacityBufferClient) *StatusUpdater {
 	}
 }
 
-// Update updates the buffer status with pod capacity
-func (u *StatusUpdater) Update(buffers []*v1.CapacityBuffer) []error {
+// Update updates the buffer status and returns the updated buffers objects and list of errors
+func (u *StatusUpdater) Update(buffers []*v1.CapacityBuffer) ([]*v1.CapacityBuffer, []error) {
 	var errors []error
+	var updatedBuffers []*v1.CapacityBuffer
 	for _, buffer := range buffers {
-		_, err := u.client.UpdateCapacityBuffer(buffer)
+		updatedBuffer, err := u.client.UpdateCapacityBuffer(buffer)
 		if err != nil {
 			errors = append(errors, err)
+		} else {
+			updatedBuffers = append(updatedBuffers, updatedBuffer)
 		}
 	}
-	return errors
+	return updatedBuffers, errors
 }
 
 // CleanUp cleans up the updater's internal structures.

cluster-autoscaler/capacitybuffer/updater/status_updater_test.go

@@ -89,9 +89,10 @@ func TestStatusUpdater(t *testing.T) {
 				},
 			)
 			buffersUpdater := NewStatusUpdater(fakeCapacityBuffersClient)
-			errors := buffersUpdater.Update(test.buffers)
+			updatedBuffers, errors := buffersUpdater.Update(test.buffers)
 			assert.Equal(t, test.expectedNumberOfErrors, len(errors))
 			assert.Equal(t, test.expectedNumberOfCalls, updateCallsCount)
+			assert.Equal(t, len(test.buffers)-test.expectedNumberOfErrors, len(updatedBuffers))
 		})
 	}
 }

cluster-autoscaler/capacitybuffer/updater/updater.go

@@ -0,0 +1,27 @@
+/*
+Copyright 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 updater
+
+import (
+	v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1beta1"
+)
+
+// Updater updates the passed buffers via API server call and returns the
+// successfully updated buffers and list of errors
+type Updater interface {
+	Update(buffers []*v1.CapacityBuffer) ([]*v1.CapacityBuffer, []error)
+}