From 8a03067211e0ad3162fb7e793c4769f6f423f94e Mon Sep 17 00:00:00 2001 From: Alay Patel Date: Sat, 27 Sep 2025 17:13:13 -0400 Subject: [PATCH] fix resource claims deallocation for extended resource when pod is completed Signed-off-by: Alay Patel --- pkg/controller/resourceclaim/controller.go | 22 +- .../resourceclaim/controller_test.go | 195 ++++++++++++++++++ test/e2e/dra/dra.go | 64 ++++++ 3 files changed, 280 insertions(+), 1 deletion(-) diff --git a/pkg/controller/resourceclaim/controller.go b/pkg/controller/resourceclaim/controller.go index 37c7991b84c..f3b94b1331b 100644 --- a/pkg/controller/resourceclaim/controller.go +++ b/pkg/controller/resourceclaim/controller.go @@ -223,7 +223,15 @@ func (ec *Controller) enqueuePod(logger klog.Logger, obj interface{}, deleted bo return } - if len(pod.Spec.ResourceClaims) == 0 { + // Check if pod has any resource claims to process. + // Extended resource claims are stored in pod.Status.ExtendedResourceClaimStatus, + // not in pod.Spec.ResourceClaims, so we need to check both locations. + hasResourceClaims := len(pod.Spec.ResourceClaims) > 0 + // For cleanup of extended resource claims, we must consider claims present + // in pod status regardless of the current feature gate state. The claim may + // have been created when the feature was enabled and still needs cleanup. + hasExtendedResourceClaims := pod.Status.ExtendedResourceClaimStatus != nil + if !hasResourceClaims && !hasExtendedResourceClaims { // Nothing to do for it at all. return } @@ -257,6 +265,18 @@ func (ec *Controller) enqueuePod(logger klog.Logger, obj interface{}, deleted bo logger.V(6).Info("Nothing to do for skipped claim during pod change", "pod", klog.KObj(pod), "podClaim", podClaim.Name, "reason", reason) } } + + // Process extended resource claims for completed/deleted pods. + // Extended resource claims are created by the scheduler and stored in + // pod.Status.ExtendedResourceClaimStatus, not in pod.Spec.ResourceClaims. + // Without this, extended resource claims would never be cleaned up when + // pods complete, causing device resources to remain allocated indefinitely. + if hasExtendedResourceClaims { + claimName := pod.Status.ExtendedResourceClaimStatus.ResourceClaimName + key := claimKeyPrefix + pod.Namespace + "/" + claimName + logger.V(6).Info("Process extended resource claim", "pod", klog.KObj(pod), "claim", klog.KRef(pod.Namespace, claimName), "key", key, "reason", reason) + ec.queue.Add(key) + } } needsWork, reason := ec.podNeedsWork(pod) diff --git a/pkg/controller/resourceclaim/controller_test.go b/pkg/controller/resourceclaim/controller_test.go index bdcda37be32..b8b465245e1 100644 --- a/pkg/controller/resourceclaim/controller_test.go +++ b/pkg/controller/resourceclaim/controller_test.go @@ -34,15 +34,18 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/diff" + utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes/fake" resourcelisters "k8s.io/client-go/listers/resource/v1" k8stesting "k8s.io/client-go/testing" + featuregatetesting "k8s.io/component-base/featuregate/testing" "k8s.io/component-base/metrics" "k8s.io/component-base/metrics/testutil" "k8s.io/klog/v2" "k8s.io/kubernetes/pkg/controller" resourceclaimmetrics "k8s.io/kubernetes/pkg/controller/resourceclaim/metrics" + "k8s.io/kubernetes/pkg/features" "k8s.io/kubernetes/test/utils/ktesting" "k8s.io/utils/ptr" ) @@ -1127,3 +1130,195 @@ func (em numMetrics) withUpdates(notAllocatedDelta, notAllocatedWithAdminDelta, lister: em.lister, } } + +func TestEnqueuePodExtendedResourceClaims(t *testing.T) { + tests := []struct { + name string + pod *v1.Pod + featureGateEnabled bool + deleted bool + expectEarlyReturn bool + expectExtendedClaimEnqueued bool + }{ + { + name: "pod with no resource claims", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{}, + Status: v1.PodStatus{}, + }, + featureGateEnabled: true, + expectEarlyReturn: true, + expectExtendedClaimEnqueued: false, + }, + { + name: "pod with regular resource claims only", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{ + ResourceClaims: []v1.PodResourceClaim{{Name: "regular-claim"}}, + }, + Status: v1.PodStatus{Phase: v1.PodRunning}, + }, + featureGateEnabled: true, + expectEarlyReturn: false, + expectExtendedClaimEnqueued: false, + }, + { + name: "pod with extended resource claim, feature enabled, running pod", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{}, + Status: v1.PodStatus{ + Phase: v1.PodRunning, + ExtendedResourceClaimStatus: &v1.PodExtendedResourceClaimStatus{ + ResourceClaimName: "test-extended-claim", + }, + }, + }, + featureGateEnabled: true, + expectEarlyReturn: false, + expectExtendedClaimEnqueued: false, + }, + { + name: "pod with extended resource claim, feature enabled, completed pod", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{}, + Status: v1.PodStatus{ + Phase: v1.PodSucceeded, + ExtendedResourceClaimStatus: &v1.PodExtendedResourceClaimStatus{ + ResourceClaimName: "test-extended-claim", + }, + }, + }, + featureGateEnabled: true, + expectEarlyReturn: false, + expectExtendedClaimEnqueued: true, + }, + { + name: "pod with extended resource claim, feature enabled, failed pod", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{}, + Status: v1.PodStatus{ + Phase: v1.PodFailed, // Failed pod + ExtendedResourceClaimStatus: &v1.PodExtendedResourceClaimStatus{ + ResourceClaimName: "test-extended-claim", + }, + }, + }, + featureGateEnabled: true, + expectEarlyReturn: false, + expectExtendedClaimEnqueued: true, + }, + { + name: "pod with extended resource claim, feature disabled", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{}, + Status: v1.PodStatus{ + Phase: v1.PodSucceeded, + ExtendedResourceClaimStatus: &v1.PodExtendedResourceClaimStatus{ + ResourceClaimName: "test-extended-claim", + }, + }, + }, + featureGateEnabled: false, + expectEarlyReturn: false, + expectExtendedClaimEnqueued: true, + }, + { + name: "deleted pod with extended resource claim", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{}, + Status: v1.PodStatus{ + Phase: v1.PodSucceeded, + ExtendedResourceClaimStatus: &v1.PodExtendedResourceClaimStatus{ + ResourceClaimName: "test-extended-claim", + }, + }, + }, + featureGateEnabled: true, + deleted: true, + expectEarlyReturn: false, + expectExtendedClaimEnqueued: true, + }, + { + name: "pod with both regular and extended resource claims, completed", + pod: &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-pod", Namespace: "default"}, + Spec: v1.PodSpec{ + ResourceClaims: []v1.PodResourceClaim{{Name: "regular-claim"}}, + }, + Status: v1.PodStatus{ + Phase: v1.PodSucceeded, + ExtendedResourceClaimStatus: &v1.PodExtendedResourceClaimStatus{ + ResourceClaimName: "test-extended-claim", + }, + }, + }, + featureGateEnabled: true, + expectEarlyReturn: false, + expectExtendedClaimEnqueued: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.DRAExtendedResource, test.featureGateEnabled) + + tCtx := ktesting.Init(t) + tCtx = ktesting.WithCancel(tCtx) + + fakeKubeClient := createTestClient() + informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, controller.NoResyncPeriodFunc()) + podInformer := informerFactory.Core().V1().Pods() + claimInformer := informerFactory.Resource().V1().ResourceClaims() + templateInformer := informerFactory.Resource().V1().ResourceClaimTemplates() + + setupMetrics() + + ec, err := NewController(tCtx.Logger(), Features{}, fakeKubeClient, podInformer, claimInformer, templateInformer) + if err != nil { + t.Fatalf("error creating controller: %v", err) + } + + ec.enqueuePod(tCtx.Logger(), test.pod, test.deleted) + + var keys []string + for ec.queue.Len() > 0 { + k, _ := ec.queue.Get() + keys = append(keys, k) + ec.queue.Forget(k) + ec.queue.Done(k) + } + + if test.expectEarlyReturn { + if len(keys) != 0 { + t.Errorf("expected no keys enqueued on early return, got: %v", keys) + } + return + } + + var expectedClaimKey string + if test.pod.Status.ExtendedResourceClaimStatus != nil { + expectedClaimKey = claimKeyPrefix + test.pod.Namespace + "/" + test.pod.Status.ExtendedResourceClaimStatus.ResourceClaimName + } + found := false + for _, k := range keys { + if k == expectedClaimKey { + found = true + break + } + } + if test.expectExtendedClaimEnqueued && !found { + t.Errorf("expected extended claim key %q to be enqueued, got keys: %v", expectedClaimKey, keys) + } + if !test.expectExtendedClaimEnqueued && found { + t.Errorf("did not expect extended claim key %q to be enqueued, got keys: %v", expectedClaimKey, keys) + } + }) + } +} diff --git a/test/e2e/dra/dra.go b/test/e2e/dra/dra.go index 6d780d7f45f..1cfb33efbe0 100644 --- a/test/e2e/dra/dra.go +++ b/test/e2e/dra/dra.go @@ -1924,6 +1924,48 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), func() { drautils.TestContainerEnv(ctx, f, pod, pod.Spec.Containers[0].Name, false, containerEnv...) } + runExtendedClaimCleanup := func(ctx context.Context, b *drautils.Builder, shellScript string, waitFn func(context.Context, *v1.Pod) error, cleanupMessage string) { + pod := b.Pod() + res := v1.ResourceList{} + res[v1.ResourceName(b.ExtendedResourceName(0))] = resource.MustParse("1") + pod.Spec.Containers[0].Resources.Requests = res + pod.Spec.Containers[0].Resources.Limits = res + + pod.Spec.Containers[0].Command = []string{"/bin/sh"} + pod.Spec.Containers[0].Args = []string{"-c", shellScript} + pod.Spec.RestartPolicy = v1.RestartPolicyNever + + ginkgo.By("Creating the pod with extended resource") + b.Create(ctx, pod) + err := e2epod.WaitForPodRunningInNamespace(ctx, f.ClientSet, pod) + framework.ExpectNoError(err, "start pod") + + ginkgo.By("Verifying extended resource claim exists") + var extendedResourceClaim *resourceapi.ResourceClaim + gomega.Eventually(ctx, func(ctx context.Context) bool { + updatedPod, err := f.ClientSet.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{}) + if err != nil { + return false + } + if updatedPod.Status.ExtendedResourceClaimStatus == nil { + return false + } + claimName := updatedPod.Status.ExtendedResourceClaimStatus.ResourceClaimName + extendedResourceClaim, err = b.ClientV1().ResourceClaims(pod.Namespace).Get(ctx, claimName, metav1.GetOptions{}) + return err == nil && extendedResourceClaim != nil + }).WithTimeout(time.Minute).Should(gomega.BeTrueBecause("extended resource claim should be created")) + + ginkgo.By("Waiting for the pod to reach terminal state") + err = waitFn(ctx, pod) + framework.ExpectNoError(err, "waiting for pod to reach terminal state") + + ginkgo.By("Verifying extended resource claim is cleaned up") + gomega.Eventually(ctx, func(ctx context.Context) bool { + _, err := b.ClientV1().ResourceClaims(pod.Namespace).Get(ctx, extendedResourceClaim.Name, metav1.GetOptions{}) + return apierrors.IsNotFound(err) + }).WithTimeout(time.Minute).Should(gomega.BeTrueBecause("extended resource claim should be automatically deleted when pod %s", cleanupMessage)) + } + framework.Context(f.WithFeatureGate(features.DRAExtendedResource), func() { nodes := drautils.NewNodes(f, 1, 1) driver := drautils.NewDriver(f, nodes, drautils.NetworkResources(10, false)) @@ -2046,6 +2088,28 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), func() { } drautils.TestContainerEnv(ctx, f, pod, pod.Spec.Containers[2].Name, false, containerEnv...) }) + + ginkgo.It("must cleanup extended resource claims when pods complete", func(ctx context.Context) { + runExtendedClaimCleanup(ctx, b, + "echo 'Pod started with extended resource'; sleep 2; echo 'Pod completed successfully'", + func(ctx context.Context, p *v1.Pod) error { + return e2epod.WaitForPodSuccessInNamespace(ctx, f.ClientSet, p.Name, p.Namespace) + }, + "completes", + ) + }) + + ginkgo.It("must cleanup extended resource claims when pods fail", func(ctx context.Context) { + runExtendedClaimCleanup(ctx, b, + "echo 'Pod started with extended resource'; sleep 2; echo 'Pod failing now'; exit 1", + func(ctx context.Context, p *v1.Pod) error { + return e2epod.WaitForPodCondition(ctx, f.ClientSet, p.Namespace, p.Name, "pod failed", framework.PodStartTimeout, func(pod *v1.Pod) (bool, error) { + return pod.Status.Phase == v1.PodFailed, nil + }) + }, + "fails", + ) + }) }) framework.Context(f.WithFeatureGate(features.DRAExtendedResource), func() {