github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-24 03:40:56 +00:00
Code Issues Projects Releases Wiki Activity

DRA allocator: promote experimental -> incubating -> stable

Browse Source 
The previous incubating becomes stable, experimental the new incubating. Now
experimental and incubating are identical until we merge more experimental
changes again.

Specifically, these commands where used:

    rm -rf stable
    mv incubating stable
    mv stable/allocator_incubating.go stable/allocator_stable.go
    mv stable/pools_incubating.go stable/pools_stable.go
    sed -i -e 's/package incubating/package stable/' stable/*.go
    cp -a experimental incubating
    mv incubating/allocator_experimental.go incubating/allocator_incubating.go
    mv incubating/pools_experimental.go incubating/pools_incubating.go
    sed -i -e 's/package experimental/package incubating/' incubating/*.go

Some other packages then need to be adapted, in particular the
TestAllocatorSelection test.
This commit is contained in:
Patrick Ohly
2026-01-29 10:33:46 +01:00
parent 0d1b02490b
commit 5c19239290
10 changed files with 1496 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
staging/src/k8s.io/dynamic-resource-allocation/structured/allocator.go
View File

@@ -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,
},

362
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_incubating.go
View File

@@ -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) {

4
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/allocator_test.go
View File

@@ -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)
},
)
}

131
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/constraint.go Normal file
View File

@@ -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
}

211
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity.go Normal file
View File

@@ -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
}

243
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/consumable_capacity_test.go Normal file
View File

@@ -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())
})
}
}

54
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/incubating/pools_incubating.go
View File

@@ -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.
//

301
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/allocator_stable.go
View File

@@ -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)

396
staging/src/k8s.io/dynamic-resource-allocation/structured/internal/stable/pools_stable.go
View File

@@ -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
}