Support GCEPD driver for Windows

This PR adds the support for gcepd driver to work on Windows node. The
driver will call csi-proxy API to perform file, volume and disk
operations.

Author: jingxu97

Makefile

@@ -17,15 +17,23 @@
 
 STAGINGIMAGE=${GCE_PD_CSI_STAGING_IMAGE}
 STAGINGVERSION=${GCE_PD_CSI_STAGING_VERSION}
+DRIVERBINARY=gce-pd-csi-driver
+DRIVERWINDOWSBINARY=${DRIVERBINARY}.exe
 
 all: gce-pd-driver
-
 gce-pd-driver:
 	mkdir -p bin
 ifndef GCE_PD_CSI_STAGING_VERSION
 	$(error "Must set environment variable GCE_PD_CSI_STAGING_VERSION to staging version")
 endif
-	go build -ldflags "-X main.vendorVersion=${STAGINGVERSION}" -o bin/gce-pd-csi-driver ./cmd/
+	go build -ldflags "-X main.vendorVersion=${STAGINGVERSION}" -o bin/${DRIVERBINARY} ./cmd/
+
+build-windows:
+	mkdir -p bin
+ifndef GCE_PD_CSI_STAGING_VERSION
+	$(error "Must set environment variable GCE_PD_CSI_STAGING_VERSION to staging version")
+endif
+	GOOS=windows go build -ldflags "-X main.vendorVersion=${STAGINGVERSION}" -o bin/${DRIVERWINDOWSBINARY} ./cmd/
 
 build-container:
 ifndef GCE_PD_CSI_STAGING_IMAGE

cmd/main.go

@@ -90,7 +90,10 @@ func handle() {
 	//Initialize requirements for the node service
 	var nodeServer *driver.GCENodeServer
 	if *runNodeService {
-		mounter := mountmanager.NewSafeMounter()
+		mounter, err := mountmanager.NewSafeMounter()
+		if err != nil {
+			klog.Fatalf("Failed to get safe mounter: %v", err)
+		}
 		deviceUtils := mountmanager.NewDeviceUtils()
 		statter := mountmanager.NewStatter()
 		meta, err := metadataservice.NewMetadataService()

go.mod

@@ -6,17 +6,19 @@ require (
 	cloud.google.com/go v0.45.1
 	github.com/GoogleCloudPlatform/k8s-cloud-provider v0.0.0-20190612171043-2e19bb35a278
 	github.com/container-storage-interface/spec v1.2.0
-	github.com/golang/protobuf v1.3.2
+	github.com/golang/protobuf v1.3.4
 	github.com/google/uuid v1.1.1
 	github.com/hashicorp/go-multierror v1.0.0 // indirect
+	github.com/kubernetes-csi/csi-proxy/client v0.0.0-20200330215040-9eff16441b2a
 	github.com/kubernetes-csi/csi-test/v3 v3.0.0
 	github.com/onsi/ginkgo v1.10.3
 	github.com/onsi/gomega v1.7.1
 	golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45
 	golang.org/x/sys v0.0.0-20191113165036-4c7a9d0fe056
+	golang.org/x/tools/gopls v0.3.3 // indirect
 	google.golang.org/api v0.10.0
 	google.golang.org/genproto v0.0.0-20191114150713-6bbd007550de
-	google.golang.org/grpc v1.25.1
+	google.golang.org/grpc v1.27.1
 	gopkg.in/gcfg.v1 v1.2.3
 	gopkg.in/warnings.v0 v0.1.2 // indirect
 	k8s.io/apimachinery v0.17.1

go.sum

@@ -17,6 +17,7 @@ package gceGCEDriver
 import (
 	"fmt"
 	"os"
+	"runtime"
 	"strconv"
 	"strings"
 
@@ -57,10 +58,20 @@ var _ csi.NodeServer = &GCENodeServer{}
 // node boot disk is considered an attachable disk so effective attach limit is
 // one less.
 const (
-	volumeLimitSmall int64 = 15
-	volumeLimitBig   int64 = 127
+	volumeLimitSmall     int64 = 15
+	volumeLimitBig       int64 = 127
+	defaultLinuxFsType         = "ext4"
+	defaultWindowsFsType       = "ntfs"
 )
 
+func getDefaultFsType() string {
+	if runtime.GOOS == "windows" {
+		return defaultWindowsFsType
+	} else {
+		return defaultLinuxFsType
+	}
+}
+
 func (ns *GCENodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
 	// Validate Arguments
 	targetPath := req.GetTargetPath()
@@ -129,10 +140,10 @@ func (ns *GCENodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePub
 		}
 
 		sourcePath = stagingTargetPath
-
-		if err := os.MkdirAll(targetPath, 0750); err != nil {
+		if err := preparePublishPath(targetPath, ns.Mounter); err != nil {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("mkdir failed on disk %s (%v)", targetPath, err))
 		}
+
 	} else if blk := volumeCapability.GetBlock(); blk != nil {
 		klog.V(4).Infof("NodePublishVolume with block volume mode")
 
@@ -141,7 +152,7 @@ func (ns *GCENodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePub
 			partition = part
 		}
 
-		sourcePath, err = ns.getDevicePath(volumeID, partition)
+		sourcePath, err = getDevicePath(ns, volumeID, partition)
 		if err != nil {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("Error when getting device path: %v", err))
 		}
@@ -218,11 +229,9 @@ func (ns *GCENodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeU
 	}
 	defer ns.volumeLocks.Release(volumeID)
 
-	err := mount.CleanupMountPoint(targetPath, ns.Mounter.Interface, false /* bind mount */)
-	if err != nil {
+	if err := cleanupPublishPath(targetPath, ns.Mounter); err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Unmount failed: %v\nUnmounting arguments: %s\n", err, targetPath))
 	}
-
 	klog.V(4).Infof("NodeUnpublishVolume succeded on %v from %s", volumeID, targetPath)
 	return &csi.NodeUnpublishVolumeResponse{}, nil
 }
@@ -264,27 +273,19 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 	if part, ok := req.GetVolumeContext()[common.VolumeAttributePartition]; ok {
 		partition = part
 	}
+	devicePath, err := getDevicePath(ns, volumeID, partition)
 
-	devicePath, err := ns.getDevicePath(volumeID, partition)
 	if err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Error when getting device path: %v", err))
 	}
 
 	klog.V(4).Infof("Successfully found attached GCE PD %q at device path %s.", volumeKey.Name, devicePath)
 
-	// Part 2: Check if mount already exists at targetpath
+	// Part 2: Check if mount already exists at stagingTargetPath
 	notMnt, err := ns.Mounter.Interface.IsLikelyNotMountPoint(stagingTargetPath)
-	if err != nil {
-		if os.IsNotExist(err) {
-			if err := os.MkdirAll(stagingTargetPath, 0750); err != nil {
-				return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to create directory (%q): %v", stagingTargetPath, err))
-			}
-			notMnt = true
-		} else {
-			return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown error when checking mount point (%q): %v", stagingTargetPath, err))
-		}
+	if err != nil && !os.IsNotExist(err) {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("cannot validate mount point: %s %v", stagingTargetPath, err))
 	}
-
 	if !notMnt {
 		// TODO(#95): Check who is mounted here. No error if its us
 		/*
@@ -293,15 +294,17 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 			3) Readonly MUST match
 
 		*/
-
 		klog.V(4).Infof("NodeStageVolume succeded on %v to %s, mount already exists.", volumeID, stagingTargetPath)
 		return &csi.NodeStageVolumeResponse{}, nil
 
 	}
+	if err := prepareStagePath(stagingTargetPath, ns.Mounter); err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("mkdir failed on disk %s (%v)", stagingTargetPath, err))
+	}
 
 	// Part 3: Mount device to stagingTargetPath
-	// Default fstype is ext4
-	fstype := "ext4"
+	fstype := getDefaultFsType()
+
 	options := []string{}
 	if mnt := volumeCapability.GetMount(); mnt != nil {
 		if mnt.FsType != "" {
@@ -316,7 +319,7 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		return &csi.NodeStageVolumeResponse{}, nil
 	}
 
-	err = ns.Mounter.FormatAndMount(devicePath, stagingTargetPath, fstype, options)
+	err = formatAndMount(devicePath, stagingTargetPath, fstype, options, ns.Mounter)
 	if err != nil {
 		return nil, status.Error(codes.Internal,
 			fmt.Sprintf("Failed to format and mount device from (%q) to (%q) with fstype (%q) and options (%q): %v",
@@ -343,9 +346,8 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 	}
 	defer ns.volumeLocks.Release(volumeID)
 
-	err := mount.CleanupMountPoint(stagingTargetPath, ns.Mounter.Interface, false /* bind mount */)
-	if err != nil {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("NodeUnstageVolume failed to unmount at path %s: %v", stagingTargetPath, err))
+	if err := cleanupStagePath(stagingTargetPath, ns.Mounter); err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("NodeUnstageVolume failed: %v\nUnmounting arguments: %s\n", err, stagingTargetPath))
 	}
 
 	klog.V(4).Infof("NodeUnstageVolume succeded on %v from %s", volumeID, stagingTargetPath)
@@ -454,7 +456,7 @@ func (ns *GCENodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpa
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("volume ID is invalid: %v", err))
 	}
 
-	devicePath, err := ns.getDevicePath(volumeID, "")
+	devicePath, err := getDevicePath(ns, volumeID, "")
 	if err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("error when getting device path for %s: %v", volumeID, err))
 	}
@@ -517,28 +519,6 @@ func (ns *GCENodeServer) GetVolumeLimits() (int64, error) {
 	return volumeLimitBig, nil
 }
 
-func (ns *GCENodeServer) getDevicePath(volumeID string, partition string) (string, error) {
-	volumeKey, err := common.VolumeIDToKey(volumeID)
-	if err != nil {
-		return "", err
-	}
-	deviceName, err := common.GetDeviceName(volumeKey)
-	if err != nil {
-		return "", fmt.Errorf("error getting device name: %v", err)
-	}
-
-	devicePaths := ns.DeviceUtils.GetDiskByIdPaths(deviceName, partition)
-	devicePath, err := ns.DeviceUtils.VerifyDevicePath(devicePaths, deviceName)
-
-	if err != nil {
-		return "", fmt.Errorf("error verifying GCE PD (%q) is attached: %v", volumeKey.Name, err)
-	}
-	if devicePath == "" {
-		return "", fmt.Errorf("unable to find device path out of attempted paths: %v", devicePaths)
-	}
-	return devicePath, nil
-}
-
 func (ns *GCENodeServer) getBlockSizeBytes(devicePath string) (int64, error) {
 	output, err := ns.Mounter.Exec.Command("blockdev", "--getsize64", devicePath).CombinedOutput()
 	if err != nil {

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

@@ -0,0 +1,66 @@
+// +build !windows
+
+/*
+Copyright 2020 The Kubernetes Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package gceGCEDriver
+
+import (
+	"fmt"
+	"os"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"k8s.io/utils/mount"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+)
+
+func getDevicePath(ns *GCENodeServer, volumeID, partition string) (string, error) {
+	volumeKey, err := common.VolumeIDToKey(volumeID)
+	if err != nil {
+		return "", err
+	}
+	deviceName, err := common.GetDeviceName(volumeKey)
+	if err != nil {
+		return "", fmt.Errorf("error getting device name: %v", err)
+	}
+	devicePaths := ns.DeviceUtils.GetDiskByIdPaths(deviceName, partition)
+	devicePath, err := ns.DeviceUtils.VerifyDevicePath(devicePaths, deviceName)
+	if err != nil {
+		return "", status.Error(codes.Internal, fmt.Sprintf("error verifying GCE PD (%q) is attached: %v", deviceName, err))
+	}
+	if devicePath == "" {
+		return "", status.Error(codes.Internal, fmt.Sprintf("Unable to find device path out of attempted paths: %v", devicePaths))
+	}
+	return devicePath, nil
+}
+
+func formatAndMount(source, target, fstype string, options []string, m *mount.SafeFormatAndMount) error {
+	return m.FormatAndMount(source, target, fstype, options)
+}
+
+func preparePublishPath(path string, m *mount.SafeFormatAndMount) error {
+	return os.MkdirAll(path, 0750)
+}
+
+func prepareStagePath(path string, m *mount.SafeFormatAndMount) error {
+	return os.MkdirAll(path, 0750)
+}
+
+func cleanupPublishPath(path string, m *mount.SafeFormatAndMount) error {
+	return mount.CleanupMountPoint(path, m, false /* bind mount */)
+}
+
+func cleanupStagePath(path string, m *mount.SafeFormatAndMount) error {
+	return cleanupPublishPath(path, m)
+}

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

@@ -0,0 +1,88 @@
+// +build windows
+
+/*
+Copyright 2020 The Kubernetes Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package gceGCEDriver
+
+import (
+	"fmt"
+	"k8s.io/utils/mount"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+	mounter "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
+)
+
+func formatAndMount(source, target, fstype string, options []string, m *mount.SafeFormatAndMount) error {
+	proxy, ok := m.Interface.(*mounter.CSIProxyMounter)
+	if !ok {
+		return fmt.Errorf("could not cast to csi proxy class")
+	}
+	return proxy.FormatAndMount(source, target, fstype, options)
+}
+
+// Before mounting (which means creating symlink) in Windows, the targetPath should
+// not exist. Currently kubelet creates the path beforehand, this is a workaround to
+// remove the path first.
+func preparePublishPath(path string, m *mount.SafeFormatAndMount) error {
+	proxy, ok := m.Interface.(*mounter.CSIProxyMounter)
+	if !ok {
+		return fmt.Errorf("could not cast to csi proxy class")
+	}
+	exists, err := proxy.ExistsPath(path)
+	if err != nil {
+		return err
+	}
+	if exists {
+		return proxy.RemovePodDir(path)
+	}
+	return nil
+}
+
+// Before staging (which means creating symlink) in Windows, the targetPath should
+// not exist. 
+func prepareStagePath(path string, m *mount.SafeFormatAndMount) error {
+	return nil
+}
+
+func cleanupPublishPath(path string, m *mount.SafeFormatAndMount) error {
+	proxy, ok := m.Interface.(*mounter.CSIProxyMounter)
+	if !ok {
+		return fmt.Errorf("could not cast to csi proxy class")
+	}
+	return proxy.RemovePodDir(path)
+}
+
+func cleanupStagePath(path string, m *mount.SafeFormatAndMount) error {
+	proxy, ok := m.Interface.(*mounter.CSIProxyMounter)
+	if !ok {
+		return fmt.Errorf("could not cast to csi proxy class")
+	}
+	return proxy.RemovePluginDir(path)
+}
+
+// search Windows disk number by volumeID
+func getDevicePath(ns *GCENodeServer, volumeID, partition string) (string, error) {
+	volumeKey, err := common.VolumeIDToKey(volumeID)
+	if err != nil {
+		return "", err
+	}
+	deviceName, err := common.GetDeviceName(volumeKey)
+	if err != nil {
+		return "", fmt.Errorf("error getting device name: %v", err)
+	}
+	proxy, ok := ns.Mounter.Interface.(*mounter.CSIProxyMounter)
+	if !ok {
+		return "", fmt.Errorf("could not cast to csi proxy class")
+	}
+	return proxy.GetDevicePath(deviceName, partition, volumeKey.Name)
+}