Merge pull request #41906 from gnufied/implement-mount-options

Automatic merge from submit-queue

Implement support for mount options in PVs

**What this PR does / why we need it**:

This PR implements support for mount options in PersistentVolume via `volume.beta.kubernetes.io/mount-options` annotation.

**Which issue this PR fixes** 

Fixes https://github.com/kubernetes/features/issues/168

**Release note**:
```
Enable additional, custom mount options to be passed to PersistentVolume objects via volume.beta.kubernetes.io/mount-options annotation.
```

pkg/api/validation/BUILD

@@ -15,6 +15,7 @@ go_library(
         "events.go",
         "schema.go",
         "validation.go",
+        "volume_plugins.go",
     ],
     tags = ["automanaged"],
     deps = [
@@ -27,6 +28,30 @@ go_library(
         "//pkg/capabilities:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/security/apparmor:go_default_library",
+        "//pkg/volume:go_default_library",
+        "//pkg/volume/aws_ebs:go_default_library",
+        "//pkg/volume/azure_dd:go_default_library",
+        "//pkg/volume/azure_file:go_default_library",
+        "//pkg/volume/cephfs:go_default_library",
+        "//pkg/volume/cinder:go_default_library",
+        "//pkg/volume/configmap:go_default_library",
+        "//pkg/volume/downwardapi:go_default_library",
+        "//pkg/volume/empty_dir:go_default_library",
+        "//pkg/volume/fc:go_default_library",
+        "//pkg/volume/flexvolume:go_default_library",
+        "//pkg/volume/flocker:go_default_library",
+        "//pkg/volume/gce_pd:go_default_library",
+        "//pkg/volume/git_repo:go_default_library",
+        "//pkg/volume/glusterfs:go_default_library",
+        "//pkg/volume/host_path:go_default_library",
+        "//pkg/volume/iscsi:go_default_library",
+        "//pkg/volume/nfs:go_default_library",
+        "//pkg/volume/photon_pd:go_default_library",
+        "//pkg/volume/projected:go_default_library",
+        "//pkg/volume/quobyte:go_default_library",
+        "//pkg/volume/rbd:go_default_library",
+        "//pkg/volume/secret:go_default_library",
+        "//pkg/volume/vsphere_volume:go_default_library",
         "//vendor:github.com/emicklei/go-restful/swagger",
         "//vendor:github.com/exponent-io/jsonpath",
         "//vendor:github.com/golang/glog",
@@ -77,6 +102,7 @@ go_test(
         "//pkg/apis/storage/util:go_default_library",
         "//pkg/capabilities:go_default_library",
         "//pkg/security/apparmor:go_default_library",
+        "//pkg/volume:go_default_library",
         "//vendor:github.com/ghodss/yaml",
         "//vendor:k8s.io/apimachinery/pkg/api/resource",
         "//vendor:k8s.io/apimachinery/pkg/api/testing",

pkg/api/validation/validation.go

@@ -48,6 +48,7 @@ import (
 	"k8s.io/kubernetes/pkg/capabilities"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/security/apparmor"
+	"k8s.io/kubernetes/pkg/volume"
 )
 
 // TODO: delete this global variable when we enable the validation of common
@@ -64,6 +65,11 @@ var volumeModeErrorMsg string = "must be a number between 0 and 0777 (octal), bo
 
 // BannedOwners is a black list of object that are not allowed to be owners.
 var BannedOwners = genericvalidation.BannedOwners
+var volumePlugins []volume.VolumePlugin
+
+func init() {
+	volumePlugins = probeVolumePlugins()
+}
 
 // ValidateHasLabel requires that metav1.ObjectMeta has a Label with key and expectedValue
 func ValidateHasLabel(meta metav1.ObjectMeta, fldPath *field.Path, key, expectedValue string) field.ErrorList {
@@ -1032,6 +1038,20 @@ func ValidatePersistentVolume(pv *api.PersistentVolume) field.ErrorList {
 		}
 	}
 
+	volumePlugin := findPluginBySpec(volumePlugins, pv)
+	mountOptions := volume.MountOptionFromApiPV(pv)
+
+	metaField := field.NewPath("metadata")
+	if volumePlugin == nil && len(mountOptions) > 0 {
+		allErrs = append(allErrs, field.Forbidden(metaField.Child("annotations", volume.MountOptionAnnotation), "may not specify mount options for this volume type"))
+	}
+
+	if volumePlugin != nil {
+		if !volumePlugin.SupportsMountOption() && len(mountOptions) > 0 {
+			allErrs = append(allErrs, field.Forbidden(metaField.Child("annotations", volume.MountOptionAnnotation), "may not specify mount options for this volume type"))
+		}
+	}
+
 	numVolumes := 0
 	if pv.Spec.HostPath != nil {
 		if numVolumes > 0 {

pkg/api/validation/validation_test.go

@@ -31,6 +31,7 @@ import (
 	storageutil "k8s.io/kubernetes/pkg/apis/storage/util"
 	"k8s.io/kubernetes/pkg/capabilities"
 	"k8s.io/kubernetes/pkg/security/apparmor"
+	"k8s.io/kubernetes/pkg/volume"
 )
 
 const (
@@ -205,6 +206,30 @@ func TestValidatePersistentVolumes(t *testing.T) {
 				PersistentVolumeReclaimPolicy: api.PersistentVolumeReclaimRecycle,
 			}),
 		},
+		"volume with valid mount option for nfs": {
+			isExpectedFailure: false,
+			volume: testVolumeWithMountOption("good-nfs-mount-volume", "", "ro,nfsvers=3", api.PersistentVolumeSpec{
+				Capacity: api.ResourceList{
+					api.ResourceName(api.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
+				PersistentVolumeSource: api.PersistentVolumeSource{
+					NFS: &api.NFSVolumeSource{Server: "localhost", Path: "/srv", ReadOnly: false},
+				},
+			}),
+		},
+		"volume with mount option for host path": {
+			isExpectedFailure: true,
+			volume: testVolumeWithMountOption("bad-hostpath-mount-volume", "", "ro,nfsvers=3", api.PersistentVolumeSpec{
+				Capacity: api.ResourceList{
+					api.ResourceName(api.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
+				PersistentVolumeSource: api.PersistentVolumeSource{
+					HostPath: &api.HostPathVolumeSource{Path: "/a/.."},
+				},
+			}),
+		},
 	}
 
 	for name, scenario := range scenarios {
@@ -241,6 +266,25 @@ func testVolumeClaimStorageClass(name string, namespace string, annval string, s
 	}
 }
 
+func testVolumeWithMountOption(name string, namespace string, mountOptions string, spec api.PersistentVolumeSpec) *api.PersistentVolume {
+	annotations := map[string]string{
+		volume.MountOptionAnnotation: mountOptions,
+	}
+	objMeta := metav1.ObjectMeta{
+		Name:        name,
+		Annotations: annotations,
+	}
+
+	if namespace != "" {
+		objMeta.Namespace = namespace
+	}
+
+	return &api.PersistentVolume{
+		ObjectMeta: objMeta,
+		Spec:       spec,
+	}
+}
+
 func testVolumeClaimAnnotation(name string, namespace string, ann string, annval string, spec api.PersistentVolumeClaimSpec) *api.PersistentVolumeClaim {
 	annotations := map[string]string{
 		ann: annval,

pkg/api/validation/volume_plugins.go

@@ -0,0 +1,105 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package validation
+
+import (
+	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/v1"
+	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/aws_ebs"
+	"k8s.io/kubernetes/pkg/volume/azure_dd"
+	"k8s.io/kubernetes/pkg/volume/azure_file"
+	"k8s.io/kubernetes/pkg/volume/cephfs"
+	"k8s.io/kubernetes/pkg/volume/cinder"
+	"k8s.io/kubernetes/pkg/volume/configmap"
+	"k8s.io/kubernetes/pkg/volume/downwardapi"
+	"k8s.io/kubernetes/pkg/volume/empty_dir"
+	"k8s.io/kubernetes/pkg/volume/fc"
+	"k8s.io/kubernetes/pkg/volume/flexvolume"
+	"k8s.io/kubernetes/pkg/volume/flocker"
+	"k8s.io/kubernetes/pkg/volume/gce_pd"
+	"k8s.io/kubernetes/pkg/volume/git_repo"
+	"k8s.io/kubernetes/pkg/volume/glusterfs"
+	"k8s.io/kubernetes/pkg/volume/host_path"
+	"k8s.io/kubernetes/pkg/volume/iscsi"
+	"k8s.io/kubernetes/pkg/volume/nfs"
+	"k8s.io/kubernetes/pkg/volume/photon_pd"
+	"k8s.io/kubernetes/pkg/volume/projected"
+	"k8s.io/kubernetes/pkg/volume/quobyte"
+	"k8s.io/kubernetes/pkg/volume/rbd"
+	"k8s.io/kubernetes/pkg/volume/secret"
+	"k8s.io/kubernetes/pkg/volume/vsphere_volume"
+)
+
+func probeVolumePlugins() []volume.VolumePlugin {
+	allPlugins := []volume.VolumePlugin{}
+
+	// list of volume plugins to probe for
+	allPlugins = append(allPlugins, aws_ebs.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, empty_dir.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, gce_pd.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, git_repo.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, host_path.ProbeVolumePlugins(volume.VolumeConfig{})...)
+	allPlugins = append(allPlugins, nfs.ProbeVolumePlugins(volume.VolumeConfig{})...)
+	allPlugins = append(allPlugins, secret.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, iscsi.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, glusterfs.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, rbd.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, cinder.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, quobyte.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, cephfs.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, downwardapi.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, fc.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, flocker.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, flexvolume.ProbeVolumePlugins("")...)
+	allPlugins = append(allPlugins, azure_file.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, configmap.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, vsphere_volume.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, azure_dd.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, photon_pd.ProbeVolumePlugins()...)
+	allPlugins = append(allPlugins, projected.ProbeVolumePlugins()...)
+	return allPlugins
+}
+
+func findPluginBySpec(volumePlugins []volume.VolumePlugin, pv *api.PersistentVolume) volume.VolumePlugin {
+	matches := []volume.VolumePlugin{}
+	v1Pv := &v1.PersistentVolume{}
+	err := v1.Convert_api_PersistentVolume_To_v1_PersistentVolume(pv, v1Pv, nil)
+	if err != nil {
+		glog.Errorf("Error converting to v1.PersistentVolume: %v", err)
+		return nil
+	}
+	volumeSpec := &volume.Spec{PersistentVolume: v1Pv}
+	for _, plugin := range volumePlugins {
+		if plugin.CanSupport(volumeSpec) {
+			matches = append(matches, plugin)
+		}
+	}
+
+	if len(matches) == 0 {
+		glog.V(5).Infof("No matching plugin found for : %s", pv.Name)
+		return nil
+	}
+
+	if len(matches) > 1 {
+		glog.V(3).Infof("multiple volume plugins matched for : %s ", pv.Name)
+		return nil
+	}
+
+	return matches[0]
+}

pkg/controller/volume/persistentvolume/framework_test.go

@@ -1130,6 +1130,10 @@ func (plugin *mockVolumePlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *mockVolumePlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *mockVolumePlugin) ConstructVolumeSpec(volumeName, mountPath string) (*vol.Spec, error) {
 	return nil, nil
 }

pkg/kubelet/events/event.go

@@ -65,6 +65,7 @@ const (
 	ImageGCFailed                        = "ImageGCFailed"
 	FailedNodeAllocatableEnforcement     = "FailedNodeAllocatableEnforcement"
 	SuccessfulNodeAllocatableEnforcement = "NodeAllocatableEnforced"
+	UnsupportedMountOption               = "UnsupportedMountOption"
 
 	// Image manager event reason list
 	InvalidDiskCapacity = "InvalidDiskCapacity"

pkg/volume/BUILD

@@ -24,6 +24,7 @@ go_library(
     ],
     tags = ["automanaged"],
     deps = [
+        "//pkg/api:go_default_library",
         "//pkg/api/v1:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/cloudprovider:go_default_library",
@@ -58,6 +59,7 @@ go_test(
     deps = [
         "//pkg/api:go_default_library",
         "//pkg/api/v1:go_default_library",
+        "//pkg/util/slice:go_default_library",
         "//vendor:k8s.io/apimachinery/pkg/api/errors",
         "//vendor:k8s.io/apimachinery/pkg/api/resource",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",

pkg/volume/aws_ebs/attacher.go

@@ -193,7 +193,8 @@ func (attacher *awsElasticBlockStoreAttacher) MountDevice(spec *volume.Spec, dev
 	}
 	if notMnt {
 		diskMounter := &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()}
-		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, options)
+		mountOptions := volume.MountOptionFromSpec(spec, options...)
+		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, mountOptions)
 		if err != nil {
 			os.Remove(deviceMountPath)
 			return err

pkg/volume/aws_ebs/aws_ebs.go

@@ -87,6 +87,10 @@ func (plugin *awsElasticBlockStorePlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *awsElasticBlockStorePlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *awsElasticBlockStorePlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/azure_dd/attacher.go

@@ -218,7 +218,8 @@ func (attacher *azureDiskAttacher) MountDevice(spec *volume.Spec, devicePath str
 	}
 	if notMnt {
 		diskMounter := &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()}
-		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, *volumeSource.FSType, options)
+		mountOptions := volume.MountOptionFromSpec(spec, options...)
+		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, *volumeSource.FSType, mountOptions)
 		if err != nil {
 			os.Remove(deviceMountPath)
 			return err

pkg/volume/azure_dd/azure_dd.go

@@ -103,6 +103,10 @@ func (plugin *azureDataDiskPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *azureDataDiskPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *azureDataDiskPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/azure_file/azure_file.go

@@ -79,6 +79,10 @@ func (plugin *azureFilePlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *azureFilePlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *azureFilePlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -105,10 +109,11 @@ func (plugin *azureFilePlugin) newMounterInternal(spec *volume.Spec, pod *v1.Pod
 			plugin:          plugin,
 			MetricsProvider: volume.NewMetricsStatFS(getPath(pod.UID, spec.Name(), plugin.host)),
 		},
-		util:       util,
+		util:         util,
-		secretName: source.SecretName,
+		secretName:   source.SecretName,
-		shareName:  source.ShareName,
+		shareName:    source.ShareName,
-		readOnly:   readOnly,
+		readOnly:     readOnly,
+		mountOptions: volume.MountOptionFromSpec(spec),
 	}, nil
 }
 
@@ -154,10 +159,11 @@ func (azureFileVolume *azureFile) GetPath() string {
 
 type azureFileMounter struct {
 	*azureFile
-	util       azureUtil
+	util         azureUtil
-	secretName string
+	secretName   string
-	shareName  string
+	shareName    string
-	readOnly   bool
+	readOnly     bool
+	mountOptions []string
 }
 
 var _ volume.Mounter = &azureFileMounter{}
@@ -202,7 +208,8 @@ func (b *azureFileMounter) SetUpAt(dir string, fsGroup *int64) error {
 	if b.readOnly {
 		options = append(options, "ro")
 	}
-	err = b.mounter.Mount(source, dir, "cifs", options)
+	mountOptions := volume.JoinMountOptions(b.mountOptions, options)
+	err = b.mounter.Mount(source, dir, "cifs", mountOptions)
 	if err != nil {
 		notMnt, mntErr := b.mounter.IsLikelyNotMountPoint(dir)
 		if mntErr != nil {

pkg/volume/cephfs/cephfs.go

@@ -72,6 +72,10 @@ func (plugin *cephfsPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *cephfsPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *cephfsPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -129,16 +133,18 @@ func (plugin *cephfsPlugin) newMounterInternal(spec *volume.Spec, podUID types.U
 
 	return &cephfsMounter{
 		cephfs: &cephfs{
-			podUID:      podUID,
+			podUID:       podUID,
-			volName:     spec.Name(),
+			volName:      spec.Name(),
-			mon:         cephvs.Monitors,
+			mon:          cephvs.Monitors,
-			path:        path,
+			path:         path,
-			secret:      secret,
+			secret:       secret,
-			id:          id,
+			id:           id,
-			secret_file: secret_file,
+			secret_file:  secret_file,
-			readonly:    cephvs.ReadOnly,
+			readonly:     cephvs.ReadOnly,
-			mounter:     mounter,
+			mounter:      mounter,
-			plugin:      plugin},
+			plugin:       plugin,
+			mountOptions: volume.MountOptionFromSpec(spec),
+		},
 	}, nil
 }
 
@@ -182,6 +188,7 @@ type cephfs struct {
 	mounter     mount.Interface
 	plugin      *cephfsPlugin
 	volume.MetricsNil
+	mountOptions []string
 }
 
 type cephfsMounter struct {
@@ -282,7 +289,8 @@ func (cephfsVolume *cephfs) execMount(mountpoint string) error {
 	}
 	src += hosts[i] + ":" + cephfsVolume.path
 
-	if err := cephfsVolume.mounter.Mount(src, mountpoint, "ceph", opt); err != nil {
+	mountOptions := volume.JoinMountOptions(cephfsVolume.mountOptions, opt)
+	if err := cephfsVolume.mounter.Mount(src, mountpoint, "ceph", mountOptions); err != nil {
 		return fmt.Errorf("CephFS: mount failed: %v", err)
 	}
 

pkg/volume/cinder/attacher.go

@@ -221,7 +221,8 @@ func (attacher *cinderDiskAttacher) MountDevice(spec *volume.Spec, devicePath st
 	}
 	if notMnt {
 		diskMounter := &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()}
-		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, options)
+		mountOptions := volume.MountOptionFromSpec(spec, options...)
+		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, mountOptions)
 		if err != nil {
 			os.Remove(deviceMountPath)
 			return err

pkg/volume/cinder/cinder.go

@@ -99,6 +99,10 @@ func (plugin *cinderPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *cinderPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *cinderPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/configmap/configmap.go

@@ -76,6 +76,10 @@ func (plugin *configMapPlugin) RequiresRemount() bool {
 	return true
 }
 
+func (plugin *configMapPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *configMapPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, opts volume.VolumeOptions) (volume.Mounter, error) {
 	return &configMapVolumeMounter{
 		configMapVolume: &configMapVolume{spec.Name(), pod.UID, plugin, plugin.host.GetMounter(), plugin.host.GetWriter(), volume.MetricsNil{}},

pkg/volume/downwardapi/downwardapi.go

@@ -82,6 +82,10 @@ func (plugin *downwardAPIPlugin) RequiresRemount() bool {
 	return true
 }
 
+func (plugin *downwardAPIPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *downwardAPIPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, opts volume.VolumeOptions) (volume.Mounter, error) {
 	v := &downwardAPIVolume{
 		volName: spec.Name(),

pkg/volume/empty_dir/empty_dir.go

@@ -90,6 +90,10 @@ func (plugin *emptyDirPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *emptyDirPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *emptyDirPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, opts volume.VolumeOptions) (volume.Mounter, error) {
 	return plugin.newMounterInternal(spec, pod, plugin.host.GetMounter(), &realMountDetector{plugin.host.GetMounter()}, opts)
 }

pkg/volume/fc/fc.go

@@ -78,6 +78,10 @@ func (plugin *fcPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *fcPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *fcPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/flexvolume/plugin.go

@@ -166,6 +166,10 @@ func (plugin *flexVolumePlugin) ConstructVolumeSpec(volumeName, mountPath string
 	return volume.NewSpecFromVolume(flexVolume), nil
 }
 
+func (plugin *flexVolumePlugin) SupportsMountOption() bool {
+	return false
+}
+
 // Mark the given commands as unsupported.
 func (plugin *flexVolumePlugin) unsupported(commands ...string) {
 	plugin.Lock()

pkg/volume/flocker/flocker.go

@@ -112,6 +112,10 @@ func (p *flockerPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (p *flockerPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *flockerPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/gce_pd/attacher.go

@@ -209,7 +209,8 @@ func (attacher *gcePersistentDiskAttacher) MountDevice(spec *volume.Spec, device
 	}
 	if notMnt {
 		diskMounter := &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()}
-		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, options)
+		mountOptions := volume.MountOptionFromSpec(spec, options...)
+		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, mountOptions)
 		if err != nil {
 			os.Remove(deviceMountPath)
 			return err

pkg/volume/gce_pd/gce_pd.go

@@ -82,6 +82,10 @@ func (plugin *gcePersistentDiskPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *gcePersistentDiskPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *gcePersistentDiskPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/git_repo/git_repo.go

@@ -81,6 +81,10 @@ func (plugin *gitRepoPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *gitRepoPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *gitRepoPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, opts volume.VolumeOptions) (volume.Mounter, error) {
 	return &gitRepoVolumeMounter{
 		gitRepoVolume: &gitRepoVolume{

pkg/volume/glusterfs/glusterfs.go

@@ -116,6 +116,10 @@ func (plugin *glusterfsPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *glusterfsPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *glusterfsPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -161,10 +165,12 @@ func (plugin *glusterfsPlugin) newMounterInternal(spec *volume.Spec, ep *v1.Endp
 			pod:     pod,
 			plugin:  plugin,
 		},
-		hosts:    ep,
+		hosts:        ep,
-		path:     source.Path,
+		path:         source.Path,
-		readOnly: readOnly,
+		readOnly:     readOnly,
-		exe:      exe}, nil
+		exe:          exe,
+		mountOptions: volume.MountOptionFromSpec(spec),
+	}, nil
 }
 
 func (plugin *glusterfsPlugin) NewUnmounter(volName string, podUID types.UID) (volume.Unmounter, error) {
@@ -209,10 +215,11 @@ type glusterfs struct {
 
 type glusterfsMounter struct {
 	*glusterfs
-	hosts    *v1.Endpoints
+	hosts        *v1.Endpoints
-	path     string
+	path         string
-	readOnly bool
+	readOnly     bool
-	exe      exec.Interface
+	exe          exec.Interface
+	mountOptions []string
 }
 
 var _ volume.Mounter = &glusterfsMounter{}
@@ -322,7 +329,8 @@ func (b *glusterfsMounter) setUpAtInternal(dir string) error {
 		// Avoid mount storm, pick a host randomly.
 		// Iterate all hosts until mount succeeds.
 		for _, ip := range addrlist {
-			errs = b.mounter.Mount(ip+":"+b.path, dir, "glusterfs", options)
+			mountOptions := volume.JoinMountOptions(b.mountOptions, options)
+			errs = b.mounter.Mount(ip+":"+b.path, dir, "glusterfs", mountOptions)
 			if errs == nil {
 				glog.Infof("glusterfs: successfully mounted %s", dir)
 				return nil

pkg/volume/host_path/host_path.go

@@ -83,6 +83,10 @@ func (plugin *hostPathPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *hostPathPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *hostPathPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/iscsi/disk_manager.go

@@ -60,7 +60,8 @@ func diskSetUp(manager diskManager, b iscsiDiskMounter, volPath string, mounter
 	if b.readOnly {
 		options = append(options, "ro")
 	}
-	err = mounter.Mount(globalPDPath, volPath, "", options)
+	mountOptions := volume.JoinMountOptions(b.mountOptions, options)
+	err = mounter.Mount(globalPDPath, volPath, "", mountOptions)
 	if err != nil {
 		glog.Errorf("failed to bind mount:%s", globalPDPath)
 		return err

pkg/volume/iscsi/iscsi.go

@@ -82,6 +82,10 @@ func (plugin *iscsiPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *iscsiPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *iscsiPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -121,10 +125,11 @@ func (plugin *iscsiPlugin) newMounterInternal(spec *volume.Spec, podUID types.UI
 			iface:   iface,
 			manager: manager,
 			plugin:  plugin},
-		fsType:     iscsi.FSType,
+		fsType:       iscsi.FSType,
-		readOnly:   readOnly,
+		readOnly:     readOnly,
-		mounter:    &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()},
+		mounter:      &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()},
-		deviceUtil: ioutil.NewDeviceHandler(ioutil.NewIOHandler()),
+		deviceUtil:   ioutil.NewDeviceHandler(ioutil.NewIOHandler()),
+		mountOptions: volume.MountOptionFromSpec(spec),
 	}, nil
 }
 
@@ -184,10 +189,11 @@ func (iscsi *iscsiDisk) GetPath() string {
 
 type iscsiDiskMounter struct {
 	*iscsiDisk
-	readOnly   bool
+	readOnly     bool
-	fsType     string
+	fsType       string
-	mounter    *mount.SafeFormatAndMount
+	mounter      *mount.SafeFormatAndMount
-	deviceUtil ioutil.DeviceUtil
+	deviceUtil   ioutil.DeviceUtil
+	mountOptions []string
 }
 
 var _ volume.Mounter = &iscsiDiskMounter{}

pkg/volume/nfs/nfs.go

@@ -88,6 +88,10 @@ func (plugin *nfsPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *nfsPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *nfsPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -113,9 +117,10 @@ func (plugin *nfsPlugin) newMounterInternal(spec *volume.Spec, pod *v1.Pod, moun
 			pod:     pod,
 			plugin:  plugin,
 		},
-		server:     source.Server,
+		server:       source.Server,
-		exportPath: source.Path,
+		exportPath:   source.Path,
-		readOnly:   readOnly,
+		readOnly:     readOnly,
+		mountOptions: volume.MountOptionFromSpec(spec),
 	}, nil
 }
 
@@ -207,9 +212,10 @@ func (nfsMounter *nfsMounter) CanMount() error {
 
 type nfsMounter struct {
 	*nfs
-	server     string
+	server       string
-	exportPath string
+	exportPath   string
-	readOnly   bool
+	readOnly     bool
+	mountOptions []string
 }
 
 var _ volume.Mounter = &nfsMounter{}
@@ -242,7 +248,8 @@ func (b *nfsMounter) SetUpAt(dir string, fsGroup *int64) error {
 	if b.readOnly {
 		options = append(options, "ro")
 	}
-	err = b.mounter.Mount(source, dir, "nfs", options)
+	mountOptions := volume.JoinMountOptions(b.mountOptions, options)
+	err = b.mounter.Mount(source, dir, "nfs", mountOptions)
 	if err != nil {
 		notMnt, mntErr := b.mounter.IsLikelyNotMountPoint(dir)
 		if mntErr != nil {

pkg/volume/photon_pd/attacher.go

@@ -203,7 +203,8 @@ func (attacher *photonPersistentDiskAttacher) MountDevice(spec *volume.Spec, dev
 
 	if notMnt {
 		diskMounter := &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()}
-		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, options)
+		mountOptions := volume.MountOptionFromSpec(spec)
+		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, mountOptions)
 		if err != nil {
 			os.Remove(deviceMountPath)
 			return err

pkg/volume/photon_pd/photon_pd.go

@@ -79,6 +79,10 @@ func (plugin *photonPersistentDiskPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *photonPersistentDiskPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *photonPersistentDiskPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, _ volume.VolumeOptions) (volume.Mounter, error) {
 	return plugin.newMounterInternal(spec, pod.UID, &PhotonDiskUtil{}, plugin.host.GetMounter())
 }

pkg/volume/plugins.go

@@ -107,6 +107,11 @@ type VolumePlugin interface {
 	// information from input. This function is used by volume manager to reconstruct
 	// volume spec by reading the volume directories from disk
 	ConstructVolumeSpec(volumeName, mountPath string) (*Spec, error)
+
+	// SupportsMountOption returns true if volume plugins supports Mount options
+	// Specifying mount options in a volume plugin that doesn't support
+	// user specified mount options will result in error creating persistent volumes
+	SupportsMountOption() bool
 }
 
 // PersistentVolumePlugin is an extended interface of VolumePlugin and is used
@@ -146,6 +151,8 @@ const (
 	// Name of a volume in external cloud that is being provisioned and thus
 	// should be ignored by rest of Kubernetes.
 	ProvisionedVolumeName = "placeholder-for-provisioning"
+	// Mount options annotations
+	MountOptionAnnotation = "volume.beta.kubernetes.io/mount-options"
 )
 
 // ProvisionableVolumePlugin is an extended interface of VolumePlugin and is

pkg/volume/plugins_test.go

@@ -77,6 +77,10 @@ func (plugin *testPlugins) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *testPlugins) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *testPlugins) NewMounter(spec *Spec, podRef *v1.Pod, opts VolumeOptions) (Mounter, error) {
 	return nil, nil
 }

pkg/volume/portworx/portworx.go

@@ -175,6 +175,10 @@ func (plugin *portworxVolumePlugin) ConstructVolumeSpec(volumeName, mountPath st
 	return volume.NewSpecFromVolume(portworxVolume), nil
 }
 
+func (plugin *portworxVolumePlugin) SupportsMountOption() bool {
+	return false
+}
+
 func getVolumeSource(
 	spec *volume.Spec) (*v1.PortworxVolumeSource, bool, error) {
 	if spec.Volume != nil && spec.Volume.PortworxVolume != nil {

pkg/volume/projected/projected.go

@@ -92,6 +92,10 @@ func (plugin *projectedPlugin) RequiresRemount() bool {
 	return true
 }
 
+func (plugin *projectedPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *projectedPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, opts volume.VolumeOptions) (volume.Mounter, error) {
 	return &projectedVolumeMounter{
 		projectedVolume: &projectedVolume{

pkg/volume/quobyte/quobyte.go

@@ -118,6 +118,10 @@ func (plugin *quobytePlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *quobytePlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *quobytePlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -169,8 +173,9 @@ func (plugin *quobytePlugin) newMounterInternal(spec *volume.Spec, pod *v1.Pod,
 			volume:  source.Volume,
 			plugin:  plugin,
 		},
-		registry: source.Registry,
+		registry:     source.Registry,
-		readOnly: readOnly,
+		readOnly:     readOnly,
+		mountOptions: volume.MountOptionFromSpec(spec),
 	}, nil
 }
 
@@ -205,8 +210,9 @@ type quobyte struct {
 
 type quobyteMounter struct {
 	*quobyte
-	registry string
+	registry     string
-	readOnly bool
+	readOnly     bool
+	mountOptions []string
 }
 
 var _ volume.Mounter = &quobyteMounter{}
@@ -249,7 +255,8 @@ func (mounter *quobyteMounter) SetUpAt(dir string, fsGroup *int64) error {
 	}
 
 	//if a trailing slash is missing we add it here
-	if err := mounter.mounter.Mount(mounter.correctTraillingSlash(mounter.registry), dir, "quobyte", options); err != nil {
+	mountOptions := volume.JoinMountOptions(mounter.mountOptions, options)
+	if err := mounter.mounter.Mount(mounter.correctTraillingSlash(mounter.registry), dir, "quobyte", mountOptions); err != nil {
 		return fmt.Errorf("quobyte: mount failed: %v", err)
 	}
 

pkg/volume/rbd/disk_manager.go

@@ -71,7 +71,8 @@ func diskSetUp(manager diskManager, b rbdMounter, volPath string, mounter mount.
 	if (&b).GetAttributes().ReadOnly {
 		options = append(options, "ro")
 	}
-	err = mounter.Mount(globalPDPath, volPath, "", options)
+	mountOptions := volume.JoinMountOptions(b.mountOptions, options)
+	err = mounter.Mount(globalPDPath, volPath, "", mountOptions)
 	if err != nil {
 		glog.Errorf("failed to bind mount:%s", globalPDPath)
 		return err

pkg/volume/rbd/rbd.go

@@ -86,6 +86,10 @@ func (plugin *rbdPlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *rbdPlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *rbdPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -136,11 +140,12 @@ func (plugin *rbdPlugin) newMounterInternal(spec *volume.Spec, podUID types.UID,
 			mounter:  &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()},
 			plugin:   plugin,
 		},
-		Mon:     source.CephMonitors,
+		Mon:          source.CephMonitors,
-		Id:      id,
+		Id:           id,
-		Keyring: keyring,
+		Keyring:      keyring,
-		Secret:  secret,
+		Secret:       secret,
-		fsType:  source.FSType,
+		fsType:       source.FSType,
+		mountOptions: volume.MountOptionFromSpec(spec),
 	}, nil
 }
 
@@ -360,13 +365,14 @@ func (rbd *rbd) GetPath() string {
 type rbdMounter struct {
 	*rbd
 	// capitalized so they can be exported in persistRBD()
-	Mon         []string
+	Mon          []string
-	Id          string
+	Id           string
-	Keyring     string
+	Keyring      string
-	Secret      string
+	Secret       string
-	fsType      string
+	fsType       string
-	adminSecret string
+	adminSecret  string
-	adminId     string
+	adminId      string
+	mountOptions []string
 }
 
 var _ volume.Mounter = &rbdMounter{}

pkg/volume/secret/secret.go

@@ -85,6 +85,10 @@ func (plugin *secretPlugin) RequiresRemount() bool {
 	return true
 }
 
+func (plugin *secretPlugin) SupportsMountOption() bool {
+	return false
+}
+
 func (plugin *secretPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, opts volume.VolumeOptions) (volume.Mounter, error) {
 	return &secretVolumeMounter{
 		secretVolume: &secretVolume{

pkg/volume/testing/testing.go

@@ -203,6 +203,10 @@ func (plugin *FakeVolumePlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *FakeVolumePlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *FakeVolumePlugin) NewMounter(spec *Spec, pod *v1.Pod, opts VolumeOptions) (Mounter, error) {
 	plugin.Lock()
 	defer plugin.Unlock()

pkg/volume/util.go

@@ -23,6 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/v1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 
@@ -372,3 +373,43 @@ func UnmountViaEmptyDir(dir string, host VolumeHost, volName string, volSpec Spe
 	}
 	return wrapped.TearDownAt(dir)
 }
+
+// MountOptionFromSpec extracts and joins mount options from volume spec with supplied options
+func MountOptionFromSpec(spec *Spec, options ...string) []string {
+	pv := spec.PersistentVolume
+
+	if pv != nil {
+		if mo, ok := pv.Annotations[MountOptionAnnotation]; ok {
+			moList := strings.Split(mo, ",")
+			return JoinMountOptions(moList, options)
+		}
+
+	}
+	return options
+}
+
+// MountOptionFromApiPV extracts mount options from api.PersistentVolume
+func MountOptionFromApiPV(pv *api.PersistentVolume) []string {
+	mountOptions := []string{}
+	if mo, ok := pv.Annotations[MountOptionAnnotation]; ok {
+		moList := strings.Split(mo, ",")
+		return JoinMountOptions(moList, mountOptions)
+	}
+	return mountOptions
+}
+
+// JoinMountOptions joins mount options eliminating duplicates
+func JoinMountOptions(userOptions []string, systemOptions []string) []string {
+	allMountOptions := sets.NewString()
+
+	for _, mountOption := range userOptions {
+		if len(mountOption) > 0 {
+			allMountOptions.Insert(mountOption)
+		}
+	}
+
+	for _, mountOption := range systemOptions {
+		allMountOptions.Insert(mountOption)
+	}
+	return allMountOptions.UnsortedList()
+}

pkg/volume/util/operationexecutor/operation_generator.go

@@ -383,6 +383,12 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 			newMounterErr)
 	}
 
+	mountCheckError := checkMountOptionSupport(og, volumeToMount, volumePlugin)
+
+	if mountCheckError != nil {
+		return nil, mountCheckError
+	}
+
 	// Get attacher, if possible
 	attachableVolumePlugin, _ :=
 		og.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec)
@@ -867,3 +873,20 @@ func (og *operationGenerator) verifyVolumeIsSafeToDetach(
 		volumeToDetach.NodeName)
 	return nil
 }
+
+func checkMountOptionSupport(og *operationGenerator, volumeToMount VolumeToMount, plugin volume.VolumePlugin) error {
+	mountOptions := volume.MountOptionFromSpec(volumeToMount.VolumeSpec)
+
+	if len(mountOptions) > 0 && !plugin.SupportsMountOption() {
+		err := fmt.Errorf(
+			"MountVolume.checkMountOptionSupport failed for volume %q (spec.Name: %q) pod %q (UID: %q) with: %q",
+			volumeToMount.VolumeName,
+			volumeToMount.VolumeSpec.Name(),
+			volumeToMount.PodName,
+			volumeToMount.Pod.UID,
+			"Mount options are not supported for this volume type")
+		og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeWarning, kevents.UnsupportedMountOption, err.Error())
+		return err
+	}
+	return nil
+}

pkg/volume/util_test.go

@@ -19,6 +19,7 @@ package volume
 import (
 	"fmt"
 	"hash/fnv"
+	"reflect"
 	"strings"
 	"testing"
 
@@ -29,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/v1"
+	"k8s.io/kubernetes/pkg/util/slice"
 )
 
 type testcase struct {
@@ -304,7 +306,74 @@ func TestGenerateVolumeName(t *testing.T) {
 	if v3 != expect {
 		t.Errorf("Expected %s, got %s", expect, v3)
 	}
+}
 
+func TestMountOptionFromSpec(t *testing.T) {
+	scenarios := map[string]struct {
+		volume            *Spec
+		expectedMountList []string
+		systemOptions     []string
+	}{
+		"volume-with-mount-options": {
+			volume: createVolumeSpecWithMountOption("good-mount-opts", "ro,nfsvers=3", v1.PersistentVolumeSpec{
+				PersistentVolumeSource: v1.PersistentVolumeSource{
+					NFS: &v1.NFSVolumeSource{Server: "localhost", Path: "/srv", ReadOnly: false},
+				},
+			}),
+			expectedMountList: []string{"ro", "nfsvers=3"},
+			systemOptions:     nil,
+		},
+		"volume-with-bad-mount-options": {
+			volume: createVolumeSpecWithMountOption("good-mount-opts", "", v1.PersistentVolumeSpec{
+				PersistentVolumeSource: v1.PersistentVolumeSource{
+					NFS: &v1.NFSVolumeSource{Server: "localhost", Path: "/srv", ReadOnly: false},
+				},
+			}),
+			expectedMountList: []string{},
+			systemOptions:     nil,
+		},
+		"vol-with-sys-opts": {
+			volume: createVolumeSpecWithMountOption("good-mount-opts", "ro,nfsvers=3", v1.PersistentVolumeSpec{
+				PersistentVolumeSource: v1.PersistentVolumeSource{
+					NFS: &v1.NFSVolumeSource{Server: "localhost", Path: "/srv", ReadOnly: false},
+				},
+			}),
+			expectedMountList: []string{"ro", "nfsvers=3", "fsid=100", "hard"},
+			systemOptions:     []string{"fsid=100", "hard"},
+		},
+		"vol-with-sys-opts-with-dup": {
+			volume: createVolumeSpecWithMountOption("good-mount-opts", "ro,nfsvers=3", v1.PersistentVolumeSpec{
+				PersistentVolumeSource: v1.PersistentVolumeSource{
+					NFS: &v1.NFSVolumeSource{Server: "localhost", Path: "/srv", ReadOnly: false},
+				},
+			}),
+			expectedMountList: []string{"ro", "nfsvers=3", "fsid=100"},
+			systemOptions:     []string{"fsid=100", "ro"},
+		},
+	}
+
+	for name, scenario := range scenarios {
+		mountOptions := MountOptionFromSpec(scenario.volume, scenario.systemOptions...)
+		if !reflect.DeepEqual(slice.SortStrings(mountOptions), slice.SortStrings(scenario.expectedMountList)) {
+			t.Errorf("for %s expected mount options : %v got %v", name, scenario.expectedMountList, mountOptions)
+		}
+	}
+}
+
+func createVolumeSpecWithMountOption(name string, mountOptions string, spec v1.PersistentVolumeSpec) *Spec {
+	annotations := map[string]string{
+		MountOptionAnnotation: mountOptions,
+	}
+	objMeta := metav1.ObjectMeta{
+		Name:        name,
+		Annotations: annotations,
+	}
+
+	pv := &v1.PersistentVolume{
+		ObjectMeta: objMeta,
+		Spec:       spec,
+	}
+	return &Spec{PersistentVolume: pv}
 }
 
 func checkFnv32(t *testing.T, s string, expected int) {

pkg/volume/vsphere_volume/attacher.go

@@ -195,7 +195,8 @@ func (attacher *vsphereVMDKAttacher) MountDevice(spec *volume.Spec, devicePath s
 
 	if notMnt {
 		diskMounter := &mount.SafeFormatAndMount{Interface: mounter, Runner: exec.New()}
-		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, options)
+		mountOptions := volume.MountOptionFromSpec(spec, options...)
+		err = diskMounter.FormatAndMount(devicePath, deviceMountPath, volumeSource.FSType, mountOptions)
 		if err != nil {
 			os.Remove(deviceMountPath)
 			return err

pkg/volume/vsphere_volume/vsphere_volume.go

@@ -80,6 +80,10 @@ func (plugin *vsphereVolumePlugin) RequiresRemount() bool {
 	return false
 }
 
+func (plugin *vsphereVolumePlugin) SupportsMountOption() bool {
+	return true
+}
+
 func (plugin *vsphereVolumePlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, _ volume.VolumeOptions) (volume.Mounter, error) {
 	return plugin.newMounterInternal(spec, pod.UID, &VsphereDiskUtil{}, plugin.host.GetMounter())
 }