Do not fail volume attach operation at kubelet if staging target path already exists on windows worker node.

pkg/util/filesystem/defaultfs.go

@@ -17,8 +17,10 @@ limitations under the License.
 package filesystem
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
 	"strings"
 	"time"
 )
@@ -75,6 +77,32 @@ func (fs *DefaultFs) MkdirAll(path string, perm os.FileMode) error {
 	return os.MkdirAll(fs.prefix(path), perm)
 }
 
+// MkdirAllWithPathCheck checks if path exists already. If not, it creates a directory
+// named path, along with any necessary parents, and returns nil, or else returns an error.
+// Permission bits perm (before umask) are used for all directories that
+// MkdirAllWithPathCheck creates.
+// If path is already a directory, MkdirAllWithPathCheck does nothing and returns nil.
+// NOTE: In case of Windows NTFS, mount points are implemented as reparse-point
+// (similar to symlink) and do not represent actual directory. Hence Directory existence
+// check for windows NTFS will NOT check for dir, but for symlink presence.
+func MkdirAllWithPathCheck(path string, perm os.FileMode) error {
+	if dir, err := os.Lstat(path); err == nil {
+		// If the path exists already,
+		// 1. for Unix/Linux OS, check if the path is directory.
+		// 2. for windows NTFS, check if the path is symlink instead of directory.
+		if dir.IsDir() ||
+			(runtime.GOOS == "windows" && (dir.Mode()&os.ModeSymlink != 0)) {
+			return nil
+		}
+		return fmt.Errorf("path %v exists but is not a directory", path)
+	}
+	// If existence of path not known, attempt to create it.
+	if err := os.MkdirAll(path, perm); err != nil {
+		return err
+	}
+	return nil
+}
+
 // Chtimes via os.Chtimes
 func (fs *DefaultFs) Chtimes(name string, atime time.Time, mtime time.Time) error {
 	return os.Chtimes(fs.prefix(name), atime, mtime)

pkg/volume/csi/csi_attacher.go

@@ -37,6 +37,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/util/filesystem"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/util"
 	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
@@ -341,9 +342,10 @@ 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 = filesystem.MkdirAllWithPathCheck(deviceMountPath, 0750); err != nil {
 		return errors.New(log("attacher.MountDevice failed to create dir %#v:  %v", deviceMountPath, err))
 	}
+
 	klog.V(4).Info(log("created target path successfully [%s]", deviceMountPath))
 	dataDir := filepath.Dir(deviceMountPath)
 	data := map[string]string{