Merge pull request #71314 from saad-ali/csi03Compat

hack/.golint_failures

@@ -394,6 +394,7 @@ pkg/version/verflag
 pkg/volume
 pkg/volume/azure_dd
 pkg/volume/azure_file
+pkg/volume/csi/csiv0
 pkg/volume/csi/fake
 pkg/volume/git_repo
 pkg/volume/host_path

pkg/kubelet/cm/devicemanager/manager.go

@@ -246,7 +246,7 @@ func (m *ManagerImpl) GetWatcherHandler() watcher.PluginHandler {
 }
 
 // ValidatePlugin validates a plugin if the version is correct and the name has the format of an extended resource
-func (m *ManagerImpl) ValidatePlugin(pluginName string, endpoint string, versions []string) error {
+func (m *ManagerImpl) ValidatePlugin(pluginName string, endpoint string, versions []string, foundInDeprecatedDir bool) error {
 	klog.V(2).Infof("Got Plugin %s at endpoint %s with versions %v", pluginName, endpoint, versions)
 
 	if !m.isVersionCompatibleWithPlugin(versions) {
@@ -263,7 +263,7 @@ func (m *ManagerImpl) ValidatePlugin(pluginName string, endpoint string, version
 // RegisterPlugin starts the endpoint and registers it
 // TODO: Start the endpoint and wait for the First ListAndWatch call
 //       before registering the plugin
-func (m *ManagerImpl) RegisterPlugin(pluginName string, endpoint string) error {
+func (m *ManagerImpl) RegisterPlugin(pluginName string, endpoint string, versions []string) error {
 	klog.V(2).Infof("Registering Plugin %s at endpoint %s", pluginName, endpoint)
 
 	e, err := newEndpointImpl(endpoint, pluginName, m.callback)

pkg/kubelet/cm/devicemanager/manager_test.go

@@ -248,7 +248,7 @@ func setupDevicePlugin(t *testing.T, devs []*pluginapi.Device, pluginSocketName
 }
 
 func setupPluginWatcher(pluginSocketName string, m Manager) *pluginwatcher.Watcher {
-	w := pluginwatcher.NewWatcher(filepath.Dir(pluginSocketName))
+	w := pluginwatcher.NewWatcher(filepath.Dir(pluginSocketName), "" /* deprecatedSockDir */)
 	w.AddHandler(watcherapi.DevicePlugin, m.GetWatcherHandler())
 	w.Start()
 

pkg/kubelet/kubelet.go

@@ -789,7 +789,10 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		return nil, err
 	}
 	if klet.enablePluginsWatcher {
-		klet.pluginWatcher = pluginwatcher.NewWatcher(klet.getPluginsRegistrationDir())
+		klet.pluginWatcher = pluginwatcher.NewWatcher(
+			klet.getPluginsRegistrationDir(), /* sockDir */
+			klet.getPluginsDir(),             /* deprecatedSockDir */
+		)
 	}
 
 	// If the experimentalMounterPathFlag is set, we do not want to

pkg/kubelet/util/pluginwatcher/example_handler.go

@@ -38,6 +38,8 @@ type exampleHandler struct {
 
 	m     sync.Mutex
 	count int
+
+	permitDeprecatedDir bool
 }
 
 type examplePluginEvent int
@@ -50,16 +52,21 @@ const (
 )
 
 // NewExampleHandler provide a example handler
-func NewExampleHandler(supportedVersions []string) *exampleHandler {
+func NewExampleHandler(supportedVersions []string, permitDeprecatedDir bool) *exampleHandler {
 	return &exampleHandler{
 		SupportedVersions: supportedVersions,
 		ExpectedNames:     make(map[string]int),
 
-		eventChans: make(map[string]chan examplePluginEvent),
+		eventChans:          make(map[string]chan examplePluginEvent),
+		permitDeprecatedDir: permitDeprecatedDir,
 	}
 }
 
-func (p *exampleHandler) ValidatePlugin(pluginName string, endpoint string, versions []string) error {
+func (p *exampleHandler) ValidatePlugin(pluginName string, endpoint string, versions []string, foundInDeprecatedDir bool) error {
+	if foundInDeprecatedDir && !p.permitDeprecatedDir {
+		return fmt.Errorf("device plugin socket was found in a directory that is no longer supported and this test does not permit plugins from deprecated dir")
+	}
+
 	p.SendEvent(pluginName, exampleEventValidate)
 
 	n, ok := p.DecreasePluginCount(pluginName)
@@ -79,7 +86,7 @@ func (p *exampleHandler) ValidatePlugin(pluginName string, endpoint string, vers
 	return nil
 }
 
-func (p *exampleHandler) RegisterPlugin(pluginName, endpoint string) error {
+func (p *exampleHandler) RegisterPlugin(pluginName, endpoint string, versions []string) error {
 	p.SendEvent(pluginName, exampleEventRegister)
 
 	// Verifies the grpcServer is ready to serve services.

pkg/kubelet/util/pluginwatcher/plugin_watcher.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -36,11 +37,12 @@ import (
 
 // Watcher is the plugin watcher
 type Watcher struct {
-	path      string
-	stopCh    chan interface{}
-	fs        utilfs.Filesystem
-	fsWatcher *fsnotify.Watcher
-	wg        sync.WaitGroup
+	path           string
+	deprecatedPath string
+	stopCh         chan interface{}
+	fs             utilfs.Filesystem
+	fsWatcher      *fsnotify.Watcher
+	wg             sync.WaitGroup
 
 	mutex       sync.Mutex
 	handlers    map[string]PluginHandler
@@ -54,10 +56,13 @@ type pathInfo struct {
 }
 
 // NewWatcher provides a new watcher
-func NewWatcher(sockDir string) *Watcher {
+// deprecatedSockDir refers to a pre-GA directory that was used by older plugins
+// for socket registration. New plugins should not use this directory.
+func NewWatcher(sockDir string, deprecatedSockDir string) *Watcher {
 	return &Watcher{
-		path: sockDir,
-		fs:   &utilfs.DefaultFs{},
+		path:           sockDir,
+		deprecatedPath: deprecatedSockDir,
+		fs:             &utilfs.DefaultFs{},
 
 		handlers:    make(map[string]PluginHandler),
 		plugins:     make(map[string]pathInfo),
@@ -137,7 +142,15 @@ func (w *Watcher) Start() error {
 	// Traverse plugin dir after starting the plugin processing goroutine
 	if err := w.traversePluginDir(w.path); err != nil {
 		w.Stop()
-		return fmt.Errorf("failed to traverse plugin socket path, err: %v", err)
+		return fmt.Errorf("failed to traverse plugin socket path %q, err: %v", w.path, err)
+	}
+
+	// Traverse deprecated plugin dir, if specified.
+	if len(w.deprecatedPath) != 0 {
+		if err := w.traversePluginDir(w.deprecatedPath); err != nil {
+			w.Stop()
+			return fmt.Errorf("failed to traverse deprecated plugin socket path %q, err: %v", w.deprecatedPath, err)
+		}
 	}
 
 	return nil
@@ -190,6 +203,10 @@ func (w *Watcher) traversePluginDir(dir string) error {
 
 		switch mode := info.Mode(); {
 		case mode.IsDir():
+			if w.containsBlacklistedDir(path) {
+				return filepath.SkipDir
+			}
+
 			if err := w.fsWatcher.Add(path); err != nil {
 				return fmt.Errorf("failed to watch %s, err: %v", path, err)
 			}
@@ -216,6 +233,10 @@ func (w *Watcher) traversePluginDir(dir string) error {
 func (w *Watcher) handleCreateEvent(event fsnotify.Event) error {
 	klog.V(6).Infof("Handling create event: %v", event)
 
+	if w.containsBlacklistedDir(event.Name) {
+		return nil
+	}
+
 	fi, err := os.Stat(event.Name)
 	if err != nil {
 		return fmt.Errorf("stat file %s failed: %v", event.Name, err)
@@ -271,8 +292,10 @@ func (w *Watcher) handlePluginRegistration(socketPath string) error {
 		infoResp.Endpoint = socketPath
 	}
 
+	foundInDeprecatedDir := w.foundInDeprecatedDir(socketPath)
+
 	// calls handler callback to verify registration request
-	if err := handler.ValidatePlugin(infoResp.Name, infoResp.Endpoint, infoResp.SupportedVersions); err != nil {
+	if err := handler.ValidatePlugin(infoResp.Name, infoResp.Endpoint, infoResp.SupportedVersions, foundInDeprecatedDir); err != nil {
 		return w.notifyPlugin(client, false, fmt.Sprintf("plugin validation failed with err: %v", err))
 	}
 
@@ -280,7 +303,7 @@ func (w *Watcher) handlePluginRegistration(socketPath string) error {
 	// so that if we receive a delete event during Register Plugin, we can process it as a DeRegister call.
 	w.registerPlugin(socketPath, infoResp.Type, infoResp.Name)
 
-	if err := handler.RegisterPlugin(infoResp.Name, infoResp.Endpoint); err != nil {
+	if err := handler.RegisterPlugin(infoResp.Name, infoResp.Endpoint, infoResp.SupportedVersions); err != nil {
 		return w.notifyPlugin(client, false, fmt.Sprintf("plugin registration failed with err: %v", err))
 	}
 
@@ -417,3 +440,27 @@ func dial(unixSocketPath string, timeout time.Duration) (registerapi.Registratio
 
 	return registerapi.NewRegistrationClient(c), c, nil
 }
+
+// While deprecated dir is supported, to add extra protection around #69015
+// we will explicitly blacklist kubernetes.io directory.
+func (w *Watcher) containsBlacklistedDir(path string) bool {
+	return strings.HasPrefix(path, w.deprecatedPath+"/kubernetes.io/") ||
+		path == w.deprecatedPath+"/kubernetes.io"
+}
+
+func (w *Watcher) foundInDeprecatedDir(socketPath string) bool {
+	if len(w.deprecatedPath) != 0 {
+		if socketPath == w.deprecatedPath {
+			return true
+		}
+
+		deprecatedPath := w.deprecatedPath
+		if !strings.HasSuffix(deprecatedPath, "/") {
+			deprecatedPath = deprecatedPath + "/"
+		}
+		if strings.HasPrefix(socketPath, deprecatedPath) {
+			return true
+		}
+	}
+	return false
+}

pkg/kubelet/util/pluginwatcher/types.go

@@ -48,11 +48,11 @@ package pluginwatcher
 type PluginHandler interface {
 	// Validate returns an error if the information provided by
 	// the potential plugin is erroneous (unsupported version, ...)
-	ValidatePlugin(pluginName string, endpoint string, versions []string) error
+	ValidatePlugin(pluginName string, endpoint string, versions []string, foundInDeprecatedDir bool) error
 	// RegisterPlugin is called so that the plugin can be register by any
 	// plugin consumer
 	// Error encountered here can still be Notified to the plugin.
-	RegisterPlugin(pluginName, endpoint string) error
+	RegisterPlugin(pluginName, endpoint string, versions []string) error
 	// DeRegister is called once the pluginwatcher observes that the socket has
 	// been deleted.
 	DeRegisterPlugin(pluginName string)

pkg/volume/csi/BUILD

@@ -16,6 +16,7 @@ go_library(
         "//pkg/features:go_default_library",
         "//pkg/util/strings:go_default_library",
         "//pkg/volume:go_default_library",
+        "//pkg/volume/csi/csiv0:go_default_library",
         "//pkg/volume/csi/nodeinfomanager:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
@@ -23,6 +24,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/version:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
@@ -86,6 +88,7 @@ filegroup(
     name = "all-srcs",
     srcs = [
         ":package-srcs",
+        "//pkg/volume/csi/csiv0:all-srcs",
         "//pkg/volume/csi/fake:all-srcs",
         "//pkg/volume/csi/nodeinfomanager:all-srcs",
     ],

pkg/volume/csi/csi_attacher.go

@@ -29,7 +29,6 @@ import (
 
 	"k8s.io/klog"
 
-	csipb "github.com/container-storage-interface/spec/lib/go/csi"
 	"k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1beta1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
@@ -342,14 +341,18 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 	}()
 
 	if c.csiClient == nil {
-		c.csiClient = newCsiDriverClient(csiSource.Driver)
+		c.csiClient, err = newCsiDriverClient(csiDriverName(csiSource.Driver))
+		if err != nil {
+			klog.Errorf(log("attacher.MountDevice failed to create newCsiDriverClient: %v", err))
+			return err
+		}
 	}
 	csi := c.csiClient
 
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()
 	// Check whether "STAGE_UNSTAGE_VOLUME" is set
-	stageUnstageSet, err := hasStageUnstageCapability(ctx, csi)
+	stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)
 	if err != nil {
 		return err
 	}
@@ -361,7 +364,7 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 
 	// Start MountDevice
 	nodeName := string(c.plugin.host.GetNodeName())
-	publishVolumeInfo, err := c.plugin.getPublishVolumeInfo(c.k8s, csiSource.VolumeHandle, csiSource.Driver, nodeName)
+	publishContext, err := c.plugin.getPublishContext(c.k8s, csiSource.VolumeHandle, csiSource.Driver, nodeName)
 
 	nodeStageSecrets := map[string]string{}
 	if csiSource.NodeStageSecretRef != nil {
@@ -382,7 +385,7 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 	fsType := csiSource.FSType
 	err = csi.NodeStageVolume(ctx,
 		csiSource.VolumeHandle,
-		publishVolumeInfo,
+		publishContext,
 		deviceMountPath,
 		fsType,
 		accessMode,
@@ -522,14 +525,18 @@ func (c *csiAttacher) UnmountDevice(deviceMountPath string) error {
 	}
 
 	if c.csiClient == nil {
-		c.csiClient = newCsiDriverClient(driverName)
+		c.csiClient, err = newCsiDriverClient(csiDriverName(driverName))
+		if err != nil {
+			klog.Errorf(log("attacher.UnmountDevice failed to create newCsiDriverClient: %v", err))
+			return err
+		}
 	}
 	csi := c.csiClient
 
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()
 	// Check whether "STAGE_UNSTAGE_VOLUME" is set
-	stageUnstageSet, err := hasStageUnstageCapability(ctx, csi)
+	stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)
 	if err != nil {
 		klog.Errorf(log("attacher.UnmountDevice failed to check whether STAGE_UNSTAGE_VOLUME set: %v", err))
 		return err
@@ -563,24 +570,6 @@ func (c *csiAttacher) UnmountDevice(deviceMountPath string) error {
 	return nil
 }
 
-func hasStageUnstageCapability(ctx context.Context, csi csiClient) (bool, error) {
-	capabilities, err := csi.NodeGetCapabilities(ctx)
-	if err != nil {
-		return false, err
-	}
-
-	stageUnstageSet := false
-	if capabilities == nil {
-		return false, nil
-	}
-	for _, capability := range capabilities {
-		if capability.GetRpc().GetType() == csipb.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME {
-			stageUnstageSet = true
-		}
-	}
-	return stageUnstageSet, nil
-}
-
 // getAttachmentName returns csi-<sha252(volName,csiDriverName,NodeName>
 func getAttachmentName(volName, csiDriverName, nodeName string) string {
 	result := sha256.Sum256([]byte(fmt.Sprintf("%s%s%s", volName, csiDriverName, nodeName)))

pkg/volume/csi/csi_block.go

@@ -40,7 +40,7 @@ type csiBlockMapper struct {
 	k8s        kubernetes.Interface
 	csiClient  csiClient
 	plugin     *csiPlugin
-	driverName string
+	driverName csiDriverName
 	specName   string
 	volumeID   string
 	readOnly   bool
@@ -96,7 +96,7 @@ func (m *csiBlockMapper) stageVolumeForBlock(
 	klog.V(4).Infof(log("blockMapper.stageVolumeForBlock stagingPath set [%s]", stagingPath))
 
 	// Check whether "STAGE_UNSTAGE_VOLUME" is set
-	stageUnstageSet, err := hasStageUnstageCapability(ctx, csi)
+	stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)
 	if err != nil {
 		klog.Error(log("blockMapper.stageVolumeForBlock failed to check STAGE_UNSTAGE_VOLUME capability: %v", err))
 		return "", err
@@ -287,7 +287,7 @@ func (m *csiBlockMapper) unpublishVolumeForBlock(ctx context.Context, csi csiCli
 // unstageVolumeForBlock unstages a block volume from stagingPath
 func (m *csiBlockMapper) unstageVolumeForBlock(ctx context.Context, csi csiClient, stagingPath string) error {
 	// Check whether "STAGE_UNSTAGE_VOLUME" is set
-	stageUnstageSet, err := hasStageUnstageCapability(ctx, csi)
+	stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)
 	if err != nil {
 		klog.Error(log("blockMapper.unstageVolumeForBlock failed to check STAGE_UNSTAGE_VOLUME capability: %v", err))
 		return err

pkg/volume/csi/csi_block_test.go

@@ -33,7 +33,8 @@ import (
 	volumetest "k8s.io/kubernetes/pkg/volume/testing"
 )
 
-func prepareBlockMapperTest(plug *csiPlugin, specVolumeName string) (*csiBlockMapper, *volume.Spec, *api.PersistentVolume, error) {
+func prepareBlockMapperTest(plug *csiPlugin, specVolumeName string, t *testing.T) (*csiBlockMapper, *volume.Spec, *api.PersistentVolume, error) {
+	registerFakePlugin(testDriver, "endpoint", []string{"1.0.0"}, t)
 	pv := makeTestPV(specVolumeName, 10, testDriver, testVol)
 	spec := volume.NewSpecFromPersistentVolume(pv, pv.Spec.PersistentVolumeSource.CSI.ReadOnly)
 	mapper, err := plug.NewBlockVolumeMapper(
@@ -73,7 +74,7 @@ func TestBlockMapperGetGlobalMapPath(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
-		csiMapper, spec, _, err := prepareBlockMapperTest(plug, tc.specVolumeName)
+		csiMapper, spec, _, err := prepareBlockMapperTest(plug, tc.specVolumeName, t)
 		if err != nil {
 			t.Fatalf("Failed to make a new Mapper: %v", err)
 		}
@@ -113,7 +114,7 @@ func TestBlockMapperGetStagingPath(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
-		csiMapper, _, _, err := prepareBlockMapperTest(plug, tc.specVolumeName)
+		csiMapper, _, _, err := prepareBlockMapperTest(plug, tc.specVolumeName, t)
 		if err != nil {
 			t.Fatalf("Failed to make a new Mapper: %v", err)
 		}
@@ -150,7 +151,7 @@ func TestBlockMapperGetPublishPath(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
-		csiMapper, _, _, err := prepareBlockMapperTest(plug, tc.specVolumeName)
+		csiMapper, _, _, err := prepareBlockMapperTest(plug, tc.specVolumeName, t)
 		if err != nil {
 			t.Fatalf("Failed to make a new Mapper: %v", err)
 		}
@@ -187,7 +188,7 @@ func TestBlockMapperGetDeviceMapPath(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
-		csiMapper, _, _, err := prepareBlockMapperTest(plug, tc.specVolumeName)
+		csiMapper, _, _, err := prepareBlockMapperTest(plug, tc.specVolumeName, t)
 		if err != nil {
 			t.Fatalf("Failed to make a new Mapper: %v", err)
 		}
@@ -219,7 +220,7 @@ func TestBlockMapperSetupDevice(t *testing.T) {
 	)
 	plug.host = host
 
-	csiMapper, _, pv, err := prepareBlockMapperTest(plug, "test-pv")
+	csiMapper, _, pv, err := prepareBlockMapperTest(plug, "test-pv", t)
 	if err != nil {
 		t.Fatalf("Failed to make a new Mapper: %v", err)
 	}
@@ -229,7 +230,7 @@ func TestBlockMapperSetupDevice(t *testing.T) {
 
 	csiMapper.csiClient = setupClient(t, true)
 
-	attachID := getAttachmentName(csiMapper.volumeID, csiMapper.driverName, string(nodeName))
+	attachID := getAttachmentName(csiMapper.volumeID, string(csiMapper.driverName), string(nodeName))
 	attachment := makeTestAttachment(attachID, nodeName, pvName)
 	attachment.Status.Attached = true
 	_, err = csiMapper.k8s.StorageV1beta1().VolumeAttachments().Create(attachment)
@@ -286,7 +287,7 @@ func TestBlockMapperMapDevice(t *testing.T) {
 	)
 	plug.host = host
 
-	csiMapper, _, pv, err := prepareBlockMapperTest(plug, "test-pv")
+	csiMapper, _, pv, err := prepareBlockMapperTest(plug, "test-pv", t)
 	if err != nil {
 		t.Fatalf("Failed to make a new Mapper: %v", err)
 	}
@@ -296,7 +297,7 @@ func TestBlockMapperMapDevice(t *testing.T) {
 
 	csiMapper.csiClient = setupClient(t, true)
 
-	attachID := getAttachmentName(csiMapper.volumeID, csiMapper.driverName, string(nodeName))
+	attachID := getAttachmentName(csiMapper.volumeID, string(csiMapper.driverName), string(nodeName))
 	attachment := makeTestAttachment(attachID, nodeName, pvName)
 	attachment.Status.Attached = true
 	_, err = csiMapper.k8s.StorageV1beta1().VolumeAttachments().Create(attachment)
@@ -369,7 +370,7 @@ func TestBlockMapperTearDownDevice(t *testing.T) {
 	)
 	plug.host = host
 
-	_, spec, pv, err := prepareBlockMapperTest(plug, "test-pv")
+	_, spec, pv, err := prepareBlockMapperTest(plug, "test-pv", t)
 	if err != nil {
 		t.Fatalf("Failed to make a new Mapper: %v", err)
 	}

pkg/volume/csi/csi_client_test.go

@@ -23,7 +23,7 @@ import (
 	"reflect"
 	"testing"
 
-	csipb "github.com/container-storage-interface/spec/lib/go/csi"
+	csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
 	api "k8s.io/api/core/v1"
 	"k8s.io/kubernetes/pkg/volume/csi/fake"
 )
@@ -43,10 +43,14 @@ func newFakeCsiDriverClient(t *testing.T, stagingCapable bool) *fakeCsiDriverCli
 func (c *fakeCsiDriverClient) NodeGetInfo(ctx context.Context) (
 	nodeID string,
 	maxVolumePerNode int64,
-	accessibleTopology *csipb.Topology,
+	accessibleTopology map[string]string,
 	err error) {
-	resp, err := c.nodeClient.NodeGetInfo(ctx, &csipb.NodeGetInfoRequest{})
-	return resp.GetNodeId(), resp.GetMaxVolumesPerNode(), resp.GetAccessibleTopology(), err
+	resp, err := c.nodeClient.NodeGetInfo(ctx, &csipbv1.NodeGetInfoRequest{})
+	topology := resp.GetAccessibleTopology()
+	if topology != nil {
+		accessibleTopology = topology.Segments
+	}
+	return resp.GetNodeId(), resp.GetMaxVolumesPerNode(), accessibleTopology, err
 }
 
 func (c *fakeCsiDriverClient) NodePublishVolume(
@@ -56,26 +60,26 @@ func (c *fakeCsiDriverClient) NodePublishVolume(
 	stagingTargetPath string,
 	targetPath string,
 	accessMode api.PersistentVolumeAccessMode,
-	volumeInfo map[string]string,
+	publishContext map[string]string,
 	volumeContext map[string]string,
 	secrets map[string]string,
 	fsType string,
 	mountOptions []string,
 ) error {
 	c.t.Log("calling fake.NodePublishVolume...")
-	req := &csipb.NodePublishVolumeRequest{
+	req := &csipbv1.NodePublishVolumeRequest{
 		VolumeId:       volID,
 		TargetPath:     targetPath,
 		Readonly:       readOnly,
-		PublishContext: volumeInfo,
+		PublishContext: publishContext,
 		VolumeContext:  volumeContext,
 		Secrets:        secrets,
-		VolumeCapability: &csipb.VolumeCapability{
-			AccessMode: &csipb.VolumeCapability_AccessMode{
-				Mode: asCSIAccessMode(accessMode),
+		VolumeCapability: &csipbv1.VolumeCapability{
+			AccessMode: &csipbv1.VolumeCapability_AccessMode{
+				Mode: asCSIAccessModeV1(accessMode),
 			},
-			AccessType: &csipb.VolumeCapability_Mount{
-				Mount: &csipb.VolumeCapability_MountVolume{
+			AccessType: &csipbv1.VolumeCapability_Mount{
+				Mount: &csipbv1.VolumeCapability_MountVolume{
 					FsType:     fsType,
 					MountFlags: mountOptions,
 				},
@@ -89,7 +93,7 @@ func (c *fakeCsiDriverClient) NodePublishVolume(
 
 func (c *fakeCsiDriverClient) NodeUnpublishVolume(ctx context.Context, volID string, targetPath string) error {
 	c.t.Log("calling fake.NodeUnpublishVolume...")
-	req := &csipb.NodeUnpublishVolumeRequest{
+	req := &csipbv1.NodeUnpublishVolumeRequest{
 		VolumeId:   volID,
 		TargetPath: targetPath,
 	}
@@ -108,16 +112,16 @@ func (c *fakeCsiDriverClient) NodeStageVolume(ctx context.Context,
 	volumeContext map[string]string,
 ) error {
 	c.t.Log("calling fake.NodeStageVolume...")
-	req := &csipb.NodeStageVolumeRequest{
+	req := &csipbv1.NodeStageVolumeRequest{
 		VolumeId:          volID,
 		PublishContext:    publishContext,
 		StagingTargetPath: stagingTargetPath,
-		VolumeCapability: &csipb.VolumeCapability{
-			AccessMode: &csipb.VolumeCapability_AccessMode{
-				Mode: asCSIAccessMode(accessMode),
+		VolumeCapability: &csipbv1.VolumeCapability{
+			AccessMode: &csipbv1.VolumeCapability_AccessMode{
+				Mode: asCSIAccessModeV1(accessMode),
 			},
-			AccessType: &csipb.VolumeCapability_Mount{
-				Mount: &csipb.VolumeCapability_MountVolume{
+			AccessType: &csipbv1.VolumeCapability_Mount{
+				Mount: &csipbv1.VolumeCapability_MountVolume{
 					FsType: fsType,
 				},
 			},
@@ -132,7 +136,7 @@ func (c *fakeCsiDriverClient) NodeStageVolume(ctx context.Context,
 
 func (c *fakeCsiDriverClient) NodeUnstageVolume(ctx context.Context, volID, stagingTargetPath string) error {
 	c.t.Log("calling fake.NodeUnstageVolume...")
-	req := &csipb.NodeUnstageVolumeRequest{
+	req := &csipbv1.NodeUnstageVolumeRequest{
 		VolumeId:          volID,
 		StagingTargetPath: stagingTargetPath,
 	}
@@ -140,14 +144,26 @@ func (c *fakeCsiDriverClient) NodeUnstageVolume(ctx context.Context, volID, stag
 	return err
 }
 
-func (c *fakeCsiDriverClient) NodeGetCapabilities(ctx context.Context) ([]*csipb.NodeServiceCapability, error) {
-	c.t.Log("calling fake.NodeGetCapabilities...")
-	req := &csipb.NodeGetCapabilitiesRequest{}
+func (c *fakeCsiDriverClient) NodeSupportsStageUnstage(ctx context.Context) (bool, error) {
+	c.t.Log("calling fake.NodeGetCapabilities for NodeSupportsStageUnstage...")
+	req := &csipbv1.NodeGetCapabilitiesRequest{}
 	resp, err := c.nodeClient.NodeGetCapabilities(ctx, req)
 	if err != nil {
-		return nil, err
+		return false, err
+	}
+
+	capabilities := resp.GetCapabilities()
+
+	stageUnstageSet := false
+	if capabilities == nil {
+		return false, nil
+	}
+	for _, capability := range capabilities {
+		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME {
+			stageUnstageSet = true
+		}
 	}
-	return resp.GetCapabilities(), nil
+	return stageUnstageSet, nil
 }
 
 func setupClient(t *testing.T, stageUnstageSet bool) csiClient {
@@ -173,17 +189,15 @@ func TestClientNodeGetInfo(t *testing.T) {
 		name                       string
 		expectedNodeID             string
 		expectedMaxVolumePerNode   int64
-		expectedAccessibleTopology *csipb.Topology
+		expectedAccessibleTopology map[string]string
 		mustFail                   bool
 		err                        error
 	}{
 		{
-			name:                     "test ok",
-			expectedNodeID:           "node1",
-			expectedMaxVolumePerNode: 16,
-			expectedAccessibleTopology: &csipb.Topology{
-				Segments: map[string]string{"com.example.csi-topology/zone": "zone1"},
-			},
+			name:                       "test ok",
+			expectedNodeID:             "node1",
+			expectedMaxVolumePerNode:   16,
+			expectedAccessibleTopology: map[string]string{"com.example.csi-topology/zone": "zone1"},
 		},
 		{
 			name:     "grpc error",
@@ -198,13 +212,15 @@ func TestClientNodeGetInfo(t *testing.T) {
 		fakeCloser := fake.NewCloser(t)
 		client := &csiDriverClient{
 			driverName: "Fake Driver Name",
-			nodeClientCreator: func(driverName string) (csipb.NodeClient, io.Closer, error) {
+			nodeV1ClientCreator: func(addr csiAddr) (csipbv1.NodeClient, io.Closer, error) {
 				nodeClient := fake.NewNodeClient(false /* stagingCapable */)
 				nodeClient.SetNextError(tc.err)
-				nodeClient.SetNodeGetInfoResp(&csipb.NodeGetInfoResponse{
-					NodeId:             tc.expectedNodeID,
-					MaxVolumesPerNode:  tc.expectedMaxVolumePerNode,
-					AccessibleTopology: tc.expectedAccessibleTopology,
+				nodeClient.SetNodeGetInfoResp(&csipbv1.NodeGetInfoResponse{
+					NodeId:            tc.expectedNodeID,
+					MaxVolumesPerNode: tc.expectedMaxVolumePerNode,
+					AccessibleTopology: &csipbv1.Topology{
+						Segments: tc.expectedAccessibleTopology,
+					},
 				})
 				return nodeClient, fakeCloser, nil
 			},
@@ -222,7 +238,7 @@ func TestClientNodeGetInfo(t *testing.T) {
 		}
 
 		if !reflect.DeepEqual(accessibleTopology, tc.expectedAccessibleTopology) {
-			t.Errorf("expected accessibleTopology: %v; got: %v", *tc.expectedAccessibleTopology, *accessibleTopology)
+			t.Errorf("expected accessibleTopology: %v; got: %v", tc.expectedAccessibleTopology, accessibleTopology)
 		}
 
 		if !tc.mustFail {
@@ -252,7 +268,7 @@ func TestClientNodePublishVolume(t *testing.T) {
 		fakeCloser := fake.NewCloser(t)
 		client := &csiDriverClient{
 			driverName: "Fake Driver Name",
-			nodeClientCreator: func(driverName string) (csipb.NodeClient, io.Closer, error) {
+			nodeV1ClientCreator: func(addr csiAddr) (csipbv1.NodeClient, io.Closer, error) {
 				nodeClient := fake.NewNodeClient(false /* stagingCapable */)
 				nodeClient.SetNextError(tc.err)
 				return nodeClient, fakeCloser, nil
@@ -299,7 +315,7 @@ func TestClientNodeUnpublishVolume(t *testing.T) {
 		fakeCloser := fake.NewCloser(t)
 		client := &csiDriverClient{
 			driverName: "Fake Driver Name",
-			nodeClientCreator: func(driverName string) (csipb.NodeClient, io.Closer, error) {
+			nodeV1ClientCreator: func(addr csiAddr) (csipbv1.NodeClient, io.Closer, error) {
 				nodeClient := fake.NewNodeClient(false /* stagingCapable */)
 				nodeClient.SetNextError(tc.err)
 				return nodeClient, fakeCloser, nil
@@ -337,7 +353,7 @@ func TestClientNodeStageVolume(t *testing.T) {
 		fakeCloser := fake.NewCloser(t)
 		client := &csiDriverClient{
 			driverName: "Fake Driver Name",
-			nodeClientCreator: func(driverName string) (csipb.NodeClient, io.Closer, error) {
+			nodeV1ClientCreator: func(addr csiAddr) (csipbv1.NodeClient, io.Closer, error) {
 				nodeClient := fake.NewNodeClient(false /* stagingCapable */)
 				nodeClient.SetNextError(tc.err)
 				return nodeClient, fakeCloser, nil
@@ -381,7 +397,7 @@ func TestClientNodeUnstageVolume(t *testing.T) {
 		fakeCloser := fake.NewCloser(t)
 		client := &csiDriverClient{
 			driverName: "Fake Driver Name",
-			nodeClientCreator: func(driverName string) (csipb.NodeClient, io.Closer, error) {
+			nodeV1ClientCreator: func(addr csiAddr) (csipbv1.NodeClient, io.Closer, error) {
 				nodeClient := fake.NewNodeClient(false /* stagingCapable */)
 				nodeClient.SetNextError(tc.err)
 				return nodeClient, fakeCloser, nil

pkg/volume/csi/csi_mounter.go

@@ -55,18 +55,18 @@ var (
 )
 
 type csiMountMgr struct {
-	csiClient    csiClient
-	k8s          kubernetes.Interface
-	plugin       *csiPlugin
-	driverName   string
-	volumeID     string
-	specVolumeID string
-	readOnly     bool
-	spec         *volume.Spec
-	pod          *api.Pod
-	podUID       types.UID
-	options      volume.VolumeOptions
-	volumeInfo   map[string]string
+	csiClient      csiClient
+	k8s            kubernetes.Interface
+	plugin         *csiPlugin
+	driverName     csiDriverName
+	volumeID       string
+	specVolumeID   string
+	readOnly       bool
+	spec           *volume.Spec
+	pod            *api.Pod
+	podUID         types.UID
+	options        volume.VolumeOptions
+	publishContext map[string]string
 	volume.MetricsNil
 }
 
@@ -121,7 +121,7 @@ func (c *csiMountMgr) SetUpAt(dir string, fsGroup *int64) error {
 
 	// Check for STAGE_UNSTAGE_VOLUME set and populate deviceMountPath if so
 	deviceMountPath := ""
-	stageUnstageSet, err := hasStageUnstageCapability(ctx, csi)
+	stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)
 	if err != nil {
 		klog.Error(log("mounter.SetUpAt failed to check for STAGE_UNSTAGE_VOLUME capabilty: %v", err))
 		return err
@@ -135,9 +135,9 @@ func (c *csiMountMgr) SetUpAt(dir string, fsGroup *int64) error {
 		}
 	}
 	// search for attachment by VolumeAttachment.Spec.Source.PersistentVolumeName
-	if c.volumeInfo == nil {
+	if c.publishContext == nil {
 		nodeName := string(c.plugin.host.GetNodeName())
-		c.volumeInfo, err = c.plugin.getPublishVolumeInfo(c.k8s, c.volumeID, c.driverName, nodeName)
+		c.publishContext, err = c.plugin.getPublishContext(c.k8s, c.volumeID, string(c.driverName), nodeName)
 		if err != nil {
 			return err
 		}
@@ -191,7 +191,7 @@ func (c *csiMountMgr) SetUpAt(dir string, fsGroup *int64) error {
 		deviceMountPath,
 		dir,
 		accessMode,
-		c.volumeInfo,
+		c.publishContext,
 		attribs,
 		nodePublishSecrets,
 		fsType,
@@ -239,7 +239,7 @@ func (c *csiMountMgr) podAttributes() (map[string]string, error) {
 		return nil, errors.New("CSIDriver lister does not exist")
 	}
 
-	csiDriver, err := c.plugin.csiDriverLister.Get(c.driverName)
+	csiDriver, err := c.plugin.csiDriverLister.Get(string(c.driverName))
 	if err != nil {
 		if apierrs.IsNotFound(err) {
 			klog.V(4).Infof(log("CSIDriver %q not found, not adding pod information", c.driverName))

pkg/volume/csi/csi_mounter_test.go

@@ -75,6 +75,7 @@ func TestMounterGetPath(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
+		registerFakePlugin(testDriver, "endpoint", []string{"1.0.0"}, t)
 		pv := makeTestPV(tc.specVolumeName, 10, testDriver, testVol)
 		spec := volume.NewSpecFromPersistentVolume(pv, pv.Spec.PersistentVolumeSource.CSI.ReadOnly)
 		mounter, err := plug.NewMounter(
@@ -161,6 +162,7 @@ func MounterSetUpTests(t *testing.T, podInfoEnabled bool) {
 				})
 			}
 
+			registerFakePlugin(test.driver, "endpoint", []string{"1.0.0"}, t)
 			pv := makeTestPV("test-pv", 10, test.driver, testVol)
 			pv.Spec.CSI.VolumeAttributes = test.volumeContext
 			pv.Spec.MountOptions = []string{"foo=bar", "baz=qux"}
@@ -187,7 +189,7 @@ func MounterSetUpTests(t *testing.T, podInfoEnabled bool) {
 			csiMounter := mounter.(*csiMountMgr)
 			csiMounter.csiClient = setupClient(t, true)
 
-			attachID := getAttachmentName(csiMounter.volumeID, csiMounter.driverName, string(plug.host.GetNodeName()))
+			attachID := getAttachmentName(csiMounter.volumeID, string(csiMounter.driverName), string(plug.host.GetNodeName()))
 
 			attachment := &storage.VolumeAttachment{
 				ObjectMeta: meta.ObjectMeta{
@@ -331,6 +333,7 @@ func TestMounterSetUpWithFSGroup(t *testing.T) {
 		t.Logf("Running test %s", tc.name)
 
 		volName := fmt.Sprintf("test-vol-%d", i)
+		registerFakePlugin(testDriver, "endpoint", []string{"1.0.0"}, t)
 		pv := makeTestPV("test-pv", 10, testDriver, volName)
 		pv.Spec.AccessModes = tc.accessModes
 		pvName := pv.GetName()
@@ -357,7 +360,7 @@ func TestMounterSetUpWithFSGroup(t *testing.T) {
 		csiMounter := mounter.(*csiMountMgr)
 		csiMounter.csiClient = setupClient(t, true)
 
-		attachID := getAttachmentName(csiMounter.volumeID, csiMounter.driverName, string(plug.host.GetNodeName()))
+		attachID := getAttachmentName(csiMounter.volumeID, string(csiMounter.driverName), string(plug.host.GetNodeName()))
 		attachment := makeTestAttachment(attachID, "test-node", pvName)
 
 		_, err = csiMounter.k8s.StorageV1beta1().VolumeAttachments().Create(attachment)
@@ -392,6 +395,7 @@ func TestMounterSetUpWithFSGroup(t *testing.T) {
 func TestUnmounterTeardown(t *testing.T) {
 	plug, tmpDir := newTestPlugin(t, nil, nil)
 	defer os.RemoveAll(tmpDir)
+	registerFakePlugin(testDriver, "endpoint", []string{"1.0.0"}, t)
 	pv := makeTestPV("test-pv", 10, testDriver, testVol)
 
 	// save the data file prior to unmount

pkg/volume/csi/csiv0/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["csi.pb.go"],
+    importpath = "k8s.io/kubernetes/pkg/volume/csi/csiv0",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//vendor/github.com/golang/protobuf/proto:go_default_library",
+        "//vendor/github.com/golang/protobuf/ptypes/wrappers:go_default_library",
+        "//vendor/golang.org/x/net/context:go_default_library",
+        "//vendor/google.golang.org/grpc:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

pkg/volume/csi/nodeinfomanager/BUILD

@@ -21,7 +21,6 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
         "//staging/src/k8s.io/csi-api/pkg/client/clientset/versioned:go_default_library",
-        "//vendor/github.com/container-storage-interface/spec/lib/go/csi:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
 )
@@ -63,7 +62,6 @@ go_test(
         "//staging/src/k8s.io/client-go/util/testing:go_default_library",
         "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
         "//staging/src/k8s.io/csi-api/pkg/client/clientset/versioned/fake: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",
     ],
 )

pkg/volume/csi/nodeinfomanager/nodeinfomanager.go

@@ -23,7 +23,8 @@ import (
 	"fmt"
 	"strings"
 
-	csipb "github.com/container-storage-interface/spec/lib/go/csi"
+	"time"
+
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -40,7 +41,6 @@ import (
 	nodeutil "k8s.io/kubernetes/pkg/util/node"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/util"
-	"time"
 )
 
 const (
@@ -75,7 +75,7 @@ type Interface interface {
 	// Record in the cluster the given node information from the CSI driver with the given name.
 	// Concurrent calls to InstallCSIDriver() is allowed, but they should not be intertwined with calls
 	// to other methods in this interface.
-	InstallCSIDriver(driverName string, driverNodeID string, maxVolumeLimit int64, topology *csipb.Topology) error
+	InstallCSIDriver(driverName string, driverNodeID string, maxVolumeLimit int64, topology map[string]string) error
 
 	// Remove in the cluster node information from the CSI driver with the given name.
 	// Concurrent calls to UninstallCSIDriver() is allowed, but they should not be intertwined with calls
@@ -97,7 +97,7 @@ func NewNodeInfoManager(
 // CSINodeInfo object. If the CSINodeInfo object doesn't yet exist, it will be created.
 // If multiple calls to InstallCSIDriver() are made in parallel, some calls might receive Node or
 // CSINodeInfo update conflicts, which causes the function to retry the corresponding update.
-func (nim *nodeInfoManager) InstallCSIDriver(driverName string, driverNodeID string, maxAttachLimit int64, topology *csipb.Topology) error {
+func (nim *nodeInfoManager) InstallCSIDriver(driverName string, driverNodeID string, maxAttachLimit int64, topology map[string]string) error {
 	if driverNodeID == "" {
 		return fmt.Errorf("error adding CSI driver node info: driverNodeID must not be empty")
 	}
@@ -133,12 +133,14 @@ func (nim *nodeInfoManager) InstallCSIDriver(driverName string, driverNodeID str
 // If multiple calls to UninstallCSIDriver() are made in parallel, some calls might receive Node or
 // CSINodeInfo update conflicts, which causes the function to retry the corresponding update.
 func (nim *nodeInfoManager) UninstallCSIDriver(driverName string) error {
-	err := nim.uninstallDriverFromCSINodeInfo(driverName)
-	if err != nil {
-		return fmt.Errorf("error uninstalling CSI driver from CSINodeInfo object %v", err)
+	if utilfeature.DefaultFeatureGate.Enabled(features.CSINodeInfo) {
+		err := nim.uninstallDriverFromCSINodeInfo(driverName)
+		if err != nil {
+			return fmt.Errorf("error uninstalling CSI driver from CSINodeInfo object %v", err)
+		}
 	}
 
-	err = nim.updateNode(
+	err := nim.updateNode(
 		removeMaxAttachLimit(driverName),
 		removeNodeIDFromNode(driverName),
 	)
@@ -320,13 +322,13 @@ func removeNodeIDFromNode(csiDriverName string) nodeUpdateFunc {
 
 // updateTopologyLabels returns a function that updates labels of a Node object with the given
 // topology information.
-func updateTopologyLabels(topology *csipb.Topology) nodeUpdateFunc {
+func updateTopologyLabels(topology map[string]string) nodeUpdateFunc {
 	return func(node *v1.Node) (*v1.Node, bool, error) {
-		if topology == nil || len(topology.Segments) == 0 {
+		if topology == nil || len(topology) == 0 {
 			return node, false, nil
 		}
 
-		for k, v := range topology.Segments {
+		for k, v := range topology {
 			if curVal, exists := node.Labels[k]; exists && curVal != v {
 				return nil, false, fmt.Errorf("detected topology value collision: driver reported %q:%q but existing label is %q:%q", k, v, k, curVal)
 			}
@@ -335,7 +337,7 @@ func updateTopologyLabels(topology *csipb.Topology) nodeUpdateFunc {
 		if node.Labels == nil {
 			node.Labels = make(map[string]string)
 		}
-		for k, v := range topology.Segments {
+		for k, v := range topology {
 			node.Labels[k] = v
 		}
 		return node, true, nil
@@ -345,7 +347,7 @@ func updateTopologyLabels(topology *csipb.Topology) nodeUpdateFunc {
 func (nim *nodeInfoManager) updateCSINodeInfo(
 	driverName string,
 	driverNodeID string,
-	topology *csipb.Topology) error {
+	topology map[string]string) error {
 
 	csiKubeClient := nim.volumeHost.GetCSIClient()
 	if csiKubeClient == nil {
@@ -370,7 +372,7 @@ func (nim *nodeInfoManager) tryUpdateCSINodeInfo(
 	csiKubeClient csiclientset.Interface,
 	driverName string,
 	driverNodeID string,
-	topology *csipb.Topology) error {
+	topology map[string]string) error {
 
 	nodeInfo, err := csiKubeClient.CsiV1alpha1().CSINodeInfos().Get(string(nim.nodeName), metav1.GetOptions{})
 	if nodeInfo == nil || errors.IsNotFound(err) {
@@ -427,7 +429,7 @@ func (nim *nodeInfoManager) installDriverToCSINodeInfo(
 	nodeInfo *csiv1alpha1.CSINodeInfo,
 	driverName string,
 	driverNodeID string,
-	topology *csipb.Topology) error {
+	topology map[string]string) error {
 
 	csiKubeClient := nim.volumeHost.GetCSIClient()
 	if csiKubeClient == nil {
@@ -435,10 +437,8 @@ func (nim *nodeInfoManager) installDriverToCSINodeInfo(
 	}
 
 	topologyKeys := make(sets.String)
-	if topology != nil {
-		for k := range topology.Segments {
-			topologyKeys.Insert(k)
-		}
+	for k := range topology {
+		topologyKeys.Insert(k)
 	}
 
 	specModified := true

pkg/volume/csi/nodeinfomanager/nodeinfomanager_test.go

@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/stretchr/testify/assert"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -49,7 +48,7 @@ type testcase struct {
 	existingNode        *v1.Node
 	existingNodeInfo    *csiv1alpha1.CSINodeInfo
 	inputNodeID         string
-	inputTopology       *csi.Topology
+	inputTopology       map[string]string
 	inputVolumeLimit    int64
 	expectedNodeIDMap   map[string]string
 	expectedTopologyMap map[string]sets.String
@@ -71,10 +70,8 @@ func TestInstallCSIDriver(t *testing.T) {
 			driverName:   "com.example.csi/driver1",
 			existingNode: generateNode(nil /* nodeIDs */, nil /* labels */, nil /*capacity*/),
 			inputNodeID:  "com.example.csi/csi-node1",
-			inputTopology: &csi.Topology{
-				Segments: map[string]string{
-					"com.example.csi/zone": "zoneA",
-				},
+			inputTopology: map[string]string{
+				"com.example.csi/zone": "zoneA",
 			},
 			expectedNodeIDMap: map[string]string{
 				"com.example.csi/driver1": "com.example.csi/csi-node1",
@@ -104,10 +101,8 @@ func TestInstallCSIDriver(t *testing.T) {
 				},
 			),
 			inputNodeID: "com.example.csi/csi-node1",
-			inputTopology: &csi.Topology{
-				Segments: map[string]string{
-					"com.example.csi/zone": "zoneA",
-				},
+			inputTopology: map[string]string{
+				"com.example.csi/zone": "zoneA",
 			},
 			expectedNodeIDMap: map[string]string{
 				"com.example.csi/driver1": "com.example.csi/csi-node1",
@@ -134,10 +129,8 @@ func TestInstallCSIDriver(t *testing.T) {
 				nil, /* topologyKeys */
 			),
 			inputNodeID: "com.example.csi/csi-node1",
-			inputTopology: &csi.Topology{
-				Segments: map[string]string{
-					"com.example.csi/zone": "zoneA",
-				},
+			inputTopology: map[string]string{
+				"com.example.csi/zone": "zoneA",
 			},
 			expectedNodeIDMap: map[string]string{
 				"com.example.csi/driver1": "com.example.csi/csi-node1",
@@ -168,10 +161,8 @@ func TestInstallCSIDriver(t *testing.T) {
 				},
 			),
 			inputNodeID: "com.example.csi/csi-node1",
-			inputTopology: &csi.Topology{
-				Segments: map[string]string{
-					"com.example.csi/zone": "zoneA",
-				},
+			inputTopology: map[string]string{
+				"com.example.csi/zone": "zoneA",
 			},
 			expectedNodeIDMap: map[string]string{
 				"com.example.csi/driver1":          "com.example.csi/csi-node1",
@@ -205,10 +196,8 @@ func TestInstallCSIDriver(t *testing.T) {
 				},
 			),
 			inputNodeID: "com.example.csi/csi-node1",
-			inputTopology: &csi.Topology{
-				Segments: map[string]string{
-					"com.example.csi/zone": "other-zone",
-				},
+			inputTopology: map[string]string{
+				"com.example.csi/zone": "other-zone",
 			},
 			expectFail: true,
 		},
@@ -231,10 +220,8 @@ func TestInstallCSIDriver(t *testing.T) {
 				},
 			),
 			inputNodeID: "com.example.csi/other-node",
-			inputTopology: &csi.Topology{
-				Segments: map[string]string{
-					"com.example.csi/rack": "rack1",
-				},
+			inputTopology: map[string]string{
+				"com.example.csi/rack": "rack1",
 			},
 			expectedNodeIDMap: map[string]string{
 				"com.example.csi/driver1": "com.example.csi/other-node",

test/e2e/storage/csi_volumes.go

@@ -47,6 +47,7 @@ var csiTestDrivers = []func() drivers.TestDriver{
 	drivers.InitHostPathCSIDriver,
 	drivers.InitGcePDCSIDriver,
 	drivers.InitGcePDExternalCSIDriver,
+	drivers.InitHostV0PathCSIDriver,
 }
 
 // List of testSuites to be executed in below loop

test/e2e/storage/drivers/csi.go

@@ -134,6 +134,92 @@ func (h *hostpathCSIDriver) CleanupDriver() {
 	}
 }
 
+// hostpathV0CSIDriver
+type hostpathV0CSIDriver struct {
+	cleanup    func()
+	driverInfo DriverInfo
+}
+
+var _ TestDriver = &hostpathV0CSIDriver{}
+var _ DynamicPVTestDriver = &hostpathV0CSIDriver{}
+
+// InitHostPathV0CSIDriver returns hostpathV0CSIDriver that implements TestDriver interface
+func InitHostV0PathCSIDriver() TestDriver {
+	return &hostpathV0CSIDriver{
+		driverInfo: DriverInfo{
+			Name:        "csi-hostpath-v0",
+			FeatureTag:  "",
+			MaxFileSize: testpatterns.FileSizeMedium,
+			SupportedFsType: sets.NewString(
+				"", // Default fsType
+			),
+			IsPersistent:       true,
+			IsFsGroupSupported: false,
+			IsBlockSupported:   false,
+		},
+	}
+}
+
+func (h *hostpathV0CSIDriver) GetDriverInfo() *DriverInfo {
+	return &h.driverInfo
+}
+
+func (h *hostpathV0CSIDriver) SkipUnsupportedTest(pattern testpatterns.TestPattern) {
+}
+
+func (h *hostpathV0CSIDriver) GetDynamicProvisionStorageClass(fsType string) *storagev1.StorageClass {
+	provisioner := GetUniqueDriverName(h)
+	parameters := map[string]string{}
+	ns := h.driverInfo.Framework.Namespace.Name
+	suffix := fmt.Sprintf("%s-sc", provisioner)
+
+	return getStorageClass(provisioner, parameters, nil, ns, suffix)
+}
+
+func (h *hostpathV0CSIDriver) CreateDriver() {
+	By("deploying csi hostpath v0 driver")
+	f := h.driverInfo.Framework
+	cs := f.ClientSet
+
+	// pods should be scheduled on the node
+	nodes := framework.GetReadySchedulableNodesOrDie(cs)
+	node := nodes.Items[rand.Intn(len(nodes.Items))]
+	h.driverInfo.Config.ClientNodeName = node.Name
+	h.driverInfo.Config.ServerNodeName = node.Name
+
+	// TODO (?): the storage.csi.image.version and storage.csi.image.registry
+	// settings are ignored for this test. We could patch the image definitions.
+	o := utils.PatchCSIOptions{
+		OldDriverName:            h.driverInfo.Name,
+		NewDriverName:            GetUniqueDriverName(h),
+		DriverContainerName:      "hostpath",
+		ProvisionerContainerName: "csi-provisioner-v0",
+		NodeName:                 h.driverInfo.Config.ServerNodeName,
+	}
+	cleanup, err := h.driverInfo.Framework.CreateFromManifests(func(item interface{}) error {
+		return utils.PatchCSIDeployment(h.driverInfo.Framework, o, item)
+	},
+		"test/e2e/testing-manifests/storage-csi/driver-registrar/rbac.yaml",
+		"test/e2e/testing-manifests/storage-csi/external-attacher/rbac.yaml",
+		"test/e2e/testing-manifests/storage-csi/external-provisioner/rbac.yaml",
+		"test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/csi-hostpath-attacher.yaml",
+		"test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/csi-hostpath-provisioner.yaml",
+		"test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/csi-hostpathplugin.yaml",
+		"test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/e2e-test-rbac.yaml",
+	)
+	h.cleanup = cleanup
+	if err != nil {
+		framework.Failf("deploying csi hostpath v0 driver: %v", err)
+	}
+}
+
+func (h *hostpathV0CSIDriver) CleanupDriver() {
+	if h.cleanup != nil {
+		By("uninstalling csi hostpath v0 driver")
+		h.cleanup()
+	}
+}
+
 // gce-pd
 type gcePDCSIDriver struct {
 	cleanup    func()

test/e2e/testing-manifests/storage-csi/gce-pd/node_ds.yaml

@@ -14,7 +14,7 @@ spec:
       serviceAccountName: csi-node-sa
       containers:
         - name: csi-driver-registrar
-          image: gcr.io/gke-release/csi-driver-registrar:v1.0.0-gke.0
+          image: gcr.io/gke-release/csi-driver-registrar:v1.0.1-gke.0
           args:
             - "--v=5"
             - "--csi-address=/csi/csi.sock"

test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/csi-hostpath-attacher.yaml

+kind: Service
+apiVersion: v1
+metadata:
+  name: csi-hostpath-attacher
+  labels:
+    app: csi-hostpath-attacher
+spec:
+  selector:
+    app: csi-hostpath-attacher
+  ports:
+    - name: dummy
+      port: 12345
+
+---
+kind: StatefulSet
+apiVersion: apps/v1
+metadata:
+  name: csi-hostpath-attacher
+spec:
+  serviceName: "csi-hostpath-attacher"
+  replicas: 1
+  selector:
+    matchLabels:
+      app: csi-hostpath-attacher
+  template:
+    metadata:
+      labels:
+        app: csi-hostpath-attacher
+    spec:
+      serviceAccountName: csi-attacher
+      containers:
+        - name: csi-attacher
+          image: gcr.io/gke-release/csi-attacher:v0.4.1-gke.0
+          args:
+            - --v=5
+            - --csi-address=$(ADDRESS)
+          env:
+            - name: ADDRESS
+              value: /csi/csi.sock
+          imagePullPolicy: Always
+          volumeMounts:
+          - mountPath: /csi
+            name: socket-dir
+      volumes:
+        - hostPath:
+            path: /var/lib/kubelet/plugins/csi-hostpath-v0
+            type: DirectoryOrCreate
+          name: socket-dir

test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/csi-hostpath-provisioner.yaml

+kind: Service
+apiVersion: v1
+metadata:
+  name: csi-hostpath-provisioner 
+  labels:
+    app: csi-hostpath-provisioner 
+spec:
+  selector:
+    app: csi-hostpath-provisioner 
+  ports:
+    - name: dummy
+      port: 12345
+
+---
+kind: StatefulSet
+apiVersion: apps/v1
+metadata:
+  name: csi-hostpath-provisioner
+spec:
+  serviceName: "csi-hostpath-provisioner"
+  replicas: 1
+  selector:
+    matchLabels:
+      app: csi-hostpath-provisioner
+  template:
+    metadata:
+      labels:
+        app: csi-hostpath-provisioner
+    spec:
+      serviceAccountName: csi-provisioner
+      containers:
+        - name: csi-provisioner-v0
+          image: gcr.io/gke-release/csi-provisioner:v0.4.1-gke.0
+          args:
+            - "--provisioner=csi-hostpath-v0"
+            - "--csi-address=$(ADDRESS)"
+            - "--connection-timeout=15s"
+          env:
+            - name: ADDRESS
+              value: /csi/csi.sock
+          imagePullPolicy: Always
+          volumeMounts:
+            - mountPath: /csi
+              name: socket-dir
+      volumes:
+        - hostPath:
+            path: /var/lib/kubelet/plugins/csi-hostpath-v0
+            type: DirectoryOrCreate
+          name: socket-dir

test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/csi-hostpathplugin.yaml

+kind: DaemonSet
+apiVersion: apps/v1
+metadata:
+  name: csi-hostpathplugin
+spec:
+  selector:
+    matchLabels:
+      app: csi-hostpathplugin
+  template:
+    metadata:
+      labels:
+        app: csi-hostpathplugin
+    spec:
+      serviceAccountName: csi-node-sa
+      hostNetwork: true
+      containers:
+        - name: driver-registrar
+          image: gcr.io/gke-release/csi-driver-registrar:v0.4.1-gke.0
+          args:
+            - --v=5
+            - --csi-address=/csi/csi.sock
+            - --kubelet-registration-path=/var/lib/kubelet/plugins/csi-hostpath-v0/csi.sock
+          env:
+            - name: KUBE_NODE_NAME
+              valueFrom:
+                fieldRef:
+                  apiVersion: v1
+                  fieldPath: spec.nodeName
+          imagePullPolicy: Always
+          volumeMounts:
+          - mountPath: /csi
+            name: socket-dir
+          - mountPath: /registration
+            name: registration-dir
+        - name: hostpath
+          image: quay.io/k8scsi/hostpathplugin:v0.4.1
+          args:
+            - "--v=5"
+            - "--endpoint=$(CSI_ENDPOINT)"
+            - "--nodeid=$(KUBE_NODE_NAME)"
+          env:
+            - name: CSI_ENDPOINT
+              value: unix:///csi/csi.sock
+            - name: KUBE_NODE_NAME
+              valueFrom:
+                fieldRef:
+                  apiVersion: v1
+                  fieldPath: spec.nodeName
+          imagePullPolicy: Always
+          securityContext:
+            privileged: true
+          volumeMounts:
+            - mountPath: /csi
+              name: socket-dir
+            - mountPath: /var/lib/kubelet/pods
+              mountPropagation: Bidirectional
+              name: mountpoint-dir
+      volumes:
+        - hostPath:
+            path: /var/lib/kubelet/plugins/csi-hostpath-v0
+            type: DirectoryOrCreate
+          name: socket-dir
+        - hostPath:
+            path: /var/lib/kubelet/pods
+            type: DirectoryOrCreate
+          name: mountpoint-dir
+        - hostPath:
+            path: /var/lib/kubelet/plugins
+            type: Directory
+          name: registration-dir

test/e2e/testing-manifests/storage-csi/hostpath/hostpath-v0/e2e-test-rbac.yaml

+# priviledged Pod Security Policy, previously defined just for gcePD via PrivilegedTestPSPClusterRoleBinding()
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: psp-csi-hostpath-role
+subjects:
+  - kind: ServiceAccount
+    name: csi-attacher
+    namespace: default
+  - kind: ServiceAccount
+    name: csi-node-sa
+    namespace: default
+  - kind: ServiceAccount
+    name: csi-provisioner
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: e2e-test-privileged-psp
+  apiGroup: rbac.authorization.k8s.io

test/e2e/testing-manifests/storage-csi/hostpath/hostpath/csi-hostpathplugin.yaml

@@ -15,7 +15,7 @@ spec:
       hostNetwork: true
       containers:
         - name: driver-registrar
-          image: gcr.io/gke-release/csi-driver-registrar:v1.0.0-gke.0
+          image: gcr.io/gke-release/csi-driver-registrar:v1.0.1-gke.0
           args:
             - --v=5
             - --csi-address=/csi/csi.sock