Merge pull request #141 from verult/final-top

Topology support

Author: k8s-ci-robot

.gitignore

@@ -41,6 +41,7 @@ flycheck_*.el
 
 .idea/
 /.project
+*.iml
 
 # ignore build directory
 _output/

cmd/csi-provisioner/csi-provisioner.go

@@ -29,6 +29,7 @@ import (
 	ctrl "github.com/kubernetes-csi/external-provisioner/pkg/controller"
 	snapclientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
 	"github.com/kubernetes-incubator/external-storage/lib/controller"
+	csiclientset "k8s.io/csi-api/pkg/client/clientset/versioned"
 
 	"google.golang.org/grpc"
 
@@ -92,6 +93,10 @@ func init() {
 	if err != nil {
 		glog.Fatalf("Failed to create snapshot client: %v", err)
 	}
+	csiAPIClient, err := csiclientset.NewForConfig(config)
+	if err != nil {
+		glog.Fatalf("Failed to create CSI API client: %v", err)
+	}
 
 	// The controller needs to know what the server version is because out-of-tree
 	// provisioners aren't officially supported until 1.5
@@ -118,7 +123,7 @@ func init() {
 	}
 	// Create the provisioner: it implements the Provisioner interface expected by
 	// the controller
-	csiProvisioner := ctrl.NewCSIProvisioner(clientset, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient, snapClient)
+	csiProvisioner := ctrl.NewCSIProvisioner(clientset, csiAPIClient, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient, snapClient)
 	provisionController = controller.NewProvisionController(
 		clientset,
 		*provisioner,

pkg/controller/controller.go

@@ -49,6 +49,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/container-storage-interface/spec/lib/go/csi/v0"
+	csiclientset "k8s.io/csi-api/pkg/client/clientset/versioned"
 )
 
 const (
@@ -81,6 +82,7 @@ const (
 type csiProvisioner struct {
 	client               kubernetes.Interface
 	csiClient            csi.ControllerClient
+	csiAPIClient         csiclientset.Interface
 	grpcClient           *grpc.ClientConn
 	snapshotClient       snapclientset.Interface
 	timeout              time.Duration
@@ -90,6 +92,18 @@ type csiProvisioner struct {
 	config               *rest.Config
 }
 
+type driverState struct {
+	driverName   string
+	capabilities sets.Int
+}
+
+const (
+	PluginCapability_CONTROLLER_SERVICE = iota
+	PluginCapability_ACCESSIBILITY_CONSTRAINTS
+	ControllerCapability_CREATE_DELETE_VOLUME
+	ControllerCapability_CREATE_DELETE_SNAPSHOT
+)
+
 var _ controller.Provisioner = &csiProvisioner{}
 
 var (
@@ -163,61 +177,66 @@ func getDriverName(conn *grpc.ClientConn, timeout time.Duration) (string, error)
 	return name, nil
 }
 
-func supportsPluginControllerService(conn *grpc.ClientConn, timeout time.Duration) (bool, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
-	client := csi.NewIdentityClient(conn)
-	req := csi.GetPluginCapabilitiesRequest{}
+func getDriverCapabilities(conn *grpc.ClientConn, timeout time.Duration) (sets.Int, error) {
+	pluginCaps, err := getPluginCapabilities(conn, timeout)
+	if err != nil {
+		return nil, err
+	}
 
-	rsp, err := client.GetPluginCapabilities(ctx, &req)
+	controllerCaps, err := getControllerCapabilities(conn, timeout)
 	if err != nil {
-		return false, err
+		return nil, err
 	}
-	caps := rsp.GetCapabilities()
-	for _, cap := range caps {
+
+	capabilities := make(sets.Int)
+	for _, cap := range pluginCaps {
 		if cap == nil {
 			continue
 		}
 		service := cap.GetService()
 		if service == nil {
 			continue
 		}
-		if service.GetType() == csi.PluginCapability_Service_CONTROLLER_SERVICE {
-			return true, nil
+		switch service.GetType() {
+		case csi.PluginCapability_Service_CONTROLLER_SERVICE:
+			capabilities.Insert(PluginCapability_CONTROLLER_SERVICE)
+		case csi.PluginCapability_Service_ACCESSIBILITY_CONSTRAINTS:
+			capabilities.Insert(PluginCapability_ACCESSIBILITY_CONSTRAINTS)
 		}
 	}
-	return false, nil
-}
-
-func supportsControllerCreateVolume(conn *grpc.ClientConn, timeout time.Duration) (bool, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
-	client := csi.NewControllerClient(conn)
-	req := csi.ControllerGetCapabilitiesRequest{}
-
-	rsp, err := client.ControllerGetCapabilities(ctx, &req)
-	if err != nil {
-		return false, err
-	}
-	caps := rsp.GetCapabilities()
-	for _, cap := range caps {
+	for _, cap := range controllerCaps {
 		if cap == nil {
 			continue
 		}
 		rpc := cap.GetRpc()
 		if rpc == nil {
 			continue
 		}
-		if rpc.GetType() == csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME {
-			return true, nil
+		switch rpc.GetType() {
+		case csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME:
+			capabilities.Insert(ControllerCapability_CREATE_DELETE_VOLUME)
+		case csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT:
+			capabilities.Insert(ControllerCapability_CREATE_DELETE_SNAPSHOT)
 		}
 	}
-	return false, nil
+	return capabilities, nil
+}
+
+func getPluginCapabilities(conn *grpc.ClientConn, timeout time.Duration) ([]*csi.PluginCapability, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	client := csi.NewIdentityClient(conn)
+	req := csi.GetPluginCapabilitiesRequest{}
+
+	rsp, err := client.GetPluginCapabilities(ctx, &req)
+	if err != nil {
+		return nil, err
+	}
+	return rsp.GetCapabilities(), nil
 }
 
-func supportsControllerCreateSnapshot(conn *grpc.ClientConn, timeout time.Duration) (bool, error) {
+func getControllerCapabilities(conn *grpc.ClientConn, timeout time.Duration) ([]*csi.ControllerServiceCapability, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
@@ -226,26 +245,14 @@ func supportsControllerCreateSnapshot(conn *grpc.ClientConn, timeout time.Durati
 
 	rsp, err := client.ControllerGetCapabilities(ctx, &req)
 	if err != nil {
-		return false, err
-	}
-	caps := rsp.GetCapabilities()
-	for _, cap := range caps {
-		if cap == nil {
-			continue
-		}
-		rpc := cap.GetRpc()
-		if rpc == nil {
-			continue
-		}
-		if rpc.GetType() == csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT {
-			return true, nil
-		}
+		return nil, err
 	}
-	return false, nil
+	return rsp.GetCapabilities(), nil
 }
 
 // NewCSIProvisioner creates new CSI provisioner
 func NewCSIProvisioner(client kubernetes.Interface,
+	csiAPIClient csiclientset.Interface,
 	csiEndpoint string,
 	connectionTimeout time.Duration,
 	identity string,
@@ -259,6 +266,7 @@ func NewCSIProvisioner(client kubernetes.Interface,
 		client:               client,
 		grpcClient:           grpcClient,
 		csiClient:            csiClient,
+		csiAPIClient:         csiAPIClient,
 		snapshotClient:       snapshotClient,
 		timeout:              connectionTimeout,
 		identity:             identity,
@@ -268,27 +276,21 @@ func NewCSIProvisioner(client kubernetes.Interface,
 	return provisioner
 }
 
-// This function get called before any attepmt to communicate with the driver.
+// This function get called before any attempt to communicate with the driver.
 // Before initiating Create/Delete API calls provisioner checks if Capabilities:
 // PluginControllerService,  ControllerCreateVolume sre supported and gets the  driver name.
-func checkDriverState(grpcClient *grpc.ClientConn, timeout time.Duration, needSnapshotSupport bool) (string, error) {
-	ok, err := supportsPluginControllerService(grpcClient, timeout)
+func checkDriverState(grpcClient *grpc.ClientConn, timeout time.Duration, needSnapshotSupport bool) (*driverState, error) {
+	capabilities, err := getDriverCapabilities(grpcClient, timeout)
 	if err != nil {
-		glog.Errorf("failed to get support info :%v", err)
-		return "", err
-	}
-	if !ok {
-		glog.Errorf("no plugin controller service support detected")
-		return "", fmt.Errorf("no plugin controller service support detected")
+		return nil, fmt.Errorf("failed to get capabilities: %v", err)
 	}
-	ok, err = supportsControllerCreateVolume(grpcClient, timeout)
-	if err != nil {
-		glog.Errorf("failed to get support info :%v", err)
-		return "", err
+
+	if !capabilities.Has(PluginCapability_CONTROLLER_SERVICE) {
+		return nil, fmt.Errorf("no plugin controller service support detected")
 	}
-	if !ok {
-		glog.Error("no create/delete volume support detected")
-		return "", fmt.Errorf("no create/delete volume support detected")
+
+	if !capabilities.Has(ControllerCapability_CREATE_DELETE_VOLUME) {
+		return nil, fmt.Errorf("no create/delete volume support detected")
 	}
 
 	// If PVC.Spec.DataSource is not nil, it indicates the request is to create volume
@@ -297,23 +299,19 @@ func checkDriverState(grpcClient *grpc.ClientConn, timeout time.Duration, needSn
 	if needSnapshotSupport {
 		// Check whether plugin supports create snapshot
 		// If not, create volume from snapshot cannot proceed
-		ok, err = supportsControllerCreateSnapshot(grpcClient, timeout)
-		if err != nil {
-			glog.Errorf("failed to get support info :%v", err)
-			return "", err
-		}
-		if !ok {
-			glog.Error("no create/delete snapshot support detected. Cannot create volume from shapshot")
-			return "", fmt.Errorf("no create/delete snapshot support detected. Cannot create volume from shapshot")
+		if !capabilities.Has(ControllerCapability_CREATE_DELETE_SNAPSHOT) {
+			return nil, fmt.Errorf("no create/delete snapshot support detected. Cannot create volume from snapshot")
 		}
 	}
 
 	driverName, err := getDriverName(grpcClient, timeout)
 	if err != nil {
-		glog.Errorf("failed to get driver info :%v", err)
-		return "", err
+		return nil, fmt.Errorf("failed to get driver info: %v", err)
 	}
-	return driverName, nil
+	return &driverState{
+		driverName:   driverName,
+		capabilities: capabilities,
+	}, nil
 }
 
 func makeVolumeName(prefix, pvcUID string, volumeNameUUIDLength int) (string, error) {
@@ -354,7 +352,7 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		}
 		needSnapshotSupport = true
 	}
-	driverName, err := checkDriverState(p.grpcClient, p.timeout, needSnapshotSupport)
+	driverState, err := checkDriverState(p.grpcClient, p.timeout, needSnapshotSupport)
 	if err != nil {
 		return nil, err
 	}
@@ -414,6 +412,19 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		req.VolumeContentSource = volumeContentSource
 	}
 
+	if driverState.capabilities.Has(PluginCapability_ACCESSIBILITY_CONSTRAINTS) {
+		requirements, err := GenerateAccessibilityRequirements(
+			p.client,
+			p.csiAPIClient,
+			driverState.driverName,
+			options.AllowedTopologies,
+			options.SelectedNode)
+		if err != nil {
+			return nil, fmt.Errorf("error generating accessibility requirements: %v", err)
+		}
+		req.AccessibilityRequirements = requirements
+	}
+
 	glog.V(5).Infof("CreateVolumeRequest %+v", req)
 
 	rep := &csi.CreateVolumeResponse{}
@@ -502,6 +513,7 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 	if len(fsType) == 0 {
 		fsType = defaultFSType
 	}
+
 	pv := &v1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: pvName,
@@ -515,7 +527,7 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 			// TODO wait for CSI VolumeSource API
 			PersistentVolumeSource: v1.PersistentVolumeSource{
 				CSI: &v1.CSIPersistentVolumeSource{
-					Driver:                     driverName,
+					Driver:                     driverState.driverName,
 					VolumeHandle:               p.volumeIdToHandle(rep.Volume.Id),
 					FSType:                     fsType,
 					VolumeAttributes:           volumeAttributes,
@@ -527,6 +539,10 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		},
 	}
 
+	if driverState.capabilities.Has(PluginCapability_ACCESSIBILITY_CONSTRAINTS) {
+		pv.Spec.NodeAffinity = GenerateVolumeNodeAffinity(rep.Volume.AccessibleTopology)
+	}
+
 	glog.Infof("successfully created PV %+v", pv.Spec.PersistentVolumeSource)
 
 	return pv, nil