Merge pull request #39732 from tsmetana/issue_34242

pkg/controller/volume/attachdetach/BUILD

@@ -31,6 +31,7 @@ go_library(
         "//pkg/volume/util/operationexecutor:go_default_library",
         "//pkg/volume/util/volumehelper:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
@@ -46,9 +47,14 @@ go_test(
     library = ":go_default_library",
     tags = ["automanaged"],
     deps = [
+        "//pkg/api/v1:go_default_library",
         "//pkg/client/informers/informers_generated/externalversions:go_default_library",
         "//pkg/controller:go_default_library",
+        "//pkg/controller/volume/attachdetach/cache:go_default_library",
         "//pkg/controller/volume/attachdetach/testing:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
     ],
 )
 

pkg/controller/volume/attachdetach/attach_detach_controller.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -102,12 +103,16 @@ func NewAttachDetachController(
 	// dropped pods so they are continuously processed until it is accepted or
 	// deleted (probably can't do this with sharedInformer), etc.
 	adc := &attachDetachController{
-		kubeClient: kubeClient,
-		pvcLister:  pvcInformer.Lister(),
-		pvcsSynced: pvcInformer.Informer().HasSynced,
-		pvLister:   pvInformer.Lister(),
-		pvsSynced:  pvInformer.Informer().HasSynced,
-		cloud:      cloud,
+		kubeClient:  kubeClient,
+		pvcLister:   pvcInformer.Lister(),
+		pvcsSynced:  pvcInformer.Informer().HasSynced,
+		pvLister:    pvInformer.Lister(),
+		pvsSynced:   pvInformer.Informer().HasSynced,
+		podLister:   podInformer.Lister(),
+		podsSynced:  podInformer.Informer().HasSynced,
+		nodeLister:  nodeInformer.Lister(),
+		nodesSynced: nodeInformer.Informer().HasSynced,
+		cloud:       cloud,
 	}
 
 	if err := adc.volumePluginMgr.InitPlugins(plugins, adc); err != nil {
@@ -155,14 +160,12 @@ func NewAttachDetachController(
 		UpdateFunc: adc.podUpdate,
 		DeleteFunc: adc.podDelete,
 	})
-	adc.podsSynced = podInformer.Informer().HasSynced
 
 	nodeInformer.Informer().AddEventHandler(kcache.ResourceEventHandlerFuncs{
 		AddFunc:    adc.nodeAdd,
 		UpdateFunc: adc.nodeUpdate,
 		DeleteFunc: adc.nodeDelete,
 	})
-	adc.nodesSynced = nodeInformer.Informer().HasSynced
 
 	return adc, nil
 }
@@ -184,7 +187,10 @@ type attachDetachController struct {
 	pvLister  corelisters.PersistentVolumeLister
 	pvsSynced kcache.InformerSynced
 
-	podsSynced  kcache.InformerSynced
+	podLister  corelisters.PodLister
+	podsSynced kcache.InformerSynced
+
+	nodeLister  corelisters.NodeLister
 	nodesSynced kcache.InformerSynced
 
 	// cloud provider used by volume host
@@ -239,12 +245,136 @@ func (adc *attachDetachController) Run(stopCh <-chan struct{}) {
 		return
 	}
 
+	err := adc.populateActualStateOfWorld()
+	if err != nil {
+		glog.Errorf("Error populating the actual state of world: %v", err)
+	}
+	err = adc.populateDesiredStateOfWorld()
+	if err != nil {
+		glog.Errorf("Error populating the desired state of world: %v", err)
+	}
 	go adc.reconciler.Run(stopCh)
 	go adc.desiredStateOfWorldPopulator.Run(stopCh)
 
 	<-stopCh
 }
 
+func (adc *attachDetachController) populateActualStateOfWorld() error {
+	glog.V(5).Infof("Populating ActualStateOfworld")
+	nodes, err := adc.nodeLister.List(labels.Everything())
+	if err != nil {
+		return err
+	}
+
+	for _, node := range nodes {
+		nodeName := types.NodeName(node.Name)
+		for _, attachedVolume := range node.Status.VolumesAttached {
+			uniqueName := attachedVolume.Name
+			// The nil VolumeSpec is safe only in the case the volume is not in use by any pod.
+			// In such a case it should be detached in the first reconciliation cycle and the
+			// volume spec is not needed to detach a volume. If the volume is used by a pod, it
+			// its spec can be: this would happen during in the populateDesiredStateOfWorld which
+			// scans the pods and updates their volumes in the ActualStateOfWorld too.
+			err = adc.actualStateOfWorld.MarkVolumeAsAttached(uniqueName, nil /* VolumeSpec */, nodeName, attachedVolume.DevicePath)
+			if err != nil {
+				glog.Errorf("Failed to mark the volume as attached: %v", err)
+				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
+			}
+		}
+	}
+	return nil
+}
+
+func (adc *attachDetachController) getNodeVolumeDevicePath(
+	volumeName v1.UniqueVolumeName, nodeName types.NodeName) (string, error) {
+	var devicePath string
+	var found bool
+	node, err := adc.nodeLister.Get(string(nodeName))
+	if err != nil {
+		return devicePath, err
+	}
+	for _, attachedVolume := range node.Status.VolumesAttached {
+		if volumeName == attachedVolume.Name {
+			devicePath = attachedVolume.DevicePath
+			found = true
+			break
+		}
+	}
+	if !found {
+		err = fmt.Errorf("Volume %s not found on node %s", volumeName, nodeName)
+	}
+
+	return devicePath, err
+}
+
+func (adc *attachDetachController) populateDesiredStateOfWorld() error {
+	glog.V(5).Infof("Populating DesiredStateOfworld")
+
+	pods, err := adc.podLister.List(labels.Everything())
+	if err != nil {
+		return err
+	}
+	for _, pod := range pods {
+		podToAdd := pod
+		adc.podAdd(&podToAdd)
+		for _, podVolume := range podToAdd.Spec.Volumes {
+			// The volume specs present in the ActualStateOfWorld are nil, let's replace those
+			// with the correct ones found on pods. The present in the ASW with no corresponding
+			// pod will be detached and the spec is irrelevant.
+			volumeSpec, err := util.CreateVolumeSpec(podVolume, podToAdd.Namespace, adc.pvcLister, adc.pvLister)
+			if err != nil {
+				glog.Errorf(
+					"Error creating spec for volume %q, pod %q/%q: %v",
+					podVolume.Name,
+					podToAdd.Namespace,
+					podToAdd.Name,
+					err)
+				continue
+			}
+			nodeName := types.NodeName(podToAdd.Spec.NodeName)
+			plugin, err := adc.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
+			if err != nil || plugin == nil {
+				glog.V(10).Infof(
+					"Skipping volume %q for pod %q/%q: it does not implement attacher interface. err=%v",
+					podVolume.Name,
+					podToAdd.Namespace,
+					podToAdd.Name,
+					err)
+				continue
+			}
+			volumeName, err := volumehelper.GetUniqueVolumeNameFromSpec(plugin, volumeSpec)
+			if err != nil {
+				glog.Errorf(
+					"Failed to find unique name for volume %q, pod %q/%q: %v",
+					podVolume.Name,
+					podToAdd.Namespace,
+					podToAdd.Name,
+					err)
+				continue
+			}
+			if adc.actualStateOfWorld.VolumeNodeExists(volumeName, nodeName) {
+				devicePath, err := adc.getNodeVolumeDevicePath(volumeName, nodeName)
+				if err != nil {
+					glog.Errorf("Failed to find device path: %v", err)
+					continue
+				}
+				err = adc.actualStateOfWorld.MarkVolumeAsAttached(volumeName, volumeSpec, nodeName, devicePath)
+				if err != nil {
+					glog.Errorf("Failed to update volume spec for node %s: %v", nodeName, err)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
 func (adc *attachDetachController) podAdd(obj interface{}) {
 	pod, ok := obj.(*v1.Pod)
 	if pod == nil || !ok {
@@ -308,7 +438,7 @@ func (adc *attachDetachController) nodeUpdate(oldObj, newObj interface{}) {
 		// detach controller. Add it to desired state of world.
 		adc.desiredStateOfWorld.AddNode(nodeName)
 	}
-	adc.processVolumesInUse(nodeName, node.Status.VolumesInUse)
+	adc.processVolumesInUse(nodeName, node.Status.VolumesInUse, false /* forceUnmount */)
 }
 
 func (adc *attachDetachController) nodeDelete(obj interface{}) {
@@ -322,7 +452,7 @@ func (adc *attachDetachController) nodeDelete(obj interface{}) {
 		glog.V(10).Infof("%v", err)
 	}
 
-	adc.processVolumesInUse(nodeName, node.Status.VolumesInUse)
+	adc.processVolumesInUse(nodeName, node.Status.VolumesInUse, false /* forceUnmount */)
 }
 
 // processVolumesInUse processes the list of volumes marked as "in-use"
@@ -330,7 +460,7 @@ func (adc *attachDetachController) nodeDelete(obj interface{}) {
 // corresponding volume in the actual state of the world to indicate that it is
 // mounted.
 func (adc *attachDetachController) processVolumesInUse(
-	nodeName types.NodeName, volumesInUse []v1.UniqueVolumeName) {
+	nodeName types.NodeName, volumesInUse []v1.UniqueVolumeName, forceUnmount bool) {
 	glog.V(4).Infof("processVolumesInUse for node %q", nodeName)
 	for _, attachedVolume := range adc.actualStateOfWorld.GetAttachedVolumesForNode(nodeName) {
 		mounted := false
@@ -341,7 +471,7 @@ func (adc *attachDetachController) processVolumesInUse(
 			}
 		}
 		err := adc.actualStateOfWorld.SetVolumeMountedByNode(
-			attachedVolume.VolumeName, nodeName, mounted)
+			attachedVolume.VolumeName, nodeName, mounted, forceUnmount)
 		if err != nil {
 			glog.Warningf(
 				"SetVolumeMountedByNode(%q, %q, %q) returned an error: %v",

pkg/controller/volume/attachdetach/attach_detach_controller_test.go

@@ -17,11 +17,17 @@ limitations under the License.
 package attachdetach
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/kubernetes/pkg/api/v1"
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/externalversions"
 	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/controller/volume/attachdetach/cache"
 	controllervolumetesting "k8s.io/kubernetes/pkg/controller/volume/attachdetach/testing"
 )
 
@@ -47,3 +53,231 @@ func Test_NewAttachDetachController_Positive(t *testing.T) {
 		t.Fatalf("Run failed with error. Expected: <no error> Actual: <%v>", err)
 	}
 }
+
+func Test_AttachDetachControllerStateOfWolrdPopulators_Positive(t *testing.T) {
+	// Arrange
+	fakeKubeClient := controllervolumetesting.CreateTestClient()
+	informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, controller.NoResyncPeriodFunc())
+	podInformer := informerFactory.Core().V1().Pods()
+	nodeInformer := informerFactory.Core().V1().Nodes()
+	pvcInformer := informerFactory.Core().V1().PersistentVolumeClaims()
+	pvInformer := informerFactory.Core().V1().PersistentVolumes()
+
+	adc := &attachDetachController{
+		kubeClient:  fakeKubeClient,
+		pvcLister:   pvcInformer.Lister(),
+		pvcsSynced:  pvcInformer.Informer().HasSynced,
+		pvLister:    pvInformer.Lister(),
+		pvsSynced:   pvInformer.Informer().HasSynced,
+		podLister:   podInformer.Lister(),
+		podsSynced:  podInformer.Informer().HasSynced,
+		nodeLister:  nodeInformer.Lister(),
+		nodesSynced: nodeInformer.Informer().HasSynced,
+		cloud:       nil,
+	}
+
+	// Act
+	plugins := controllervolumetesting.CreateTestPlugin()
+
+	if err := adc.volumePluginMgr.InitPlugins(plugins, adc); err != nil {
+		t.Fatalf("Could not initialize volume plugins for Attach/Detach Controller: %+v", err)
+	}
+
+	adc.actualStateOfWorld = cache.NewActualStateOfWorld(&adc.volumePluginMgr)
+	adc.desiredStateOfWorld = cache.NewDesiredStateOfWorld(&adc.volumePluginMgr)
+
+	err := adc.populateActualStateOfWorld()
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: <%v>", err)
+	}
+
+	err = adc.populateDesiredStateOfWorld()
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: %v", err)
+	}
+
+	// Test the ActualStateOfWorld contains all the node volumes
+	nodes, err := adc.nodeLister.List(labels.Everything())
+	for _, node := range nodes {
+		nodeName := types.NodeName(node.Name)
+		for _, attachedVolume := range node.Status.VolumesAttached {
+			found := adc.actualStateOfWorld.VolumeNodeExists(attachedVolume.Name, nodeName)
+			if !found {
+				t.Fatalf("Run failed with error. Node %s, volume %s not found", nodeName, attachedVolume.Name)
+			}
+		}
+	}
+
+	pods, err := adc.podLister.List(labels.Everything())
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: %v", err)
+	}
+	for _, pod := range pods {
+		uniqueName := fmt.Sprintf("%s/%s", controllervolumetesting.TestPluginName, pod.Spec.Volumes[0].Name)
+		nodeName := types.NodeName(pod.Spec.NodeName)
+		found := adc.desiredStateOfWorld.VolumeExists(v1.UniqueVolumeName(uniqueName), nodeName)
+		if !found {
+			t.Fatalf("Run failed with error. Volume %s, node %s not found in DesiredStateOfWorld",
+				pod.Spec.Volumes[0].Name,
+				pod.Spec.NodeName)
+		}
+	}
+}
+
+func Test_AttachDetachControllerRecovery(t *testing.T) {
+	attachDetachRecoveryTestCase(t, []*v1.Pod{}, []*v1.Pod{})
+	newPod1 := controllervolumetesting.NewPodWithVolume("newpod-1", "volumeName2", "mynode-1")
+	attachDetachRecoveryTestCase(t, []*v1.Pod{newPod1}, []*v1.Pod{})
+	newPod1 = controllervolumetesting.NewPodWithVolume("newpod-1", "volumeName2", "mynode-1")
+	attachDetachRecoveryTestCase(t, []*v1.Pod{}, []*v1.Pod{newPod1})
+	newPod1 = controllervolumetesting.NewPodWithVolume("newpod-1", "volumeName2", "mynode-1")
+	newPod2 := controllervolumetesting.NewPodWithVolume("newpod-2", "volumeName3", "mynode-1")
+	attachDetachRecoveryTestCase(t, []*v1.Pod{newPod1}, []*v1.Pod{newPod2})
+}
+
+func attachDetachRecoveryTestCase(t *testing.T, extraPods1 []*v1.Pod, extraPods2 []*v1.Pod) {
+	fakeKubeClient := controllervolumetesting.CreateTestClient()
+	informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, time.Second*1)
+	//informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, time.Second*1)
+	plugins := controllervolumetesting.CreateTestPlugin()
+	nodeInformer := informerFactory.Core().V1().Nodes().Informer()
+	podInformer := informerFactory.Core().V1().Pods().Informer()
+	var podsNum, extraPodsNum, nodesNum, i int
+
+	stopCh := make(chan struct{})
+
+	pods, err := fakeKubeClient.Core().Pods(v1.NamespaceAll).List(metav1.ListOptions{})
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: %v", err)
+	}
+
+	for _, pod := range pods.Items {
+		podToAdd := pod
+		podInformer.GetIndexer().Add(&podToAdd)
+		podsNum++
+	}
+	nodes, err := fakeKubeClient.Core().Nodes().List(metav1.ListOptions{})
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: %v", err)
+	}
+	for _, node := range nodes.Items {
+		nodeToAdd := node
+		nodeInformer.GetIndexer().Add(&nodeToAdd)
+		nodesNum++
+	}
+
+	informerFactory.Start(stopCh)
+
+	// Make sure the nodes and pods are in the inforer cache
+	i = 0
+	nodeList, err := informerFactory.Core().V1().Nodes().Lister().List(labels.Everything())
+	for len(nodeList) < nodesNum {
+		if err != nil {
+			t.Fatalf("Error getting list of nodes %v", err)
+		}
+		if i > 100 {
+			t.Fatalf("Time out while waiting for the node informer sync: found %d nodes, expected %d nodes", len(nodeList), nodesNum)
+		}
+		time.Sleep(100 * time.Millisecond)
+		nodeList, err = informerFactory.Core().V1().Nodes().Lister().List(labels.Everything())
+		i++
+	}
+	i = 0
+	podList, err := informerFactory.Core().V1().Pods().Lister().List(labels.Everything())
+	for len(podList) < podsNum {
+		if err != nil {
+			t.Fatalf("Error getting list of nodes %v", err)
+		}
+		if i > 100 {
+			t.Fatalf("Time out while waiting for the pod informer sync: found %d pods, expected %d pods", len(podList), podsNum)
+		}
+		time.Sleep(100 * time.Millisecond)
+		podList, err = informerFactory.Core().V1().Pods().Lister().List(labels.Everything())
+		i++
+	}
+
+	// Create the controller
+	adcObj, err := NewAttachDetachController(
+		fakeKubeClient,
+		informerFactory.Core().V1().Pods(),
+		informerFactory.Core().V1().Nodes(),
+		informerFactory.Core().V1().PersistentVolumeClaims(),
+		informerFactory.Core().V1().PersistentVolumes(),
+		nil, /* cloud */
+		plugins,
+		false,
+		time.Second*1)
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: <%v>", err)
+	}
+
+	adc := adcObj.(*attachDetachController)
+
+	// Populate ASW
+	err = adc.populateActualStateOfWorld()
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: <%v>", err)
+	}
+
+	for _, newPod := range extraPods1 {
+		// Add a new pod between ASW and DSW ppoulators
+		_, err = adc.kubeClient.Core().Pods(newPod.ObjectMeta.Namespace).Create(newPod)
+		if err != nil {
+			t.Fatalf("Run failed with error. Failed to create a new pod: <%v>", err)
+		}
+		extraPodsNum++
+		podInformer.GetIndexer().Add(newPod)
+
+	}
+
+	// Populate DSW
+	err = adc.populateDesiredStateOfWorld()
+	if err != nil {
+		t.Fatalf("Run failed with error. Expected: <no error> Actual: %v", err)
+	}
+
+	for _, newPod := range extraPods2 {
+		// Add a new pod between DSW ppoulator and reconciler run
+		_, err = adc.kubeClient.Core().Pods(newPod.ObjectMeta.Namespace).Create(newPod)
+		if err != nil {
+			t.Fatalf("Run failed with error. Failed to create a new pod: <%v>", err)
+		}
+		extraPodsNum++
+		podInformer.GetIndexer().Add(newPod)
+	}
+
+	go adc.reconciler.Run(stopCh)
+	go adc.desiredStateOfWorldPopulator.Run(stopCh)
+	defer close(stopCh)
+
+	time.Sleep(time.Second * 1) // Wait so the reconciler calls sync at least once
+
+	testPlugin := plugins[0].(*controllervolumetesting.TestPlugin)
+	for i = 0; i <= 10; i++ {
+		var attachedVolumesNum int = 0
+		var detachedVolumesNum int = 0
+
+		time.Sleep(time.Second * 1) // Wait for a second
+		for _, volumeList := range testPlugin.GetAttachedVolumes() {
+			attachedVolumesNum += len(volumeList)
+		}
+		for _, volumeList := range testPlugin.GetDetachedVolumes() {
+			detachedVolumesNum += len(volumeList)
+		}
+
+		// All the "extra pods" should result in volume to be attached, the pods all share one volume
+		// which should be attached (+1), the volumes found only in the nodes status should be detached
+		if attachedVolumesNum == 1+extraPodsNum && detachedVolumesNum == nodesNum {
+			break
+		}
+		if i == 10 { // 10 seconds time out
+			t.Fatalf("Waiting for the volumes to attach/detach timed out: attached %d (expected %d); detached %d (%d)",
+				attachedVolumesNum, 1+extraPodsNum, detachedVolumesNum, nodesNum)
+		}
+	}
+
+	if testPlugin.GetErrorEncountered() {
+		t.Fatalf("Fatal error encountered in the testing volume plugin")
+	}
+
+}

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

@@ -56,16 +56,19 @@ type ActualStateOfWorld interface {
 	// added.
 	// If no node with the name nodeName exists in list of attached nodes for
 	// the specified volume, the node is added.
-	AddVolumeNode(volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) (v1.UniqueVolumeName, error)
+	AddVolumeNode(uniqueName v1.UniqueVolumeName, volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) (v1.UniqueVolumeName, error)
 
 	// SetVolumeMountedByNode sets the MountedByNode value for the given volume
-	// and node. When set to true this value indicates the volume is mounted by
-	// the given node, indicating it may not be safe to detach.
+	// and node. When set to true the mounted parameter indicates the volume
+	// is mounted by the given node, indicating it may not be safe to detach.
+	// If the forceUnmount is set to true the MountedByNode value would be reset
+	// to false even it was not set yet (this is required during a controller
+	// crash recovery).
 	// If no volume with the name volumeName exists in the store, an error is
 	// returned.
 	// If no node with the name nodeName exists in list of attached nodes for
 	// the specified volume, an error is returned.
-	SetVolumeMountedByNode(volumeName v1.UniqueVolumeName, nodeName types.NodeName, mounted bool) error
+	SetVolumeMountedByNode(volumeName v1.UniqueVolumeName, nodeName types.NodeName, mounted bool, forceUnmount bool) error
 
 	// SetNodeStatusUpdateNeeded sets statusUpdateNeeded for the specified
 	// node to true indicating the AttachedVolume field in the Node's Status
@@ -230,8 +233,8 @@ type nodeToUpdateStatusFor struct {
 }
 
 func (asw *actualStateOfWorld) MarkVolumeAsAttached(
-	_ v1.UniqueVolumeName, volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) error {
-	_, err := asw.AddVolumeNode(volumeSpec, nodeName, devicePath)
+	uniqueName v1.UniqueVolumeName, volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) error {
+	_, err := asw.AddVolumeNode(uniqueName, volumeSpec, nodeName, devicePath)
 	return err
 }
 
@@ -255,25 +258,34 @@ func (asw *actualStateOfWorld) AddVolumeToReportAsAttached(
 }
 
 func (asw *actualStateOfWorld) AddVolumeNode(
-	volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) (v1.UniqueVolumeName, error) {
+	uniqueName v1.UniqueVolumeName, volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) (v1.UniqueVolumeName, error) {
 	asw.Lock()
 	defer asw.Unlock()
 
-	attachableVolumePlugin, err := asw.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
-	if err != nil || attachableVolumePlugin == nil {
-		return "", fmt.Errorf(
-			"failed to get AttachablePlugin from volumeSpec for volume %q err=%v",
-			volumeSpec.Name(),
-			err)
-	}
+	var volumeName v1.UniqueVolumeName
+	if volumeSpec != nil {
+		attachableVolumePlugin, err := asw.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
+		if err != nil || attachableVolumePlugin == nil {
+			return "", fmt.Errorf(
+				"failed to get AttachablePlugin from volumeSpec for volume %q err=%v",
+				volumeSpec.Name(),
+				err)
+		}
 
-	volumeName, err := volumehelper.GetUniqueVolumeNameFromSpec(
-		attachableVolumePlugin, volumeSpec)
-	if err != nil {
-		return "", fmt.Errorf(
-			"failed to GetUniqueVolumeNameFromSpec for volumeSpec %q err=%v",
-			volumeSpec.Name(),
-			err)
+		volumeName, err = volumehelper.GetUniqueVolumeNameFromSpec(
+			attachableVolumePlugin, volumeSpec)
+		if err != nil {
+			return "", fmt.Errorf(
+				"failed to GetUniqueVolumeNameFromSpec for volumeSpec %q err=%v",
+				volumeSpec.Name(),
+				err)
+		}
+	} else {
+		// volumeSpec is nil
+		// This happens only on controller startup when reading the volumes from node
+		// status; if the pods using the volume have been removed and are unreachable
+		// the volumes should be detached immediately and the spec is not needed
+		volumeName = uniqueName
 	}
 
 	volumeObj, volumeExists := asw.attachedVolumes[volumeName]
@@ -316,7 +328,7 @@ func (asw *actualStateOfWorld) AddVolumeNode(
 }
 
 func (asw *actualStateOfWorld) SetVolumeMountedByNode(
-	volumeName v1.UniqueVolumeName, nodeName types.NodeName, mounted bool) error {
+	volumeName v1.UniqueVolumeName, nodeName types.NodeName, mounted bool, forceUnmount bool) error {
 	asw.Lock()
 	defer asw.Unlock()
 
@@ -330,7 +342,7 @@ func (asw *actualStateOfWorld) SetVolumeMountedByNode(
 		nodeObj.mountedByNodeSetCount = nodeObj.mountedByNodeSetCount + 1
 	} else {
 		// Do not allow value to be reset unless it has been set at least once
-		if nodeObj.mountedByNodeSetCount == 0 {
+		if nodeObj.mountedByNodeSetCount == 0 && !forceUnmount {
 			return nil
 		}
 	}

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

@@ -184,9 +184,8 @@ func (rc *reconciler) reconcile() {
 				// Ignore nestedpendingoperations.IsAlreadyExists && exponentialbackoff.IsExponentialBackoff errors, they are expected.
 				// Log all other errors.
 				glog.Errorf(
-					"operationExecutor.DetachVolume failed to start for volume %q (spec.Name: %q) from node %q with err: %v",
+					"operationExecutor.DetachVolume failed to start for volume %q from node %q with err: %v",
 					attachedVolume.VolumeName,
-					attachedVolume.VolumeSpec.Name(),
 					attachedVolume.NodeName,
 					err)
 			}

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

@@ -170,8 +170,8 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithUnmountedVolume(t *te
 			generatedVolumeName,
 			nodeName)
 	}
-	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, true /* mounted */)
-	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, false /* mounted */)
+	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, true /* mounted */, false)
+	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, false /* mounted */, false)
 
 	// Assert
 	waitForNewDetacherCallCount(t, 1 /* expectedCallCount */, fakePlugin)
@@ -302,8 +302,8 @@ func Test_Run_Negative_OneDesiredVolumeAttachThenDetachWithUnmountedVolumeUpdate
 			generatedVolumeName,
 			nodeName)
 	}
-	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, true /* mounted */)
-	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, false /* mounted */)
+	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, true /* mounted */, false)
+	asw.SetVolumeMountedByNode(generatedVolumeName, nodeName, false /* mounted */, false)
 
 	// Assert
 	verifyNewDetacherCallCount(t, true /* expectZeroNewDetacherCallCount */, fakePlugin)

pkg/controller/volume/attachdetach/testing/BUILD

@@ -15,6 +15,8 @@ go_library(
         "//pkg/api/v1:go_default_library",
         "//pkg/client/clientset_generated/clientset/fake:go_default_library",
         "//pkg/volume:go_default_library",
+        "//pkg/volume/util/volumehelper:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",

pkg/volume/plugins.go

@@ -68,9 +68,10 @@ type VolumePlugin interface {
 	// depend on this.
 	Init(host VolumeHost) error
 
-	// Name returns the plugin's name.  Plugins should use namespaced names
-	// such as "example.com/volume".  The "kubernetes.io" namespace is
-	// reserved for plugins which are bundled with kubernetes.
+	// Name returns the plugin's name.  Plugins must use namespaced names
+	// such as "example.com/volume" and contain exactly one '/' character.
+	// The "kubernetes.io" namespace is reserved for plugins which are
+	// bundled with kubernetes.
 	GetPluginName() string
 
 	// GetVolumeName returns the name/ID to uniquely identifying the actual

pkg/volume/util/operationexecutor/operation_executor.go

@@ -403,6 +403,10 @@ func (oe *operationExecutor) VerifyVolumesAreAttached(
 
 	for node, nodeAttachedVolumes := range attachedVolumes {
 		for _, volumeAttached := range nodeAttachedVolumes {
+			if volumeAttached.VolumeSpec == nil {
+				glog.Errorf("VerifyVolumesAreAttached: nil spec for volume %s", volumeAttached.VolumeName)
+				continue
+			}
 			volumePlugin, err :=
 				oe.operationGenerator.GetVolumePluginMgr().FindPluginBySpec(volumeAttached.VolumeSpec)
 

pkg/volume/util/operationexecutor/operation_generator.go

@@ -30,6 +30,7 @@ import (
 	kevents "k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
 )
 
 var _ OperationGenerator = &operationGenerator{}
@@ -314,35 +315,48 @@ func (og *operationGenerator) GenerateDetachVolumeFunc(
 	volumeToDetach AttachedVolume,
 	verifySafeToDetach bool,
 	actualStateOfWorld ActualStateOfWorldAttacherUpdater) (func() error, error) {
-	// Get attacher plugin
-	attachableVolumePlugin, err :=
-		og.volumePluginMgr.FindAttachablePluginBySpec(volumeToDetach.VolumeSpec)
-	if err != nil || attachableVolumePlugin == nil {
-		return nil, fmt.Errorf(
-			"DetachVolume.FindAttachablePluginBySpec failed for volume %q (spec.Name: %q) from node %q with: %v",
-			volumeToDetach.VolumeName,
-			volumeToDetach.VolumeSpec.Name(),
-			volumeToDetach.NodeName,
-			err)
-	}
+	var volumeName string
+	var attachableVolumePlugin volume.AttachableVolumePlugin
+	var err error
+
+	if volumeToDetach.VolumeSpec != nil {
+		// Get attacher plugin
+		attachableVolumePlugin, err =
+			og.volumePluginMgr.FindAttachablePluginBySpec(volumeToDetach.VolumeSpec)
+		if err != nil || attachableVolumePlugin == nil {
+			return nil, fmt.Errorf(
+				"DetachVolume.FindAttachablePluginBySpec failed for volume %q (spec.Name: %q) from node %q with: %v",
+				volumeToDetach.VolumeName,
+				volumeToDetach.VolumeSpec.Name(),
+				volumeToDetach.NodeName,
+				err)
+		}
 
-	volumeName, err :=
-		attachableVolumePlugin.GetVolumeName(volumeToDetach.VolumeSpec)
-	if err != nil {
-		return nil, fmt.Errorf(
-			"DetachVolume.GetVolumeName failed for volume %q (spec.Name: %q) from node %q with: %v",
-			volumeToDetach.VolumeName,
-			volumeToDetach.VolumeSpec.Name(),
-			volumeToDetach.NodeName,
-			err)
+		volumeName, err =
+			attachableVolumePlugin.GetVolumeName(volumeToDetach.VolumeSpec)
+	} else {
+		var pluginName string
+		// Get attacher plugin and the volumeName by splitting the volume unique name in case
+		// there's no VolumeSpec: this happens only on attach/detach controller crash recovery
+		// when a pod has been deleted during the controller downtime
+		pluginName, volumeName, err = volumehelper.SplitUniqueName(volumeToDetach.VolumeName)
+		if err != nil {
+			return nil, err
+		}
+		attachableVolumePlugin, err = og.volumePluginMgr.FindAttachablePluginByName(pluginName)
+		if err != nil {
+			return nil, fmt.Errorf(
+				"Failed to find plugin for volume %q from node %q with: %v",
+				volumeToDetach.VolumeName,
+				volumeToDetach.NodeName,
+				err)
+		}
 	}
-
 	volumeDetacher, err := attachableVolumePlugin.NewDetacher()
 	if err != nil {
 		return nil, fmt.Errorf(
-			"DetachVolume.NewDetacher failed for volume %q (spec.Name: %q) from node %q with: %v",
+			"DetachVolume.NewDetacher failed for volume %q from node %q with: %v",
 			volumeToDetach.VolumeName,
-			volumeToDetach.VolumeSpec.Name(),
 			volumeToDetach.NodeName,
 			err)
 	}
@@ -360,17 +374,15 @@ func (og *operationGenerator) GenerateDetachVolumeFunc(
 			actualStateOfWorld.AddVolumeToReportAsAttached(
 				volumeToDetach.VolumeName, volumeToDetach.NodeName)
 			return fmt.Errorf(
-				"DetachVolume.Detach failed for volume %q (spec.Name: %q) from node %q with: %v",
+				"DetachVolume.Detach failed for volume %q from node %q with: %v",
 				volumeToDetach.VolumeName,
-				volumeToDetach.VolumeSpec.Name(),
 				volumeToDetach.NodeName,
 				err)
 		}
 
 		glog.Infof(
-			"DetachVolume.Detach succeeded for volume %q (spec.Name: %q) from node %q.",
+			"DetachVolume.Detach succeeded for volume %q from node %q.",
 			volumeToDetach.VolumeName,
-			volumeToDetach.VolumeSpec.Name(),
 			volumeToDetach.NodeName)
 
 		// Update actual state of world
@@ -865,16 +877,14 @@ func (og *operationGenerator) verifyVolumeIsSafeToDetach(
 		if errors.IsNotFound(fetchErr) {
 			glog.Warningf("Node %q not found on API server. DetachVolume will skip safe to detach check.",
 				volumeToDetach.NodeName,
-				volumeToDetach.VolumeName,
-				volumeToDetach.VolumeSpec.Name())
+				volumeToDetach.VolumeName)
 			return nil
 		}
 
 		// On failure, return error. Caller will log and retry.
 		return fmt.Errorf(
-			"DetachVolume failed fetching node from API server for volume %q (spec.Name: %q) from node %q with: %v",
+			"DetachVolume failed fetching node from API server for volume %q from node %q with: %v",
 			volumeToDetach.VolumeName,
-			volumeToDetach.VolumeSpec.Name(),
 			volumeToDetach.NodeName,
 			fetchErr)
 	}
@@ -882,25 +892,22 @@ func (og *operationGenerator) verifyVolumeIsSafeToDetach(
 	if node == nil {
 		// On failure, return error. Caller will log and retry.
 		return fmt.Errorf(
-			"DetachVolume failed fetching node from API server for volume %q (spec.Name: %q) from node %q. Error: node object retrieved from API server is nil",
+			"DetachVolume failed fetching node from API server for volume %q from node %q. Error: node object retrieved from API server is nil",
 			volumeToDetach.VolumeName,
-			volumeToDetach.VolumeSpec.Name(),
 			volumeToDetach.NodeName)
 	}
 
 	for _, inUseVolume := range node.Status.VolumesInUse {
 		if inUseVolume == volumeToDetach.VolumeName {
-			return fmt.Errorf("DetachVolume failed for volume %q (spec.Name: %q) from node %q. Error: volume is still in use by node, according to Node status",
+			return fmt.Errorf("DetachVolume failed for volume %q from node %q. Error: volume is still in use by node, according to Node status",
 				volumeToDetach.VolumeName,
-				volumeToDetach.VolumeSpec.Name(),
 				volumeToDetach.NodeName)
 		}
 	}
 
 	// Volume is not marked as in use by node
-	glog.Infof("Verified volume is safe to detach for volume %q (spec.Name: %q) from node %q.",
+	glog.Infof("Verified volume is safe to detach for volume %q from node %q.",
 		volumeToDetach.VolumeName,
-		volumeToDetach.VolumeSpec.Name(),
 		volumeToDetach.NodeName)
 	return nil
 }

pkg/volume/util/volumehelper/volumehelper.go

@@ -20,6 +20,7 @@ package volumehelper
 
 import (
 	"fmt"
+	"strings"
 
 	"k8s.io/kubernetes/pkg/api/v1"
 	"k8s.io/kubernetes/pkg/volume"
@@ -103,3 +104,18 @@ func notRunning(statuses []v1.ContainerStatus) bool {
 	}
 	return true
 }
+
+// SplitUniqueName splits the unique name to plugin name and volume name strings. It expects the uniqueName to follow
+// the fromat plugin_name/volume_name and the plugin name must be namespaced as descibed by the plugin interface,
+// i.e. namespace/plugin containing exactly one '/'. This means the unique name will always be in the form of
+// plugin_namespace/plugin/volume_name, see k8s.io/kubernetes/pkg/volume/plugins.go VolumePlugin interface
+// description and pkg/volume/util/volumehelper/volumehelper.go GetUniqueVolumeNameFromSpec that constructs
+// the unique volume names.
+func SplitUniqueName(uniqueName v1.UniqueVolumeName) (string, string, error) {
+	components := strings.SplitN(string(uniqueName), "/", 3)
+	if len(components) != 3 {
+		return "", "", fmt.Errorf("cannot split volume unique name %s to plugin/volume components", uniqueName)
+	}
+	pluginName := fmt.Sprintf("%s/%s", components[0], components[1])
+	return pluginName, components[2], nil
+}