feat(cinder-csi-plugin): add possibility to set custom topology keys (k…

…ubernetes#2035)

* feat(cinder-csi-plugin-node): Additionnal topology keys `--additionnal-topology` are announced on NodeGetInfo to create proper CSINode object
    and could used in storage class allowedTopologies field, in combination with csi-provisioner `--feature-gates Topology=true`
    created PV will have nodeAffinity set with topologies presents in storageClass allowedTopologies and in `--additionnal-topology` argument.

Signed-off-by: MatthieuFin <[email protected]>

cmd/cinder-csi-plugin/main.go

@@ -35,6 +35,7 @@ var (
 	nodeID                   string
 	cloudConfig              []string
 	cloudNames               []string
+	additionalTopologies     map[string]string
 	cluster                  string
 	httpEndpoint             string
 	provideControllerService bool
@@ -67,6 +68,7 @@ func main() {
 	}
 
 	cmd.PersistentFlags().StringSliceVar(&cloudNames, "cloud-name", []string{""}, "CSI driver cloud name in config files. This option can be given multiple times to manage multiple openstack clouds")
+	cmd.PersistentFlags().StringToStringVar(&additionalTopologies, "additionnal-topology", map[string]string{}, "CSI driver topology example topology.kubernetes.io/region=REGION1. This option can be given multiple times to manage multiple topology keys")
 
 	cmd.PersistentFlags().StringVar(&cluster, "cluster", "", "The identifier of the cluster that the plugin is running in.")
 	cmd.PersistentFlags().StringVar(&httpEndpoint, "http-endpoint", "", "The TCP network address where the HTTP server for providing metrics for diagnostics, will listen (example: `:8080`). The default is empty string, which means the server is disabled.")
@@ -106,7 +108,7 @@ func handle() {
 		//Initialize Metadata
 		metadata := metadata.GetMetadataProvider(clouds[cloudNames[0]].GetMetadataOpts().SearchOrder)
 
-		d.SetupNodeService(clouds[cloudNames[0]], mount, metadata)
+		d.SetupNodeService(clouds[cloudNames[0]], mount, metadata, additionalTopologies)
 	}
 
 	d.Run()

pkg/csi/cinder/driver.go

@@ -177,9 +177,9 @@ func (d *Driver) SetupControllerService(clouds map[string]openstack.IOpenStack)
 	d.cs = NewControllerServer(d, clouds)
 }
 
-func (d *Driver) SetupNodeService(cloud openstack.IOpenStack, mount mount.IMount, metadata metadata.IMetadata) {
+func (d *Driver) SetupNodeService(cloud openstack.IOpenStack, mount mount.IMount, metadata metadata.IMetadata, topologies map[string]string) {
 	klog.Info("Providing node service")
-	d.ns = NewNodeServer(d, mount, metadata, cloud)
+	d.ns = NewNodeServer(d, mount, metadata, cloud, topologies)
 }
 
 func (d *Driver) Run() {

pkg/csi/cinder/nodeserver.go

@@ -41,10 +41,11 @@ import (
 )
 
 type nodeServer struct {
-	Driver   *Driver
-	Mount    mount.IMount
-	Metadata metadata.IMetadata
-	Cloud    openstack.IOpenStack
+	Driver     *Driver
+	Mount      mount.IMount
+	Metadata   metadata.IMetadata
+	Cloud      openstack.IOpenStack
+	Topologies map[string]string
 }
 
 func (ns *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
@@ -467,7 +468,12 @@ func (ns *nodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoReque
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "[NodeGetInfo] Unable to retrieve availability zone of node %v", err)
 	}
-	topology := &csi.Topology{Segments: map[string]string{topologyKey: zone}}
+	topologyMap := make(map[string]string, len(ns.Topologies)+1)
+	topologyMap[topologyKey] = zone
+	for k, v := range ns.Topologies {
+		topologyMap[k] = v
+	}
+	topology := &csi.Topology{Segments: topologyMap}
 
 	maxVolume := ns.Cloud.GetMaxVolLimit()
 

pkg/csi/cinder/nodeserver_test.go

@@ -52,7 +52,7 @@ func init() {
 			"": omock,
 		}
 
-		fakeNs = NewNodeServer(d, mount.MInstance, metadata.MetadataService, openstack.OsInstances[""])
+		fakeNs = NewNodeServer(d, mount.MInstance, metadata.MetadataService, openstack.OsInstances[""], map[string]string{})
 	}
 }
 
@@ -147,7 +147,7 @@ func TestNodePublishVolumeEphermeral(t *testing.T) {
 	}
 
 	d := NewDriver(&DriverOpts{Endpoint: FakeEndpoint, ClusterID: FakeCluster})
-	fakeNse := NewNodeServer(d, mount.MInstance, metadata.MetadataService, openstack.OsInstances[""])
+	fakeNse := NewNodeServer(d, mount.MInstance, metadata.MetadataService, openstack.OsInstances[""], map[string]string{})
 
 	// Init assert
 	assert := assert.New(t)
@@ -299,7 +299,7 @@ func TestNodeUnpublishVolumeEphermeral(t *testing.T) {
 	omock.On("DeleteVolume", FakeVolID).Return(nil)
 
 	d := NewDriver(&DriverOpts{Endpoint: FakeEndpoint, ClusterID: FakeCluster})
-	fakeNse := NewNodeServer(d, mount.MInstance, metadata.MetadataService, osmock[""])
+	fakeNse := NewNodeServer(d, mount.MInstance, metadata.MetadataService, osmock[""], map[string]string{})
 
 	// Init assert
 	assert := assert.New(t)

pkg/csi/cinder/utils.go

@@ -57,12 +57,13 @@ func NewIdentityServer(d *Driver) *identityServer {
 	}
 }
 
-func NewNodeServer(d *Driver, mount mount.IMount, metadata metadata.IMetadata, cloud openstack.IOpenStack) *nodeServer {
+func NewNodeServer(d *Driver, mount mount.IMount, metadata metadata.IMetadata, cloud openstack.IOpenStack, topologies map[string]string) *nodeServer {
 	return &nodeServer{
-		Driver:   d,
-		Mount:    mount,
-		Metadata: metadata,
-		Cloud:    cloud,
+		Driver:     d,
+		Mount:      mount,
+		Metadata:   metadata,
+		Cloud:      cloud,
+		Topologies: topologies,
 	}
 }
 

tests/sanity/cinder/sanity_test.go

@@ -30,7 +30,7 @@ func TestDriver(t *testing.T) {
 	fakemet := &fakemetadata{}
 
 	d.SetupControllerService(openstack.OsInstances)
-	d.SetupNodeService(fakecloudprovider, fakemnt, fakemet)
+	d.SetupNodeService(fakecloudprovider, fakemnt, fakemet, map[string]string{})
 
 	// TODO: Stop call