diff --git a/pkg/scheduler/framework/plugins/noderesources/balanced_allocation.go b/pkg/scheduler/framework/plugins/noderesources/balanced_allocation.go index 48e4caf6c9b..d2b2829e308 100644 --- a/pkg/scheduler/framework/plugins/noderesources/balanced_allocation.go +++ b/pkg/scheduler/framework/plugins/noderesources/balanced_allocation.go @@ -22,7 +22,9 @@ import ( "math" v1 "k8s.io/api/core/v1" + resourceapi "k8s.io/api/resource/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/dynamic-resource-allocation/structured" fwk "k8s.io/kube-scheduler/framework" "k8s.io/kubernetes/pkg/scheduler/apis/config" "k8s.io/kubernetes/pkg/scheduler/apis/config/validation" @@ -48,11 +50,21 @@ const ( balancedAllocationPreScoreStateKey = "PreScore" + BalancedAllocationName ) +// draPreScoreState holds the pre-computed data for DRA extended resources scoring. +type draPreScoreState struct { + // allocatedState holds the DRA allocated state for DRA extended resources scoring. + allocatedState *structured.AllocatedState + // resourceSlices holds the list of resource slices for DRA extended resource scoring. + resourceSlices []*resourceapi.ResourceSlice +} + // balancedAllocationPreScoreState computed at PreScore and used at Score. type balancedAllocationPreScoreState struct { // podRequests have the same order of the resources defined in NodeResourcesFitArgs.Resources, // same for other place we store a list like that. podRequests []int64 + // DRA extended resource scoring state. + *draPreScoreState } // Clone implements the mandatory Clone interface. We don't really copy the data since @@ -73,6 +85,15 @@ func (ba *BalancedAllocation) PreScore(ctx context.Context, cycleState fwk.Cycle state := &balancedAllocationPreScoreState{ podRequests: podRequests, } + if ba.enableDRAExtendedResource { + draPreScoreState, status := getDRAPreScoredParams(ba.draManager, ba.resources) + if status != nil { + return status + } + if draPreScoreState != nil { + state.draPreScoreState = draPreScoreState + } + } cycleState.Write(balancedAllocationPreScoreStateKey, state) return nil } @@ -103,6 +124,15 @@ func (ba *BalancedAllocation) Score(ctx context.Context, state fwk.CycleState, p if ba.isBestEffortPod(s.podRequests) { return 0, nil } + if ba.enableDRAExtendedResource { + draPreScoreState, status := getDRAPreScoredParams(ba.draManager, ba.resources) + if status != nil { + return 0, status + } + if draPreScoreState != nil { + s.draPreScoreState = draPreScoreState + } + } } // ba.score favors nodes with balanced resource usage rate. @@ -110,7 +140,7 @@ func (ba *BalancedAllocation) Score(ctx context.Context, state fwk.CycleState, p // Detail: score = (1 - std) * MaxNodeScore, where std is calculated by the root square of Σ((fraction(i)-mean)^2)/len(resources) // The algorithm is partly inspired by: // "Wei Huang et al. An Energy Efficient Virtual Machine Placement Algorithm with Balanced Resource Utilization" - return ba.score(ctx, pod, nodeInfo, s.podRequests) + return ba.score(ctx, pod, nodeInfo, s.podRequests, s.draPreScoreState) } // ScoreExtensions of the Score plugin. diff --git a/pkg/scheduler/framework/plugins/noderesources/fit.go b/pkg/scheduler/framework/plugins/noderesources/fit.go index ba584a4471b..be2ede7cd54 100644 --- a/pkg/scheduler/framework/plugins/noderesources/fit.go +++ b/pkg/scheduler/framework/plugins/noderesources/fit.go @@ -97,7 +97,7 @@ type Fit struct { enablePodLevelResources bool enableDRAExtendedResource bool handle fwk.Handle - resourceAllocationScorer + *resourceAllocationScorer } // ScoreExtensions of the Score plugin. @@ -120,6 +120,8 @@ type preScoreState struct { // podRequests have the same order as the resources defined in NodeResourcesBalancedAllocationArgs.Resources, // same for other place we store a list like that. podRequests []int64 + // DRA extended resource scoring related info. + *draPreScoreState } // Clone implements the mandatory Clone interface. We don't really copy the data since @@ -130,9 +132,20 @@ func (s *preScoreState) Clone() fwk.StateData { // PreScore calculates incoming pod's resource requests and writes them to the cycle state used. func (f *Fit) PreScore(ctx context.Context, cycleState fwk.CycleState, pod *v1.Pod, nodes []fwk.NodeInfo) *fwk.Status { + podRequests := f.calculatePodResourceRequestList(pod, f.resources) state := &preScoreState{ - podRequests: f.calculatePodResourceRequestList(pod, f.resources), + podRequests: podRequests, } + if f.enableDRAExtendedResource { + draPreScoreState, status := getDRAPreScoredParams(f.draManager, f.resources) + if status != nil { + return status + } + if draPreScoreState != nil { + state.draPreScoreState = draPreScoreState + } + } + cycleState.Write(preScoreStateKey, state) return nil } @@ -182,7 +195,7 @@ func NewFit(_ context.Context, plArgs runtime.Object, h fwk.Handle, fts feature. scorer.draFeatures = dynamicresources.AllocatorFeatures(fts) // Create a CEL cache for device class selector compilation // This cache improves performance by avoiding recompilation of the same CEL expressions - scorer.celCache = cel.NewCache(10, cel.Features{EnableConsumableCapacity: fts.EnableDRAConsumableCapacity}) + scorer.DRACaches.celCache = cel.NewCache(10, cel.Features{EnableConsumableCapacity: fts.EnableDRAConsumableCapacity}) } return &Fit{ @@ -194,7 +207,7 @@ func NewFit(_ context.Context, plArgs runtime.Object, h fwk.Handle, fts feature. handle: h, enablePodLevelResources: fts.EnablePodLevelResources, enableDRAExtendedResource: fts.EnableDRAExtendedResource, - resourceAllocationScorer: *scorer, + resourceAllocationScorer: scorer, }, nil } @@ -713,7 +726,16 @@ func (f *Fit) Score(ctx context.Context, state fwk.CycleState, pod *v1.Pod, node s = &preScoreState{ podRequests: f.calculatePodResourceRequestList(pod, f.resources), } + if f.enableDRAExtendedResource { + draPreScoreState, status := getDRAPreScoredParams(f.draManager, f.resources) + if status != nil { + return 0, status + } + if draPreScoreState != nil { + s.draPreScoreState = draPreScoreState + } + } } - return f.score(ctx, pod, nodeInfo, s.podRequests) + return f.score(ctx, pod, nodeInfo, s.podRequests, s.draPreScoreState) } diff --git a/pkg/scheduler/framework/plugins/noderesources/resource_allocation.go b/pkg/scheduler/framework/plugins/noderesources/resource_allocation.go index aaf7fa0faba..69d6337f714 100644 --- a/pkg/scheduler/framework/plugins/noderesources/resource_allocation.go +++ b/pkg/scheduler/framework/plugins/noderesources/resource_allocation.go @@ -18,6 +18,8 @@ package noderesources import ( "context" + "strings" + "sync" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" @@ -30,13 +32,22 @@ import ( "k8s.io/dynamic-resource-allocation/structured" fwk "k8s.io/kube-scheduler/framework" "k8s.io/kubernetes/pkg/scheduler/apis/config" - "k8s.io/kubernetes/pkg/scheduler/framework/plugins/dynamicresources/extended" schedutil "k8s.io/kubernetes/pkg/scheduler/util" ) // scorer is decorator for resourceAllocationScorer type scorer func(args *config.NodeResourcesFitArgs) *resourceAllocationScorer +// DRACaches holds various caches used for DRA-related computations +type DRACaches struct { + // celCache is a cache for compiled CEL expressions used in device class selectors. + celCache *cel.Cache + // Cache for DeviceMatches results to avoid expensive repeated evaluations + deviceMatchCache sync.Map // map[deviceMatchCacheKey]bool + // Cache for NodeMatches results to avoid expensive repeated node selector evaluations + nodeMatchCache sync.Map // map[nodeMatchCacheKey]bool +} + // resourceAllocationScorer contains information to calculate resource allocation score. type resourceAllocationScorer struct { Name string @@ -50,7 +61,76 @@ type resourceAllocationScorer struct { resources []config.ResourceSpec draFeatures structured.Features draManager fwk.SharedDRAManager - celCache *cel.Cache + // Caches for DRA-related computations + DRACaches +} + +// buildNodeMatchCacheKey creates a string cache key for node matching results +// Using a string key is significantly faster than struct keys with sync.Map +func buildNodeMatchCacheKey(nodeName string, nodeNameToMatch string, allNodesMatch bool, nodeSelectorHash string) string { + // Pre-allocate sufficient capacity to avoid reallocation + var b strings.Builder + b.Grow(len(nodeName) + len(nodeNameToMatch) + len(nodeSelectorHash) + 4) + + b.WriteString(nodeName) + b.WriteByte('|') + b.WriteString(nodeNameToMatch) + b.WriteByte('|') + if allNodesMatch { + b.WriteByte('1') + } else { + b.WriteByte('0') + } + b.WriteByte('|') + b.WriteString(nodeSelectorHash) + + return b.String() +} + +// buildDeviceMatchCacheKey creates a string cache key for device matching results +// Using a string key is significantly faster than struct keys with sync.Map +// This concatenates expression|driver|poolName|deviceName with pipe separators +func buildDeviceMatchCacheKey(expression string, driver string, poolName string, deviceName string) string { + // Pre-allocate sufficient capacity to avoid reallocation + var b strings.Builder + b.Grow(len(expression) + len(driver) + len(poolName) + len(deviceName) + 3) + + b.WriteString(expression) + b.WriteByte('|') + b.WriteString(driver) + b.WriteByte('|') + b.WriteString(poolName) + b.WriteByte('|') + b.WriteString(deviceName) + + return b.String() +} + +// nodeMatches is a cached wrapper around structured.NodeMatches +func (r *resourceAllocationScorer) nodeMatches(node *v1.Node, nodeNameToMatch string, allNodesMatch bool, nodeSelector *v1.NodeSelector) (bool, error) { + + var nodeName string + if node != nil { + nodeName = node.Name + } + + nodeSelectorStr := nodeSelector.String() + key := buildNodeMatchCacheKey(nodeName, nodeNameToMatch, allNodesMatch, nodeSelectorStr) + + // Check cache first + if matches, ok := r.nodeMatchCache.Load(key); ok { + return matches.(bool), nil + } + + // Call the original function + matches, err := structured.NodeMatches(r.draFeatures, node, nodeNameToMatch, allNodesMatch, nodeSelector) + + // Cache the result (even if there was an error, to avoid repeated failures) + if err == nil { + r.nodeMatchCache.Store(key, matches) + } + + return matches, err } // score will use `scorer` function to calculate the score. @@ -58,7 +138,9 @@ func (r *resourceAllocationScorer) score( ctx context.Context, pod *v1.Pod, nodeInfo fwk.NodeInfo, - podRequests []int64) (int64, *fwk.Status) { + podRequests []int64, + draPreScoreState *draPreScoreState, +) (int64, *fwk.Status) { logger := klog.FromContext(ctx) node := nodeInfo.Node() @@ -70,7 +152,7 @@ func (r *resourceAllocationScorer) score( requested := make([]int64, len(r.resources)) allocatable := make([]int64, len(r.resources)) for i := range r.resources { - alloc, req := r.calculateResourceAllocatableRequest(ctx, nodeInfo, v1.ResourceName(r.resources[i].Name), podRequests[i]) + alloc, req := r.calculateResourceAllocatableRequest(ctx, nodeInfo, v1.ResourceName(r.resources[i].Name), podRequests[i], draPreScoreState) // Only fill the extended resource entry when it's non-zero. if alloc == 0 { continue @@ -95,7 +177,13 @@ func (r *resourceAllocationScorer) score( // - 1st param: quantity of allocatable resource on the node. // - 2nd param: aggregated quantity of requested resource on the node. // Note: if it's an extended resource, and the pod doesn't request it, (0, 0) is returned. -func (r *resourceAllocationScorer) calculateResourceAllocatableRequest(ctx context.Context, nodeInfo fwk.NodeInfo, resource v1.ResourceName, podRequest int64) (int64, int64) { +func (r *resourceAllocationScorer) calculateResourceAllocatableRequest( + ctx context.Context, + nodeInfo fwk.NodeInfo, + resource v1.ResourceName, + podRequest int64, + draPreScoreState *draPreScoreState, +) (int64, int64) { requested := nodeInfo.GetNonZeroRequested() if r.useRequested { requested = nodeInfo.GetRequested() @@ -115,10 +203,10 @@ func (r *resourceAllocationScorer) calculateResourceAllocatableRequest(ctx conte return nodeInfo.GetAllocatable().GetEphemeralStorage(), (nodeInfo.GetRequested().GetEphemeralStorage() + podRequest) default: allocatable, exists := nodeInfo.GetAllocatable().GetScalarResources()[resource] - if allocatable == 0 && r.enableDRAExtendedResource { + if allocatable == 0 && r.enableDRAExtendedResource && draPreScoreState != nil { // Allocatable 0 means that this resource is not handled by device plugin. // Calculate allocatable and requested for resources backed by DRA. - allocatable, allocated := r.calculateDRAExtendedResourceAllocatableRequest(ctx, nodeInfo.Node(), resource) + allocatable, allocated := r.calculateDRAExtendedResourceAllocatableRequest(ctx, nodeInfo.Node(), resource, draPreScoreState) if allocatable > 0 { return allocatable, allocated + podRequest } @@ -174,36 +262,63 @@ func (r *resourceAllocationScorer) isBestEffortPod(podRequests []int64) bool { return true } +// getDRAPreScoredParams returns the DRA allocated state and resource slices for DRA extended resource scoring. +func getDRAPreScoredParams(draManager fwk.SharedDRAManager, resources []config.ResourceSpec) (*draPreScoreState, *fwk.Status) { + anyBackedByDRA := false + for _, resource := range resources { + resourceName := v1.ResourceName(resource.Name) + if !schedutil.IsDRAExtendedResourceName(resourceName) { + continue + } + deviceClass := draManager.DeviceClassResolver().GetDeviceClass(resourceName) + if deviceClass != nil { + anyBackedByDRA = true + break + } + } + // There's no point in returning DRA data as there are no resources backed by DRA. + if !anyBackedByDRA { + return nil, nil + } + + allocatedState, err := draManager.ResourceClaims().GatherAllocatedState() + if err != nil { + return nil, fwk.AsStatus(err) + } + resourceSlices, err := draManager.ResourceSlices().ListWithDeviceTaintRules() + if err != nil { + return nil, fwk.AsStatus(err) + } + + return &draPreScoreState{ + allocatedState: allocatedState, + resourceSlices: resourceSlices, + }, nil +} + // calculateDRAExtendedResourceAllocatableRequest calculates allocatable and allocated // quantities for extended resources backed by DRA. -func (r *resourceAllocationScorer) calculateDRAExtendedResourceAllocatableRequest(ctx context.Context, node *v1.Node, resource v1.ResourceName) (int64, int64) { +func (r *resourceAllocationScorer) calculateDRAExtendedResourceAllocatableRequest( + ctx context.Context, + node *v1.Node, + resource v1.ResourceName, + draPreScoreState *draPreScoreState, +) (int64, int64) { logger := klog.FromContext(ctx) - // Get device class mapping to find the device class for this resource - deviceClassMapping, err := extended.DeviceClassMapping(r.draManager) + deviceClass := r.draManager.DeviceClassResolver().GetDeviceClass(resource) + if deviceClass == nil { + // This resource is not backed by DRA. + logger.V(7).Info("Extended resource not found in device class mapping", "resource", resource) + return 0, 0 + } + + capacity, allocated, err := r.calculateDRAResourceTotals(ctx, node, deviceClass, draPreScoreState.allocatedState, draPreScoreState.resourceSlices) if err != nil { - logger.Error(err, "Failed to get device class mapping for DRA extended resource scoring") + logger.Error(err, "Failed to calculate DRA resource capacity and allocated", "node", node.Name, "resource", resource, "deviceClass", deviceClass.Name) return 0, 0 } - deviceClassName, exists := deviceClassMapping[resource] - if !exists { - logger.Error(nil, "Extended resource not found in device class mapping", "resource", resource) - return 0, 0 - } - - deviceClass, err := r.draManager.DeviceClasses().Get(deviceClassName) - if err != nil { - logger.Error(err, "Failed to get device class for DRA extended resource scoring", "resource", resource, "deviceClass", deviceClassName) - return 0, 0 - } - - capacity, allocated, err := r.calculateDRAResourceTotals(ctx, node, deviceClass) - if err != nil { - logger.Error(err, "Failed to calculate DRA resource capacity and allocated", "node", node.Name, "resource", resource, "deviceClass", deviceClassName) - return 0, 0 - } - - logger.V(7).Info("DRA extended resource calculation", "node", node.Name, "resource", resource, "deviceClass", deviceClassName, "capacity", capacity, "allocated", allocated) + logger.V(7).Info("DRA extended resource calculation", "node", node.Name, "resource", resource, "deviceClass", deviceClass.Name, "capacity", capacity, "allocated", allocated) return capacity, allocated } @@ -223,62 +338,92 @@ func (r *resourceAllocationScorer) calculateDRAExtendedResourceAllocatableReques // totalCapacity - total number of devices matching the device class on the node // totalAllocated - number of devices currently allocated from the matching set // error - any error encountered during processing -func (r *resourceAllocationScorer) calculateDRAResourceTotals(ctx context.Context, node *v1.Node, deviceClass *resourceapi.DeviceClass) (int64, int64, error) { - allocatedState, err := r.draManager.ResourceClaims().GatherAllocatedState() - if err != nil { - return 0, 0, err - } - - resourceSlices, err := r.draManager.ResourceSlices().ListWithDeviceTaintRules() - if err != nil { - return 0, 0, err - } - +func (r *resourceAllocationScorer) calculateDRAResourceTotals(ctx context.Context, node *v1.Node, deviceClass *resourceapi.DeviceClass, allocatedState *structured.AllocatedState, resourceSlices []*resourceapi.ResourceSlice, +) (int64, int64, error) { var totalCapacity, totalAllocated int64 + nodeName := node.Name + for _, slice := range resourceSlices { - driver := slice.Spec.Driver - pool := slice.Spec.Pool.Name + // Early filtering: check if slice applies to this node + perDeviceNodeSelection := ptr.Deref(slice.Spec.PerDeviceNodeSelection, false) + var devices []resourceapi.Device - // Handle per-device node selection vs slice-level node selection - if ptr.Deref(slice.Spec.PerDeviceNodeSelection, false) { - devices = []resourceapi.Device{} - // When per-device node selection is enabled, check each device individually + + if perDeviceNodeSelection { + // Per-device node selection: filter devices individually + devices = make([]resourceapi.Device, 0, len(slice.Spec.Devices)) for _, device := range slice.Spec.Devices { - // Check if this specific device matches the node - deviceMatches, err := structured.NodeMatches(r.draFeatures, node, ptr.Deref(device.NodeName, ""), ptr.Deref(device.AllNodes, false), device.NodeSelector) - if err != nil { - return 0, 0, err - } - if deviceMatches { + deviceNodeName := ptr.Deref(device.NodeName, "") + deviceAllNodes := ptr.Deref(device.AllNodes, false) + + // Fast path: check AllNodes or exact name match first + if deviceAllNodes || (deviceNodeName != "" && deviceNodeName == nodeName) { devices = append(devices, device) + continue + } + + // Slow path: only if we have a node selector + if device.NodeSelector != nil { + deviceMatches, err := r.nodeMatches(node, deviceNodeName, deviceAllNodes, device.NodeSelector) + if err != nil { + return 0, 0, err + } + if deviceMatches { + devices = append(devices, device) + } } } } else { - // When per-device node selection is disabled, check slice-level node selection first - matches, err := structured.NodeMatches(r.draFeatures, node, ptr.Deref(slice.Spec.NodeName, ""), ptr.Deref(slice.Spec.AllNodes, false), slice.Spec.NodeSelector) - if err != nil { - return 0, 0, err - } - if !matches { - // Skip this slice as it doesn't match the node + // Slice-level node selection + sliceNodeName := ptr.Deref(slice.Spec.NodeName, "") + sliceAllNodes := ptr.Deref(slice.Spec.AllNodes, false) + + // Fast path: check AllNodes or exact name match first + if !sliceAllNodes && sliceNodeName != nodeName && slice.Spec.NodeSelector != nil { + // Need to check node selector + matches, err := r.nodeMatches(node, sliceNodeName, sliceAllNodes, slice.Spec.NodeSelector) + if err != nil { + return 0, 0, err + } + if !matches { + continue // Skip this slice + } + } else if !sliceAllNodes && sliceNodeName != "" && sliceNodeName != nodeName { + // Node name specified but doesn't match continue } + devices = slice.Spec.Devices } - // Count devices that match the device class - for _, device := range devices { - matches, err := r.deviceMatchesClass(ctx, device, deviceClass, driver) - if err != nil { - return 0, 0, err - } - if matches { + + // Fast path for device class with no selectors + if len(deviceClass.Spec.Selectors) == 0 { + driver := slice.Spec.Driver + pool := slice.Spec.Pool.Name + for _, device := range devices { totalCapacity++ - // Count allocated devices (both fully allocated and partially consumed) deviceID := structured.MakeDeviceID(driver, pool, device.Name) if structured.IsDeviceAllocated(deviceID, allocatedState) { totalAllocated++ } } + } else { + // Slow path: check device class selectors + driver := slice.Spec.Driver + pool := slice.Spec.Pool.Name + for _, device := range devices { + matches, err := r.deviceMatchesClass(ctx, device, deviceClass, driver, pool) + if err != nil { + return 0, 0, err + } + if matches { + totalCapacity++ + deviceID := structured.MakeDeviceID(driver, pool, device.Name) + if structured.IsDeviceAllocated(deviceID, allocatedState) { + totalAllocated++ + } + } + } } } @@ -289,35 +434,56 @@ func (r *resourceAllocationScorer) calculateDRAResourceTotals(ctx context.Contex // Note: This method assumes the device class has ExtendedResourceName set, as filtering // should be done by the caller to ensure we only process DRA resources meant for extended // resource scoring. -func (r *resourceAllocationScorer) deviceMatchesClass(ctx context.Context, device resourceapi.Device, deviceClass *resourceapi.DeviceClass, driver string) (bool, error) { +func (r *resourceAllocationScorer) deviceMatchesClass(ctx context.Context, device resourceapi.Device, deviceClass *resourceapi.DeviceClass, driver string, poolName string) (bool, error) { // If no selectors are defined, all devices match if len(deviceClass.Spec.Selectors) == 0 { return true, nil } + // Lazily create the CEL device only when needed (first CEL selector that's not cached) + var celDevice cel.Device + celDeviceCreated := false + // All selectors must match for the device to be considered a match for _, selector := range deviceClass.Spec.Selectors { if selector.CEL == nil { continue } + key := buildDeviceMatchCacheKey(selector.CEL.Expression, driver, poolName, device.Name) + + // Check if result is already cached + if matches, ok := r.deviceMatchCache.Load(key); ok { + if !matches.(bool) { + return false, nil + } + continue // This selector matches, check the next one + } + + // Cache miss - need to evaluate CEL expression + // Create CEL device if we haven't already + if !celDeviceCreated { + celDevice = cel.Device{ + Driver: driver, + Attributes: device.Attributes, + Capacity: device.Capacity, + } + celDeviceCreated = true + } + // Use cached CEL compilation for performance result := r.celCache.GetOrCompile(selector.CEL.Expression) if result.Error != nil { return false, result.Error } - // Evaluate the expression against the device - celDevice := cel.Device{ - Driver: driver, - Attributes: device.Attributes, - Capacity: device.Capacity, - } - matches, _, err := result.DeviceMatches(ctx, celDevice) if err != nil || !matches { - return false, nil + return false, err } + + // Cache the result for future use + r.deviceMatchCache.Store(key, matches) } return true, nil diff --git a/pkg/scheduler/framework/plugins/noderesources/resource_allocation_test.go b/pkg/scheduler/framework/plugins/noderesources/resource_allocation_test.go index 25e2eb996d9..f6b6c2aa3ae 100644 --- a/pkg/scheduler/framework/plugins/noderesources/resource_allocation_test.go +++ b/pkg/scheduler/framework/plugins/noderesources/resource_allocation_test.go @@ -20,20 +20,26 @@ import ( "context" "testing" + "github.com/google/go-cmp/cmp" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" v1 "k8s.io/api/core/v1" resourceapi "k8s.io/api/resource/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" apiruntime "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" + utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes/fake" + featuregatetesting "k8s.io/component-base/featuregate/testing" "k8s.io/dynamic-resource-allocation/cel" + "k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache" "k8s.io/dynamic-resource-allocation/resourceslice/tracker" "k8s.io/dynamic-resource-allocation/structured" "k8s.io/klog/v2/ktesting" + fwk "k8s.io/kube-scheduler/framework" + "k8s.io/kubernetes/pkg/features" + "k8s.io/kubernetes/pkg/scheduler/apis/config" "k8s.io/kubernetes/pkg/scheduler/framework" "k8s.io/kubernetes/pkg/scheduler/framework/plugins/dynamicresources" st "k8s.io/kubernetes/pkg/scheduler/testing" @@ -532,12 +538,13 @@ func TestCalculateResourceAllocatableRequest(t *testing.T) { for name, tc := range tests { t.Run(name, func(t *testing.T) { - t.Parallel() // Setup environment, create required objects logger, tCtx := ktesting.NewTestContext(t) tCtx, cancel := context.WithCancel(tCtx) defer cancel() + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.DRAExtendedResource, tc.enableDRAExtendedResource) + client := fake.NewClientset(tc.objects...) informerFactory := informers.NewSharedInformerFactory(client, 0) resourceSliceTrackerOpts := tracker.Options{ @@ -547,7 +554,9 @@ func TestCalculateResourceAllocatableRequest(t *testing.T) { KubeClient: client, } resourceSliceTracker, err := tracker.StartTracker(tCtx, resourceSliceTrackerOpts) - require.NoError(t, err, "couldn't start resource slice tracker") + if err != nil { + t.Fatalf("couldn't start resource slice tracker: %v", err) + } draManager := dynamicresources.NewDRAManager( tCtx, assumecache.NewAssumeCache( @@ -559,6 +568,13 @@ func TestCalculateResourceAllocatableRequest(t *testing.T) { resourceSliceTracker, informerFactory) + if tc.enableDRAExtendedResource { + cache := draManager.DeviceClassResolver().(*extendedresourcecache.ExtendedResourceCache) + if _, err := informerFactory.Resource().V1().DeviceClasses().Informer().AddEventHandler(cache); err != nil { + logger.Error(err, "failed to add device class informer event handler") + } + } + informerFactory.Start(tCtx.Done()) t.Cleanup(func() { // Now we can wait for all goroutines to stop. @@ -573,13 +589,323 @@ func TestCalculateResourceAllocatableRequest(t *testing.T) { enableDRAExtendedResource: tc.enableDRAExtendedResource, draManager: draManager, draFeatures: draFeatures, - celCache: celCache, + DRACaches: DRACaches{ + celCache: celCache, + }, + } + + var draPreScoreState *draPreScoreState + if tc.enableDRAExtendedResource { + var status *fwk.Status + draPreScoreState, status = getDRAPreScoredParams(draManager, []config.ResourceSpec{{Name: string(tc.extendedResource)}}) + if status != nil { + t.Fatalf("getting DRA pre-scored params failed: %v", status) + } } // Test calculateResourceAllocatableRequest API - allocatable, requested := scorer.calculateResourceAllocatableRequest(tCtx, nodeInfo, tc.extendedResource, tc.podRequest) - assert.Equal(t, tc.expectedAllocatable, allocatable) - assert.Equal(t, tc.expectedRequested, requested) + allocatable, requested := scorer.calculateResourceAllocatableRequest(tCtx, nodeInfo, tc.extendedResource, tc.podRequest, draPreScoreState) + if !cmp.Equal(allocatable, tc.expectedAllocatable) { + t.Errorf("Expected allocatable=%v, but got allocatable=%v", tc.expectedAllocatable, allocatable) + } + if !cmp.Equal(requested, tc.expectedRequested) { + t.Errorf("Expected requested=%v, but got requested=%v", tc.expectedRequested, requested) + } }) } } + +// getCachedDeviceMatch checks the cache for a DeviceMatches result +// returns (matches, found) +func (r *resourceAllocationScorer) getCachedDeviceMatch(expression string, driver string, poolName string, deviceName string) (bool, bool) { + key := buildDeviceMatchCacheKey(expression, driver, poolName, deviceName) + + if value, ok := r.deviceMatchCache.Load(key); ok { + return value.(bool), true + } + + return false, false +} + +// setCachedDeviceMatch stores a DeviceMatches result in the cache +func (r *resourceAllocationScorer) setCachedDeviceMatch(expression string, driver string, poolName string, deviceName string, matches bool) { + key := buildDeviceMatchCacheKey(expression, driver, poolName, deviceName) + r.deviceMatchCache.Store(key, matches) +} + +func TestDeviceMatchCaching(t *testing.T) { + // Create a scorer with caching enabled + scorer := &resourceAllocationScorer{ + DRACaches: DRACaches{ + celCache: cel.NewCache(10, cel.Features{}), + }, + } + + expression := `device.attributes["example.com"].test_attr == "test-value"` + driverName := "example.com/test-driver" + poolName := "example-pool" + deviceName := "device-1" + + // Test cache operations + // Initially, cache should be empty + matches, found := scorer.getCachedDeviceMatch(expression, driverName, poolName, deviceName) + if found { + t.Errorf("Cache should be empty initially, but found an entry") + } + if matches { + t.Errorf("Matches should be false initially, but got true") + } + + // Store a result in cache + scorer.setCachedDeviceMatch(expression, driverName, poolName, deviceName, true) + + // Retrieve from cache + matches, found = scorer.getCachedDeviceMatch(expression, driverName, poolName, deviceName) + if !found { + t.Errorf("Result should be found in cache, but was not") + } + if !matches { + t.Errorf("Cached result should match what we stored, expected true but got false") + } + + // Test caching with error + scorer.setCachedDeviceMatch(expression, driverName, poolName, deviceName, false) + + matches, found = scorer.getCachedDeviceMatch(expression, driverName, poolName, deviceName) + if !found { + t.Errorf("Result should be found in cache") + } + if matches { + t.Errorf("Cached result should match what we stored, expected false but got true") + } + + // Test that different devices have different cache keys + matches, found = scorer.getCachedDeviceMatch(expression, driverName, poolName, "device-2") + if found { + t.Errorf("Different device should not hit cache") + } + if matches { + t.Errorf("Matches should be false for uncached entry") + } + + // Test that different pools have different cache keys + matches, found = scorer.getCachedDeviceMatch(expression, driverName, "other-pool", deviceName) + if found { + t.Errorf("Different pool should not hit cache") + } + if matches { + t.Errorf("Matches should be false for uncached entry") + } +} + +func BenchmarkDeviceMatchCaching(b *testing.B) { + expression := `device.attributes["example.com"].test_attr == "test-value"` + driverName := "example.com/test-driver" + poolName := "example-pool" + deviceName := "device-1" + + // Create a scorer with caching enabled + scorer := &resourceAllocationScorer{ + DRACaches: DRACaches{ + celCache: cel.NewCache(10, cel.Features{}), + }, + } + + b.Run("WithoutCache", func(b *testing.B) { + // Disable caching by creating a new scorer each time + for i := 0; i < b.N; i++ { + freshScorer := &resourceAllocationScorer{ + DRACaches: DRACaches{ + celCache: cel.NewCache(10, cel.Features{}), + }, + } + + // This will always be a cache miss + _, _ = freshScorer.getCachedDeviceMatch(expression, driverName, poolName, deviceName) + } + }) + + b.Run("WithCache", func(b *testing.B) { + // Pre-warm the cache + scorer.setCachedDeviceMatch(expression, driverName, poolName, deviceName, true) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + // This should always be a cache hit + _, _ = scorer.getCachedDeviceMatch(expression, driverName, poolName, deviceName) + } + }) +} + +// getCachedNodeMatch checks the cache for a NodeMatches result +func (r *resourceAllocationScorer) getCachedNodeMatch(nodeName string, nodeNameToMatch string, allNodesMatch bool, nodeSelectorHash string) (bool, bool) { + key := buildNodeMatchCacheKey(nodeName, nodeNameToMatch, allNodesMatch, nodeSelectorHash) + + if value, ok := r.nodeMatchCache.Load(key); ok { + return value.(bool), true + } + + return false, false +} + +// setCachedNodeMatch stores a NodeMatches result in the cache +func (r *resourceAllocationScorer) setCachedNodeMatch(nodeName string, nodeNameToMatch string, allNodesMatch bool, nodeSelectorHash string, matches bool) { + key := buildNodeMatchCacheKey(nodeName, nodeNameToMatch, allNodesMatch, nodeSelectorHash) + r.nodeMatchCache.Store(key, matches) +} + +func TestNodeMatchCaching(t *testing.T) { + // Create a scorer with caching enabled + scorer := &resourceAllocationScorer{ + DRACaches: DRACaches{ + celCache: cel.NewCache(10, cel.Features{}), + }, + } + + // Create test node + testNode := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-node-1", + Labels: map[string]string{ + "zone": "us-east-1a", + "instance.type": "gpu-xlarge", + }, + }, + } + + // Test cases for different NodeMatches scenarios + testCases := []struct { + name string + nodeNameToMatch string + allNodesMatch bool + nodeSelector *v1.NodeSelector + expectedMatch bool + }{ + { + name: "exact node name match", + nodeNameToMatch: "test-node-1", + allNodesMatch: false, + nodeSelector: nil, + expectedMatch: true, + }, + { + name: "node name mismatch", + nodeNameToMatch: "different-node", + allNodesMatch: false, + nodeSelector: nil, + expectedMatch: false, + }, + { + name: "all nodes match", + nodeNameToMatch: "", + allNodesMatch: true, + nodeSelector: nil, + expectedMatch: true, + }, + { + name: "node selector match", + nodeNameToMatch: "", + allNodesMatch: false, + nodeSelector: &v1.NodeSelector{ + NodeSelectorTerms: []v1.NodeSelectorTerm{ + { + MatchExpressions: []v1.NodeSelectorRequirement{ + { + Key: "zone", + Operator: v1.NodeSelectorOpIn, + Values: []string{"us-east-1a", "us-east-1b"}, + }, + }, + }, + }, + }, + expectedMatch: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + nodeSelectorStr := tc.nodeSelector.String() + + // First call should be a cache miss + _, found1 := scorer.getCachedNodeMatch(testNode.Name, tc.nodeNameToMatch, tc.allNodesMatch, nodeSelectorStr) + assert.False(t, found1, "Cache should be empty initially") + + // Simulate setting a cached result + scorer.setCachedNodeMatch(testNode.Name, tc.nodeNameToMatch, tc.allNodesMatch, nodeSelectorStr, tc.expectedMatch) + + // Second call should be a cache hit + matches2, found2 := scorer.getCachedNodeMatch(testNode.Name, tc.nodeNameToMatch, tc.allNodesMatch, nodeSelectorStr) + assert.True(t, found2, "Result should be found in cache") + assert.Equal(t, tc.expectedMatch, matches2, "Cached result should match expected value") + }) + } +} + +func BenchmarkNodeMatchCaching(b *testing.B) { + // Create test node + testNode := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-node-1", + Labels: map[string]string{ + "zone": "us-east-1a", + "instance.type": "gpu-xlarge", + }, + }, + } + + // Create a complex NodeSelector to make the test meaningful + nodeSelector := &v1.NodeSelector{ + NodeSelectorTerms: []v1.NodeSelectorTerm{ + { + MatchExpressions: []v1.NodeSelectorRequirement{ + { + Key: "zone", + Operator: v1.NodeSelectorOpIn, + Values: []string{"us-east-1a", "us-east-1b", "us-west-2a"}, + }, + { + Key: "instance.type", + Operator: v1.NodeSelectorOpExists, + }, + }, + }, + }, + } + + b.Run("WithoutCache", func(b *testing.B) { + nodeSelectorStr := nodeSelector.String() + + // Create a new scorer for each iteration to avoid caching + for i := 0; i < b.N; i++ { + freshScorer := &resourceAllocationScorer{ + DRACaches: DRACaches{ + celCache: cel.NewCache(10, cel.Features{}), + }, + } + + // This will always be a cache miss + _, _ = freshScorer.getCachedNodeMatch(testNode.Name, "", false, nodeSelectorStr) + } + }) + + b.Run("WithCache", func(b *testing.B) { + nodeSelectorStr := nodeSelector.String() + + // Create a scorer and pre-warm the cache + scorer := &resourceAllocationScorer{ + DRACaches: DRACaches{ + celCache: cel.NewCache(10, cel.Features{}), + }, + } + + // Pre-warm the cache + scorer.setCachedNodeMatch(testNode.Name, "", false, nodeSelectorStr, true) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + // This should always be a cache hit + _, _ = scorer.getCachedNodeMatch(testNode.Name, "", false, nodeSelectorStr) + } + }) +} diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/allocator.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/allocator.go index 3c804d7b818..c4c8f00bafe 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/structured/allocator.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/allocator.go @@ -140,18 +140,18 @@ func NewAllocator(ctx context.Context, // file name!) into "stable", or individual chunks can be copied over. // // Unit tests are shared between all implementations. - enabledAllocators := []string{} - for _, api := range availableAPIs { + var enabledAllocators []string + for _, allocator := range availableAllocators { // Disabled? - if !allocatorEnabled(api.name) { + if !allocatorEnabled(allocator.name) { continue } - enabledAllocators = append(enabledAllocators, api.name) + enabledAllocators = append(enabledAllocators, allocator.name) // All required features supported? - if api.supportedFeatures.Set().IsSuperset(features.Set()) { + if allocator.supportedFeatures.Set().IsSuperset(features.Set()) { // Use it! - return api.newAllocator(ctx, features, allocatedState, classLister, slices, celCache) + return allocator.newAllocator(ctx, features, allocatedState, classLister, slices, celCache) } } return nil, fmt.Errorf("internal error: no allocator available for feature set %+v, enabled allocators: %s", features, strings.Join(enabledAllocators, ", ")) @@ -172,7 +172,7 @@ func allocatorEnabled(name string) bool { return len(explicitlyEnabledAllocators) == 0 || explicitlyEnabledAllocators.Has(name) } -var availableAPIs = []struct { +var availableAllocators = []struct { name string supportedFeatures Features newAllocator func(ctx context.Context, @@ -237,13 +237,13 @@ var availableAPIs = []struct { // It calls one of the available implementations(stable, incubating, experimental) based // on the provided DRA features. func NodeMatches(features Features, node *v1.Node, nodeNameToMatch string, allNodesMatch bool, nodeSelector *v1.NodeSelector) (bool, error) { - for _, api := range availableAPIs { - if api.supportedFeatures.Set().IsSuperset(features.Set()) { - return api.nodeMatches(node, nodeNameToMatch, allNodesMatch, nodeSelector) + for _, allocator := range availableAllocators { + if allocator.supportedFeatures.Set().IsSuperset(features.Set()) { + return allocator.nodeMatches(node, nodeNameToMatch, allNodesMatch, nodeSelector) } } - return false, fmt.Errorf("internal error: no NodeMatches API available for feature set %v", features) + return false, fmt.Errorf("internal error: no NodeMatches implementation available for feature set %v", features) } // IsDeviceAllocated checks if a device is allocated, considering both fully allocated devices @@ -254,8 +254,8 @@ func IsDeviceAllocated(deviceID DeviceID, allocatedState *AllocatedState) bool { return true } - // Check if device is partially consumed via shared allocations (consumable capacity case) - // We need to check if any shared device ID corresponds to our device + // Check if device is partially consumed via shared allocations (consumable capacity case). + // We need to check if any shared device ID corresponds to our device. for sharedDeviceID := range allocatedState.AllocatedSharedDeviceIDs { // Extract the base device ID from the shared device ID by recreating it baseDeviceID := MakeDeviceID( diff --git a/test/integration/scheduler_perf/dra/extendedresource/scoring/performance-config.yaml b/test/integration/scheduler_perf/dra/extendedresource/scoring/performance-config.yaml index 4308f23d690..d1c9cb8f9a9 100644 --- a/test/integration/scheduler_perf/dra/extendedresource/scoring/performance-config.yaml +++ b/test/integration/scheduler_perf/dra/extendedresource/scoring/performance-config.yaml @@ -1,22 +1,26 @@ -# Copyright 2025 The Kubernetes Authors. +# The following labels are used in this file. (listed in ascending order of the number of covered test cases) # -# 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 +# - integration-test: test cases to run as the integration test, usually to spot some issues in the scheduler implementation or scheduler-perf itself. +# - performance: test cases to run in the performance test. +# - short: supplemental label for the above two labels (must not used alone), which literally means short execution time test cases. # -# http://www.apache.org/licenses/LICENSE-2.0 +# Specifically, the CIs use labels like the following: +# - `ci-kubernetes-integration-master` (`integration-test`): Test cases are chosen based on a tradeoff between code coverage and overall runtime. +# It basically covers all test cases but with their smallest workload. +# - `pull-kubernetes-integration` (`integration-test`,`short`): Test cases are chosen so that they should take less than total 5 min to complete. +# - `ci-benchmark-scheduler-perf` (`performance`): Long enough test cases are chosen (ideally, longer than 10 seconds) +# to provide meaningful samples for the pod scheduling rate. # -# 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. +# Also, `performance`+`short` isn't used in the CIs, but it's used to test the performance test locally. +# (Sometimes, the test cases with `integration-test` are too small to spot issues.) +# +# Combining `performance` and `short` selects suitable workloads for a local +# before/after comparisons with benchstat. -# Configuration specifically for testing DRA extended resource scoring performance. -# This config ensures that multiple nodes can satisfy each pod's extended resource +# This config is specific for testing DRA extended resource scoring performance. +# It ensures that multiple nodes can satisfy each pod's extended resource # requirements, forcing the scheduler to use scoring to choose between nodes. - # ExtendedResourceScoring uses pods with extended resources that can be satisfied # by multiple nodes, triggering scoring competition to test the DRA scoring implementation. - name: ExtendedResourceScoring