fix(controller): skip reaping succeeded Job pods

internal/controller/pod_placement_controller_test.go

@@ -960,3 +960,44 @@ func TestReap_NonNebulaTerminalPodIsIgnored(t *testing.T) {
 		t.Fatal("must not reap a non-Nebula Pod")
 	}
 }
+
+func TestReap_SucceededJobPodIsKept(t *testing.T) {
+	// Deleting a Succeeded Job pod makes the Job lose its success record and read as Failed.
+	pod := terminalOwnedPod("p1", "default", "uid-1", corev1.PodSucceeded, true)
+	pod.OwnerReferences[0].APIVersion = "batch/v1"
+	pod.OwnerReferences[0].Kind = "Job"
+	r, c := newPlacementReconciler(t, []client.Object{pod})
+
+	reconcilePod(t, r, "default", "p1")
+
+	if !podPresent(c, "default", "p1") {
+		t.Fatal("expected a Succeeded Job Pod to be kept")
+	}
+}
+
+func TestReap_FailedJobPodIsDeleted(t *testing.T) {
+	pod := terminalOwnedPod("p1", "default", "uid-1", corev1.PodFailed, true)
+	pod.OwnerReferences[0].APIVersion = "batch/v1"
+	pod.OwnerReferences[0].Kind = "Job"
+	r, c := newPlacementReconciler(t, []client.Object{pod})
+
+	reconcilePod(t, r, "default", "p1")
+
+	if podPresent(c, "default", "p1") {
+		t.Fatal("expected a Failed Job Pod to be deleted")
+	}
+}
+
+func TestReap_SucceededPodOfNonBatchJobIsDeleted(t *testing.T) {
+	// Only a batch Job is kept; a "Job" kind from another API group is not.
+	pod := terminalOwnedPod("p1", "default", "uid-1", corev1.PodSucceeded, true)
+	pod.OwnerReferences[0].APIVersion = "example.com/v1"
+	pod.OwnerReferences[0].Kind = "Job"
+	r, c := newPlacementReconciler(t, []client.Object{pod})
+
+	reconcilePod(t, r, "default", "p1")
+
+	if podPresent(c, "default", "p1") {
+		t.Fatal("expected a Succeeded Pod of a non-batch Job to be deleted")
+	}
+}

internal/controller/pod_placement_helpers.go

@@ -23,9 +23,11 @@
 	"strings"
 	"time"
 
+	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -418,6 +420,7 @@
 //   - terminal phase: only Failed/Succeeded (a live Pod is never touched).
 //   - controller-owned: only Pods a ReplicaSet/Job will recreate; a bare Pod is
 //     left as an inspectable record.
+//   - not a succeeded Job Pod: the Job counts it as a completion.
 //
 // Delete is UID-pinned so a Pod already replaced by a same-name recreate is not
 // clobbered, and a NotFound (already gone) is treated as success.
@@ -434,11 +437,14 @@
 	if !isControllerOwned(pod) {
 		return false, nil // bare Pod: leave it as a record
 	}
+	if isSucceededPodOwnedByJob(pod) {
+		return false, nil // the Job counts it as a completion; deleting it loses that
+	}
 
 	preconditions := metav1.Preconditions{UID: &pod.UID}
 	if err := r.Delete(ctx, pod, &client.DeleteOptions{Preconditions: &preconditions}); err != nil {
 		return false, client.IgnoreNotFound(err)
 	}
 	return true, nil
 }
 
@@ -448,6 +454,20 @@
 	return metav1.GetControllerOf(pod) != nil
 }
 
+// isSucceededPodOwnedByJob reports whether the Pod has succeeded and is
+// controlled by a batch Job.
+func isSucceededPodOwnedByJob(pod *corev1.Pod) bool {
+	if pod.Status.Phase != corev1.PodSucceeded {
+		return false
+	}
+	owner := metav1.GetControllerOf(pod)
+	if owner == nil || owner.Kind != "Job" {
+		return false
+	}
+	gv, err := schema.ParseGroupVersion(owner.APIVersion)
+	return err == nil && gv.Group == batchv1.GroupName
+}
+
 // SetupWithManager wires the controller. It watches Pods; NodePool edits are not
 // watched because a Pod already gated will re-reconcile on the periodic resync,
 // and a newly-created Pod always triggers its own event.