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

Merge pull request #137607 from 0xMH/fix/133531-dra-queueing-hint

Browse Source 
DRA: return error on filter timeout instead of unschedulable
This commit is contained in:
Kubernetes Prow Robot
2026-03-18 14:26:31 +05:30
committed by GitHub
parent 7d56731021 67deec3bfa
commit 5953c57c6b
5 changed files with 48 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go
View File

@@ -721,7 +721,13 @@ func (pl *DynamicResources) Filter(ctx context.Context, cs fwk.CycleState, pod *
a, err := state.allocator.Allocate(allocCtx, node, claimsToAllocate)
switch {
case errors.Is(err, context.DeadlineExceeded):
return statusUnschedulable(logger, "timed out trying to allocate devices", "pod", klog.KObj(pod), "node", klog.KObj(node), "resourceclaims", klog.KObjSlice(claimsToAllocate))
// Timeouts are potentially transient. Return Error
// so the pod retries via backoff instead of sitting in the
// unschedulable queue waiting for a cluster event.
//
// The timeout might be caused by ResourceSlices for the node,
// so including the node name may help with diagnosing the failure.
return statusError(logger, fmt.Errorf("node %s: timed out trying to allocate devices", node.Name), "pod", klog.KObj(pod), "node", klog.KObj(node), "resourceclaims", klog.KObjSlice(claimsToAllocate))
case errors.Is(err, structured.ErrFailedAllocationOnNode):
// Not a fatal error, allocation on other nodes may proceed.
// The error is only surfaced if allocation fails on all nodes.

7
pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go
View File

@@ -2114,12 +2114,11 @@ func testPlugin(tCtx ktesting.TContext) {
want: want{
filter: perNodeResult{
workerNode.Name: {
status: fwk.NewStatus(fwk.UnschedulableAndUnresolvable, `timed out trying to allocate devices`),
// Timeouts return Error so the pod retries via backoff.
status: fwk.AsStatus(fmt.Errorf("node %s: timed out trying to allocate devices", workerNode.Name)),
},
},
postfilter: result{
status: fwk.NewStatus(fwk.Unschedulable, `still not schedulable`),
},
// No postfilter: Error aborts scheduling immediately.
},
// Skipping this test case on Windows as a 1ns timeout is not guaranteed to
// expire immediately on Windows due to its coarser timer granularity -

39
test/integration/dra/core.go
View File

@@ -95,23 +95,30 @@ func testConvert(tCtx ktesting.TContext) {
// testFilterTimeout covers the scheduler plugin's filter timeout configuration and behavior.
//
// It runs the scheduler with non-standard settings and thus cannot run in parallel.
func testFilterTimeout(tCtx ktesting.TContext, devicesPerSlice int) {
func testFilterTimeout(tCtx ktesting.TContext, requestDeviceCount int) {
namespace := createTestNamespace(tCtx, nil)
class, driverName := createTestClass(tCtx, namespace)
deviceNames := make([]string, devicesPerSlice)
for i := range devicesPerSlice {
deviceNames := make([]string, requestDeviceCount)
for i := range requestDeviceCount {
deviceNames[i] = fmt.Sprintf("dev-%d", i)
}
slice := st.MakeResourceSlice("worker-0", driverName).Devices(deviceNames...)
createSlice(tCtx, slice.Obj())
otherSlice := st.MakeResourceSlice("worker-1", driverName).Devices(deviceNames...)
otherSlice := st.MakeResourceSlice("worker-1", driverName).Devices(deviceNames[:requestDeviceCount-1]...)
createdOtherSlice := createSlice(tCtx, otherSlice.Obj())
claim := claim.DeepCopy()
claim.Spec.Devices.Requests[0].Exactly.Count = int64(devicesPerSlice + 1) // Impossible to allocate.
claim = createClaim(tCtx, namespace, "", class, claim)
// Impossible to allocate on worker-1: not enough devices, but allocation is too
// dumb to notice that upfront and keeps trying until it times out.
// On worker-0 we can allocate, but don't schedule because of the timeout on worker-1.
newClaim := func(suffix string) *resourceapi.ResourceClaim {
c := claim.DeepCopy()
c.Spec.Devices.Requests[0].Exactly.Count = int64(requestDeviceCount)
return createClaim(tCtx, namespace, suffix, class, c)
}
runSubTest(tCtx, "disabled", func(tCtx ktesting.TContext) {
pod := createPod(tCtx, namespace, "", podWithClaimName, claim)
cl := newClaim("-disabled")
pod := createPod(tCtx, namespace, "-disabled", podWithClaimName, cl)
startSchedulerWithConfig(tCtx, `
profiles:
- schedulerName: default-scheduler
@@ -120,11 +127,14 @@ profiles:
args:
filterTimeout: 0s
`)
expectPodUnschedulable(tCtx, pod, "cannot allocate all claims")
// Without a timeout, the allocator runs to completion on both nodes.
// worker-0 has enough devices and succeeds, so the pod gets scheduled.
tCtx.ExpectNoError(e2epod.WaitForPodScheduled(tCtx, tCtx.Client(), namespace, pod.Name))
})
runSubTest(tCtx, "enabled", func(tCtx ktesting.TContext) {
pod := createPod(tCtx, namespace, "", podWithClaimName, claim)
cl := newClaim("-enabled")
pod := createPod(tCtx, namespace, "-enabled", podWithClaimName, cl)
startSchedulerWithConfig(tCtx, `
profiles:
- schedulerName: default-scheduler
@@ -133,12 +143,13 @@ profiles:
args:
filterTimeout: 10ms
`)
expectPodUnschedulable(tCtx, pod, "timed out trying to allocate devices")
expectPodSchedulerError(tCtx, pod, "timed out trying to allocate devices")
// Update one slice such that allocation succeeds.
// The scheduler must retry and should succeed now.
// Update the smaller slice such that allocation also succeeds.
// The scheduler retries automatically (timeouts go through
// backoff queue, not unschedulable pool) and should succeed now.
createdOtherSlice.Spec.Devices = append(createdOtherSlice.Spec.Devices, resourceapi.Device{
Name: fmt.Sprintf("dev-%d", devicesPerSlice),
Name: deviceNames[requestDeviceCount-1],
})
_, err := tCtx.Client().ResourceV1().ResourceSlices().Update(tCtx, createdOtherSlice, metav1.UpdateOptions{})
tCtx.ExpectNoError(err, "update worker-1's ResourceSlice")

4
test/integration/dra/dra.go
View File

@@ -133,7 +133,7 @@ func run(tCtx ktesting.TContext, whatRE string) {
runSubTest(tCtx, "EvictClusterWithSlices", func(tCtx ktesting.TContext) { testEvictCluster(tCtx, useNoRule) })
// Number of devices per slice is chosen so that Filter takes a few seconds:
// without a timeout, the test doesn't run too long, but long enough that a short timeout triggers.
runSubTest(tCtx, "FilterTimeout", func(tCtx ktesting.TContext) { testFilterTimeout(tCtx, 20) })
runSubTest(tCtx, "FilterTimeout", func(tCtx ktesting.TContext) { testFilterTimeout(tCtx, 21) })
runSubTest(tCtx, "UsesAllResources", testUsesAllResources)
},
},
@@ -243,7 +243,7 @@ func run(tCtx ktesting.TContext, whatRE string) {
// Number of devices per slice is chosen so that Filter takes a few seconds: The allocator
// in the experimental channel has an improvement that requires a higher number here than
// in the incubating and stable channels.
runSubTest(tCtx, "FilterTimeout", func(tCtx ktesting.TContext) { testFilterTimeout(tCtx, 20) })
runSubTest(tCtx, "FilterTimeout", func(tCtx ktesting.TContext) { testFilterTimeout(tCtx, 21) })
runSubTest(tCtx, "ShareResourceClaimSequentially", testShareResourceClaimSequentially)
runSubTest(tCtx, "UsesAllResources", testUsesAllResources)
},

21
test/integration/dra/helpers.go
View File

@@ -272,17 +272,18 @@ func waitForClaimAllocatedToDevice(tCtx ktesting.TContext, namespace, claimName
)
}
func expectPodUnschedulable(tCtx ktesting.TContext, pod *v1.Pod, reason string) {
func expectPodSchedulerError(tCtx ktesting.TContext, pod *v1.Pod, reason string) {
tCtx.Helper()
tCtx.ExpectNoError(e2epod.WaitForPodNameUnschedulableInNamespace(tCtx, tCtx.Client(), pod.Name, pod.Namespace), fmt.Sprintf("expected pod to be unschedulable because %q", reason))
pod, err := tCtx.Client().CoreV1().Pods(pod.Namespace).Get(tCtx, pod.Name, metav1.GetOptions{})
tCtx.ExpectNoError(err)
gomega.NewWithT(tCtx).Expect(pod).To(gomega.HaveField("Status.Conditions", gomega.ContainElement(gstruct.MatchFields(gstruct.IgnoreExtras, gstruct.Fields{
"Type": gomega.Equal(v1.PodScheduled),
"Status": gomega.Equal(v1.ConditionFalse),
"Reason": gomega.Equal(v1.PodReasonUnschedulable),
"Message": gomega.ContainSubstring(reason),
}))))
tCtx.ExpectNoError(e2epod.WaitForPodCondition(tCtx, tCtx.Client(), pod.Namespace, pod.Name, v1.PodReasonSchedulerError, time.Minute, func(pod *v1.Pod) (bool, error) {
if pod.Status.Phase == v1.PodPending {
for _, cond := range pod.Status.Conditions {
if cond.Type == v1.PodScheduled && cond.Status == v1.ConditionFalse && cond.Reason == v1.PodReasonSchedulerError && strings.Contains(cond.Message, reason) {
return true, nil
}
}
}
return false, nil
}), fmt.Sprintf("expected pod to have scheduler error because %q", reason))
}
type nodeInfo struct {