Merge pull request #45286 from gnufied/fix-terminated-pods-detach

Automatic merge from submit-queue

detach the volume when pod is terminated

When pods are terminated we should detach the volume. 

Fixes https://github.com/kubernetes/kubernetes/issues/45191

**Release note**:
```
Detach the volume when pods are terminated.  
```

cmd/kube-controller-manager/app/controllermanager.go

@@ -569,8 +569,7 @@ func StartControllers(controllers map[string]InitFunc, s *options.CMServer, root
 				cloud,
 				ProbeAttachableVolumePlugins(s.VolumeConfiguration),
 				s.DisableAttachDetachReconcilerSync,
-				s.ReconcilerSyncLoopPeriod.Duration,
-			)
+				s.ReconcilerSyncLoopPeriod.Duration)
 		if attachDetachControllerErr != nil {
 			return fmt.Errorf("failed to start attach/detach controller: %v", attachDetachControllerErr)
 		}

hack/local-up-cluster.sh

@@ -208,6 +208,7 @@ CPU_CFS_QUOTA=${CPU_CFS_QUOTA:-true}
 ENABLE_HOSTPATH_PROVISIONER=${ENABLE_HOSTPATH_PROVISIONER:-"false"}
 CLAIM_BINDER_SYNC_PERIOD=${CLAIM_BINDER_SYNC_PERIOD:-"15s"} # current k8s default
 ENABLE_CONTROLLER_ATTACH_DETACH=${ENABLE_CONTROLLER_ATTACH_DETACH:-"true"} # current default
+KEEP_TERMINATED_POD_VOLUMES=${KEEP_TERMINATED_POD_VOLUMES:-"true"}
 # This is the default dir and filename where the apiserver will generate a self-signed cert
 # which should be able to be used as the CA to verify itself
 CERT_DIR=${CERT_DIR:-"/var/run/kubernetes"}
@@ -638,7 +639,7 @@ function start_kubelet {
         --enable-controller-attach-detach="${ENABLE_CONTROLLER_ATTACH_DETACH}" \
         --cgroups-per-qos=${CGROUPS_PER_QOS} \
         --cgroup-driver=${CGROUP_DRIVER} \
-        --keep-terminated-pod-volumes=true \
+        --keep-terminated-pod-volumes=${KEEP_TERMINATED_POD_VOLUMES} \
         --eviction-hard=${EVICTION_HARD} \
         --eviction-soft=${EVICTION_SOFT} \
         --eviction-pressure-transition-period=${EVICTION_PRESSURE_TRANSITION_PERIOD} \

pkg/controller/volume/attachdetach/attach_detach_controller.go

@@ -281,11 +281,7 @@ func (adc *attachDetachController) populateActualStateOfWorld() error {
 				continue
 			}
 			adc.processVolumesInUse(nodeName, node.Status.VolumesInUse, true /* forceUnmount */)
-			if _, exists := node.Annotations[volumehelper.ControllerManagedAttachAnnotation]; exists {
-				// Node specifies annotation indicating it should be managed by
-				// attach detach controller. Add it to desired state of world.
-				adc.desiredStateOfWorld.AddNode(types.NodeName(node.Name)) // Needed for DesiredStateOfWorld population
-			}
+			adc.addNodeToDswp(node, types.NodeName(node.Name))
 		}
 	}
 	return nil
@@ -385,7 +381,12 @@ func (adc *attachDetachController) podAdd(obj interface{}) {
 		return
 	}
 
-	util.ProcessPodVolumes(pod, true, /* addVolumes */
+	volumeActionFlag := util.DetermineVolumeAction(
+		pod,
+		adc.desiredStateOfWorld,
+		true /* default volume action */)
+
+	util.ProcessPodVolumes(pod, volumeActionFlag, /* addVolumes */
 		adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
 }
 
@@ -395,8 +396,22 @@ func (adc *attachDetachController) GetDesiredStateOfWorld() cache.DesiredStateOf
 }
 
 func (adc *attachDetachController) podUpdate(oldObj, newObj interface{}) {
-	// The flow for update is the same as add.
-	adc.podAdd(newObj)
+	pod, ok := newObj.(*v1.Pod)
+	if pod == nil || !ok {
+		return
+	}
+	if pod.Spec.NodeName == "" {
+		// Ignore pods without NodeName, indicating they are not scheduled.
+		return
+	}
+
+	volumeActionFlag := util.DetermineVolumeAction(
+		pod,
+		adc.desiredStateOfWorld,
+		true /* default volume action */)
+
+	util.ProcessPodVolumes(pod, volumeActionFlag, /* addVolumes */
+		adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
 }
 
 func (adc *attachDetachController) podDelete(obj interface{}) {
@@ -433,11 +448,7 @@ func (adc *attachDetachController) nodeUpdate(oldObj, newObj interface{}) {
 	}
 
 	nodeName := types.NodeName(node.Name)
-	if _, exists := node.Annotations[volumehelper.ControllerManagedAttachAnnotation]; exists {
-		// Node specifies annotation indicating it should be managed by attach
-		// detach controller. Add it to desired state of world.
-		adc.desiredStateOfWorld.AddNode(nodeName)
-	}
+	adc.addNodeToDswp(node, nodeName)
 	adc.processVolumesInUse(nodeName, node.Status.VolumesInUse, false /* forceUnmount */)
 }
 
@@ -539,3 +550,17 @@ func (adc *attachDetachController) GetSecretFunc() func(namespace, name string)
 		return nil, fmt.Errorf("GetSecret unsupported in attachDetachController")
 	}
 }
+
+func (adc *attachDetachController) addNodeToDswp(node *v1.Node, nodeName types.NodeName) {
+	if _, exists := node.Annotations[volumehelper.ControllerManagedAttachAnnotation]; exists {
+		keepTerminatedPodVolumes := false
+
+		if t, ok := node.Annotations[volumehelper.KeepTerminatedPodVolumesAnnotation]; ok {
+			keepTerminatedPodVolumes = (t == "true")
+		}
+
+		// Node specifies annotation indicating it should be managed by attach
+		// detach controller. Add it to desired state of world.
+		adc.desiredStateOfWorld.AddNode(nodeName, keepTerminatedPodVolumes)
+	}
+}

pkg/controller/volume/attachdetach/attach_detach_controller_test.go

@@ -213,6 +213,7 @@ func attachDetachRecoveryTestCase(t *testing.T, extraPods1 []*v1.Pod, extraPods2
 		plugins,
 		false,
 		time.Second*1)
+
 	if err != nil {
 		t.Fatalf("Run failed with error. Expected: <no error> Actual: <%v>", err)
 	}

pkg/controller/volume/attachdetach/cache/desired_state_of_world.go

@@ -46,7 +46,9 @@ type DesiredStateOfWorld interface {
 	// AddNode adds the given node to the list of nodes managed by the attach/
 	// detach controller.
 	// If the node already exists this is a no-op.
-	AddNode(nodeName k8stypes.NodeName)
+	// keepTerminatedPodVolumes is a property of the node that determines
+	// if for terminated pods volumes should be mounted and attached.
+	AddNode(nodeName k8stypes.NodeName, keepTerminatedPodVolumes bool)
 
 	// AddPod adds the given pod to the list of pods that reference the
 	// specified volume and is scheduled to the specified node.
@@ -95,6 +97,10 @@ type DesiredStateOfWorld interface {
 	// GetPodToAdd generates and returns a map of pods based on the current desired
 	// state of world
 	GetPodToAdd() map[types.UniquePodName]PodToAdd
+
+	// GetKeepTerminatedPodVolumesForNode determines if node wants volumes to be
+	// mounted and attached for terminated pods
+	GetKeepTerminatedPodVolumesForNode(k8stypes.NodeName) bool
 }
 
 // VolumeToAttach represents a volume that should be attached to a node.
@@ -144,6 +150,10 @@ type nodeManaged struct {
 	// attached to this node. The key in the map is the name of the volume and
 	// the value is a pod object containing more information about the volume.
 	volumesToAttach map[v1.UniqueVolumeName]volumeToAttach
+
+	// keepTerminatedPodVolumes determines if for terminated pods(on this node) - volumes
+	// should be kept mounted and attached.
+	keepTerminatedPodVolumes bool
 }
 
 // The volume object represents a volume that should be attached to a node.
@@ -173,14 +183,15 @@ type pod struct {
 	podObj *v1.Pod
 }
 
-func (dsw *desiredStateOfWorld) AddNode(nodeName k8stypes.NodeName) {
+func (dsw *desiredStateOfWorld) AddNode(nodeName k8stypes.NodeName, keepTerminatedPodVolumes bool) {
 	dsw.Lock()
 	defer dsw.Unlock()
 
 	if _, nodeExists := dsw.nodesManaged[nodeName]; !nodeExists {
 		dsw.nodesManaged[nodeName] = nodeManaged{
-			nodeName:        nodeName,
-			volumesToAttach: make(map[v1.UniqueVolumeName]volumeToAttach),
+			nodeName:                 nodeName,
+			volumesToAttach:          make(map[v1.UniqueVolumeName]volumeToAttach),
+			keepTerminatedPodVolumes: keepTerminatedPodVolumes,
 		}
 	}
 }
@@ -313,6 +324,21 @@ func (dsw *desiredStateOfWorld) VolumeExists(
 	return false
 }
 
+// GetKeepTerminatedPodVolumesForNode determines if node wants volumes to be
+// mounted and attached for terminated pods
+func (dsw *desiredStateOfWorld) GetKeepTerminatedPodVolumesForNode(nodeName k8stypes.NodeName) bool {
+	dsw.RLock()
+	defer dsw.RUnlock()
+
+	if nodeName == "" {
+		return false
+	}
+	if node, ok := dsw.nodesManaged[nodeName]; ok {
+		return node.keepTerminatedPodVolumes
+	}
+	return false
+}
+
 func (dsw *desiredStateOfWorld) GetVolumesToAttach() []VolumeToAttach {
 	dsw.RLock()
 	defer dsw.RUnlock()

pkg/controller/volume/attachdetach/cache/desired_state_of_world_test.go

@@ -35,7 +35,7 @@ func Test_AddNode_Positive_NewNode(t *testing.T) {
 	nodeName := k8stypes.NodeName("node-name")
 
 	// Act
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 
 	// Assert
 	nodeExists := dsw.NodeExists(nodeName)
@@ -60,7 +60,7 @@ func Test_AddNode_Positive_ExistingNode(t *testing.T) {
 	nodeName := k8stypes.NodeName("node-name")
 
 	// Act
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 
 	// Assert
 	nodeExists := dsw.NodeExists(nodeName)
@@ -69,7 +69,7 @@ func Test_AddNode_Positive_ExistingNode(t *testing.T) {
 	}
 
 	// Act
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 
 	// Assert
 	nodeExists = dsw.NodeExists(nodeName)
@@ -94,7 +94,7 @@ func Test_AddPod_Positive_NewPodNodeExistsVolumeDoesntExist(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	volumeExists := dsw.VolumeExists(volumeName, nodeName)
 	if volumeExists {
 		t.Fatalf(
@@ -142,7 +142,7 @@ func Test_AddPod_Positive_NewPodNodeExistsVolumeExists(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	volumeExists := dsw.VolumeExists(volumeName, nodeName)
 	if volumeExists {
 		t.Fatalf(
@@ -215,7 +215,7 @@ func Test_AddPod_Positive_PodExistsNodeExistsVolumeExists(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	volumeExists := dsw.VolumeExists(volumeName, nodeName)
 	if volumeExists {
 		t.Fatalf(
@@ -319,7 +319,7 @@ func Test_DeleteNode_Positive_NodeExists(t *testing.T) {
 	volumePluginMgr, _ := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := NewDesiredStateOfWorld(volumePluginMgr)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 
 	// Act
 	err := dsw.DeleteNode(nodeName)
@@ -375,7 +375,7 @@ func Test_DeleteNode_Negative_NodeExistsHasChildVolumes(t *testing.T) {
 	volumePluginMgr, _ := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := NewDesiredStateOfWorld(volumePluginMgr)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	podName := "pod-uid"
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
@@ -419,7 +419,7 @@ func Test_DeletePod_Positive_PodExistsNodeExistsVolumeExists(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	generatedVolumeName, podAddErr := dsw.AddPod(types.UniquePodName(podName), controllervolumetesting.NewPod(podName, podName), volumeSpec, nodeName)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -467,7 +467,7 @@ func Test_DeletePod_Positive_2PodsExistNodeExistsVolumesExist(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	generatedVolumeName1, pod1AddErr := dsw.AddPod(types.UniquePodName(pod1Name), controllervolumetesting.NewPod(pod1Name, pod1Name), volumeSpec, nodeName)
 	if pod1AddErr != nil {
 		t.Fatalf(
@@ -528,7 +528,7 @@ func Test_DeletePod_Positive_PodDoesNotExist(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	generatedVolumeName, pod1AddErr := dsw.AddPod(types.UniquePodName(pod1Name), controllervolumetesting.NewPod(pod1Name, pod1Name), volumeSpec, nodeName)
 	if pod1AddErr != nil {
 		t.Fatalf(
@@ -576,7 +576,7 @@ func Test_DeletePod_Positive_NodeDoesNotExist(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	node1Name := k8stypes.NodeName("node1-name")
-	dsw.AddNode(node1Name)
+	dsw.AddNode(node1Name, false /*keepTerminatedPodVolumes*/)
 	generatedVolumeName, podAddErr := dsw.AddPod(types.UniquePodName(podName), controllervolumetesting.NewPod(podName, podName), volumeSpec, node1Name)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -631,7 +631,7 @@ func Test_DeletePod_Positive_VolumeDoesNotExist(t *testing.T) {
 	volume1Name := v1.UniqueVolumeName("volume1-name")
 	volume1Spec := controllervolumetesting.GetTestVolumeSpec(string(volume1Name), volume1Name)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	generatedVolume1Name, podAddErr := dsw.AddPod(types.UniquePodName(podName), controllervolumetesting.NewPod(podName, podName), volume1Spec, nodeName)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -705,7 +705,7 @@ func Test_NodeExists_Positive_NodeDoesntExist(t *testing.T) {
 	volumePluginMgr, _ := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := NewDesiredStateOfWorld(volumePluginMgr)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 
 	// Act
 	nodeExists := dsw.NodeExists(nodeName)
@@ -729,7 +729,7 @@ func Test_VolumeExists_Positive_VolumeExistsNodeExists(t *testing.T) {
 	volumePluginMgr, _ := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := NewDesiredStateOfWorld(volumePluginMgr)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	podName := "pod-uid"
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
@@ -759,7 +759,7 @@ func Test_VolumeExists_Positive_VolumeDoesntExistNodeExists(t *testing.T) {
 	volumePluginMgr, _ := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := NewDesiredStateOfWorld(volumePluginMgr)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	podName := "pod-uid"
 	volume1Name := v1.UniqueVolumeName("volume1-name")
 	volume1Spec := controllervolumetesting.GetTestVolumeSpec(string(volume1Name), volume1Name)
@@ -836,8 +836,8 @@ func Test_GetVolumesToAttach_Positive_TwoNodes(t *testing.T) {
 	dsw := NewDesiredStateOfWorld(volumePluginMgr)
 	node1Name := k8stypes.NodeName("node1-name")
 	node2Name := k8stypes.NodeName("node2-name")
-	dsw.AddNode(node1Name)
-	dsw.AddNode(node2Name)
+	dsw.AddNode(node1Name, false /*keepTerminatedPodVolumes*/)
+	dsw.AddNode(node2Name, false /*keepTerminatedPodVolumes*/)
 
 	// Act
 	volumesToAttach := dsw.GetVolumesToAttach()
@@ -859,7 +859,7 @@ func Test_GetVolumesToAttach_Positive_TwoNodesOneVolumeEach(t *testing.T) {
 	pod1Name := "pod1-uid"
 	volume1Name := v1.UniqueVolumeName("volume1-name")
 	volume1Spec := controllervolumetesting.GetTestVolumeSpec(string(volume1Name), volume1Name)
-	dsw.AddNode(node1Name)
+	dsw.AddNode(node1Name, false /*keepTerminatedPodVolumes*/)
 	generatedVolume1Name, podAddErr := dsw.AddPod(types.UniquePodName(pod1Name), controllervolumetesting.NewPod(pod1Name, pod1Name), volume1Spec, node1Name)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -871,7 +871,7 @@ func Test_GetVolumesToAttach_Positive_TwoNodesOneVolumeEach(t *testing.T) {
 	pod2Name := "pod2-uid"
 	volume2Name := v1.UniqueVolumeName("volume2-name")
 	volume2Spec := controllervolumetesting.GetTestVolumeSpec(string(volume2Name), volume2Name)
-	dsw.AddNode(node2Name)
+	dsw.AddNode(node2Name, false /*keepTerminatedPodVolumes*/)
 	generatedVolume2Name, podAddErr := dsw.AddPod(types.UniquePodName(pod2Name), controllervolumetesting.NewPod(pod2Name, pod2Name), volume2Spec, node2Name)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -904,7 +904,7 @@ func Test_GetVolumesToAttach_Positive_TwoNodesOneVolumeEachExtraPod(t *testing.T
 	pod1Name := "pod1-uid"
 	volume1Name := v1.UniqueVolumeName("volume1-name")
 	volume1Spec := controllervolumetesting.GetTestVolumeSpec(string(volume1Name), volume1Name)
-	dsw.AddNode(node1Name)
+	dsw.AddNode(node1Name, false /*keepTerminatedPodVolumes*/)
 	generatedVolume1Name, podAddErr := dsw.AddPod(types.UniquePodName(pod1Name), controllervolumetesting.NewPod(pod1Name, pod1Name), volume1Spec, node1Name)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -916,7 +916,7 @@ func Test_GetVolumesToAttach_Positive_TwoNodesOneVolumeEachExtraPod(t *testing.T
 	pod2Name := "pod2-uid"
 	volume2Name := v1.UniqueVolumeName("volume2-name")
 	volume2Spec := controllervolumetesting.GetTestVolumeSpec(string(volume2Name), volume2Name)
-	dsw.AddNode(node2Name)
+	dsw.AddNode(node2Name, false /*keepTerminatedPodVolumes*/)
 	generatedVolume2Name, podAddErr := dsw.AddPod(types.UniquePodName(pod2Name), controllervolumetesting.NewPod(pod2Name, pod2Name), volume2Spec, node2Name)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -958,7 +958,7 @@ func Test_GetVolumesToAttach_Positive_TwoNodesThreeVolumes(t *testing.T) {
 	pod1Name := "pod1-uid"
 	volume1Name := v1.UniqueVolumeName("volume1-name")
 	volume1Spec := controllervolumetesting.GetTestVolumeSpec(string(volume1Name), volume1Name)
-	dsw.AddNode(node1Name)
+	dsw.AddNode(node1Name, false /*keepTerminatedPodVolumes*/)
 	generatedVolume1Name, podAddErr := dsw.AddPod(types.UniquePodName(pod1Name), controllervolumetesting.NewPod(pod1Name, pod1Name), volume1Spec, node1Name)
 	if podAddErr != nil {
 		t.Fatalf(
@@ -970,7 +970,7 @@ func Test_GetVolumesToAttach_Positive_TwoNodesThreeVolumes(t *testing.T) {
 	pod2aName := "pod2a-name"
 	volume2Name := v1.UniqueVolumeName("volume2-name")
 	volume2Spec := controllervolumetesting.GetTestVolumeSpec(string(volume2Name), volume2Name)
-	dsw.AddNode(node2Name)
+	dsw.AddNode(node2Name, false /*keepTerminatedPodVolumes*/)
 	generatedVolume2Name1, podAddErr := dsw.AddPod(types.UniquePodName(pod2aName), controllervolumetesting.NewPod(pod2aName, pod2aName), volume2Spec, node2Name)
 	if podAddErr != nil {
 		t.Fatalf(

pkg/controller/volume/attachdetach/populator/desired_state_of_world_populator.go

@@ -127,11 +127,18 @@ func (dswp *desiredStateOfWorldPopulator) findAndRemoveDeletedPods() {
 			glog.Errorf("podLister Get failed for pod %q (UID %q) with %v", dswPodKey, dswPodUID, err)
 			continue
 		default:
-			informerPodUID := volumehelper.GetUniquePodName(informerPod)
-			// Check whether the unique identifier of the pod from dsw matches the one retrieved from pod informer
-			if informerPodUID == dswPodUID {
-				glog.V(10).Infof("Verified pod %q (UID %q) from dsw exists in pod informer.", dswPodKey, dswPodUID)
-				continue
+			volumeActionFlag := util.DetermineVolumeAction(
+				informerPod,
+				dswp.desiredStateOfWorld,
+				true /* default volume action */)
+
+			if volumeActionFlag {
+				informerPodUID := volumehelper.GetUniquePodName(informerPod)
+				// Check whether the unique identifier of the pod from dsw matches the one retrieved from pod informer
+				if informerPodUID == dswPodUID {
+					glog.V(10).Infof("Verified pod %q (UID %q) from dsw exists in pod informer.", dswPodKey, dswPodUID)
+					continue
+				}
 			}
 		}
 

pkg/controller/volume/attachdetach/populator/desired_state_of_world_populator_test.go

@@ -84,7 +84,7 @@ func TestFindAndAddActivePods_FindAndRemoveDeletedPods(t *testing.T) {
 	}
 
 	//add the given node to the list of nodes managed by dsw
-	dswp.desiredStateOfWorld.AddNode(k8stypes.NodeName(pod.Spec.NodeName))
+	dswp.desiredStateOfWorld.AddNode(k8stypes.NodeName(pod.Spec.NodeName), false /*keepTerminatedPodVolumes*/)
 
 	dswp.findAndAddActivePods()
 

pkg/controller/volume/attachdetach/reconciler/reconciler_test.go

@@ -88,7 +88,7 @@ func Test_Run_Positive_OneDesiredVolumeAttach(t *testing.T) {
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	volumeExists := dsw.VolumeExists(volumeName, nodeName)
 	if volumeExists {
 		t.Fatalf(
@@ -134,7 +134,7 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithUnmountedVolume(t *te
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	volumeExists := dsw.VolumeExists(volumeName, nodeName)
 	if volumeExists {
 		t.Fatalf(
@@ -201,7 +201,7 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithMountedVolume(t *test
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	volumeExists := dsw.VolumeExists(volumeName, nodeName)
 	if volumeExists {
 		t.Fatalf(
@@ -268,7 +268,7 @@ func Test_Run_Negative_OneDesiredVolumeAttachThenDetachWithUnmountedVolumeUpdate
 	volumeName := v1.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
 	nodeName := k8stypes.NodeName("node-name")
-	dsw.AddNode(nodeName)
+	dsw.AddNode(nodeName, false /*keepTerminatedPodVolumes*/)
 	volumeExists := dsw.VolumeExists(volumeName, nodeName)
 	if volumeExists {
 		t.Fatalf(

pkg/controller/volume/attachdetach/util/util.go

@@ -159,6 +159,23 @@ func getPVSpecFromCache(name string, pvcReadOnly bool, expectedClaimUID types.UI
 	return volume.NewSpecFromPersistentVolume(clonedPV, pvcReadOnly), nil
 }
 
+// DetermineVolumeAction returns true if volume and pod needs to be added to dswp
+// and it returns false if volume and pod needs to be removed from dswp
+func DetermineVolumeAction(pod *v1.Pod, desiredStateOfWorld cache.DesiredStateOfWorld, defaultAction bool) bool {
+	if pod == nil || len(pod.Spec.Volumes) <= 0 {
+		return defaultAction
+	}
+	nodeName := types.NodeName(pod.Spec.NodeName)
+	keepTerminatedPodVolume := desiredStateOfWorld.GetKeepTerminatedPodVolumesForNode(nodeName)
+
+	if volumehelper.IsPodTerminated(pod, pod.Status) {
+		// if pod is terminate we let kubelet policy dictate if volume
+		// should be detached or not
+		return keepTerminatedPodVolume
+	}
+	return defaultAction
+}
+
 // ProcessPodVolumes processes the volumes in the given pod and adds them to the
 // desired state of the world if addVolumes is true, otherwise it removes them.
 func ProcessPodVolumes(pod *v1.Pod, addVolumes bool, desiredStateOfWorld cache.DesiredStateOfWorld, volumePluginMgr *volume.VolumePluginMgr, pvcLister corelisters.PersistentVolumeClaimLister, pvLister corelisters.PersistentVolumeLister) {

pkg/kubelet/kubelet_node_status.go

@@ -246,6 +246,14 @@ func (kl *Kubelet) initialNode() (*v1.Node, error) {
 		glog.Infof("Controller attach/detach is disabled for this node; Kubelet will attach and detach volumes")
 	}
 
+	if kl.kubeletConfiguration.KeepTerminatedPodVolumes {
+		if node.Annotations == nil {
+			node.Annotations = make(map[string]string)
+		}
+		glog.Infof("Setting node annotation to keep pod volumes of terminated pods attached to the node")
+		node.Annotations[volumehelper.KeepTerminatedPodVolumesAnnotation] = "true"
+	}
+
 	// @question: should this be place after the call to the cloud provider? which also applies labels
 	for k, v := range kl.nodeLabels {
 		if cv, found := node.ObjectMeta.Labels[k]; found {

pkg/volume/util/volumehelper/volumehelper.go

@@ -33,6 +33,10 @@ const (
 	// managed by the attach/detach controller
 	ControllerManagedAttachAnnotation string = "volumes.kubernetes.io/controller-managed-attach-detach"
 
+	// KeepTerminatedPodVolumesAnnotation is the key of the annotation on Node
+	// that decides if pod volumes are unmounted when pod is terminated
+	KeepTerminatedPodVolumesAnnotation string = "volumes.kubernetes.io/keep-terminated-pod-volumes"
+
 	// VolumeGidAnnotationKey is the of the annotation on the PersistentVolume
 	// object that specifies a supplemental GID.
 	VolumeGidAnnotationKey = "pv.beta.kubernetes.io/gid"

test/integration/volume/attach_detach_test.go

@@ -149,6 +149,149 @@ func TestPodDeletionWithDswp(t *testing.T) {
 	close(stopCh)
 }
 
+func TestPodUpdateWithWithADC(t *testing.T) {
+	_, server, closeFn := framework.RunAMaster(nil)
+	defer closeFn()
+	namespaceName := "test-pod-update"
+
+	node := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "node-sandbox",
+			Annotations: map[string]string{
+				volumehelper.ControllerManagedAttachAnnotation: "true",
+			},
+		},
+	}
+
+	ns := framework.CreateTestingNamespace(namespaceName, server, t)
+	defer framework.DeleteTestingNamespace(ns, server, t)
+
+	testClient, ctrl, informers := createAdClients(ns, t, server, defaultSyncPeriod)
+
+	pod := fakePodWithVol(namespaceName)
+	podStopCh := make(chan struct{})
+
+	if _, err := testClient.Core().Nodes().Create(node); err != nil {
+		t.Fatalf("Failed to created node : %v", err)
+	}
+
+	go informers.Core().V1().Nodes().Informer().Run(podStopCh)
+
+	if _, err := testClient.Core().Pods(ns.Name).Create(pod); err != nil {
+		t.Errorf("Failed to create pod : %v", err)
+	}
+
+	podInformer := informers.Core().V1().Pods().Informer()
+	go podInformer.Run(podStopCh)
+
+	// start controller loop
+	stopCh := make(chan struct{})
+	go informers.Core().V1().PersistentVolumeClaims().Informer().Run(stopCh)
+	go informers.Core().V1().PersistentVolumes().Informer().Run(stopCh)
+	go ctrl.Run(stopCh)
+
+	waitToObservePods(t, podInformer, 1)
+	podKey, err := cache.MetaNamespaceKeyFunc(pod)
+	if err != nil {
+		t.Fatalf("MetaNamespaceKeyFunc failed with : %v", err)
+	}
+
+	_, _, err = podInformer.GetStore().GetByKey(podKey)
+
+	if err != nil {
+		t.Fatalf("Pod not found in Pod Informer cache : %v", err)
+	}
+
+	waitForPodsInDSWP(t, ctrl.GetDesiredStateOfWorld())
+
+	pod.Status.Phase = v1.PodSucceeded
+
+	if _, err := testClient.Core().Pods(ns.Name).UpdateStatus(pod); err != nil {
+		t.Errorf("Failed to update pod : %v", err)
+	}
+
+	time.Sleep(20 * time.Second)
+	podsToAdd := ctrl.GetDesiredStateOfWorld().GetPodToAdd()
+	if len(podsToAdd) != 0 {
+		t.Fatalf("All pods should have been removed")
+	}
+
+	close(podStopCh)
+	close(stopCh)
+}
+
+func TestPodUpdateWithKeepTerminatedPodVolumes(t *testing.T) {
+	_, server, closeFn := framework.RunAMaster(nil)
+	defer closeFn()
+	namespaceName := "test-pod-update"
+
+	node := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "node-sandbox",
+			Annotations: map[string]string{
+				volumehelper.ControllerManagedAttachAnnotation:  "true",
+				volumehelper.KeepTerminatedPodVolumesAnnotation: "true",
+			},
+		},
+	}
+
+	ns := framework.CreateTestingNamespace(namespaceName, server, t)
+	defer framework.DeleteTestingNamespace(ns, server, t)
+
+	testClient, ctrl, informers := createAdClients(ns, t, server, defaultSyncPeriod)
+
+	pod := fakePodWithVol(namespaceName)
+	podStopCh := make(chan struct{})
+
+	if _, err := testClient.Core().Nodes().Create(node); err != nil {
+		t.Fatalf("Failed to created node : %v", err)
+	}
+
+	go informers.Core().V1().Nodes().Informer().Run(podStopCh)
+
+	if _, err := testClient.Core().Pods(ns.Name).Create(pod); err != nil {
+		t.Errorf("Failed to create pod : %v", err)
+	}
+
+	podInformer := informers.Core().V1().Pods().Informer()
+	go podInformer.Run(podStopCh)
+
+	// start controller loop
+	stopCh := make(chan struct{})
+	go informers.Core().V1().PersistentVolumeClaims().Informer().Run(stopCh)
+	go informers.Core().V1().PersistentVolumes().Informer().Run(stopCh)
+	go ctrl.Run(stopCh)
+
+	waitToObservePods(t, podInformer, 1)
+	podKey, err := cache.MetaNamespaceKeyFunc(pod)
+	if err != nil {
+		t.Fatalf("MetaNamespaceKeyFunc failed with : %v", err)
+	}
+
+	_, _, err = podInformer.GetStore().GetByKey(podKey)
+
+	if err != nil {
+		t.Fatalf("Pod not found in Pod Informer cache : %v", err)
+	}
+
+	waitForPodsInDSWP(t, ctrl.GetDesiredStateOfWorld())
+
+	pod.Status.Phase = v1.PodSucceeded
+
+	if _, err := testClient.Core().Pods(ns.Name).UpdateStatus(pod); err != nil {
+		t.Errorf("Failed to update pod : %v", err)
+	}
+
+	time.Sleep(20 * time.Second)
+	podsToAdd := ctrl.GetDesiredStateOfWorld().GetPodToAdd()
+	if len(podsToAdd) == 0 {
+		t.Fatalf("The pod should not be removed if KeepTerminatedPodVolumesAnnotation is set")
+	}
+
+	close(podStopCh)
+	close(stopCh)
+}
+
 // wait for the podInformer to observe the pods. Call this function before
 // running the RC manager to prevent the rc manager from creating new pods
 // rather than adopting the existing ones.
@@ -213,8 +356,8 @@ func createAdClients(ns *v1.Namespace, t *testing.T, server *httptest.Server, sy
 		cloud,
 		plugins,
 		false,
-		time.Second*5,
-	)
+		time.Second*5)
+
 	if err != nil {
 		t.Fatalf("Error creating AttachDetach : %v", err)
 	}