Add logic to force release volume in NodeUnstageVolume if it has not been released due to an error for more than 30 seconds

Author: davis-haba

pkg/gce-pd-csi-driver/node.go

@@ -51,6 +51,10 @@ type GCENodeServer struct {
 	// for that same volume (as defined by VolumeID) return an Aborted error
 	volumeLocks *common.VolumeLocks
 
+	// deviceInUseErrors keeps tracks of device names and a timestamp for when an error is
+	// encounted for that device
+	deviceInUseErrors map[string]time.Time
+
 	// If set, this semaphore will be used to serialize formatAndMount. It will be raised
 	// when the operation starts, and lowered either when finished, or when
 	// formatAndMountTimeout has expired.
@@ -81,6 +85,8 @@ const (
 	fsTypeExt3                 = "ext3"
 
 	readAheadKBMountFlagRegexPattern = "^read_ahead_kb=(.+)$"
+
+	deviceInUseTimeout = 30
 )
 
 var (
@@ -94,6 +100,36 @@ func getDefaultFsType() string {
 		return defaultLinuxFsType
 	}
 }
+
+// checkDeviceErrorTimeout returns true an error was encounted for the specified deviceName,
+// where the error happened at least `deviceInUseTimeout` seconds ago.
+func (ns *GCENodeServer) checkDeviceErrorTimeout(deviceName string) bool {
+	if ns.deviceInUseErrors == nil {
+		ns.deviceInUseErrors = make(map[string]time.Time)
+	}
+
+	lastErrTime, exists := ns.deviceInUseErrors[deviceName]
+	if exists && time.Now().Sub(lastErrTime).Seconds() >= deviceInUseTimeout {
+		return true
+	}
+	return false
+}
+
+// markDeviceError updates the internal `deviceInUseErrors` map to denote an error was encounted
+// for the specified deviceName at the current time
+func (ns *GCENodeServer) markDeviceError(deviceName string) {
+	if ns.deviceInUseErrors == nil {
+		ns.deviceInUseErrors = make(map[string]time.Time)
+	}
+
+	// If an earlier error has already been recorded, do not overwrite it
+	if _, exists := ns.deviceInUseErrors[deviceName]; !exists {
+		now := time.Now()
+		klog.V(4).Infof("Recording in-use error for device %s at time %s", deviceName, now)
+		ns.deviceInUseErrors[deviceName] = now
+	}
+}
+
 func (ns *GCENodeServer) isVolumePathMounted(path string) bool {
 	notMnt, err := ns.Mounter.Interface.IsLikelyNotMountPoint(path)
 	klog.V(4).Infof("Checking volume path %s is mounted %t: error %v", path, !notMnt, err)
@@ -459,11 +495,15 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 		var targetErr *ignoreableError
 		if errors.As(err, &targetErr) {
 			klog.Warningf("Unabled to check if device for %s is unused. Device has been unmounted successfully. Ignoring and continuing with unstaging. (%v)", volumeID, err)
+		} else if ns.checkDeviceErrorTimeout(volumeID) {
+			klog.Warningf("Device %s could not be released after timeout of %d seconds. NodeUnstageVolume will return success.", volumeID, deviceInUseTimeout)
 		} else {
+			ns.markDeviceError(volumeID)
 			return nil, status.Errorf(codes.Internal, "NodeUnstageVolume for volume %s failed: %v", volumeID, err)
 		}
 	}
 
+	delete(ns.deviceInUseErrors, volumeID)
 	klog.V(4).Infof("NodeUnstageVolume succeeded on %v from %s", volumeID, stagingTargetPath)
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }

test/e2e/tests/single_zone_e2e_test.go

@@ -1437,6 +1437,82 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}()
 	})
 
+	It("Should unpublish if there is an error unpublishing and device has been in use longer than timeout", func() {
+		testContext := getRandomTestContext()
+
+		p, z, _ := testContext.Instance.GetIdentity()
+		client := testContext.Client
+		instance := testContext.Instance
+
+		// Create Disk
+		volName, volID := createAndValidateUniqueZonalDisk(client, p, z, standardDiskType)
+
+		defer func() {
+			// Delete Disk
+			err := client.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, z, volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+
+		// Attach Disk
+		err := client.ControllerPublishVolumeReadWrite(volID, instance.GetNodeID(), false /* forceAttach */)
+		Expect(err).To(BeNil(), "ControllerPublishVolume failed with error for disk %v on node %v: %v", volID, instance.GetNodeID(), err)
+
+		defer func() {
+			// Detach Disk
+			err = client.ControllerUnpublishVolume(volID, instance.GetNodeID())
+			if err != nil {
+				klog.Errorf("Failed to detach disk: %v", err)
+			}
+		}()
+
+		// Stage Disk
+		stageDir := filepath.Join("/tmp/", volName, "stage")
+		err = client.NodeStageExt4Volume(volID, stageDir)
+		Expect(err).To(BeNil(), "failed to stage volume: %v", err)
+
+		// Create private bind mount to keep the device's state as "in use"
+		boundMountStageDir := filepath.Join("/tmp/bindmount", volName, "bindmount")
+		boundMountStageMkdirOutput, err := instance.SSH("mkdir", "-p", boundMountStageDir)
+		Expect(err).To(BeNil(), "mkdir failed on instance %v: output: %v: %v", instance.GetNodeID(), boundMountStageMkdirOutput, err)
+		bindMountOutput, err := instance.SSH("mount", "--rbind", "--make-private", stageDir, boundMountStageDir)
+		Expect(err).To(BeNil(), "Bind mount failed on instance %v: output: %v: %v", instance.GetNodeID(), bindMountOutput, err)
+
+		privateBindMountRemoved := false
+		unmountAndRmPrivateBindMount := func() {
+			if !privateBindMountRemoved {
+				// Umount and delete private mount staging directory
+				bindUmountOutput, err := instance.SSH("umount", boundMountStageDir)
+				Expect(err).To(BeNil(), "Bind mount failed on instance %v: output: %v: %v", instance.GetNodeID(), bindUmountOutput, err)
+				err = testutils.RmAll(instance, boundMountStageDir)
+				Expect(err).To(BeNil(), "Failed to rm mount stage dir %s: %v", boundMountStageDir, err)
+			}
+			privateBindMountRemoved = true
+		}
+
+		defer func() {
+			unmountAndRmPrivateBindMount()
+		}()
+
+		// Unstage Disk. This should record a "deviceInUse" error
+		err = client.NodeUnstageVolume(volID, stageDir)
+		Expect(err).ToNot(BeNil(), "Expected failure during unstage")
+		Expect(err).To(MatchError(ContainSubstring(("is still in use"))))
+
+		// Wait 35s (30s timeout + 5s buffer) and try again
+		time.Sleep(35 * time.Second)
+		err = client.NodeUnstageVolume(volID, stageDir)
+		Expect(err).To(BeNil(), "Failed to unpublish after 30s in-use timeout for volume: %s, stageDir: %s", volID, stageDir)
+
+		// Unstage Disk
+		fp := filepath.Join("/tmp/", volName)
+		err = testutils.RmAll(instance, fp)
+		Expect(err).To(BeNil(), "Failed to rm file path %s: %v", fp, err)
+	})
+
 	It("Should block unstage if filesystem mounted", func() {
 		testContext := getRandomTestContext()