ScaleIO Volume Plugin - volume attribute updates

This commit introduces the following updates and fixes:
- Enable scaleIO volume multip-mapping based on accessMode
- No longer uses "default" as default values for storagepool & protection domain
- validates capacity when capacity is zero
- Better naming for PV and volume
- make mount ro when accessModes contains ROM

pkg/volume/scaleio/BUILD

@@ -22,6 +22,7 @@ go_test(
         "//pkg/volume:go_default_library",
         "//pkg/volume/testing:go_default_library",
         "//vendor/github.com/codedellemc/goscaleio/types/v1:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",

pkg/volume/scaleio/sio_client.go

@@ -45,7 +45,7 @@ type sioInterface interface {
 	FindVolume(name string) (*siotypes.Volume, error)
 	Volume(sioVolumeID) (*siotypes.Volume, error)
 	CreateVolume(name string, sizeGB int64) (*siotypes.Volume, error)
-	AttachVolume(sioVolumeID) error
+	AttachVolume(sioVolumeID, bool) error
 	DetachVolume(sioVolumeID) error
 	DeleteVolume(sioVolumeID) error
 	IID() (string, error)
@@ -217,8 +217,9 @@ func (c *sioClient) CreateVolume(name string, sizeGB int64) (*siotypes.Volume, e
 	return c.Volume(sioVolumeID(createResponse.ID))
 }
 
-// AttachVolume maps the scaleio volume to an sdc node.
+// AttachVolume maps the scaleio volume to an sdc node.  If the multipleMappings flag
-func (c *sioClient) AttachVolume(id sioVolumeID) error {
+// is true, ScaleIO will allow other SDC to map to that volume.
+func (c *sioClient) AttachVolume(id sioVolumeID, multipleMappings bool) error {
 	if err := c.init(); err != nil {
 		glog.Error(log("failed to init'd client in attach volume: %v", err))
 		return err
@@ -232,7 +233,7 @@ func (c *sioClient) AttachVolume(id sioVolumeID) error {
 
 	params := &siotypes.MapVolumeSdcParam{
 		SdcID: iid,
-		AllowMultipleMappings: "false",
+		AllowMultipleMappings: strconv.FormatBool(multipleMappings),
 		AllSdcs:               "",
 	}
 	volClient := sio.NewVolume(c.client)

pkg/volume/scaleio/sio_mgr.go

@@ -27,7 +27,7 @@ import (
 
 type storageInterface interface {
 	CreateVolume(string, int64) (*siotypes.Volume, error)
-	AttachVolume(string) (string, error)
+	AttachVolume(string, bool) (string, error)
 	IsAttached(string) (bool, error)
 	DetachVolume(string) error
 	DeleteVolume(string) error
@@ -103,8 +103,9 @@ func (m *sioMgr) CreateVolume(volName string, sizeGB int64) (*siotypes.Volume, e
 	return vol, nil
 }
 
-// AttachVolume maps a ScaleIO volume to the running node
+// AttachVolume maps a ScaleIO volume to the running node.  If flag multiMaps,
-func (m *sioMgr) AttachVolume(volName string) (string, error) {
+// ScaleIO will allow other SDC to map to volume.
+func (m *sioMgr) AttachVolume(volName string, multipleMappings bool) (string, error) {
 	client, err := m.getClient()
 	if err != nil {
 		glog.Error(log("attach volume failed: %v", err))
@@ -139,7 +140,7 @@ func (m *sioMgr) AttachVolume(volName string) (string, error) {
 	}
 
 	// attach volume, get deviceName
-	if err := client.AttachVolume(sioVolumeID(vol.ID)); err != nil {
+	if err := client.AttachVolume(sioVolumeID(vol.ID), multipleMappings); err != nil {
 		glog.Error(log("attachment for volume %s failed :%v", volName, err))
 		return "", err
 	}

pkg/volume/scaleio/sio_mgr_test.go

@@ -99,7 +99,7 @@ func TestMgrCreateVolume(t *testing.T) {
 func TestMgrAttachVolume(t *testing.T) {
 	mgr := newTestMgr(t)
 	mgr.CreateVolume("test-vol-0001", 8*1024*1024)
-	device, err := mgr.AttachVolume("test-vol-0001")
+	device, err := mgr.AttachVolume("test-vol-0001", false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -111,8 +111,8 @@ func TestMgrAttachVolume(t *testing.T) {
 func TestMgrAttachVolume_AlreadyAttached(t *testing.T) {
 	mgr := newTestMgr(t)
 	mgr.CreateVolume("test-vol-0001", 8*1024*1024)
-	mgr.AttachVolume("test-vol-0001")
+	mgr.AttachVolume("test-vol-0001", false)
-	dev, err := mgr.AttachVolume("test-vol-0001")
+	dev, err := mgr.AttachVolume("test-vol-0001", false)
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -124,7 +124,8 @@ func TestMgrAttachVolume_AlreadyAttached(t *testing.T) {
 func TestMgrAttachVolume_VolumeNotFoundError(t *testing.T) {
 	mgr := newTestMgr(t)
 	mgr.CreateVolume("test-vol-0001", 8*1024*1024)
-	_, err := mgr.AttachVolume("test-vol-0002")
+	_, err := mgr.AttachVolume("test-vol-0002", false)
+
 	if err == nil {
 		t.Error("attachVolume should fail with volume not found error")
 	}
@@ -137,7 +138,7 @@ func TestMgrAttachVolume_WaitForAttachError(t *testing.T) {
 		c := mgr.client.(*fakeSio)
 		close(c.waitAttachCtrl)
 	}()
-	_, err := mgr.AttachVolume("test-vol-0001")
+	_, err := mgr.AttachVolume("test-vol-0001", false)
 	if err == nil {
 		t.Error("attachVolume should fail with attach timeout error")
 	}
@@ -146,7 +147,7 @@ func TestMgrAttachVolume_WaitForAttachError(t *testing.T) {
 func TestMgrDetachVolume(t *testing.T) {
 	mgr := newTestMgr(t)
 	mgr.CreateVolume("test-vol-0001", 8*1024*1024)
-	mgr.AttachVolume("test-vol-0001")
+	mgr.AttachVolume("test-vol-0001", false)
 	if err := mgr.DetachVolume("test-vol-0001"); err != nil {
 		t.Fatal(err)
 	}
@@ -162,7 +163,7 @@ func TestMgrDetachVolume(t *testing.T) {
 func TestMgrDetachVolume_VolumeNotFound(t *testing.T) {
 	mgr := newTestMgr(t)
 	mgr.CreateVolume("test-vol-0001", 8*1024*1024)
-	mgr.AttachVolume("test-vol-0001")
+	mgr.AttachVolume("test-vol-0001", false)
 	err := mgr.DetachVolume("test-vol-0002")
 	if err == nil {
 		t.Fatal("expected a volume not found failure")
@@ -181,7 +182,7 @@ func TestMgrDetachVolume_VolumeNotAttached(t *testing.T) {
 func TestMgrDetachVolume_VolumeAlreadyDetached(t *testing.T) {
 	mgr := newTestMgr(t)
 	mgr.CreateVolume("test-vol-0001", 8*1024*1024)
-	mgr.AttachVolume("test-vol-0001")
+	mgr.AttachVolume("test-vol-0001", false)
 	mgr.DetachVolume("test-vol-0001")
 	err := mgr.DetachVolume("test-vol-0001")
 	if err != nil {
@@ -192,7 +193,7 @@ func TestMgrDetachVolume_VolumeAlreadyDetached(t *testing.T) {
 func TestMgrDetachVolume_WaitForDetachError(t *testing.T) {
 	mgr := newTestMgr(t)
 	mgr.CreateVolume("test-vol-0001", 8*1024*1024)
-	mgr.AttachVolume("test-vol-0001")
+	mgr.AttachVolume("test-vol-0001", false)
 	err := mgr.DetachVolume("test-vol-0001")
 	if err != nil {
 		t.Error("detachVolume failed")
@@ -227,6 +228,7 @@ type fakeSio struct {
 	waitAttachCtrl chan struct{}
 	waitDetachCtrl chan struct{}
 	devs           map[string]string
+	isMultiMap     bool
 }
 
 func newFakeSio() *fakeSio {
@@ -261,7 +263,8 @@ func (f *fakeSio) CreateVolume(volName string, sizeGB int64) (*siotypes.Volume,
 	return f.volume, nil
 }
 
-func (f *fakeSio) AttachVolume(id sioVolumeID) error {
+func (f *fakeSio) AttachVolume(id sioVolumeID, multiMaps bool) error {
+	f.isMultiMap = multiMaps
 	_, err := f.Volume(id)
 	if err != nil {
 		return err

pkg/volume/scaleio/sio_plugin.go

@@ -149,6 +149,7 @@ var _ volume.PersistentVolumePlugin = &sioPlugin{}
 func (p *sioPlugin) GetAccessModes() []api.PersistentVolumeAccessMode {
 	return []api.PersistentVolumeAccessMode{
 		api.ReadWriteOnce,
+		api.ReadOnlyMany,
 	}
 }
 

pkg/volume/scaleio/sio_util.go

@@ -70,9 +70,11 @@ var (
 
 	secretNotFoundErr              = errors.New("secret not found")
 	configMapNotFoundErr           = errors.New("configMap not found")
-	gatewayNotProvidedErr   = errors.New("gateway not provided")
+	gatewayNotProvidedErr          = errors.New("ScaleIO gateway not provided")
 	secretRefNotProvidedErr        = errors.New("secret ref not provided")
-	systemNotProvidedErr    = errors.New("secret not provided")
+	systemNotProvidedErr           = errors.New("ScaleIO system not provided")
+	storagePoolNotProvidedErr      = errors.New("ScaleIO storage pool not provided")
+	protectionDomainNotProvidedErr = errors.New("ScaleIO protection domain not provided")
 )
 
 // mapScaleIOVolumeSource maps attributes from a ScaleIOVolumeSource to config
@@ -107,6 +109,12 @@ func validateConfigs(config map[string]string) error {
 	if config[confKey.system] == "" {
 		return systemNotProvidedErr
 	}
+	if config[confKey.storagePool] == "" {
+		return storagePoolNotProvidedErr
+	}
+	if config[confKey.protectionDomain] == "" {
+		return protectionDomainNotProvidedErr
+	}
 
 	return nil
 }
@@ -119,8 +127,6 @@ func applyConfigDefaults(config map[string]string) {
 		b = false
 	}
 	config[confKey.sslEnabled] = strconv.FormatBool(b)
-	config[confKey.protectionDomain] = defaultString(config[confKey.protectionDomain], "default")
-	config[confKey.storagePool] = defaultString(config[confKey.storagePool], "default")
 	config[confKey.storageMode] = defaultString(config[confKey.storageMode], "ThinProvisioned")
 	config[confKey.fsType] = defaultString(config[confKey.fsType], "xfs")
 	b, err = strconv.ParseBool(config[confKey.readOnly])

pkg/volume/scaleio/sio_util_test.go

@@ -115,10 +115,10 @@ func TestUtilApplyConfigDefaults(t *testing.T) {
 	if data[confKey.system] != "sio" {
 		t.Error("Unexpected system value")
 	}
-	if data[confKey.protectionDomain] != "default" {
+	if data[confKey.protectionDomain] != "" {
 		t.Error("Unexpected protection domain value")
 	}
-	if data[confKey.storagePool] != "default" {
+	if data[confKey.storagePool] != "" {
 		t.Error("Unexpected storage pool value")
 	}
 	if data[confKey.volumeName] != "sio-vol" {

pkg/volume/scaleio/sio_volume.go

@@ -88,7 +88,7 @@ func (v *sioVolume) SetUpAt(dir string, fsGroup *int64) error {
 	v.plugin.volumeMtx.LockKey(v.volSpecName)
 	defer v.plugin.volumeMtx.UnlockKey(v.volSpecName)
 
-	glog.V(4).Info(log("setting up volume %s", v.volSpecName))
+	glog.V(4).Info(log("setting up volume for PV.spec %s", v.volSpecName))
 	if err := v.setSioMgr(); err != nil {
 		glog.Error(log("setup failed to create scalio manager: %v", err))
 		return err
@@ -104,18 +104,36 @@ func (v *sioVolume) SetUpAt(dir string, fsGroup *int64) error {
 		return nil
 	}
 
-	// attach the volume and mount
+	// should multiple-mapping be enabled
+	enableMultiMaps := false
+	isROM := false
+	if v.spec.PersistentVolume != nil {
+		ams := v.spec.PersistentVolume.Spec.AccessModes
+		for _, am := range ams {
+			if am == api.ReadOnlyMany {
+				enableMultiMaps = true
+				isROM = true
+			}
+		}
+	}
+	glog.V(4).Info(log("multiple mapping enabled = %v", enableMultiMaps))
+
 	volName := v.volName
-	devicePath, err := v.sioMgr.AttachVolume(volName)
+	devicePath, err := v.sioMgr.AttachVolume(volName, enableMultiMaps)
 	if err != nil {
 		glog.Error(log("setup of volume %v:  %v", v.volSpecName, err))
 		return err
 	}
 	options := []string{}
-	if v.source.ReadOnly {
+	switch {
-		options = append(options, "ro")
+	default:
-	} else {
 		options = append(options, "rw")
+	case isROM && !v.source.ReadOnly:
+		options = append(options, "rw")
+	case isROM:
+		options = append(options, "ro")
+	case v.source.ReadOnly:
+		options = append(options, "ro")
 	}
 
 	glog.V(4).Info(log("mounting device  %s -> %s", devicePath, dir))
@@ -140,7 +158,12 @@ func (v *sioVolume) SetUpAt(dir string, fsGroup *int64) error {
 		return err
 	}
 
-	glog.V(4).Info(log("successfully setup volume %s attached %s:%s as %s", v.volSpecName, v.volName, devicePath, dir))
+	if !v.readOnly && fsGroup != nil {
+		glog.V(4).Info(log("applying  value FSGroup ownership"))
+		volume.SetVolumeOwnership(v, fsGroup)
+	}
+
+	glog.V(4).Info(log("successfully setup PV %s: volume %s mapped as %s mounted at %s", v.volSpecName, v.volName, devicePath, dir))
 	return nil
 }
 
@@ -191,7 +214,7 @@ func (v *sioVolume) TearDownAt(dir string) error {
 	// use "last attempt wins" strategy to detach volume from node
 	// only allow volume to detach when it is not busy (not being used by other pods)
 	if !deviceBusy {
-		glog.V(4).Info(log("teardown is attempting to detach/unmap volume for %s", v.volSpecName))
+		glog.V(4).Info(log("teardown is attempting to detach/unmap volume for PV %s", v.volSpecName))
 		if err := v.resetSioMgr(); err != nil {
 			glog.Error(log("teardown failed, unable to reset scalio mgr: %v", err))
 		}
@@ -224,7 +247,7 @@ func (v *sioVolume) Delete() error {
 		return err
 	}
 
-	glog.V(4).Info(log("successfully deleted pvc %s", v.volSpecName))
+	glog.V(4).Info(log("successfully deleted PV %s with volume %s", v.volSpecName, v.volName))
 	return nil
 }
 
@@ -234,17 +257,30 @@ func (v *sioVolume) Delete() error {
 var _ volume.Provisioner = &sioVolume{}
 
 func (v *sioVolume) Provision() (*api.PersistentVolume, error) {
-	glog.V(4).Info(log("attempting to dynamically provision pvc %v", v.options.PVName))
+	glog.V(4).Info(log("attempting to dynamically provision pvc %v", v.options.PVC.Name))
 
 	if !volume.AccessModesContainedInAll(v.plugin.GetAccessModes(), v.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", v.options.PVC.Spec.AccessModes, v.plugin.GetAccessModes())
 	}
 
 	// setup volume attrributes
-	name := v.generateVolName()
+	genName := v.generateName("k8svol", 11)
+	var oneGig int64 = 1024 * 1024 * 1024
+	var eightGig int64 = 8 * oneGig
+
 	capacity := v.options.PVC.Spec.Resources.Requests[api.ResourceName(api.ResourceStorage)]
 	volSizeBytes := capacity.Value()
-	volSizeGB := int64(volume.RoundUpSize(volSizeBytes, 1024*1024*1024))
+	volSizeGB := int64(volume.RoundUpSize(volSizeBytes, oneGig))
+
+	if volSizeBytes == 0 {
+		return nil, fmt.Errorf("invalid volume size of 0 specified")
+	}
+
+	if volSizeBytes < eightGig {
+		volSizeGB = int64(volume.RoundUpSize(eightGig, oneGig))
+		glog.V(4).Info(log("capacity less than 8Gi found, adjusted to %dGi", volSizeGB))
+
+	}
 
 	// create sio manager
 	if err := v.setSioMgrFromConfig(); err != nil {
@@ -253,14 +289,15 @@ func (v *sioVolume) Provision() (*api.PersistentVolume, error) {
 	}
 
 	// create volume
-	vol, err := v.sioMgr.CreateVolume(name, volSizeGB)
+	volName := genName
+	vol, err := v.sioMgr.CreateVolume(volName, volSizeGB)
 	if err != nil {
 		glog.Error(log("provision failed while creating volume: %v", err))
 		return nil, err
 	}
 
 	// prepare data for pv
-	v.configData[confKey.volumeName] = name
+	v.configData[confKey.volumeName] = volName
 	sslEnabled, err := strconv.ParseBool(v.configData[confKey.sslEnabled])
 	if err != nil {
 		glog.Warning(log("failed to parse parameter sslEnabled, setting to false"))
@@ -273,9 +310,10 @@ func (v *sioVolume) Provision() (*api.PersistentVolume, error) {
 	}
 
 	// describe created pv
+	pvName := genName
 	pv := &api.PersistentVolume{
 		ObjectMeta: meta.ObjectMeta{
-			Name:      v.options.PVName,
+			Name:      pvName,
 			Namespace: v.options.PVC.Namespace,
 			Labels:    map[string]string{},
 			Annotations: map[string]string{
@@ -299,7 +337,7 @@ func (v *sioVolume) Provision() (*api.PersistentVolume, error) {
 					ProtectionDomain: v.configData[confKey.protectionDomain],
 					StoragePool:      v.configData[confKey.storagePool],
 					StorageMode:      v.configData[confKey.storageMode],
-					VolumeName:       name,
+					VolumeName:       volName,
 					FSType:           v.configData[confKey.fsType],
 					ReadOnly:         readOnly,
 				},
@@ -310,14 +348,14 @@ func (v *sioVolume) Provision() (*api.PersistentVolume, error) {
 		pv.Spec.AccessModes = v.plugin.GetAccessModes()
 	}
 
-	glog.V(4).Info(log("provisioner dynamically created pvc %v with volume %s successfully", pv.Name, vol.Name))
+	glog.V(4).Info(log("provisioner created pv %v and volume %s successfully", pvName, vol.Name))
 	return pv, nil
 }
 
 // setSioMgr creates scaleio mgr from cached config data if found
 // otherwise, setups new config data and create mgr
 func (v *sioVolume) setSioMgr() error {
-	glog.V(4).Info(log("setting up sio mgr for vol  %s", v.volSpecName))
+	glog.V(4).Info(log("setting up sio mgr for spec  %s", v.volSpecName))
 	podDir := v.plugin.host.GetPodPluginDir(v.podUID, sioPluginName)
 	configName := path.Join(podDir, sioConfigFileName)
 	if v.sioMgr == nil {
@@ -455,6 +493,6 @@ func (v *sioVolume) setSioMgrFromSpec() error {
 	return nil
 }
 
-func (v *sioVolume) generateVolName() string {
+func (v *sioVolume) generateName(prefix string, size int) string {
-	return "sio-" + strings.Replace(string(uuid.NewUUID()), "-", "", -1)[0:25]
+	return fmt.Sprintf("%s-%s", prefix, strings.Replace(string(uuid.NewUUID()), "-", "", -1)[0:size])
 }

pkg/volume/scaleio/sio_volume_test.go

@@ -23,6 +23,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/golang/glog"
+
 	api "k8s.io/api/core/v1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -149,6 +151,7 @@ func TestVolumeMounterUnmounter(t *testing.T) {
 				VolumeName:       testSioVol,
 				FSType:           "ext4",
 				SecretRef:        &api.LocalObjectReference{Name: "sio-secret"},
+				ReadOnly:         false,
 			},
 		},
 	}
@@ -191,6 +194,10 @@ func TestVolumeMounterUnmounter(t *testing.T) {
 		}
 	}
 
+	if sio.isMultiMap {
+		t.Errorf("SetUp() - expecting multiple volume disabled by default")
+	}
+
 	// rebuild spec
 	builtSpec, err := sioPlug.ConstructVolumeSpec(volume.NewSpecFromVolume(vol).Name(), path)
 	if err != nil {
@@ -235,25 +242,23 @@ func TestVolumeProvisioner(t *testing.T) {
 
 	plug, err := plugMgr.FindPluginByName(sioPluginName)
 	if err != nil {
-		t.Errorf("Can't find the plugin %v", sioPluginName)
+		t.Fatalf("Can't find the plugin %v", sioPluginName)
 	}
 	sioPlug, ok := plug.(*sioPlugin)
 	if !ok {
-		t.Errorf("Cannot assert plugin to be type sioPlugin")
+		t.Fatal("Cannot assert plugin to be type sioPlugin")
 	}
 
 	options := volume.VolumeOptions{
 		ClusterName: "testcluster",
-		PVName:      "pvc-sio-dynamic-vol",
 		PVC:         volumetest.CreateTestPVC("100Mi", []api.PersistentVolumeAccessMode{api.ReadWriteOnce}),
 		PersistentVolumeReclaimPolicy: api.PersistentVolumeReclaimDelete,
 	}
+	options.PVC.Name = "testpvc"
 	options.PVC.Namespace = testns
 
-	// incomplete options, test should fail
+	options.PVC.Spec.AccessModes = []api.PersistentVolumeAccessMode{
-	_, err = sioPlug.NewProvisioner(options)
+		api.ReadOnlyMany,
-	if err == nil {
-		t.Fatal("expected failure due to incomplete options")
 	}
 
 	options.Parameters = map[string]string{
@@ -288,10 +293,9 @@ func TestVolumeProvisioner(t *testing.T) {
 	// validate provision
 	actualSpecName := spec.Name
 	actualVolName := spec.Spec.PersistentVolumeSource.ScaleIO.VolumeName
-	if !strings.HasPrefix(actualSpecName, "pvc-") {
+	if !strings.HasPrefix(actualSpecName, "k8svol-") {
-		t.Errorf("expecting volume name to start with pov-, got %s", actualSpecName)
+		t.Errorf("expecting volume name to start with k8svol-, got %s", actualSpecName)
 	}
-
 	vol, err := sio.FindVolume(actualVolName)
 	if err != nil {
 		t.Fatalf("failed getting volume %v: %v", actualVolName, err)
@@ -299,6 +303,9 @@ func TestVolumeProvisioner(t *testing.T) {
 	if vol.Name != actualVolName {
 		t.Errorf("expected volume name to be %s, got %s", actualVolName, vol.Name)
 	}
+	if vol.SizeInKb != 8*1024*1024 {
+		glog.V(4).Info(log("unexpected volume size"))
+	}
 
 	// mount dynamic vol
 	sioMounter, err := sioPlug.NewMounter(
@@ -315,8 +322,14 @@ func TestVolumeProvisioner(t *testing.T) {
 	}
 	sioVol.sioMgr.client = sio
 	if err := sioMounter.SetUp(nil); err != nil {
-		t.Errorf("Expected success, got: %v", err)
+		t.Fatalf("Expected success, got: %v", err)
 	}
+
+	// isMultiMap applied
+	if !sio.isMultiMap {
+		t.Errorf("SetUp()  expecting attached volume with multi-mapping")
+	}
+
 	// teardown dynamic vol
 	sioUnmounter, err := sioPlug.NewUnmounter(spec.Name, podUID)
 	if err != nil {
@@ -351,3 +364,83 @@ func TestVolumeProvisioner(t *testing.T) {
 		t.Errorf("Deleter did not delete path %v: %v", path, err)
 	}
 }
+
+func TestVolumeProvisionerWithIncompleteConfig(t *testing.T) {
+	plugMgr, tmpDir := newPluginMgr(t)
+	defer os.RemoveAll(tmpDir)
+
+	plug, err := plugMgr.FindPluginByName(sioPluginName)
+	if err != nil {
+		t.Fatalf("Can't find the plugin %v", sioPluginName)
+	}
+	sioPlug, ok := plug.(*sioPlugin)
+	if !ok {
+		t.Fatal("Cannot assert plugin to be type sioPlugin")
+	}
+
+	options := volume.VolumeOptions{
+		ClusterName: "testcluster",
+		PVName:      "pvc-sio-dynamic-vol",
+		PVC:         volumetest.CreateTestPVC("100Mi", []api.PersistentVolumeAccessMode{api.ReadWriteOnce}),
+		PersistentVolumeReclaimPolicy: api.PersistentVolumeReclaimDelete,
+	}
+	options.PVC.Namespace = testns
+
+	options.PVC.Spec.AccessModes = []api.PersistentVolumeAccessMode{
+		api.ReadWriteOnce,
+	}
+
+	// incomplete options, test should fail
+	_, err = sioPlug.NewProvisioner(options)
+	if err == nil {
+		t.Fatal("expected failure due to incomplete options")
+	}
+}
+
+func TestVolumeProvisionerWithZeroCapacity(t *testing.T) {
+	plugMgr, tmpDir := newPluginMgr(t)
+	defer os.RemoveAll(tmpDir)
+
+	plug, err := plugMgr.FindPluginByName(sioPluginName)
+	if err != nil {
+		t.Fatalf("Can't find the plugin %v", sioPluginName)
+	}
+	sioPlug, ok := plug.(*sioPlugin)
+	if !ok {
+		t.Fatal("Cannot assert plugin to be type sioPlugin")
+	}
+
+	options := volume.VolumeOptions{
+		ClusterName: "testcluster",
+		PVName:      "pvc-sio-dynamic-vol",
+		PVC:         volumetest.CreateTestPVC("0Mi", []api.PersistentVolumeAccessMode{api.ReadWriteOnce}),
+		PersistentVolumeReclaimPolicy: api.PersistentVolumeReclaimDelete,
+	}
+	options.PVC.Namespace = testns
+
+	options.PVC.Spec.AccessModes = []api.PersistentVolumeAccessMode{
+		api.ReadWriteOnce,
+	}
+
+	options.Parameters = map[string]string{
+		confKey.gateway:          "http://test.scaleio:11111",
+		confKey.system:           "sio",
+		confKey.protectionDomain: testSioPD,
+		confKey.storagePool:      "default",
+		confKey.secretRef:        "sio-secret",
+	}
+
+	provisioner, _ := sioPlug.NewProvisioner(options)
+	sio := newFakeSio()
+	sioVol := provisioner.(*sioVolume)
+	if err := sioVol.setSioMgrFromConfig(); err != nil {
+		t.Fatalf("failed to create scaleio mgr from config: %v", err)
+	}
+	sioVol.sioMgr.client = sio
+
+	_, err = provisioner.Provision()
+	if err == nil {
+		t.Fatalf("call to Provision() should fail with invalid capacity")
+	}
+
+}