Enforce ReadWriteOncePod access mode during mount

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/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)