DRA: Update allocator for Prioritized Alternatives in Device Requests

Author: mortent

pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/google/go-cmp/cmp" //nolint:depguard
@@ -103,6 +104,7 @@ type informationForClaim struct {
 type DynamicResources struct {
 	enabled                   bool
 	enableAdminAccess         bool
+	enablePrioritizedList     bool
 	enableSchedulingQueueHint bool
 
 	fh         framework.Handle
@@ -121,6 +123,7 @@ func New(ctx context.Context, plArgs runtime.Object, fh framework.Handle, fts fe
 	pl := &DynamicResources{
 		enabled:                   true,
 		enableAdminAccess:         fts.EnableDRAAdminAccess,
+		enablePrioritizedList:     fts.EnableDRAPrioritizedList,
 		enableSchedulingQueueHint: fts.EnableSchedulingQueueHint,
 
 		fh:        fh,
@@ -405,20 +408,19 @@ func (pl *DynamicResources) PreFilter(ctx context.Context, state *framework.Cycl
 			// initial set of potential nodes before we ask the
 			// driver(s) for information about the specific pod.
 			for _, request := range claim.Spec.Devices.Requests {
-				if request.DeviceClassName == "" {
-					return nil, statusError(logger, fmt.Errorf("request %s: unsupported request type", request.Name))
-				}
-
-				_, err := pl.draManager.DeviceClasses().Get(request.DeviceClassName)
-				if err != nil {
-					// If the class cannot be retrieved, allocation cannot proceed.
-					if apierrors.IsNotFound(err) {
-						// Here we mark the pod as "unschedulable", so it'll sleep in
-						// the unscheduleable queue until a DeviceClass event occurs.
-						return nil, statusUnschedulable(logger, fmt.Sprintf("request %s: device class %s does not exist", request.Name, request.DeviceClassName))
+				// The requirements differ depending on whether the request has a list of
+				// alternative subrequests defined in the firstAvailable field.
+				if len(request.FirstAvailable) == 0 {
+					if status := pl.validateDeviceClass(logger, request.DeviceClassName, request.Name); status != nil {
+						return nil, status
+					}
+				} else {
+					for _, subRequest := range request.FirstAvailable {
+						qualRequestName := strings.Join([]string{request.Name, subRequest.Name}, "/")
+						if status := pl.validateDeviceClass(logger, subRequest.DeviceClassName, qualRequestName); status != nil {
+							return nil, status
+						}
 					}
-					// Other error, retry with backoff.
-					return nil, statusError(logger, fmt.Errorf("request %s: look up device class: %w", request.Name, err))
 				}
 			}
 		}
@@ -447,7 +449,7 @@ func (pl *DynamicResources) PreFilter(ctx context.Context, state *framework.Cycl
 		if err != nil {
 			return nil, statusError(logger, err)
 		}
-		allocator, err := structured.NewAllocator(ctx, pl.enableAdminAccess, allocateClaims, allAllocatedDevices, pl.draManager.DeviceClasses(), slices, pl.celCache)
+		allocator, err := structured.NewAllocator(ctx, pl.enableAdminAccess, pl.enablePrioritizedList, allocateClaims, allAllocatedDevices, pl.draManager.DeviceClasses(), slices, pl.celCache)
 		if err != nil {
 			return nil, statusError(logger, err)
 		}
@@ -459,6 +461,23 @@ func (pl *DynamicResources) PreFilter(ctx context.Context, state *framework.Cycl
 	return nil, nil
 }
 
+func (pl *DynamicResources) validateDeviceClass(logger klog.Logger, deviceClassName, requestName string) *framework.Status {
+	if deviceClassName == "" {
+		return statusError(logger, fmt.Errorf("request %s: unsupported request type", requestName))
+	}
+
+	_, err := pl.draManager.DeviceClasses().Get(deviceClassName)
+	if err != nil {
+		// If the class cannot be retrieved, allocation cannot proceed.
+		if apierrors.IsNotFound(err) {
+			// Here we mark the pod as "unschedulable", so it'll sleep in
+			// the unscheduleable queue until a DeviceClass event occurs.
+			return statusUnschedulable(logger, fmt.Sprintf("request %s: device class %s does not exist", requestName, deviceClassName))
+		}
+	}
+	return nil
+}
+
 // PreFilterExtensions returns prefilter extensions, pod add and remove.
 func (pl *DynamicResources) PreFilterExtensions() framework.PreFilterExtensions {
 	return nil

pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go

@@ -117,9 +117,17 @@ var (
 		Namespace(namespace).
 		Request(className).
 		Obj()
+	claimWithPrioritzedList = st.MakeResourceClaim().
+				Name(claimName).
+				Namespace(namespace).
+				RequestWithPrioritizedList(className).
+				Obj()
 	pendingClaim = st.FromResourceClaim(claim).
 			OwnerReference(podName, podUID, podKind).
 			Obj()
+	pendingClaimWithPrioritizedList = st.FromResourceClaim(claimWithPrioritzedList).
+					OwnerReference(podName, podUID, podKind).
+					Obj()
 	allocationResult = &resourceapi.AllocationResult{
 		Devices: resourceapi.DeviceAllocationResult{
 			Results: []resourceapi.DeviceRequestAllocationResult{{
@@ -133,13 +141,33 @@ var (
 			return st.MakeNodeSelector().In("metadata.name", []string{nodeName}, st.NodeSelectorTypeMatchFields).Obj()
 		}(),
 	}
+	allocationResultWithPrioritizedList = &resourceapi.AllocationResult{
+		Devices: resourceapi.DeviceAllocationResult{
+			Results: []resourceapi.DeviceRequestAllocationResult{{
+				Driver:  driver,
+				Pool:    nodeName,
+				Device:  "instance-1",
+				Request: "req-1/subreq-1",
+			}},
+		},
+		NodeSelector: func() *v1.NodeSelector {
+			return st.MakeNodeSelector().In("metadata.name", []string{nodeName}, st.NodeSelectorTypeMatchFields).Obj()
+		}(),
+	}
 	inUseClaim = st.FromResourceClaim(pendingClaim).
 			Allocation(allocationResult).
 			ReservedForPod(podName, types.UID(podUID)).
 			Obj()
+	inUseClaimWithPrioritizedList = st.FromResourceClaim(pendingClaimWithPrioritizedList).
+					Allocation(allocationResultWithPrioritizedList).
+					ReservedForPod(podName, types.UID(podUID)).
+					Obj()
 	allocatedClaim = st.FromResourceClaim(pendingClaim).
 			Allocation(allocationResult).
 			Obj()
+	allocatedClaimWithPrioritizedList = st.FromResourceClaim(pendingClaimWithPrioritizedList).
+						Allocation(allocationResultWithPrioritizedList).
+						Obj()
 
 	allocatedClaimWithWrongTopology = st.FromResourceClaim(allocatedClaim).
 					Allocation(&resourceapi.AllocationResult{NodeSelector: st.MakeNodeSelector().In("no-such-label", []string{"no-such-value"}, st.NodeSelectorTypeMatchExpressions).Obj()}).
@@ -201,6 +229,24 @@ func breakCELInClass(class *resourceapi.DeviceClass) *resourceapi.DeviceClass {
 	return class
 }
 
+func updateDeviceClassName(claim *resourceapi.ResourceClaim, deviceClassName string) *resourceapi.ResourceClaim {
+	claim = claim.DeepCopy()
+	for i := range claim.Spec.Devices.Requests {
+		// If the firstAvailable list is empty we update the device class name
+		// on the base request.
+		if len(claim.Spec.Devices.Requests[i].FirstAvailable) == 0 {
+			claim.Spec.Devices.Requests[i].DeviceClassName = deviceClassName
+		} else {
+			// If subrequests are specified, update the device class name on
+			// all of them.
+			for j := range claim.Spec.Devices.Requests[i].FirstAvailable {
+				claim.Spec.Devices.Requests[i].FirstAvailable[j].DeviceClassName = deviceClassName
+			}
+		}
+	}
+	return claim
+}
+
 // result defines the expected outcome of some operation. It covers
 // operation's status and the state of the world (= objects).
 type result struct {
@@ -295,6 +341,8 @@ func TestPlugin(t *testing.T) {
 		// Feature gates. False is chosen so that the uncommon case
 		// doesn't need to be set.
 		disableDRA bool
+
+		enableDRAPrioritizedList bool
 	}{
 		"empty": {
 			pod: st.MakePod().Name("foo").Namespace("default").Obj(),
@@ -795,6 +843,69 @@ func TestPlugin(t *testing.T) {
 			},
 			disableDRA: true,
 		},
+		"claim-with-request-with-unknown-device-class": {
+			pod:    podWithClaimName,
+			claims: []*resourceapi.ResourceClaim{updateDeviceClassName(claim, "does-not-exist")},
+			want: want{
+				prefilter: result{
+					status: framework.NewStatus(framework.Unschedulable, `request req-1: device class does-not-exist does not exist`),
+				},
+				postfilter: result{
+					status: framework.NewStatus(framework.Unschedulable, `no new claims to deallocate`),
+				},
+			},
+		},
+		"claim-with-prioritized-list-feature-disabled": {
+			enableDRAPrioritizedList: false,
+			pod:                      podWithClaimName,
+			claims:                   []*resourceapi.ResourceClaim{claimWithPrioritzedList},
+			classes:                  []*resourceapi.DeviceClass{deviceClass},
+			want: want{
+				filter: perNodeResult{
+					workerNode.Name: {
+						status: framework.NewStatus(framework.UnschedulableAndUnresolvable, `claim default/my-pod-my-resource, request req-1: has subrequests, but the feature is disabled`),
+					},
+				},
+			},
+		},
+		"claim-with-prioritized-list-unknown-device-class": {
+			enableDRAPrioritizedList: true,
+			pod:                      podWithClaimName,
+			claims:                   []*resourceapi.ResourceClaim{updateDeviceClassName(claimWithPrioritzedList, "does-not-exist")},
+			want: want{
+				prefilter: result{
+					status: framework.NewStatus(framework.Unschedulable, `request req-1/subreq-1: device class does-not-exist does not exist`),
+				},
+				postfilter: result{
+					status: framework.NewStatus(framework.Unschedulable, `no new claims to deallocate`),
+				},
+			},
+		},
+		"claim-with-prioritized-list": {
+			enableDRAPrioritizedList: true,
+			pod:                      podWithClaimName,
+			claims:                   []*resourceapi.ResourceClaim{pendingClaimWithPrioritizedList},
+			classes:                  []*resourceapi.DeviceClass{deviceClass},
+			objs:                     []apiruntime.Object{workerNodeSlice},
+			want: want{
+				reserve: result{
+					inFlightClaim: allocatedClaimWithPrioritizedList,
+				},
+				prebind: result{
+					assumedClaim: reserve(allocatedClaimWithPrioritizedList, podWithClaimName),
+					changes: change{
+						claim: func(claim *resourceapi.ResourceClaim) *resourceapi.ResourceClaim {
+							if claim.Name == claimName {
+								claim = claim.DeepCopy()
+								claim.Finalizers = allocatedClaimWithPrioritizedList.Finalizers
+								claim.Status = inUseClaimWithPrioritizedList.Status
+							}
+							return claim
+						},
+					},
+				},
+			},
+		},
 	}
 
 	for name, tc := range testcases {
@@ -809,6 +920,7 @@ func TestPlugin(t *testing.T) {
 			features := feature.Features{
 				EnableDRAAdminAccess:            tc.enableDRAAdminAccess,
 				EnableDynamicResourceAllocation: !tc.disableDRA,
+				EnableDRAPrioritizedList:        tc.enableDRAPrioritizedList,
 			}
 			testCtx := setup(t, nodes, tc.claims, tc.classes, tc.objs, features)
 			initialObjects := testCtx.listAll(t)

pkg/scheduler/framework/plugins/feature/feature.go

@@ -20,6 +20,7 @@ package feature
 // This struct allows us to break the dependency of the plugins on
 // the internal k8s features pkg.
 type Features struct {
+	EnableDRAPrioritizedList                     bool
 	EnableDRAAdminAccess                         bool
 	EnableDynamicResourceAllocation              bool
 	EnableVolumeCapacityPriority                 bool

pkg/scheduler/framework/plugins/registry.go

@@ -46,6 +46,7 @@ import (
 // through the WithFrameworkOutOfTreeRegistry option.
 func NewInTreeRegistry() runtime.Registry {
 	fts := plfeature.Features{
+		EnableDRAPrioritizedList:                     feature.DefaultFeatureGate.Enabled(features.DRAPrioritizedList),
 		EnableDRAAdminAccess:                         feature.DefaultFeatureGate.Enabled(features.DRAAdminAccess),
 		EnableDynamicResourceAllocation:              feature.DefaultFeatureGate.Enabled(features.DynamicResourceAllocation),
 		EnableVolumeCapacityPriority:                 feature.DefaultFeatureGate.Enabled(features.VolumeCapacityPriority),

pkg/scheduler/testing/wrappers.go

@@ -1104,6 +1104,28 @@ func (wrapper *ResourceClaimWrapper) Request(deviceClassName string) *ResourceCl
 	return wrapper
 }
 
+// RequestWithPrioritizedList adds one device request with one subrequest
+// per provided deviceClassName.
+func (wrapper *ResourceClaimWrapper) RequestWithPrioritizedList(deviceClassNames ...string) *ResourceClaimWrapper {
+	var prioritizedList []resourceapi.DeviceSubRequest
+	for i, deviceClassName := range deviceClassNames {
+		prioritizedList = append(prioritizedList, resourceapi.DeviceSubRequest{
+			Name:            fmt.Sprintf("subreq-%d", i+1),
+			AllocationMode:  resourceapi.DeviceAllocationModeExactCount,
+			Count:           1,
+			DeviceClassName: deviceClassName,
+		})
+	}
+
+	wrapper.Spec.Devices.Requests = append(wrapper.Spec.Devices.Requests,
+		resourceapi.DeviceRequest{
+			Name:           fmt.Sprintf("req-%d", len(wrapper.Spec.Devices.Requests)+1),
+			FirstAvailable: prioritizedList,
+		},
+	)
+	return wrapper
+}
+
 // Allocation sets the allocation of the inner object.
 func (wrapper *ResourceClaimWrapper) Allocation(allocation *resourceapi.AllocationResult) *ResourceClaimWrapper {
 	if !slices.Contains(wrapper.ResourceClaim.Finalizers, resourceapi.Finalizer) {