Merge pull request #9356 from abdelrahman882/buffers_metric

Add capacity buffer oldest reconciliation timestamp metric

Author: k8s-ci-robot

cluster-autoscaler/builder/autoscaler.go

@@ -199,8 +199,7 @@ func (b *AutoscalerBuilder) Build(ctx context.Context) (core.Autoscaler, *loop.L
 			} else {
 				fakePodsResolver = fakepods.NewDefaultingResolver()
 			}
-			nodeBufferController := cbctrl.NewDefaultBufferController(capacitybufferClient, fakePodsResolver)
-			go nodeBufferController.Run(ctx.Done())
+			cbctrl.InitializeAndRunDefaultBufferController(ctx, capacitybufferClient, fakePodsResolver)
 		}
 	}
 

cluster-autoscaler/capacitybuffer/controller/controller.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controller
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"sort"
@@ -36,8 +37,10 @@ import (
 	cbclient "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/client"
 	"k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/fakepods"
 	filters "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/filters"
+	cbmetrics "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/metrics"
 	translators "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/translators"
 	updater "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/updater"
+	"k8s.io/utils/clock"
 )
 
 // BufferController performs updates on Buffers and convert them to pods to be injected
@@ -47,12 +50,14 @@ type BufferController interface {
 }
 
 type bufferController struct {
-	client         *cbclient.CapacityBufferClient
-	strategyFilter filters.Filter
-	translator     translators.Translator
-	quotaAllocator *resourceQuotaAllocator
-	updater        updater.StatusUpdater
-	queue          workqueue.TypedRateLimitingInterface[string]
+	client                  *cbclient.CapacityBufferClient
+	strategyFilter          filters.Filter
+	translator              translators.Translator
+	quotaAllocator          *resourceQuotaAllocator
+	updater                 updater.StatusUpdater
+	queue                   workqueue.TypedRateLimitingInterface[string]
+	clock                   clock.Clock
+	reconciliationTimeCache *cbmetrics.ReconciliationCache
 }
 
 // NewBufferController creates new bufferController object
@@ -61,6 +66,8 @@ func NewBufferController(
 	strategyFilter filters.Filter,
 	translator translators.Translator,
 	updater updater.StatusUpdater,
+	clock clock.Clock,
+	reconciliationTimeCache *cbmetrics.ReconciliationCache,
 ) BufferController {
 	bc := &bufferController{
 		client:         client,
@@ -71,20 +78,42 @@ func NewBufferController(
 		queue: workqueue.NewTypedRateLimitingQueueWithConfig(
 			workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{Name: "CapacityBuffers"},
 		),
+		clock:                   clock,
+		reconciliationTimeCache: reconciliationTimeCache,
 	}
 	bc.configureEventHandlers()
 	return bc
 }
 
+// InitializeAndRunDefaultBufferController creates the default Capacity buffer controller and processing interval metric collector
+// and runs each of them asyncrounsly
+func InitializeAndRunDefaultBufferController(
+	ctx context.Context,
+	client *cbclient.CapacityBufferClient,
+	resolver fakepods.Resolver,
+
+) {
+	realClock := clock.RealClock{}
+	reconciledBuffersCache := cbmetrics.NewReconciliationCache()
+	// Accepting empty string as it represents nil value for ProvisioningStrategy
+	defaultStrategies := []string{capacitybuffer.ActiveProvisioningStrategy, ""}
+	controller := NewDefaultBufferController(client, resolver, defaultStrategies, reconciledBuffersCache, realClock)
+	go controller.Run(ctx.Done())
+
+	cbmetrics.RegisterReconciliationTimestampCollector(client, defaultStrategies, reconciledBuffersCache, realClock)
+}
+
 // NewDefaultBufferController creates bufferController with default configs
 func NewDefaultBufferController(
 	client *cbclient.CapacityBufferClient,
 	resolver fakepods.Resolver,
+	strategies []string,
+	reconciliationTimeCache *cbmetrics.ReconciliationCache,
+	clock clock.Clock,
 ) BufferController {
 	bc := &bufferController{
-		client: client,
-		// Accepting empty string as it represents nil value for ProvisioningStrategy
-		strategyFilter: filters.NewStrategyFilter([]string{capacitybuffer.ActiveProvisioningStrategy, ""}),
+		client:         client,
+		strategyFilter: filters.NewStrategyFilter(strategies),
 		translator: translators.NewCombinedTranslator(
 			[]translators.Translator{
 				translators.NewPodTemplateBufferTranslator(client, resolver),
@@ -96,6 +125,8 @@ func NewDefaultBufferController(
 		queue: workqueue.NewTypedRateLimitingQueueWithConfig(
 			workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{Name: "CapacityBuffers"},
 		),
+		clock:                   clock,
+		reconciliationTimeCache: reconciliationTimeCache,
 	}
 	bc.configureEventHandlers()
 	return bc
@@ -278,7 +309,10 @@ func (c *bufferController) reconcileNamespace(namespace string) error {
 
 	// Filter the desired provisioning strategy
 	// Note: We process ALL buffers in the namespace that match the strategy.
-	filteredBuffers, _ := c.strategyFilter.Filter(buffers)
+	filteredBuffers, filteredOutBuffers := c.strategyFilter.Filter(buffers)
+
+	// Update reconciliation time for filtered out buffers
+	c.updateReconciliationTimeCache(filteredOutBuffers)
 
 	if len(filteredBuffers) == 0 {
 		return nil
@@ -306,7 +340,8 @@ func (c *bufferController) reconcileNamespace(namespace string) error {
 	}
 
 	// Update buffer status by calling API server
-	updateErrors := c.updater.Update(filteredBuffers)
+	updatedBuffers, updateErrors := c.updater.Update(filteredBuffers)
+	c.updateReconciliationTimeCache(updatedBuffers)
 	for _, err := range updateErrors {
 		runtime.HandleError(fmt.Errorf("capacity buffer controller error: %w", err))
 	}
@@ -318,3 +353,10 @@ func (c *bufferController) reconcileNamespace(namespace string) error {
 
 	return nil
 }
+
+func (c *bufferController) updateReconciliationTimeCache(buffers []*v1.CapacityBuffer) {
+	if c.reconciliationTimeCache == nil || len(buffers) == 0 {
+		return
+	}
+	c.reconciliationTimeCache.Update(buffers, c.clock.Now())
+}

cluster-autoscaler/capacitybuffer/controller/controller_test.go

@@ -29,18 +29,74 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	v1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1beta1"
 	fakebuffers "k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/client/clientset/versioned/fake"
 	"k8s.io/autoscaler/cluster-autoscaler/capacitybuffer"
 	cbclient "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/client"
 	"k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/fakepods"
+	"k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/metrics"
 	"k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/testutil"
 	fakek8s "k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/utils/clock"
+	testclock "k8s.io/utils/clock/testing"
 	"k8s.io/utils/ptr"
 )
 
+func TestController_CacheUpdates(t *testing.T) {
+	now := metav1.Now().Time
+	fakeClock := testclock.NewFakeClock(now)
+	reconciliationCache := metrics.NewReconciliationCache()
+
+	bSupported := &v1.CapacityBuffer{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "supported",
+			Namespace: "default",
+			UID:       types.UID("uid-supported"),
+		},
+		Spec: v1.CapacityBufferSpec{
+			ProvisioningStrategy: ptr.To(capacitybuffer.ActiveProvisioningStrategy),
+		},
+	}
+	bUnsupported := &v1.CapacityBuffer{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "unsupported",
+			Namespace: "default",
+			UID:       types.UID("uid-unsupported"),
+		},
+		Spec: v1.CapacityBufferSpec{
+			ProvisioningStrategy: ptr.To("unsupported-strategy"),
+		},
+	}
+
+	buffersClient := fakebuffers.NewSimpleClientset(bSupported, bUnsupported)
+	k8sClient := fakek8s.NewSimpleClientset()
+	client, _ := cbclient.NewCapacityBufferClientFromClients(buffersClient, k8sClient, nil, nil)
+
+	resolver := fakepods.NewDefaultingResolver()
+	controller := NewDefaultBufferController(
+		client,
+		resolver,
+		[]string{capacitybuffer.ActiveProvisioningStrategy},
+		reconciliationCache,
+		fakeClock,
+	).(*bufferController)
+
+	// Reconcile
+	err := controller.reconcileNamespace("default")
+	assert.NoError(t, err)
+
+	// Verify Cache
+	snapshot := reconciliationCache.Snapshot()
+	assert.Len(t, snapshot, 2)
+	assert.Contains(t, snapshot, types.UID("uid-supported"))
+	assert.Contains(t, snapshot, types.UID("uid-unsupported"))
+	assert.Equal(t, now, snapshot[types.UID("uid-supported")])
+	assert.Equal(t, now, snapshot[types.UID("uid-unsupported")])
+}
+
 func TestControllerIntegration_ResourceQuotas(t *testing.T) {
 	// TODO: refactor to ginkgo and envtest
 	// Setup
@@ -91,7 +147,7 @@ func TestControllerIntegration_ResourceQuotas(t *testing.T) {
 
 	// TODO: use DryRunResolver once migrated to envtest
 	resolver := fakepods.NewDefaultingResolver()
-	controller := NewDefaultBufferController(client, resolver).(*bufferController)
+	controller := NewDefaultBufferController(client, resolver, []string{capacitybuffer.ActiveProvisioningStrategy}, metrics.NewReconciliationCache(), clock.RealClock{}).(*bufferController)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

cluster-autoscaler/capacitybuffer/metrics/reconciliation_cache.go

@@ -0,0 +1,82 @@
+/*
+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 metrics
+
+import (
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/autoscaler/cluster-autoscaler/apis/capacitybuffer/autoscaling.x-k8s.io/v1beta1"
+)
+
+// ReconciliationCache is a thread safe map for buffers last reconciliation time
+type ReconciliationCache struct {
+	reconciledBuffers map[types.UID]time.Time
+	mutex             sync.RWMutex
+}
+
+// NewReconciliationCache creates a new instance of ReconciliationCache
+func NewReconciliationCache() *ReconciliationCache {
+	return &ReconciliationCache{
+		reconciledBuffers: make(map[types.UID]time.Time),
+	}
+}
+
+// Update updates the underlying map with the provided buffers and time.
+func (r *ReconciliationCache) Update(buffers []*v1beta1.CapacityBuffer, t time.Time) {
+	if len(buffers) == 0 {
+		return
+	}
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+
+	for _, buffer := range buffers {
+		r.reconciledBuffers[buffer.UID] = t
+	}
+}
+
+// Prune removes entries from the cache that are not present in the provided buffers list.
+func (r *ReconciliationCache) Prune(buffers []*v1beta1.CapacityBuffer) {
+	existingBuffers := sets.New[types.UID]()
+	for _, buffer := range buffers {
+		existingBuffers.Insert(buffer.UID)
+	}
+
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+
+	for uid := range r.reconciledBuffers {
+		if !existingBuffers.Has(uid) {
+			delete(r.reconciledBuffers, uid)
+		}
+	}
+}
+
+// Snapshot creates and returns a copy of the current map.
+func (r *ReconciliationCache) Snapshot() map[types.UID]time.Time {
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+
+	snapshot := make(map[types.UID]time.Time, len(r.reconciledBuffers))
+	for key, value := range r.reconciledBuffers {
+		snapshot[key] = value
+	}
+
+	return snapshot
+}