From ba7637c194bf723919c76139ae159cd5e21978c1 Mon Sep 17 00:00:00 2001 From: Kevin Hannon Date: Thu, 20 Nov 2025 22:43:53 -0500 Subject: [PATCH] Add integration test for MutablePodResourcesForSuspendedJobs with PodReplacementPolicy=Failed This test verifies that when a job with PodReplacementPolicy=Failed is suspended and pods are terminating, resource updates can be made to the job but new pods are only created after terminating pods are removed. The new pods should have the updated resources. --- test/integration/job/job_test.go | 192 +++++++++++++++++++++++++++++++ 1 file changed, 192 insertions(+) diff --git a/test/integration/job/job_test.go b/test/integration/job/job_test.go index 798ab81a925..c68b82fa191 100644 --- a/test/integration/job/job_test.go +++ b/test/integration/job/job_test.go @@ -5302,3 +5302,195 @@ func failTerminatingPods(ctx context.Context, t *testing.T, clientSet clientset. t.Fatalf("Failed to update pod statuses: %v", err) } } + +// TestMutablePodResourcesWithPodReplacementPolicyFailed tests that when a job with +// PodReplacementPolicy=Failed is suspended and pods are terminating, resource updates +// can be made to the job but new pods are only created after terminating pods are removed. +// The new pods should have the updated resources. +func TestMutablePodResourcesWithPodReplacementPolicyFailed(t *testing.T) { + const blockDeletionFinalizerForTest string = "fake.example.com/blockDeletion" + + // Enable MutablePodResourcesForSuspendedJobs feature gate + featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.MutablePodResourcesForSuspendedJobs, true) + + closeFn, restConfig, cs, ns := setup(t, "mutable-resources-pod-replacement") + defer closeFn() + + ctx, cancel := startJobControllerAndWaitForCaches(t, restConfig) + defer cancel() + + // Create a job with PodReplacementPolicy=Failed and a finalizer to block pod deletion + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-job", + Namespace: ns.Name, + }, + Spec: batchv1.JobSpec{ + Parallelism: ptr.To[int32](2), + Completions: ptr.To[int32](2), + PodReplacementPolicy: ptr.To(batchv1.Failed), + Template: v1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Finalizers: []string{blockDeletionFinalizerForTest}, + }, + Spec: v1.PodSpec{ + Containers: []v1.Container{{ + Name: "test-container", + Image: "busybox", + Resources: v1.ResourceRequirements{ + Requests: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("100m"), + v1.ResourceMemory: resource.MustParse("128Mi"), + }, + Limits: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("200m"), + v1.ResourceMemory: resource.MustParse("256Mi"), + }, + }, + }}, + RestartPolicy: v1.RestartPolicyNever, + }, + }, + }, + } + + job, err := createJobWithDefaults(ctx, cs, ns.Name, job) + if err != nil { + t.Fatalf("Failed to create Job: %v", err) + } + + jobClient := cs.BatchV1().Jobs(ns.Name) + + // Wait for pods to be active + t.Logf("Waiting for 2 pods to be active") + waitForPodsToBeActive(ctx, t, jobClient, 2, job) + + // Suspend the job + t.Logf("Suspending the job") + _, err = updateJob(ctx, jobClient, job.Name, func(j *batchv1.Job) { + j.Spec.Suspend = ptr.To(true) + }) + if err != nil { + t.Fatalf("Failed to suspend Job: %v", err) + } + + // Wait for the job to be suspended + err = wait.PollUntilContextTimeout(ctx, time.Second, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) { + j, err := jobClient.Get(ctx, job.Name, metav1.GetOptions{}) + if err != nil { + return false, err + } + return j.Spec.Suspend != nil && *j.Spec.Suspend, nil + }) + if err != nil { + t.Fatalf("Failed to wait for job to be suspended: %v", err) + } + t.Logf("Job is suspended") + + // Delete the pods to make them go to Terminating state + t.Logf("Deleting pods to make them go to Terminating state") + deletePods(ctx, t, cs, ns.Name) + + // Verify pods are in Terminating state (because of the finalizer) + validateJobsPodsStatusOnly(ctx, t, cs, job, podsByStatus{ + Terminating: ptr.To[int32](2), + Active: 0, + Ready: ptr.To[int32](0), + }) + t.Logf("Pods are in Terminating state") + + // Update the job's pod template with new resources + t.Logf("Updating job resources") + _, err = updateJob(ctx, jobClient, job.Name, func(j *batchv1.Job) { + for i := range j.Spec.Template.Spec.Containers { + j.Spec.Template.Spec.Containers[i].Resources = v1.ResourceRequirements{ + Requests: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("300m"), + v1.ResourceMemory: resource.MustParse("384Mi"), + }, + Limits: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("600m"), + v1.ResourceMemory: resource.MustParse("768Mi"), + }, + } + } + }) + if err != nil { + t.Fatalf("Failed to update Job resources: %v", err) + } + t.Logf("Job resources updated") + + // Wait a bit to ensure the controller has time to process the update + // With PodReplacementPolicy=Failed, no new pods should be created while old pods are terminating + time.Sleep(sleepDurationForControllerLatency) + + // Verify that no new pods are created while old pods are terminating + validateJobsPodsStatusOnly(ctx, t, cs, job, podsByStatus{ + Terminating: ptr.To[int32](2), + Active: 0, + Ready: ptr.To[int32](0), + }) + t.Logf("Verified: No new pods created while old pods are terminating") + + // Remove the finalizers to allow pods to finish terminating + t.Logf("Removing finalizers from terminating pods") + err = removePodsFinalizers(ctx, cs, ns.Name, []string{blockDeletionFinalizerForTest}) + if err != nil { + t.Fatalf("Failed to remove finalizers: %v", err) + } + + // Wait for pods to be deleted and verify terminating count drops to 0 + err = wait.PollUntilContextTimeout(ctx, time.Second, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) { + updatedJob, err := jobClient.Get(ctx, job.Name, metav1.GetOptions{}) + if err != nil { + return false, err + } + return updatedJob.Status.Terminating == nil || *updatedJob.Status.Terminating == 0, nil + }) + if err != nil { + t.Fatalf("Failed to wait for terminating pods to be removed: %v", err) + } + t.Logf("Terminating pods have been removed") + + // Unsuspend the job + t.Logf("Unsuspending the job") + _, err = updateJob(ctx, jobClient, job.Name, func(j *batchv1.Job) { + j.Spec.Suspend = ptr.To(false) + }) + if err != nil { + t.Fatalf("Failed to unsuspend Job: %v", err) + } + + // Wait for new pods to be created + waitForPodsToBeActive(ctx, t, jobClient, 2, job) + t.Logf("New pods are active") + + // Verify that the new pods have the updated resources + pods, err := getJobPods(ctx, t, cs, job, func(s v1.PodStatus) bool { return true }) + if err != nil { + t.Fatalf("Failed to get Job pods: %v", err) + } + if len(pods) != 2 { + t.Fatalf("Expected 2 pods, got %d", len(pods)) + } + + expectedResources := v1.ResourceRequirements{ + Requests: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("300m"), + v1.ResourceMemory: resource.MustParse("384Mi"), + }, + Limits: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("600m"), + v1.ResourceMemory: resource.MustParse("768Mi"), + }, + } + + for _, pod := range pods { + for i := range pod.Spec.Containers { + if diff := cmp.Diff(expectedResources, pod.Spec.Containers[i].Resources); diff != "" { + t.Errorf("Unexpected resources in pod %s for container %q (-want +got):\n%s", pod.Name, pod.Spec.Containers[i].Name, diff) + } + } + } + t.Logf("Verified: New pods have the updated resources") +}