mirror of
https://github.com/k3s-io/kubernetes.git
synced 2026-08-08 23:37:11 +00:00
fix resource claims deallocation for extended resource when pod is completed
Signed-off-by: Alay Patel <alayp@nvidia.com>
This commit is contained in:
Alay Patel
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
|
||||
Reference in New Issue
Block a user