Merge pull request #71276 from jingxu97/Oct/uncertain

pkg/controller/volume/attachdetach/attach_detach_controller.go

@@ -435,7 +435,7 @@ func (adc *attachDetachController) populateDesiredStateOfWorld() error {
 					err)
 				continue
 			}
-			if adc.actualStateOfWorld.VolumeNodeExists(volumeName, nodeName) {
+			if adc.actualStateOfWorld.IsVolumeAttachedToNode(volumeName, nodeName) {
 				devicePath, err := adc.getNodeVolumeDevicePath(volumeName, nodeName)
 				if err != nil {
 					klog.Errorf("Failed to find device path: %v", err)

pkg/controller/volume/attachdetach/attach_detach_controller_test.go

@@ -110,7 +110,7 @@ func Test_AttachDetachControllerStateOfWolrdPopulators_Positive(t *testing.T) {
 	for _, node := range nodes {
 		nodeName := types.NodeName(node.Name)
 		for _, attachedVolume := range node.Status.VolumesAttached {
-			found := adc.actualStateOfWorld.VolumeNodeExists(attachedVolume.Name, nodeName)
+			found := adc.actualStateOfWorld.IsVolumeAttachedToNode(attachedVolume.Name, nodeName)
 			if !found {
 				t.Fatalf("Run failed with error. Node %s, volume %s not found", nodeName, attachedVolume.Name)
 			}

pkg/controller/volume/attachdetach/cache/BUILD

@@ -34,6 +34,7 @@ go_test(
     deps = [
         "//pkg/controller/volume/attachdetach/testing:go_default_library",
         "//pkg/volume/testing:go_default_library",
+        "//pkg/volume/util:go_default_library",
         "//pkg/volume/util/types:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",

pkg/controller/volume/attachdetach/metrics/metrics_test.go

@@ -143,7 +143,7 @@ func TestTotalVolumesMetricCollection(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
-	asw.AddVolumeNode(volumeName, volumeSpec, nodeName, "")
+	asw.AddVolumeNode(volumeName, volumeSpec, nodeName, "", true)
 
 	metricCollector := newAttachDetachStateCollector(
 		nil,

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

@@ -178,7 +178,6 @@ func (rc *reconciler) reconcile() {
 	for _, attachedVolume := range rc.actualStateOfWorld.GetAttachedVolumes() {
 		if !rc.desiredStateOfWorld.VolumeExists(
 			attachedVolume.VolumeName, attachedVolume.NodeName) {
-
 			// Don't even try to start an operation if there is already one running
 			// This check must be done before we do any other checks, as otherwise the other checks
 			// may pass while at the same time the volume leaves the pending state, resulting in
@@ -198,7 +197,7 @@ func (rc *reconciler) reconcile() {
 			timeout := elapsedTime > rc.maxWaitForUnmountDuration
 			// Check whether volume is still mounted. Skip detach if it is still mounted unless timeout
 			if attachedVolume.MountedByNode && !timeout {
-				klog.V(12).Infof(attachedVolume.GenerateMsgDetailed("Cannot detach volume because it is still mounted", ""))
+				klog.V(5).Infof(attachedVolume.GenerateMsgDetailed("Cannot detach volume because it is still mounted", ""))
 				continue
 			}
 
@@ -253,7 +252,7 @@ func (rc *reconciler) reconcile() {
 func (rc *reconciler) attachDesiredVolumes() {
 	// Ensure volumes that should be attached are attached.
 	for _, volumeToAttach := range rc.desiredStateOfWorld.GetVolumesToAttach() {
-		if rc.actualStateOfWorld.VolumeNodeExists(volumeToAttach.VolumeName, volumeToAttach.NodeName) {
+		if rc.actualStateOfWorld.IsVolumeAttachedToNode(volumeToAttach.VolumeName, volumeToAttach.NodeName) {
 			// Volume/Node exists, touch it to reset detachRequestedTime
 			if klog.V(5) {
 				klog.Infof(volumeToAttach.GenerateMsgDetailed("Volume attached--touching", ""))
@@ -270,7 +269,7 @@ func (rc *reconciler) attachDesiredVolumes() {
 		}
 
 		if rc.isMultiAttachForbidden(volumeToAttach.VolumeSpec) {
-			nodes := rc.actualStateOfWorld.GetNodesForVolume(volumeToAttach.VolumeName)
+			nodes := rc.actualStateOfWorld.GetNodesForAttachedVolume(volumeToAttach.VolumeName)
 			if len(nodes) > 0 {
 				if !volumeToAttach.MultiAttachErrorReported {
 					rc.reportMultiAttachError(volumeToAttach, nodes)

pkg/kubelet/volumemanager/cache/actual_state_of_world.go

@@ -313,6 +313,11 @@ func (asw *actualStateOfWorld) MarkVolumeAsAttached(
 	return asw.addVolume(volumeName, volumeSpec, devicePath)
 }
 
+func (asw *actualStateOfWorld) MarkVolumeAsUncertain(
+	volumeName v1.UniqueVolumeName, volumeSpec *volume.Spec, _ types.NodeName) error {
+	return nil
+}
+
 func (asw *actualStateOfWorld) MarkVolumeAsDetached(
 	volumeName v1.UniqueVolumeName, nodeName types.NodeName) {
 	asw.DeleteVolume(volumeName)

pkg/volume/testing/testing.go

@@ -46,9 +46,19 @@ import (
 	"k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
 )
 
-// A hook specified in storage class to indicate it's provisioning
-// is expected to fail.
-const ExpectProvisionFailureKey = "expect-provision-failure"
+const (
+	// A hook specified in storage class to indicate it's provisioning
+	// is expected to fail.
+	ExpectProvisionFailureKey = "expect-provision-failure"
+	// The node is marked as uncertain. The attach operation will fail and return timeout error
+	// for the first attach call. The following call will return sucesssfully.
+	UncertainAttachNode = "uncertain-attach-node"
+	// The node is marked as timeout. The attach operation will always fail and return timeout error
+	// but the operation is actually succeeded.
+	TimeoutAttachNode = "timeout-attach-node"
+	// The node is marked as multi-attach which means it is allowed to attach the volume to multiple nodes.
+	MultiAttachNode = "multi-attach-node"
+)
 
 // fakeVolumeHost is useful for testing volume plugins.
 type fakeVolumeHost struct {
@@ -273,10 +283,20 @@ var _ DeviceMountableVolumePlugin = &FakeVolumePlugin{}
 var _ FSResizableVolumePlugin = &FakeVolumePlugin{}
 
 func (plugin *FakeVolumePlugin) getFakeVolume(list *[]*FakeVolume) *FakeVolume {
+	volumeList := *list
+	if list != nil && len(volumeList) > 0 {
+		volume := volumeList[0]
+		volume.Lock()
+		defer volume.Unlock()
+		volume.WaitForAttachHook = plugin.WaitForAttachHook
+		volume.UnmountDeviceHook = plugin.UnmountDeviceHook
+		return volume
+	}
 	volume := &FakeVolume{
 		WaitForAttachHook: plugin.WaitForAttachHook,
 		UnmountDeviceHook: plugin.UnmountDeviceHook,
 	}
+	volume.VolumesAttached = make(map[string]types.NodeName)
 	*list = append(*list, volume)
 	return volume
 }
@@ -333,6 +353,8 @@ func (plugin *FakeVolumePlugin) NewMounter(spec *Spec, pod *v1.Pod, opts VolumeO
 	plugin.Lock()
 	defer plugin.Unlock()
 	volume := plugin.getFakeVolume(&plugin.Mounters)
+	volume.Lock()
+	defer volume.Unlock()
 	volume.PodUID = pod.UID
 	volume.VolName = spec.Name()
 	volume.Plugin = plugin
@@ -350,6 +372,8 @@ func (plugin *FakeVolumePlugin) NewUnmounter(volName string, podUID types.UID) (
 	plugin.Lock()
 	defer plugin.Unlock()
 	volume := plugin.getFakeVolume(&plugin.Unmounters)
+	volume.Lock()
+	defer volume.Unlock()
 	volume.PodUID = podUID
 	volume.VolName = volName
 	volume.Plugin = plugin
@@ -368,6 +392,8 @@ func (plugin *FakeVolumePlugin) NewBlockVolumeMapper(spec *Spec, pod *v1.Pod, op
 	plugin.Lock()
 	defer plugin.Unlock()
 	volume := plugin.getFakeVolume(&plugin.BlockVolumeMappers)
+	volume.Lock()
+	defer volume.Unlock()
 	if pod != nil {
 		volume.PodUID = pod.UID
 	}
@@ -388,6 +414,8 @@ func (plugin *FakeVolumePlugin) NewBlockVolumeUnmapper(volName string, podUID ty
 	plugin.Lock()
 	defer plugin.Unlock()
 	volume := plugin.getFakeVolume(&plugin.BlockVolumeUnmappers)
+	volume.Lock()
+	defer volume.Unlock()
 	volume.PodUID = podUID
 	volume.VolName = volName
 	volume.Plugin = plugin
@@ -428,7 +456,16 @@ func (plugin *FakeVolumePlugin) NewDetacher() (Detacher, error) {
 	plugin.Lock()
 	defer plugin.Unlock()
 	plugin.NewDetacherCallCount = plugin.NewDetacherCallCount + 1
-	return plugin.getFakeVolume(&plugin.Detachers), nil
+	detacher := plugin.getFakeVolume(&plugin.Detachers)
+	attacherList := plugin.Attachers
+	if attacherList != nil && len(attacherList) > 0 {
+		detacherList := plugin.Detachers
+		if detacherList != nil && len(detacherList) > 0 {
+			detacherList[0].VolumesAttached = attacherList[0].VolumesAttached
+		}
+
+	}
+	return detacher, nil
 }
 
 func (plugin *FakeVolumePlugin) NewDeviceUnmounter() (DeviceUnmounter, error) {
@@ -657,6 +694,7 @@ type FakeVolume struct {
 	VolName string
 	Plugin  *FakeVolumePlugin
 	MetricsNil
+	VolumesAttached map[string]types.NodeName
 
 	// Add callbacks as needed
 	WaitForAttachHook func(spec *Spec, devicePath string, pod *v1.Pod, spectimeout time.Duration) (string, error)
@@ -677,6 +715,20 @@ type FakeVolume struct {
 	PodDeviceMapPathCallCount   int
 }
 
+func getUniqueVolumeName(spec *Spec) (string, error) {
+	var volumeName string
+	if spec.Volume != nil && spec.Volume.GCEPersistentDisk != nil {
+		volumeName = spec.Volume.GCEPersistentDisk.PDName
+	} else if spec.PersistentVolume != nil &&
+		spec.PersistentVolume.Spec.GCEPersistentDisk != nil {
+		volumeName = spec.PersistentVolume.Spec.GCEPersistentDisk.PDName
+	}
+	if volumeName == "" {
+		volumeName = spec.Name()
+	}
+	return volumeName, nil
+}
+
 func (_ *FakeVolume) GetAttributes() Attributes {
 	return Attributes{
 		ReadOnly:        false,
@@ -822,6 +874,25 @@ func (fv *FakeVolume) Attach(spec *Spec, nodeName types.NodeName) (string, error
 	fv.Lock()
 	defer fv.Unlock()
 	fv.AttachCallCount++
+	volumeName, err := getUniqueVolumeName(spec)
+	if err != nil {
+		return "", err
+	}
+	volumeNode, exist := fv.VolumesAttached[volumeName]
+	if exist {
+		if nodeName == UncertainAttachNode {
+			return "/dev/vdb-test", nil
+		}
+		if volumeNode == nodeName || volumeNode == MultiAttachNode || nodeName == MultiAttachNode {
+			return "/dev/vdb-test", nil
+		}
+		return "", fmt.Errorf("volume %q trying to attach to node %q is already attached to node %q", volumeName, nodeName, volumeNode)
+	}
+
+	fv.VolumesAttached[volumeName] = nodeName
+	if nodeName == UncertainAttachNode || nodeName == TimeoutAttachNode {
+		return "", fmt.Errorf("Timed out to attach volume %q to node %q", volumeName, nodeName)
+	}
 	return "/dev/vdb-test", nil
 }
 
@@ -871,6 +942,10 @@ func (fv *FakeVolume) Detach(volumeName string, nodeName types.NodeName) error {
 	fv.Lock()
 	defer fv.Unlock()
 	fv.DetachCallCount++
+	if _, exist := fv.VolumesAttached[volumeName]; !exist {
+		return fmt.Errorf("Trying to detach volume %q that is not attached to the node %q", volumeName, nodeName)
+	}
+	delete(fv.VolumesAttached, volumeName)
 	return nil
 }
 
@@ -1037,7 +1112,7 @@ func VerifyAttachCallCount(
 	fakeVolumePlugin *FakeVolumePlugin) error {
 	for _, attacher := range fakeVolumePlugin.GetAttachers() {
 		actualCallCount := attacher.GetAttachCallCount()
-		if actualCallCount == expectedAttachCallCount {
+		if actualCallCount >= expectedAttachCallCount {
 			return nil
 		}
 	}
@@ -1070,7 +1145,7 @@ func VerifyWaitForAttachCallCount(
 	fakeVolumePlugin *FakeVolumePlugin) error {
 	for _, attacher := range fakeVolumePlugin.GetAttachers() {
 		actualCallCount := attacher.GetWaitForAttachCallCount()
-		if actualCallCount == expectedWaitForAttachCallCount {
+		if actualCallCount >= expectedWaitForAttachCallCount {
 			return nil
 		}
 	}
@@ -1103,7 +1178,7 @@ func VerifyMountDeviceCallCount(
 	fakeVolumePlugin *FakeVolumePlugin) error {
 	for _, attacher := range fakeVolumePlugin.GetAttachers() {
 		actualCallCount := attacher.GetMountDeviceCallCount()
-		if actualCallCount == expectedMountDeviceCallCount {
+		if actualCallCount >= expectedMountDeviceCallCount {
 			return nil
 		}
 	}

pkg/volume/util/operationexecutor/operation_executor.go

@@ -192,6 +192,12 @@ type ActualStateOfWorldAttacherUpdater interface {
 	// volumes.  See issue 29695.
 	MarkVolumeAsAttached(volumeName v1.UniqueVolumeName, volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) error
 
+	// Marks the specified volume as *possibly* attached to the specified node.
+	// If an attach operation fails, the attach/detach controller does not know for certain if the volume is attached or not.
+	// If the volume name is supplied, that volume name will be used.  If not, the
+	// volume name is computed using the result from querying the plugin.
+	MarkVolumeAsUncertain(volumeName v1.UniqueVolumeName, volumeSpec *volume.Spec, nodeName types.NodeName) error
+
 	// Marks the specified volume as detached from the specified node
 	MarkVolumeAsDetached(volumeName v1.UniqueVolumeName, nodeName types.NodeName)
 

pkg/volume/util/operationexecutor/operation_generator.go

@@ -327,6 +327,12 @@ func (og *operationGenerator) GenerateAttachVolumeFunc(
 					klog.Errorf("AttachVolume.MarkVolumeAsAttached failed to fix dangling volume error for volume %q with %s", volumeToAttach.VolumeName, addErr)
 				}
 
+			} else {
+				addErr := actualStateOfWorld.MarkVolumeAsUncertain(
+					v1.UniqueVolumeName(""), volumeToAttach.VolumeSpec, volumeToAttach.NodeName)
+				if addErr != nil {
+					klog.Errorf("AttachVolume.MarkVolumeAsUncertain fail to add the volume %q to actual state with %s", volumeToAttach.VolumeName, addErr)
+				}
 			}
 			// On failure, return error. Caller will log and retry.
 			return volumeToAttach.GenerateError("AttachVolume.Attach failed", attachErr)