Cache node deltas in resource quotas

Author: norbertcyran

cluster-autoscaler/resourcequotas/cache.go

@@ -0,0 +1,52 @@
+/*
+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 resourcequotas
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	cacontext "k8s.io/autoscaler/cluster-autoscaler/context"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/customresources"
+)
+
+type nodeCache struct {
+	crp    customresources.CustomResourcesProcessor
+	deltas map[string]resourceList
+}
+
+func newNodeCache(crp customresources.CustomResourcesProcessor) *nodeCache {
+	return &nodeCache{
+		crp:    crp,
+		deltas: make(map[string]resourceList),
+	}
+}
+
+func (nc *nodeCache) nodeResources(autoscalingCtx *cacontext.AutoscalingContext, node *corev1.Node, nodeGroup cloudprovider.NodeGroup) (resourceList, error) {
+	if nodeGroup != nil {
+		if delta, ok := nc.deltas[nodeGroup.Id()]; ok {
+			return delta, nil
+		}
+	}
+	delta, err := nodeResources(autoscalingCtx, nc.crp, node, nodeGroup)
+	if err != nil {
+		return nil, err
+	}
+	if nodeGroup != nil {
+		nc.deltas[nodeGroup.Id()] = delta
+	}
+	return delta, nil
+}

cluster-autoscaler/resourcequotas/cache_test.go

@@ -0,0 +1,119 @@
+/*
+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 resourcequotas
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/mock"
+	apiv1 "k8s.io/api/core/v1"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	cptest "k8s.io/autoscaler/cluster-autoscaler/cloudprovider/test"
+	"k8s.io/autoscaler/cluster-autoscaler/context"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/customresources"
+	drasnapshot "k8s.io/autoscaler/cluster-autoscaler/simulator/dynamicresources/snapshot"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/test"
+)
+
+type mockCustomResourcesProcessor struct {
+	mock.Mock
+}
+
+func (m *mockCustomResourcesProcessor) FilterOutNodesWithUnreadyResources(autoscalingCtx *context.AutoscalingContext, allNodes, readyNodes []*apiv1.Node, draSnapshot *drasnapshot.Snapshot) ([]*apiv1.Node, []*apiv1.Node) {
+	return allNodes, readyNodes
+}
+
+func (m *mockCustomResourcesProcessor) GetNodeResourceTargets(autoscalingCtx *context.AutoscalingContext, node *apiv1.Node, nodeGroup cloudprovider.NodeGroup) ([]customresources.CustomResourceTarget, errors.AutoscalerError) {
+	args := m.Called(autoscalingCtx, node, nodeGroup)
+	return args.Get(0).([]customresources.CustomResourceTarget), nil
+}
+
+func (m *mockCustomResourcesProcessor) CleanUp() {
+	return
+}
+
+func TestNodeCacheNodeResources(t *testing.T) {
+	node := test.BuildTestNode("n1", 1000, 2000)
+	autoscalingCtx := &context.AutoscalingContext{}
+	ng1 := cptest.NewTestNodeGroup("ng1", 1, 10, 1, true, false, "n1-template", nil, nil)
+	ng2 := cptest.NewTestNodeGroup("ng2", 1, 10, 1, true, false, "n2-template", nil, nil)
+	resourceTargets := []customresources.CustomResourceTarget{
+		{ResourceType: "gpu", ResourceCount: 1},
+	}
+	wantResources := resourceList{"cpu": 1, "memory": 2000, "nodes": 1, "gpu": 1}
+
+	type nodeResourcesCall struct {
+		node      *apiv1.Node
+		nodeGroup cloudprovider.NodeGroup
+	}
+
+	testCases := []struct {
+		name                 string
+		calls                []nodeResourcesCall
+		setupCRPExpectations func(*mock.Mock)
+	}{
+		{
+			name: "cache hit",
+			calls: []nodeResourcesCall{
+				{node: node, nodeGroup: ng1},
+				{node: node, nodeGroup: ng1},
+			},
+			setupCRPExpectations: func(m *mock.Mock) {
+				m.On("GetNodeResourceTargets", autoscalingCtx, node, ng1).Return(resourceTargets, nil).Once()
+			},
+		},
+		{
+			name: "cache miss on different node group",
+			calls: []nodeResourcesCall{
+				{node: node, nodeGroup: ng1},
+				{node: node, nodeGroup: ng2},
+			},
+			setupCRPExpectations: func(m *mock.Mock) {
+				m.On("GetNodeResourceTargets", autoscalingCtx, node, ng1).Return(resourceTargets, nil).Once().
+					On("GetNodeResourceTargets", autoscalingCtx, node, ng2).Return(resourceTargets, nil).Once()
+			},
+		},
+		{
+			name: "no node group bypasses cache",
+			calls: []nodeResourcesCall{
+				{node: node, nodeGroup: nil},
+				{node: node, nodeGroup: nil},
+			},
+			setupCRPExpectations: func(m *mock.Mock) {
+				m.On("GetNodeResourceTargets", autoscalingCtx, node, nil).Return(resourceTargets, nil).Twice()
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockCRP := &mockCustomResourcesProcessor{}
+			tc.setupCRPExpectations(&mockCRP.Mock)
+			nc := newNodeCache(mockCRP)
+			for _, call := range tc.calls {
+				resources, err := nc.nodeResources(autoscalingCtx, call.node, call.nodeGroup)
+				if err != nil {
+					t.Fatalf("nodeResources unexpected error: %v", err)
+				}
+				if diff := cmp.Diff(wantResources, resources); diff != "" {
+					t.Errorf("nodeResources mismatch (-want, +got):\n%s", diff)
+				}
+			}
+		})
+	}
+}

cluster-autoscaler/resourcequotas/factory.go

@@ -24,9 +24,9 @@ import (
 
 // TrackerFactory builds quota trackers.
 type TrackerFactory struct {
-	crp             customresources.CustomResourcesProcessor
-	quotasProvider  Provider
-	usageCalculator *usageCalculator
+	crp            customresources.CustomResourcesProcessor
+	quotasProvider Provider
+	nodeFilter     NodeFilter
 }
 
 // TrackerOptions stores configuration for quota tracking.
@@ -38,11 +38,10 @@ type TrackerOptions struct {
 
 // NewTrackerFactory creates a new TrackerFactory.
 func NewTrackerFactory(opts TrackerOptions) *TrackerFactory {
-	uc := newUsageCalculator(opts.CustomResourcesProcessor, opts.NodeFilter)
 	return &TrackerFactory{
-		crp:             opts.CustomResourcesProcessor,
-		quotasProvider:  opts.QuotaProvider,
-		usageCalculator: uc,
+		crp:            opts.CustomResourcesProcessor,
+		quotasProvider: opts.QuotaProvider,
+		nodeFilter:     opts.NodeFilter,
 	}
 }
 
@@ -56,7 +55,9 @@ func (f *TrackerFactory) NewQuotasTracker(autoscalingCtx *context.AutoscalingCon
 	if err != nil {
 		return nil, err
 	}
-	usages, err := f.usageCalculator.calculateUsages(autoscalingCtx, nodes, quotas)
+	nc := newNodeCache(f.crp)
+	uc := newUsageCalculator(f.nodeFilter, nc)
+	usages, err := uc.calculateUsages(autoscalingCtx, nodes, quotas)
 	if err != nil {
 		return nil, err
 	}
@@ -73,6 +74,6 @@ func (f *TrackerFactory) NewQuotasTracker(autoscalingCtx *context.AutoscalingCon
 			limitsLeft: limitsLeft,
 		})
 	}
-	tracker := newTracker(f.crp, quotaStatuses)
+	tracker := newTracker(quotaStatuses, nc)
 	return tracker, nil
 }

cluster-autoscaler/resourcequotas/tracker.go

@@ -45,8 +45,8 @@ type resourceList map[string]int64
 
 // Tracker tracks resource quotas.
 type Tracker struct {
-	crp           customresources.CustomResourcesProcessor
 	quotaStatuses []*quotaStatus
+	nodeCache     *nodeCache
 }
 
 type quotaStatus struct {
@@ -55,10 +55,10 @@ type quotaStatus struct {
 }
 
 // newTracker creates a new Tracker.
-func newTracker(crp customresources.CustomResourcesProcessor, quotaStatuses []*quotaStatus) *Tracker {
+func newTracker(quotaStatuses []*quotaStatus, nodeCache *nodeCache) *Tracker {
 	return &Tracker{
-		crp:           crp,
 		quotaStatuses: quotaStatuses,
+		nodeCache:     nodeCache,
 	}
 }
 
@@ -67,7 +67,7 @@ func newTracker(crp customresources.CustomResourcesProcessor, quotaStatuses []*q
 func (t *Tracker) ApplyDelta(
 	autoscalingCtx *context.AutoscalingContext, nodeGroup cloudprovider.NodeGroup, node *corev1.Node, nodeDelta int,
 ) (*CheckDeltaResult, error) {
-	delta, err := nodeResources(autoscalingCtx, t.crp, node, nodeGroup)
+	delta, err := t.nodeCache.nodeResources(autoscalingCtx, node, nodeGroup)
 	if err != nil {
 		return nil, err
 	}
@@ -100,8 +100,7 @@ func (t *Tracker) ApplyDelta(
 func (t *Tracker) CheckDelta(
 	autoscalingCtx *context.AutoscalingContext, nodeGroup cloudprovider.NodeGroup, node *corev1.Node, nodeDelta int,
 ) (*CheckDeltaResult, error) {
-	// TODO: cache deltas
-	delta, err := nodeResources(autoscalingCtx, t.crp, node, nodeGroup)
+	delta, err := t.nodeCache.nodeResources(autoscalingCtx, node, nodeGroup)
 	if err != nil {
 		return nil, err
 	}