mirror of
https://github.com/k3s-io/kubernetes.git
synced 2026-07-18 04:54:54 +00:00
Merge pull request #135381 from kannon92/mutable-pod-replacement-policy
[KEP-5440]: Add integration test for MutablePodResourcesForSuspendedJobs with Pod Replacement Policy = Failed
This commit is contained in:
Kubernetes Prow Robot
GitHub
@@ -5293,3 +5293,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")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user