Merge pull request kubernetes#93567 from gnufied/fix-stale-attachments

Make AttachDisk for EBS idempotent again

Author: k8s-ci-robot

staging/src/k8s.io/legacy-cloud-providers/aws/aws.go

@@ -257,6 +257,14 @@ const (
 	filterNodeLimit = 150
 )
 
+const (
+	// represents expected attachment status of a volume after attach
+	volumeAttachedStatus = "attached"
+
+	// represents expected attachment status of a volume after detach
+	volumeDetachedStatus = "detached"
+)
+
 // awsTagNameMasterRoles is a set of well-known AWS tag names that indicate the instance is a master
 // The major consequence is that it is then not considered for AWS zone discovery for dynamic volume creation.
 var awsTagNameMasterRoles = sets.NewString("kubernetes.io/role/master", "k8s.io/role/master")
@@ -1943,7 +1951,6 @@ func (c *Cloud) getMountDevice(
 				// AWS API returns consistent result next time (i.e. the volume is detached).
 				status := volumeStatus[mappingVolumeID]
 				klog.Warningf("Got assignment call for already-assigned volume: %s@%s, volume status: %s", mountDevice, mappingVolumeID, status)
-				return mountDevice, false, fmt.Errorf("volume is still being detached from the node")
 			}
 			return mountDevice, true, nil
 		}
@@ -2144,7 +2151,7 @@ func (c *Cloud) applyUnSchedulableTaint(nodeName types.NodeName, reason string)
 
 // waitForAttachmentStatus polls until the attachment status is the expected value
 // On success, it returns the last attachment state.
-func (d *awsDisk) waitForAttachmentStatus(status string, expectedInstance, expectedDevice string) (*ec2.VolumeAttachment, error) {
+func (d *awsDisk) waitForAttachmentStatus(status string, expectedInstance, expectedDevice string, alreadyAttached bool) (*ec2.VolumeAttachment, error) {
 	backoff := wait.Backoff{
 		Duration: volumeAttachmentStatusPollDelay,
 		Factor:   volumeAttachmentStatusFactor,
@@ -2169,7 +2176,7 @@ func (d *awsDisk) waitForAttachmentStatus(status string, expectedInstance, expec
 		if err != nil {
 			// The VolumeNotFound error is special -- we don't need to wait for it to repeat
 			if isAWSErrorVolumeNotFound(err) {
-				if status == "detached" {
+				if status == volumeDetachedStatus {
 					// The disk doesn't exist, assume it's detached, log warning and stop waiting
 					klog.Warningf("Waiting for volume %q to be detached but the volume does not exist", d.awsID)
 					stateStr := "detached"
@@ -2178,7 +2185,7 @@ func (d *awsDisk) waitForAttachmentStatus(status string, expectedInstance, expec
 					}
 					return true, nil
 				}
-				if status == "attached" {
+				if status == volumeAttachedStatus {
 					// The disk doesn't exist, complain, give up waiting and report error
 					klog.Warningf("Waiting for volume %q to be attached but the volume does not exist", d.awsID)
 					return false, err
@@ -2213,7 +2220,7 @@ func (d *awsDisk) waitForAttachmentStatus(status string, expectedInstance, expec
 			}
 		}
 		if attachmentStatus == "" {
-			attachmentStatus = "detached"
+			attachmentStatus = volumeDetachedStatus
 		}
 		if attachment != nil {
 			// AWS eventual consistency can go back in time.
@@ -2242,6 +2249,13 @@ func (d *awsDisk) waitForAttachmentStatus(status string, expectedInstance, expec
 			}
 		}
 
+		// if we expected volume to be attached and it was reported as already attached via DescribeInstance call
+		// but DescribeVolume told us volume is detached, we will short-circuit this long wait loop and return error
+		// so as AttachDisk can be retried without waiting for 20 minutes.
+		if (status == volumeAttachedStatus) && alreadyAttached && (attachmentStatus != status) {
+			return false, fmt.Errorf("attachment of disk %q failed, expected device to be attached but was %s", d.name, attachmentStatus)
+		}
+
 		if attachmentStatus == status {
 			// Attachment is in requested state, finish waiting
 			return true, nil
@@ -2387,7 +2401,7 @@ func (c *Cloud) AttachDisk(diskName KubernetesVolumeID, nodeName types.NodeName)
 		klog.V(2).Infof("AttachVolume volume=%q instance=%q request returned %v", disk.awsID, awsInstance.awsID, attachResponse)
 	}
 
-	attachment, err := disk.waitForAttachmentStatus("attached", awsInstance.awsID, ec2Device)
+	attachment, err := disk.waitForAttachmentStatus("attached", awsInstance.awsID, ec2Device, alreadyAttached)
 
 	if err != nil {
 		if err == wait.ErrWaitTimeout {
@@ -2465,7 +2479,7 @@ func (c *Cloud) DetachDisk(diskName KubernetesVolumeID, nodeName types.NodeName)
 		return "", errors.New("no response from DetachVolume")
 	}
 
-	attachment, err := diskInfo.disk.waitForAttachmentStatus("detached", awsInstance.awsID, "")
+	attachment, err := diskInfo.disk.waitForAttachmentStatus("detached", awsInstance.awsID, "", false)
 	if err != nil {
 		return "", err
 	}
@@ -4773,7 +4787,7 @@ func setNodeDisk(
 }
 
 func getInitialAttachDetachDelay(status string) time.Duration {
-	if status == "detached" {
+	if status == volumeDetachedStatus {
 		return volumeDetachmentStatusInitialDelay
 	}
 	return volumeAttachmentStatusInitialDelay

test/e2e/storage/pd.go

@@ -451,6 +451,37 @@ var _ = utils.SIGDescribe("Pod Disks", func() {
 		ginkgo.By("delete a PD")
 		framework.ExpectNoError(e2epv.DeletePDWithRetry("non-exist"))
 	})
+
+	// This test is marked to run as serial so as device selection on AWS does not
+	// conflict with other concurrent attach operations.
+	ginkgo.It("[Serial] attach on previously attached volumes should work", func() {
+		e2eskipper.SkipUnlessProviderIs("gce", "gke", "aws")
+		ginkgo.By("creating PD")
+		diskName, err := e2epv.CreatePDWithRetry()
+		framework.ExpectNoError(err, "Error creating PD")
+
+		// this should be safe to do because if attach fails then detach will be considered
+		// successful and we will delete the volume.
+		defer func() {
+			detachAndDeletePDs(diskName, []types.NodeName{host0Name})
+		}()
+
+		ginkgo.By("Attaching volume to a node")
+		err = attachPD(host0Name, diskName)
+		framework.ExpectNoError(err, "Error attaching PD")
+
+		pod := testPDPod([]string{diskName}, host0Name /*readOnly*/, false, 1)
+		ginkgo.By("Creating test pod with same volume")
+		_, err = podClient.Create(context.TODO(), pod, metav1.CreateOptions{})
+		framework.ExpectNoError(err, "Failed to create pod")
+		framework.ExpectNoError(e2epod.WaitForPodRunningInNamespaceSlow(f.ClientSet, pod.Name, f.Namespace.Name))
+
+		ginkgo.By("deleting the pod")
+		framework.ExpectNoError(podClient.Delete(context.TODO(), pod.Name, *metav1.NewDeleteOptions(0)), "Failed to delete pod")
+		framework.Logf("deleted pod %q", pod.Name)
+		ginkgo.By("waiting for PD to detach")
+		framework.ExpectNoError(waitForPDDetach(diskName, host0Name))
+	})
 })
 
 func countReadyNodes(c clientset.Interface, hostName types.NodeName) int {
@@ -474,6 +505,7 @@ func verifyPDContentsViaContainer(namespace string, f *framework.Framework, podN
 	}
 }
 
+// TODO: move detachPD to standard cloudprovider functions so as these tests can run on other cloudproviders too
 func detachPD(nodeName types.NodeName, pdName string) error {
 	if framework.TestContext.Provider == "gce" || framework.TestContext.Provider == "gke" {
 		gceCloud, err := gce.GetGCECloud()
@@ -512,6 +544,38 @@ func detachPD(nodeName types.NodeName, pdName string) error {
 	}
 }
 
+// TODO: move attachPD to standard cloudprovider functions so as these tests can run on other cloudproviders too
+func attachPD(nodeName types.NodeName, pdName string) error {
+	if framework.TestContext.Provider == "gce" || framework.TestContext.Provider == "gke" {
+		gceCloud, err := gce.GetGCECloud()
+		if err != nil {
+			return err
+		}
+		err = gceCloud.AttachDisk(pdName, nodeName, false /*readOnly*/, false /*regional*/)
+		if err != nil {
+			framework.Logf("Error attaching PD %q: %v", pdName, err)
+		}
+		return err
+
+	} else if framework.TestContext.Provider == "aws" {
+		awsSession, err := session.NewSession()
+		if err != nil {
+			return fmt.Errorf("error creating session: %v", err)
+		}
+		client := ec2.New(awsSession)
+		tokens := strings.Split(pdName, "/")
+		awsVolumeID := tokens[len(tokens)-1]
+		ebsUtil := utils.NewEBSUtil(client)
+		err = ebsUtil.AttachDisk(awsVolumeID, string(nodeName))
+		if err != nil {
+			return fmt.Errorf("error attaching volume %s to node %s: %v", awsVolumeID, nodeName, err)
+		}
+		return nil
+	} else {
+		return fmt.Errorf("Provider does not support volume attaching")
+	}
+}
+
 // Returns pod spec suitable for api Create call. Handles gce, gke and aws providers only and
 // escapes if a different provider is supplied.
 // The first container name is hard-coded to "mycontainer". Subsequent containers are named:

test/e2e/storage/utils/BUILD

@@ -7,6 +7,7 @@ go_library(
     srcs = [
         "create.go",
         "deployment.go",
+        "ebs.go",
         "framework.go",
         "host_exec.go",
         "local.go",
@@ -36,9 +37,12 @@ go_library(
         "//test/e2e/framework/ssh:go_default_library",
         "//test/e2e/framework/testfiles:go_default_library",
         "//test/utils/image:go_default_library",
+        "//vendor/github.com/aws/aws-sdk-go/aws:go_default_library",
+        "//vendor/github.com/aws/aws-sdk-go/service/ec2:go_default_library",
         "//vendor/github.com/onsi/ginkgo:go_default_library",
         "//vendor/github.com/onsi/gomega:go_default_library",
         "//vendor/github.com/pkg/errors:go_default_library",
+        "//vendor/k8s.io/klog/v2:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",
     ],
 )