Check if publish target path already mounted to be idempotent

hack/update-gomock

@@ -13,10 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 set -euo pipefail
 
 IMPORT_PATH=github.com/kubernetes-sigs/aws-efs-csi-driver
-mockgen -package=mocks -destination=./pkg/driver/mocks/mock_mount.go ${IMPORT_PATH}/pkg/driver Mounter 
-mockgen -package=mocks -destination=./pkg/cloud/mocks/mock_ec2metadata.go ${IMPORT_PATH}/pkg/cloud EC2Metadata
-mockgen -package=mocks -destination=./pkg/cloud/mocks/mock_taskmetadata.go ${IMPORT_PATH}/pkg/cloud TaskMetadataService
+mockgen -package=mocks -destination=./pkg/driver/mocks/mock_mount.go --build_flags=--mod=mod ${IMPORT_PATH}/pkg/driver Mounter
+mockgen -package=mocks -destination=./pkg/cloud/mocks/mock_ec2metadata.go --build_flags=--mod=mod ${IMPORT_PATH}/pkg/cloud EC2Metadata
+mockgen -package=mocks -destination=./pkg/cloud/mocks/mock_taskmetadata.go --build_flags=--mod=mod ${IMPORT_PATH}/pkg/cloud TaskMetadataService

pkg/driver/mounter.go

@@ -22,7 +22,9 @@ import (
 // Mounter is an interface for mount operations
 type Mounter interface {
 	mount.Interface
+	IsCorruptedMnt(err error) bool
 	MakeDir(pathname string) error
+	PathExists(path string) (bool, error)
 	GetDeviceName(mountPath string) (string, int, error)
 }
 
@@ -49,3 +51,14 @@ func (m *NodeMounter) MakeDir(pathname string) error {
 func (m *NodeMounter) GetDeviceName(mountPath string) (string, int, error) {
 	return mount.GetDeviceNameFromMount(m, mountPath)
 }
+
+// IsCorruptedMnt return true if err is about corrupted mount point
+func (m NodeMounter) IsCorruptedMnt(err error) bool {
+	return mount.IsCorruptedMnt(err)
+}
+
+// This function is mirrored in ./sanity_test.go to make sure sanity test covered this block of code
+// Please mirror the change to func MakeFile in ./sanity_test.go
+func (m *NodeMounter) PathExists(path string) (bool, error) {
+	return mount.PathExists(path)
+}

pkg/driver/node.go

@@ -175,25 +175,75 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 		return nil, status.Errorf(codes.Internal, "Could not create dir %q: %v", target, err)
 	}
 
-	klog.V(5).Infof("NodePublishVolume: mounting %s at %s with options %v", source, target, mountOptions)
-	if err := d.mounter.Mount(source, target, "efs", mountOptions); err != nil {
-		os.Remove(target)
-		return nil, status.Errorf(codes.Internal, "Could not mount %q at %q: %v", source, target, err)
+	//Checking if the target directory is already mounted with a volume.
+	mounted, err := d.isMounted(source, target)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Could not check if %q is mounted: %v", target, err)
 	}
-	klog.V(5).Infof("NodePublishVolume: %s was mounted", target)
 
-	//Increment volume Id counter
-	if d.volMetricsOptIn {
-		if value, ok := volumeIdCounter[req.GetVolumeId()]; ok {
-			volumeIdCounter[req.GetVolumeId()] = value + 1
-		} else {
-			volumeIdCounter[req.GetVolumeId()] = 1
+	if !mounted {
+		klog.V(5).Infof("NodePublishVolume: mounting %s at %s with options %v", source, target, mountOptions)
+		if err := d.mounter.Mount(source, target, "efs", mountOptions); err != nil {
+			os.Remove(target)
+			return nil, status.Errorf(codes.Internal, "Could not mount %q at %q: %v", source, target, err)
+		}
+		klog.V(5).Infof("NodePublishVolume: %s was mounted", target)
+
+		//Increment volume Id counter
+		if d.volMetricsOptIn {
+			if value, ok := volumeIdCounter[req.GetVolumeId()]; ok {
+				volumeIdCounter[req.GetVolumeId()] = value + 1
+			} else {
+				volumeIdCounter[req.GetVolumeId()] = 1
+			}
 		}
 	}
 
 	return &csi.NodePublishVolumeResponse{}, nil
 }
 
+// isMounted checks if target is mounted. It does NOT return an error if target
+// doesn't exist.
+func (d *Driver) isMounted(source string, target string) (bool, error) {
+	/*
+		Checking if it's a mount point using IsLikelyNotMountPoint. There are three different return values,
+		1. true, err when the directory does not exist or corrupted.
+		2. false, nil when the path is already mounted with a device.
+		3. true, nil when the path is not mounted with any device.
+	*/
+	notMnt, err := d.mounter.IsLikelyNotMountPoint(target)
+	if err != nil && !os.IsNotExist(err) {
+		//Checking if the path exists and error is related to Corrupted Mount, in that case, the system could unmount and mount.
+		_, pathErr := d.mounter.PathExists(target)
+		if pathErr != nil && d.mounter.IsCorruptedMnt(pathErr) {
+			klog.V(4).Infof("NodePublishVolume: Target path %q is a corrupted mount. Trying to unmount.", target)
+			if mntErr := d.mounter.Unmount(target); mntErr != nil {
+				return false, status.Errorf(codes.Internal, "Unable to unmount the target %q : %v", target, mntErr)
+			}
+			//After successful unmount, the device is ready to be mounted.
+			return false, nil
+		}
+		return false, status.Errorf(codes.Internal, "Could not check if %q is a mount point: %v, %v", target, err, pathErr)
+	}
+
+	// Do not return os.IsNotExist error. Other errors were handled above.  The
+	// Existence of the target should be checked by the caller explicitly and
+	// independently because sometimes prior to mount it is expected not to exist
+	// (in Windows, the target must NOT exist before a symlink is created at it)
+	// and in others it is an error (in Linux, the target mount directory must
+	// exist before mount is called on it)
+	if err != nil && os.IsNotExist(err) {
+		klog.V(5).Infof("[Debug] NodePublishVolume: Target path %q does not exist", target)
+		return false, nil
+	}
+
+	if !notMnt {
+		klog.V(4).Infof("NodePublishVolume: Target path %q is already mounted", target)
+	}
+
+	return !notMnt, nil
+}
+
 func (d *Driver) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
 	klog.V(4).Infof("NodeUnpublishVolume: called with args %+v", req)