[WIP] Add Block volume support for CSI provisioner

In CSI provisioner, below three logics need to be implemented,
to add Block volume support to CSI provisioner:

  1. Add SupportsBlock that properly returns whether Storage
     Provider's plugin supports block (this is checked
     by using ValidateVolumeCapabilities),
  2. Pass BlockVolume instead of MountVolume to CreateVolume
     if volumeMode is set to be Block on Provision,
  3. Set volumeMode to PV returned by Provision.

vendor/github.com/kubernetes-incubator/external-storage/*
should be updated from upstream, but manually changed for
testing purpose, in this commit.

Fixes kubernetes-csi#110

Author: mkimuram

pkg/controller/controller.go

@@ -31,6 +31,7 @@ import (
 	"github.com/golang/glog"
 
 	"github.com/kubernetes-incubator/external-storage/lib/controller"
+	"github.com/kubernetes-incubator/external-storage/lib/util"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -89,9 +90,20 @@ var (
 	accessMode = &csi.VolumeCapability_AccessMode{
 		Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 	}
-	accessType = &csi.VolumeCapability_Mount{
+	accessTypeMount = &csi.VolumeCapability_Mount{
 		Mount: &csi.VolumeCapability_MountVolume{},
 	}
+	accessTypeBlock = &csi.VolumeCapability_Block{
+		Block: &csi.VolumeCapability_BlockVolume{},
+	}
+	volumeCapabilityMountSingle = &csi.VolumeCapability{
+		AccessType: accessTypeMount,
+		AccessMode: accessMode,
+	}
+	volumeCapabilityBlockSingle = &csi.VolumeCapability{
+		AccessType: accessTypeBlock,
+		AccessMode: accessMode,
+	}
 	// Each provisioner have a identify string to distinguish with others. This
 	// identify string will be added in PV annoations under this key.
 	provisionerIDKey = "storage.kubernetes.io/csiProvisionerIdentity"
@@ -295,21 +307,27 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 	capacity := options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
 	volSizeBytes := capacity.Value()
 
+	volumeCapabilities := []*csi.VolumeCapability{
+		volumeCapabilityMountSingle,
+	}
+
+	if util.CheckPersistentVolumeClaimModeBlock(options.PVC) {
+		volumeCapabilities = []*csi.VolumeCapability{
+			volumeCapabilityBlockSingle,
+		}
+	}
+
 	// Create a CSI CreateVolumeRequest and Response
 	req := csi.CreateVolumeRequest{
 
-		Name:       pvName,
-		Parameters: options.Parameters,
-		VolumeCapabilities: []*csi.VolumeCapability{
-			{
-				AccessType: accessType,
-				AccessMode: accessMode,
-			},
-		},
+		Name:               pvName,
+		Parameters:         options.Parameters,
+		VolumeCapabilities: volumeCapabilities,
 		CapacityRange: &csi.CapacityRange{
 			RequiredBytes: int64(volSizeBytes),
 		},
 	}
+
 	rep := &csi.CreateVolumeResponse{}
 
 	// Resolve provision secret credentials.
@@ -421,6 +439,10 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		},
 	}
 
+	if options.PVC.Spec.VolumeMode != nil {
+		pv.Spec.VolumeMode = options.PVC.Spec.VolumeMode
+	}
+
 	glog.Infof("successfully created PV %+v", pv.Spec.PersistentVolumeSource)
 
 	return pv, nil
@@ -466,6 +488,25 @@ func (p *csiProvisioner) Delete(volume *v1.PersistentVolume) error {
 	return err
 }
 
+func (p *csiProvisioner) SupportsBlock() bool {
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+
+	client := csi.NewControllerClient(p.grpcClient)
+	req := csi.ValidateVolumeCapabilitiesRequest{
+		VolumeCapabilities: []*csi.VolumeCapability{
+			volumeCapabilityBlockSingle,
+		},
+	}
+
+	rsp, err := client.ValidateVolumeCapabilities(ctx, &req)
+	if err != nil {
+		return false
+	}
+
+	return rsp.Supported
+}
+
 //TODO use a unique volume handle from and to Id
 func (p *csiProvisioner) volumeIdToHandle(id string) string {
 	return id