Merge branch 'master' into testgetdrivename

Author: lpabon

.travis.yml

@@ -14,7 +14,7 @@ script:
 - go test $(go list ./... | grep -v vendor)
 after_success:
   - if [ "${TRAVIS_BRANCH}" == "master" ] && [ "${TRAVIS_PULL_REQUEST}" == "false" ]; then
-      sudo ./hack/build-container.sh;
+      make container;
       docker login -u "${DOCKER_USERNAME}" -p "${DOCKER_PASSWORD}" quay.io;
       docker push quay.io/k8scsi/csi-provisioner:canary;
     fi

Makefile

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 IMAGE_NAME = quay.io/k8scsi/csi-provisioner
-IMAGE_VERSION = v0.2.0
+IMAGE_VERSION = canary
 
 ifdef V
 TESTARGS = -v -args -alsologtostderr -v 5
@@ -25,7 +25,7 @@ all: csi-provisioner
 
 csi-provisioner:
 	mkdir -p bin
-	CGO_ENABLED=0 GOOS=linux go build -a -ldflags '-extldflags "-static"' -o ./bin/csi-provisioner ./cmd/csi-provisioner 
+	CGO_ENABLED=0 GOOS=linux go build -a -ldflags '-extldflags "-static"' -o ./bin/csi-provisioner ./cmd/csi-provisioner
 
 clean:
 	rm -rf bin deploy/docker/csi-provisioner

README.md

@@ -1,3 +1,4 @@
+[![Build Status](https://travis-ci.org/kubernetes-csi/external-provisioner.svg?branch=master)](https://travis-ci.org/kubernetes-csi/external-provisioner)
 # Kubernetes external provisioner that works with CSI volumes.
 
 This is an example external provisioner for Kubernetes which provisions using CSI Volume drivers..  It's under heavy development, so at this time README.md is notes for the developers coding.  Once complete this will change to something user friendly.

cmd/csi-provisioner/csi-provisioner.go

@@ -28,6 +28,8 @@ import (
 	ctrl "github.com/kubernetes-csi/external-provisioner/pkg/controller"
 	"github.com/kubernetes-incubator/external-storage/lib/controller"
 
+	"google.golang.org/grpc"
+
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -38,7 +40,7 @@ var (
 	provisioner          = flag.String("provisioner", "", "Name of the provisioner. The provisioner will only provision volumes for claims that request a StorageClass with a provisioner field set equal to this name.")
 	master               = flag.String("master", "", "Master URL to build a client config from. Either this or kubeconfig needs to be set if the provisioner is being run out of cluster.")
 	kubeconfig           = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Either this or master needs to be set if the provisioner is being run out of cluster.")
-	csiEndpoint          = flag.String("csi-address", "", "The gRPC endpoint for Target CSI Volume")
+	csiEndpoint          = flag.String("csi-address", "/run/csi/socket", "The gRPC endpoint for Target CSI Volume")
 	connectionTimeout    = flag.Duration("connection-timeout", 10*time.Second, "Timeout for waiting for CSI driver socket.")
 	volumeNamePrefix     = flag.String("volume-name-prefix", "kubernetes-dynamic-pv", "Prefix to apply to the name of a created volume")
 	volumeNameUUIDLength = flag.Int("volume-name-uuid-length", 16, "Length in characters for the generated uuid of a created volume")
@@ -87,9 +89,21 @@ func init() {
 	timeStamp := time.Now().UnixNano() / int64(time.Millisecond)
 	identity := strconv.FormatInt(timeStamp, 10) + "-" + strconv.Itoa(rand.Intn(10000)) + "-" + *provisioner
 
+	// Provisioner will stay in Init until driver opens csi socket, once it's done
+	// controller will exit this loop and proceed normally.
+	socketDown := true
+	grpcClient := &grpc.ClientConn{}
+	for socketDown {
+		grpcClient, err = ctrl.Connect(*csiEndpoint, *connectionTimeout)
+		if err == nil {
+			socketDown = false
+			continue
+		}
+		time.Sleep(10 * time.Second)
+	}
 	// Create the provisioner: it implements the Provisioner interface expected by
 	// the controller
-	csiProvisioner := ctrl.NewCSIProvisioner(clientset, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength)
+	csiProvisioner := ctrl.NewCSIProvisioner(clientset, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient)
 	provisionController = controller.NewProvisionController(
 		clientset,
 		*provisioner,

hack/build-container.sh

@@ -44,10 +44,11 @@ const (
 	secretNamespaceKey = "csiProvisionerSecretNamespace"
 )
 
+// CSIProvisioner struct
 type csiProvisioner struct {
 	client               kubernetes.Interface
 	csiClient            csi.ControllerClient
-	driverName           string
+	grpcClient           *grpc.ClientConn
 	timeout              time.Duration
 	identity             string
 	volumeNamePrefix     string
@@ -80,7 +81,7 @@ func logGRPC(ctx context.Context, method string, req, reply interface{}, cc *grp
 	return err
 }
 
-func connect(address string, timeout time.Duration) (*grpc.ClientConn, error) {
+func Connect(address string, timeout time.Duration) (*grpc.ClientConn, error) {
 	glog.V(2).Infof("Connecting to %s", address)
 	dialOptions := []grpc.DialOption{
 		grpc.WithInsecure(),
@@ -102,7 +103,7 @@ func connect(address string, timeout time.Duration) (*grpc.ClientConn, error) {
 	for {
 		if !conn.WaitForStateChange(ctx, conn.GetState()) {
 			glog.V(4).Infof("Connection timed out")
-			return conn, nil // return nil, subsequent GetPluginInfo will show the real connection error
+			return conn, fmt.Errorf("Connection timed out")
 		}
 		if conn.GetState() == connectivity.Ready {
 			glog.V(3).Infof("Connected")
@@ -185,34 +186,19 @@ func supportsControllerCreateVolume(conn *grpc.ClientConn, timeout time.Duration
 	return false, nil
 }
 
-func NewCSIProvisioner(client kubernetes.Interface, csiEndpoint string, connectionTimeout time.Duration, identity string, volumeNamePrefix string, volumeNameUUIDLength int) controller.Provisioner {
-	grpcClient, err := connect(csiEndpoint, connectionTimeout)
-	if err != nil || grpcClient == nil {
-		glog.Fatalf("failed to connect to csi endpoint :%v", err)
-	}
-	ok, err := supportsPluginControllerService(grpcClient, connectionTimeout)
-	if err != nil {
-		glog.Fatalf("failed to get support info :%v", err)
-	}
-	if !ok {
-		glog.Fatalf("no plugin controller service support detected")
-	}
-	ok, err = supportsControllerCreateVolume(grpcClient, connectionTimeout)
-	if err != nil {
-		glog.Fatalf("failed to get support info :%v", err)
-	}
-	if !ok {
-		glog.Fatalf("no create/delete volume support detected")
-	}
-	driver, err := getDriverName(grpcClient, connectionTimeout)
-	if err != nil {
-		glog.Fatalf("failed to get driver info :%v", err)
-	}
+// NewCSIProvisioner creates new CSI provisioner
+func NewCSIProvisioner(client kubernetes.Interface,
+	csiEndpoint string,
+	connectionTimeout time.Duration,
+	identity string,
+	volumeNamePrefix string,
+	volumeNameUUIDLength int,
+	grpcClient *grpc.ClientConn) controller.Provisioner {
 
 	csiClient := csi.NewControllerClient(grpcClient)
 	provisioner := &csiProvisioner{
 		client:               client,
-		driverName:           driver,
+		grpcClient:           grpcClient,
 		csiClient:            csiClient,
 		timeout:              connectionTimeout,
 		identity:             identity,
@@ -222,10 +208,45 @@ func NewCSIProvisioner(client kubernetes.Interface, csiEndpoint string, connecti
 	return provisioner
 }
 
+// This function get called before any attepmt 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) (string, error) {
+	ok, err := supportsPluginControllerService(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")
+	}
+	ok, err = supportsControllerCreateVolume(grpcClient, timeout)
+	if err != nil {
+		glog.Errorf("failed to get support info :%v", err)
+		return "", err
+	}
+	if !ok {
+		glog.Error("no create/delete volume support detected")
+		return "", fmt.Errorf("no create/delete volume support detected")
+	}
+	driverName, err := getDriverName(grpcClient, timeout)
+	if err != nil {
+		glog.Errorf("failed to get driver info :%v", err)
+		return "", err
+	}
+	return driverName, nil
+}
+
 func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.PersistentVolume, error) {
 	if options.PVC.Spec.Selector != nil {
 		return nil, fmt.Errorf("claim Selector is not supported")
 	}
+
+	driverName, err := checkDriverState(p.grpcClient, p.timeout)
+	if err != nil {
+		return nil, err
+	}
 	// create random share name
 	share := fmt.Sprintf("%s-%s", p.volumeNamePrefix, strings.Replace(string(uuid.NewUUID()), "-", "", -1)[0:p.volumeNameUUIDLength])
 	capacity := options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
@@ -249,7 +270,11 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 	}
 	secret := v1.SecretReference{}
 	if options.Parameters != nil {
-		req.ControllerCreateSecrets = getCredentialsFromParameters(p.client, options.Parameters)
+		credentials, err := getCredentialsFromParameters(p.client, options.Parameters)
+		if err != nil {
+			return nil, err
+		}
+		req.ControllerCreateSecrets = credentials
 		secret.Name, secret.Namespace, _ = getSecretAndNamespaceFromParameters(options.Parameters)
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
@@ -279,7 +304,7 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 			// TODO wait for CSI VolumeSource API
 			PersistentVolumeSource: v1.PersistentVolumeSource{
 				CSI: &v1.CSIPersistentVolumeSource{
-					Driver:           p.driverName,
+					Driver:           driverName,
 					VolumeHandle:     p.volumeIdToHandle(rep.Volume.Id),
 					VolumeAttributes: volumeAttributes,
 				},
@@ -300,21 +325,30 @@ func (p *csiProvisioner) Delete(volume *v1.PersistentVolume) error {
 	}
 	volumeId := p.volumeHandleToId(volume.Spec.CSI.VolumeHandle)
 
+	_, err := checkDriverState(p.grpcClient, p.timeout)
+	if err != nil {
+		return err
+	}
+
 	req := csi.DeleteVolumeRequest{
 		VolumeId: volumeId,
 	}
 	// get secrets if StorageClass specifies it
 	storageClassName := volume.Spec.StorageClassName
 	if len(storageClassName) != 0 {
 		if storageClass, err := p.client.StorageV1().StorageClasses().Get(storageClassName, metav1.GetOptions{}); err == nil {
-			req.ControllerDeleteSecrets = getCredentialsFromParameters(p.client, storageClass.Parameters)
+			credentials, err := getCredentialsFromParameters(p.client, storageClass.Parameters)
+			if err != nil {
+				return err
+			}
+			req.ControllerDeleteSecrets = credentials
 		}
 
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
 	defer cancel()
 
-	_, err := p.csiClient.DeleteVolume(ctx, &req)
+	_, err = p.csiClient.DeleteVolume(ctx, &req)
 
 	return err
 }
@@ -330,11 +364,11 @@ func getSecretAndNamespaceFromParameters(parameters map[string]string) (string,
 	return "", "", false
 }
 
-func getCredentialsFromParameters(k8s kubernetes.Interface, parameters map[string]string) map[string]string {
+func getCredentialsFromParameters(k8s kubernetes.Interface, parameters map[string]string) (map[string]string, error) {
 	if secretName, namespace, found := getSecretAndNamespaceFromParameters(parameters); found {
 		return getCredentialsFromSecret(k8s, secretName, namespace)
 	}
-	return map[string]string{}
+	return map[string]string{}, nil
 }
 
 //TODO use a unique volume handle from and to Id
@@ -346,19 +380,18 @@ func (p *csiProvisioner) volumeHandleToId(handle string) string {
 	return handle
 }
 
-func getCredentialsFromSecret(k8s kubernetes.Interface, secretName, nameSpace string) map[string]string {
+func getCredentialsFromSecret(k8s kubernetes.Interface, secretName, nameSpace string) (map[string]string, error) {
 	credentials := map[string]string{}
 	if len(secretName) == 0 {
-		return credentials
+		return credentials, nil
 	}
 	secret, err := k8s.CoreV1().Secrets(nameSpace).Get(secretName, metav1.GetOptions{})
 	if err != nil {
-		glog.Warningf("failed to find the secret %s in the namespace %s with error: %v\n", secretName, nameSpace, err)
-		return credentials
+		return credentials, fmt.Errorf("failed to find the secret %s in the namespace %s with error: %v\n", secretName, nameSpace, err)
 	}
 	for key, value := range secret.Data {
 		credentials[key] = string(value)
 	}
 
-	return credentials
+	return credentials, nil
 }

main.go

@@ -36,7 +36,7 @@ type csiConnection struct {
 }
 
 func New(address string, timeout time.Duration) (csiConnection, error) {
-	conn, err := connect(address, timeout)
+	conn, err := Connect(address, timeout)
 	if err != nil {
 		return csiConnection{}, err
 	}
@@ -70,6 +70,64 @@ func createMockServer(t *testing.T) (*gomock.Controller,
 	return mockController, drv, identityServer, controllerServer, csiConn, nil
 }
 
+func TestGetPluginName(t *testing.T) {
+	test := struct {
+		name   string
+		output []*csi.GetPluginInfoResponse
+	}{
+		name: "success",
+		output: []*csi.GetPluginInfoResponse{
+			{
+				Name:          "csi/example-1",
+				VendorVersion: "0.2.0",
+				Manifest: map[string]string{
+					"hello": "world",
+				},
+			},
+			{
+				Name:          "csi/example-2",
+				VendorVersion: "0.2.0",
+				Manifest: map[string]string{
+					"hello": "world",
+				},
+			},
+		},
+	}
+
+	mockController, driver, identityServer, _, csiConn, err := createMockServer(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer mockController.Finish()
+	defer driver.Stop()
+
+	in := &csi.GetPluginInfoRequest{}
+	out := test.output[0]
+
+	identityServer.EXPECT().GetPluginInfo(gomock.Any(), in).Return(out, nil).Times(1)
+	oldName, err := getDriverName(csiConn.conn, timeout)
+	if err != nil {
+		t.Errorf("test %q: Failed to get driver's name", test.name)
+	}
+	if oldName != test.output[0].Name {
+		t.Errorf("test %s: failed, expected %s got %s", test.name, test.output[0].Name, oldName)
+	}
+
+	out = test.output[1]
+	identityServer.EXPECT().GetPluginInfo(gomock.Any(), in).Return(out, nil).Times(1)
+	newName, err := getDriverName(csiConn.conn, timeout)
+	if err != nil {
+		t.Errorf("test %s: Failed to get driver's name", test.name)
+	}
+	if newName != test.output[1].Name {
+		t.Errorf("test %q: failed, expected %s got %s", test.name, test.output[1].Name, newName)
+	}
+
+	if oldName == newName {
+		t.Errorf("test: %s failed, driver's names should not match", test.name)
+	}
+}
+
 func TestSupportsControllerCreateVolume(t *testing.T) {
 
 	tests := []struct {