Parallelize attach operations across different nodes for volumes that allow multi-attach

Author: verult

pkg/controller/volume/attachdetach/reconciler/BUILD

@@ -16,7 +16,7 @@ go_library(
         "//pkg/controller/volume/attachdetach/statusupdater:go_default_library",
         "//pkg/kubelet/events:go_default_library",
         "//pkg/util/goroutinemap/exponentialbackoff:go_default_library",
-        "//pkg/volume:go_default_library",
+        "//pkg/volume/util:go_default_library",
         "//pkg/volume/util/operationexecutor:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",

pkg/controller/volume/attachdetach/reconciler/reconciler.go

@@ -24,7 +24,7 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/record"
@@ -34,7 +34,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/volume/attachdetach/statusupdater"
 	kevents "k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/util/goroutinemap/exponentialbackoff"
-	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
 )
 
@@ -134,42 +134,6 @@ func (rc *reconciler) syncStates() {
 	rc.attacherDetacher.VerifyVolumesAreAttached(volumesPerNode, rc.actualStateOfWorld)
 }
 
-// isMultiAttachForbidden checks if attaching this volume to multiple nodes is definitely not allowed/possible.
-// In its current form, this function can only reliably say for which volumes it's definitely forbidden. If it returns
-// false, it is not guaranteed that multi-attach is actually supported by the volume type and we must rely on the
-// attacher to fail fast in such cases.
-// Please see https://github.com/kubernetes/kubernetes/issues/40669 and https://github.com/kubernetes/kubernetes/pull/40148#discussion_r98055047
-func (rc *reconciler) isMultiAttachForbidden(volumeSpec *volume.Spec) bool {
-	if volumeSpec.Volume != nil {
-		// Check for volume types which are known to fail slow or cause trouble when trying to multi-attach
-		if volumeSpec.Volume.AzureDisk != nil ||
-			volumeSpec.Volume.Cinder != nil {
-			return true
-		}
-	}
-
-	// Only if this volume is a persistent volume, we have reliable information on whether it's allowed or not to
-	// multi-attach. We trust in the individual volume implementations to not allow unsupported access modes
-	if volumeSpec.PersistentVolume != nil {
-		// Check for persistent volume types which do not fail when trying to multi-attach
-		if len(volumeSpec.PersistentVolume.Spec.AccessModes) == 0 {
-			// No access mode specified so we don't know for sure. Let the attacher fail if needed
-			return false
-		}
-
-		// check if this volume is allowed to be attached to multiple PODs/nodes, if yes, return false
-		for _, accessMode := range volumeSpec.PersistentVolume.Spec.AccessModes {
-			if accessMode == v1.ReadWriteMany || accessMode == v1.ReadOnlyMany {
-				return false
-			}
-		}
-		return true
-	}
-
-	// we don't know if it's supported or not and let the attacher fail later in cases it's not supported
-	return false
-}
-
 func (rc *reconciler) reconcile() {
 	// Detaches are triggered before attaches so that volumes referenced by
 	// pods that are rescheduled to a different node are detached first.
@@ -182,9 +146,16 @@ func (rc *reconciler) reconcile() {
 			// This check must be done before we do any other checks, as otherwise the other checks
 			// may pass while at the same time the volume leaves the pending state, resulting in
 			// double detach attempts
-			if rc.attacherDetacher.IsOperationPending(attachedVolume.VolumeName, "") {
-				klog.V(10).Infof("Operation for volume %q is already running. Can't start detach for %q", attachedVolume.VolumeName, attachedVolume.NodeName)
-				continue
+			if util.IsMultiAttachForbidden(attachedVolume.VolumeSpec) {
+				if rc.attacherDetacher.IsOperationPending(attachedVolume.VolumeName, "" /* podName */, "" /* nodeName */) {
+					klog.V(10).Infof("Operation for volume %q is already running in the cluster. Can't start detach for %q", attachedVolume.VolumeName, attachedVolume.NodeName)
+					continue
+				}
+			} else {
+				if rc.attacherDetacher.IsOperationPending(attachedVolume.VolumeName, "" /* podName */, attachedVolume.NodeName) {
+					klog.V(10).Infof("Operation for volume %q is already running for node %q. Can't start detach", attachedVolume.VolumeName, attachedVolume.NodeName)
+					continue
+				}
 			}
 
 			// Set the detach request time
@@ -260,15 +231,17 @@ func (rc *reconciler) attachDesiredVolumes() {
 			rc.actualStateOfWorld.ResetDetachRequestTime(volumeToAttach.VolumeName, volumeToAttach.NodeName)
 			continue
 		}
-		// Don't even try to start an operation if there is already one running
-		if rc.attacherDetacher.IsOperationPending(volumeToAttach.VolumeName, "") {
-			if klog.V(10) {
-				klog.Infof("Operation for volume %q is already running. Can't start attach for %q", volumeToAttach.VolumeName, volumeToAttach.NodeName)
+
+		if util.IsMultiAttachForbidden(volumeToAttach.VolumeSpec) {
+
+			// Don't even try to start an operation if there is already one running for the given volume
+			if rc.attacherDetacher.IsOperationPending(volumeToAttach.VolumeName, "" /* podName */, "" /* nodeName */) {
+				if klog.V(10) {
+					klog.Infof("Operation for volume %q is already running. Can't start attach for %q", volumeToAttach.VolumeName, volumeToAttach.NodeName)
+				}
+				continue
 			}
-			continue
-		}
 
-		if rc.isMultiAttachForbidden(volumeToAttach.VolumeSpec) {
 			nodes := rc.actualStateOfWorld.GetNodesForAttachedVolume(volumeToAttach.VolumeName)
 			if len(nodes) > 0 {
 				if !volumeToAttach.MultiAttachErrorReported {
@@ -277,6 +250,17 @@ func (rc *reconciler) attachDesiredVolumes() {
 				}
 				continue
 			}
+
+		} else {
+
+			// Don't even try to start an operation if there is already one running for the given volume and node.
+			if rc.attacherDetacher.IsOperationPending(volumeToAttach.VolumeName, "" /* podName */, volumeToAttach.NodeName) {
+				if klog.V(10) {
+					klog.Infof("Operation for volume %q is already running for node %q. Can't start attach", volumeToAttach.VolumeName, volumeToAttach.NodeName)
+				}
+				continue
+			}
+
 		}
 
 		// Volume/Node doesn't exist, spawn a goroutine to attach it

pkg/kubelet/volumemanager/reconciler/reconciler.go

@@ -294,7 +294,7 @@ func (rc *reconciler) unmountDetachDevices() {
 	for _, attachedVolume := range rc.actualStateOfWorld.GetUnmountedVolumes() {
 		// Check IsOperationPending to avoid marking a volume as detached if it's in the process of mounting.
 		if !rc.desiredStateOfWorld.VolumeExists(attachedVolume.VolumeName) &&
-			!rc.operationExecutor.IsOperationPending(attachedVolume.VolumeName, nestedpendingoperations.EmptyUniquePodName) {
+			!rc.operationExecutor.IsOperationPending(attachedVolume.VolumeName, nestedpendingoperations.EmptyUniquePodName, nestedpendingoperations.EmptyNodeName) {
 			if attachedVolume.DeviceMayBeMounted() {
 				// Volume is globally mounted to device, unmount it
 				klog.V(5).Infof(attachedVolume.GenerateMsgDetailed("Starting operationExecutor.UnmountDevice", ""))
@@ -422,7 +422,7 @@ func (rc *reconciler) syncStates() {
 			continue
 		}
 		// There is no pod that uses the volume.
-		if rc.operationExecutor.IsOperationPending(reconstructedVolume.volumeName, nestedpendingoperations.EmptyUniquePodName) {
+		if rc.operationExecutor.IsOperationPending(reconstructedVolume.volumeName, nestedpendingoperations.EmptyUniquePodName, nestedpendingoperations.EmptyNodeName) {
 			klog.Warning("Volume is in pending operation, skip cleaning up mounts")
 		}
 		klog.V(2).Infof(

pkg/volume/util/nestedpendingoperations/BUILD

@@ -14,6 +14,7 @@ go_library(
         "//pkg/util/goroutinemap/exponentialbackoff:go_default_library",
         "//pkg/volume/util/types:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
@@ -27,6 +28,7 @@ go_test(
         "//pkg/util/goroutinemap/exponentialbackoff:go_default_library",
         "//pkg/volume/util/types:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
     ],
 )

pkg/volume/util/nestedpendingoperations/nestedpendingoperations.go

@@ -29,45 +29,73 @@ import (
 	"sync"
 
 	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	k8sRuntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/util/goroutinemap/exponentialbackoff"
-	"k8s.io/kubernetes/pkg/volume/util/types"
+	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
 )
 
 const (
 	// EmptyUniquePodName is a UniquePodName for empty string.
-	EmptyUniquePodName types.UniquePodName = types.UniquePodName("")
+	EmptyUniquePodName volumetypes.UniquePodName = volumetypes.UniquePodName("")
 
 	// EmptyUniqueVolumeName is a UniqueVolumeName for empty string
 	EmptyUniqueVolumeName v1.UniqueVolumeName = v1.UniqueVolumeName("")
+
+	// EmptyNodeName is a NodeName for empty string
+	EmptyNodeName types.NodeName = types.NodeName("")
 )
 
 // NestedPendingOperations defines the supported set of operations.
 type NestedPendingOperations interface {
-	// Run adds the concatenation of volumeName and podName to the list of
-	// running operations and spawns a new go routine to execute operationFunc.
-	// If an operation with the same volumeName, same or empty podName
-	// and same operationName exits, an AlreadyExists or ExponentialBackoff
-	// error is returned. If an operation with same volumeName and podName
-	// has ExponentialBackoff error but operationName is different, exponential
-	// backoff is reset and operation is allowed to proceed.
-	// This enables multiple operations to execute in parallel for the same
-	// volumeName as long as they have different podName.
+
+	// Run adds the concatenation of volumeName and one of podName or nodeName to
+	// the list of running operations and spawns a new go routine to execute
+	// OperationFunc inside generatedOperations.
+
+	// volumeName, podName, and nodeName collectively form the operation key.
+	// The following forms of operation keys are supported:
+	// - volumeName empty, podName empty, nodeName empty
+	//   This key does not have any conflicting keys.
+	// - volumeName exists, podName empty, nodeName empty
+	//   This key conflicts with all other keys with the same volumeName.
+	// - volumeName exists, podName exists, nodeName empty
+	//   This key conflicts with:
+	//   - the same volumeName and podName
+	//   - the same volumeName, but no podName
+	// - volumeName exists, podName empty, nodeName exists
+	//   This key conflicts with:
+	//   - the same volumeName and nodeName
+	//   - the same volumeName but no nodeName
+
+	// If an operation with the same operationName and a conflicting key exists,
+	// an AlreadyExists or ExponentialBackoff error is returned.
+	// If an operation with a conflicting key has ExponentialBackoff error but
+	// operationName is different, exponential backoff is reset and operation is
+	// allowed to proceed.
+
 	// Once the operation is complete, the go routine is terminated and the
-	// concatenation of volumeName and podName is removed from the list of
-	// executing operations allowing a new operation to be started with the
-	// volumeName without error.
-	Run(volumeName v1.UniqueVolumeName, podName types.UniquePodName, generatedOperations types.GeneratedOperations) error
+	// concatenation of volumeName and (podName or nodeName) is removed from the
+	// list of executing operations allowing a new operation to be started with
+	// the volumeName without error.
+	Run(
+		volumeName v1.UniqueVolumeName,
+		podName volumetypes.UniquePodName,
+		nodeName types.NodeName,
+		generatedOperations volumetypes.GeneratedOperations) error
 
 	// Wait blocks until all operations are completed. This is typically
 	// necessary during tests - the test should wait until all operations finish
 	// and evaluate results after that.
 	Wait()
 
-	// IsOperationPending returns true if an operation for the given volumeName and podName is pending,
-	// otherwise it returns false
-	IsOperationPending(volumeName v1.UniqueVolumeName, podName types.UniquePodName) bool
+	// IsOperationPending returns true if an operation for the given volumeName
+	// and one of podName or nodeName is pending, otherwise it returns false
+	IsOperationPending(
+		volumeName v1.UniqueVolumeName,
+		podName volumetypes.UniquePodName,
+		nodeName types.NodeName) bool
 }
 
 // NewNestedPendingOperations returns a new instance of NestedPendingOperations.
@@ -96,12 +124,13 @@ type operation struct {
 
 func (grm *nestedPendingOperations) Run(
 	volumeName v1.UniqueVolumeName,
-	podName types.UniquePodName,
-	generatedOperations types.GeneratedOperations) error {
+	podName volumetypes.UniquePodName,
+	nodeName types.NodeName,
+	generatedOperations volumetypes.GeneratedOperations) error {
 	grm.lock.Lock()
 	defer grm.lock.Unlock()
 
-	opKey := operationKey{volumeName, podName}
+	opKey := operationKey{volumeName, podName, nodeName}
 
 	opExists, previousOpIndex := grm.isOperationExists(opKey)
 	if opExists {
@@ -149,12 +178,13 @@ func (grm *nestedPendingOperations) Run(
 
 func (grm *nestedPendingOperations) IsOperationPending(
 	volumeName v1.UniqueVolumeName,
-	podName types.UniquePodName) bool {
+	podName volumetypes.UniquePodName,
+	nodeName types.NodeName) bool {
 
 	grm.lock.RLock()
 	defer grm.lock.RUnlock()
 
-	opKey := operationKey{volumeName, podName}
+	opKey := operationKey{volumeName, podName, nodeName}
 	exist, previousOpIndex := grm.isOperationExists(opKey)
 	if exist && grm.operations[previousOpIndex].operationPending {
 		return true
@@ -177,8 +207,11 @@ func (grm *nestedPendingOperations) isOperationExists(key operationKey) (bool, i
 			key.podName == EmptyUniquePodName ||
 			previousOp.key.podName == key.podName
 
+		nodeNameMatch := previousOp.key.nodeName == EmptyNodeName ||
+			key.nodeName == EmptyNodeName ||
+			previousOp.key.nodeName == key.nodeName
 
-		if volumeNameMatch && podNameMatch {
+		if volumeNameMatch && podNameMatch && nodeNameMatch {
 			return true, previousOpIndex
 		}
 	}
@@ -264,15 +297,15 @@ func (grm *nestedPendingOperations) Wait() {
 
 type operationKey struct {
 	volumeName v1.UniqueVolumeName
-	podName    types.UniquePodName
+	podName    volumetypes.UniquePodName
+	nodeName   types.NodeName
 }
 
 func (key operationKey) String() string {
-	podNameStr := fmt.Sprintf(" (%q)", key.podName)
-
-	return fmt.Sprintf("%q%s",
+	return fmt.Sprintf("{volumeName=%q, podName=%q, nodeName=%q}",
 		key.volumeName,
-		podNameStr)
+		key.podName,
+		key.nodeName)
 }
 
 // NewAlreadyExistsError returns a new instance of AlreadyExists error.