Controller wait for attach and exponential backoff

pkg/controller/persistentvolume/controller_base.go

@@ -65,7 +65,7 @@ func NewPersistentVolumeController(
 		claims:                        cache.NewStore(framework.DeletionHandlingMetaNamespaceKeyFunc),
 		kubeClient:                    kubeClient,
 		eventRecorder:                 eventRecorder,
-		runningOperations:             goroutinemap.NewGoRoutineMap(),
+		runningOperations:             goroutinemap.NewGoRoutineMap(false /* exponentialBackOffOnError */),
 		cloud:                         cloud,
 		provisioner:                   provisioner,
 		clusterName:                   clusterName,

pkg/controller/volume/attach_detach_controller.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/framework"
 	"k8s.io/kubernetes/pkg/controller/volume/cache"
 	"k8s.io/kubernetes/pkg/controller/volume/reconciler"
+	"k8s.io/kubernetes/pkg/controller/volume/statusupdater"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util/io"
 	"k8s.io/kubernetes/pkg/util/mount"
@@ -105,13 +106,18 @@ func NewAttachDetachController(
 	adc.desiredStateOfWorld = cache.NewDesiredStateOfWorld(&adc.volumePluginMgr)
 	adc.actualStateOfWorld = cache.NewActualStateOfWorld(&adc.volumePluginMgr)
 	adc.attacherDetacher =
-		operationexecutor.NewOperationExecutor(&adc.volumePluginMgr)
+		operationexecutor.NewOperationExecutor(
+			kubeClient,
+			&adc.volumePluginMgr)
+	adc.nodeStatusUpdater = statusupdater.NewNodeStatusUpdater(
+		kubeClient, nodeInformer, adc.actualStateOfWorld)
 	adc.reconciler = reconciler.NewReconciler(
 		reconcilerLoopPeriod,
 		reconcilerMaxWaitForUnmountDuration,
 		adc.desiredStateOfWorld,
 		adc.actualStateOfWorld,
-		adc.attacherDetacher)
+		adc.attacherDetacher,
+		adc.nodeStatusUpdater)
 
 	return adc, nil
 }
@@ -160,6 +166,10 @@ type attachDetachController struct {
 	// desiredStateOfWorld with the actualStateOfWorld by triggering attach
 	// detach operations using the attacherDetacher.
 	reconciler reconciler.Reconciler
+
+	// nodeStatusUpdater is used to update node status with the list of attached
+	// volumes
+	nodeStatusUpdater statusupdater.NodeStatusUpdater
 }
 
 func (adc *attachDetachController) Run(stopCh <-chan struct{}) {

pkg/controller/volume/attach_detach_controller_test.go

@@ -17,21 +17,16 @@ limitations under the License.
 package volume
 
 import (
-	"fmt"
 	"testing"
 	"time"
 
-	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
-	"k8s.io/kubernetes/pkg/client/testing/core"
 	"k8s.io/kubernetes/pkg/controller/framework/informers"
-	"k8s.io/kubernetes/pkg/runtime"
-	"k8s.io/kubernetes/pkg/watch"
+	controllervolumetesting "k8s.io/kubernetes/pkg/controller/volume/testing"
 )
 
 func Test_NewAttachDetachController_Positive(t *testing.T) {
 	// Arrange
-	fakeKubeClient := createTestClient()
+	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	resyncPeriod := 5 * time.Minute
 	podInformer := informers.CreateSharedPodIndexInformer(fakeKubeClient, resyncPeriod)
 	nodeInformer := informers.CreateSharedNodeIndexInformer(fakeKubeClient, resyncPeriod)
@@ -53,62 +48,3 @@ func Test_NewAttachDetachController_Positive(t *testing.T) {
 		t.Fatalf("Run failed with error. Expected: <no error> Actual: <%v>", err)
 	}
 }
-
-func createTestClient() *fake.Clientset {
-	fakeClient := &fake.Clientset{}
-
-	fakeClient.AddReactor("list", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
-		obj := &api.PodList{}
-		podNamePrefix := "mypod"
-		namespace := "mynamespace"
-		for i := 0; i < 5; i++ {
-			podName := fmt.Sprintf("%s-%d", podNamePrefix, i)
-			pod := api.Pod{
-				Status: api.PodStatus{
-					Phase: api.PodRunning,
-				},
-				ObjectMeta: api.ObjectMeta{
-					Name:      podName,
-					Namespace: namespace,
-					Labels: map[string]string{
-						"name": podName,
-					},
-				},
-				Spec: api.PodSpec{
-					Containers: []api.Container{
-						{
-							Name:  "containerName",
-							Image: "containerImage",
-							VolumeMounts: []api.VolumeMount{
-								{
-									Name:      "volumeMountName",
-									ReadOnly:  false,
-									MountPath: "/mnt",
-								},
-							},
-						},
-					},
-					Volumes: []api.Volume{
-						{
-							Name: "volumeName",
-							VolumeSource: api.VolumeSource{
-								GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{
-									PDName:   "pdName",
-									FSType:   "ext4",
-									ReadOnly: false,
-								},
-							},
-						},
-					},
-				},
-			}
-			obj.Items = append(obj.Items, pod)
-		}
-		return true, obj, nil
-	})
-
-	fakeWatch := watch.NewFake()
-	fakeClient.AddWatchReactor("*", core.DefaultWatchReactor(fakeWatch, nil))
-
-	return fakeClient
-}

pkg/controller/volume/cache/actual_state_of_world.go

@@ -55,7 +55,7 @@ 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 string) (api.UniqueVolumeName, error)
+	AddVolumeNode(volumeSpec *volume.Spec, nodeName string, devicePath string) (api.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
@@ -75,6 +75,13 @@ type ActualStateOfWorld interface {
 	// the specified volume, an error is returned.
 	MarkDesireToDetach(volumeName api.UniqueVolumeName, nodeName string) (time.Duration, error)
 
+	// ResetNodeStatusUpdateNeeded resets statusUpdateNeeded for the specified
+	// node to false indicating the AttachedVolume field of the Node's Status
+	// object has been updated.
+	// If no node with the name nodeName exists in list of attached nodes for
+	// the specified volume, an error is returned.
+	ResetNodeStatusUpdateNeeded(nodeName string) error
+
 	// DeleteVolumeNode removes the given volume and node from the underlying
 	// store indicating the specified volume is no longer attached to the
 	// specified node.
@@ -97,6 +104,15 @@ type ActualStateOfWorld interface {
 	// the specified node reflecting which volumes are attached to that node
 	// based on the current actual state of the world.
 	GetAttachedVolumesForNode(nodeName string) []AttachedVolume
+
+	// GetVolumesToReportAttached returns a map containing the set of nodes for
+	// which the VolumesAttached Status field in the Node API object should be
+	// updated. The key in this map is the name of the node to update and the
+	// value is list of volumes that should be reported as attached (note that
+	// this may differ from the actual list of attached volumes for the node
+	// since volumes should be removed from this list as soon a detach operation
+	// is considered, before the detach operation is triggered).
+	GetVolumesToReportAttached() map[string][]api.AttachedVolume
 }
 
 // AttachedVolume represents a volume that is attached to a node.
@@ -119,8 +135,9 @@ type AttachedVolume struct {
 // NewActualStateOfWorld returns a new instance of ActualStateOfWorld.
 func NewActualStateOfWorld(volumePluginMgr *volume.VolumePluginMgr) ActualStateOfWorld {
 	return &actualStateOfWorld{
-		attachedVolumes: make(map[api.UniqueVolumeName]attachedVolume),
-		volumePluginMgr: volumePluginMgr,
+		attachedVolumes:        make(map[api.UniqueVolumeName]attachedVolume),
+		nodesToUpdateStatusFor: make(map[string]nodeToUpdateStatusFor),
+		volumePluginMgr:        volumePluginMgr,
 	}
 }
 
@@ -130,9 +147,17 @@ type actualStateOfWorld struct {
 	// managing. The key in this map is the name of the volume and the value is
 	// an object containing more information about the attached volume.
 	attachedVolumes map[api.UniqueVolumeName]attachedVolume
+
+	// nodesToUpdateStatusFor is a map containing the set of nodes for which to
+	// update the VolumesAttached Status field. The key in this map is the name
+	// of the node and the value is an object containing more information about
+	// the node (including the list of volumes to report attached).
+	nodesToUpdateStatusFor map[string]nodeToUpdateStatusFor
+
 	// volumePluginMgr is the volume plugin manager used to create volume
 	// plugin objects.
 	volumePluginMgr *volume.VolumePluginMgr
+
 	sync.RWMutex
 }
 
@@ -152,9 +177,12 @@ type attachedVolume struct {
 	// node and the value is a node object containing more information about
 	// the node.
 	nodesAttachedTo map[string]nodeAttachedTo
+
+	// devicePath contains the path on the node where the volume is attached
+	devicePath string
 }
 
-// The nodeAttachedTo object represents a node that .
+// The nodeAttachedTo object represents a node that has volumes attached to it.
 type nodeAttachedTo struct {
 	// nodeName contains the name of this node.
 	nodeName string
@@ -173,9 +201,31 @@ type nodeAttachedTo struct {
 	detachRequestedTime time.Time
 }
 
+// nodeToUpdateStatusFor is an object that reflects a node that has one or more
+// volume attached. It keeps track of the volumes that should be reported as
+// attached in the Node's Status API object.
+type nodeToUpdateStatusFor struct {
+	// nodeName contains the name of this node.
+	nodeName string
+
+	// statusUpdateNeeded indicates that the value of the VolumesAttached field
+	// in the Node's Status API object should be updated. This should be set to
+	// true whenever a volume is added or deleted from
+	// volumesToReportAsAttached. It should be reset whenever the status is
+	// updated.
+	statusUpdateNeeded bool
+
+	// volumesToReportAsAttached is the list of volumes that should be reported
+	// as attached in the Node's status (note that this may differ from the
+	// actual list of attached volumes since volumes should be removed from this
+	// list as soon a detach operation is considered, before the detach
+	// operation is triggered).
+	volumesToReportAsAttached map[api.UniqueVolumeName]api.UniqueVolumeName
+}
+
 func (asw *actualStateOfWorld) MarkVolumeAsAttached(
-	volumeSpec *volume.Spec, nodeName string) error {
-	_, err := asw.AddVolumeNode(volumeSpec, nodeName)
+	volumeSpec *volume.Spec, nodeName string, devicePath string) error {
+	_, err := asw.AddVolumeNode(volumeSpec, nodeName, devicePath)
 	return err
 }
 
@@ -185,7 +235,7 @@ func (asw *actualStateOfWorld) MarkVolumeAsDetached(
 }
 
 func (asw *actualStateOfWorld) AddVolumeNode(
-	volumeSpec *volume.Spec, nodeName string) (api.UniqueVolumeName, error) {
+	volumeSpec *volume.Spec, nodeName string, devicePath string) (api.UniqueVolumeName, error) {
 	asw.Lock()
 	defer asw.Unlock()
 
@@ -212,6 +262,7 @@ func (asw *actualStateOfWorld) AddVolumeNode(
 			volumeName:      volumeName,
 			spec:            volumeSpec,
 			nodesAttachedTo: make(map[string]nodeAttachedTo),
+			devicePath:      devicePath,
 		}
 		asw.attachedVolumes[volumeName] = volumeObj
 	}
@@ -231,6 +282,24 @@ func (asw *actualStateOfWorld) AddVolumeNode(
 		volumeObj.nodesAttachedTo[nodeName] = nodeObj
 	}
 
+	nodeToUpdate, nodeToUpdateExists := asw.nodesToUpdateStatusFor[nodeName]
+	if !nodeToUpdateExists {
+		// Create object if it doesn't exist
+		nodeToUpdate = nodeToUpdateStatusFor{
+			nodeName:                  nodeName,
+			statusUpdateNeeded:        true,
+			volumesToReportAsAttached: make(map[api.UniqueVolumeName]api.UniqueVolumeName),
+		}
+		asw.nodesToUpdateStatusFor[nodeName] = nodeToUpdate
+	}
+	_, nodeToUpdateVolumeExists :=
+		nodeToUpdate.volumesToReportAsAttached[volumeName]
+	if !nodeToUpdateVolumeExists {
+		nodeToUpdate.statusUpdateNeeded = true
+		nodeToUpdate.volumesToReportAsAttached[volumeName] = volumeName
+		asw.nodesToUpdateStatusFor[nodeName] = nodeToUpdate
+	}
+
 	return volumeName, nil
 }
 
@@ -298,9 +367,38 @@ func (asw *actualStateOfWorld) MarkDesireToDetach(
 		volumeObj.nodesAttachedTo[nodeName] = nodeObj
 	}
 
+	// Remove volume from volumes to report as attached
+	nodeToUpdate, nodeToUpdateExists := asw.nodesToUpdateStatusFor[nodeName]
+	if nodeToUpdateExists {
+		_, nodeToUpdateVolumeExists :=
+			nodeToUpdate.volumesToReportAsAttached[volumeName]
+		if nodeToUpdateVolumeExists {
+			nodeToUpdate.statusUpdateNeeded = true
+			delete(nodeToUpdate.volumesToReportAsAttached, volumeName)
+			asw.nodesToUpdateStatusFor[nodeName] = nodeToUpdate
+		}
+	}
+
 	return time.Since(volumeObj.nodesAttachedTo[nodeName].detachRequestedTime), nil
 }
 
+func (asw *actualStateOfWorld) ResetNodeStatusUpdateNeeded(
+	nodeName string) error {
+	asw.Lock()
+	defer asw.Unlock()
+	// Remove volume from volumes to report as attached
+	nodeToUpdate, nodeToUpdateExists := asw.nodesToUpdateStatusFor[nodeName]
+	if !nodeToUpdateExists {
+		return fmt.Errorf(
+			"failed to ResetNodeStatusUpdateNeeded(nodeName=%q) nodeName does not exist",
+			nodeName)
+	}
+
+	nodeToUpdate.statusUpdateNeeded = false
+	asw.nodesToUpdateStatusFor[nodeName] = nodeToUpdate
+	return nil
+}
+
 func (asw *actualStateOfWorld) DeleteVolumeNode(
 	volumeName api.UniqueVolumeName, nodeName string) {
 	asw.Lock()
@@ -319,6 +417,18 @@ func (asw *actualStateOfWorld) DeleteVolumeNode(
 	if len(volumeObj.nodesAttachedTo) == 0 {
 		delete(asw.attachedVolumes, volumeName)
 	}
+
+	// Remove volume from volumes to report as attached
+	nodeToUpdate, nodeToUpdateExists := asw.nodesToUpdateStatusFor[nodeName]
+	if nodeToUpdateExists {
+		_, nodeToUpdateVolumeExists :=
+			nodeToUpdate.volumesToReportAsAttached[volumeName]
+		if nodeToUpdateVolumeExists {
+			nodeToUpdate.statusUpdateNeeded = true
+			delete(nodeToUpdate.volumesToReportAsAttached, volumeName)
+			asw.nodesToUpdateStatusFor[nodeName] = nodeToUpdate
+		}
+	}
 }
 
 func (asw *actualStateOfWorld) VolumeNodeExists(
@@ -372,6 +482,31 @@ func (asw *actualStateOfWorld) GetAttachedVolumesForNode(
 	return attachedVolumes
 }
 
+func (asw *actualStateOfWorld) GetVolumesToReportAttached() map[string][]api.AttachedVolume {
+	asw.RLock()
+	defer asw.RUnlock()
+
+	volumesToReportAttached := make(map[string][]api.AttachedVolume)
+	for _, nodeToUpdateObj := range asw.nodesToUpdateStatusFor {
+		if nodeToUpdateObj.statusUpdateNeeded {
+			attachedVolumes := make(
+				[]api.AttachedVolume,
+				len(nodeToUpdateObj.volumesToReportAsAttached) /* len */)
+			i := 0
+			for _, volume := range nodeToUpdateObj.volumesToReportAsAttached {
+				attachedVolumes[i] = api.AttachedVolume{
+					Name:       volume,
+					DevicePath: asw.attachedVolumes[volume].devicePath,
+				}
+				i++
+			}
+			volumesToReportAttached[nodeToUpdateObj.nodeName] = attachedVolumes
+		}
+	}
+
+	return volumesToReportAttached
+}
+
 func getAttachedVolume(
 	attachedVolume *attachedVolume,
 	nodeAttachedTo *nodeAttachedTo) AttachedVolume {

pkg/controller/volume/reconciler/reconciler.go

@@ -24,6 +24,8 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/controller/volume/cache"
+	"k8s.io/kubernetes/pkg/controller/volume/statusupdater"
+	"k8s.io/kubernetes/pkg/util/goroutinemap"
 	"k8s.io/kubernetes/pkg/util/wait"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
 )
@@ -55,13 +57,15 @@ func NewReconciler(
 	maxWaitForUnmountDuration time.Duration,
 	desiredStateOfWorld cache.DesiredStateOfWorld,
 	actualStateOfWorld cache.ActualStateOfWorld,
-	attacherDetacher operationexecutor.OperationExecutor) Reconciler {
+	attacherDetacher operationexecutor.OperationExecutor,
+	nodeStatusUpdater statusupdater.NodeStatusUpdater) Reconciler {
 	return &reconciler{
 		loopPeriod:                loopPeriod,
 		maxWaitForUnmountDuration: maxWaitForUnmountDuration,
 		desiredStateOfWorld:       desiredStateOfWorld,
 		actualStateOfWorld:        actualStateOfWorld,
 		attacherDetacher:          attacherDetacher,
+		nodeStatusUpdater:         nodeStatusUpdater,
 	}
 }
 
@@ -71,6 +75,7 @@ type reconciler struct {
 	desiredStateOfWorld       cache.DesiredStateOfWorld
 	actualStateOfWorld        cache.ActualStateOfWorld
 	attacherDetacher          operationexecutor.OperationExecutor
+	nodeStatusUpdater         statusupdater.NodeStatusUpdater
 }
 
 func (rc *reconciler) Run(stopCh <-chan struct{}) {
@@ -88,10 +93,22 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 				attachedVolume.VolumeName, attachedVolume.NodeName) {
 				// Volume exists in actual state of world but not desired
 				if !attachedVolume.MountedByNode {
-					glog.V(5).Infof("Attempting to start DetachVolume for volume %q to node %q", attachedVolume.VolumeName, attachedVolume.NodeName)
+					glog.V(5).Infof("Attempting to start DetachVolume for volume %q from node %q", attachedVolume.VolumeName, attachedVolume.NodeName)
 					err := rc.attacherDetacher.DetachVolume(attachedVolume.AttachedVolume, rc.actualStateOfWorld)
 					if err == nil {
-						glog.Infof("Started DetachVolume for volume %q to node %q", attachedVolume.VolumeName, attachedVolume.NodeName)
+						glog.Infof("Started DetachVolume for volume %q from node %q", attachedVolume.VolumeName, attachedVolume.NodeName)
+					}
+					if err != nil &&
+						!goroutinemap.IsAlreadyExists(err) &&
+						!goroutinemap.IsExponentialBackoff(err) {
+						// Ignore goroutinemap.IsAlreadyExists && goroutinemap.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",
+							attachedVolume.VolumeName,
+							attachedVolume.VolumeSpec.Name(),
+							attachedVolume.NodeName,
+							err)
 					}
 				} else {
 					// If volume is not safe to detach (is mounted) wait a max amount of time before detaching any way.
@@ -100,10 +117,22 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 						glog.Errorf("Unexpected error actualStateOfWorld.MarkDesireToDetach(): %v", err)
 					}
 					if timeElapsed > rc.maxWaitForUnmountDuration {
-						glog.V(5).Infof("Attempting to start DetachVolume for volume %q to node %q. Volume is not safe to detach, but maxWaitForUnmountDuration expired.", attachedVolume.VolumeName, attachedVolume.NodeName)
+						glog.V(5).Infof("Attempting to start DetachVolume for volume %q from node %q. Volume is not safe to detach, but maxWaitForUnmountDuration expired.", attachedVolume.VolumeName, attachedVolume.NodeName)
 						err := rc.attacherDetacher.DetachVolume(attachedVolume.AttachedVolume, rc.actualStateOfWorld)
 						if err == nil {
-							glog.Infof("Started DetachVolume for volume %q to node %q due to maxWaitForUnmountDuration expiry.", attachedVolume.VolumeName, attachedVolume.NodeName)
+							glog.Infof("Started DetachVolume for volume %q from node %q due to maxWaitForUnmountDuration expiry.", attachedVolume.VolumeName, attachedVolume.NodeName)
+						}
+						if err != nil &&
+							!goroutinemap.IsAlreadyExists(err) &&
+							!goroutinemap.IsExponentialBackoff(err) {
+							// Ignore goroutinemap.IsAlreadyExists && goroutinemap.IsExponentialBackoff errors, they are expected.
+							// Log all other errors.
+							glog.Errorf(
+								"operationExecutor.DetachVolume failed to start (maxWaitForUnmountDuration expiry) for volume %q (spec.Name: %q) from node %q with err: %v",
+								attachedVolume.VolumeName,
+								attachedVolume.VolumeSpec.Name(),
+								attachedVolume.NodeName,
+								err)
 						}
 					}
 				}
@@ -117,7 +146,7 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 				// Volume/Node exists, touch it to reset detachRequestedTime
 				glog.V(12).Infof("Volume %q/Node %q is attached--touching.", volumeToAttach.VolumeName, volumeToAttach.NodeName)
 				_, err := rc.actualStateOfWorld.AddVolumeNode(
-					volumeToAttach.VolumeSpec, volumeToAttach.NodeName)
+					volumeToAttach.VolumeSpec, volumeToAttach.NodeName, "" /* devicePath */)
 				if err != nil {
 					glog.Errorf("Unexpected error on actualStateOfWorld.AddVolumeNode(): %v", err)
 				}
@@ -128,7 +157,25 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 				if err == nil {
 					glog.Infof("Started AttachVolume for volume %q to node %q", volumeToAttach.VolumeName, volumeToAttach.NodeName)
 				}
+				if err != nil &&
+					!goroutinemap.IsAlreadyExists(err) &&
+					!goroutinemap.IsExponentialBackoff(err) {
+					// Ignore goroutinemap.IsAlreadyExists && goroutinemap.IsExponentialBackoff errors, they are expected.
+					// Log all other errors.
+					glog.Errorf(
+						"operationExecutor.AttachVolume failed to start for volume %q (spec.Name: %q) to node %q with err: %v",
+						volumeToAttach.VolumeName,
+						volumeToAttach.VolumeSpec.Name(),
+						volumeToAttach.NodeName,
+						err)
+				}
 			}
 		}
+
+		// Update Node Status
+		err := rc.nodeStatusUpdater.UpdateNodeStatuses()
+		if err != nil {
+			glog.Infof("UpdateNodeStatuses failed with: %v", err)
+		}
 	}
 }

pkg/controller/volume/reconciler/reconciler_test.go

@@ -21,7 +21,9 @@ import (
 	"time"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/controller/framework/informers"
 	"k8s.io/kubernetes/pkg/controller/volume/cache"
+	"k8s.io/kubernetes/pkg/controller/volume/statusupdater"
 	controllervolumetesting "k8s.io/kubernetes/pkg/controller/volume/testing"
 	"k8s.io/kubernetes/pkg/util/wait"
 	volumetesting "k8s.io/kubernetes/pkg/volume/testing"
@@ -32,6 +34,7 @@ import (
 const (
 	reconcilerLoopPeriod      time.Duration = 0 * time.Millisecond
 	maxWaitForUnmountDuration time.Duration = 50 * time.Millisecond
+	resyncPeriod              time.Duration = 5 * time.Minute
 )
 
 // Calls Run()
@@ -41,9 +44,15 @@ func Test_Run_Positive_DoNothing(t *testing.T) {
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
-	ad := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	fakeKubeClient := controllervolumetesting.CreateTestClient()
+	ad := operationexecutor.NewOperationExecutor(
+		fakeKubeClient, volumePluginMgr)
+	nodeInformer := informers.CreateSharedNodeIndexInformer(
+		fakeKubeClient, resyncPeriod)
+	nsu := statusupdater.NewNodeStatusUpdater(
+		fakeKubeClient, nodeInformer, asw)
 	reconciler := NewReconciler(
-		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad)
+		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad, nsu)
 
 	// Act
 	go reconciler.Run(wait.NeverStop)
@@ -64,9 +73,14 @@ func Test_Run_Positive_OneDesiredVolumeAttach(t *testing.T) {
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
-	ad := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	fakeKubeClient := controllervolumetesting.CreateTestClient()
+	ad := operationexecutor.NewOperationExecutor(fakeKubeClient, volumePluginMgr)
+	nodeInformer := informers.CreateSharedNodeIndexInformer(
+		fakeKubeClient, resyncPeriod)
+	nsu := statusupdater.NewNodeStatusUpdater(
+		fakeKubeClient, nodeInformer, asw)
 	reconciler := NewReconciler(
-		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad)
+		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad, nsu)
 	podName := types.UniquePodName("pod-uid")
 	volumeName := api.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
@@ -105,9 +119,14 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithUnmountedVolume(t *te
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
-	ad := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	fakeKubeClient := controllervolumetesting.CreateTestClient()
+	ad := operationexecutor.NewOperationExecutor(fakeKubeClient, volumePluginMgr)
+	nodeInformer := informers.CreateSharedNodeIndexInformer(
+		fakeKubeClient, resyncPeriod)
+	nsu := statusupdater.NewNodeStatusUpdater(
+		fakeKubeClient, nodeInformer, asw)
 	reconciler := NewReconciler(
-		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad)
+		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad, nsu)
 	podName := types.UniquePodName("pod-uid")
 	volumeName := api.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
@@ -167,9 +186,14 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithMountedVolume(t *test
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
-	ad := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	fakeKubeClient := controllervolumetesting.CreateTestClient()
+	ad := operationexecutor.NewOperationExecutor(fakeKubeClient, volumePluginMgr)
+	nodeInformer := informers.CreateSharedNodeIndexInformer(
+		fakeKubeClient, resyncPeriod)
+	nsu := statusupdater.NewNodeStatusUpdater(
+		fakeKubeClient, nodeInformer, asw)
 	reconciler := NewReconciler(
-		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad)
+		reconcilerLoopPeriod, maxWaitForUnmountDuration, dsw, asw, ad, nsu)
 	podName := types.UniquePodName("pod-uid")
 	volumeName := api.UniqueVolumeName("volume-name")
 	volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
@@ -379,6 +403,3 @@ func retryWithExponentialBackOff(initialDuration time.Duration, fn wait.Conditio
 	}
 	return wait.ExponentialBackoff(backoff, fn)
 }
-
-// t.Logf("asw: %v", asw.GetAttachedVolumes())
-// t.Logf("dsw: %v", dsw.GetVolumesToAttach())

pkg/controller/volume/statusupdater/node_status_updater.go

+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package statusupdater implements interfaces that enable updating the status
+// of API objects.
+package statusupdater
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/golang/glog"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
+	"k8s.io/kubernetes/pkg/controller/framework"
+	"k8s.io/kubernetes/pkg/controller/volume/cache"
+	"k8s.io/kubernetes/pkg/util/strategicpatch"
+)
+
+// NodeStatusUpdater defines a set of operations for updating the
+// VolumesAttached field in the Node Status.
+type NodeStatusUpdater interface {
+	// Gets a list of node statuses that should be updated from the actual state
+	// of the world and updates them.
+	UpdateNodeStatuses() error
+}
+
+// NewNodeStatusUpdater returns a new instance of NodeStatusUpdater.
+func NewNodeStatusUpdater(
+	kubeClient internalclientset.Interface,
+	nodeInformer framework.SharedInformer,
+	actualStateOfWorld cache.ActualStateOfWorld) NodeStatusUpdater {
+	return &nodeStatusUpdater{
+		actualStateOfWorld: actualStateOfWorld,
+		nodeInformer:       nodeInformer,
+		kubeClient:         kubeClient,
+	}
+}
+
+type nodeStatusUpdater struct {
+	kubeClient         internalclientset.Interface
+	nodeInformer       framework.SharedInformer
+	actualStateOfWorld cache.ActualStateOfWorld
+}
+
+func (nsu *nodeStatusUpdater) UpdateNodeStatuses() error {
+	nodesToUpdate := nsu.actualStateOfWorld.GetVolumesToReportAttached()
+	for nodeName, attachedVolumes := range nodesToUpdate {
+		nodeObj, exists, err := nsu.nodeInformer.GetStore().GetByKey(nodeName)
+		if nodeObj == nil || !exists || err != nil {
+			return fmt.Errorf(
+				"failed to find node %q in NodeInformer cache. %v",
+				nodeName,
+				err)
+		}
+
+		node, ok := nodeObj.(*api.Node)
+		if !ok || node == nil {
+			return fmt.Errorf(
+				"failed to cast %q object %#v to Node",
+				nodeName,
+				nodeObj)
+		}
+
+		oldData, err := json.Marshal(node)
+		if err != nil {
+			return fmt.Errorf(
+				"failed to Marshal oldData for node %q. %v",
+				nodeName,
+				err)
+		}
+
+		node.Status.VolumesAttached = attachedVolumes
+
+		newData, err := json.Marshal(node)
+		if err != nil {
+			return fmt.Errorf(
+				"failed to Marshal newData for node %q. %v",
+				nodeName,
+				err)
+		}
+
+		patchBytes, err :=
+			strategicpatch.CreateStrategicMergePatch(oldData, newData, node)
+		if err != nil {
+			return fmt.Errorf(
+				"failed to CreateStrategicMergePatch for node %q. %v",
+				nodeName,
+				err)
+		}
+
+		_, err = nsu.kubeClient.Core().Nodes().PatchStatus(nodeName, patchBytes)
+		if err != nil {
+			return fmt.Errorf(
+				"failed to kubeClient.Core().Nodes().Patch for node %q. %v",
+				nodeName,
+				err)
+		}
+
+		err = nsu.actualStateOfWorld.ResetNodeStatusUpdateNeeded(nodeName)
+		if err != nil {
+			return fmt.Errorf(
+				"failed to ResetNodeStatusUpdateNeeded for node %q. %v",
+				nodeName,
+				err)
+		}
+
+		glog.V(3).Infof(
+			"Updating status for node %q succeeded. patchBytes: %q",
+			string(patchBytes))
+	}
+	return nil
+}

pkg/controller/volume/testing/testvolumespec.go

@@ -17,8 +17,14 @@ limitations under the License.
 package testing
 
 import (
+	"fmt"
+
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
+	"k8s.io/kubernetes/pkg/client/testing/core"
+	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/watch"
 )
 
 // GetTestVolumeSpec returns a test volume spec
@@ -36,3 +42,62 @@ func GetTestVolumeSpec(volumeName string, diskName api.UniqueVolumeName) *volume
 		},
 	}
 }
+
+func CreateTestClient() *fake.Clientset {
+	fakeClient := &fake.Clientset{}
+
+	fakeClient.AddReactor("list", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		obj := &api.PodList{}
+		podNamePrefix := "mypod"
+		namespace := "mynamespace"
+		for i := 0; i < 5; i++ {
+			podName := fmt.Sprintf("%s-%d", podNamePrefix, i)
+			pod := api.Pod{
+				Status: api.PodStatus{
+					Phase: api.PodRunning,
+				},
+				ObjectMeta: api.ObjectMeta{
+					Name:      podName,
+					Namespace: namespace,
+					Labels: map[string]string{
+						"name": podName,
+					},
+				},
+				Spec: api.PodSpec{
+					Containers: []api.Container{
+						{
+							Name:  "containerName",
+							Image: "containerImage",
+							VolumeMounts: []api.VolumeMount{
+								{
+									Name:      "volumeMountName",
+									ReadOnly:  false,
+									MountPath: "/mnt",
+								},
+							},
+						},
+					},
+					Volumes: []api.Volume{
+						{
+							Name: "volumeName",
+							VolumeSource: api.VolumeSource{
+								GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{
+									PDName:   "pdName",
+									FSType:   "ext4",
+									ReadOnly: false,
+								},
+							},
+						},
+					},
+				},
+			}
+			obj.Items = append(obj.Items, pod)
+		}
+		return true, obj, nil
+	})
+
+	fakeWatch := watch.NewFake()
+	fakeClient.AddWatchReactor("*", core.DefaultWatchReactor(fakeWatch, nil))
+
+	return fakeClient
+}

pkg/kubelet/kubelet_test.go

@@ -297,7 +297,7 @@ func newTestKubeletWithImageList(
 		controllerAttachDetachEnabled,
 		kubelet.hostname,
 		kubelet.podManager,
-		kubelet.kubeClient,
+		fakeKubeClient,
 		kubelet.volumePluginMgr)
 	if err != nil {
 		t.Fatalf("failed to initialize volume manager: %v", err)
@@ -617,6 +617,24 @@ func TestVolumeAttachAndMountControllerEnabled(t *testing.T) {
 	testKubelet := newTestKubelet(t, true /* controllerAttachDetachEnabled */)
 	kubelet := testKubelet.kubelet
 	kubelet.mounter = &mount.FakeMounter{}
+	kubeClient := testKubelet.fakeKubeClient
+	kubeClient.AddReactor("get", "nodes",
+		func(action core.Action) (bool, runtime.Object, error) {
+			return true, &api.Node{
+				ObjectMeta: api.ObjectMeta{Name: testKubeletHostname},
+				Status: api.NodeStatus{
+					VolumesAttached: []api.AttachedVolume{
+						{
+							Name:       "fake/vol1",
+							DevicePath: "fake/path",
+						},
+					}},
+				Spec: api.NodeSpec{ExternalID: testKubeletHostname},
+			}, nil
+		})
+	kubeClient.AddReactor("*", "*", func(action core.Action) (bool, runtime.Object, error) {
+		return true, nil, fmt.Errorf("no reaction implemented for %s", action)
+	})
 
 	pod := podWithUidNameNsSpec("12345678", "foo", "test", api.PodSpec{
 		Volumes: []api.Volume{
@@ -687,6 +705,24 @@ func TestVolumeUnmountAndDetachControllerEnabled(t *testing.T) {
 	testKubelet := newTestKubelet(t, true /* controllerAttachDetachEnabled */)
 	kubelet := testKubelet.kubelet
 	kubelet.mounter = &mount.FakeMounter{}
+	kubeClient := testKubelet.fakeKubeClient
+	kubeClient.AddReactor("get", "nodes",
+		func(action core.Action) (bool, runtime.Object, error) {
+			return true, &api.Node{
+				ObjectMeta: api.ObjectMeta{Name: testKubeletHostname},
+				Status: api.NodeStatus{
+					VolumesAttached: []api.AttachedVolume{
+						{
+							Name:       "fake/vol1",
+							DevicePath: "fake/path",
+						},
+					}},
+				Spec: api.NodeSpec{ExternalID: testKubeletHostname},
+			}, nil
+		})
+	kubeClient.AddReactor("*", "*", func(action core.Action) (bool, runtime.Object, error) {
+		return true, nil, fmt.Errorf("no reaction implemented for %s", action)
+	})
 
 	pod := podWithUidNameNsSpec("12345678", "foo", "test", api.PodSpec{
 		Volumes: []api.Volume{

pkg/kubelet/volume/cache/actual_state_of_world.go

@@ -57,7 +57,7 @@ type ActualStateOfWorld interface {
 	// If a volume with the same generated name already exists, this is a noop.
 	// If no volume plugin can support the given volumeSpec or more than one
 	// plugin can support it, an error is returned.
-	AddVolume(volumeSpec *volume.Spec) (api.UniqueVolumeName, error)
+	AddVolume(volumeSpec *volume.Spec, devicePath string) (api.UniqueVolumeName, error)
 
 	// AddPodToVolume adds the given pod to the given volume in the cache
 	// indicating the specified volume has been successfully mounted to the
@@ -108,14 +108,14 @@ type ActualStateOfWorld interface {
 	// If a volume with the name volumeName does not exist in the list of
 	// attached volumes, a volumeNotAttachedError is returned indicating the
 	// given volume is not yet attached.
-	// If a the given volumeName/podName combo exists but the value of
+	// If the given volumeName/podName combo exists but the value of
 	// remountRequired is true, a remountRequiredError is returned indicating
 	// the given volume has been successfully mounted to this pod but should be
 	// remounted to reflect changes in the referencing pod. Atomically updating
 	// volumes, depend on this to update the contents of the volume.
 	// All volume mounting calls should be idempotent so a second mount call for
 	// volumes that do not need to update contents should not fail.
-	PodExistsInVolume(podName volumetypes.UniquePodName, volumeName api.UniqueVolumeName) (bool, error)
+	PodExistsInVolume(podName volumetypes.UniquePodName, volumeName api.UniqueVolumeName) (bool, string, error)
 
 	// GetMountedVolumes generates and returns a list of volumes and the pods
 	// they are successfully attached and mounted for based on the current
@@ -224,9 +224,13 @@ type attachedVolume struct {
 	pluginIsAttachable bool
 
 	// globallyMounted indicates that the volume is mounted to the underlying
-	// device at a global mount point. This global mount point must unmounted
+	// device at a global mount point. This global mount point must be unmounted
 	// prior to detach.
 	globallyMounted bool
+
+	// devicePath contains the path on the node where the volume is attached for
+	// attachable volumes
+	devicePath string
 }
 
 // The mountedPod object represents a pod for which the kubelet volume manager
@@ -260,8 +264,8 @@ type mountedPod struct {
 }
 
 func (asw *actualStateOfWorld) MarkVolumeAsAttached(
-	volumeSpec *volume.Spec, nodeName string) error {
-	_, err := asw.AddVolume(volumeSpec)
+	volumeSpec *volume.Spec, nodeName string, devicePath string) error {
+	_, err := asw.AddVolume(volumeSpec, devicePath)
 	return err
 }
 
@@ -302,7 +306,7 @@ func (asw *actualStateOfWorld) MarkDeviceAsUnmounted(
 }
 
 func (asw *actualStateOfWorld) AddVolume(
-	volumeSpec *volume.Spec) (api.UniqueVolumeName, error) {
+	volumeSpec *volume.Spec, devicePath string) (api.UniqueVolumeName, error) {
 	asw.Lock()
 	defer asw.Unlock()
 
@@ -338,6 +342,7 @@ func (asw *actualStateOfWorld) AddVolume(
 			pluginName:         volumePlugin.GetPluginName(),
 			pluginIsAttachable: pluginIsAttachable,
 			globallyMounted:    false,
+			devicePath:         devicePath,
 		}
 		asw.attachedVolumes[volumeName] = volumeObj
 	}
@@ -469,21 +474,22 @@ func (asw *actualStateOfWorld) DeleteVolume(volumeName api.UniqueVolumeName) err
 }
 
 func (asw *actualStateOfWorld) PodExistsInVolume(
-	podName volumetypes.UniquePodName, volumeName api.UniqueVolumeName) (bool, error) {
+	podName volumetypes.UniquePodName,
+	volumeName api.UniqueVolumeName) (bool, string, error) {
 	asw.RLock()
 	defer asw.RUnlock()
 
 	volumeObj, volumeExists := asw.attachedVolumes[volumeName]
 	if !volumeExists {
-		return false, newVolumeNotAttachedError(volumeName)
+		return false, "", newVolumeNotAttachedError(volumeName)
 	}
 
 	podObj, podExists := volumeObj.mountedPods[podName]
 	if podExists && podObj.remountRequired {
-		return true, newRemountRequiredError(volumeObj.volumeName, podObj.podName)
+		return true, volumeObj.devicePath, newRemountRequiredError(volumeObj.volumeName, podObj.podName)
 	}
 
-	return podExists, nil
+	return podExists, volumeObj.devicePath, nil
 }
 
 func (asw *actualStateOfWorld) GetMountedVolumes() []MountedVolume {

pkg/kubelet/volume/cache/actual_state_of_world_test.go

@@ -51,9 +51,10 @@ func Test_AddVolume_Positive_NewVolume(t *testing.T) {
 		},
 	}
 	volumeSpec := &volume.Spec{Volume: &pod.Spec.Volumes[0]}
+	devicePath := "fake/device/path"
 
 	// Act
-	generatedVolumeName, err := asw.AddVolume(volumeSpec)
+	generatedVolumeName, err := asw.AddVolume(volumeSpec, devicePath)
 
 	// Assert
 	if err != nil {
@@ -69,6 +70,7 @@ func Test_AddVolume_Positive_NewVolume(t *testing.T) {
 func Test_AddVolume_Positive_ExistingVolume(t *testing.T) {
 	// Arrange
 	volumePluginMgr, _ := volumetesting.GetTestVolumePluginMgr(t)
+	devicePath := "fake/device/path"
 	asw := NewActualStateOfWorld("mynode" /* nodeName */, volumePluginMgr)
 	pod := &api.Pod{
 		ObjectMeta: api.ObjectMeta{
@@ -90,13 +92,13 @@ func Test_AddVolume_Positive_ExistingVolume(t *testing.T) {
 	}
 
 	volumeSpec := &volume.Spec{Volume: &pod.Spec.Volumes[0]}
-	generatedVolumeName, err := asw.AddVolume(volumeSpec)
+	generatedVolumeName, err := asw.AddVolume(volumeSpec, devicePath)
 	if err != nil {
 		t.Fatalf("AddVolume failed. Expected: <no error> Actual: <%v>", err)
 	}
 
 	// Act
-	generatedVolumeName, err = asw.AddVolume(volumeSpec)
+	generatedVolumeName, err = asw.AddVolume(volumeSpec, devicePath)
 
 	// Assert
 	if err != nil {
@@ -113,6 +115,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeNewNode(t *testing.T) {
 	// Arrange
 	volumePluginMgr, plugin := volumetesting.GetTestVolumePluginMgr(t)
 	asw := NewActualStateOfWorld("mynode" /* nodeName */, volumePluginMgr)
+	devicePath := "fake/device/path"
 
 	pod := &api.Pod{
 		ObjectMeta: api.ObjectMeta{
@@ -137,7 +140,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeNewNode(t *testing.T) {
 	volumeName, err := volumehelper.GetUniqueVolumeNameFromSpec(
 		plugin, volumeSpec)
 
-	generatedVolumeName, err := asw.AddVolume(volumeSpec)
+	generatedVolumeName, err := asw.AddVolume(volumeSpec, devicePath)
 	if err != nil {
 		t.Fatalf("AddVolume failed. Expected: <no error> Actual: <%v>", err)
 	}
@@ -158,7 +161,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeNewNode(t *testing.T) {
 	}
 
 	verifyVolumeExistsInAttachedVolumes(t, generatedVolumeName, asw)
-	verifyPodExistsInVolumeAsw(t, podName, generatedVolumeName, asw)
+	verifyPodExistsInVolumeAsw(t, podName, generatedVolumeName, "fake/device/path" /* expectedDevicePath */, asw)
 }
 
 // Populates data struct with a volume
@@ -169,6 +172,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeExistingNode(t *testing.T) {
 	// Arrange
 	volumePluginMgr, plugin := volumetesting.GetTestVolumePluginMgr(t)
 	asw := NewActualStateOfWorld("mynode" /* nodeName */, volumePluginMgr)
+	devicePath := "fake/device/path"
 
 	pod := &api.Pod{
 		ObjectMeta: api.ObjectMeta{
@@ -193,7 +197,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeExistingNode(t *testing.T) {
 	volumeName, err := volumehelper.GetUniqueVolumeNameFromSpec(
 		plugin, volumeSpec)
 
-	generatedVolumeName, err := asw.AddVolume(volumeSpec)
+	generatedVolumeName, err := asw.AddVolume(volumeSpec, devicePath)
 	if err != nil {
 		t.Fatalf("AddVolume failed. Expected: <no error> Actual: <%v>", err)
 	}
@@ -220,7 +224,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeExistingNode(t *testing.T) {
 	}
 
 	verifyVolumeExistsInAttachedVolumes(t, generatedVolumeName, asw)
-	verifyPodExistsInVolumeAsw(t, podName, generatedVolumeName, asw)
+	verifyPodExistsInVolumeAsw(t, podName, generatedVolumeName, "fake/device/path" /* expectedDevicePath */, asw)
 }
 
 // Calls AddPodToVolume() to add pod to empty data stuct
@@ -316,8 +320,9 @@ func verifyPodExistsInVolumeAsw(
 	t *testing.T,
 	expectedPodName volumetypes.UniquePodName,
 	expectedVolumeName api.UniqueVolumeName,
+	expectedDevicePath string,
 	asw ActualStateOfWorld) {
-	podExistsInVolume, err :=
+	podExistsInVolume, devicePath, err :=
 		asw.PodExistsInVolume(expectedPodName, expectedVolumeName)
 	if err != nil {
 		t.Fatalf(
@@ -329,6 +334,13 @@ func verifyPodExistsInVolumeAsw(
 			"ASW PodExistsInVolume result invalid. Expected: <true> Actual: <%v>",
 			podExistsInVolume)
 	}
+
+	if devicePath != expectedDevicePath {
+		t.Fatalf(
+			"Invalid devicePath. Expected: <%q> Actual: <%q> ",
+			expectedDevicePath,
+			devicePath)
+	}
 }
 
 func verifyPodDoesntExistInVolumeAsw(
@@ -337,7 +349,7 @@ func verifyPodDoesntExistInVolumeAsw(
 	volumeToCheck api.UniqueVolumeName,
 	expectVolumeToExist bool,
 	asw ActualStateOfWorld) {
-	podExistsInVolume, err :=
+	podExistsInVolume, devicePath, err :=
 		asw.PodExistsInVolume(podToCheck, volumeToCheck)
 	if !expectVolumeToExist && err == nil {
 		t.Fatalf(
@@ -354,4 +366,10 @@ func verifyPodDoesntExistInVolumeAsw(
 			"ASW PodExistsInVolume result invalid. Expected: <false> Actual: <%v>",
 			podExistsInVolume)
 	}
+
+	if devicePath != "" {
+		t.Fatalf(
+			"Invalid devicePath. Expected: <\"\"> Actual: <%q> ",
+			devicePath)
+	}
 }

pkg/kubelet/volume/cache/desired_state_of_world.go

@@ -84,8 +84,8 @@ type DesiredStateOfWorld interface {
 	GetVolumesToMount() []VolumeToMount
 }
 
-// VolumeToMount represents a volume that should be attached to this node and
-// mounted to the PodName.
+// VolumeToMount represents a volume that is attached to this node and needs to
+// be mounted to PodName.
 type VolumeToMount struct {
 	operationexecutor.VolumeToMount
 }

pkg/kubelet/volume/reconciler/reconciler.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/kubelet/volume/cache"
 	"k8s.io/kubernetes/pkg/util/goroutinemap"
 	"k8s.io/kubernetes/pkg/util/wait"
@@ -62,6 +63,7 @@ type Reconciler interface {
 //   safely (prevents more than one operation from being triggered on the same
 //   volume)
 func NewReconciler(
+	kubeClient internalclientset.Interface,
 	controllerAttachDetachEnabled bool,
 	loopSleepDuration time.Duration,
 	waitForAttachTimeout time.Duration,
@@ -70,6 +72,7 @@ func NewReconciler(
 	actualStateOfWorld cache.ActualStateOfWorld,
 	operationExecutor operationexecutor.OperationExecutor) Reconciler {
 	return &reconciler{
+		kubeClient:                    kubeClient,
 		controllerAttachDetachEnabled: controllerAttachDetachEnabled,
 		loopSleepDuration:             loopSleepDuration,
 		waitForAttachTimeout:          waitForAttachTimeout,
@@ -81,6 +84,7 @@ func NewReconciler(
 }
 
 type reconciler struct {
+	kubeClient                    internalclientset.Interface
 	controllerAttachDetachEnabled bool
 	loopSleepDuration             time.Duration
 	waitForAttachTimeout          time.Duration
@@ -112,8 +116,10 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 					mountedVolume.PodUID)
 				err := rc.operationExecutor.UnmountVolume(
 					mountedVolume.MountedVolume, rc.actualStateOfWorld)
-				if err != nil && !goroutinemap.IsAlreadyExists(err) {
-					// Ignore goroutinemap.IsAlreadyExists errors, they are expected.
+				if err != nil &&
+					!goroutinemap.IsAlreadyExists(err) &&
+					!goroutinemap.IsExponentialBackoff(err) {
+					// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
 					// Log all other errors.
 					glog.Errorf(
 						"operationExecutor.UnmountVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
@@ -136,25 +142,37 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 
 		// Ensure volumes that should be attached/mounted are attached/mounted.
 		for _, volumeToMount := range rc.desiredStateOfWorld.GetVolumesToMount() {
-			volMounted, err := rc.actualStateOfWorld.PodExistsInVolume(volumeToMount.PodName, volumeToMount.VolumeName)
+			volMounted, devicePath, err := rc.actualStateOfWorld.PodExistsInVolume(volumeToMount.PodName, volumeToMount.VolumeName)
+			volumeToMount.DevicePath = devicePath
 			if cache.IsVolumeNotAttachedError(err) {
-				// Volume is not attached, it should be
 				if rc.controllerAttachDetachEnabled || !volumeToMount.PluginIsAttachable {
-					// Kubelet not responsible for attaching or this volume has a non-attachable volume plugin,
-					// so just add it to actualStateOfWorld without attach.
-					markVolumeAttachErr := rc.actualStateOfWorld.MarkVolumeAsAttached(
-						volumeToMount.VolumeSpec, rc.hostName)
-					if markVolumeAttachErr != nil {
+					// Volume is not attached (or doesn't implement attacher), kubelet attach is disabled, wait
+					// for controller to finish attaching volume.
+					glog.V(12).Infof("Attempting to start VerifyControllerAttachedVolume for volume %q (spec.Name: %q) pod %q (UID: %q)",
+						volumeToMount.VolumeName,
+						volumeToMount.VolumeSpec.Name(),
+						volumeToMount.PodName,
+						volumeToMount.Pod.UID)
+					err := rc.operationExecutor.VerifyControllerAttachedVolume(
+						volumeToMount.VolumeToMount,
+						rc.hostName,
+						rc.actualStateOfWorld)
+					if err != nil &&
+						!goroutinemap.IsAlreadyExists(err) &&
+						!goroutinemap.IsExponentialBackoff(err) {
+						// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
+						// Log all other errors.
 						glog.Errorf(
-							"actualStateOfWorld.MarkVolumeAsAttached failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
+							"operationExecutor.VerifyControllerAttachedVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
 							volumeToMount.VolumeName,
 							volumeToMount.VolumeSpec.Name(),
 							volumeToMount.PodName,
 							volumeToMount.Pod.UID,
 							rc.controllerAttachDetachEnabled,
-							markVolumeAttachErr)
-					} else {
-						glog.V(12).Infof("actualStateOfWorld.MarkVolumeAsAttached succeeded for volume %q (spec.Name: %q) pod %q (UID: %q)",
+							err)
+					}
+					if err == nil {
+						glog.Infof("VerifyControllerAttachedVolume operation started for volume %q (spec.Name: %q) pod %q (UID: %q)",
 							volumeToMount.VolumeName,
 							volumeToMount.VolumeSpec.Name(),
 							volumeToMount.PodName,
@@ -174,8 +192,10 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 						volumeToMount.PodName,
 						volumeToMount.Pod.UID)
 					err := rc.operationExecutor.AttachVolume(volumeToAttach, rc.actualStateOfWorld)
-					if err != nil && !goroutinemap.IsAlreadyExists(err) {
-						// Ignore goroutinemap.IsAlreadyExists errors, they are expected.
+					if err != nil &&
+						!goroutinemap.IsAlreadyExists(err) &&
+						!goroutinemap.IsExponentialBackoff(err) {
+						// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
 						// Log all other errors.
 						glog.Errorf(
 							"operationExecutor.AttachVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
@@ -210,8 +230,10 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 					rc.waitForAttachTimeout,
 					volumeToMount.VolumeToMount,
 					rc.actualStateOfWorld)
-				if err != nil && !goroutinemap.IsAlreadyExists(err) {
-					// Ignore goroutinemap.IsAlreadyExists errors, they are expected.
+				if err != nil &&
+					!goroutinemap.IsAlreadyExists(err) &&
+					!goroutinemap.IsExponentialBackoff(err) {
+					// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
 					// Log all other errors.
 					glog.Errorf(
 						"operationExecutor.MountVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
@@ -243,8 +265,10 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 						attachedVolume.VolumeSpec.Name())
 					err := rc.operationExecutor.UnmountDevice(
 						attachedVolume.AttachedVolume, rc.actualStateOfWorld)
-					if err != nil && !goroutinemap.IsAlreadyExists(err) {
-						// Ignore goroutinemap.IsAlreadyExists errors, they are expected.
+					if err != nil &&
+						!goroutinemap.IsAlreadyExists(err) &&
+						!goroutinemap.IsExponentialBackoff(err) {
+						// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
 						// Log all other errors.
 						glog.Errorf(
 							"operationExecutor.UnmountDevice failed for volume %q (spec.Name: %q) controllerAttachDetachEnabled: %v with err: %v",
@@ -272,8 +296,10 @@ func (rc *reconciler) reconciliationLoopFunc() func() {
 							attachedVolume.VolumeSpec.Name())
 						err := rc.operationExecutor.DetachVolume(
 							attachedVolume.AttachedVolume, rc.actualStateOfWorld)
-						if err != nil && !goroutinemap.IsAlreadyExists(err) {
-							// Ignore goroutinemap.IsAlreadyExists errors, they are expected.
+						if err != nil &&
+							!goroutinemap.IsAlreadyExists(err) &&
+							!goroutinemap.IsExponentialBackoff(err) {
+							// Ignore goroutinemap.IsAlreadyExists && goroutinemap.IsExponentialBackoff errors, they are expected.
 							// Log all other errors.
 							glog.Errorf(
 								"operationExecutor.DetachVolume failed for volume %q (spec.Name: %q) controllerAttachDetachEnabled: %v with err: %v",

pkg/kubelet/volume/reconciler/reconciler_test.go

@@ -17,12 +17,16 @@ limitations under the License.
 package reconciler
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
+	"k8s.io/kubernetes/pkg/client/testing/core"
 	"k8s.io/kubernetes/pkg/kubelet/volume/cache"
+	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/util/wait"
 	"k8s.io/kubernetes/pkg/volume"
 	volumetesting "k8s.io/kubernetes/pkg/volume/testing"
@@ -38,18 +42,20 @@ const (
 	// waitForAttachTimeout is the maximum amount of time a
 	// operationexecutor.Mount call will wait for a volume to be attached.
 	waitForAttachTimeout time.Duration = 1 * time.Second
+	nodeName             string        = "myhostname"
 )
 
 // Calls Run()
 // Verifies there are no calls to attach, detach, mount, unmount, etc.
 func Test_Run_Positive_DoNothing(t *testing.T) {
 	// Arrange
-	nodeName := "myhostname"
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
-	oex := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	kubeClient := createTestClient()
+	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
 	reconciler := NewReconciler(
+		kubeClient,
 		false, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
 		waitForAttachTimeout,
@@ -75,12 +81,13 @@ func Test_Run_Positive_DoNothing(t *testing.T) {
 // Verifies there is are attach/mount/etc calls and no detach/unmount calls.
 func Test_Run_Positive_VolumeAttachAndMount(t *testing.T) {
 	// Arrange
-	nodeName := "myhostname"
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
-	oex := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	kubeClient := createTestClient()
+	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
 	reconciler := NewReconciler(
+		kubeClient,
 		false, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
 		waitForAttachTimeout,
@@ -141,12 +148,13 @@ func Test_Run_Positive_VolumeAttachAndMount(t *testing.T) {
 // Verifies there are no attach/detach calls.
 func Test_Run_Positive_VolumeMountControllerAttachEnabled(t *testing.T) {
 	// Arrange
-	nodeName := "myhostname"
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
-	oex := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	kubeClient := createTestClient()
+	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
 	reconciler := NewReconciler(
+		kubeClient,
 		true, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
 		waitForAttachTimeout,
@@ -206,12 +214,13 @@ func Test_Run_Positive_VolumeMountControllerAttachEnabled(t *testing.T) {
 // Verifies detach/unmount calls are issued.
 func Test_Run_Positive_VolumeAttachMountUnmountDetach(t *testing.T) {
 	// Arrange
-	nodeName := "myhostname"
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
-	oex := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	kubeClient := createTestClient()
+	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
 	reconciler := NewReconciler(
+		kubeClient,
 		false, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
 		waitForAttachTimeout,
@@ -284,12 +293,13 @@ func Test_Run_Positive_VolumeAttachMountUnmountDetach(t *testing.T) {
 // Verifies there are no attach/detach calls made.
 func Test_Run_Positive_VolumeUnmountControllerAttachEnabled(t *testing.T) {
 	// Arrange
-	nodeName := "myhostname"
 	volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
 	dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
-	oex := operationexecutor.NewOperationExecutor(volumePluginMgr)
+	kubeClient := createTestClient()
+	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
 	reconciler := NewReconciler(
+		kubeClient,
 		true, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
 		waitForAttachTimeout,
@@ -402,3 +412,25 @@ func retryWithExponentialBackOff(initialDuration time.Duration, fn wait.Conditio
 	}
 	return wait.ExponentialBackoff(backoff, fn)
 }
+
+func createTestClient() *fake.Clientset {
+	fakeClient := &fake.Clientset{}
+	fakeClient.AddReactor("get", "nodes",
+		func(action core.Action) (bool, runtime.Object, error) {
+			return true, &api.Node{
+				ObjectMeta: api.ObjectMeta{Name: nodeName},
+				Status: api.NodeStatus{
+					VolumesAttached: []api.AttachedVolume{
+						{
+							Name:       "fake-plugin/volume-name",
+							DevicePath: "fake/path",
+						},
+					}},
+				Spec: api.NodeSpec{ExternalID: nodeName},
+			}, nil
+		})
+	fakeClient.AddReactor("*", "*", func(action core.Action) (bool, runtime.Object, error) {
+		return true, nil, fmt.Errorf("no reaction implemented for %s", action)
+	})
+	return fakeClient
+}

pkg/kubelet/volume/volume_manager.go

@@ -126,10 +126,13 @@ func NewVolumeManager(
 		volumePluginMgr:     volumePluginMgr,
 		desiredStateOfWorld: cache.NewDesiredStateOfWorld(volumePluginMgr),
 		actualStateOfWorld:  cache.NewActualStateOfWorld(hostName, volumePluginMgr),
-		operationExecutor:   operationexecutor.NewOperationExecutor(volumePluginMgr),
+		operationExecutor: operationexecutor.NewOperationExecutor(
+			kubeClient,
+			volumePluginMgr),
 	}
 
 	vm.reconciler = reconciler.NewReconciler(
+		kubeClient,
 		controllerAttachDetachEnabled,
 		reconcilerLoopSleepPeriod,
 		waitForAttachTimeout,

pkg/util/goroutinemap/goroutinemap.go

@@ -23,9 +23,24 @@ package goroutinemap
 
 import (
 	"fmt"
+	"runtime"
 	"sync"
+	"time"
 
-	"k8s.io/kubernetes/pkg/util/runtime"
+	"github.com/golang/glog"
+	k8sRuntime "k8s.io/kubernetes/pkg/util/runtime"
+)
+
+const (
+	// initialDurationBeforeRetry is the amount of time after an error occurs
+	// that GoRoutineMap will refuse to allow another operation to start with
+	// the same operationName (if exponentialBackOffOnError is enabled). Each
+	// successive error results in a wait 2x times the previous.
+	initialDurationBeforeRetry time.Duration = 500 * time.Millisecond
+
+	// maxDurationBeforeRetry is the maximum amount of time that
+	// durationBeforeRetry will grow to due to exponential backoff.
+	maxDurationBeforeRetry time.Duration = 2 * time.Minute
 )
 
 // GoRoutineMap defines the supported set of operations.
@@ -36,7 +51,7 @@ type GoRoutineMap interface {
 	// go routine is terminated and the operationName is removed from the list
 	// of executing operations allowing a new operation to be started with the
 	// same name without error.
-	Run(operationName string, operation func() error) error
+	Run(operationName string, operationFunc func() error) error
 
 	// Wait blocks until all operations are completed. This is typically
 	// necessary during tests - the test should wait until all operations finish
@@ -45,50 +60,127 @@ type GoRoutineMap interface {
 }
 
 // NewGoRoutineMap returns a new instance of GoRoutineMap.
-func NewGoRoutineMap() GoRoutineMap {
+func NewGoRoutineMap(exponentialBackOffOnError bool) GoRoutineMap {
 	return &goRoutineMap{
-		operations: make(map[string]bool),
+		operations:                make(map[string]operation),
+		exponentialBackOffOnError: exponentialBackOffOnError,
 	}
 }
 
 type goRoutineMap struct {
-	operations map[string]bool
+	operations                map[string]operation
+	exponentialBackOffOnError bool
+	wg                        sync.WaitGroup
 	sync.Mutex
-	wg sync.WaitGroup
 }
 
-func (grm *goRoutineMap) Run(operationName string, operation func() error) error {
+type operation struct {
+	operationPending    bool
+	lastError           error
+	lastErrorTime       time.Time
+	durationBeforeRetry time.Duration
+}
+
+func (grm *goRoutineMap) Run(operationName string, operationFunc func() error) error {
 	grm.Lock()
 	defer grm.Unlock()
-	if grm.operations[operationName] {
+	existingOp, exists := grm.operations[operationName]
+	if exists {
 		// Operation with name exists
-		return newAlreadyExistsError(operationName)
+		if existingOp.operationPending {
+			return newAlreadyExistsError(operationName)
+		}
+
+		if time.Since(existingOp.lastErrorTime) <= existingOp.durationBeforeRetry {
+			return newExponentialBackoffError(operationName, existingOp)
+		}
 	}
 
-	grm.operations[operationName] = true
+	grm.operations[operationName] = operation{
+		operationPending:    true,
+		lastError:           existingOp.lastError,
+		lastErrorTime:       existingOp.lastErrorTime,
+		durationBeforeRetry: existingOp.durationBeforeRetry,
+	}
 	grm.wg.Add(1)
-	go func() {
-		defer grm.operationComplete(operationName)
-		defer runtime.HandleCrash()
-		operation()
+	go func() (err error) {
+		// Handle unhandled panics (very unlikely)
+		defer k8sRuntime.HandleCrash()
+		// Handle completion of and error, if any, from operationFunc()
+		defer grm.operationComplete(operationName, &err)
+		// Handle panic, if any, from operationFunc()
+		defer recoverFromPanic(operationName, &err)
+		return operationFunc()
 	}()
 
 	return nil
 }
 
-func (grm *goRoutineMap) operationComplete(operationName string) {
+func (grm *goRoutineMap) operationComplete(operationName string, err *error) {
 	defer grm.wg.Done()
 	grm.Lock()
 	defer grm.Unlock()
-	delete(grm.operations, operationName)
+
+	if *err == nil || !grm.exponentialBackOffOnError {
+		// Operation completed without error, or exponentialBackOffOnError disabled
+		delete(grm.operations, operationName)
+		if *err != nil {
+			// Log error
+			glog.Errorf("operation for %q failed with: %v",
+				operationName,
+				*err)
+		}
+	} else {
+		// Operation completed with error and exponentialBackOffOnError Enabled
+		existingOp := grm.operations[operationName]
+		if existingOp.durationBeforeRetry == 0 {
+			existingOp.durationBeforeRetry = initialDurationBeforeRetry
+		} else {
+			existingOp.durationBeforeRetry = 2 * existingOp.durationBeforeRetry
+			if existingOp.durationBeforeRetry > maxDurationBeforeRetry {
+				existingOp.durationBeforeRetry = maxDurationBeforeRetry
+			}
+		}
+		existingOp.lastError = *err
+		existingOp.lastErrorTime = time.Now()
+		existingOp.operationPending = false
+
+		grm.operations[operationName] = existingOp
+
+		// Log error
+		glog.Errorf("Operation for %q failed. No retries permitted until %v (durationBeforeRetry %v). error: %v",
+			operationName,
+			existingOp.lastErrorTime.Add(existingOp.durationBeforeRetry),
+			existingOp.durationBeforeRetry,
+			*err)
+	}
 }
 
 func (grm *goRoutineMap) Wait() {
 	grm.wg.Wait()
 }
 
-// alreadyExistsError is specific error returned when NewGoRoutine()
-// detects that operation with given name is already running.
+func recoverFromPanic(operationName string, err *error) {
+	if r := recover(); r != nil {
+		callers := ""
+		for i := 0; true; i++ {
+			_, file, line, ok := runtime.Caller(i)
+			if !ok {
+				break
+			}
+			callers = callers + fmt.Sprintf("%v:%v\n", file, line)
+		}
+		*err = fmt.Errorf(
+			"operation for %q recovered from panic %q. (err=%v) Call stack:\n%v",
+			operationName,
+			r,
+			*err,
+			callers)
+	}
+}
+
+// alreadyExistsError is the error returned when NewGoRoutine() detects that
+// an operation with the given name is already running.
 type alreadyExistsError struct {
 	operationName string
 }
@@ -96,7 +188,7 @@ type alreadyExistsError struct {
 var _ error = alreadyExistsError{}
 
 func (err alreadyExistsError) Error() string {
-	return fmt.Sprintf("Failed to create operation with name %q. An operation with that name already exists", err.operationName)
+	return fmt.Sprintf("Failed to create operation with name %q. An operation with that name is already executing.", err.operationName)
 }
 
 func newAlreadyExistsError(operationName string) error {
@@ -113,3 +205,43 @@ func IsAlreadyExists(err error) bool {
 		return false
 	}
 }
+
+// exponentialBackoffError is the error returned when NewGoRoutine() detects
+// that the previous operation for given name failed less then
+// durationBeforeRetry.
+type exponentialBackoffError struct {
+	operationName string
+	failedOp      operation
+}
+
+var _ error = exponentialBackoffError{}
+
+func (err exponentialBackoffError) Error() string {
+	return fmt.Sprintf(
+		"Failed to create operation with name %q. An operation with that name failed at %v. No retries permitted until %v (%v). Last error: %q.",
+		err.operationName,
+		err.failedOp.lastErrorTime,
+		err.failedOp.lastErrorTime.Add(err.failedOp.durationBeforeRetry),
+		err.failedOp.durationBeforeRetry,
+		err.failedOp.lastError)
+}
+
+func newExponentialBackoffError(
+	operationName string, failedOp operation) error {
+	return exponentialBackoffError{
+		operationName: operationName,
+		failedOp:      failedOp,
+	}
+}
+
+// IsExponentialBackoff returns true if an error returned from NewGoRoutine()
+// indicates that the previous operation for given name failed less then
+// durationBeforeRetry.
+func IsExponentialBackoff(err error) bool {
+	switch err.(type) {
+	case exponentialBackoffError:
+		return true
+	default:
+		return false
+	}
+}

pkg/volume/aws_ebs/attacher.go

@@ -65,12 +65,7 @@ func (attacher *awsElasticBlockStoreAttacher) Attach(spec *volume.Spec, hostName
 	return devicePath, nil
 }
 
-func (attacher *awsElasticBlockStoreAttacher) WaitForAttach(spec *volume.Spec, timeout time.Duration) (string, error) {
-	awsCloud, err := getCloudProvider(attacher.host.GetCloudProvider())
-	if err != nil {
-		return "", err
-	}
-
+func (attacher *awsElasticBlockStoreAttacher) WaitForAttach(spec *volume.Spec, devicePath string, timeout time.Duration) (string, error) {
 	volumeSource, _, err := getVolumeSource(spec)
 	if err != nil {
 		return "", err
@@ -82,11 +77,8 @@ func (attacher *awsElasticBlockStoreAttacher) WaitForAttach(spec *volume.Spec, t
 		partition = strconv.Itoa(int(volumeSource.Partition))
 	}
 
-	devicePath := ""
-	if d, err := awsCloud.GetDiskPath(volumeID); err == nil {
-		devicePath = d
-	} else {
-		glog.Errorf("GetDiskPath %q gets error %v", volumeID, err)
+	if devicePath == "" {
+		return "", fmt.Errorf("WaitForAttach failed for AWS Volume %q: devicePath is empty.", volumeID)
 	}
 
 	ticker := time.NewTicker(checkSleepDuration)
@@ -98,13 +90,6 @@ func (attacher *awsElasticBlockStoreAttacher) WaitForAttach(spec *volume.Spec, t
 		select {
 		case <-ticker.C:
 			glog.V(5).Infof("Checking AWS Volume %q is attached.", volumeID)
-			if devicePath == "" {
-				if d, err := awsCloud.GetDiskPath(volumeID); err == nil {
-					devicePath = d
-				} else {
-					glog.Errorf("GetDiskPath %q gets error %v", volumeID, err)
-				}
-			}
 			if devicePath != "" {
 				devicePaths := getDiskByIdPaths(partition, devicePath)
 				path, err := verifyDevicePath(devicePaths)

pkg/volume/cinder/attacher.go

@@ -71,7 +71,7 @@ func (attacher *cinderDiskAttacher) Attach(spec *volume.Spec, hostName string) (
 	attached, err := cloud.DiskIsAttached(volumeID, instanceid)
 	if err != nil {
 		// Log error and continue with attach
-		glog.Errorf(
+		glog.Warningf(
 			"Error checking if volume (%q) is already attached to current node (%q). Will continue and try attach anyway. err=%v",
 			volumeID, instanceid, err)
 	}
@@ -81,41 +81,33 @@ func (attacher *cinderDiskAttacher) Attach(spec *volume.Spec, hostName string) (
 		glog.Infof("Attach operation is successful. volume %q is already attached to node %q.", volumeID, instanceid)
 	} else {
 		_, err = cloud.AttachDisk(instanceid, volumeID)
-		if err != nil {
-			glog.Infof("attach volume %q to instance %q gets %v", volumeID, instanceid, err)
+		if err == nil {
+			glog.Infof("Attach operation successful: volume %q attached to node %q.", volumeID, instanceid)
+		} else {
+			glog.Infof("Attach volume %q to instance %q failed with %v", volumeID, instanceid, err)
+			return "", err
 		}
 	}
 
-	glog.Infof("attached volume %q to instance %q", volumeID, instanceid)
 	devicePath, err := cloud.GetAttachmentDiskPath(instanceid, volumeID)
 	if err != nil {
+		glog.Infof("Attach volume %q to instance %q failed with %v", volumeID, instanceid, err)
 		return "", err
 	}
 
 	return devicePath, err
 }
 
-func (attacher *cinderDiskAttacher) WaitForAttach(spec *volume.Spec, timeout time.Duration) (string, error) {
-	cloud, err := getCloudProvider(attacher.host.GetCloudProvider())
-	if err != nil {
-		return "", err
-	}
-
+func (attacher *cinderDiskAttacher) WaitForAttach(spec *volume.Spec, devicePath string, timeout time.Duration) (string, error) {
 	volumeSource, _, err := getVolumeSource(spec)
 	if err != nil {
 		return "", err
 	}
 
 	volumeID := volumeSource.VolumeID
-	instanceid, err := cloud.InstanceID()
-	if err != nil {
-		return "", err
-	}
-	devicePath := ""
-	if d, err := cloud.GetAttachmentDiskPath(instanceid, volumeID); err == nil {
-		devicePath = d
-	} else {
-		glog.Errorf("%q GetAttachmentDiskPath (%q) gets error %v", instanceid, volumeID, err)
+
+	if devicePath == "" {
+		return "", fmt.Errorf("WaitForAttach failed for Cinder disk %q: devicePath is empty.", volumeID)
 	}
 
 	ticker := time.NewTicker(checkSleepDuration)
@@ -128,25 +120,14 @@ func (attacher *cinderDiskAttacher) WaitForAttach(spec *volume.Spec, timeout tim
 		select {
 		case <-ticker.C:
 			glog.V(5).Infof("Checking Cinder disk %q is attached.", volumeID)
-			if devicePath == "" {
-				if d, err := cloud.GetAttachmentDiskPath(instanceid, volumeID); err == nil {
-					devicePath = d
-				} else {
-					glog.Errorf("%q GetAttachmentDiskPath (%q) gets error %v", instanceid, volumeID, err)
-				}
-			}
-			if devicePath == "" {
-				glog.V(5).Infof("Cinder disk (%q) is not attached yet", volumeID)
+			probeAttachedVolume()
+			exists, err := pathExists(devicePath)
+			if exists && err == nil {
+				glog.Infof("Successfully found attached Cinder disk %q.", volumeID)
+				return devicePath, nil
 			} else {
-				probeAttachedVolume()
-				exists, err := pathExists(devicePath)
-				if exists && err == nil {
-					glog.Infof("Successfully found attached Cinder disk %q.", volumeID)
-					return devicePath, nil
-				} else {
-					//Log error, if any, and continue checking periodically
-					glog.Errorf("Error Stat Cinder disk (%q) is attached: %v", volumeID, err)
-				}
+				//Log error, if any, and continue checking periodically
+				glog.Errorf("Error Stat Cinder disk (%q) is attached: %v", volumeID, err)
 			}
 		case <-timer.C:
 			return "", fmt.Errorf("Could not find attached Cinder disk %q. Timeout waiting for mount paths to be created.", volumeID)

pkg/volume/gce_pd/attacher.go

@@ -89,7 +89,7 @@ func (attacher *gcePersistentDiskAttacher) Attach(spec *volume.Spec, hostName st
 	return path.Join(diskByIdPath, diskGooglePrefix+pdName), nil
 }
 
-func (attacher *gcePersistentDiskAttacher) WaitForAttach(spec *volume.Spec, timeout time.Duration) (string, error) {
+func (attacher *gcePersistentDiskAttacher) WaitForAttach(spec *volume.Spec, devicePath string, timeout time.Duration) (string, error) {
 	ticker := time.NewTicker(checkSleepDuration)
 	defer ticker.Stop()
 	timer := time.NewTimer(timeout)

pkg/volume/testing/testing.go

@@ -371,7 +371,7 @@ func (fv *FakeVolume) GetAttachCallCount() int {
 	return fv.AttachCallCount
 }
 
-func (fv *FakeVolume) WaitForAttach(spec *Spec, spectimeout time.Duration) (string, error) {
+func (fv *FakeVolume) WaitForAttach(spec *Spec, devicePath string, spectimeout time.Duration) (string, error) {
 	fv.Lock()
 	defer fv.Unlock()
 	fv.WaitForAttachCallCount++

pkg/volume/volume.go

@@ -143,7 +143,7 @@ type Attacher interface {
 	// node. If it successfully attaches, the path to the device
 	// is returned. Otherwise, if the device does not attach after
 	// the given timeout period, an error will be returned.
-	WaitForAttach(spec *Spec, timeout time.Duration) (string, error)
+	WaitForAttach(spec *Spec, devicePath string, timeout time.Duration) (string, error)
 
 	// GetDeviceMountPath returns a path where the device should
 	// be mounted after it is attached. This is a global mount