Automated cherry pick of #88569: fix: corrupted mount point in csi driver

pkg/volume/csi/BUILD

@@ -17,6 +17,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/features:go_default_library",
+        "//pkg/util/mount:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/csi/csiv0:go_default_library",
         "//pkg/volume/csi/nodeinfomanager:go_default_library",
@@ -81,6 +82,7 @@ go_test(
         "//staging/src/k8s.io/client-go/util/testing:go_default_library",
         "//staging/src/k8s.io/component-base/featuregate/testing:go_default_library",
         "//vendor/github.com/container-storage-interface/spec/lib/go/csi:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
 )

pkg/volume/csi/csi_attacher.go

@@ -291,13 +291,19 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 		return err
 	}
 
+	corruptedDir := false
 	mounted, err := isDirMounted(c.plugin, deviceMountPath)
 	if err != nil {
 		klog.Error(log("attacher.MountDevice failed while checking mount status for dir [%s]", deviceMountPath))
-		return err
+		if isCorruptedDir(deviceMountPath) {
+			corruptedDir = true // leave to CSI driver to handle corrupted mount
+			klog.Warning(log("attacher.MountDevice detected corrupted mount for dir [%s]", deviceMountPath))
+		} else {
+			return err
+		}
 	}
 
-	if mounted {
+	if mounted && !corruptedDir {
 		klog.V(4).Info(log("attacher.MountDevice skipping mount, dir already mounted [%s]", deviceMountPath))
 		return nil
 	}
@@ -314,7 +320,7 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 
 	// Store volume metadata for UnmountDevice. Keep it around even if the
 	// driver does not support NodeStage, UnmountDevice still needs it.
-	if err = os.MkdirAll(deviceMountPath, 0750); err != nil {
+	if err = os.MkdirAll(deviceMountPath, 0750); err != nil && !corruptedDir {
 		klog.Error(log("attacher.MountDevice failed to create dir %#v:  %v", deviceMountPath, err))
 		return err
 	}

pkg/volume/csi/csi_mounter.go

@@ -32,6 +32,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
 	utilstrings "k8s.io/utils/strings"
 )
@@ -100,13 +101,18 @@ func (c *csiMountMgr) SetUp(mounterArgs volume.MounterArgs) error {
 func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error {
 	klog.V(4).Infof(log("Mounter.SetUpAt(%s)", dir))
 
+	corruptedDir := false
 	mounted, err := isDirMounted(c.plugin, dir)
 	if err != nil {
-		klog.Error(log("mounter.SetUpAt failed while checking mount status for dir [%s]", dir))
-		return err
+		if isCorruptedDir(dir) {
+			corruptedDir = true // leave to CSI driver to handle corrupted mount
+			klog.Warning(log("mounter.SetUpAt detected corrupted mount for dir [%s]", dir))
+		} else {
+			return err
+		}
 	}
 
-	if mounted {
+	if mounted && !corruptedDir {
 		klog.V(4).Info(log("mounter.SetUpAt skipping mount, dir already mounted [%s]", dir))
 		return nil
 	}
@@ -209,7 +215,7 @@ func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error
 	}
 
 	// create target_dir before call to NodePublish
-	if err := os.MkdirAll(dir, 0750); err != nil {
+	if err := os.MkdirAll(dir, 0750); err != nil && !corruptedDir {
 		klog.Error(log("mouter.SetUpAt failed to create dir %#v:  %v", dir, err))
 		return err
 	}
@@ -425,6 +431,11 @@ func isDirMounted(plug *csiPlugin, dir string) (bool, error) {
 	return !notMnt, nil
 }
 
+func isCorruptedDir(dir string) bool {
+	_, pathErr := mount.PathExists(dir)
+	return pathErr != nil && mount.IsCorruptedMnt(pathErr)
+}
+
 // removeMountDir cleans the mount dir when dir is not mounted and removed the volume data file in dir
 func removeMountDir(plug *csiPlugin, mountPath string) error {
 	klog.V(4).Info(log("removing mount path [%s]", mountPath))

pkg/volume/csi/csi_mounter_test.go

@@ -18,6 +18,7 @@ package csi
 
 import (
 	"fmt"
+	"io/ioutil"
 	"math/rand"
 	"os"
 	"path"
@@ -26,6 +27,8 @@ import (
 
 	"reflect"
 
+	"github.com/stretchr/testify/assert"
+
 	api "k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -696,3 +699,33 @@ func TestUnmounterTeardown(t *testing.T) {
 	}
 
 }
+
+func TestIsCorruptedDir(t *testing.T) {
+	existingMountPath, err := ioutil.TempDir(os.TempDir(), "blobfuse-csi-mount-test")
+	if err != nil {
+		t.Fatalf("failed to create tmp dir: %v", err)
+	}
+	defer os.RemoveAll(existingMountPath)
+
+	tests := []struct {
+		desc           string
+		dir            string
+		expectedResult bool
+	}{
+		{
+			desc:           "NotExist dir",
+			dir:            "/tmp/NotExist",
+			expectedResult: false,
+		},
+		{
+			desc:           "Existing dir",
+			dir:            existingMountPath,
+			expectedResult: false,
+		},
+	}
+
+	for i, test := range tests {
+		isCorruptedDir := isCorruptedDir(test.dir)
+		assert.Equal(t, test.expectedResult, isCorruptedDir, "TestCase[%d]: %s", i, test.desc)
+	}
+}