Merge pull request #27567 from saad-ali/blockKubeletOnAttachController

api/swagger-spec/v1.json

@@ -16768,6 +16768,13 @@
        "$ref": "v1.UniqueVolumeName"
       },
       "description": "List of attachable volumes in use (mounted) by the node."
+     },
+     "volumesAttached": {
+      "type": "array",
+      "items": {
+       "$ref": "v1.AttachedVolume"
+      },
+      "description": "List of volumes that are attached to the node."
      }
     }
    },
@@ -16930,6 +16937,24 @@
     "id": "v1.UniqueVolumeName",
     "properties": {}
    },
+   "v1.AttachedVolume": {
+    "id": "v1.AttachedVolume",
+    "description": "AttachedVolume describes a volume attached to a node",
+    "required": [
+     "name",
+     "devicePath"
+    ],
+    "properties": {
+     "name": {
+      "type": "string",
+      "description": "Name of the attached volume"
+     },
+     "devicePath": {
+      "type": "string",
+      "description": "DevicePath represents the device path where the volume should be avilable"
+     }
+    }
+   },
    "v1.PersistentVolumeClaimList": {
     "id": "v1.PersistentVolumeClaimList",
     "description": "PersistentVolumeClaimList is a list of PersistentVolumeClaim items.",

pkg/api/deep_copy_generated.go

@@ -35,6 +35,7 @@ func init() {
 	if err := Scheme.AddGeneratedDeepCopyFuncs(
 		DeepCopy_api_AWSElasticBlockStoreVolumeSource,
 		DeepCopy_api_Affinity,
+		DeepCopy_api_AttachedVolume,
 		DeepCopy_api_AzureFileVolumeSource,
 		DeepCopy_api_Binding,
 		DeepCopy_api_Capabilities,
@@ -228,6 +229,12 @@ func DeepCopy_api_Affinity(in Affinity, out *Affinity, c *conversion.Cloner) err
 	return nil
 }
 
+func DeepCopy_api_AttachedVolume(in AttachedVolume, out *AttachedVolume, c *conversion.Cloner) error {
+	out.Name = in.Name
+	out.DevicePath = in.DevicePath
+	return nil
+}
+
 func DeepCopy_api_AzureFileVolumeSource(in AzureFileVolumeSource, out *AzureFileVolumeSource, c *conversion.Cloner) error {
 	out.SecretName = in.SecretName
 	out.ShareName = in.ShareName
@@ -1610,6 +1617,17 @@ func DeepCopy_api_NodeStatus(in NodeStatus, out *NodeStatus, c *conversion.Clone
 	} else {
 		out.VolumesInUse = nil
 	}
+	if in.VolumesAttached != nil {
+		in, out := in.VolumesAttached, &out.VolumesAttached
+		*out = make([]AttachedVolume, len(in))
+		for i := range in {
+			if err := DeepCopy_api_AttachedVolume(in[i], &(*out)[i], c); err != nil {
+				return err
+			}
+		}
+	} else {
+		out.VolumesAttached = nil
+	}
 	return nil
 }
 

pkg/api/types.go

@@ -1989,10 +1989,21 @@ type NodeStatus struct {
 	Images []ContainerImage `json:"images,omitempty"`
 	// List of attachable volumes in use (mounted) by the node.
 	VolumesInUse []UniqueVolumeName `json:"volumesInUse,omitempty"`
+	// List of volumes that are attached to the node.
+	VolumesAttached []AttachedVolume `json:"volumesAttached,omitempty"`
 }
 
 type UniqueVolumeName string
 
+// AttachedVolume describes a volume attached to a node
+type AttachedVolume struct {
+	// Name of the attached volume
+	Name UniqueVolumeName `json:"name"`
+
+	// DevicePath represents the device path where the volume should be avilable
+	DevicePath string `json:"devicePath"`
+}
+
 // Describe a container image
 type ContainerImage struct {
 	// Names by which this image is known.

pkg/api/v1/conversion_generated.go

@@ -33,6 +33,8 @@ func init() {
 		Convert_api_AWSElasticBlockStoreVolumeSource_To_v1_AWSElasticBlockStoreVolumeSource,
 		Convert_v1_Affinity_To_api_Affinity,
 		Convert_api_Affinity_To_v1_Affinity,
+		Convert_v1_AttachedVolume_To_api_AttachedVolume,
+		Convert_api_AttachedVolume_To_v1_AttachedVolume,
 		Convert_v1_AzureFileVolumeSource_To_api_AzureFileVolumeSource,
 		Convert_api_AzureFileVolumeSource_To_v1_AzureFileVolumeSource,
 		Convert_v1_Binding_To_api_Binding,
@@ -425,6 +427,26 @@ func Convert_api_Affinity_To_v1_Affinity(in *api.Affinity, out *Affinity, s conv
 	return autoConvert_api_Affinity_To_v1_Affinity(in, out, s)
 }
 
+func autoConvert_v1_AttachedVolume_To_api_AttachedVolume(in *AttachedVolume, out *api.AttachedVolume, s conversion.Scope) error {
+	out.Name = api.UniqueVolumeName(in.Name)
+	out.DevicePath = in.DevicePath
+	return nil
+}
+
+func Convert_v1_AttachedVolume_To_api_AttachedVolume(in *AttachedVolume, out *api.AttachedVolume, s conversion.Scope) error {
+	return autoConvert_v1_AttachedVolume_To_api_AttachedVolume(in, out, s)
+}
+
+func autoConvert_api_AttachedVolume_To_v1_AttachedVolume(in *api.AttachedVolume, out *AttachedVolume, s conversion.Scope) error {
+	out.Name = UniqueVolumeName(in.Name)
+	out.DevicePath = in.DevicePath
+	return nil
+}
+
+func Convert_api_AttachedVolume_To_v1_AttachedVolume(in *api.AttachedVolume, out *AttachedVolume, s conversion.Scope) error {
+	return autoConvert_api_AttachedVolume_To_v1_AttachedVolume(in, out, s)
+}
+
 func autoConvert_v1_AzureFileVolumeSource_To_api_AzureFileVolumeSource(in *AzureFileVolumeSource, out *api.AzureFileVolumeSource, s conversion.Scope) error {
 	out.SecretName = in.SecretName
 	out.ShareName = in.ShareName
@@ -3397,6 +3419,17 @@ func autoConvert_v1_NodeStatus_To_api_NodeStatus(in *NodeStatus, out *api.NodeSt
 	} else {
 		out.VolumesInUse = nil
 	}
+	if in.VolumesAttached != nil {
+		in, out := &in.VolumesAttached, &out.VolumesAttached
+		*out = make([]api.AttachedVolume, len(*in))
+		for i := range *in {
+			if err := Convert_v1_AttachedVolume_To_api_AttachedVolume(&(*in)[i], &(*out)[i], s); err != nil {
+				return err
+			}
+		}
+	} else {
+		out.VolumesAttached = nil
+	}
 	return nil
 }
 
@@ -3480,6 +3513,17 @@ func autoConvert_api_NodeStatus_To_v1_NodeStatus(in *api.NodeStatus, out *NodeSt
 	} else {
 		out.VolumesInUse = nil
 	}
+	if in.VolumesAttached != nil {
+		in, out := &in.VolumesAttached, &out.VolumesAttached
+		*out = make([]AttachedVolume, len(*in))
+		for i := range *in {
+			if err := Convert_api_AttachedVolume_To_v1_AttachedVolume(&(*in)[i], &(*out)[i], s); err != nil {
+				return err
+			}
+		}
+	} else {
+		out.VolumesAttached = nil
+	}
 	return nil
 }
 

pkg/api/v1/deep_copy_generated.go

@@ -34,6 +34,7 @@ func init() {
 	if err := api.Scheme.AddGeneratedDeepCopyFuncs(
 		DeepCopy_v1_AWSElasticBlockStoreVolumeSource,
 		DeepCopy_v1_Affinity,
+		DeepCopy_v1_AttachedVolume,
 		DeepCopy_v1_AzureFileVolumeSource,
 		DeepCopy_v1_Binding,
 		DeepCopy_v1_Capabilities,
@@ -225,6 +226,12 @@ func DeepCopy_v1_Affinity(in Affinity, out *Affinity, c *conversion.Cloner) erro
 	return nil
 }
 
+func DeepCopy_v1_AttachedVolume(in AttachedVolume, out *AttachedVolume, c *conversion.Cloner) error {
+	out.Name = in.Name
+	out.DevicePath = in.DevicePath
+	return nil
+}
+
 func DeepCopy_v1_AzureFileVolumeSource(in AzureFileVolumeSource, out *AzureFileVolumeSource, c *conversion.Cloner) error {
 	out.SecretName = in.SecretName
 	out.ShareName = in.ShareName
@@ -1557,6 +1564,17 @@ func DeepCopy_v1_NodeStatus(in NodeStatus, out *NodeStatus, c *conversion.Cloner
 	} else {
 		out.VolumesInUse = nil
 	}
+	if in.VolumesAttached != nil {
+		in, out := in.VolumesAttached, &out.VolumesAttached
+		*out = make([]AttachedVolume, len(in))
+		for i := range in {
+			if err := DeepCopy_v1_AttachedVolume(in[i], &(*out)[i], c); err != nil {
+				return err
+			}
+		}
+	} else {
+		out.VolumesAttached = nil
+	}
 	return nil
 }
 

pkg/api/v1/generated.pb.go

@@ -27,6 +27,7 @@ limitations under the License.
 	It has these top-level messages:
 		AWSElasticBlockStoreVolumeSource
 		Affinity
+		AttachedVolume
 		AzureFileVolumeSource
 		Binding
 		Capabilities
@@ -201,6 +202,10 @@ func (m *Affinity) Reset()         { *m = Affinity{} }
 func (m *Affinity) String() string { return proto.CompactTextString(m) }
 func (*Affinity) ProtoMessage()    {}
 
+func (m *AttachedVolume) Reset()         { *m = AttachedVolume{} }
+func (m *AttachedVolume) String() string { return proto.CompactTextString(m) }
+func (*AttachedVolume) ProtoMessage()    {}
+
 func (m *AzureFileVolumeSource) Reset()         { *m = AzureFileVolumeSource{} }
 func (m *AzureFileVolumeSource) String() string { return proto.CompactTextString(m) }
 func (*AzureFileVolumeSource) ProtoMessage()    {}
@@ -788,6 +793,7 @@ func (*WeightedPodAffinityTerm) ProtoMessage()    {}
 func init() {
 	proto.RegisterType((*AWSElasticBlockStoreVolumeSource)(nil), "k8s.io.kubernetes.pkg.api.v1.AWSElasticBlockStoreVolumeSource")
 	proto.RegisterType((*Affinity)(nil), "k8s.io.kubernetes.pkg.api.v1.Affinity")
+	proto.RegisterType((*AttachedVolume)(nil), "k8s.io.kubernetes.pkg.api.v1.AttachedVolume")
 	proto.RegisterType((*AzureFileVolumeSource)(nil), "k8s.io.kubernetes.pkg.api.v1.AzureFileVolumeSource")
 	proto.RegisterType((*Binding)(nil), "k8s.io.kubernetes.pkg.api.v1.Binding")
 	proto.RegisterType((*Capabilities)(nil), "k8s.io.kubernetes.pkg.api.v1.Capabilities")
@@ -1020,6 +1026,32 @@ func (m *Affinity) MarshalTo(data []byte) (int, error) {
 	return i, nil
 }
 
+func (m *AttachedVolume) Marshal() (data []byte, err error) {
+	size := m.Size()
+	data = make([]byte, size)
+	n, err := m.MarshalTo(data)
+	if err != nil {
+		return nil, err
+	}
+	return data[:n], nil
+}
+
+func (m *AttachedVolume) MarshalTo(data []byte) (int, error) {
+	var i int
+	_ = i
+	var l int
+	_ = l
+	data[i] = 0xa
+	i++
+	i = encodeVarintGenerated(data, i, uint64(len(m.Name)))
+	i += copy(data[i:], m.Name)
+	data[i] = 0x12
+	i++
+	i = encodeVarintGenerated(data, i, uint64(len(m.DevicePath)))
+	i += copy(data[i:], m.DevicePath)
+	return i, nil
+}
+
 func (m *AzureFileVolumeSource) Marshal() (data []byte, err error) {
 	size := m.Size()
 	data = make([]byte, size)
@@ -4174,6 +4206,18 @@ func (m *NodeStatus) MarshalTo(data []byte) (int, error) {
 			i += copy(data[i:], s)
 		}
 	}
+	if len(m.VolumesAttached) > 0 {
+		for _, msg := range m.VolumesAttached {
+			data[i] = 0x52
+			i++
+			i = encodeVarintGenerated(data, i, uint64(msg.Size()))
+			n, err := msg.MarshalTo(data[i:])
+			if err != nil {
+				return 0, err
+			}
+			i += n
+		}
+	}
 	return i, nil
 }
 
@@ -7735,6 +7779,16 @@ func (m *Affinity) Size() (n int) {
 	return n
 }
 
+func (m *AttachedVolume) Size() (n int) {
+	var l int
+	_ = l
+	l = len(m.Name)
+	n += 1 + l + sovGenerated(uint64(l))
+	l = len(m.DevicePath)
+	n += 1 + l + sovGenerated(uint64(l))
+	return n
+}
+
 func (m *AzureFileVolumeSource) Size() (n int) {
 	var l int
 	_ = l
@@ -8887,6 +8941,12 @@ func (m *NodeStatus) Size() (n int) {
 			n += 1 + l + sovGenerated(uint64(l))
 		}
 	}
+	if len(m.VolumesAttached) > 0 {
+		for _, e := range m.VolumesAttached {
+			l = e.Size()
+			n += 1 + l + sovGenerated(uint64(l))
+		}
+	}
 	return n
 }
 
@@ -10492,6 +10552,114 @@ func (m *Affinity) Unmarshal(data []byte) error {
 	}
 	return nil
 }
+func (m *AttachedVolume) Unmarshal(data []byte) error {
+	l := len(data)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowGenerated
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := data[iNdEx]
+			iNdEx++
+			wire |= (uint64(b) & 0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: AttachedVolume: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: AttachedVolume: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Name", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := data[iNdEx]
+				iNdEx++
+				stringLen |= (uint64(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Name = UniqueVolumeName(data[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field DevicePath", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := data[iNdEx]
+				iNdEx++
+				stringLen |= (uint64(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.DevicePath = string(data[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipGenerated(data[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *AzureFileVolumeSource) Unmarshal(data []byte) error {
 	l := len(data)
 	iNdEx := 0
@@ -21635,6 +21803,37 @@ func (m *NodeStatus) Unmarshal(data []byte) error {
 			}
 			m.VolumesInUse = append(m.VolumesInUse, UniqueVolumeName(data[iNdEx:postIndex]))
 			iNdEx = postIndex
+		case 10:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field VolumesAttached", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := data[iNdEx]
+				iNdEx++
+				msglen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + msglen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.VolumesAttached = append(m.VolumesAttached, AttachedVolume{})
+			if err := m.VolumesAttached[len(m.VolumesAttached)-1].Unmarshal(data[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipGenerated(data[iNdEx:])

pkg/api/v1/generated.proto

@@ -71,6 +71,15 @@ message Affinity {
   optional PodAntiAffinity podAntiAffinity = 3;
 }
 
+// AttachedVolume describes a volume attached to a node
+message AttachedVolume {
+  // Name of the attached volume
+  optional string name = 1;
+
+  // DevicePath represents the device path where the volume should be avilable
+  optional string devicePath = 2;
+}
+
 // AzureFile represents an Azure File Service mount on the host and bind mount to the pod.
 message AzureFileVolumeSource {
   // the name of secret that contains Azure Storage Account Name and Key
@@ -1306,6 +1315,9 @@ message NodeStatus {
 
   // List of attachable volumes in use (mounted) by the node.
   repeated string volumesInUse = 9;
+
+  // List of volumes that are attached to the node.
+  repeated AttachedVolume volumesAttached = 10;
 }
 
 // NodeSystemInfo is a set of ids/uuids to uniquely identify the node.

pkg/api/v1/types.go

@@ -2388,10 +2388,21 @@ type NodeStatus struct {
 	Images []ContainerImage `json:"images,omitempty" protobuf:"bytes,8,rep,name=images"`
 	// List of attachable volumes in use (mounted) by the node.
 	VolumesInUse []UniqueVolumeName `json:"volumesInUse,omitempty" protobuf:"bytes,9,rep,name=volumesInUse"`
+	// List of volumes that are attached to the node.
+	VolumesAttached []AttachedVolume `json:"volumesAttached,omitempty" protobuf:"bytes,10,rep,name=volumesAttached"`
 }
 
 type UniqueVolumeName string
 
+// AttachedVolume describes a volume attached to a node
+type AttachedVolume struct {
+	// Name of the attached volume
+	Name UniqueVolumeName `json:"name" protobuf:"bytes,1,rep,name=name"`
+
+	// DevicePath represents the device path where the volume should be avilable
+	DevicePath string `json:"devicePath" protobuf:"bytes,2,rep,name=devicePath"`
+}
+
 // Describe a container image
 type ContainerImage struct {
 	// Names by which this image is known.

pkg/api/v1/types_swagger_doc_generated.go

@@ -50,6 +50,16 @@ func (Affinity) SwaggerDoc() map[string]string {
 	return map_Affinity
 }
 
+var map_AttachedVolume = map[string]string{
+	"":           "AttachedVolume describes a volume attached to a node",
+	"name":       "Name of the attached volume",
+	"devicePath": "DevicePath represents the device path where the volume should be avilable",
+}
+
+func (AttachedVolume) SwaggerDoc() map[string]string {
+	return map_AttachedVolume
+}
+
 var map_AzureFileVolumeSource = map[string]string{
 	"":           "AzureFile represents an Azure File Service mount on the host and bind mount to the pod.",
 	"secretName": "the name of secret that contains Azure Storage Account Name and Key",
@@ -881,6 +891,7 @@ var map_NodeStatus = map[string]string{
 	"nodeInfo":        "Set of ids/uuids to uniquely identify the node. More info: http://releases.k8s.io/HEAD/docs/admin/node.md#node-info",
 	"images":          "List of container images on this node",
 	"volumesInUse":    "List of attachable volumes in use (mounted) by the node.",
+	"volumesAttached": "List of volumes that are attached to the node.",
 }
 
 func (NodeStatus) SwaggerDoc() map[string]string {

pkg/client/clientset_generated/internalclientset/typed/core/unversioned/fake/fake_node_expansion.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 fake
+
+import (
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/testing/core"
+)
+
+func (c *FakeNodes) PatchStatus(nodeName string, data []byte) (*api.Node, error) {
+	obj, err := c.Fake.Invokes(
+		core.NewPatchSubresourceAction(nodesResource, "status"), &api.Node{})
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*api.Node), err
+}

pkg/client/clientset_generated/internalclientset/typed/core/unversioned/generated_expansion.go

@@ -22,8 +22,6 @@ type EndpointsExpansion interface{}
 
 type LimitRangeExpansion interface{}
 
-type NodeExpansion interface{}
-
 type PersistentVolumeExpansion interface{}
 
 type PersistentVolumeClaimExpansion interface{}

pkg/client/clientset_generated/internalclientset/typed/core/unversioned/node_expansion.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 unversioned
+
+import "k8s.io/kubernetes/pkg/api"
+
+// The NodeExpansion interface allows manually adding extra methods to the NodeInterface.
+type NodeExpansion interface {
+	// PatchStatus modifies the status of an existing node. It returns the copy
+	// of the node that the server returns, or an error.
+	PatchStatus(nodeName string, data []byte) (*api.Node, error)
+}
+
+// PatchStatus modifies the status of an existing node. It returns the copy of
+// the node that the server returns, or an error.
+func (c *nodes) PatchStatus(nodeName string, data []byte) (*api.Node, error) {
+	result := &api.Node{}
+	err := c.client.Patch(api.StrategicMergePatchType).
+		Resource("nodes").
+		Name(nodeName).
+		SubResource("status").
+		Body(data).
+		Do().
+		Into(result)
+	return result, err
+}

pkg/client/testing/core/actions.go

@@ -137,6 +137,15 @@ func NewPatchAction(resource unversioned.GroupVersionResource, namespace string,
 	return action
 }
 
+func NewPatchSubresourceAction(resource unversioned.GroupVersionResource, subresource string) PatchActionImpl {
+	action := PatchActionImpl{}
+	action.Verb = "patch"
+	action.Resource = resource
+	action.Subresource = subresource
+
+	return action
+}
+
 func NewRootUpdateSubresourceAction(resource unversioned.GroupVersionResource, subresource string, object runtime.Object) UpdateActionImpl {
 	action := UpdateActionImpl{}
 	action.Verb = "update"

pkg/controller/node/nodecontroller_test.go

@@ -157,6 +157,11 @@ func (m *FakeNodeHandler) UpdateStatus(node *api.Node) (*api.Node, error) {
 	return node, nil
 }
 
+func (m *FakeNodeHandler) PatchStatus(nodeName string, data []byte) (*api.Node, error) {
+	m.RequestCount++
+	return &api.Node{}, nil
+}
+
 func (m *FakeNodeHandler) Watch(opts api.ListOptions) (watch.Interface, error) {
 	return nil, nil
 }

pkg/controller/persistentvolume/controller_base.go

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

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

@@ -41,46 +41,31 @@ func (plugin *awsElasticBlockStorePlugin) NewAttacher() (volume.Attacher, error)
 	return &awsElasticBlockStoreAttacher{host: plugin.host}, nil
 }
 
-func (attacher *awsElasticBlockStoreAttacher) Attach(spec *volume.Spec, hostName string) error {
+func (attacher *awsElasticBlockStoreAttacher) Attach(spec *volume.Spec, hostName string) (string, error) {
 	volumeSource, readOnly, err := getVolumeSource(spec)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	volumeID := volumeSource.VolumeID
 
 	awsCloud, err := getCloudProvider(attacher.host.GetCloudProvider())
 	if err != nil {
-		return err
+		return "", err
 	}
 
-	attached, err := awsCloud.DiskIsAttached(volumeID, hostName)
+	// awsCloud.AttachDisk checks if disk is already attached to node and
+	// succeeds in that case, so no need to do that separately.
+	devicePath, err := awsCloud.AttachDisk(volumeID, hostName, readOnly)
 	if err != nil {
-		// Log error and continue with attach
-		glog.Errorf(
-			"Error checking if volume (%q) is already attached to current node (%q). Will continue and try attach anyway. err=%v",
-			volumeID, hostName, err)
-	}
-
-	if err == nil && attached {
-		// Volume is already attached to node.
-		glog.Infof("Attach operation is successful. volume %q is already attached to node %q.", volumeID, hostName)
-		return nil
-	}
-
-	if _, err = awsCloud.AttachDisk(volumeID, hostName, readOnly); err != nil {
 		glog.Errorf("Error attaching volume %q: %+v", volumeID, err)
-		return err
-	}
-	return nil
-}
-
-func (attacher *awsElasticBlockStoreAttacher) WaitForAttach(spec *volume.Spec, timeout time.Duration) (string, error) {
-	awsCloud, err := getCloudProvider(attacher.host.GetCloudProvider())
-	if err != nil {
 		return "", err
 	}
 
+	return devicePath, nil
+}
+
+func (attacher *awsElasticBlockStoreAttacher) WaitForAttach(spec *volume.Spec, devicePath string, timeout time.Duration) (string, error) {
 	volumeSource, _, err := getVolumeSource(spec)
 	if err != nil {
 		return "", err
@@ -92,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)
@@ -108,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

@@ -45,25 +45,25 @@ func (plugin *cinderPlugin) NewAttacher() (volume.Attacher, error) {
 	return &cinderDiskAttacher{host: plugin.host}, nil
 }
 
-func (attacher *cinderDiskAttacher) Attach(spec *volume.Spec, hostName string) error {
+func (attacher *cinderDiskAttacher) Attach(spec *volume.Spec, hostName string) (string, error) {
 	volumeSource, _, err := getVolumeSource(spec)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	volumeID := volumeSource.VolumeID
 
 	cloud, err := getCloudProvider(attacher.host.GetCloudProvider())
 	if err != nil {
-		return err
+		return "", err
 	}
 	instances, res := cloud.Instances()
 	if !res {
-		return fmt.Errorf("failed to list openstack instances")
+		return "", fmt.Errorf("failed to list openstack instances")
 	}
 	instanceid, err := instances.InstanceID(hostName)
 	if err != nil {
-		return err
+		return "", err
 	}
 	if ind := strings.LastIndex(instanceid, "/"); ind >= 0 {
 		instanceid = instanceid[(ind + 1):]
@@ -71,7 +71,7 @@ func (attacher *cinderDiskAttacher) Attach(spec *volume.Spec, hostName string) e
 	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)
 	}
@@ -79,38 +79,35 @@ func (attacher *cinderDiskAttacher) Attach(spec *volume.Spec, hostName string) e
 	if err == nil && attached {
 		// Volume is already attached to node.
 		glog.Infof("Attach operation is successful. volume %q is already attached to node %q.", volumeID, instanceid)
-		return nil
-	}
-
-	_, err = cloud.AttachDisk(instanceid, volumeID)
-	if err != nil {
-		glog.Infof("attach volume %q to instance %q gets %v", volumeID, instanceid, err)
+	} else {
+		_, err = cloud.AttachDisk(instanceid, volumeID)
+		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)
-	return err
-}
 
-func (attacher *cinderDiskAttacher) WaitForAttach(spec *volume.Spec, timeout time.Duration) (string, error) {
-	cloud, err := getCloudProvider(attacher.host.GetCloudProvider())
+	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, 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)
@@ -123,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

@@ -60,10 +60,10 @@ func (plugin *gcePersistentDiskPlugin) NewAttacher() (volume.Attacher, error) {
 // Callers are responsible for retryinging on failure.
 // Callers are responsible for thread safety between concurrent attach and
 // detach operations.
-func (attacher *gcePersistentDiskAttacher) Attach(spec *volume.Spec, hostName string) error {
+func (attacher *gcePersistentDiskAttacher) Attach(spec *volume.Spec, hostName string) (string, error) {
 	volumeSource, readOnly, err := getVolumeSource(spec)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	pdName := volumeSource.PDName
@@ -79,18 +79,17 @@ func (attacher *gcePersistentDiskAttacher) Attach(spec *volume.Spec, hostName st
 	if err == nil && attached {
 		// Volume is already attached to node.
 		glog.Infof("Attach operation is successful. PD %q is already attached to node %q.", pdName, hostName)
-		return nil
-	}
-
-	if err = attacher.gceDisks.AttachDisk(pdName, hostName, readOnly); err != nil {
-		glog.Errorf("Error attaching PD %q to node %q: %+v", pdName, hostName, err)
-		return err
+	} else {
+		if err := attacher.gceDisks.AttachDisk(pdName, hostName, readOnly); err != nil {
+			glog.Errorf("Error attaching PD %q to node %q: %+v", pdName, hostName, err)
+			return "", err
+		}
 	}
 
-	return nil
+	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/gce_pd/attacher_test.go

@@ -18,6 +18,7 @@ package gce_pd
 
 import (
 	"errors"
+	"fmt"
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
@@ -86,7 +87,11 @@ func TestAttachDetach(t *testing.T) {
 			attach:         attachCall{diskName, instanceID, readOnly, nil},
 			test: func(testcase *testcase) error {
 				attacher := newAttacher(testcase)
-				return attacher.Attach(spec, instanceID)
+				devicePath, err := attacher.Attach(spec, instanceID)
+				if devicePath != "/dev/disk/by-id/google-disk" {
+					return fmt.Errorf("devicePath incorrect. Expected<\"/dev/disk/by-id/google-disk\"> Actual: <%q>", devicePath)
+				}
+				return err
 			},
 		},
 
@@ -96,7 +101,11 @@ func TestAttachDetach(t *testing.T) {
 			diskIsAttached: diskIsAttachedCall{diskName, instanceID, true, nil},
 			test: func(testcase *testcase) error {
 				attacher := newAttacher(testcase)
-				return attacher.Attach(spec, instanceID)
+				devicePath, err := attacher.Attach(spec, instanceID)
+				if devicePath != "/dev/disk/by-id/google-disk" {
+					return fmt.Errorf("devicePath incorrect. Expected<\"/dev/disk/by-id/google-disk\"> Actual: <%q>", devicePath)
+				}
+				return err
 			},
 		},
 
@@ -107,7 +116,11 @@ func TestAttachDetach(t *testing.T) {
 			attach:         attachCall{diskName, instanceID, readOnly, nil},
 			test: func(testcase *testcase) error {
 				attacher := newAttacher(testcase)
-				return attacher.Attach(spec, instanceID)
+				devicePath, err := attacher.Attach(spec, instanceID)
+				if devicePath != "/dev/disk/by-id/google-disk" {
+					return fmt.Errorf("devicePath incorrect. Expected<\"/dev/disk/by-id/google-disk\"> Actual: <%q>", devicePath)
+				}
+				return err
 			},
 		},
 
@@ -118,7 +131,11 @@ func TestAttachDetach(t *testing.T) {
 			attach:         attachCall{diskName, instanceID, readOnly, attachError},
 			test: func(testcase *testcase) error {
 				attacher := newAttacher(testcase)
-				return attacher.Attach(spec, instanceID)
+				devicePath, err := attacher.Attach(spec, instanceID)
+				if devicePath != "" {
+					return fmt.Errorf("devicePath incorrect. Expected<\"\"> Actual: <%q>", devicePath)
+				}
+				return err
 			},
 			expectedReturn: attachError,
 		},

pkg/volume/testing/testing.go

@@ -358,11 +358,11 @@ func (fv *FakeVolume) TearDownAt(dir string) error {
 	return os.RemoveAll(dir)
 }
 
-func (fv *FakeVolume) Attach(spec *Spec, hostName string) error {
+func (fv *FakeVolume) Attach(spec *Spec, hostName string) (string, error) {
 	fv.Lock()
 	defer fv.Unlock()
 	fv.AttachCallCount++
-	return nil
+	return "", nil
 }
 
 func (fv *FakeVolume) GetAttachCallCount() int {
@@ -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

@@ -134,14 +134,16 @@ type Deleter interface {
 
 // Attacher can attach a volume to a node.
 type Attacher interface {
-	// Attach the volume specified by the given spec to the given host
-	Attach(spec *Spec, hostName string) error
+	// Attaches the volume specified by the given spec to the given host.
+	// On success, returns the device path where the device was attache don the
+	// node.
+	Attach(spec *Spec, hostName string) (string, error)
 
 	// WaitForAttach blocks until the device is attached to this
 	// 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