Merge pull request #98956 from gjkim42/fix-race-KillPod

kubelet: Fix race when KillPod followed by IsPodPendingTermination

pkg/kubelet/kubelet_pods.go

@@ -1143,7 +1143,7 @@ func (kl *Kubelet) HandlePodCleanups() error {
 // PodKiller handles requests for killing pods
 type PodKiller interface {
 	// KillPod receives pod speficier representing the pod to kill
-	KillPod(pair *kubecontainer.PodPair)
+	KillPod(podPair *kubecontainer.PodPair)
 	// PerformPodKillingWork performs the actual pod killing work via calling CRI
 	// It returns after its Close() func is called and all outstanding pod killing requests are served
 	PerformPodKillingWork()
@@ -1221,10 +1221,9 @@ func (pk *podKillerWithChannel) markPodTerminated(uid string) {
 	delete(pk.podTerminationMap, uid)
 }
 
-// checkAndMarkPodPendingTerminationByPod checks to see if the pod is being
+// KillPod sends pod killing request to the killer after marks the pod
-// killed and returns true if it is, otherwise the pod is added to the map and
+// unless the given pod has been marked to be killed
-// returns false
+func (pk *podKillerWithChannel) KillPod(podPair *kubecontainer.PodPair) {
-func (pk *podKillerWithChannel) checkAndMarkPodPendingTerminationByPod(podPair *kubecontainer.PodPair) bool {
 	pk.podKillingLock.Lock()
 	defer pk.podKillingLock.Unlock()
 	var apiPodExists bool
@@ -1255,9 +1254,10 @@ func (pk *podKillerWithChannel) checkAndMarkPodPendingTerminationByPod(podPair *
 		} else {
 			klog.V(4).Infof("running pod %q is pending termination", podPair.RunningPod.ID)
 		}
-		return true
+		return
 	}
-	return false
+	// Limit to one request per pod
+	pk.podKillingCh <- podPair
 }
 
 // Close closes the channel through which requests are delivered
@@ -1265,20 +1265,10 @@ func (pk *podKillerWithChannel) Close() {
 	close(pk.podKillingCh)
 }
 
-// KillPod sends pod killing request to the killer
-func (pk *podKillerWithChannel) KillPod(pair *kubecontainer.PodPair) {
-	pk.podKillingCh <- pair
-}
-
 // PerformPodKillingWork launches a goroutine to kill a pod received from the channel if
 // another goroutine isn't already in action.
 func (pk *podKillerWithChannel) PerformPodKillingWork() {
 	for podPair := range pk.podKillingCh {
-		if pk.checkAndMarkPodPendingTerminationByPod(podPair) {
-			// Pod is already being killed
-			continue
-		}
-
 		runningPod := podPair.RunningPod
 		apiPod := podPair.APIPod
 

pkg/kubelet/kubelet_test.go

@@ -444,6 +444,9 @@ func TestSyncPodsDeletesWhenSourcesAreReadyPerQOS(t *testing.T) {
 	// assert that unwanted pods were killed
 	fakeRuntime.AssertKilledPods([]string{"12345678"})
 
+	// simulate Runtime.KillPod
+	fakeRuntime.PodList = nil
+
 	kubelet.HandlePodCleanups()
 	kubelet.HandlePodCleanups()
 	kubelet.HandlePodCleanups()
@@ -500,6 +503,39 @@ func TestSyncPodsDeletesWhenSourcesAreReady(t *testing.T) {
 	fakeRuntime.AssertKilledPods([]string{"12345678"})
 }
 
+func TestKillPodFollwedByIsPodPendingTermination(t *testing.T) {
+	testKubelet := newTestKubelet(t, false /* controllerAttachDetachEnabled */)
+	defer testKubelet.Cleanup()
+	defer testKubelet.kubelet.podKiller.Close()
+	go testKubelet.kubelet.podKiller.PerformPodKillingWork()
+
+	pod := &kubecontainer.Pod{
+		ID:        "12345678",
+		Name:      "foo",
+		Namespace: "new",
+		Containers: []*kubecontainer.Container{
+			{Name: "bar"},
+		},
+	}
+
+	fakeRuntime := testKubelet.fakeRuntime
+	fakeContainerManager := testKubelet.fakeContainerManager
+	fakeContainerManager.PodContainerManager.AddPodFromCgroups(pod) // add pod to mock cgroup
+	fakeRuntime.PodList = []*containertest.FakePod{
+		{Pod: pod},
+	}
+
+	kl := testKubelet.kubelet
+	kl.podKiller.KillPod(&kubecontainer.PodPair{
+		APIPod:     nil,
+		RunningPod: pod,
+	})
+
+	if !(kl.podKiller.IsPodPendingTerminationByUID(pod.ID) || fakeRuntime.AssertKilledPods([]string{"12345678"}) == nil) {
+		t.Fatal("Race condition: When KillPod is complete, the pod should be pending termination or be killed")
+	}
+}
+
 type testNodeLister struct {
 	nodes []*v1.Node
 }