Skip to content

Commit 66ec7fc

Browse files
cgetzencgetzen
committed
fix: pod deletion race condition
1 parent 89d49fd commit 66ec7fc

2 files changed

Lines changed: 135 additions & 14 deletions

File tree

‎internal/controller/pod/pod_sync.go‎

Lines changed: 22 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ func (r *PodReconciler) syncKubernetes(ctx context.Context, req reconcile.Reques
9292
}
9393

9494
// syncSlurm reconciles the Slurm Job with Kubernetes Pods.
95-
// It will terminate the job corresponding to a terminat[ed,ing] pod.
95+
// It will terminate the job corresponding to a terminated pod.
9696
func (r *PodReconciler) syncSlurm(ctx context.Context, req reconcile.Request) error {
9797
logger := log.FromContext(ctx)
9898
podKey := req.String()
@@ -106,8 +106,8 @@ func (r *PodReconciler) syncSlurm(ctx context.Context, req reconcile.Request) er
106106
return err
107107
}
108108

109-
if pod.DeletionTimestamp == nil && !podv1.IsPodTerminal(pod) {
110-
logger.V(2).Info("Pod is not terminated or terminating, skipping", "pod", podKey)
109+
if !podv1.IsPodTerminal(pod) {
110+
logger.V(2).Info("Pod is not terminated, skipping", "pod", podKey)
111111
return nil
112112
}
113113

@@ -122,31 +122,39 @@ func (r *PodReconciler) syncSlurm(ctx context.Context, req reconcile.Request) er
122122
return err
123123
}
124124

125-
// If there are no pods labeled with this jobId the Slurm Job may
125+
// If there are no non-terminated pods labeled with this jobId the Slurm Job may
126126
// be terminated.
127-
activePods := 0
127+
nonTerminalPods := 0
128+
terminatingPods := 0
128129
for _, p := range podList.Items {
129-
if p.DeletionTimestamp == nil && !podv1.IsPodTerminal(&p) {
130-
activePods++
130+
if podv1.IsPodTerminal(&p) {
131+
continue
132+
}
133+
nonTerminalPods++
134+
if p.DeletionTimestamp != nil {
135+
terminatingPods++
131136
}
132137
}
133-
if activePods == 0 {
134-
jobId := slurmjobir.ParseSlurmJobId(pod.Labels[wellknown.LabelPlaceholderJobId])
138+
jobId := slurmjobir.ParseSlurmJobId(pod.Labels[wellknown.LabelPlaceholderJobId])
139+
if nonTerminalPods == 0 {
135140
logger.Info("Terminate Slurm Job for Pod", "pod", klog.KObj(pod), "jobId", jobId)
136141
if err := r.slurmControl.TerminateJob(ctx, jobId); err != nil {
137142
logger.Error(err, "failed to terminate Slurm Job without corresponding Pod",
138143
"jobId", jobId, "pod", podKey)
139144
return err
140145
}
146+
} else if terminatingPods > 0 {
147+
logger.V(4).Info("Retaining Slurm Job until non-terminating Pods complete",
148+
"jobId", jobId, "nonTerminalPods", nonTerminalPods, "terminatingPods", terminatingPods)
141149
}
142150

143151
return nil
144152
}
145153

146154
// prepareTerminalPod will remove the finalizer and resource claims from the pod
147-
// if it is to be deleted. This is done to ensure syncSlurm is able to get the
148-
// pod labels to determine if the pod has a placeholder JobId and cleanup
149-
// resource claims that were generated by slurm-bridge.
155+
// once the pod reaches a terminal phase. This is done to ensure syncSlurm is
156+
// able to get the pod labels to determine if the pod has a placeholder JobId
157+
// and cleanup resource claims that were generated by slurm-bridge.
150158
func (r *PodReconciler) prepareTerminalPod(ctx context.Context, req reconcile.Request) error {
151159
logger := log.FromContext(ctx)
152160
podKey := req.String()
@@ -160,8 +168,8 @@ func (r *PodReconciler) prepareTerminalPod(ctx context.Context, req reconcile.Re
160168
return err
161169
}
162170

163-
if pod.DeletionTimestamp == nil && !podv1.IsPodTerminal(pod) {
164-
logger.V(2).Info("Pod is not terminated or terminating, skipping", "pod", podKey)
171+
if !podv1.IsPodTerminal(pod) {
172+
logger.V(2).Info("Pod is not terminated, skipping", "pod", podKey)
165173
return nil
166174
}
167175

‎internal/controller/pod/pod_sync_test.go‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,19 @@ func newPod(name string, jobId int32) *corev1.Pod {
6666
}
6767
}
6868

69+
func newTerminatingPod(name string, jobId int32) *corev1.Pod {
70+
pod := newPod(name, jobId)
71+
now := metav1.Now()
72+
pod.DeletionTimestamp = &now
73+
return pod
74+
}
75+
76+
func newTerminalPod(name string, jobId int32) *corev1.Pod {
77+
pod := newPod(name, jobId)
78+
pod.Status.Phase = corev1.PodSucceeded
79+
return pod
80+
}
81+
6982
func newRequest(name string) ctrl.Request {
7083
return ctrl.Request{
7184
NamespacedName: types.NamespacedName{
@@ -225,6 +238,65 @@ var _ = Describe("syncSlurm()", func() {
225238
Expect(exists).To(BeTrue())
226239
})
227240

241+
It("Should not terminate the job for a terminating pod", func() {
242+
By("Setting the pod as terminating but non-terminal")
243+
key := types.NamespacedName{Namespace: corev1.NamespaceDefault, Name: podName}
244+
pod := &corev1.Pod{}
245+
err := controller.Get(ctx, key, pod)
246+
Expect(apierrors.IsNotFound(err)).ToNot(BeTrue())
247+
now := metav1.Now()
248+
pod.DeletionTimestamp = &now
249+
err = controller.Update(ctx, pod)
250+
Expect(err).NotTo(HaveOccurred())
251+
252+
By("Reconciling")
253+
err = controller.syncSlurm(ctx, req)
254+
Expect(err).NotTo(HaveOccurred())
255+
256+
By("Check job is still running")
257+
exists, err := controller.slurmControl.IsJobRunning(ctx, pod)
258+
Expect(err).ToNot(HaveOccurred())
259+
Expect(exists).To(BeTrue())
260+
})
261+
262+
It("Should not terminate the job while another pod is terminating", func() {
263+
By("Creating a terminal pod and a non-terminal terminating pod for the same Slurm job")
264+
terminalPod := newTerminalPod("foo-terminal", jobId)
265+
terminatingPod := newTerminatingPod("foo-terminating", jobId)
266+
podList := &corev1.PodList{
267+
Items: []corev1.Pod{*terminalPod, *terminatingPod},
268+
}
269+
jobList := &slurmtypes.V0044JobInfoList{
270+
Items: []slurmtypes.V0044JobInfo{
271+
{
272+
V0044JobInfo: api.V0044JobInfo{
273+
JobId: ptr.To(jobId),
274+
JobState: &[]api.V0044JobInfoJobState{api.V0044JobInfoJobStateRUNNING},
275+
AdminComment: ptr.To(newPlaceholderInfo("foo-terminal").ToString()),
276+
},
277+
},
278+
},
279+
}
280+
c := slurmclientfake.NewClientBuilder().WithLists(jobList).Build()
281+
localController := &PodReconciler{
282+
Client: fake.NewFakeClient(podList),
283+
Scheme: scheme.Scheme,
284+
SlurmClient: c,
285+
EventCh: make(chan event.GenericEvent, 5),
286+
slurmControl: slurmcontrol.NewControl(c),
287+
eventRecorder: record.NewFakeRecorder(10),
288+
}
289+
290+
By("Reconciling the terminal pod")
291+
err := localController.syncSlurm(ctx, newRequest("foo-terminal"))
292+
Expect(err).NotTo(HaveOccurred())
293+
294+
By("Check job is still running because a non-terminal pod still exists")
295+
exists, err := localController.slurmControl.IsJobRunning(ctx, terminatingPod)
296+
Expect(err).ToNot(HaveOccurred())
297+
Expect(exists).To(BeTrue())
298+
})
299+
228300
It("Should terminate the job", func() {
229301
By("Terminating the corresponding Slurm job")
230302
key := types.NamespacedName{Namespace: corev1.NamespaceDefault, Name: "bar"}
@@ -402,5 +474,46 @@ var _ = Describe("prepareTerminalPod()", func() {
402474
err = controller.Get(ctx, claimKey, claim)
403475
Expect(apierrors.IsNotFound(err)).To(BeTrue())
404476
})
477+
478+
It("Should not remove finalizer for terminating pod", func() {
479+
By("Creating a non-terminal terminating pod with finalizer and claim")
480+
podName := "terminating"
481+
terminatingPod := newTerminatingPod(podName, 1)
482+
terminatingPod.Finalizers = []string{wellknown.FinalizerScheduler}
483+
terminatingPod.Status.ExtendedResourceClaimStatus = &corev1.PodExtendedResourceClaimStatus{
484+
ResourceClaimName: "terminating-claim",
485+
}
486+
claim := &resourcev1.ResourceClaim{
487+
ObjectMeta: metav1.ObjectMeta{
488+
Name: "terminating-claim",
489+
Namespace: metav1.NamespaceDefault,
490+
},
491+
}
492+
localController := &PodReconciler{
493+
Client: fake.NewFakeClient(terminatingPod, claim),
494+
Scheme: scheme.Scheme,
495+
SlurmClient: slurmclientfake.NewFakeClient(),
496+
EventCh: make(chan event.GenericEvent, 5),
497+
slurmControl: slurmcontrol.NewControl(slurmclientfake.NewFakeClient()),
498+
eventRecorder: record.NewFakeRecorder(10),
499+
}
500+
501+
By("Reconciling")
502+
err := localController.prepareTerminalPod(ctx, newRequest(podName))
503+
Expect(err).NotTo(HaveOccurred())
504+
505+
By("Check finalizer still exists")
506+
podKey := types.NamespacedName{Namespace: metav1.NamespaceDefault, Name: podName}
507+
pod := &corev1.Pod{}
508+
err = localController.Get(ctx, podKey, pod)
509+
Expect(apierrors.IsNotFound(err)).ToNot(BeTrue())
510+
Expect(pod.ObjectMeta.Finalizers).To(Equal([]string{wellknown.FinalizerScheduler}))
511+
512+
By("Check ResourceClaim still exists")
513+
claimKey := types.NamespacedName{Namespace: metav1.NamespaceDefault, Name: "terminating-claim"}
514+
gotClaim := &resourcev1.ResourceClaim{}
515+
err = localController.Get(ctx, claimKey, gotClaim)
516+
Expect(err).NotTo(HaveOccurred())
517+
})
405518
})
406519
})

0 commit comments

Comments
 (0)