diff --git a/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go b/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go index 65d7851e570..d19ab4dd218 100644 --- a/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go +++ b/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go @@ -3284,13 +3284,13 @@ func TestAllocatorSelection(t *testing.T) { // is used. "default": { features: "", - expectImplementation: "incubating", + expectImplementation: "stable", }, // Alpha features need the experimental implementation. "alpha": { features: "AllAlpha=true,AllBeta=true", - expectImplementation: "experimental", + expectImplementation: "incubating", }, } { t.Run(name, func(t *testing.T) { 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 82b9d153d53..ca252a9bbb4 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/structured/allocator.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/allocator.go @@ -227,7 +227,7 @@ var availableAllocators = []struct { slices []*resourceapi.ResourceSlice, celCache *cel.Cache, ) (Allocator, error) { - return incubating.NewAllocator(ctx, features, allocatedState.AllocatedDevices, classLister, slices, celCache) + return incubating.NewAllocator(ctx, features, allocatedState, classLister, slices, celCache) }, nodeMatches: incubating.NodeMatches, }, diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_incubating.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_incubating.go index 2fd2a09c456..6c3648ec99d 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_incubating.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_incubating.go @@ -28,6 +28,8 @@ import ( v1 "k8s.io/api/core/v1" resourceapi "k8s.io/api/resource/v1" + "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" draapi "k8s.io/dynamic-resource-allocation/api" "k8s.io/dynamic-resource-allocation/cel" @@ -46,22 +48,48 @@ func MakeDeviceID(driver, pool, device string) DeviceID { return internal.MakeDeviceID(driver, pool, device) } +// types_experimental +type SharedDeviceID = internal.SharedDeviceID +type DeviceConsumedCapacity = internal.DeviceConsumedCapacity +type ConsumedCapacityCollection = internal.ConsumedCapacityCollection +type ConsumedCapacity = internal.ConsumedCapacity +type AllocatedState = internal.AllocatedState + +func GenerateNewShareID() *types.UID { + return internal.GenerateShareID() +} + +func NewConsumedCapacity() ConsumedCapacity { + return internal.NewConsumedCapacity() +} + +func NewDeviceConsumedCapacity(deviceID DeviceID, + consumedCapacity map[resourceapi.QualifiedName]resource.Quantity) DeviceConsumedCapacity { + return internal.NewDeviceConsumedCapacity(deviceID, consumedCapacity) +} + +func NewConsumedCapacityCollection() ConsumedCapacityCollection { + return internal.NewConsumedCapacityCollection() +} + // SupportedFeatures includes all additional features, // making this the variant that is used when any of those // are enabled. var SupportedFeatures = internal.Features{ - AdminAccess: true, - PrioritizedList: true, - PartitionableDevices: true, - DeviceTaints: true, + AdminAccess: true, + PrioritizedList: true, + PartitionableDevices: true, + DeviceTaints: true, + DeviceBindingAndStatus: true, + ConsumableCapacity: true, } type Allocator struct { - features Features - allocatedDevices sets.Set[DeviceID] - classLister DeviceClassLister - slices []*resourceapi.ResourceSlice - celCache *cel.Cache + features Features + allocatedState AllocatedState + classLister DeviceClassLister + slices []*resourceapi.ResourceSlice + celCache *cel.Cache // availableCounters contains the available counters for each // resource pool. It acts as a cache that is updated the first time // the available counters are needed for each pool. The information @@ -88,14 +116,14 @@ var _ internal.AllocatorExtended = &Allocator{} // The returned Allocator can be used multiple times and is thread-safe. func NewAllocator(ctx context.Context, features Features, - allocatedDevices sets.Set[DeviceID], + allocatedState AllocatedState, classLister DeviceClassLister, slices []*resourceapi.ResourceSlice, celCache *cel.Cache, ) (*Allocator, error) { return &Allocator{ features: features, - allocatedDevices: allocatedDevices, + allocatedState: allocatedState, classLister: classLister, slices: slices, celCache: celCache, @@ -104,7 +132,7 @@ func NewAllocator(ctx context.Context, } func (a *Allocator) Channel() internal.AllocatorChannel { - return internal.Incubating + return internal.Experimental } func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resourceapi.ResourceClaim) (finalResult []resourceapi.AllocationResult, finalErr error) { @@ -119,12 +147,14 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou consumedCounters: make(map[draapi.UniqueString]counterSets), requestData: make(map[requestIndices]requestData), result: make([]internalAllocationResult, len(claims)), + allocatingCapacity: NewConsumedCapacityCollection(), } alloc.logger.V(5).Info("Starting allocation", "numClaims", len(alloc.claimsToAllocate), "numSlices", len(alloc.slices)) defer func() { alloc.logger.V(5).Info("Done with allocation", "success", len(finalResult) == len(alloc.claimsToAllocate), "err", finalErr) }() + alloc.logger.V(5).Info("Gathering pools", "slices", alloc.slices) // First determine all eligible pools. pools, err := GatherPools(ctx, alloc.slices, node, a.features) if err != nil { @@ -174,12 +204,27 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou return nil, fmt.Errorf("claim %s, request %s: has subrequests, but the DRAPrioritizedList feature is disabled", klog.KObj(claim), request.Name) } + // Error out if the consumableCapacity feature is not enabled + // and the request contains capacity requests. + if !a.features.ConsumableCapacity { + if request.Exactly != nil && request.Exactly.Capacity != nil { + return nil, fmt.Errorf("claim %s, request %s: has capacity requests, but the DRAConsumableCapacity feature is disabled", + klog.KObj(claim), request.Name) + } + for _, subReq := range request.FirstAvailable { + // Error out if any subrequest contains capacity requests. + if subReq.Capacity != nil { + return nil, fmt.Errorf("claim %s, subrequest %s: has capacity requests, but the DRAConsumableCapacity feature is disabled", + klog.KObj(claim), subReq.Name) + } + } + } + if hasSubRequests { // We need to find the minimum number of devices that can be allocated // for the request, so setting this to a high number so we can do the // easy comparison in the loop. minDevicesPerRequest := math.MaxInt - // A request with subrequests gets one entry per subrequest in alloc.requestData. // We can only predict a lower number of devices because it depends on which // subrequest gets chosen. @@ -236,6 +281,20 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou attributeName: matchAttribute, } constraints[i] = m + case constraint.DistinctAttribute != nil: + distinctAttribute := resourceapi.FullyQualifiedName(*constraint.DistinctAttribute) + logger := alloc.logger + if loggerV := alloc.logger.V(6); loggerV.Enabled() { + logger = klog.LoggerWithName(logger, "distinctAttributeConstraint") + logger = klog.LoggerWithValues(logger, "distinctAttribute", distinctAttribute) + } + m := &distinctAttributeConstraint{ + logger: logger, + requestNames: sets.New(constraint.Requests...), + attributeName: distinctAttribute, + attributes: make(map[string]resourceapi.DeviceAttribute), + } + constraints[i] = m default: // Unknown constraint type! return nil, fmt.Errorf("claim %s, constraint #%d: empty constraint (unsupported constraint type?)", klog.KObj(claim), i) @@ -255,7 +314,7 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou // We can estimate the size based on what we need to allocate. alloc.allocatingDevices = make(map[DeviceID]sets.Set[int], minDevicesTotal) - alloc.logger.V(5).Info("Gathered information about devices", "allocatedDevices", logAllocatedDevices(alloc.logger, alloc.allocatedDevices), "minDevicesToBeAllocated", minDevicesTotal) + alloc.logger.V(6).Info("Gathered information about devices", "allocatedDevices", logAllocatedDevices(alloc.logger, alloc.allocatedState.AllocatedDevices), "minDevicesToBeAllocated", minDevicesTotal) // In practice, there aren't going to be many different CEL // expressions. Most likely, there is going to be handful of different @@ -270,7 +329,7 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou // All errors get created such that they can be returned by Allocate // without further wrapping. - done, err := alloc.allocateOne(deviceIndices{}, false) + done, err := alloc.allocateOne(deviceIndices{}, false, deviceLocation{}) if errors.Is(err, errStop) { return nil, nil } @@ -298,13 +357,36 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou allocationResult := &result[claimIndex] allocationResult.Devices.Results = make([]resourceapi.DeviceRequestAllocationResult, len(internalResult.devices)) for i, internal := range internalResult.devices { + var consumedCapacity map[resourceapi.QualifiedName]resource.Quantity + if internal.consumedCapacity != nil { + consumedCapacity = make(map[resourceapi.QualifiedName]resource.Quantity, len(internal.consumedCapacity)) + for key, val := range internal.consumedCapacity { + consumedCapacity[key] = val.DeepCopy() + } + } allocationResult.Devices.Results[i] = resourceapi.DeviceRequestAllocationResult{ - Request: internal.requestName(), - Driver: internal.id.Driver.String(), - Pool: internal.id.Pool.String(), - Device: internal.id.Device.String(), - AdminAccess: internal.adminAccess, - Tolerations: internal.lookupRequest(claim).tolerations(), + Request: internal.requestName(), + Driver: internal.id.Driver.String(), + Pool: internal.id.Pool.String(), + Device: internal.id.Device.String(), + AdminAccess: internal.adminAccess, + Tolerations: internal.lookupRequest(claim).tolerations(), + ShareID: internal.shareID, + ConsumedCapacity: consumedCapacity, + } + // Performance optimization: skip the for loop if the feature is off. + // Not needed for correctness because if the feature is off, the selected + // device should not have binding conditions. + if a.features.DeviceBindingAndStatus { + allocationResult.Devices.Results[i].BindingConditions = internal.BindingConditions + allocationResult.Devices.Results[i].BindingFailureConditions = internal.BindingFailureConditions + } + // Performance optimization: skip the for loop if the feature is off. + // Not needed for correctness because if the feature is off, the selected + // device should not have binding conditions. + if a.features.DeviceBindingAndStatus { + allocationResult.Devices.Results[i].BindingConditions = internal.BindingConditions + allocationResult.Devices.Results[i].BindingFailureConditions = internal.BindingFailureConditions } } @@ -388,7 +470,7 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou } // Determine node selector. - nodeSelector, err := alloc.createNodeSelector(internalResult.devices) + nodeSelector, err := alloc.createNodeSelector(internalResult.devices, node.Name) if err != nil { return nil, fmt.Errorf("create NodeSelector for claim %s: %w", claim.Name, err) } @@ -459,7 +541,6 @@ func (alloc *allocator) validateDeviceRequest(request requestAccessor, parentReq if pool.IsInvalid { return requestData, fmt.Errorf("claim %s, request %s: asks for all devices, but resource pool %s is currently invalid", klog.KObj(claim), request.name(), pool.PoolID) } - for _, slice := range pool.DeviceSlicesTargetingNode { for deviceIndex := range slice.Spec.Devices { selectable, err := alloc.isSelectable(requestKey, requestData, slice, deviceIndex) @@ -472,6 +553,20 @@ func (alloc *allocator) validateDeviceRequest(request requestAccessor, parentReq Device: &slice.Spec.Devices[deviceIndex], slice: slice, } + if alloc.features.ConsumableCapacity { + // Next validate whether resource request over capacity + device := slice.Spec.Devices[deviceIndex] + success, err := alloc.CmpRequestOverCapacity(requestData.request, slice, device) + if err != nil { + alloc.logger.V(7).Info("Skip comparing device capacity request", + "device", device, "request", requestData.request.name(), "err", err) + continue + } + if !success { + alloc.logger.V(7).Info("Device capacity not enough", "device", device) + continue + } + } requestData.allDevices = append(requestData.allDevices, device) } } @@ -518,7 +613,12 @@ type allocator struct { // be allocated. // Claims are identified by their index in claimsToAllocate. allocatingDevices map[DeviceID]sets.Set[int] - result []internalAllocationResult + // allocatingCapacity tracks the amount of device capacity that will be newly allocated + // for a particular attempt to find a solution. + // The map is indexed by device ID, and each value represents the accumulated capacity + // requested by all allocations targeting that device. + allocatingCapacity ConsumedCapacityCollection + result []internalAllocationResult } // counterSets is a map with the name of counter sets to the counters in @@ -551,6 +651,13 @@ type deviceIndices struct { deviceIndex int // The index of a device within a request or subrequest. } +// deviceLocation identifies a device by its position in the allocator's pools. +type deviceLocation struct { + poolIndex int + sliceIndex int + deviceIndex int +} + type requestData struct { // The request or subrequest which needs to be allocated. // Never nil. @@ -588,11 +695,13 @@ type internalAllocationResult struct { type internalDeviceResult struct { *draapi.Device - request string // name of the request (if no subrequests) or the subrequest - parentRequest string // name of the request which contains the subrequest, empty otherwise - id DeviceID - slice *draapi.ResourceSlice - adminAccess *bool + request string // name of the request (if no subrequests) or the subrequest + parentRequest string // name of the request which contains the subrequest, empty otherwise + id DeviceID + shareID *types.UID + slice *draapi.ResourceSlice + consumedCapacity map[resourceapi.QualifiedName]resource.Quantity + adminAccess *bool } func (idr internalDeviceResult) requestName() string { @@ -762,7 +871,24 @@ func lookupAttribute(device *draapi.Device, deviceID DeviceID, attributeName res // allocateSubRequest is true when trying to allocate one particular subrequest. // This allows the logic for subrequests to call allocateOne with the same // device index without causing infinite recursion. -func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (bool, error) { +// +// startLocation provides the starting point when searching for the next device to +// allocate. Since we have a list of pools, each with a list of slices which again +// contains a list of devices, we have defined an order in which the devices will +// be attempted for allocation. Since the order of devices doesn't matter within a +// single request, we want to make sure we only attempt all combinations of devices, +// and not search through all permutations (since many of them will have the same set +// of devices and therefore be identical allocations). Note that the order of devices +// does matter across requests and claims. So by providing the startLocation, we +// make sure that the allocator only attempts allocations that follows the order +// of the available devices, thereby avoiding different permutations. For example, +// this means that for the devices [1, 2, 3], we will only attempt the following +// possible allocations [1], [2], [3], [1, 2], [1, 3], [2, 3], and [1, 2, 3]. +// +// The null startLocation (= all indices zero) means that all devices are considered. +// The only situation where a non-null startLocation is used is when looking for the +// next device within the same request. +func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool, startLocation deviceLocation) (bool, error) { alloc.numAllocateOneInvocations.Add(1) if alloc.ctx.Err() != nil { @@ -780,7 +906,7 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b claim := alloc.claimsToAllocate[r.claimIndex] if r.requestIndex >= len(claim.Spec.Devices.Requests) { // Done with the claim, continue with the next one. - success, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex + 1}, false) + success, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex + 1}, false, deviceLocation{}) if errors.Is(err, errAllocationResultMaxSizeExceeded) { // We don't need to propagate this further because // this is not a fatal error. Retrying the claim under @@ -827,7 +953,7 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b } r.subRequestIndex = subRequestIndex - success, err := alloc.allocateOne(r, true /* prevent infinite recusion */) + success, err := alloc.allocateOne(r, true /* prevent infinite recusion */, deviceLocation{}) // If we reached the allocation result limit, we can try // with the next subrequest if there is one. It might request // fewer devices, so it might succeed. @@ -868,7 +994,7 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b // Done with request, continue with next one. We have completed the work for // the request or subrequest, so we can no longer be allocating devices for // a subrequest. - success, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex + 1}, false) + success, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex + 1}, false, deviceLocation{}) // We want to propagate any errAllocationResultMaxSizeExceeded to the caller. If // that error is returned here, it means none of the requests/subrequests after this one // could be allocated while staying within the limit on the number of devices, so there @@ -888,7 +1014,11 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b return false, errAllocationResultMaxSizeExceeded } - alloc.logger.V(6).Info("Allocating one device", "currentClaim", r.claimIndex, "totalClaims", len(alloc.claimsToAllocate), "currentRequest", r.requestIndex, "currentSubRequest", r.subRequestIndex, "totalRequestsPerClaim", len(claim.Spec.Devices.Requests), "currentDevice", r.deviceIndex, "devicesPerRequest", requestData.numDevices, "allDevices", doAllDevices, "adminAccess", request.adminAccess()) + alloc.logger.V(6).Info("Allocating one device", "currentClaim", r.claimIndex, + "totalClaims", len(alloc.claimsToAllocate), "currentRequest", r.requestIndex, + "currentSubRequest", r.subRequestIndex, "totalRequestsPerClaim", len(claim.Spec.Devices.Requests), + "currentDevice", r.deviceIndex, "devicesPerRequest", requestData.numDevices, "allDevices", doAllDevices, "adminAccess", request.adminAccess(), "capacities", request.capacities()) + if doAllDevices { // For "all" devices we already know which ones we need. We // just need to check whether we can use them. @@ -903,7 +1033,8 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b // get all of them, then there is no solution and we have to stop. return false, nil } - done, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex, deviceIndex: r.deviceIndex + 1}, allocateSubRequest) + // No need to propagate the startLocation here since we only try one combination of devices. + done, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex, deviceIndex: r.deviceIndex + 1}, allocateSubRequest, deviceLocation{}) if err != nil || !done { // If we get an error or didn't complete, we need to backtrack. Depending // on the situation we might be able to retry, so we make sure we @@ -915,15 +1046,33 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b } // We need to find suitable devices. - for _, pool := range alloc.pools { + for poolIndex := startLocation.poolIndex; poolIndex < len(alloc.pools); poolIndex++ { + pool := alloc.pools[poolIndex] // We don't allocate devices from invalid or incomplete pools, but // don't error out here since there might be available devices in other // pools. if pool.IsIncomplete || pool.IsInvalid { continue } - for _, slice := range pool.DeviceSlicesTargetingNode { - for deviceIndex := range slice.Spec.Devices { + // We should start at the slice provided by startLocation unless we searched through + // all devices in the previous pool and have started at the next pool. When this happens, + // we should start at the first slice in the pool. + sliceStart := 0 + if poolIndex == startLocation.poolIndex { + sliceStart = startLocation.sliceIndex + } + for sliceIndex := sliceStart; sliceIndex < len(pool.DeviceSlicesTargetingNode); sliceIndex++ { + slice := pool.DeviceSlicesTargetingNode[sliceIndex] + + // We should start at the device provided by startLocation unless we searched through + // all devices in the previous slice and have started at the next slice. When this happens, + // we should start at the first device in the pool. + deviceStart := 0 + if poolIndex == startLocation.poolIndex && + sliceIndex == startLocation.sliceIndex { + deviceStart = startLocation.deviceIndex + } + for deviceIndex := deviceStart; deviceIndex < len(slice.Spec.Devices); deviceIndex++ { deviceID := DeviceID{Driver: pool.Driver, Pool: pool.Pool, Device: slice.Spec.Devices[deviceIndex].Name} // Checking for "in use" is cheap and thus gets done first. @@ -946,6 +1095,20 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b alloc.logger.V(7).Info("Device not selectable", "device", deviceID) continue } + if alloc.features.ConsumableCapacity { + // Next validate whether resource request over capacity + device := slice.Spec.Devices[deviceIndex] + success, err := alloc.CmpRequestOverCapacity(requestData.request, slice, device) + if err != nil { + alloc.logger.V(7).Info("Skip comparing device capacity request", + "device", deviceID, "request", requestData.request.name(), "err", err) + continue + } + if !success { + alloc.logger.V(7).Info("Device capacity not enough", "device", deviceID) + continue + } + } // Finally treat as allocated and move on to the next device. device := deviceWithID{ @@ -969,7 +1132,16 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b subRequestIndex: r.subRequestIndex, deviceIndex: r.deviceIndex + 1, } - done, err := alloc.allocateOne(deviceKey, allocateSubRequest) + nextLocation := deviceLocation{ + poolIndex: poolIndex, + sliceIndex: sliceIndex, + deviceIndex: deviceIndex + 1, + } + // This is the allocation attempt for the next device in the same request. + // If allocateOne finds out that it is done with the request, it moves to + // the next without setting a start location, so each request is free to try + // all devices. + done, err := alloc.allocateOne(deviceKey, allocateSubRequest, nextLocation) // If we found a solution, we can stop. if err == nil && done { return done, nil @@ -1047,6 +1219,20 @@ func (alloc *allocator) isSelectable(r requestIndices, requestData requestData, } +// CmpRequestOverCapacity checks if the given device has sufficient remaining capacity +// to satisfy the resource request. +// Return true if success. +func (alloc *allocator) CmpRequestOverCapacity(request requestAccessor, slice *draapi.ResourceSlice, device draapi.Device) (bool, error) { + deviceID := DeviceID{Driver: slice.Spec.Driver, Pool: slice.Spec.Pool.Name, Device: device.Name} + allocatingCapacity := alloc.allocatingCapacity[deviceID] + allowMultipleAllocations := device.AllowMultipleAllocations + capacities := device.Capacity + if allocatedCapacity, found := alloc.allocatedState.AggregatedCapacity[deviceID]; found { + return CmpRequestOverCapacity(allocatedCapacity, request.capacities(), allowMultipleAllocations, capacities, allocatingCapacity) + } + return CmpRequestOverCapacity(NewConsumedCapacity(), request.capacities(), allowMultipleAllocations, capacities, allocatingCapacity) +} + func (alloc *allocator) selectorsMatch(r requestIndices, device *draapi.Device, deviceID DeviceID, class *resourceapi.DeviceClass, selectors []resourceapi.DeviceSelector) (bool, error) { for i, selector := range selectors { expr := alloc.celCache.GetOrCompile(selector.CEL.Expression) @@ -1068,7 +1254,7 @@ func (alloc *allocator) selectorsMatch(r requestIndices, device *draapi.Device, if err := draapi.Convert_api_Device_To_v1_Device(device, &d, nil); err != nil { return false, fmt.Errorf("convert Device: %w", err) } - matches, details, err := expr.DeviceMatches(alloc.ctx, cel.Device{Driver: deviceID.Driver.String(), Attributes: d.Attributes, Capacity: d.Capacity}) + matches, details, err := expr.DeviceMatches(alloc.ctx, cel.Device{Driver: deviceID.Driver.String(), AllowMultipleAllocations: d.AllowMultipleAllocations, Attributes: d.Attributes, Capacity: d.Capacity}) if class != nil { alloc.logger.V(7).Info("CEL result", "device", deviceID, "class", klog.KObj(class), "selector", i, "expression", selector.CEL.Expression, "matches", matches, "actualCost", ptr.Deref(details.ActualCost(), 0), "err", err) } else { @@ -1094,6 +1280,11 @@ func (alloc *allocator) selectorsMatch(r requestIndices, device *draapi.Device, // allocateDevice checks device availability and constraints for one // candidate. The device must be selectable. // +// r identifies the request for which allocation is attempted. +// The device is the candidate under consideration. +// If "must" is true, then it is okay (but not required) to return an error which describes in more detail why allocation wasn't possible. +// This is used when the caller wants exactly this device and will give up if it cannot allocate it. +// // If that candidate works out okay, the shared state gets updated // as if that candidate had been allocated. If allocation cannot continue later // and must try something else, then the rollback function can be invoked to @@ -1106,7 +1297,11 @@ func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, mus requestKey := requestIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex, subRequestIndex: r.subRequestIndex} requestData := alloc.requestData[requestKey] request := requestData.request - if request.adminAccess() && alloc.allocatingDeviceForClaim(device.id, r.claimIndex) { + allowMultipleAllocations := false + if alloc.features.ConsumableCapacity { + allowMultipleAllocations = device.AllowMultipleAllocations != nil && *device.AllowMultipleAllocations + } + if !allowMultipleAllocations && request.adminAccess() && alloc.allocatingDeviceForClaim(device.id, r.claimIndex) { alloc.logger.V(7).Info("Device in use in same claim", "device", device.id) return false, nil, nil } @@ -1122,8 +1317,11 @@ func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, mus return false, nil, nil } + // Skip counter availability check for devices that allow multiple allocation and some capacity has already in-use. + skipCounterCheck := allowMultipleAllocations && alloc.deviceCapacityInUse(device.id) + // The API validation logic has checked the ConsumesCounters referred should exist inside SharedCounters. - if len(device.ConsumesCounters) > 0 { + if !skipCounterCheck && len(device.ConsumesCounters) > 0 { // If a device consumes counters from a counter set, verify that // there is sufficient counters available. ok, err := alloc.checkAvailableCounters(device) @@ -1171,14 +1369,41 @@ func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, mus } } - // All constraints satisfied. Mark as in use (unless we do admin access) + // All constraints satisfied. Mark as in use (unless we do admin access or allow multiple allocations) // and record the result. alloc.logger.V(7).Info("Device allocated", "device", device.id) if alloc.allocatingDevices[device.id] == nil { alloc.allocatingDevices[device.id] = make(sets.Set[int]) } - alloc.allocatingDevices[device.id].Insert(r.claimIndex) + if !allowMultipleAllocations { + alloc.allocatingDevices[device.id].Insert(r.claimIndex) + } + + consumedCapacity := make(map[resourceapi.QualifiedName]resource.Quantity, 0) + var shareID *types.UID + if alloc.features.ConsumableCapacity { + // Validate whether resource request over capacity + success, err := alloc.CmpRequestOverCapacity(requestData.request, device.slice, *device.Device) + // The error should not occur at this point as it should be detected in the previous step. + if err != nil { + alloc.logger.V(7).Info("Failed to compare device capacity request on allocateDevice", + "device", device, "request", requestData.request.name(), "err", err) + return false, nil, nil + } + if !success { + alloc.logger.V(7).Info("Device capacity not enough", "device", device) + return false, nil, nil + } + + if allowMultipleAllocations { + consumedCapacity = GetConsumedCapacityFromRequest(request.capacities(), device.Capacity) + shareID = GenerateNewShareID() + alloc.logger.V(7).Info("Device capacity allocated", "device", device.id, + "consumed capacity", klog.Format(consumedCapacity)) + alloc.allocatingCapacity.Insert(NewDeviceConsumedCapacity(device.id, consumedCapacity)) + } + } result := internalDeviceResult{ request: request.name(), @@ -1186,10 +1411,14 @@ func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, mus id: device.id, Device: device.Device, slice: device.slice, + shareID: shareID, } if request.adminAccess() { result.adminAccess = ptr.To(request.adminAccess()) } + if len(consumedCapacity) > 0 { + result.consumedCapacity = consumedCapacity + } previousNumResults := len(alloc.result[r.claimIndex].devices) alloc.result[r.claimIndex].devices = append(alloc.result[r.claimIndex].devices, result) @@ -1198,8 +1427,16 @@ func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, mus constraint.remove(baseRequestName, subRequestName, device.Device, device.id) } alloc.allocatingDevices[device.id].Delete(r.claimIndex) - if alloc.features.PartitionableDevices && len(device.ConsumesCounters) > 0 { - alloc.deallocateCountersForDevice(device) + if allowMultipleAllocations { + requestedResource := alloc.result[r.claimIndex].devices[previousNumResults].consumedCapacity + if requestedResource != nil { + alloc.allocatingCapacity.Remove(NewDeviceConsumedCapacity(device.id, requestedResource)) + } + } else { + alloc.allocatingDevices[device.id].Delete(r.claimIndex) + if alloc.features.PartitionableDevices && len(device.ConsumesCounters) > 0 { + alloc.deallocateCountersForDevice(device) + } } // Truncate, but keep the underlying slice. alloc.result[r.claimIndex].devices = alloc.result[r.claimIndex].devices[:previousNumResults] @@ -1271,7 +1508,7 @@ func (alloc *allocator) checkAvailableCounters(device deviceWithID) (bool, error } // Devices that aren't allocated doesn't consume any counters, so we don't // need to consider them. - if !alloc.allocatedDevices.Has(deviceID) { + if !alloc.allocatedState.AllocatedDevices.Has(deviceID) { continue } for _, deviceCounterConsumption := range device.ConsumesCounters { @@ -1346,7 +1583,12 @@ func (alloc *allocator) checkAvailableCounters(device deviceWithID) (bool, error } func (alloc *allocator) deviceInUse(deviceID DeviceID) bool { - return alloc.allocatedDevices.Has(deviceID) || alloc.allocatingDeviceForAnyClaim(deviceID) + return alloc.allocatedState.AllocatedDevices.Has(deviceID) || alloc.allocatingDeviceForAnyClaim(deviceID) +} + +func (alloc *allocator) deviceCapacityInUse(deviceID DeviceID) bool { + _, found := alloc.allocatedState.AggregatedCapacity[deviceID] + return found || alloc.allocatingCapacityForAnyClaim(deviceID) } func (alloc *allocator) allocatingDeviceForAnyClaim(deviceID DeviceID) bool { @@ -1357,6 +1599,11 @@ func (alloc *allocator) allocatingDeviceForClaim(deviceID DeviceID, claimIndex i return alloc.allocatingDevices[deviceID].Has(claimIndex) } +func (alloc *allocator) allocatingCapacityForAnyClaim(deviceID DeviceID) bool { + _, found := alloc.allocatingCapacity[deviceID] + return found +} + // deallocateCountersForDevice subtracts the consumed counters of the provided // device from the consumedCounters data structure. func (alloc *allocator) deallocateCountersForDevice(device deviceWithID) { @@ -1376,7 +1623,7 @@ func (alloc *allocator) deallocateCountersForDevice(device deviceWithID) { // createNodeSelector constructs a node selector for the allocation, if needed, // otherwise it returns nil. -func (alloc *allocator) createNodeSelector(result []internalDeviceResult) (*v1.NodeSelector, error) { +func (alloc *allocator) createNodeSelector(result []internalDeviceResult, targetNodeName string) (*v1.NodeSelector, error) { // Selector with one term. That term gets extended with additional // requirements from the different devices. ns := &v1.NodeSelector{ @@ -1394,15 +1641,15 @@ func (alloc *allocator) createNodeSelector(result []internalDeviceResult) (*v1.N nodeName = slice.Spec.NodeName nodeSelector = slice.Spec.NodeSelector } - if nodeName != nil { - // At least one device is local to one node. This - // restricts the allocation to that node. + if nodeName != nil || result[i].BindsToNode { + // At least one device is local to one node or binds to a node, + // so we need to restrict the allocation to that node. return &v1.NodeSelector{ NodeSelectorTerms: []v1.NodeSelectorTerm{{ MatchFields: []v1.NodeSelectorRequirement{{ Key: "metadata.name", Operator: v1.NodeSelectorOpIn, - Values: []string{*nodeName}, + Values: []string{targetNodeName}, }}, }}, }, nil @@ -1444,6 +1691,7 @@ type requestAccessor interface { hasAdminAccess() bool selectors() []resourceapi.DeviceSelector tolerations() []resourceapi.DeviceToleration + capacities() *resourceapi.CapacityRequirements } // exactDeviceRequestAccessor is an implementation of the @@ -1484,6 +1732,10 @@ func (d *exactDeviceRequestAccessor) tolerations() []resourceapi.DeviceToleratio return d.request.Exactly.Tolerations } +func (d *exactDeviceRequestAccessor) capacities() *resourceapi.CapacityRequirements { + return d.request.Exactly.Capacity +} + // deviceSubRequestAccessor is an implementation of the // requestAccessor interface for DeviceSubRequests. type deviceSubRequestAccessor struct { @@ -1522,6 +1774,10 @@ func (d *deviceSubRequestAccessor) tolerations() []resourceapi.DeviceToleration return d.subRequest.Tolerations } +func (d *deviceSubRequestAccessor) capacities() *resourceapi.CapacityRequirements { + return d.subRequest.Capacity +} + func addNewNodeSelectorRequirements(from []v1.NodeSelectorRequirement, to *[]v1.NodeSelectorRequirement) { for _, requirement := range from { if !containsNodeSelectorRequirement(*to, requirement) { diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_test.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_test.go index 28a7884a4f1..7ba5bedb18b 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_test.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_test.go @@ -32,12 +32,12 @@ func TestAllocator(t *testing.T) { func( ctx context.Context, features Features, - allocatedState internal.AllocatedState, + allocatedState AllocatedState, classLister DeviceClassLister, slices []*resourceapi.ResourceSlice, celCache *cel.Cache, ) (internal.Allocator, error) { - return NewAllocator(ctx, features, allocatedState.AllocatedDevices, classLister, slices, celCache) + return NewAllocator(ctx, features, allocatedState, classLister, slices, celCache) }, ) } diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/constraint.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/constraint.go new file mode 100644 index 00000000000..7324d6a50ae --- /dev/null +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/constraint.go @@ -0,0 +1,131 @@ +/* +Copyright 2024 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 incubating + +import ( + "fmt" + + resourceapi "k8s.io/api/resource/v1" + "k8s.io/apimachinery/pkg/util/sets" + draapi "k8s.io/dynamic-resource-allocation/api" + "k8s.io/klog/v2" +) + +// distinctAttributeConstraint compares an attribute value across devices. +// All devices must share the same value. When the set of devices is +// empty, any device that has the attribute can be added. After that, +// only matching devices can be added. +// +// We don't need to track *which* devices are part of the set, only +// how many. +type distinctAttributeConstraint struct { + logger klog.Logger // Includes name and attribute name, so no need to repeat in log messages. + requestNames sets.Set[string] + attributeName resourceapi.FullyQualifiedName + + attributes map[string]resourceapi.DeviceAttribute + numDevices int +} + +func (m *distinctAttributeConstraint) add(requestName, subRequestName string, device *draapi.Device, deviceID DeviceID) bool { + if m.requestNames.Len() > 0 && !m.matches(requestName, subRequestName) { + // Device not affected by constraint. + return true + } + + attribute := lookupAttribute(device, deviceID, m.attributeName) + if attribute == nil { + // Doesn't have the attribute. + m.logger.V(7).Info("Constraint not satisfied, attribute not set") + return false + } + + if m.numDevices == 0 { + // The first device can always get picked. + m.attributes[requestName] = *attribute + m.numDevices = 1 + m.logger.V(7).Info("First attribute added") + return true + } + + if !m.matchesAttribute(*attribute) { + m.logger.V(7).Info("Constraint not satisfied, has some duplicated attributes") + return false + } + m.attributes[requestName] = *attribute + m.numDevices++ + m.logger.V(7).Info("Constraint satisfied by device", "device", deviceID, "numDevices", m.numDevices) + return true + +} + +func (m *distinctAttributeConstraint) remove(requestName, subRequestName string, device *draapi.Device, deviceID DeviceID) { + if m.requestNames.Len() > 0 && !m.matches(requestName, subRequestName) { + // Device not affected by constraint. + return + } + delete(m.attributes, requestName) + m.numDevices-- + m.logger.V(7).Info("Device removed from constraint set", "device", deviceID, "numDevices", m.numDevices) +} + +func (m *distinctAttributeConstraint) matches(requestName, subRequestName string) bool { + if subRequestName == "" { + return m.requestNames.Has(requestName) + } else { + fullSubRequestName := fmt.Sprintf("%s/%s", requestName, subRequestName) + return m.requestNames.Has(requestName) || m.requestNames.Has(fullSubRequestName) + } +} + +func (m *distinctAttributeConstraint) matchesAttribute(attribute resourceapi.DeviceAttribute) bool { + for _, attr := range m.attributes { + switch { + case attribute.StringValue != nil: + if attr.StringValue != nil && *attribute.StringValue == *attr.StringValue { + m.logger.V(7).Info("String values duplicated") + return false + } + case attribute.IntValue != nil: + if attr.IntValue != nil && *attribute.IntValue == *attr.IntValue { + m.logger.V(7).Info("Int values duplicated") + return false + } + case attribute.BoolValue != nil: + if attr.BoolValue != nil && *attribute.BoolValue == *attr.BoolValue { + m.logger.V(7).Info("Bool values duplicated") + return false + } + case attribute.VersionValue != nil: + // semver 2.0.0 requires that version strings are in their + // minimal form (in particular, no leading zeros). Therefore a + // strict "exact equal" check can do a string comparison. + if attr.VersionValue != nil && *attribute.VersionValue == *attr.VersionValue { + m.logger.V(7).Info("Version values duplicated") + return false + } + default: + // Unknown value type, cannot match. + // This condition should not be reached + // as the unknown value type should be failed on CEL compile (getAttributeValue). + m.logger.V(7).Info("Distinct attribute type unknown") + return false + } + } + // All distinct + return true +} diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity.go new file mode 100644 index 00000000000..2bb3239f587 --- /dev/null +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity.go @@ -0,0 +1,211 @@ +/* +Copyright 2024 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 incubating + +import ( + "errors" + + resourceapi "k8s.io/api/resource/v1" + "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/utils/ptr" +) + +// CmpRequestOverCapacity checks whether the new capacity request can be added within the given capacity, +// and checks whether the requested value is against the capacity requestPolicy. +func CmpRequestOverCapacity(currentConsumedCapacity ConsumedCapacity, deviceRequestCapacity *resourceapi.CapacityRequirements, + allowMultipleAllocations *bool, capacity map[resourceapi.QualifiedName]resourceapi.DeviceCapacity, allocatingCapacity ConsumedCapacity) (bool, error) { + if requestsContainNonExistCapacity(deviceRequestCapacity, capacity) { + return false, errors.New("some requested capacity has not been defined") + } + clone := currentConsumedCapacity.Clone() + for name, cap := range capacity { + var requestedValPtr *resource.Quantity + if deviceRequestCapacity != nil && deviceRequestCapacity.Requests != nil { + if requestedVal, requestedFound := deviceRequestCapacity.Requests[name]; requestedFound { + requestedValPtr = &requestedVal + } + } + consumedCapacity := calculateConsumedCapacity(requestedValPtr, cap) + if violatesPolicy(consumedCapacity, cap.RequestPolicy) { + return false, nil + } + // If the current clone already contains an entry for this capacity, add the consumedCapacity to it. + // Otherwise, initialize it with calculated consumedCapacity. + if _, allocatedFound := clone[name]; allocatedFound { + clone[name].Add(consumedCapacity) + } else { + clone[name] = ptr.To(consumedCapacity) + } + // If allocatingCapacity contains an entry for this capacity, add its value to clone as well. + if allocatingVal, allocatingFound := allocatingCapacity[name]; allocatingFound { + clone[name].Add(*allocatingVal) + } + if clone[name].Cmp(cap.Value) > 0 { + return false, nil + } + } + return true, nil +} + +// requestsNonExistCapacity returns true if requests contain non-exist capacity. +func requestsContainNonExistCapacity(deviceRequestCapacity *resourceapi.CapacityRequirements, + capacity map[resourceapi.QualifiedName]resourceapi.DeviceCapacity) bool { + if deviceRequestCapacity == nil || deviceRequestCapacity.Requests == nil { + return false + } + for name := range deviceRequestCapacity.Requests { + if _, found := capacity[name]; !found { + return true + } + } + return false +} + +// calculateConsumedCapacity returns valid capacity to be consumed regarding the requested capacity and device capacity policy. +// +// If no requestPolicy, return capacity.Value. +// If no requestVal, fill the quantity by fillEmptyRequest function +// Otherwise, use requestPolicy to calculate the consumed capacity from request if applicable. +func calculateConsumedCapacity(requestedVal *resource.Quantity, capacity resourceapi.DeviceCapacity) resource.Quantity { + if requestedVal == nil { + return fillEmptyRequest(capacity) + } + if capacity.RequestPolicy == nil { + return requestedVal.DeepCopy() + } + switch { + case capacity.RequestPolicy.ValidRange != nil && capacity.RequestPolicy.ValidRange.Min != nil: + return roundUpRange(requestedVal, capacity.RequestPolicy.ValidRange) + case capacity.RequestPolicy.ValidValues != nil: + return roundUpValidValues(requestedVal, capacity.RequestPolicy.ValidValues) + } + return *requestedVal +} + +// fillEmptyRequest +// return requestPolicy.default if defined. +// Otherwise, return capacity value. +func fillEmptyRequest(capacity resourceapi.DeviceCapacity) resource.Quantity { + if capacity.RequestPolicy != nil && capacity.RequestPolicy.Default != nil { + return capacity.RequestPolicy.Default.DeepCopy() + } + return capacity.Value.DeepCopy() +} + +// roundUpRange rounds the requestedVal up to fit within the specified validRange. +// - If requestedVal is less than Min, it returns Min. +// - If Step is specified, it rounds requestedVal up to the nearest multiple of Step +// starting from Min. +// - If no Step is specified and requestedVal >= Min, it returns requestedVal as is. +func roundUpRange(requestedVal *resource.Quantity, validRange *resourceapi.CapacityRequestPolicyRange) resource.Quantity { + if requestedVal.Cmp(*validRange.Min) < 0 { + return validRange.Min.DeepCopy() + } + if validRange.Step == nil { + return *requestedVal + } + requestedInt64 := requestedVal.Value() + step := validRange.Step.Value() + min := validRange.Min.Value() + added := (requestedInt64 - min) + n := added / step + mod := added % step + if mod != 0 { + n += 1 + } + val := min + step*n + return *resource.NewQuantity(val, resource.BinarySI) +} + +// roundUpValidValues returns the first value in validValues that is greater than or equal to requestedVal. +// If no such value exists, it returns requestedVal itself. +func roundUpValidValues(requestedVal *resource.Quantity, validValues []resource.Quantity) resource.Quantity { + // Simple sequential search is used as the maximum entry of validValues is finite and small (≤10), + // and the list must already be sorted in ascending order, ensured by API validation. + // Note: A binary search could alternatively be used for better efficiency if the list grows larger. + for _, validValue := range validValues { + if requestedVal.Cmp(validValue) <= 0 { + return validValue.DeepCopy() + } + } + return *requestedVal +} + +// GetConsumedCapacityFromRequest returns valid consumed capacity, +// according to claim request and defined capacity. +func GetConsumedCapacityFromRequest(requestedCapacity *resourceapi.CapacityRequirements, + consumableCapacity map[resourceapi.QualifiedName]resourceapi.DeviceCapacity) map[resourceapi.QualifiedName]resource.Quantity { + consumedCapacity := make(map[resourceapi.QualifiedName]resource.Quantity) + for name, cap := range consumableCapacity { + var requestedValPtr *resource.Quantity + if requestedCapacity != nil && requestedCapacity.Requests != nil { + if requestedVal, requestedFound := requestedCapacity.Requests[name]; requestedFound { + requestedValPtr = &requestedVal + } + } + capacity := calculateConsumedCapacity(requestedValPtr, cap) + consumedCapacity[name] = capacity + } + return consumedCapacity +} + +// violatesPolicy checks whether the request violate the requestPolicy. +func violatesPolicy(requestedVal resource.Quantity, policy *resourceapi.CapacityRequestPolicy) bool { + if policy == nil { + // no policy to check + return false + } + if policy.Default != nil && requestedVal == *policy.Default { + return false + } + switch { + case policy.ValidRange != nil: + return violateValidRange(requestedVal, *policy.ValidRange) + case len(policy.ValidValues) > 0: + return violateValidValues(requestedVal, policy.ValidValues) + } + // no policy violated through to completion. + return false +} + +func violateValidRange(requestedVal resource.Quantity, validRange resourceapi.CapacityRequestPolicyRange) bool { + if validRange.Max != nil && + requestedVal.Cmp(*validRange.Max) > 0 { + return true + } + if validRange.Step != nil { + requestedInt64 := requestedVal.Value() + step := validRange.Step.Value() + min := validRange.Min.Value() + added := (requestedInt64 - min) + mod := added % step + // must be a multiply of step + if mod != 0 { + return true + } + } + return false +} + +func violateValidValues(requestedVal resource.Quantity, validValues []resource.Quantity) bool { + for _, validVal := range validValues { + if requestedVal.Cmp(validVal) == 0 { + return false + } + } + return true +} diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity_test.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity_test.go new file mode 100644 index 00000000000..df618737649 --- /dev/null +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity_test.go @@ -0,0 +1,243 @@ +/* +Copyright 2024 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 incubating + +import ( + "testing" + + . "github.com/onsi/gomega" + resourceapi "k8s.io/api/resource/v1" + "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/utils/ptr" +) + +const ( + driverA = "driver-a" + pool1 = "pool-1" + device1 = "device-1" + capacity0 = "capacity-0" + capacity1 = "capacity-1" +) + +var ( + one = resource.MustParse("1") + two = resource.MustParse("2") + three = resource.MustParse("3") +) + +func deviceConsumedCapacity(deviceID DeviceID) DeviceConsumedCapacity { + capaicty := map[resourceapi.QualifiedName]resource.Quantity{ + capacity0: one, + } + return NewDeviceConsumedCapacity(deviceID, capaicty) +} + +func TestConsumableCapacity(t *testing.T) { + + t.Run("add-sub-allocating-consumed-capacity", func(t *testing.T) { + g := NewWithT(t) + allocatedCapacity := NewConsumedCapacity() + g.Expect(allocatedCapacity.Empty()).To(BeTrueBecause("allocated capacity should start from zero")) + oneAllocated := ConsumedCapacity{ + capacity0: &one, + } + allocatedCapacity.Add(oneAllocated) + g.Expect(allocatedCapacity.Empty()).To(BeFalseBecause("capacity is added")) + allocatedCapacity.Sub(oneAllocated) + g.Expect(allocatedCapacity.Empty()).To(BeTrueBecause("capacity is subtracted to zero")) + }) + + t.Run("insert-remove-allocating-consumed-capacity-collection", func(t *testing.T) { + g := NewWithT(t) + deviceID := MakeDeviceID(driverA, pool1, device1) + aggregatedCapacity := NewConsumedCapacityCollection() + aggregatedCapacity.Insert(deviceConsumedCapacity(deviceID)) + aggregatedCapacity.Insert(deviceConsumedCapacity(deviceID)) + allocatedCapacity, found := aggregatedCapacity[deviceID] + g.Expect(found).To(BeTrueBecause("expected deviceID to be found")) + g.Expect(allocatedCapacity[capacity0].Cmp(two)).To(BeZero()) + aggregatedCapacity.Remove(deviceConsumedCapacity(deviceID)) + g.Expect(allocatedCapacity[capacity0].Cmp(one)).To(BeZero()) + }) + + t.Run("get-consumed-capacity-from-request", func(t *testing.T) { + requestedCapacity := &resourceapi.CapacityRequirements{ + Requests: map[resourceapi.QualifiedName]resource.Quantity{ + capacity0: one, + "dummy": one, + }, + } + consumableCapacity := map[resourceapi.QualifiedName]resourceapi.DeviceCapacity{ + capacity0: { // with request and with default, expect requested value + Value: two, + RequestPolicy: &resourceapi.CapacityRequestPolicy{ + Default: ptr.To(two), + ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one)}, + }, + }, + capacity1: { // no request but with default, expect default + Value: two, + RequestPolicy: &resourceapi.CapacityRequestPolicy{ + Default: ptr.To(one), + ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one)}, + }, + }, + "dummy": { + Value: one, // no request and no policy (no default), expect capacity value + }, + } + consumedCapacity := GetConsumedCapacityFromRequest(requestedCapacity, consumableCapacity) + g := NewWithT(t) + g.Expect(consumedCapacity).To(HaveLen(3)) + for name, val := range consumedCapacity { + g.Expect(string(name)).Should(BeElementOf([]string{capacity0, capacity1, "dummy"})) + g.Expect(val.Cmp(one)).To(BeZero()) + } + }) + + t.Run("violate-capacity-sharing", testViolateCapacityRequestPolicy) + + t.Run("calculate-consumed-capacity", testCalculateConsumedCapacity) + +} + +func testViolateCapacityRequestPolicy(t *testing.T) { + testcases := map[string]struct { + requestedVal resource.Quantity + requestPolicy *resourceapi.CapacityRequestPolicy + + expectResult bool + }{ + "no constraint": {one, nil, false}, + "less than maximum": { + one, + &resourceapi.CapacityRequestPolicy{ + Default: ptr.To(one), + ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one), Max: &two}, + }, + false, + }, + "more than maximum": { + two, + &resourceapi.CapacityRequestPolicy{ + Default: ptr.To(one), + ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one), Max: &one}, + }, + true, + }, + "in set": { + one, + &resourceapi.CapacityRequestPolicy{ + Default: ptr.To(one), + ValidValues: []resource.Quantity{one}, + }, + false, + }, + "not in set": { + two, + &resourceapi.CapacityRequestPolicy{ + Default: ptr.To(one), + ValidValues: []resource.Quantity{one}, + }, + true, + }, + } + for name, tc := range testcases { + t.Run(name, func(t *testing.T) { + g := NewWithT(t) + violate := violatesPolicy(tc.requestedVal, tc.requestPolicy) + g.Expect(violate).To(BeEquivalentTo(tc.expectResult)) + }) + } +} + +func testCalculateConsumedCapacity(t *testing.T) { + testcases := map[string]struct { + requestedVal *resource.Quantity + capacityValue resource.Quantity + requestPolicy *resourceapi.CapacityRequestPolicy + + expectResult resource.Quantity + }{ + "empty": {nil, one, &resourceapi.CapacityRequestPolicy{}, one}, + "min in range": { + nil, + two, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one)}}, + one, + }, + "default in set": { + nil, + two, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidValues: []resource.Quantity{one}}, + one, + }, + "more than min in range": { + &two, + two, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one)}}, + two, + }, + "less than min in range": { + &one, + two, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(two)}}, + two, + }, + "with step (round up)": { + &two, + three, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one), Step: ptr.To(two.DeepCopy())}}, + three, + }, + "with step (no remaining)": { + &two, + two, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidRange: &resourceapi.CapacityRequestPolicyRange{Min: ptr.To(one), Step: ptr.To(one.DeepCopy())}}, + two, + }, + "valid value in set": { + &two, + three, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidValues: []resource.Quantity{one, two, three}}, + two, + }, + "set (round up)": { + &two, + three, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidValues: []resource.Quantity{one, three}}, + three, + }, + "larger than set": { + &three, + three, + &resourceapi.CapacityRequestPolicy{Default: ptr.To(one), ValidValues: []resource.Quantity{one, two}}, + three, + }, + } + for name, tc := range testcases { + t.Run(name, func(t *testing.T) { + g := NewWithT(t) + capacity := resourceapi.DeviceCapacity{ + Value: tc.capacityValue, + RequestPolicy: tc.requestPolicy, + } + consumedCapacity := calculateConsumedCapacity(tc.requestedVal, capacity) + g.Expect(consumedCapacity.Cmp(tc.expectResult)).To(BeZero()) + }) + } +} diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/pools_incubating.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/pools_incubating.go index ac9b05df03f..e7befc87093 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/pools_incubating.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/pools_incubating.go @@ -57,6 +57,7 @@ func NodeMatches(node *v1.Node, nodeNameToMatch string, allNodesMatch bool, node // Both is recorded in the result. func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node *v1.Node, features Features) ([]*Pool, error) { pools := make(map[PoolID][]*draapi.ResourceSlice) + var slicesWithBindingConditions []*resourceapi.ResourceSlice for _, slice := range slices { if !features.PartitionableDevices && (slice.Spec.PerDeviceNodeSelection != nil || len(slice.Spec.SharedCounters) > 0) { @@ -75,6 +76,14 @@ func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node return nil, fmt.Errorf("failed to perform node selection for slice %s: %w", slice.Name, err) } if match { + if hasBindingConditions(slice) { + // If there is a Device in the ResourceSlice that contains BindingConditions, + // the ResourceSlice should be sorted to be after the ResourceSlice without BindingConditions + // because then the allocation is going to prefer the simpler devices without + // binding conditions. + slicesWithBindingConditions = append(slicesWithBindingConditions, slice) + continue + } if err := addSlice(pools, slice); err != nil { return nil, fmt.Errorf("failed to add node slice %s: %w", slice.Name, err) } @@ -87,6 +96,12 @@ func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node device.String(), slice.Name, err) } if match { + if hasBindingConditions(slice) { + // If there is a Device in the ResourceSlice that contains BindingConditions, + // the ResourceSlice should be sorted to be after the ResourceSlice without BindingConditions. + slicesWithBindingConditions = append(slicesWithBindingConditions, slice) + break + } if err := addSlice(pools, slice); err != nil { return nil, fmt.Errorf("failed to add node slice %s: %w", slice.Name, err) } @@ -106,6 +121,12 @@ func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node } + for _, slice := range slicesWithBindingConditions { + if err := addSlice(pools, slice); err != nil { + return nil, fmt.Errorf("failed to add node slice %s: %w", slice.Name, err) + } + } + // Find incomplete pools and flatten into a single slice. // // When we get here, we only have slices relevant for the node. @@ -114,6 +135,7 @@ func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node // if they are not relevant for the node, so we have to be // careful with the "is incomplete" check. result := make([]*Pool, 0, len(pools)) + var resultWithBindingConditions []*Pool for poolID, slicesForPool := range pools { // If we have all slices, we are done. isComplete := int64(len(slicesForPool)) == slicesForPool[0].Spec.Pool.ResourceSliceCount @@ -122,6 +144,10 @@ func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node if err != nil { return nil, err } + if poolHasBindingConditions(*pool) { + resultWithBindingConditions = append(resultWithBindingConditions, pool) + continue + } result = append(result, pool) continue } @@ -159,8 +185,16 @@ func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node if err != nil { return nil, err } + // if pool has binding conditions, add the pool to the end of the result + if poolHasBindingConditions(*pool) { + resultWithBindingConditions = append(resultWithBindingConditions, pool) + continue + } result = append(result, pool) } + if len(resultWithBindingConditions) != 0 { + result = append(result, resultWithBindingConditions...) + } return result, nil } @@ -344,6 +378,26 @@ func validateDeviceCounterConsumption(counterSets map[draapi.UniqueString]*draap return nil } +func hasBindingConditions(slice *resourceapi.ResourceSlice) bool { + for _, device := range slice.Spec.Devices { + if device.BindingConditions != nil { + return true + } + } + return false +} + +func poolHasBindingConditions(pool Pool) bool { + for _, slice := range pool.DeviceSlicesTargetingNode { + for _, device := range slice.Spec.Devices { + if device.BindingConditions != nil { + return true + } + } + } + return false +} + // checkSlicesInPool is an expensive check of all slices in the pool. // The generation is what the caller wants to move ahead with. // diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/allocator_stable.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/allocator_stable.go index a5dc50a9b30..b3fcfd94925 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/allocator_stable.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/allocator_stable.go @@ -24,6 +24,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" v1 "k8s.io/api/core/v1" resourceapi "k8s.io/api/resource/v1" @@ -45,14 +46,15 @@ func MakeDeviceID(driver, pool, device string) DeviceID { return internal.MakeDeviceID(driver, pool, device) } -// SupportedFeatures does not include any additional features. -// The "stable" implementation can only be used if all of those -// are off. -// -// At least that's its purpose conceptually: in practice, all -// code supporting the additional features is still there. -// It could get removed. -var SupportedFeatures = internal.Features{} +// SupportedFeatures includes all additional features, +// making this the variant that is used when any of those +// are enabled. +var SupportedFeatures = internal.Features{ + AdminAccess: true, + PrioritizedList: true, + PartitionableDevices: true, + DeviceTaints: true, +} type Allocator struct { features Features @@ -60,19 +62,26 @@ type Allocator struct { classLister DeviceClassLister slices []*resourceapi.ResourceSlice celCache *cel.Cache - // availableCounters contains the available counters for individual - // ResourceSlices. It acts as a cache that is updated the first time - // the available counters are needed for each ResourceSlice. The information - // about each slice is never updated once set the first time. + // availableCounters contains the available counters for each + // resource pool. It acts as a cache that is updated the first time + // the available counters are needed for each pool. The information + // about each pool is never updated once set the first time. // This is computed bsed on information on the Allocator, so it will // be correct even for multiple usages of the Allocator. - // The keys in the map are ResourceSlice names. + // The keys in the map are resource pool names. // The allocator might be accessed by different goroutines, so // access to this map must be synchronized. - availableCounters map[string]counterSets + availableCounters map[draapi.UniqueString]counterSets mutex sync.RWMutex + // numAllocateOneInvocations counts the number of times the allocateOne + // function is called for the allocator. This is a measurement of the + // amount of work the allocator had to do to allocate devices + // for the claims. + numAllocateOneInvocations atomic.Int64 } +var _ internal.AllocatorExtended = &Allocator{} + // NewAllocator returns an allocator for a certain set of claims or an error if // some problem was detected which makes it impossible to allocate claims. // @@ -90,12 +99,12 @@ func NewAllocator(ctx context.Context, classLister: classLister, slices: slices, celCache: celCache, - availableCounters: make(map[string]counterSets), + availableCounters: make(map[draapi.UniqueString]counterSets), }, nil } func (a *Allocator) Channel() internal.AllocatorChannel { - return internal.Stable + return internal.Incubating } func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resourceapi.ResourceClaim) (finalResult []resourceapi.AllocationResult, finalErr error) { @@ -107,7 +116,7 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou claimsToAllocate: claims, deviceMatchesRequest: make(map[matchKey]bool), constraints: make([][]constraint, len(claims)), - consumedCounters: make(map[string]counterSets), + consumedCounters: make(map[draapi.UniqueString]counterSets), requestData: make(map[requestIndices]requestData), result: make([]internalAllocationResult, len(claims)), } @@ -122,11 +131,7 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou return nil, fmt.Errorf("gather pool information: %w", err) } alloc.pools = pools - if loggerV := alloc.logger.V(7); loggerV.Enabled() { - loggerV.Info("Gathered pool information", "numPools", len(pools), "pools", pools) - } else { - alloc.logger.V(5).Info("Gathered pool information", "numPools", len(pools)) - } + alloc.logger.V(5).Info("Gathered pool information", "pools", logPools(alloc.logger, pools)) // We allocate one claim after the other and for each claim, all of // its requests. For each individual device we pick one possible @@ -250,7 +255,7 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou // We can estimate the size based on what we need to allocate. alloc.allocatingDevices = make(map[DeviceID]sets.Set[int], minDevicesTotal) - alloc.logger.V(6).Info("Gathered information about devices", "numAllocated", len(alloc.allocatedDevices), "minDevicesToBeAllocated", minDevicesTotal) + alloc.logger.V(5).Info("Gathered information about devices", "allocatedDevices", logAllocatedDevices(alloc.logger, alloc.allocatedDevices), "minDevicesToBeAllocated", minDevicesTotal) // In practice, there aren't going to be many different CEL // expressions. Most likely, there is going to be handful of different @@ -299,6 +304,7 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou Pool: internal.id.Pool.String(), Device: internal.id.Device.String(), AdminAccess: internal.adminAccess, + Tolerations: internal.lookupRequest(claim).tolerations(), } } @@ -392,9 +398,11 @@ func (a *Allocator) Allocate(ctx context.Context, node *v1.Node, claims []*resou return result, nil } -func (a *Allocator) GetStats() *Stats { - // Not implemented. - return nil +func (a *Allocator) GetStats() Stats { + s := Stats{ + NumAllocateOneInvocations: a.numAllocateOneInvocations.Load(), + } + return s } func (alloc *allocator) validateDeviceRequest(request requestAccessor, parentRequest requestAccessor, requestKey requestIndices, pools []*Pool) (requestData, error) { @@ -452,7 +460,7 @@ func (alloc *allocator) validateDeviceRequest(request requestAccessor, parentReq return requestData, fmt.Errorf("claim %s, request %s: asks for all devices, but resource pool %s is currently invalid", klog.KObj(claim), request.name(), pool.PoolID) } - for _, slice := range pool.Slices { + for _, slice := range pool.DeviceSlicesTargetingNode { for deviceIndex := range slice.Spec.Devices { selectable, err := alloc.isSelectable(requestKey, requestData, slice, deviceIndex) if err != nil { @@ -481,6 +489,13 @@ func (alloc *allocator) validateDeviceRequest(request requestAccessor, parentReq // that allocation cannot succeed. var errStop = errors.New("stop allocation") +// errAllocationResultMaxSizeExceeded is a special error that gets return by +// allocatedOne when the number of allocated devices exceeds the max number +// allowed. This is checked by earlier invocations in the recursion and used +// to do more aggressive backtracking and avoid attempting allocations that +// we know can not succeed. +var errAllocationResultMaxSizeExceeded = errors.New("allocation max size exceeded") + // allocator is used while an [Allocator.Allocate] is running. Only a single // goroutine works with it, so there is no need for locking. type allocator struct { @@ -494,8 +509,8 @@ type allocator struct { constraints [][]constraint // one list of constraints per claim // consumedCounters keeps track of the counters consumed by all devices // that are in the process of being allocated. - // The keys in the map are ResourceSlice names. - consumedCounters map[string]counterSets + // The keys in the map are resource pool names. + consumedCounters map[draapi.UniqueString]counterSets requestData map[requestIndices]requestData // one entry per request with no subrequests and one entry per subrequest // allocatingDevices tracks which devices will be newly allocated for a // particular attempt to find a solution. The map is indexed by device @@ -564,6 +579,7 @@ type deviceWithID struct { *draapi.Device id DeviceID slice *draapi.ResourceSlice + pool *Pool } type internalAllocationResult struct { @@ -579,11 +595,36 @@ type internalDeviceResult struct { adminAccess *bool } -func (i internalDeviceResult) requestName() string { - if i.parentRequest == "" { - return i.request +func (idr internalDeviceResult) requestName() string { + if idr.parentRequest == "" { + return idr.request } - return fmt.Sprintf("%s/%s", i.parentRequest, i.request) + return fmt.Sprintf("%s/%s", idr.parentRequest, idr.request) +} + +func (idr internalDeviceResult) lookupRequest(claim *resourceapi.ResourceClaim) requestAccessor { + requestName := idr.request + if idr.parentRequest != "" { + requestName = idr.parentRequest + } + for i := range claim.Spec.Devices.Requests { + request := &claim.Spec.Devices.Requests[i] + if request.Name != requestName { + continue + } + if idr.parentRequest == "" { + // No need to check sub-requests. + return &exactDeviceRequestAccessor{request} + } + for j := range request.FirstAvailable { + subRequest := &request.FirstAvailable[j] + if subRequest.Name != idr.request { + continue + } + return &deviceSubRequestAccessor{subRequest} + } + } + return nil } type constraint interface { @@ -722,6 +763,8 @@ func lookupAttribute(device *draapi.Device, deviceID DeviceID, attributeName res // This allows the logic for subrequests to call allocateOne with the same // device index without causing infinite recursion. func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (bool, error) { + alloc.numAllocateOneInvocations.Add(1) + if alloc.ctx.Err() != nil { return false, fmt.Errorf("filter operation aborted: %w", context.Cause(alloc.ctx)) } @@ -737,7 +780,17 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b claim := alloc.claimsToAllocate[r.claimIndex] if r.requestIndex >= len(claim.Spec.Devices.Requests) { // Done with the claim, continue with the next one. - return alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex + 1}, false) + success, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex + 1}, false) + if errors.Is(err, errAllocationResultMaxSizeExceeded) { + // We don't need to propagate this further because + // this is not a fatal error. Retrying the claim under + // different circumstances may succeed if it uses + // subrequests and changing the allocation of some + // prior claim enables allocating a subrequest here + // which needs fewer devices. + return false, nil + } + return success, err } // r.subRequestIndex is zero unless the for loop below is in the @@ -750,16 +803,40 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b // hitting the first subrequest, but not if we are already working on a // specific subrequest. if !allocateSubRequest && requestData.parentRequest != nil { + // Keep track of whether all attempts to do allocation with the + // subrequests results in the allocation result limit exceeded. + // If so, there is no need to make attempts with other devices + // in the previous request (if any), except when + // it is a firstAvailable request where some sub-requests + // need less devices than others. + allAllocationExceeded := true for subRequestIndex := 0; ; subRequestIndex++ { nextSubRequestKey := requestKey nextSubRequestKey.subRequestIndex = subRequestIndex if _, ok := alloc.requestData[nextSubRequestKey]; !ok { // Past the end of the subrequests without finding a solution -> give up. + // + // Return errAllocationResultMaxSizeExceeded if all + // attempts for the subrequests failed to due to reaching + // the max size limit. This would mean that there are no + // solution that involves the previous request (if any). + if allAllocationExceeded { + return false, errAllocationResultMaxSizeExceeded + } return false, nil } r.subRequestIndex = subRequestIndex success, err := alloc.allocateOne(r, true /* prevent infinite recusion */) + // If we reached the allocation result limit, we can try + // with the next subrequest if there is one. It might request + // fewer devices, so it might succeed. + if errors.Is(err, errAllocationResultMaxSizeExceeded) { + continue + } + // If we get here, at least one of the subrequests failed for a + // different reason than errAllocationResultMaxSizeExceeded. + allAllocationExceeded = false if err != nil { return false, err } @@ -791,17 +868,24 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b // Done with request, continue with next one. We have completed the work for // the request or subrequest, so we can no longer be allocating devices for // a subrequest. - return alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex + 1}, false) + success, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex + 1}, false) + // We want to propagate any errAllocationResultMaxSizeExceeded to the caller. If + // that error is returned here, it means none of the requests/subrequests after this one + // could be allocated while staying within the limit on the number of devices, so there + // are no solution in the current request/subrequest that would work. + return success, err } + // Before trying to allocate devices, check if allocating the devices + // in the current request will put us over the threshold. // We can calculate this by adding the number of already allocated devices with the number // of devices in the current request, and then finally subtract the deviceIndex since we // don't want to double count any devices already allocated for the current request. numDevicesAfterAlloc := len(alloc.result[r.claimIndex].devices) + requestData.numDevices - r.deviceIndex if numDevicesAfterAlloc > resourceapi.AllocationResultsMaxSize { - // Don't return an error here since we want to keep searching for - // a solution that works. - return false, nil + // Return a special error so we can identify this situation in the + // callers and do more aggressive backtracking. + return false, errAllocationResultMaxSizeExceeded } alloc.logger.V(6).Info("Allocating one device", "currentClaim", r.claimIndex, "totalClaims", len(alloc.claimsToAllocate), "currentRequest", r.requestIndex, "currentSubRequest", r.subRequestIndex, "totalRequestsPerClaim", len(claim.Spec.Devices.Requests), "currentDevice", r.deviceIndex, "devicesPerRequest", requestData.numDevices, "allDevices", doAllDevices, "adminAccess", request.adminAccess()) @@ -820,13 +904,12 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b return false, nil } done, err := alloc.allocateOne(deviceIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex, deviceIndex: r.deviceIndex + 1}, allocateSubRequest) - if err != nil { - return false, err - } - if !done { - // Backtrack. + if err != nil || !done { + // If we get an error or didn't complete, we need to backtrack. Depending + // on the situation we might be able to retry, so we make sure we + // deallocate. deallocate() - return false, nil + return false, err } return done, nil } @@ -839,7 +922,7 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b if pool.IsIncomplete || pool.IsInvalid { continue } - for _, slice := range pool.Slices { + for _, slice := range pool.DeviceSlicesTargetingNode { for deviceIndex := range slice.Spec.Devices { deviceID := DeviceID{Driver: pool.Driver, Pool: pool.Pool, Device: slice.Spec.Devices[deviceIndex].Name} @@ -869,6 +952,7 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b id: deviceID, Device: &slice.Spec.Devices[deviceIndex], slice: slice, + pool: pool, } allocated, deallocate, err := alloc.allocateDevice(r, device, false) if err != nil { @@ -886,17 +970,22 @@ func (alloc *allocator) allocateOne(r deviceIndices, allocateSubRequest bool) (b deviceIndex: r.deviceIndex + 1, } done, err := alloc.allocateOne(deviceKey, allocateSubRequest) - if err != nil { - return false, err - } - - // If we found a solution, then we can stop. - if done { + // If we found a solution, we can stop. + if err == nil && done { return done, nil } - // Otherwise try some other device after rolling back. + // Otherwise we didn't find a solution, and we need to deallocate + // so the temporary allocation is correct for trying other devices. deallocate() + + if err != nil { + // If we hit an error, we return. This might be that we reached + // the allocation size limit, and if so, it will be caught further + // up the stack and other subrequests will be attempted if there + // are any. + return false, err + } } } } @@ -1009,6 +1098,9 @@ func (alloc *allocator) selectorsMatch(r requestIndices, device *draapi.Device, // as if that candidate had been allocated. If allocation cannot continue later // and must try something else, then the rollback function can be invoked to // restore the previous state. +// +// The rollback function is only provided in case of a successful allocation +// (true and no error). func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, must bool) (bool, func(), error) { claim := alloc.claimsToAllocate[r.claimIndex] requestKey := requestIndices{claimIndex: r.claimIndex, requestIndex: r.requestIndex, subRequestIndex: r.subRequestIndex} @@ -1057,7 +1149,7 @@ func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, mus // Might be tainted, in which case the taint has to be tolerated. // The check is skipped if the feature is disabled. - if alloc.features.DeviceTaints && !allTaintsTolerated(device.Device, request) { + if alloc.features.DeviceTaints && taintPreventsAllocation(device.Device, request) { return false, nil, nil } @@ -1115,13 +1207,17 @@ func (alloc *allocator) allocateDevice(r deviceIndices, device deviceWithID, mus }, nil } -func allTaintsTolerated(device *draapi.Device, request requestAccessor) bool { +func taintPreventsAllocation(device *draapi.Device, request requestAccessor) bool { for _, taint := range device.Taints { - if !taintTolerated(taint, request) { - return false + switch taint.Effect { + // Only known effects prevent allocation, others (including None) are ignored. + case resourceapi.DeviceTaintEffectNoExecute, resourceapi.DeviceTaintEffectNoSchedule: + if !taintTolerated(taint, request) { + return true + } } } - return true + return false } func taintTolerated(taint resourceapi.DeviceTaint, request requestAccessor) bool { @@ -1139,13 +1235,13 @@ func taintTolerated(taint resourceapi.DeviceTaint, request requestAccessor) bool // Gets called only if the partitionable devices feature is enabled and the device // consumes counters. func (alloc *allocator) checkAvailableCounters(device deviceWithID) (bool, error) { - slice := device.slice - sliceName := slice.Name + pool := device.pool + poolName := pool.PoolID.Pool - // Check first if the available counters for this slice have already been + // Check first if the available counters for this pool have already been // calculated. alloc.mutex.RLock() - availableCountersForSlice, found := alloc.availableCounters[sliceName] + availableCountersForPool, found := alloc.availableCounters[poolName] alloc.mutex.RUnlock() // If not, we need to do it now. But we store the result so it doesn't need // to be calculated again. @@ -1153,69 +1249,73 @@ func (alloc *allocator) checkAvailableCounters(device deviceWithID) (bool, error // might also do this work. But the input will be the same to all of them, so // the result will also always be the same. if !found { - availableCountersForSlice = make(counterSets, len(slice.Spec.SharedCounters)) - for _, counterSet := range slice.Spec.SharedCounters { + availableCountersForPool = make(counterSets, len(pool.CounterSets)) + for _, counterSet := range pool.CounterSets { availableCountersForCounterSet := make(map[string]resourceapi.Counter, len(counterSet.Counters)) for name, c := range counterSet.Counters { availableCountersForCounterSet[name] = c } - availableCountersForSlice[counterSet.Name] = availableCountersForCounterSet + availableCountersForPool[counterSet.Name] = availableCountersForCounterSet } // Update the data structure to reflect counters already consumed by allocated devices. This // only includes devices where the allocation process has completed, so this will never // change during the allocation process. - for _, device := range slice.Spec.Devices { - deviceID := DeviceID{ - Driver: slice.Spec.Driver, - Pool: slice.Spec.Pool.Name, - Device: device.Name, - } - // Devices that aren't allocated doesn't consume any counters, so we don't - // need to consider them. - if !alloc.allocatedDevices.Has(deviceID) { - continue - } - for _, deviceCounterConsumption := range device.ConsumesCounters { - availableCountersForCounterSet := availableCountersForSlice[deviceCounterConsumption.CounterSet] - for name, c := range deviceCounterConsumption.Counters { - existingCounter, ok := availableCountersForCounterSet[name] - if !ok { - // the API validation logic has been added to make sure the counters referred should exist in counter sets. + for _, resourceSlices := range [][]*draapi.ResourceSlice{pool.DeviceSlicesTargetingNode, pool.DeviceSlicesNotTargetingNode} { + for _, slice := range resourceSlices { + for _, device := range slice.Spec.Devices { + deviceID := DeviceID{ + Driver: slice.Spec.Driver, + Pool: slice.Spec.Pool.Name, + Device: device.Name, + } + // Devices that aren't allocated doesn't consume any counters, so we don't + // need to consider them. + if !alloc.allocatedDevices.Has(deviceID) { continue } - // This can potentially result in negative available counters. That is fine, - // we just treat it as no counters available. - existingCounter.Value.Sub(c.Value) - availableCountersForCounterSet[name] = existingCounter + for _, deviceCounterConsumption := range device.ConsumesCounters { + availableCountersForCounterSet := availableCountersForPool[deviceCounterConsumption.CounterSet] + for name, c := range deviceCounterConsumption.Counters { + existingCounter, ok := availableCountersForCounterSet[name] + if !ok { + // the API validation logic has been added to make sure the counters referred should exist in counter sets. + continue + } + // This can potentially result in negative available counters. That is fine, + // we just treat it as no counters available. + existingCounter.Value.Sub(c.Value) + availableCountersForCounterSet[name] = existingCounter + } + } + // Note that we don't include devices in the alloc.allocatingDevices here since + // counters consumed by devices for the current claims are tracked in + // alloc.consumedCounters } } - // Note that we don't include devices in the alloc.allocatingDevices here since - // counters consumed by devices for the current claims are tracked in - // alloc.consumedCounters } // Set the available counters on the allocator so we don't have to // compute this again. alloc.mutex.Lock() - alloc.availableCounters[sliceName] = availableCountersForSlice + alloc.availableCounters[poolName] = availableCountersForPool alloc.mutex.Unlock() } // Update the consumedCounters data structure with the counters consumed // by the current device. - consumedCountersForSlice, found := alloc.consumedCounters[sliceName] + consumedCountersForPool, found := alloc.consumedCounters[poolName] // If no devices in the allocating state have consumed any counters from the current - // slice, initialize the data structure. + // pool, initialize the data structure. if !found { - consumedCountersForSlice = make(counterSets) - alloc.consumedCounters[sliceName] = consumedCountersForSlice + consumedCountersForPool = make(counterSets) + alloc.consumedCounters[poolName] = consumedCountersForPool } for _, deviceCounterConsumption := range device.ConsumesCounters { - consumedCountersForCounterSet, found := consumedCountersForSlice[deviceCounterConsumption.CounterSet] + consumedCountersForCounterSet, found := consumedCountersForPool[deviceCounterConsumption.CounterSet] if !found { consumedCountersForCounterSet = make(map[string]resourceapi.Counter) - consumedCountersForSlice[deviceCounterConsumption.CounterSet] = consumedCountersForCounterSet + consumedCountersForPool[deviceCounterConsumption.CounterSet] = consumedCountersForCounterSet } for name, c := range deviceCounterConsumption.Counters { consumedCounters, found := consumedCountersForCounterSet[name] @@ -1231,8 +1331,8 @@ func (alloc *allocator) checkAvailableCounters(device deviceWithID) (bool, error // Check that we didn't exceed the availability of any counters by allocating // the current device. If we did, the current set of devices doesn't work, so we // update the consumed counters to no longer reflect the current device. - for availableCounterSetName, availableCounters := range availableCountersForSlice { - consumedCounters := consumedCountersForSlice[availableCounterSetName] + for availableCounterSetName, availableCounters := range availableCountersForPool { + consumedCounters := consumedCountersForPool[availableCounterSetName] for availableCounterName, availableCounter := range availableCounters { consumedCounter := consumedCounters[availableCounterName] if availableCounter.Value.Cmp(consumedCounter.Value) < 0 { @@ -1260,13 +1360,12 @@ func (alloc *allocator) allocatingDeviceForClaim(deviceID DeviceID, claimIndex i // deallocateCountersForDevice subtracts the consumed counters of the provided // device from the consumedCounters data structure. func (alloc *allocator) deallocateCountersForDevice(device deviceWithID) { - slice := device.slice - sliceName := slice.Name + poolName := device.pool.PoolID.Pool - consumedCountersForSlice := alloc.consumedCounters[sliceName] + consumedCountersForPool := alloc.consumedCounters[poolName] for _, deviceCounterConsumption := range device.ConsumesCounters { counterSetName := deviceCounterConsumption.CounterSet - consumedCounterSet := consumedCountersForSlice[counterSetName] + consumedCounterSet := consumedCountersForPool[counterSetName] for name, c := range deviceCounterConsumption.Counters { consumedCounter := consumedCounterSet[name] consumedCounter.Value.Sub(c.Value) diff --git a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/pools_stable.go b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/pools_stable.go index 951c7234219..df79f92b6ce 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/pools_stable.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/pools_stable.go @@ -19,12 +19,16 @@ package stable import ( "context" "fmt" + "slices" + + "github.com/go-logr/logr" v1 "k8s.io/api/core/v1" resourceapi "k8s.io/api/resource/v1" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/component-helpers/scheduling/corev1/nodeaffinity" draapi "k8s.io/dynamic-resource-allocation/api" + "k8s.io/klog/v2" "k8s.io/utils/ptr" ) @@ -52,14 +56,20 @@ func NodeMatches(node *v1.Node, nodeNameToMatch string, allNodesMatch bool, node // required slices available) or invalid (for example, device names not unique). // Both is recorded in the result. func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node *v1.Node, features Features) ([]*Pool, error) { - pools := make(map[PoolID]*Pool) + pools := make(map[PoolID][]*draapi.ResourceSlice) for _, slice := range slices { - if !features.PartitionableDevices && slice.Spec.PerDeviceNodeSelection != nil { + if !features.PartitionableDevices && (slice.Spec.PerDeviceNodeSelection != nil || len(slice.Spec.SharedCounters) > 0) { continue } - if nodeName, allNodes := ptr.Deref(slice.Spec.NodeName, ""), ptr.Deref(slice.Spec.AllNodes, false); nodeName != "" || allNodes || slice.Spec.NodeSelector != nil { + // Always include slices with SharedCounters since they are needed to use a pool + // regardless of their node selector. + if len(slice.Spec.SharedCounters) > 0 { + if err := addSlice(pools, slice); err != nil { + return nil, fmt.Errorf("failed to add node slice %s: %w", slice.Name, err) + } + } else if nodeName, allNodes := ptr.Deref(slice.Spec.NodeName, ""), ptr.Deref(slice.Spec.AllNodes, false); nodeName != "" || allNodes || slice.Spec.NodeSelector != nil { match, err := NodeMatches(node, nodeName, allNodes, slice.Spec.NodeSelector) if err != nil { return nil, fmt.Errorf("failed to perform node selection for slice %s: %w", slice.Name, err) @@ -104,81 +114,234 @@ func GatherPools(ctx context.Context, slices []*resourceapi.ResourceSlice, node // if they are not relevant for the node, so we have to be // careful with the "is incomplete" check. result := make([]*Pool, 0, len(pools)) - for poolID, pool := range pools { - // If we have all slices, we are done. If not, then we need to start looking for slices + for poolID, slicesForPool := range pools { + // If we have all slices, we are done. + isComplete := int64(len(slicesForPool)) == slicesForPool[0].Spec.Pool.ResourceSliceCount + if isComplete { + pool, err := buildPool(poolID, slicesForPool, features, nil) + if err != nil { + return nil, err + } + result = append(result, pool) + continue + } + // If not, then we need to start looking for slices // which were filtered out above because their node selection made them look irrelevant // for the current node. This is necessary for "allocate all" mode (it rejects incomplete // pools). - isComplete := int64(len(pool.Slices)) == pool.Slices[0].Spec.Pool.ResourceSliceCount - if !isComplete { - isObsolete, numSlices := checkSlicesInPool(slices, poolID, pool.Slices[0].Spec.Pool.Generation) - if isObsolete { - // A more thorough check determined that the DRA driver is in the process - // of replacing the current generation. The newer one didn't have any slice - // which devices for the node, or we would have noticed sooner. - // - // Let's ignore the old device information by ignoring the pool. - continue - } - // Use the more complete number of slices to check for "incomplete pool". + isObsolete, allSlicesForPool := checkSlicesInPool(slices, poolID, slicesForPool[0].Spec.Pool.Generation) + if isObsolete { + // A more thorough check determined that the DRA driver is in the process + // of replacing the current generation. The newer one didn't have any slice + // which devices for the node, or we would have noticed sooner. // - // The slices that we return to the caller still don't represent the whole - // pool, but that's okay: we *want* to limit the result to relevant devices - // so the caller doesn't need to check node selectors unnecessarily. - isComplete = numSlices == pool.Slices[0].Spec.Pool.ResourceSliceCount + // Let's ignore the old device information by ignoring the pool. + continue + } + // Use the more complete number of slices to check for "incomplete pool". + // + // The slices that we return to the caller still don't represent the whole + // pool, but that's okay: we *want* to limit the result to relevant devices + // so the caller doesn't need to check node selectors unnecessarily. + isComplete = int64(len(allSlicesForPool)) == slicesForPool[0].Spec.Pool.ResourceSliceCount + // If a pool is incomplete, we don't allow allocation so we don't need + // to do any validation. We need to keep track of the incomplete pools here + // to make sure allocationMode: All doesn't succeed without considering all + // devices. + if !isComplete { + result = append(result, &Pool{ + PoolID: poolID, + IsIncomplete: true, + }) + continue + } + pool, err := buildPool(poolID, slicesForPool, features, allSlicesForPool) + if err != nil { + return nil, err } - pool.IsIncomplete = !isComplete - pool.IsInvalid, pool.InvalidReason = poolIsInvalid(pool) result = append(result, pool) } return result, nil } -func addSlice(pools map[PoolID]*Pool, s *resourceapi.ResourceSlice) error { +func addSlice(pools map[PoolID][]*draapi.ResourceSlice, s *resourceapi.ResourceSlice) error { var slice draapi.ResourceSlice if err := draapi.Convert_v1_ResourceSlice_To_api_ResourceSlice(s, &slice, nil); err != nil { return fmt.Errorf("convert ResourceSlice: %w", err) } id := PoolID{Driver: slice.Spec.Driver, Pool: slice.Spec.Pool.Name} - pool := pools[id] - if pool == nil { + slicesForPool := pools[id] + if slicesForPool == nil { // New pool. - pool = &Pool{ - PoolID: id, - Slices: []*draapi.ResourceSlice{&slice}, - } - pools[id] = pool + pools[id] = []*draapi.ResourceSlice{&slice} return nil } - if slice.Spec.Pool.Generation < pool.Slices[0].Spec.Pool.Generation { + if slice.Spec.Pool.Generation < slicesForPool[0].Spec.Pool.Generation { // Out-dated. return nil } - if slice.Spec.Pool.Generation > pool.Slices[0].Spec.Pool.Generation { + if slice.Spec.Pool.Generation > slicesForPool[0].Spec.Pool.Generation { // Newer, replaces all old slices. - pool.Slices = nil + pools[id] = []*draapi.ResourceSlice{&slice} + return nil } // Add to pool. - pool.Slices = append(pool.Slices, &slice) + slicesForPool = append(slicesForPool, &slice) + pools[id] = slicesForPool return nil } -func poolIsInvalid(pool *Pool) (bool, string) { - devices := sets.New[draapi.UniqueString]() - for _, slice := range pool.Slices { - for _, device := range slice.Spec.Devices { - if devices.Has(device.Name) { - return true, fmt.Sprintf("duplicate device name %s", device.Name) +func buildPool(id PoolID, slices []*draapi.ResourceSlice, features Features, allSlicesForPool []*resourceapi.ResourceSlice) (*Pool, error) { + var deviceSlices []*draapi.ResourceSlice + var counterSetSlices []*draapi.ResourceSlice + if features.PartitionableDevices { + for _, slice := range slices { + if len(slice.Spec.SharedCounters) > 0 { + counterSetSlices = append(counterSetSlices, slice) + } else { + deviceSlices = append(deviceSlices, slice) } - devices.Insert(device.Name) + } + } else { + deviceSlices = slices + } + if err := validateDeviceNames(deviceSlices); err != nil { + return &Pool{ + PoolID: id, + IsInvalid: true, + InvalidReason: err.Error(), + }, nil + } + // If the partitionable devices feature is not enabled, we don't need to + // validate counter sets and consumed counters, so we are done. + if !features.PartitionableDevices { + return &Pool{ + PoolID: id, + DeviceSlicesTargetingNode: deviceSlices, + }, nil + } + + counterSets, err := getAndValidateCounterSets(counterSetSlices) + if err != nil { + return &Pool{ + PoolID: id, + IsInvalid: true, + InvalidReason: err.Error(), + }, nil + } + + if err := validateDeviceCounterConsumption(counterSets, slices); err != nil { + return &Pool{ + PoolID: id, + IsInvalid: true, + InvalidReason: err.Error(), + }, nil + } + // If we have already seen all slices (both with counter sets and devices), + // we don't need to do any more validation. + if allSlicesForPool == nil || len(slices) == len(allSlicesForPool) { + return &Pool{ + PoolID: id, + DeviceSlicesTargetingNode: deviceSlices, + CounterSets: counterSets, + }, nil + } + + // If we have slices that were discarded earlier because they didn't target the current node + // we need to check them now. They might include devices that consume counters in the pool and + // the allocator needs to know about them to correctly determine available counters. + // + // We only want to convert the slices we haven't already converted, so make it easy to + // look up the names of converted slices. + slicesTargetingNodeNames := sets.New[string]() + for _, slice := range slices { + slicesTargetingNodeNames.Insert(slice.Name) + } + var slicesNotTargetingNode []*draapi.ResourceSlice + for _, slice := range allSlicesForPool { + if slicesTargetingNodeNames.Has(slice.Name) { + continue + } + var convertedSlice draapi.ResourceSlice + if err := draapi.Convert_v1_ResourceSlice_To_api_ResourceSlice(slice, &convertedSlice, nil); err != nil { + return nil, fmt.Errorf("convert ResourceSlice: %w", err) + } + slicesNotTargetingNode = append(slicesNotTargetingNode, &convertedSlice) + } + // We need to make sure the devices here are correctly consuming counters and counter + // sets. Otherwise the allocator might make incorrect decisions. + // We don't validate the device names here. It might be that we should do that, but + // this is consistent with existing behavior where we don't validate slices that + // we don't allocate from. + if err := validateDeviceCounterConsumption(counterSets, slicesNotTargetingNode); err != nil { + return &Pool{ + PoolID: id, + IsInvalid: true, + InvalidReason: err.Error(), + }, nil + } + return &Pool{ + PoolID: id, + DeviceSlicesTargetingNode: deviceSlices, + DeviceSlicesNotTargetingNode: slicesNotTargetingNode, + CounterSets: counterSets, + }, nil +} + +func getAndValidateCounterSets(slices []*draapi.ResourceSlice) (map[draapi.UniqueString]*draapi.CounterSet, error) { + counterSets := make(map[draapi.UniqueString]*draapi.CounterSet) + // We only capture the first error we encounter. + for _, slice := range slices { + for i := range slice.Spec.SharedCounters { + counterSet := slice.Spec.SharedCounters[i] + if _, found := counterSets[counterSet.Name]; found { + return nil, fmt.Errorf("duplicate counter set name %s", counterSet.Name) + } + counterSets[counterSet.Name] = &counterSet } } - return false, "" + return counterSets, nil +} + +func validateDeviceNames(resourceSlices []*draapi.ResourceSlice) error { + deviceNames := sets.New[draapi.UniqueString]() + for _, slice := range resourceSlices { + for _, device := range slice.Spec.Devices { + // Make sure we don't have duplicate device names + if deviceNames.Has(device.Name) { + return fmt.Errorf("duplicate device name %s", device.Name) + } + deviceNames.Insert(device.Name) + } + } + return nil +} + +func validateDeviceCounterConsumption(counterSets map[draapi.UniqueString]*draapi.CounterSet, resourceSlices []*draapi.ResourceSlice) error { + for _, slice := range resourceSlices { + for _, device := range slice.Spec.Devices { + // Make sure all consumed counters for the device references counter sets that exists and + // that they consume counters that exists within those counter sets. + for _, deviceCounterConsumption := range device.ConsumesCounters { + counterSet, found := counterSets[deviceCounterConsumption.CounterSet] + if !found { + return fmt.Errorf("counter set %s not found", deviceCounterConsumption.CounterSet) + } + + for counterName := range deviceCounterConsumption.Counters { + if _, found := counterSet.Counters[counterName]; !found { + return fmt.Errorf("counter %s not found in counter set %s", counterName, counterSet.Name) + } + } + } + } + } + return nil } // checkSlicesInPool is an expensive check of all slices in the pool. @@ -186,13 +349,15 @@ func poolIsInvalid(pool *Pool) (bool, string) { // // It returns: // - current generation is obsolete -> no further checking -// - total number of slices with the generation +// - all slices with the generation in the pool // // Future TODO: detect inconsistent ResourceSliceCount, also in poolIsInvalid. -func checkSlicesInPool(slices []*resourceapi.ResourceSlice, poolID PoolID, generation int64) (isObsolete bool, numSlices int64) { +func checkSlicesInPool(slices []*resourceapi.ResourceSlice, poolID PoolID, generation int64) (bool, []*resourceapi.ResourceSlice) { // A cached index by pool ID would make this more efficient. // It may be needed long-term to support features which always have to consider all slices. - for _, slice := range slices { + var allSlicesForPool []*resourceapi.ResourceSlice + for i := range slices { + slice := slices[i] if slice.Spec.Driver != poolID.Driver.String() || slice.Spec.Pool.Name != poolID.Pool.String() { // Different pool. @@ -200,24 +365,26 @@ func checkSlicesInPool(slices []*resourceapi.ResourceSlice, poolID PoolID, gener } switch { case slice.Spec.Pool.Generation == generation: - numSlices++ + allSlicesForPool = append(allSlicesForPool, slice) case slice.Spec.Pool.Generation > generation: // The caller must have missed some other slice in the pool. // Abort! - return true, 0 + return true, nil default: // Older generation, ignore. } } - return false, numSlices + return false, allSlicesForPool } type Pool struct { PoolID - IsIncomplete bool - IsInvalid bool - InvalidReason string - Slices []*draapi.ResourceSlice + IsIncomplete bool + IsInvalid bool + InvalidReason string + DeviceSlicesTargetingNode []*draapi.ResourceSlice + DeviceSlicesNotTargetingNode []*draapi.ResourceSlice + CounterSets map[draapi.UniqueString]*draapi.CounterSet } type PoolID struct { @@ -227,3 +394,126 @@ type PoolID struct { func (p PoolID) String() string { return p.Driver.String() + "/" + p.Pool.String() } + +// At V(6), log only a limited number of devices to avoid blowing up logs. For +// many E2E tests, 10 devices is enough for all devices without having to +// truncate, at least when running the tests sequentially. +const maxDevicesLevel6 = 10 + +// logPools returns a handle for the value in a structured log call which +// includes varying amounts of information about the pools, depending on +// the verbosity of the logger. +func logPools(logger klog.Logger, pools []*Pool) any { + // We need to check verbosity here because our caller's source code + // location may be relevant (-vmodule !). + helper, logger := logger.WithCallStackHelper() + helper() + + // We always produce the same output at V <= 5. 6 adds a summary and + // 7 is a complete dump. + verbosity := 5 + for i := 7; i > verbosity; i-- { + if loggerV := logger.V(i); loggerV.Enabled() { + verbosity = i + break + } + } + return &poolsLogger{verbosity, pools} +} + +type poolsLogger struct { + verbosity int + pools []*Pool +} + +var _ logr.Marshaler = &poolsLogger{} + +func (p *poolsLogger) MarshalLog() any { + info := map[string]any{"count": len(p.pools)} + if p.verbosity == 6 { + meta := make([]map[string]any, len(p.pools)) + for i, pool := range p.pools { + meta[i] = map[string]any{ + "id": pool.PoolID.String(), + "isIncomplete": pool.IsIncomplete, + "isInvalid": pool.IsInvalid, + "InvalidReason": pool.InvalidReason, + } + } + info["meta"] = meta + info["devices"] = p.listDevices(maxDevicesLevel6) + } + if p.verbosity >= 7 { + info["devices"] = p.listDevices(-1) + info["content"] = p.pools + } + return info +} + +func (p *poolsLogger) listDevices(maxDevices int) []string { + var devices []string + for _, pool := range p.pools { + devices = p.addDevicesInSlices(devices, pool.PoolID, pool.DeviceSlicesTargetingNode, maxDevices) + devices = p.addDevicesInSlices(devices, pool.PoolID, pool.DeviceSlicesNotTargetingNode, maxDevices) + } + return devices +} + +func (p *poolsLogger) addDevicesInSlices(devices []string, poolID PoolID, slices []*draapi.ResourceSlice, maxDevices int) []string { + for _, slice := range slices { + for _, device := range slice.Spec.Devices { + if maxDevices != -1 && len(devices) >= maxDevices { + devices = append(devices, "...") + return devices + } + devices = append(devices, DeviceID{Driver: poolID.Driver, Pool: poolID.Pool, Device: device.Name}.String()) + } + } + return devices +} + +// logPools returns a handle for the value in a structured log call which +// includes varying amounts of information about the allocated devices, depending on +// the verbosity of the logger. +func logAllocatedDevices(logger klog.Logger, allocatedDevices sets.Set[DeviceID]) any { + // We need to check verbosity here because our caller's source code + // location may be relevant (-vmodule !). + helper, logger := logger.WithCallStackHelper() + helper() + + // We always produce the same output at V <= 5. 6 adds all IDs. + verbosity := 5 + for i := 7; i > verbosity; i-- { + if loggerV := logger.V(i); loggerV.Enabled() { + verbosity = i + break + } + } + + return &allocatedDevicesLogger{verbosity, allocatedDevices} +} + +type allocatedDevicesLogger struct { + verbosity int + devices sets.Set[DeviceID] +} + +var _ logr.Marshaler = &allocatedDevicesLogger{} + +func (a *allocatedDevicesLogger) MarshalLog() any { + info := map[string]any{"count": len(a.devices)} + if a.verbosity >= 6 { + ids := make([]string, 0, len(a.devices)) + for id := range a.devices { + if a.verbosity == 6 && len(ids) >= maxDevicesLevel6 { + ids = append(ids, "...") + break + } + ids = append(ids, id.String()) + } + slices.Sort(ids) + info["devices"] = ids + + } + return info +}