Merge pull request #102028 from chrishenzie/read-write-once-pod-access-mode

ReadWriteOncePod access mode for PVs and PVCs

pkg/apis/core/helper/helpers.go

@@ -303,19 +303,22 @@ func IsStandardFinalizerName(str string) bool {
 }
 
 // GetAccessModesAsString returns a string representation of an array of access modes.
-// modes, when present, are always in the same order: RWO,ROX,RWX.
+// modes, when present, are always in the same order: RWO,ROX,RWX,RWOP.
 func GetAccessModesAsString(modes []core.PersistentVolumeAccessMode) string {
 	modes = removeDuplicateAccessModes(modes)
 	modesStr := []string{}
-	if containsAccessMode(modes, core.ReadWriteOnce) {
+	if ContainsAccessMode(modes, core.ReadWriteOnce) {
 		modesStr = append(modesStr, "RWO")
 	}
-	if containsAccessMode(modes, core.ReadOnlyMany) {
+	if ContainsAccessMode(modes, core.ReadOnlyMany) {
 		modesStr = append(modesStr, "ROX")
 	}
-	if containsAccessMode(modes, core.ReadWriteMany) {
+	if ContainsAccessMode(modes, core.ReadWriteMany) {
 		modesStr = append(modesStr, "RWX")
 	}
+	if ContainsAccessMode(modes, core.ReadWriteOncePod) {
+		modesStr = append(modesStr, "RWOP")
+	}
 	return strings.Join(modesStr, ",")
 }
 
@@ -332,6 +335,8 @@ func GetAccessModesFromString(modes string) []core.PersistentVolumeAccessMode {
 			accessModes = append(accessModes, core.ReadOnlyMany)
 		case s == "RWX":
 			accessModes = append(accessModes, core.ReadWriteMany)
+		case s == "RWOP":
+			accessModes = append(accessModes, core.ReadWriteOncePod)
 		}
 	}
 	return accessModes
@@ -341,14 +346,14 @@ func GetAccessModesFromString(modes string) []core.PersistentVolumeAccessMode {
 func removeDuplicateAccessModes(modes []core.PersistentVolumeAccessMode) []core.PersistentVolumeAccessMode {
 	accessModes := []core.PersistentVolumeAccessMode{}
 	for _, m := range modes {
-		if !containsAccessMode(accessModes, m) {
+		if !ContainsAccessMode(accessModes, m) {
 			accessModes = append(accessModes, m)
 		}
 	}
 	return accessModes
 }
 
-func containsAccessMode(modes []core.PersistentVolumeAccessMode, mode core.PersistentVolumeAccessMode) bool {
+func ContainsAccessMode(modes []core.PersistentVolumeAccessMode, mode core.PersistentVolumeAccessMode) bool {
 	for _, m := range modes {
 		if m == mode {
 			return true

pkg/apis/core/helper/helpers_test.go

@@ -87,25 +87,42 @@ func TestIsStandardContainerResource(t *testing.T) {
 
 func TestGetAccessModesFromString(t *testing.T) {
 	modes := GetAccessModesFromString("ROX")
-	if !containsAccessMode(modes, core.ReadOnlyMany) {
+	if !ContainsAccessMode(modes, core.ReadOnlyMany) {
 		t.Errorf("Expected mode %s, but got %+v", core.ReadOnlyMany, modes)
 	}
 
 	modes = GetAccessModesFromString("ROX,RWX")
-	if !containsAccessMode(modes, core.ReadOnlyMany) {
+	if !ContainsAccessMode(modes, core.ReadOnlyMany) {
 		t.Errorf("Expected mode %s, but got %+v", core.ReadOnlyMany, modes)
 	}
-	if !containsAccessMode(modes, core.ReadWriteMany) {
+	if !ContainsAccessMode(modes, core.ReadWriteMany) {
 		t.Errorf("Expected mode %s, but got %+v", core.ReadWriteMany, modes)
 	}
 
 	modes = GetAccessModesFromString("RWO,ROX,RWX")
-	if !containsAccessMode(modes, core.ReadOnlyMany) {
+	if !ContainsAccessMode(modes, core.ReadWriteOnce) {
+		t.Errorf("Expected mode %s, but got %+v", core.ReadWriteOnce, modes)
+	}
+	if !ContainsAccessMode(modes, core.ReadOnlyMany) {
 		t.Errorf("Expected mode %s, but got %+v", core.ReadOnlyMany, modes)
 	}
-	if !containsAccessMode(modes, core.ReadWriteMany) {
+	if !ContainsAccessMode(modes, core.ReadWriteMany) {
 		t.Errorf("Expected mode %s, but got %+v", core.ReadWriteMany, modes)
 	}
+
+	modes = GetAccessModesFromString("RWO,ROX,RWX,RWOP")
+	if !ContainsAccessMode(modes, core.ReadWriteOnce) {
+		t.Errorf("Expected mode %s, but got %+v", core.ReadWriteOnce, modes)
+	}
+	if !ContainsAccessMode(modes, core.ReadOnlyMany) {
+		t.Errorf("Expected mode %s, but got %+v", core.ReadOnlyMany, modes)
+	}
+	if !ContainsAccessMode(modes, core.ReadWriteMany) {
+		t.Errorf("Expected mode %s, but got %+v", core.ReadWriteMany, modes)
+	}
+	if !ContainsAccessMode(modes, core.ReadWriteOncePod) {
+		t.Errorf("Expected mode %s, but got %+v", core.ReadWriteOncePod, modes)
+	}
 }
 
 func TestRemoveDuplicateAccessModes(t *testing.T) {

pkg/apis/core/types.go

@@ -511,6 +511,9 @@ const (
 	ReadOnlyMany PersistentVolumeAccessMode = "ReadOnlyMany"
 	// can be mounted in read/write mode to many hosts
 	ReadWriteMany PersistentVolumeAccessMode = "ReadWriteMany"
+	// can be mounted read/write mode to exactly 1 pod
+	// cannot be used in combination with other access modes
+	ReadWriteOncePod PersistentVolumeAccessMode = "ReadWriteOncePod"
 )
 
 // PersistentVolumePhase defines the phase in which a PV is

pkg/apis/core/v1/helper/helpers.go

@@ -170,19 +170,22 @@ func ingressEqual(lhs, rhs *v1.LoadBalancerIngress) bool {
 }
 
 // GetAccessModesAsString returns a string representation of an array of access modes.
-// modes, when present, are always in the same order: RWO,ROX,RWX.
+// modes, when present, are always in the same order: RWO,ROX,RWX,RWOP.
 func GetAccessModesAsString(modes []v1.PersistentVolumeAccessMode) string {
 	modes = removeDuplicateAccessModes(modes)
 	modesStr := []string{}
-	if containsAccessMode(modes, v1.ReadWriteOnce) {
+	if ContainsAccessMode(modes, v1.ReadWriteOnce) {
 		modesStr = append(modesStr, "RWO")
 	}
-	if containsAccessMode(modes, v1.ReadOnlyMany) {
+	if ContainsAccessMode(modes, v1.ReadOnlyMany) {
 		modesStr = append(modesStr, "ROX")
 	}
-	if containsAccessMode(modes, v1.ReadWriteMany) {
+	if ContainsAccessMode(modes, v1.ReadWriteMany) {
 		modesStr = append(modesStr, "RWX")
 	}
+	if ContainsAccessMode(modes, v1.ReadWriteOncePod) {
+		modesStr = append(modesStr, "RWOP")
+	}
 	return strings.Join(modesStr, ",")
 }
 
@@ -199,6 +202,8 @@ func GetAccessModesFromString(modes string) []v1.PersistentVolumeAccessMode {
 			accessModes = append(accessModes, v1.ReadOnlyMany)
 		case s == "RWX":
 			accessModes = append(accessModes, v1.ReadWriteMany)
+		case s == "RWOP":
+			accessModes = append(accessModes, v1.ReadWriteOncePod)
 		}
 	}
 	return accessModes
@@ -208,14 +213,14 @@ func GetAccessModesFromString(modes string) []v1.PersistentVolumeAccessMode {
 func removeDuplicateAccessModes(modes []v1.PersistentVolumeAccessMode) []v1.PersistentVolumeAccessMode {
 	accessModes := []v1.PersistentVolumeAccessMode{}
 	for _, m := range modes {
-		if !containsAccessMode(accessModes, m) {
+		if !ContainsAccessMode(accessModes, m) {
 			accessModes = append(accessModes, m)
 		}
 	}
 	return accessModes
 }
 
-func containsAccessMode(modes []v1.PersistentVolumeAccessMode, mode v1.PersistentVolumeAccessMode) bool {
+func ContainsAccessMode(modes []v1.PersistentVolumeAccessMode, mode v1.PersistentVolumeAccessMode) bool {
 	for _, m := range modes {
 		if m == mode {
 			return true

pkg/apis/core/v1/helper/helpers_test.go

@@ -213,25 +213,42 @@ func TestIsOvercommitAllowed(t *testing.T) {
 
 func TestGetAccessModesFromString(t *testing.T) {
 	modes := GetAccessModesFromString("ROX")
-	if !containsAccessMode(modes, v1.ReadOnlyMany) {
+	if !ContainsAccessMode(modes, v1.ReadOnlyMany) {
 		t.Errorf("Expected mode %s, but got %+v", v1.ReadOnlyMany, modes)
 	}
 
 	modes = GetAccessModesFromString("ROX,RWX")
-	if !containsAccessMode(modes, v1.ReadOnlyMany) {
+	if !ContainsAccessMode(modes, v1.ReadOnlyMany) {
 		t.Errorf("Expected mode %s, but got %+v", v1.ReadOnlyMany, modes)
 	}
-	if !containsAccessMode(modes, v1.ReadWriteMany) {
+	if !ContainsAccessMode(modes, v1.ReadWriteMany) {
 		t.Errorf("Expected mode %s, but got %+v", v1.ReadWriteMany, modes)
 	}
 
 	modes = GetAccessModesFromString("RWO,ROX,RWX")
-	if !containsAccessMode(modes, v1.ReadOnlyMany) {
+	if !ContainsAccessMode(modes, v1.ReadWriteOnce) {
+		t.Errorf("Expected mode %s, but got %+v", v1.ReadWriteOnce, modes)
+	}
+	if !ContainsAccessMode(modes, v1.ReadOnlyMany) {
 		t.Errorf("Expected mode %s, but got %+v", v1.ReadOnlyMany, modes)
 	}
-	if !containsAccessMode(modes, v1.ReadWriteMany) {
+	if !ContainsAccessMode(modes, v1.ReadWriteMany) {
 		t.Errorf("Expected mode %s, but got %+v", v1.ReadWriteMany, modes)
 	}
+
+	modes = GetAccessModesFromString("RWO,ROX,RWX,RWOP")
+	if !ContainsAccessMode(modes, v1.ReadWriteOnce) {
+		t.Errorf("Expected mode %s, but got %+v", v1.ReadWriteOnce, modes)
+	}
+	if !ContainsAccessMode(modes, v1.ReadOnlyMany) {
+		t.Errorf("Expected mode %s, but got %+v", v1.ReadOnlyMany, modes)
+	}
+	if !ContainsAccessMode(modes, v1.ReadWriteMany) {
+		t.Errorf("Expected mode %s, but got %+v", v1.ReadWriteMany, modes)
+	}
+	if !ContainsAccessMode(modes, v1.ReadWriteOncePod) {
+		t.Errorf("Expected mode %s, but got %+v", v1.ReadWriteOncePod, modes)
+	}
 }
 
 func TestRemoveDuplicateAccessModes(t *testing.T) {

pkg/apis/core/validation/validation.go

@@ -1608,16 +1608,17 @@ func validateEphemeralVolumeSource(ephemeral *core.EphemeralVolumeSource, fldPat
 	if ephemeral.VolumeClaimTemplate == nil {
 		allErrs = append(allErrs, field.Required(fldPath.Child("volumeClaimTemplate"), ""))
 	} else {
-		allErrs = append(allErrs, ValidatePersistentVolumeClaimTemplate(ephemeral.VolumeClaimTemplate, fldPath.Child("volumeClaimTemplate"))...)
+		opts := ValidationOptionsForPersistentVolumeClaimTemplate(ephemeral.VolumeClaimTemplate, nil)
+		allErrs = append(allErrs, ValidatePersistentVolumeClaimTemplate(ephemeral.VolumeClaimTemplate, fldPath.Child("volumeClaimTemplate"), opts)...)
 	}
 	return allErrs
 }
 
 // ValidatePersistentVolumeClaimTemplate verifies that the embedded object meta and spec are valid.
 // Checking of the object data is very minimal because only labels and annotations are used.
-func ValidatePersistentVolumeClaimTemplate(claimTemplate *core.PersistentVolumeClaimTemplate, fldPath *field.Path) field.ErrorList {
+func ValidatePersistentVolumeClaimTemplate(claimTemplate *core.PersistentVolumeClaimTemplate, fldPath *field.Path, opts PersistentVolumeClaimSpecValidationOptions) field.ErrorList {
 	allErrs := validatePersistentVolumeClaimTemplateObjectMeta(&claimTemplate.ObjectMeta, fldPath.Child("metadata"))
-	allErrs = append(allErrs, ValidatePersistentVolumeClaimSpec(&claimTemplate.Spec, fldPath.Child("spec"))...)
+	allErrs = append(allErrs, ValidatePersistentVolumeClaimSpec(&claimTemplate.Spec, fldPath.Child("spec"), opts)...)
 	return allErrs
 }
 
@@ -1638,6 +1639,12 @@ var allowedPVCTemplateObjectMetaFields = map[string]bool{
 	"Labels":      true,
 }
 
+// PersistentVolumeSpecValidationOptions contains the different settings for PeristentVolume validation
+type PersistentVolumeSpecValidationOptions struct {
+	// Allow spec to contain the "ReadWiteOncePod" access mode
+	AllowReadWriteOncePod bool
+}
+
 // ValidatePersistentVolumeName checks that a name is appropriate for a
 // PersistentVolumeName object.
 var ValidatePersistentVolumeName = apimachineryvalidation.NameIsDNSSubdomain
@@ -1648,7 +1655,22 @@ var supportedReclaimPolicy = sets.NewString(string(core.PersistentVolumeReclaimD
 
 var supportedVolumeModes = sets.NewString(string(core.PersistentVolumeBlock), string(core.PersistentVolumeFilesystem))
 
-func ValidatePersistentVolumeSpec(pvSpec *core.PersistentVolumeSpec, pvName string, validateInlinePersistentVolumeSpec bool, fldPath *field.Path) field.ErrorList {
+func ValidationOptionsForPersistentVolume(pv, oldPv *core.PersistentVolume) PersistentVolumeSpecValidationOptions {
+	opts := PersistentVolumeSpecValidationOptions{
+		AllowReadWriteOncePod: utilfeature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
+	}
+	if oldPv == nil {
+		// If there's no old PV, use the options based solely on feature enablement
+		return opts
+	}
+	if helper.ContainsAccessMode(oldPv.Spec.AccessModes, core.ReadWriteOncePod) {
+		// If the old object allowed "ReadWriteOncePod", continue to allow it in the new object
+		opts.AllowReadWriteOncePod = true
+	}
+	return opts
+}
+
+func ValidatePersistentVolumeSpec(pvSpec *core.PersistentVolumeSpec, pvName string, validateInlinePersistentVolumeSpec bool, fldPath *field.Path, opts PersistentVolumeSpecValidationOptions) field.ErrorList {
 	allErrs := field.ErrorList{}
 
 	if validateInlinePersistentVolumeSpec {
@@ -1666,10 +1688,26 @@ func ValidatePersistentVolumeSpec(pvSpec *core.PersistentVolumeSpec, pvName stri
 	if len(pvSpec.AccessModes) == 0 {
 		allErrs = append(allErrs, field.Required(fldPath.Child("accessModes"), ""))
 	}
+
+	expandedSupportedAccessModes := sets.StringKeySet(supportedAccessModes)
+	if opts.AllowReadWriteOncePod {
+		expandedSupportedAccessModes.Insert(string(core.ReadWriteOncePod))
+	}
+
+	foundReadWriteOncePod, foundNonReadWriteOncePod := false, false
 	for _, mode := range pvSpec.AccessModes {
-		if !supportedAccessModes.Has(string(mode)) {
+		if !expandedSupportedAccessModes.Has(string(mode)) {
-			allErrs = append(allErrs, field.NotSupported(fldPath.Child("accessModes"), mode, supportedAccessModes.List()))
+			allErrs = append(allErrs, field.NotSupported(fldPath.Child("accessModes"), mode, expandedSupportedAccessModes.List()))
 		}
+
+		if mode == core.ReadWriteOncePod {
+			foundReadWriteOncePod = true
+		} else if supportedAccessModes.Has(string(mode)) {
+			foundNonReadWriteOncePod = true
+		}
+	}
+	if foundReadWriteOncePod && foundNonReadWriteOncePod {
+		allErrs = append(allErrs, field.Forbidden(fldPath.Child("accessModes"), "may not use ReadWriteOncePod with other access modes"))
 	}
 
 	if !validateInlinePersistentVolumeSpec {
@@ -1922,17 +1960,17 @@ func ValidatePersistentVolumeSpec(pvSpec *core.PersistentVolumeSpec, pvName stri
 	return allErrs
 }
 
-func ValidatePersistentVolume(pv *core.PersistentVolume) field.ErrorList {
+func ValidatePersistentVolume(pv *core.PersistentVolume, opts PersistentVolumeSpecValidationOptions) field.ErrorList {
 	metaPath := field.NewPath("metadata")
 	allErrs := ValidateObjectMeta(&pv.ObjectMeta, false, ValidatePersistentVolumeName, metaPath)
-	allErrs = append(allErrs, ValidatePersistentVolumeSpec(&pv.Spec, pv.ObjectMeta.Name, false, field.NewPath("spec"))...)
+	allErrs = append(allErrs, ValidatePersistentVolumeSpec(&pv.Spec, pv.ObjectMeta.Name, false, field.NewPath("spec"), opts)...)
 	return allErrs
 }
 
 // ValidatePersistentVolumeUpdate tests to see if the update is legal for an end user to make.
 // newPv is updated with fields that cannot be changed.
-func ValidatePersistentVolumeUpdate(newPv, oldPv *core.PersistentVolume) field.ErrorList {
+func ValidatePersistentVolumeUpdate(newPv, oldPv *core.PersistentVolume, opts PersistentVolumeSpecValidationOptions) field.ErrorList {
-	allErrs := ValidatePersistentVolume(newPv)
+	allErrs := ValidatePersistentVolume(newPv, opts)
 
 	// if oldPV does not have ControllerExpandSecretRef then allow it to be set
 	if (oldPv.Spec.CSI != nil && oldPv.Spec.CSI.ControllerExpandSecretRef == nil) &&
@@ -1965,15 +2003,51 @@ func ValidatePersistentVolumeStatusUpdate(newPv, oldPv *core.PersistentVolume) f
 	return allErrs
 }
 
+// PersistentVolumeClaimSpecValidationOptions contains the different settings for PersistentVolumeClaim validation
+type PersistentVolumeClaimSpecValidationOptions struct {
+	// Allow spec to contain the "ReadWiteOncePod" access mode
+	AllowReadWriteOncePod bool
+}
+
+func ValidationOptionsForPersistentVolumeClaim(pvc, oldPvc *core.PersistentVolumeClaim) PersistentVolumeClaimSpecValidationOptions {
+	opts := PersistentVolumeClaimSpecValidationOptions{
+		AllowReadWriteOncePod: utilfeature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
+	}
+	if oldPvc == nil {
+		// If there's no old PVC, use the options based solely on feature enablement
+		return opts
+	}
+	if helper.ContainsAccessMode(oldPvc.Spec.AccessModes, core.ReadWriteOncePod) {
+		// If the old object allowed "ReadWriteOncePod", continue to allow it in the new object
+		opts.AllowReadWriteOncePod = true
+	}
+	return opts
+}
+
+func ValidationOptionsForPersistentVolumeClaimTemplate(claimTemplate, oldClaimTemplate *core.PersistentVolumeClaimTemplate) PersistentVolumeClaimSpecValidationOptions {
+	opts := PersistentVolumeClaimSpecValidationOptions{
+		AllowReadWriteOncePod: utilfeature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
+	}
+	if oldClaimTemplate == nil {
+		// If there's no old PVC template, use the options based solely on feature enablement
+		return opts
+	}
+	if helper.ContainsAccessMode(oldClaimTemplate.Spec.AccessModes, core.ReadWriteOncePod) {
+		// If the old object allowed "ReadWriteOncePod", continue to allow it in the new object
+		opts.AllowReadWriteOncePod = true
+	}
+	return opts
+}
+
 // ValidatePersistentVolumeClaim validates a PersistentVolumeClaim
-func ValidatePersistentVolumeClaim(pvc *core.PersistentVolumeClaim) field.ErrorList {
+func ValidatePersistentVolumeClaim(pvc *core.PersistentVolumeClaim, opts PersistentVolumeClaimSpecValidationOptions) field.ErrorList {
 	allErrs := ValidateObjectMeta(&pvc.ObjectMeta, true, ValidatePersistentVolumeName, field.NewPath("metadata"))
-	allErrs = append(allErrs, ValidatePersistentVolumeClaimSpec(&pvc.Spec, field.NewPath("spec"))...)
+	allErrs = append(allErrs, ValidatePersistentVolumeClaimSpec(&pvc.Spec, field.NewPath("spec"), opts)...)
 	return allErrs
 }
 
 // ValidatePersistentVolumeClaimSpec validates a PersistentVolumeClaimSpec
-func ValidatePersistentVolumeClaimSpec(spec *core.PersistentVolumeClaimSpec, fldPath *field.Path) field.ErrorList {
+func ValidatePersistentVolumeClaimSpec(spec *core.PersistentVolumeClaimSpec, fldPath *field.Path, opts PersistentVolumeClaimSpecValidationOptions) field.ErrorList {
 	allErrs := field.ErrorList{}
 	if len(spec.AccessModes) == 0 {
 		allErrs = append(allErrs, field.Required(fldPath.Child("accessModes"), "at least 1 access mode is required"))
@@ -1981,11 +2055,28 @@ func ValidatePersistentVolumeClaimSpec(spec *core.PersistentVolumeClaimSpec, fld
 	if spec.Selector != nil {
 		allErrs = append(allErrs, unversionedvalidation.ValidateLabelSelector(spec.Selector, fldPath.Child("selector"))...)
 	}
+
+	expandedSupportedAccessModes := sets.StringKeySet(supportedAccessModes)
+	if opts.AllowReadWriteOncePod {
+		expandedSupportedAccessModes.Insert(string(core.ReadWriteOncePod))
+	}
+
+	foundReadWriteOncePod, foundNonReadWriteOncePod := false, false
 	for _, mode := range spec.AccessModes {
-		if mode != core.ReadWriteOnce && mode != core.ReadOnlyMany && mode != core.ReadWriteMany {
+		if !expandedSupportedAccessModes.Has(string(mode)) {
-			allErrs = append(allErrs, field.NotSupported(fldPath.Child("accessModes"), mode, supportedAccessModes.List()))
+			allErrs = append(allErrs, field.NotSupported(fldPath.Child("accessModes"), mode, expandedSupportedAccessModes.List()))
+		}
+
+		if mode == core.ReadWriteOncePod {
+			foundReadWriteOncePod = true
+		} else if supportedAccessModes.Has(string(mode)) {
+			foundNonReadWriteOncePod = true
 		}
 	}
+	if foundReadWriteOncePod && foundNonReadWriteOncePod {
+		allErrs = append(allErrs, field.Forbidden(fldPath.Child("accessModes"), "may not use ReadWriteOncePod with other access modes"))
+	}
+
 	storageValue, ok := spec.Resources.Requests[core.ResourceStorage]
 	if !ok {
 		allErrs = append(allErrs, field.Required(fldPath.Child("resources").Key(string(core.ResourceStorage)), ""))
@@ -2024,9 +2115,9 @@ func ValidatePersistentVolumeClaimSpec(spec *core.PersistentVolumeClaimSpec, fld
 }
 
 // ValidatePersistentVolumeClaimUpdate validates an update to a PersistentVolumeClaim
-func ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc *core.PersistentVolumeClaim) field.ErrorList {
+func ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc *core.PersistentVolumeClaim, opts PersistentVolumeClaimSpecValidationOptions) field.ErrorList {
 	allErrs := ValidateObjectMetaUpdate(&newPvc.ObjectMeta, &oldPvc.ObjectMeta, field.NewPath("metadata"))
-	allErrs = append(allErrs, ValidatePersistentVolumeClaim(newPvc)...)
+	allErrs = append(allErrs, ValidatePersistentVolumeClaim(newPvc, opts)...)
 	newPvcClone := newPvc.DeepCopy()
 	oldPvcClone := oldPvc.DeepCopy()
 

pkg/apis/core/validation/validation_test.go

@@ -68,8 +68,9 @@ func TestValidatePersistentVolumes(t *testing.T) {
 	validMode := core.PersistentVolumeFilesystem
 	invalidMode := core.PersistentVolumeMode("fakeVolumeMode")
 	scenarios := map[string]struct {
-		isExpectedFailure bool
+		isExpectedFailure      bool
-		volume            *core.PersistentVolume
+		enableReadWriteOncePod bool
+		volume                 *core.PersistentVolume
 	}{
 		"good-volume": {
 			isExpectedFailure: false,
@@ -211,6 +212,54 @@ func TestValidatePersistentVolumes(t *testing.T) {
 				VolumeMode: &invalidMode,
 			}),
 		},
+		"with-read-write-once-pod-feature-gate-enabled": {
+			isExpectedFailure:      false,
+			enableReadWriteOncePod: true,
+			volume: testVolume("foo", "", core.PersistentVolumeSpec{
+				Capacity: core.ResourceList{
+					core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{"ReadWriteOncePod"},
+				PersistentVolumeSource: core.PersistentVolumeSource{
+					HostPath: &core.HostPathVolumeSource{
+						Path: "/foo",
+						Type: newHostPathType(string(core.HostPathDirectory)),
+					},
+				},
+			}),
+		},
+		"with-read-write-once-pod-feature-gate-disabled": {
+			isExpectedFailure:      true,
+			enableReadWriteOncePod: false,
+			volume: testVolume("foo", "", core.PersistentVolumeSpec{
+				Capacity: core.ResourceList{
+					core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{"ReadWriteOncePod"},
+				PersistentVolumeSource: core.PersistentVolumeSource{
+					HostPath: &core.HostPathVolumeSource{
+						Path: "/foo",
+						Type: newHostPathType(string(core.HostPathDirectory)),
+					},
+				},
+			}),
+		},
+		"with-read-write-once-pod-and-others-feature-gate-enabled": {
+			isExpectedFailure:      true,
+			enableReadWriteOncePod: true,
+			volume: testVolume("foo", "", core.PersistentVolumeSpec{
+				Capacity: core.ResourceList{
+					core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+				},
+				AccessModes: []core.PersistentVolumeAccessMode{"ReadWriteOncePod", "ReadWriteMany"},
+				PersistentVolumeSource: core.PersistentVolumeSource{
+					HostPath: &core.HostPathVolumeSource{
+						Path: "/foo",
+						Type: newHostPathType(string(core.HostPathDirectory)),
+					},
+				},
+			}),
+		},
 		"unexpected-namespace": {
 			isExpectedFailure: true,
 			volume: testVolume("foo", "unexpected-namespace", core.PersistentVolumeSpec{
@@ -415,7 +464,10 @@ func TestValidatePersistentVolumes(t *testing.T) {
 
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
-			errs := ValidatePersistentVolume(scenario.volume)
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ReadWriteOncePod, scenario.enableReadWriteOncePod)()
+
+			opts := ValidationOptionsForPersistentVolume(scenario.volume, nil)
+			errs := ValidatePersistentVolume(scenario.volume, opts)
 			if len(errs) == 0 && scenario.isExpectedFailure {
 				t.Errorf("Unexpected success for scenario: %s", name)
 			}
@@ -558,7 +610,8 @@ func TestValidatePersistentVolumeSpec(t *testing.T) {
 		},
 	}
 	for name, scenario := range scenarios {
-		errs := ValidatePersistentVolumeSpec(scenario.pvSpec, "", scenario.isInlineSpec, field.NewPath("field"))
+		opts := PersistentVolumeSpecValidationOptions{}
+		errs := ValidatePersistentVolumeSpec(scenario.pvSpec, "", scenario.isInlineSpec, field.NewPath("field"), opts)
 		if len(errs) == 0 && scenario.isExpectedFailure {
 			t.Errorf("Unexpected success for scenario: %s", name)
 		}
@@ -655,7 +708,8 @@ func TestValidatePersistentVolumeSourceUpdate(t *testing.T) {
 		},
 	}
 	for name, scenario := range scenarios {
-		errs := ValidatePersistentVolumeUpdate(scenario.newVolume, scenario.oldVolume)
+		opts := ValidationOptionsForPersistentVolume(scenario.newVolume, scenario.oldVolume)
+		errs := ValidatePersistentVolumeUpdate(scenario.newVolume, scenario.oldVolume, opts)
 		if len(errs) == 0 && scenario.isExpectedFailure {
 			t.Errorf("Unexpected success for scenario: %s", name)
 		}
@@ -665,6 +719,85 @@ func TestValidatePersistentVolumeSourceUpdate(t *testing.T) {
 	}
 }
 
+func TestValidationOptionsForPersistentVolume(t *testing.T) {
+	tests := map[string]struct {
+		oldPv                  *core.PersistentVolume
+		enableReadWriteOncePod bool
+		expectValidationOpts   PersistentVolumeSpecValidationOptions
+	}{
+		"nil old pv": {
+			oldPv:                  nil,
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop allowed because feature enabled": {
+			oldPv:                  pvWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop not allowed because not used and feature disabled": {
+			oldPv:                  pvWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
+			enableReadWriteOncePod: false,
+			expectValidationOpts: PersistentVolumeSpecValidationOptions{
+				AllowReadWriteOncePod: false,
+			},
+		},
+		"rwop allowed because used and feature enabled": {
+			oldPv:                  pvWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop allowed because used and feature disabled": {
+			oldPv:                  pvWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
+			enableReadWriteOncePod: false,
+			expectValidationOpts: PersistentVolumeSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ReadWriteOncePod, tc.enableReadWriteOncePod)()
+
+			opts := ValidationOptionsForPersistentVolume(nil, tc.oldPv)
+			if opts != tc.expectValidationOpts {
+				t.Errorf("Expected opts: %+v, received: %+v", opts, tc.expectValidationOpts)
+			}
+		})
+	}
+}
+
+func pvWithAccessModes(accessModes []core.PersistentVolumeAccessMode) *core.PersistentVolume {
+	return &core.PersistentVolume{
+		Spec: core.PersistentVolumeSpec{
+			AccessModes: accessModes,
+		},
+	}
+}
+
+func pvcWithAccessModes(accessModes []core.PersistentVolumeAccessMode) *core.PersistentVolumeClaim {
+	return &core.PersistentVolumeClaim{
+		Spec: core.PersistentVolumeClaimSpec{
+			AccessModes: accessModes,
+		},
+	}
+}
+
+func pvcTemplateWithAccessModes(accessModes []core.PersistentVolumeAccessMode) *core.PersistentVolumeClaimTemplate {
+	return &core.PersistentVolumeClaimTemplate{
+		Spec: core.PersistentVolumeClaimSpec{
+			AccessModes: accessModes,
+		},
+	}
+}
+
 func getCSIVolumeWithSecret(pv *core.PersistentVolume, secret *core.SecretReference) *core.PersistentVolume {
 	pvCopy := pv.DeepCopy()
 	if secret != nil {
@@ -729,7 +862,8 @@ func TestValidateLocalVolumes(t *testing.T) {
 	}
 
 	for name, scenario := range scenarios {
-		errs := ValidatePersistentVolume(scenario.volume)
+		opts := ValidationOptionsForPersistentVolume(scenario.volume, nil)
+		errs := ValidatePersistentVolume(scenario.volume, opts)
 		if len(errs) == 0 && scenario.isExpectedFailure {
 			t.Errorf("Unexpected success for scenario: %s", name)
 		}
@@ -808,7 +942,8 @@ func TestValidateVolumeNodeAffinityUpdate(t *testing.T) {
 	}
 
 	for name, scenario := range scenarios {
-		errs := ValidatePersistentVolumeUpdate(scenario.newPV, scenario.oldPV)
+		opts := ValidationOptionsForPersistentVolume(scenario.newPV, scenario.oldPV)
+		errs := ValidatePersistentVolumeUpdate(scenario.newPV, scenario.oldPV, opts)
 		if len(errs) == 0 && scenario.isExpectedFailure {
 			t.Errorf("Unexpected success for scenario: %s", name)
 		}
@@ -909,12 +1044,14 @@ func TestAlphaVolumeSnapshotDataSource(t *testing.T) {
 	}
 
 	for _, tc := range successTestCases {
-		if errs := ValidatePersistentVolumeClaimSpec(&tc, field.NewPath("spec")); len(errs) != 0 {
+		opts := PersistentVolumeClaimSpecValidationOptions{}
+		if errs := ValidatePersistentVolumeClaimSpec(&tc, field.NewPath("spec"), opts); len(errs) != 0 {
 			t.Errorf("expected success: %v", errs)
 		}
 	}
 	for _, tc := range failedTestCases {
-		if errs := ValidatePersistentVolumeClaimSpec(&tc, field.NewPath("spec")); len(errs) == 0 {
+		opts := PersistentVolumeClaimSpecValidationOptions{}
+		if errs := ValidatePersistentVolumeClaimSpec(&tc, field.NewPath("spec"), opts); len(errs) == 0 {
 			t.Errorf("expected failure: %v", errs)
 		}
 	}
@@ -969,8 +1106,9 @@ func testValidatePVC(t *testing.T, ephemeral bool) {
 	ten := int64(10)
 
 	scenarios := map[string]struct {
-		isExpectedFailure bool
+		isExpectedFailure      bool
-		claim             *core.PersistentVolumeClaim
+		enableReadWriteOncePod bool
+		claim                  *core.PersistentVolumeClaim
 	}{
 		"good-claim": {
 			isExpectedFailure: false,
@@ -1118,6 +1256,42 @@ func testValidatePVC(t *testing.T, ephemeral bool) {
 				return claim
 			}(),
 		},
+		"with-read-write-once-pod-feature-gate-enabled": {
+			isExpectedFailure:      false,
+			enableReadWriteOncePod: true,
+			claim: testVolumeClaim(goodName, goodNS, core.PersistentVolumeClaimSpec{
+				AccessModes: []core.PersistentVolumeAccessMode{"ReadWriteOncePod"},
+				Resources: core.ResourceRequirements{
+					Requests: core.ResourceList{
+						core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+					},
+				},
+			}),
+		},
+		"with-read-write-once-pod-feature-gate-disabled": {
+			isExpectedFailure:      true,
+			enableReadWriteOncePod: false,
+			claim: testVolumeClaim(goodName, goodNS, core.PersistentVolumeClaimSpec{
+				AccessModes: []core.PersistentVolumeAccessMode{"ReadWriteOncePod"},
+				Resources: core.ResourceRequirements{
+					Requests: core.ResourceList{
+						core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+					},
+				},
+			}),
+		},
+		"with-read-write-once-pod-and-others-feature-gate-enabled": {
+			isExpectedFailure:      true,
+			enableReadWriteOncePod: true,
+			claim: testVolumeClaim(goodName, goodNS, core.PersistentVolumeClaimSpec{
+				AccessModes: []core.PersistentVolumeAccessMode{"ReadWriteOncePod", "ReadWriteMany"},
+				Resources: core.ResourceRequirements{
+					Requests: core.ResourceList{
+						core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+					},
+				},
+			}),
+		},
 		"invalid-claim-zero-capacity": {
 			isExpectedFailure: true,
 			claim: testVolumeClaim(goodName, goodNS, core.PersistentVolumeClaimSpec{
@@ -1292,6 +1466,8 @@ func testValidatePVC(t *testing.T, ephemeral bool) {
 
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ReadWriteOncePod, scenario.enableReadWriteOncePod)()
+
 			var errs field.ErrorList
 			if ephemeral {
 				volumes := []core.Volume{
@@ -1310,7 +1486,8 @@ func testValidatePVC(t *testing.T, ephemeral bool) {
 				opts := PodValidationOptions{}
 				_, errs = ValidateVolumes(volumes, nil, field.NewPath(""), opts)
 			} else {
-				errs = ValidatePersistentVolumeClaim(scenario.claim)
+				opts := ValidationOptionsForPersistentVolumeClaim(scenario.claim, nil)
+				errs = ValidatePersistentVolumeClaim(scenario.claim, opts)
 			}
 			if len(errs) == 0 && scenario.isExpectedFailure {
 				t.Error("Unexpected success for scenario")
@@ -1393,8 +1570,9 @@ func TestAlphaPVVolumeModeUpdate(t *testing.T) {
 
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
+			opts := ValidationOptionsForPersistentVolume(scenario.newPV, scenario.oldPV)
 			// ensure we have a resource version specified for updates
-			errs := ValidatePersistentVolumeUpdate(scenario.newPV, scenario.oldPV)
+			errs := ValidatePersistentVolumeUpdate(scenario.newPV, scenario.oldPV, opts)
 			if len(errs) == 0 && scenario.isExpectedFailure {
 				t.Errorf("Unexpected success for scenario: %s", name)
 			}
@@ -1636,6 +1814,30 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 			},
 		})
 
+	validClaimRWOPAccessMode := testVolumeClaim("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOncePod,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+		VolumeName: "volume",
+	})
+
+	validClaimRWOPAccessModeAddAnnotation := testVolumeClaimAnnotation("foo", "ns", "description", "updated-or-added-foo-description", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOncePod,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+		VolumeName: "volume",
+	})
+
 	scenarios := map[string]struct {
 		isExpectedFailure bool
 		oldClaim          *core.PersistentVolumeClaim
@@ -1804,6 +2006,12 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 			newClaim:          validClaimStorageClass,
 			enableResize:      false,
 		},
+		"valid-update-rwop-used-and-rwop-feature-disabled": {
+			isExpectedFailure: false,
+			oldClaim:          validClaimRWOPAccessMode,
+			newClaim:          validClaimRWOPAccessModeAddAnnotation,
+			enableResize:      false,
+		},
 	}
 
 	for name, scenario := range scenarios {
@@ -1812,7 +2020,8 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ExpandPersistentVolumes, scenario.enableResize)()
 			scenario.oldClaim.ResourceVersion = "1"
 			scenario.newClaim.ResourceVersion = "1"
-			errs := ValidatePersistentVolumeClaimUpdate(scenario.newClaim, scenario.oldClaim)
+			opts := ValidationOptionsForPersistentVolumeClaim(scenario.newClaim, scenario.oldClaim)
+			errs := ValidatePersistentVolumeClaimUpdate(scenario.newClaim, scenario.oldClaim, opts)
 			if len(errs) == 0 && scenario.isExpectedFailure {
 				t.Errorf("Unexpected success for scenario: %s", name)
 			}
@@ -1823,6 +2032,116 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 	}
 }
 
+func TestValidationOptionsForPersistentVolumeClaim(t *testing.T) {
+	tests := map[string]struct {
+		oldPvc                 *core.PersistentVolumeClaim
+		enableReadWriteOncePod bool
+		expectValidationOpts   PersistentVolumeClaimSpecValidationOptions
+	}{
+		"nil pv": {
+			oldPvc:                 nil,
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop allowed because feature enabled": {
+			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop not allowed because not used and feature disabled": {
+			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
+			enableReadWriteOncePod: false,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: false,
+			},
+		},
+		"rwop allowed because used and feature enabled": {
+			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop allowed because used and feature disabled": {
+			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
+			enableReadWriteOncePod: false,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ReadWriteOncePod, tc.enableReadWriteOncePod)()
+
+			opts := ValidationOptionsForPersistentVolumeClaim(nil, tc.oldPvc)
+			if opts != tc.expectValidationOpts {
+				t.Errorf("Expected opts: %+v, received: %+v", opts, tc.expectValidationOpts)
+			}
+		})
+	}
+}
+
+func TestValidationOptionsForPersistentVolumeClaimTemplate(t *testing.T) {
+	tests := map[string]struct {
+		oldPvcTemplate         *core.PersistentVolumeClaimTemplate
+		enableReadWriteOncePod bool
+		expectValidationOpts   PersistentVolumeClaimSpecValidationOptions
+	}{
+		"nil pv": {
+			oldPvcTemplate:         nil,
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop allowed because feature enabled": {
+			oldPvcTemplate:         pvcTemplateWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop not allowed because not used and feature disabled": {
+			oldPvcTemplate:         pvcTemplateWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
+			enableReadWriteOncePod: false,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: false,
+			},
+		},
+		"rwop allowed because used and feature enabled": {
+			oldPvcTemplate:         pvcTemplateWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
+			enableReadWriteOncePod: true,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+		"rwop allowed because used and feature disabled": {
+			oldPvcTemplate:         pvcTemplateWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
+			enableReadWriteOncePod: false,
+			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
+				AllowReadWriteOncePod: true,
+			},
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ReadWriteOncePod, tc.enableReadWriteOncePod)()
+
+			opts := ValidationOptionsForPersistentVolumeClaimTemplate(nil, tc.oldPvcTemplate)
+			if opts != tc.expectValidationOpts {
+				t.Errorf("Expected opts: %+v, received: %+v", opts, tc.expectValidationOpts)
+			}
+		})
+	}
+}
+
 func TestValidateKeyToPath(t *testing.T) {
 	testCases := []struct {
 		kp      core.KeyToPath
@@ -4307,7 +4626,8 @@ func TestPVCVolumeMode(t *testing.T) {
 		"valid nil value":        createTestVolModePVC(nil),
 	}
 	for k, v := range successCasesPVC {
-		if errs := ValidatePersistentVolumeClaim(v); len(errs) != 0 {
+		opts := ValidationOptionsForPersistentVolumeClaim(v, nil)
+		if errs := ValidatePersistentVolumeClaim(v, opts); len(errs) != 0 {
 			t.Errorf("expected success for %s", k)
 		}
 	}
@@ -4318,7 +4638,8 @@ func TestPVCVolumeMode(t *testing.T) {
 		"empty value":   createTestVolModePVC(&empty),
 	}
 	for k, v := range errorCasesPVC {
-		if errs := ValidatePersistentVolumeClaim(v); len(errs) == 0 {
+		opts := ValidationOptionsForPersistentVolumeClaim(v, nil)
+		if errs := ValidatePersistentVolumeClaim(v, opts); len(errs) == 0 {
 			t.Errorf("expected failure for %s", k)
 		}
 	}
@@ -4337,7 +4658,8 @@ func TestPVVolumeMode(t *testing.T) {
 		"valid nil value":        createTestVolModePV(nil),
 	}
 	for k, v := range successCasesPV {
-		if errs := ValidatePersistentVolume(v); len(errs) != 0 {
+		opts := ValidationOptionsForPersistentVolume(v, nil)
+		if errs := ValidatePersistentVolume(v, opts); len(errs) != 0 {
 			t.Errorf("expected success for %s", k)
 		}
 	}
@@ -4348,7 +4670,8 @@ func TestPVVolumeMode(t *testing.T) {
 		"empty value":   createTestVolModePV(&empty),
 	}
 	for k, v := range errorCasesPV {
-		if errs := ValidatePersistentVolume(v); len(errs) == 0 {
+		opts := ValidationOptionsForPersistentVolume(v, nil)
+		if errs := ValidatePersistentVolume(v, opts); len(errs) == 0 {
 			t.Errorf("expected failure for %s", k)
 		}
 	}
@@ -16569,12 +16892,14 @@ func TestAlphaVolumePVCDataSource(t *testing.T) {
 
 	for _, tc := range testCases {
 		if tc.expectedFail {
-			if errs := ValidatePersistentVolumeClaimSpec(&tc.claimSpec, field.NewPath("spec")); len(errs) == 0 {
+			opts := PersistentVolumeClaimSpecValidationOptions{}
+			if errs := ValidatePersistentVolumeClaimSpec(&tc.claimSpec, field.NewPath("spec"), opts); len(errs) == 0 {
 				t.Errorf("expected failure: %v", errs)
 			}
 
 		} else {
-			if errs := ValidatePersistentVolumeClaimSpec(&tc.claimSpec, field.NewPath("spec")); len(errs) != 0 {
+			opts := PersistentVolumeClaimSpecValidationOptions{}
+			if errs := ValidatePersistentVolumeClaimSpec(&tc.claimSpec, field.NewPath("spec"), opts); len(errs) != 0 {
 				t.Errorf("expected success: %v", errs)
 			}
 		}

pkg/apis/storage/validation/validation.go

@@ -192,7 +192,8 @@ func validateVolumeAttachmentSource(source *storage.VolumeAttachmentSource, fldP
 			allErrs = append(allErrs, field.Required(fldPath.Child("persistentVolumeName"), "must specify non empty persistentVolumeName"))
 		}
 	case source.InlineVolumeSpec != nil:
-		allErrs = append(allErrs, apivalidation.ValidatePersistentVolumeSpec(source.InlineVolumeSpec, "", true, fldPath.Child("inlineVolumeSpec"))...)
+		opts := apivalidation.PersistentVolumeSpecValidationOptions{}
+		allErrs = append(allErrs, apivalidation.ValidatePersistentVolumeSpec(source.InlineVolumeSpec, "", true, fldPath.Child("inlineVolumeSpec"), opts)...)
 	}
 	return allErrs
 }

pkg/controller/volume/persistentvolume/index.go

@@ -152,7 +152,7 @@ func (pvIndex *persistentVolumeOrderedIndex) allPossibleMatchingAccessModes(requ
 	keys := pvIndex.store.ListIndexFuncValues("accessmodes")
 	for _, key := range keys {
 		indexedModes := v1helper.GetAccessModesFromString(key)
-		if volumeutil.AccessModesContainedInAll(indexedModes, requestedModes) {
+		if volumeutil.ContainsAllAccessModes(indexedModes, requestedModes) {
 			matchedModes = append(matchedModes, indexedModes)
 		}
 	}

pkg/controller/volume/persistentvolume/index_test.go

@@ -318,7 +318,7 @@ func TestAllPossibleAccessModes(t *testing.T) {
 		t.Errorf("Expected 3 arrays of modes that match RWO, but got %v", len(possibleModes))
 	}
 	for _, m := range possibleModes {
-		if !util.AccessModesContains(m, v1.ReadWriteOnce) {
+		if !util.ContainsAccessMode(m, v1.ReadWriteOnce) {
 			t.Errorf("AccessModes does not contain %s", v1.ReadWriteOnce)
 		}
 	}
@@ -327,7 +327,7 @@ func TestAllPossibleAccessModes(t *testing.T) {
 	if len(possibleModes) != 1 {
 		t.Errorf("Expected 1 array of modes that match RWX, but got %v", len(possibleModes))
 	}
-	if !util.AccessModesContains(possibleModes[0], v1.ReadWriteMany) {
+	if !util.ContainsAccessMode(possibleModes[0], v1.ReadWriteMany) {
 		t.Errorf("AccessModes does not contain %s", v1.ReadWriteOnce)
 	}
 

pkg/features/kube_features.go

@@ -727,6 +727,12 @@ const (
 	//
 	// Enables the PodSecurity admission plugin
 	PodSecurity featuregate.Feature = "PodSecurity"
+
+	// owner: @chrishenzie
+	// alpha: v1.22
+	//
+	// Enables usage of the ReadWriteOncePod PersistentVolume access mode.
+	ReadWriteOncePod featuregate.Feature = "ReadWriteOncePod"
 )
 
 func init() {
@@ -836,6 +842,7 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 	ExpandedDNSConfig:                              {Default: false, PreRelease: featuregate.Alpha},
 	SeccompDefault:                                 {Default: false, PreRelease: featuregate.Alpha},
 	PodSecurity:                                    {Default: false, PreRelease: featuregate.Alpha},
+	ReadWriteOncePod:                               {Default: false, PreRelease: featuregate.Alpha},
 
 	// inherited features from generic apiserver, relisted here to get a conflict if it is changed
 	// unintentionally on either side:

pkg/kubelet/volumemanager/cache/actual_state_of_world.go

@@ -425,6 +425,26 @@ func (asw *actualStateOfWorld) GetVolumeMountState(volumeName v1.UniqueVolumeNam
 	return podObj.volumeMountStateForPod
 }
 
+func (asw *actualStateOfWorld) IsVolumeMountedElsewhere(volumeName v1.UniqueVolumeName, podName volumetypes.UniquePodName) bool {
+	asw.RLock()
+	defer asw.RUnlock()
+
+	volumeObj, volumeExists := asw.attachedVolumes[volumeName]
+	if !volumeExists {
+		return false
+	}
+
+	for _, podObj := range volumeObj.mountedPods {
+		if podName != podObj.podName {
+			// Treat uncertain mount state as mounted until certain.
+			if podObj.volumeMountStateForPod != operationexecutor.VolumeNotMounted {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // addVolume adds the given volume to the cache indicating the specified
 // volume is attached to this node. If no volume name is supplied, a unique
 // volume name is generated from the volumeSpec and returned on success. If a

pkg/kubelet/volumemanager/cache/actual_state_of_world_test.go

@@ -241,6 +241,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeNewNode(t *testing.T) {
 	verifyVolumeDoesntExistInGloballyMountedVolumes(t, generatedVolumeName, asw)
 	verifyPodExistsInVolumeAsw(t, podName, generatedVolumeName, "fake/device/path" /* expectedDevicePath */, asw)
 	verifyVolumeExistsWithSpecNameInVolumeAsw(t, podName, volumeSpec.Name(), asw)
+	verifyVolumeMountedElsewhere(t, podName, generatedVolumeName, false /*expectedMountedElsewhere */, asw)
 }
 
 // Populates data struct with a volume
@@ -321,6 +322,7 @@ func Test_AddPodToVolume_Positive_ExistingVolumeExistingNode(t *testing.T) {
 	verifyVolumeDoesntExistInGloballyMountedVolumes(t, generatedVolumeName, asw)
 	verifyPodExistsInVolumeAsw(t, podName, generatedVolumeName, "fake/device/path" /* expectedDevicePath */, asw)
 	verifyVolumeExistsWithSpecNameInVolumeAsw(t, podName, volumeSpec.Name(), asw)
+	verifyVolumeMountedElsewhere(t, podName, generatedVolumeName, false /*expectedMountedElsewhere */, asw)
 }
 
 // Populates data struct with a volume
@@ -451,6 +453,8 @@ func Test_AddTwoPodsToVolume_Positive(t *testing.T) {
 	verifyVolumeExistsWithSpecNameInVolumeAsw(t, podName2, volumeSpec2.Name(), asw)
 	verifyVolumeSpecNameInVolumeAsw(t, podName1, []*volume.Spec{volumeSpec1}, asw)
 	verifyVolumeSpecNameInVolumeAsw(t, podName2, []*volume.Spec{volumeSpec2}, asw)
+	verifyVolumeMountedElsewhere(t, podName1, generatedVolumeName1, true /*expectedMountedElsewhere */, asw)
+	verifyVolumeMountedElsewhere(t, podName2, generatedVolumeName2, true /*expectedMountedElsewhere */, asw)
 }
 
 // Calls AddPodToVolume() to add pod to empty data struct
@@ -488,6 +492,10 @@ func Test_AddPodToVolume_Negative_VolumeDoesntExist(t *testing.T) {
 			err)
 	}
 
+	generatedVolumeName, err := util.GetUniqueVolumeNameFromSpec(
+		plugin, volumeSpec)
+	require.NoError(t, err)
+
 	blockplugin, err := volumePluginMgr.FindMapperPluginBySpec(volumeSpec)
 	if err != nil {
 		t.Fatalf(
@@ -538,6 +546,7 @@ func Test_AddPodToVolume_Negative_VolumeDoesntExist(t *testing.T) {
 		false, /* expectVolumeToExist */
 		asw)
 	verifyVolumeDoesntExistWithSpecNameInVolumeAsw(t, podName, volumeSpec.Name(), asw)
+	verifyVolumeMountedElsewhere(t, podName, generatedVolumeName, false /*expectedMountedElsewhere */, asw)
 }
 
 // Calls MarkVolumeAsAttached() once to add volume
@@ -773,6 +782,21 @@ func verifyPodExistsInVolumeAsw(
 	}
 }
 
+func verifyVolumeMountedElsewhere(
+	t *testing.T,
+	expectedPodName volumetypes.UniquePodName,
+	expectedVolumeName v1.UniqueVolumeName,
+	expectedMountedElsewhere bool,
+	asw ActualStateOfWorld) {
+	mountedElsewhere := asw.IsVolumeMountedElsewhere(expectedVolumeName, expectedPodName)
+	if mountedElsewhere != expectedMountedElsewhere {
+		t.Fatalf(
+			"IsVolumeMountedElsewhere assertion failure. Expected : <%t> Actual: <%t>",
+			expectedMountedElsewhere,
+			mountedElsewhere)
+	}
+}
+
 func verifyPodDoesntExistInVolumeAsw(
 	t *testing.T,
 	podToCheck volumetypes.UniquePodName,

pkg/registry/core/persistentvolume/strategy.go

@@ -71,7 +71,8 @@ func (persistentvolumeStrategy) PrepareForCreate(ctx context.Context, obj runtim
 
 func (persistentvolumeStrategy) Validate(ctx context.Context, obj runtime.Object) field.ErrorList {
 	persistentvolume := obj.(*api.PersistentVolume)
-	errorList := validation.ValidatePersistentVolume(persistentvolume)
+	opts := validation.ValidationOptionsForPersistentVolume(persistentvolume, nil)
+	errorList := validation.ValidatePersistentVolume(persistentvolume, opts)
 	return append(errorList, volumevalidation.ValidatePersistentVolume(persistentvolume)...)
 }
 
@@ -99,9 +100,11 @@ func (persistentvolumeStrategy) PrepareForUpdate(ctx context.Context, obj, old r
 
 func (persistentvolumeStrategy) ValidateUpdate(ctx context.Context, obj, old runtime.Object) field.ErrorList {
 	newPv := obj.(*api.PersistentVolume)
-	errorList := validation.ValidatePersistentVolume(newPv)
+	oldPv := old.(*api.PersistentVolume)
+	opts := validation.ValidationOptionsForPersistentVolume(newPv, oldPv)
+	errorList := validation.ValidatePersistentVolume(newPv, opts)
 	errorList = append(errorList, volumevalidation.ValidatePersistentVolume(newPv)...)
-	return append(errorList, validation.ValidatePersistentVolumeUpdate(newPv, old.(*api.PersistentVolume))...)
+	return append(errorList, validation.ValidatePersistentVolumeUpdate(newPv, oldPv, opts)...)
 }
 
 // WarningsOnUpdate returns warnings for the given update.

pkg/registry/core/persistentvolumeclaim/strategy.go

@@ -72,7 +72,8 @@ func (persistentvolumeclaimStrategy) PrepareForCreate(ctx context.Context, obj r
 
 func (persistentvolumeclaimStrategy) Validate(ctx context.Context, obj runtime.Object) field.ErrorList {
 	pvc := obj.(*api.PersistentVolumeClaim)
-	return validation.ValidatePersistentVolumeClaim(pvc)
+	opts := validation.ValidationOptionsForPersistentVolumeClaim(pvc, nil)
+	return validation.ValidatePersistentVolumeClaim(pvc, opts)
 }
 
 // WarningsOnCreate returns warnings for the creation of the given object.
@@ -98,8 +99,11 @@ func (persistentvolumeclaimStrategy) PrepareForUpdate(ctx context.Context, obj,
 }
 
 func (persistentvolumeclaimStrategy) ValidateUpdate(ctx context.Context, obj, old runtime.Object) field.ErrorList {
-	errorList := validation.ValidatePersistentVolumeClaim(obj.(*api.PersistentVolumeClaim))
+	newPvc := obj.(*api.PersistentVolumeClaim)
-	return append(errorList, validation.ValidatePersistentVolumeClaimUpdate(obj.(*api.PersistentVolumeClaim), old.(*api.PersistentVolumeClaim))...)
+	oldPvc := old.(*api.PersistentVolumeClaim)
+	opts := validation.ValidationOptionsForPersistentVolumeClaim(newPvc, oldPvc)
+	errorList := validation.ValidatePersistentVolumeClaim(newPvc, opts)
+	return append(errorList, validation.ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc, opts)...)
 }
 
 // WarningsOnUpdate returns warnings for the given update.

pkg/volume/awsebs/aws_ebs.go

@@ -485,7 +485,7 @@ type awsElasticBlockStoreProvisioner struct {
 var _ volume.Provisioner = &awsElasticBlockStoreProvisioner{}
 
 func (c *awsElasticBlockStoreProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}
 

pkg/volume/azure_file/azure_provision.go

@@ -149,7 +149,7 @@ type azureFileProvisioner struct {
 var _ volume.Provisioner = &azureFileProvisioner{}
 
 func (a *azureFileProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(a.plugin.GetAccessModes(), a.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(a.plugin.GetAccessModes(), a.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", a.options.PVC.Spec.AccessModes, a.plugin.GetAccessModes())
 	}
 	if util.CheckPersistentVolumeClaimModeBlock(a.options.PVC) {

pkg/volume/azuredd/azure_provision.go

@@ -185,7 +185,7 @@ func (p *azureDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologie
 	supportedModes := p.plugin.GetAccessModes()
 	if maxShares < 2 {
 		// only do AccessModes validation when maxShares < 2
-		if !util.AccessModesContainedInAll(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
+		if !util.ContainsAllAccessModes(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
 			return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported with maxShares(%d) < 2", p.options.PVC.Spec.AccessModes, p.plugin.GetAccessModes(), maxShares)
 		}
 

pkg/volume/cinder/cinder.go

@@ -561,7 +561,7 @@ type cinderVolumeProvisioner struct {
 var _ volume.Provisioner = &cinderVolumeProvisioner{}
 
 func (c *cinderVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}
 

pkg/volume/csi/csi_client.go

@@ -82,6 +82,7 @@ type csiClient interface {
 	NodeSupportsStageUnstage(ctx context.Context) (bool, error)
 	NodeSupportsNodeExpand(ctx context.Context) (bool, error)
 	NodeSupportsVolumeStats(ctx context.Context) (bool, error)
+	NodeSupportsSingleNodeMultiWriterAccessMode(ctx context.Context) (bool, error)
 }
 
 // Strongly typed address
@@ -120,6 +121,8 @@ type nodeV1ClientCreator func(addr csiAddr, metricsManager *MetricsManager) (
 	err error,
 )
 
+type nodeV1AccessModeMapper func(am api.PersistentVolumeAccessMode) csipbv1.VolumeCapability_AccessMode_Mode
+
 // newV1NodeClient creates a new NodeClient with the internally used gRPC
 // connection set up. It also returns a closer which must to be called to close
 // the gRPC connection when the NodeClient is not used anymore.
@@ -217,7 +220,11 @@ func (c *csiDriverClient) NodePublishVolume(
 
 	if c.nodeV1ClientCreator == nil {
 		return errors.New("failed to call NodePublishVolume. nodeV1ClientCreator is nil")
+	}
 
+	accessModeMapper, err := c.getNodeV1AccessModeMapper(ctx)
+	if err != nil {
+		return err
 	}
 
 	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr, c.metricsManager)
@@ -235,7 +242,7 @@ func (c *csiDriverClient) NodePublishVolume(
 		Secrets:        secrets,
 		VolumeCapability: &csipbv1.VolumeCapability{
 			AccessMode: &csipbv1.VolumeCapability_AccessMode{
-				Mode: asCSIAccessModeV1(accessMode),
+				Mode: accessModeMapper(accessMode),
 			},
 		},
 	}
@@ -279,6 +286,11 @@ func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResizeOp
 		return opts.newSize, errors.New("size can not be less than 0")
 	}
 
+	accessModeMapper, err := c.getNodeV1AccessModeMapper(ctx)
+	if err != nil {
+		return opts.newSize, err
+	}
+
 	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr, c.metricsManager)
 	if err != nil {
 		return opts.newSize, err
@@ -291,7 +303,7 @@ func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResizeOp
 		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: opts.newSize.Value()},
 		VolumeCapability: &csipbv1.VolumeCapability{
 			AccessMode: &csipbv1.VolumeCapability_AccessMode{
-				Mode: asCSIAccessModeV1(opts.accessMode),
+				Mode: accessModeMapper(opts.accessMode),
 			},
 		},
 	}
@@ -371,6 +383,11 @@ func (c *csiDriverClient) NodeStageVolume(ctx context.Context,
 		return errors.New("nodeV1ClientCreate is nil")
 	}
 
+	accessModeMapper, err := c.getNodeV1AccessModeMapper(ctx)
+	if err != nil {
+		return err
+	}
+
 	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr, c.metricsManager)
 	if err != nil {
 		return err
@@ -383,7 +400,7 @@ func (c *csiDriverClient) NodeStageVolume(ctx context.Context,
 		StagingTargetPath: stagingTargetPath,
 		VolumeCapability: &csipbv1.VolumeCapability{
 			AccessMode: &csipbv1.VolumeCapability_AccessMode{
-				Mode: asCSIAccessModeV1(accessMode),
+				Mode: accessModeMapper(accessMode),
 			},
 		},
 		Secrets:       secrets,
@@ -438,66 +455,23 @@ func (c *csiDriverClient) NodeUnstageVolume(ctx context.Context, volID, stagingT
 
 func (c *csiDriverClient) NodeSupportsNodeExpand(ctx context.Context) (bool, error) {
 	klog.V(4).Info(log("calling NodeGetCapabilities rpc to determine if Node has EXPAND_VOLUME capability"))
-	if c.nodeV1ClientCreator == nil {
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_EXPAND_VOLUME)
-		return false, errors.New("nodeV1ClientCreate is nil")
-	}
-
-	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr, c.metricsManager)
-	if err != nil {
-		return false, err
-	}
-	defer closer.Close()
-
-	req := &csipbv1.NodeGetCapabilitiesRequest{}
-	resp, err := nodeClient.NodeGetCapabilities(ctx, req)
-	if err != nil {
-		return false, err
-	}
-
-	capabilities := resp.GetCapabilities()
-
-	if capabilities == nil {
-		return false, nil
-	}
-	for _, capability := range capabilities {
-		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_EXPAND_VOLUME {
-			return true, nil
-		}
-	}
-	return false, nil
 }
 
 func (c *csiDriverClient) NodeSupportsStageUnstage(ctx context.Context) (bool, error) {
 	klog.V(4).Info(log("calling NodeGetCapabilities rpc to determine if NodeSupportsStageUnstage"))
-	if c.nodeV1ClientCreator == nil {
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME)
-		return false, errors.New("nodeV1ClientCreate is nil")
+}
-	}
 
-	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr, c.metricsManager)
+func (c *csiDriverClient) getNodeV1AccessModeMapper(ctx context.Context) (nodeV1AccessModeMapper, error) {
+	supported, err := c.NodeSupportsSingleNodeMultiWriterAccessMode(ctx)
 	if err != nil {
-		return false, err
+		return nil, err
 	}
-	defer closer.Close()
+	if supported {
-
+		return asSingleNodeMultiWriterCapableCSIAccessModeV1, nil
-	req := &csipbv1.NodeGetCapabilitiesRequest{}
-	resp, err := nodeClient.NodeGetCapabilities(ctx, req)
-	if err != nil {
-		return false, err
 	}
-
+	return asCSIAccessModeV1, nil
-	capabilities := resp.GetCapabilities()
-
-	stageUnstageSet := false
-	if capabilities == nil {
-		return false, nil
-	}
-	for _, capability := range capabilities {
-		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME {
-			stageUnstageSet = true
-			break
-		}
-	}
-	return stageUnstageSet, nil
 }
 
 func asCSIAccessModeV1(am api.PersistentVolumeAccessMode) csipbv1.VolumeCapability_AccessMode_Mode {
@@ -508,6 +482,25 @@ func asCSIAccessModeV1(am api.PersistentVolumeAccessMode) csipbv1.VolumeCapabili
 		return csipbv1.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY
 	case api.ReadWriteMany:
 		return csipbv1.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER
+	// This mapping exists to enable CSI drivers that lack the
+	// SINGLE_NODE_MULTI_WRITER capability to work with the
+	// ReadWriteOncePod access mode.
+	case api.ReadWriteOncePod:
+		return csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_WRITER
+	}
+	return csipbv1.VolumeCapability_AccessMode_UNKNOWN
+}
+
+func asSingleNodeMultiWriterCapableCSIAccessModeV1(am api.PersistentVolumeAccessMode) csipbv1.VolumeCapability_AccessMode_Mode {
+	switch am {
+	case api.ReadWriteOnce:
+		return csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_MULTI_WRITER
+	case api.ReadOnlyMany:
+		return csipbv1.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY
+	case api.ReadWriteMany:
+		return csipbv1.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER
+	case api.ReadWriteOncePod:
+		return csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER
 	}
 	return csipbv1.VolumeCapability_AccessMode_UNKNOWN
 }
@@ -561,30 +554,12 @@ func (c *csiClientGetter) Get() (csiClient, error) {
 
 func (c *csiDriverClient) NodeSupportsVolumeStats(ctx context.Context) (bool, error) {
 	klog.V(5).Info(log("calling NodeGetCapabilities rpc to determine if NodeSupportsVolumeStats"))
-	if c.nodeV1ClientCreator == nil {
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_GET_VOLUME_STATS)
-		return false, errors.New("nodeV1ClientCreate is nil")
+}
-	}
 
-	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr, c.metricsManager)
+func (c *csiDriverClient) NodeSupportsSingleNodeMultiWriterAccessMode(ctx context.Context) (bool, error) {
-	if err != nil {
+	klog.V(4).Info(log("calling NodeGetCapabilities rpc to determine if NodeSupportsSingleNodeMultiWriterAccessMode"))
-		return false, err
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER)
-	}
-	defer closer.Close()
-	req := &csipbv1.NodeGetCapabilitiesRequest{}
-	resp, err := nodeClient.NodeGetCapabilities(ctx, req)
-	if err != nil {
-		return false, err
-	}
-	capabilities := resp.GetCapabilities()
-	if capabilities == nil {
-		return false, nil
-	}
-	for _, capability := range capabilities {
-		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_GET_VOLUME_STATS {
-			return true, nil
-		}
-	}
-	return false, nil
 }
 
 func (c *csiDriverClient) NodeGetVolumeStats(ctx context.Context, volID string, targetPath string) (*volume.Metrics, error) {
@@ -628,7 +603,7 @@ func (c *csiDriverClient) NodeGetVolumeStats(ctx context.Context, volID string,
 	}
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.CSIVolumeHealth) {
-		isSupportNodeVolumeCondition, err := supportNodeGetVolumeCondition(ctx, nodeClient)
+		isSupportNodeVolumeCondition, err := c.nodeSupportsVolumeCondition(ctx)
 		if err != nil {
 			return nil, err
 		}
@@ -661,30 +636,47 @@ func (c *csiDriverClient) NodeGetVolumeStats(ctx context.Context, volID string,
 	return metrics, nil
 }
 
-func supportNodeGetVolumeCondition(ctx context.Context, nodeClient csipbv1.NodeClient) (supportNodeGetVolumeCondition bool, err error) {
+func (c *csiDriverClient) nodeSupportsVolumeCondition(ctx context.Context) (bool, error) {
-	req := csipbv1.NodeGetCapabilitiesRequest{}
+	klog.V(5).Info(log("calling NodeGetCapabilities rpc to determine if nodeSupportsVolumeCondition"))
-	rsp, err := nodeClient.NodeGetCapabilities(ctx, &req)
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_VOLUME_CONDITION)
+}
+
+func (c *csiDriverClient) nodeSupportsCapability(ctx context.Context, capabilityType csipbv1.NodeServiceCapability_RPC_Type) (bool, error) {
+	capabilities, err := c.nodeGetCapabilities(ctx)
 	if err != nil {
 		return false, err
 	}
 
-	for _, cap := range rsp.GetCapabilities() {
+	for _, capability := range capabilities {
-		if cap == nil {
+		if capability == nil || capability.GetRpc() == nil {
 			continue
 		}
-		rpc := cap.GetRpc()
+		if capability.GetRpc().GetType() == capabilityType {
-		if rpc == nil {
-			continue
-		}
-		t := rpc.GetType()
-		if t == csipbv1.NodeServiceCapability_RPC_VOLUME_CONDITION {
 			return true, nil
 		}
 	}
-
 	return false, nil
 }
 
+func (c *csiDriverClient) nodeGetCapabilities(ctx context.Context) ([]*csipbv1.NodeServiceCapability, error) {
+	if c.nodeV1ClientCreator == nil {
+		return []*csipbv1.NodeServiceCapability{}, errors.New("nodeV1ClientCreate is nil")
+	}
+
+	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr, c.metricsManager)
+	if err != nil {
+		return []*csipbv1.NodeServiceCapability{}, err
+	}
+	defer closer.Close()
+
+	req := &csipbv1.NodeGetCapabilitiesRequest{}
+	resp, err := nodeClient.NodeGetCapabilities(ctx, req)
+	if err != nil {
+		return []*csipbv1.NodeServiceCapability{}, err
+	}
+	return resp.GetCapabilities(), nil
+}
+
 func isFinalError(err error) bool {
 	// Sources:
 	// https://github.com/grpc/grpc/blob/master/doc/statuscodes.md

pkg/volume/csi/csi_client_test.go

@@ -106,7 +106,7 @@ func (c *fakeCsiDriverClient) NodeGetVolumeStats(ctx context.Context, volID stri
 
 	metrics := &volume.Metrics{}
 
-	isSupportNodeVolumeCondition, err := supportNodeGetVolumeCondition(ctx, c.nodeClient)
+	isSupportNodeVolumeCondition, err := c.nodeSupportsVolumeCondition(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -137,21 +137,7 @@ func (c *fakeCsiDriverClient) NodeGetVolumeStats(ctx context.Context, volID stri
 
 func (c *fakeCsiDriverClient) NodeSupportsVolumeStats(ctx context.Context) (bool, error) {
 	c.t.Log("calling fake.NodeSupportsVolumeStats...")
-	req := &csipbv1.NodeGetCapabilitiesRequest{}
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_GET_VOLUME_STATS)
-	resp, err := c.nodeClient.NodeGetCapabilities(ctx, req)
-	if err != nil {
-		return false, err
-	}
-	capabilities := resp.GetCapabilities()
-	if capabilities == nil {
-		return false, nil
-	}
-	for _, capability := range capabilities {
-		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_GET_VOLUME_STATS {
-			return true, nil
-		}
-	}
-	return false, nil
 }
 
 func (c *fakeCsiDriverClient) NodePublishVolume(
@@ -269,46 +255,12 @@ func (c *fakeCsiDriverClient) NodeUnstageVolume(ctx context.Context, volID, stag
 
 func (c *fakeCsiDriverClient) NodeSupportsNodeExpand(ctx context.Context) (bool, error) {
 	c.t.Log("calling fake.NodeSupportsNodeExpand...")
-	req := &csipbv1.NodeGetCapabilitiesRequest{}
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_EXPAND_VOLUME)
-
-	resp, err := c.nodeClient.NodeGetCapabilities(ctx, req)
-	if err != nil {
-		return false, err
-	}
-
-	capabilities := resp.GetCapabilities()
-
-	if capabilities == nil {
-		return false, nil
-	}
-	for _, capability := range capabilities {
-		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_EXPAND_VOLUME {
-			return true, nil
-		}
-	}
-	return false, nil
 }
 
 func (c *fakeCsiDriverClient) NodeSupportsStageUnstage(ctx context.Context) (bool, error) {
 	c.t.Log("calling fake.NodeGetCapabilities for NodeSupportsStageUnstage...")
-	req := &csipbv1.NodeGetCapabilitiesRequest{}
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME)
-	resp, err := c.nodeClient.NodeGetCapabilities(ctx, req)
-	if err != nil {
-		return false, err
-	}
-
-	capabilities := resp.GetCapabilities()
-
-	stageUnstageSet := false
-	if capabilities == nil {
-		return false, nil
-	}
-	for _, capability := range capabilities {
-		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME {
-			stageUnstageSet = true
-		}
-	}
-	return stageUnstageSet, nil
 }
 
 func (c *fakeCsiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResizeOptions) (resource.Quantity, error) {
@@ -344,6 +296,39 @@ func (c *fakeCsiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResi
 	return *updatedQuantity, nil
 }
 
+func (c *fakeCsiDriverClient) nodeSupportsVolumeCondition(ctx context.Context) (bool, error) {
+	c.t.Log("calling fake.nodeSupportsVolumeCondition...")
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_VOLUME_CONDITION)
+}
+
+func (c *fakeCsiDriverClient) NodeSupportsSingleNodeMultiWriterAccessMode(ctx context.Context) (bool, error) {
+	c.t.Log("calling fake.NodeSupportsSingleNodeMultiWriterAccessMode...")
+	return c.nodeSupportsCapability(ctx, csipbv1.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER)
+}
+
+func (c *fakeCsiDriverClient) nodeSupportsCapability(ctx context.Context, capabilityType csipbv1.NodeServiceCapability_RPC_Type) (bool, error) {
+	capabilities, err := c.nodeGetCapabilities(ctx)
+	if err != nil {
+		return false, err
+	}
+
+	for _, capability := range capabilities {
+		if capability.GetRpc().GetType() == capabilityType {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func (c *fakeCsiDriverClient) nodeGetCapabilities(ctx context.Context) ([]*csipbv1.NodeServiceCapability, error) {
+	req := &csipbv1.NodeGetCapabilitiesRequest{}
+	resp, err := c.nodeClient.NodeGetCapabilities(ctx, req)
+	if err != nil {
+		return []*csipbv1.NodeServiceCapability{}, err
+	}
+	return resp.GetCapabilities(), nil
+}
+
 func setupClient(t *testing.T, stageUnstageSet bool) csiClient {
 	return newFakeCsiDriverClient(t, stageUnstageSet)
 }
@@ -885,3 +870,83 @@ func TestVolumeStats(t *testing.T) {
 	}
 
 }
+
+func TestAccessModeMapping(t *testing.T) {
+	tests := []struct {
+		name                     string
+		singleNodeMultiWriterSet bool
+		accessMode               api.PersistentVolumeAccessMode
+		expectedMappedAccessMode csipbv1.VolumeCapability_AccessMode_Mode
+	}{
+		{
+			name:                     "with ReadWriteOnce and incapable driver",
+			singleNodeMultiWriterSet: false,
+			accessMode:               api.ReadWriteOnce,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+		},
+		{
+			name:                     "with ReadOnlyMany and incapable driver",
+			singleNodeMultiWriterSet: false,
+			accessMode:               api.ReadOnlyMany,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+		},
+		{
+			name:                     "with ReadWriteMany and incapable driver",
+			singleNodeMultiWriterSet: false,
+			accessMode:               api.ReadWriteMany,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
+		},
+		{
+			name:                     "with ReadWriteOncePod and incapable driver",
+			singleNodeMultiWriterSet: false,
+			accessMode:               api.ReadWriteOncePod,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+		},
+		{
+			name:                     "with ReadWriteOnce and capable driver",
+			singleNodeMultiWriterSet: true,
+			accessMode:               api.ReadWriteOnce,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_MULTI_WRITER,
+		},
+		{
+			name:                     "with ReadOnlyMany and capable driver",
+			singleNodeMultiWriterSet: true,
+			accessMode:               api.ReadOnlyMany,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+		},
+		{
+			name:                     "with ReadWriteMany and capable driver",
+			singleNodeMultiWriterSet: true,
+			accessMode:               api.ReadWriteMany,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
+		},
+		{
+			name:                     "with ReadWriteOncePod and capable driver",
+			singleNodeMultiWriterSet: true,
+			accessMode:               api.ReadWriteOncePod,
+			expectedMappedAccessMode: csipbv1.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeCloser := fake.NewCloser(t)
+			client := &csiDriverClient{
+				driverName: "Fake Driver Name",
+				nodeV1ClientCreator: func(addr csiAddr, m *MetricsManager) (csipbv1.NodeClient, io.Closer, error) {
+					nodeClient := fake.NewNodeClientWithSingleNodeMultiWriter(tc.singleNodeMultiWriterSet)
+					return nodeClient, fakeCloser, nil
+				},
+			}
+
+			accessModeMapper, err := client.getNodeV1AccessModeMapper(context.Background())
+			if err != nil {
+				t.Error(err)
+			}
+
+			mappedAccessMode := accessModeMapper(tc.accessMode)
+			if mappedAccessMode != tc.expectedMappedAccessMode {
+				t.Errorf("expected access mode: %v; got: %v", tc.expectedMappedAccessMode, mappedAccessMode)
+			}
+		})
+	}
+}

pkg/volume/csi/fake/fake_client.go

@@ -85,6 +85,7 @@ type NodeClient struct {
 	expansionSet             bool
 	volumeStatsSet           bool
 	volumeConditionSet       bool
+	singleNodeMultiWriterSet bool
 	nodeGetInfoResp          *csipb.NodeGetInfoResponse
 	nodeVolumeStatsResp      *csipb.NodeGetVolumeStatsResponse
 	FakeNodeExpansionRequest *csipb.NodeExpandVolumeRequest
@@ -123,6 +124,16 @@ func NewNodeClientWithVolumeStatsAndCondition(volumeStatsSet, volumeConditionSet
 	}
 }
 
+func NewNodeClientWithSingleNodeMultiWriter(singleNodeMultiWriterSet bool) *NodeClient {
+	return &NodeClient{
+		nodePublishedVolumes:     make(map[string]CSIVolume),
+		nodeStagedVolumes:        make(map[string]CSIVolume),
+		stageUnstageSet:          true,
+		volumeStatsSet:           true,
+		singleNodeMultiWriterSet: singleNodeMultiWriterSet,
+	}
+}
+
 // SetNextError injects next expected error
 func (f *NodeClient) SetNextError(err error) {
 	f.nextErr = err
@@ -364,6 +375,16 @@ func (f *NodeClient) NodeGetCapabilities(ctx context.Context, in *csipb.NodeGetC
 			},
 		})
 	}
+
+	if f.singleNodeMultiWriterSet {
+		resp.Capabilities = append(resp.Capabilities, &csipb.NodeServiceCapability{
+			Type: &csipb.NodeServiceCapability_Rpc{
+				Rpc: &csipb.NodeServiceCapability_RPC{
+					Type: csipb.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,
+				},
+			},
+		})
+	}
 	return resp, nil
 }
 

pkg/volume/flocker/flocker_volume.go

@@ -55,7 +55,7 @@ type flockerVolumeProvisioner struct {
 var _ volume.Provisioner = &flockerVolumeProvisioner{}
 
 func (c *flockerVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}
 

pkg/volume/gcepd/gce_pd.go

@@ -488,7 +488,7 @@ type gcePersistentDiskProvisioner struct {
 var _ volume.Provisioner = &gcePersistentDiskProvisioner{}
 
 func (c *gcePersistentDiskProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}
 

pkg/volume/glusterfs/glusterfs.go

@@ -720,7 +720,7 @@ func filterClient(client *gcli.Client, opts *proxyutil.FilteredDialOptions) *gcl
 }
 
 func (p *glusterfsVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !volutil.AccessModesContainedInAll(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
+	if !volutil.ContainsAllAccessModes(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", p.options.PVC.Spec.AccessModes, p.plugin.GetAccessModes())
 	}
 	if p.options.PVC.Spec.Selector != nil {

pkg/volume/portworx/portworx.go

@@ -389,7 +389,7 @@ type portworxVolumeProvisioner struct {
 var _ volume.Provisioner = &portworxVolumeProvisioner{}
 
 func (c *portworxVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}
 

pkg/volume/quobyte/quobyte.go

@@ -373,7 +373,7 @@ type quobyteVolumeProvisioner struct {
 }
 
 func (provisioner *quobyteVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(provisioner.plugin.GetAccessModes(), provisioner.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(provisioner.plugin.GetAccessModes(), provisioner.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", provisioner.options.PVC.Spec.AccessModes, provisioner.plugin.GetAccessModes())
 	}
 

pkg/volume/rbd/rbd.go

@@ -615,7 +615,7 @@ type rbdVolumeProvisioner struct {
 var _ volume.Provisioner = &rbdVolumeProvisioner{}
 
 func (r *rbdVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !volutil.AccessModesContainedInAll(r.plugin.GetAccessModes(), r.options.PVC.Spec.AccessModes) {
+	if !volutil.ContainsAllAccessModes(r.plugin.GetAccessModes(), r.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", r.options.PVC.Spec.AccessModes, r.plugin.GetAccessModes())
 	}
 

pkg/volume/storageos/storageos.go

@@ -570,7 +570,7 @@ type storageosProvisioner struct {
 var _ volume.Provisioner = &storageosProvisioner{}
 
 func (c *storageosProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(c.plugin.GetAccessModes(), c.options.PVC.Spec.AccessModes) {
 		return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", c.options.PVC.Spec.AccessModes, c.plugin.GetAccessModes())
 	}
 	if util.CheckPersistentVolumeClaimModeBlock(c.options.PVC) {

pkg/volume/util/operationexecutor/operation_executor.go

@@ -203,6 +203,9 @@ type ActualStateOfWorldMounterUpdater interface {
 	// GetVolumeMountState returns mount state of the volume for the Pod
 	GetVolumeMountState(volumName v1.UniqueVolumeName, podName volumetypes.UniquePodName) VolumeMountState
 
+	// IsVolumeMountedElsewhere returns whether the supplied volume is mounted in a Pod other than the supplied one
+	IsVolumeMountedElsewhere(volumeName v1.UniqueVolumeName, podName volumetypes.UniquePodName) bool
+
 	// MarkForInUseExpansionError marks the volume to have in-use error during expansion.
 	// volume expansion must not be retried for this volume
 	MarkForInUseExpansionError(volumeName v1.UniqueVolumeName)

pkg/volume/util/operationexecutor/operation_generator.go

@@ -35,6 +35,7 @@ import (
 	volerr "k8s.io/cloud-provider/volume/errors"
 	csitrans "k8s.io/csi-translation-lib"
 	"k8s.io/klog/v2"
+	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/features"
 	kevents "k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/volume"
@@ -537,12 +538,23 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 		}
 
 		mountCheckError := checkMountOptionSupport(og, volumeToMount, volumePlugin)
-
 		if mountCheckError != nil {
 			eventErr, detailedErr := volumeToMount.GenerateError("MountVolume.MountOptionSupport check failed", mountCheckError)
 			return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
 		}
 
+		// Enforce ReadWriteOncePod access mode if it is the only one present. This is also enforced during scheduling.
+		if utilfeature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod) &&
+			actualStateOfWorld.IsVolumeMountedElsewhere(volumeToMount.VolumeName, volumeToMount.PodName) &&
+			// Because we do not know what access mode the pod intends to use if there are multiple.
+			len(volumeToMount.VolumeSpec.PersistentVolume.Spec.AccessModes) == 1 &&
+			v1helper.ContainsAccessMode(volumeToMount.VolumeSpec.PersistentVolume.Spec.AccessModes, v1.ReadWriteOncePod) {
+
+			err = goerrors.New("volume uses the ReadWriteOncePod access mode and is already in use by another pod")
+			eventErr, detailedErr := volumeToMount.GenerateError("MountVolume.SetUp failed", err)
+			return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
+		}
+
 		// Get attacher, if possible
 		attachableVolumePlugin, _ :=
 			og.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec)
@@ -1027,6 +1039,18 @@ func (og *operationGenerator) GenerateMapVolumeFunc(
 
 		migrated := getMigratedStatusBySpec(volumeToMount.VolumeSpec)
 
+		// Enforce ReadWriteOncePod access mode. This is also enforced during scheduling.
+		if utilfeature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod) &&
+			actualStateOfWorld.IsVolumeMountedElsewhere(volumeToMount.VolumeName, volumeToMount.PodName) &&
+			// Because we do not know what access mode the pod intends to use if there are multiple.
+			len(volumeToMount.VolumeSpec.PersistentVolume.Spec.AccessModes) == 1 &&
+			v1helper.ContainsAccessMode(volumeToMount.VolumeSpec.PersistentVolume.Spec.AccessModes, v1.ReadWriteOncePod) {
+
+			err = goerrors.New("volume uses the ReadWriteOncePod access mode and is already in use by another pod")
+			eventErr, detailedErr := volumeToMount.GenerateError("MapVolume.SetUpDevice failed", err)
+			return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
+		}
+
 		// Set up global map path under the given plugin directory using symbolic link
 		globalMapPath, err :=
 			blockVolumeMapper.GetGlobalMapPath(volumeToMount.VolumeSpec)

pkg/volume/util/util.go

@@ -298,8 +298,8 @@ func JoinMountOptions(userOptions []string, systemOptions []string) []string {
 	return allMountOptions.List()
 }
 
-// AccessModesContains returns whether the requested mode is contained by modes
+// ContainsAccessMode returns whether the requested mode is contained by modes
-func AccessModesContains(modes []v1.PersistentVolumeAccessMode, mode v1.PersistentVolumeAccessMode) bool {
+func ContainsAccessMode(modes []v1.PersistentVolumeAccessMode, mode v1.PersistentVolumeAccessMode) bool {
 	for _, m := range modes {
 		if m == mode {
 			return true
@@ -308,10 +308,10 @@ func AccessModesContains(modes []v1.PersistentVolumeAccessMode, mode v1.Persiste
 	return false
 }
 
-// AccessModesContainedInAll returns whether all of the requested modes are contained by modes
+// ContainsAllAccessModes returns whether all of the requested modes are contained by modes
-func AccessModesContainedInAll(indexedModes []v1.PersistentVolumeAccessMode, requestedModes []v1.PersistentVolumeAccessMode) bool {
+func ContainsAllAccessModes(indexedModes []v1.PersistentVolumeAccessMode, requestedModes []v1.PersistentVolumeAccessMode) bool {
 	for _, mode := range requestedModes {
-		if !AccessModesContains(indexedModes, mode) {
+		if !ContainsAccessMode(indexedModes, mode) {
 			return false
 		}
 	}

pkg/volume/vsphere_volume/vsphere_volume.go

@@ -369,7 +369,7 @@ func (plugin *vsphereVolumePlugin) newProvisionerInternal(options volume.VolumeO
 }
 
 func (v *vsphereVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
-	if !util.AccessModesContainedInAll(v.plugin.GetAccessModes(), v.options.PVC.Spec.AccessModes) {
+	if !util.ContainsAllAccessModes(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())
 	}
 	klog.V(1).Infof("Provision with selectedNode: %s and allowedTopologies : %s", getNodeName(selectedNode), allowedTopologies)

staging/src/k8s.io/api/core/v1/types.go

@@ -560,6 +560,9 @@ const (
 	ReadOnlyMany PersistentVolumeAccessMode = "ReadOnlyMany"
 	// can be mounted in read/write mode to many hosts
 	ReadWriteMany PersistentVolumeAccessMode = "ReadWriteMany"
+	// can be mounted in read/write mode to exactly 1 pod
+	// cannot be used in combination with other access modes
+	ReadWriteOncePod PersistentVolumeAccessMode = "ReadWriteOncePod"
 )
 
 type PersistentVolumePhase string

staging/src/k8s.io/kubectl/pkg/util/storage/storage.go

@@ -45,19 +45,22 @@ func IsDefaultAnnotationText(obj metav1.ObjectMeta) string {
 }
 
 // GetAccessModesAsString returns a string representation of an array of access modes.
-// modes, when present, are always in the same order: RWO,ROX,RWX.
+// modes, when present, are always in the same order: RWO,ROX,RWX,RWOP.
 func GetAccessModesAsString(modes []v1.PersistentVolumeAccessMode) string {
 	modes = removeDuplicateAccessModes(modes)
 	modesStr := []string{}
-	if containsAccessMode(modes, v1.ReadWriteOnce) {
+	if ContainsAccessMode(modes, v1.ReadWriteOnce) {
 		modesStr = append(modesStr, "RWO")
 	}
-	if containsAccessMode(modes, v1.ReadOnlyMany) {
+	if ContainsAccessMode(modes, v1.ReadOnlyMany) {
 		modesStr = append(modesStr, "ROX")
 	}
-	if containsAccessMode(modes, v1.ReadWriteMany) {
+	if ContainsAccessMode(modes, v1.ReadWriteMany) {
 		modesStr = append(modesStr, "RWX")
 	}
+	if ContainsAccessMode(modes, v1.ReadWriteOncePod) {
+		modesStr = append(modesStr, "RWOP")
+	}
 	return strings.Join(modesStr, ",")
 }
 
@@ -65,14 +68,14 @@ func GetAccessModesAsString(modes []v1.PersistentVolumeAccessMode) string {
 func removeDuplicateAccessModes(modes []v1.PersistentVolumeAccessMode) []v1.PersistentVolumeAccessMode {
 	accessModes := []v1.PersistentVolumeAccessMode{}
 	for _, m := range modes {
-		if !containsAccessMode(accessModes, m) {
+		if !ContainsAccessMode(accessModes, m) {
 			accessModes = append(accessModes, m)
 		}
 	}
 	return accessModes
 }
 
-func containsAccessMode(modes []v1.PersistentVolumeAccessMode, mode v1.PersistentVolumeAccessMode) bool {
+func ContainsAccessMode(modes []v1.PersistentVolumeAccessMode, mode v1.PersistentVolumeAccessMode) bool {
 	for _, m := range modes {
 		if m == mode {
 			return true