Add pod to dsw if termination is not completed during reconstruction #issues/113979

pkg/kubelet/volumemanager/populator/desired_state_of_world_populator.go

@@ -205,10 +205,18 @@ func (dswp *desiredStateOfWorldPopulator) findAndAddNewPods() {
 	}
 
 	for _, pod := range dswp.podManager.GetPods() {
-		if dswp.podStateProvider.ShouldPodContainersBeTerminating(pod.UID) {
+		// Keep consistency of adding pod during reconstruction
+		if dswp.hasAddedPods && dswp.podStateProvider.ShouldPodContainersBeTerminating(pod.UID) {
 			// Do not (re)add volumes for pods that can't also be starting containers
 			continue
 		}
+
+		if !dswp.hasAddedPods && dswp.podStateProvider.ShouldPodRuntimeBeRemoved(pod.UID) {
+			// When kubelet restarts, we need to add pods to dsw if there is a possibility
+			// that the container may still be running
+			continue
+		}
+
 		dswp.processPodVolumes(pod, mountedVolumesForPod)
 	}
 }

pkg/kubelet/volumemanager/populator/desired_state_of_world_populator_test.go

@@ -46,6 +46,12 @@ import (
 	"k8s.io/kubernetes/pkg/volume/util/types"
 )
 
+const (
+	Removed     string = "removed"
+	Terminating string = "terminating"
+	Other       string = "other"
+)
+
 func pluginPVOmittingClient(dswp *desiredStateOfWorldPopulator) {
 	fakeClient := &fake.Clientset{}
 	fakeClient.AddReactor("get", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) {
@@ -57,7 +63,7 @@ func pluginPVOmittingClient(dswp *desiredStateOfWorldPopulator) {
 	dswp.kubeClient = fakeClient
 }
 
-func prepareDswpWithVolume(t *testing.T) (*desiredStateOfWorldPopulator, kubepod.Manager) {
+func prepareDswpWithVolume(t *testing.T) (*desiredStateOfWorldPopulator, kubepod.Manager, *fakePodStateProvider) {
 	// create dswp
 	mode := v1.PersistentVolumeFilesystem
 	pv := &v1.PersistentVolume{
@@ -77,8 +83,8 @@ func prepareDswpWithVolume(t *testing.T) (*desiredStateOfWorldPopulator, kubepod
 			Phase: v1.ClaimBound,
 		},
 	}
-	dswp, fakePodManager, _, _, _ := createDswpWithVolume(t, pv, pvc)
+	dswp, fakePodManager, _, _, fakePodStateProvider := createDswpWithVolume(t, pv, pvc)
-	return dswp, fakePodManager
+	return dswp, fakePodManager, fakePodStateProvider
 }
 
 func TestFindAndAddNewPods_WithRescontructedVolume(t *testing.T) {
@@ -86,7 +92,7 @@ func TestFindAndAddNewPods_WithRescontructedVolume(t *testing.T) {
 	// (i.e. with SELinuxMountReadWriteOncePod disabled)
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SELinuxMountReadWriteOncePod, false)()
 	// create dswp
-	dswp, fakePodManager := prepareDswpWithVolume(t)
+	dswp, fakePodManager, _ := prepareDswpWithVolume(t)
 
 	// create pod
 	fakeOuterVolumeName := "dswp-test-volume-name"
@@ -142,6 +148,9 @@ func TestFindAndAddNewPods_WithRescontructedVolume(t *testing.T) {
 			break
 		}
 	}
+	if dswp.hasAddedPods {
+		t.Fatalf("HasAddedPod should be false but it is true")
+	}
 	if !found {
 		t.Fatalf(
 			"Could not found pod volume %v in the list of actual state of world volumes to mount.",
@@ -150,9 +159,103 @@ func TestFindAndAddNewPods_WithRescontructedVolume(t *testing.T) {
 
 }
 
+func TestFindAndAddNewPods_WithDifferentConditions(t *testing.T) {
+	tests := []struct {
+		desc          string
+		hasAddedPods  bool
+		podState      string
+		expectedFound bool // Found pod is added to DSW
+	}{
+		{
+			desc:          "HasAddedPods is false, ShouldPodRuntimeBeRemoved and ShouldPodContainerBeTerminating are both true",
+			hasAddedPods:  false,
+			podState:      Removed,
+			expectedFound: false, // Pod should not be added to DSW
+		},
+		{
+			desc:          "HasAddedPods is false, ShouldPodRuntimeBeRemoved is false, ShouldPodContainerBeTerminating is true",
+			hasAddedPods:  false,
+			podState:      Terminating,
+			expectedFound: true, // Pod should be added to DSW
+		},
+		{
+			desc:          "HasAddedPods is false, other condition",
+			hasAddedPods:  false,
+			podState:      Other,
+			expectedFound: true, // Pod should be added to DSW
+		},
+		{
+			desc:          "HasAddedPods is true, ShouldPodRuntimeBeRemoved is false, ShouldPodContainerBeTerminating is true",
+			hasAddedPods:  true,
+			podState:      Terminating,
+			expectedFound: false, // Pod should not be added to DSW
+		},
+		{
+			desc:          "HasAddedPods is true, ShouldPodRuntimeBeRemoved and ShouldPodContainerBeTerminating are both true",
+			hasAddedPods:  true,
+			podState:      Removed,
+			expectedFound: false, // Pod should not be added to DSW
+		},
+		{
+			desc:          "HasAddedPods is true, other condition",
+			hasAddedPods:  true,
+			podState:      Other,
+			expectedFound: true, // Pod should be added to DSW
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.desc, func(t *testing.T) {
+			// create dswp
+			dswp, fakePodManager, fakePodState := prepareDswpWithVolume(t)
+
+			// create pod
+			containers := []v1.Container{
+				{
+					VolumeMounts: []v1.VolumeMount{
+						{
+							Name:      "dswp-test-volume-name",
+							MountPath: "/mnt",
+						},
+					},
+				},
+			}
+			pod := createPodWithVolume("dswp-test-pod", "dswp-test-volume-name", "file-bound", containers)
+
+			fakePodManager.AddPod(pod)
+
+			switch tc.podState {
+			case Removed:
+				fakePodState.removed = map[kubetypes.UID]struct{}{pod.UID: {}}
+			case Terminating:
+				fakePodState.terminating = map[kubetypes.UID]struct{}{pod.UID: {}}
+			case Other:
+				break
+			}
+
+			dswp.hasAddedPods = tc.hasAddedPods
+			// Action
+			dswp.findAndAddNewPods()
+
+			// Verify
+			podsInDSW := dswp.desiredStateOfWorld.GetPods()
+			found := false
+			if podsInDSW[types.UniquePodName(pod.UID)] {
+				found = true
+			}
+
+			if found != tc.expectedFound {
+				t.Fatalf(
+					"Pod with uid %v has expectedFound value %v in pods in DSW %v",
+					pod.UID, tc.expectedFound, podsInDSW)
+			}
+		})
+	}
+}
+
 func TestFindAndAddNewPods_WithReprocessPodAndVolumeRetrievalError(t *testing.T) {
 	// create dswp
-	dswp, fakePodManager := prepareDswpWithVolume(t)
+	dswp, fakePodManager, _ := prepareDswpWithVolume(t)
 
 	// create pod
 	containers := []v1.Container{
@@ -189,7 +292,7 @@ func TestFindAndAddNewPods_WithReprocessPodAndVolumeRetrievalError(t *testing.T)
 
 func TestFindAndAddNewPods_WithVolumeRetrievalError(t *testing.T) {
 	// create dswp
-	dswp, fakePodManager := prepareDswpWithVolume(t)
+	dswp, fakePodManager, _ := prepareDswpWithVolume(t)
 
 	pluginPVOmittingClient(dswp)
 
@@ -1477,6 +1580,10 @@ type fakePodStateProvider struct {
 
 func (p *fakePodStateProvider) ShouldPodContainersBeTerminating(uid kubetypes.UID) bool {
 	_, ok := p.terminating[uid]
+	// if ShouldPodRuntimeBeRemoved returns true, ShouldPodContainerBeTerminating should also return true
+	if !ok {
+		_, ok = p.removed[uid]
+	}
 	return ok
 }