Add Keepterminatedpodvolumes as a annotation on node

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,13 +381,13 @@ func (adc *attachDetachController) podAdd(obj interface{}) {
 		return
 	}
 
-	if volumehelper.IsPodTerminated(pod, pod.Status) {
-		util.ProcessPodVolumes(pod, false, /* addVolumes */
-			adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
-	} else {
-		util.ProcessPodVolumes(pod, true, /* addVolumes */
-			adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
-	}
+	volumeActionFlag := util.DetermineVolumeAction(
+		pod,
+		adc.desiredStateOfWorld,
+		true /* default volume action */)
+
+	util.ProcessPodVolumes(pod, volumeActionFlag, /* addVolumes */
+		adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
 }
 
 // GetDesiredStateOfWorld returns desired state of world associated with controller
@@ -409,13 +405,12 @@ func (adc *attachDetachController) podUpdate(oldObj, newObj interface{}) {
 		return
 	}
 
-	addPodFlag := true
-
-	if volumehelper.IsPodTerminated(pod, pod.Status) {
-		addPodFlag = false
-	}
+	volumeActionFlag := util.DetermineVolumeAction(
+		pod,
+		adc.desiredStateOfWorld,
+		true /* default volume action */)
 
-	util.ProcessPodVolumes(pod, addPodFlag, /* addVolumes */
+	util.ProcessPodVolumes(pod, volumeActionFlag, /* addVolumes */
 		adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
 }
 
@@ -453,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 */)
 }
 
@@ -559,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/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/populator/desired_state_of_world_populator.go

@@ -127,7 +127,12 @@ func (dswp *desiredStateOfWorldPopulator) findAndRemoveDeletedPods() {
 			glog.Errorf("podLister Get failed for pod %q (UID %q) with %v", dswPodKey, dswPodUID, err)
 			continue
 		default:
-			if !volumehelper.IsPodTerminated(informerPod, informerPod.Status) {
+			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 {

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

@@ -150,8 +150,8 @@ func TestPodDeletionWithDswp(t *testing.T) {
 }
 
 func TestPodUpdateWithWithADC(t *testing.T) {
-	_, server := framework.RunAMaster(nil)
-	defer server.Close()
+	_, server, closeFn := framework.RunAMaster(nil)
+	defer closeFn()
 	namespaceName := "test-pod-update"
 
 	node := &v1.Node{
@@ -220,6 +220,78 @@ func TestPodUpdateWithWithADC(t *testing.T) {
 	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.