enable topology aware provisioning for controller

src/csi/driver/controller.go

@@ -35,27 +35,15 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		parameters["backend"], parameters["cloneFrom"] = utils.SplitVolumeId(cloneFrom)
 	}
 
-	contentSource := req.GetVolumeContentSource()
-	if contentSource != nil {
-		if contentSnapshot := contentSource.GetSnapshot(); contentSnapshot != nil {
-			sourceSnapshotId := contentSnapshot.GetSnapshotId()
-			sourceBackendName, snapshotParentId, sourceSnapshotName := utils.SplitSnapshotId(sourceSnapshotId)
-			parameters["sourceSnapshotName"] = sourceSnapshotName
-			parameters["snapshotParentId"] = snapshotParentId
-			parameters["backend"] = sourceBackendName
-			log.Infof("Start to create volume from snapshot %s", sourceSnapshotName)
-		} else if contentVolume := contentSource.GetVolume(); contentVolume != nil {
-			sourceVolumeId := contentVolume.GetVolumeId()
-			sourceBackendName, sourceVolumeName := utils.SplitVolumeId(sourceVolumeId)
-			parameters["sourceVolumeName"] = sourceVolumeName
-			parameters["backend"] = sourceBackendName
-			log.Infof("Start to create volume from volume %s", sourceVolumeName)
-		} else {
-			log.Errorf("The source %s is not snapshot either volume", contentSource)
-			return nil, status.Error(codes.InvalidArgument, "no source ID provided is invalid")
-		}
+	// process volume content source
+	err := d.processVolumeContentSource(req, parameters)
+	if err != nil {
+		return nil, err
 	}
 
+	// process accessibility requirements
+	d.processAccessibilityRequirements(req, parameters)
+
 	localPool, remotePool, err := backend.SelectStoragePool(size, parameters)
 	if err != nil {
 		log.Errorf("Cannot select pool for volume creation: %v", err)
@@ -77,30 +65,105 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 
 	log.Infof("Volume %s is created", name)
 
+	volume, err := d.getCreatedVolume(req, volName, localPool)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	return &csi.CreateVolumeResponse{
+		Volume: volume,
+	}, nil
+}
+
+func (d *Driver) getCreatedVolume(req *csi.CreateVolumeRequest, volName string, pool *backend.StoragePool) (*csi.Volume, error) {
+	contentSource := req.GetVolumeContentSource()
+	size := req.GetCapacityRange().GetRequiredBytes()
+
+	accessibleTopologies := make([]*csi.Topology, 0)
+	supportedTopology := pool.GetSupportedTopologies()
+	if len(supportedTopology) > 0 {
+		for _, segment := range supportedTopology {
+			accessibleTopologies = append(accessibleTopologies, &csi.Topology{Segments: segment})
+		}
+	}
+
 	if contentSource != nil {
 		attributes := map[string]string{
-			"backend": localPool.Parent,
+			"backend": pool.Parent,
 			"name":    volName,
 		}
 
-		return &csi.CreateVolumeResponse{
-			Volume: &csi.Volume{
-				VolumeId:      localPool.Parent + "." + volName,
-				CapacityBytes: size,
-				VolumeContext: attributes,
-				ContentSource: req.VolumeContentSource,
-			},
+		return &csi.Volume{
+			VolumeId:           pool.Parent + "." + volName,
+			CapacityBytes:      size,
+			VolumeContext:      attributes,
+			ContentSource:      contentSource,
+			AccessibleTopology: accessibleTopologies,
 		}, nil
 	}
 
-	return &csi.CreateVolumeResponse{
-		Volume: &csi.Volume{
-			VolumeId:      localPool.Parent + "." + volName,
-			CapacityBytes: size,
-		},
+	return &csi.Volume{
+		VolumeId:           pool.Parent + "." + volName,
+		CapacityBytes:      size,
+		AccessibleTopology: accessibleTopologies,
 	}, nil
 }
 
+func (d *Driver) processVolumeContentSource(req *csi.CreateVolumeRequest, parameters map[string]interface{}) error {
+	contentSource := req.GetVolumeContentSource()
+	if contentSource != nil {
+		if contentSnapshot := contentSource.GetSnapshot(); contentSnapshot != nil {
+			sourceSnapshotId := contentSnapshot.GetSnapshotId()
+			sourceBackendName, snapshotParentId, sourceSnapshotName := utils.SplitSnapshotId(sourceSnapshotId)
+			parameters["sourceSnapshotName"] = sourceSnapshotName
+			parameters["snapshotParentId"] = snapshotParentId
+			parameters["backend"] = sourceBackendName
+			log.Infof("Start to create volume from snapshot %s", sourceSnapshotName)
+		} else if contentVolume := contentSource.GetVolume(); contentVolume != nil {
+			sourceVolumeId := contentVolume.GetVolumeId()
+			sourceBackendName, sourceVolumeName := utils.SplitVolumeId(sourceVolumeId)
+			parameters["sourceVolumeName"] = sourceVolumeName
+			parameters["backend"] = sourceBackendName
+			log.Infof("Start to create volume from volume %s", sourceVolumeName)
+		} else {
+			log.Errorf("The source %s is not snapshot either volume", contentSource)
+			return status.Error(codes.InvalidArgument, "no source ID provided is invalid")
+		}
+	}
+
+	return nil
+}
+
+func (d *Driver) processAccessibilityRequirements(req *csi.CreateVolumeRequest, parameters map[string]interface{}) {
+	accessibleTopology := req.GetAccessibilityRequirements()
+	if accessibleTopology == nil {
+		return
+	}
+
+	var requisiteTopologies = make([]map[string]string, 0)
+	for _, requisite := range accessibleTopology.GetRequisite() {
+		requirement := make(map[string]string)
+		for k, v := range requisite.GetSegments() {
+			requirement[k] = v
+		}
+		requisiteTopologies = append(requisiteTopologies, requirement)
+	}
+
+	var preferredTopologies = make([]map[string]string, 0)
+	for _, preferred := range accessibleTopology.GetPreferred() {
+		preference := make(map[string]string)
+		for k, v := range preferred.GetSegments() {
+			preference[k] = v
+		}
+		preferredTopologies = append(preferredTopologies, preference)
+	}
+
+	parameters[backend.TopologyRequirement] = backend.AccessibleTopology{
+		RequisiteTopologies: requisiteTopologies,
+		PreferredTopologies: preferredTopologies,
+	}
+}
+
 func (d *Driver) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
 	volumeId := req.GetVolumeId()
 

src/csi/driver/driver.go

@@ -1,17 +1,27 @@
 package driver
 
+import (
+	"strings"
+	"utils/k8sutils"
+)
+
 type Driver struct {
 	name            string
 	version         string
 	useMultiPath    bool
 	isNeedMultiPath bool
+	k8sUtils        k8sutils.Interface
+	nodeName        string
 }
 
-func NewDriver(name, version string, useMultiPath, isNeedMultiPath bool) *Driver {
+func NewDriver(name, version string, useMultiPath, isNeedMultiPath bool,
+	k8sUtils k8sutils.Interface, nodeName string) *Driver {
 	return &Driver{
 		name:            name,
 		version:         version,
 		useMultiPath:    useMultiPath,
 		isNeedMultiPath: isNeedMultiPath,
+		k8sUtils:        k8sUtils,
+		nodeName:        strings.TrimSpace(nodeName),
 	}
 }

src/csi/driver/identity.go

@@ -27,6 +27,13 @@ func (d *Driver) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCa
 					},
 				},
 			},
+			&csi.PluginCapability{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
+					},
+				},
+			},
 		},
 	}, nil
 }

src/csi/driver/node.go

@@ -171,10 +171,26 @@ func (d *Driver) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (
 		log.Errorf("Marshal node info of %s error: %v", nodeBytes, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
-
 	log.Infof("Get NodeId %s", nodeBytes)
+
+	if d.nodeName == "" {
+		return &csi.NodeGetInfoResponse{
+			NodeId: string(nodeBytes),
+		}, nil
+	}
+
+	// Get topology info from Node labels
+	topology, err := d.k8sUtils.GetNodeTopology(d.nodeName)
+	if err != nil {
+		log.Errorln(err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	return &csi.NodeGetInfoResponse{
 		NodeId: string(nodeBytes),
+		AccessibleTopology: &csi.Topology{
+			Segments: topology,
+		},
 	}, nil
 }
 

src/csi/main.go

@@ -14,6 +14,7 @@ import (
 	"runtime/debug"
 	"time"
 	"utils"
+	"utils/k8sutils"
 	"utils/log"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -30,6 +31,8 @@ const (
 
 	csiVersion        = "2.2.13"
 	defaultDriverName = "csi.huawei.com"
+
+	nodeNameEnv = "CSI_NODENAME"
 )
 
 var (
@@ -54,6 +57,12 @@ var (
 	volumeUseMultiPath = flag.Bool("volume-use-multipath",
 		true,
 		"Whether to use multipath when attach block volume")
+	kubeconfig = flag.String("kubeconfig",
+		"",
+		"absolute path to the kubeconfig file")
+	nodeName = flag.String("nodename",
+		os.Getenv(nodeNameEnv),
+		"node name in kubernetes cluster")
 
 	config CSIConfig
 	secret CSISecret
@@ -64,7 +73,7 @@ type CSIConfig struct {
 }
 
 type CSISecret struct {
-	Secrets map[string]interface{}  `json:"secrets"`
+	Secrets map[string]interface{} `json:"secrets"`
 }
 
 func init() {
@@ -97,6 +106,10 @@ func init() {
 
 	_ = mergeData(config, secret)
 
+	if "" == *nodeName {
+		logrus.Warning("Node name is empty. Topology aware volume provisioning feature may not behave normal")
+	}
+
 	if *containerized {
 		*controllerFlagFile = ""
 	}
@@ -169,14 +182,14 @@ func main() {
 	}()
 
 	if *controller || *controllerFlagFile != "" {
-		err := backend.RegisterBackend(config.Backends, true)
+		err := backend.RegisterBackend(config.Backends, true, *driverName)
 		if err != nil {
 			log.Fatalf("Register backends error: %v", err)
 		}
 
 		go updateBackendCapabilities()
 	} else {
-		err := backend.RegisterBackend(config.Backends, false)
+		err := backend.RegisterBackend(config.Backends, false, *driverName)
 		if err != nil {
 			log.Fatalf("Register backends error: %v", err)
 		}
@@ -199,10 +212,15 @@ func main() {
 		log.Fatalf("Listen on %s error: %v", *endpoint, err)
 	}
 
+	k8sUtils, err := k8sutils.NewK8SUtils(*kubeconfig)
+	if err != nil {
+		log.Fatalf("Kubernetes client initialization failed  %v", err)
+	}
+
 	isNeedMultiPath := utils.NeedMultiPath(config.Backends)
-	d := driver.NewDriver(*driverName, csiVersion, *volumeUseMultiPath, isNeedMultiPath)
-	server := grpc.NewServer()
+	d := driver.NewDriver(*driverName, csiVersion, *volumeUseMultiPath, isNeedMultiPath, k8sUtils, *nodeName)
 
+	server := grpc.NewServer()
 	csi.RegisterIdentityServer(server, d)
 	csi.RegisterControllerServer(server, d)
 	csi.RegisterNodeServer(server, d)