cleanup: add IsBlockMultiNode() helper

IsBlockMultiNode() is a new helper that takes a slice of
VolumeCapability objects and checks if it includes multi-node access
and/or block-mode support.

This can then easily be used in other services that need checking for
these particular capabilities, and preventing multi-node block-mode
access.

Signed-off-by: Niels de Vos <ndevos@redhat.com>

internal/csi-common/utils.go

@@ -290,3 +290,19 @@ func requirePositive(x int64) int64 {
 
 	return 0
 }
+
+// IsBlockMultiNode checks the volume capabilities for BlockMode and MultiNode.
+func IsBlockMultiNode(caps []*csi.VolumeCapability) (bool, bool) {
+	isMultiNode := false
+	isBlock := false
+	for _, capability := range caps {
+		if capability.GetAccessMode().GetMode() == csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER {
+			isMultiNode = true
+		}
+		if capability.GetBlock() != nil {
+			isBlock = true
+		}
+	}
+
+	return isBlock, isMultiNode
+}

internal/csi-common/utils_test.go

@@ -116,3 +116,64 @@ func TestRequirePositive(t *testing.T) {
 	assert.Equal(t, requirePositive(-1), int64(0))
 	assert.Equal(t, requirePositive(1), int64(1))
 }
+
+func TestIsBlockMultiNode(t *testing.T) {
+	t.Parallel()
+
+	blockCap := &csi.VolumeCapability{
+		AccessType: &csi.VolumeCapability_Block{
+			Block: &csi.VolumeCapability_BlockVolume{},
+		},
+	}
+
+	fsCap := &csi.VolumeCapability{
+		AccessType: &csi.VolumeCapability_Mount{
+			Mount: &csi.VolumeCapability_MountVolume{},
+		},
+	}
+
+	multiNodeCap := &csi.VolumeCapability{
+		AccessMode: &csi.VolumeCapability_AccessMode{
+			Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
+		},
+	}
+
+	singleNodeCap := &csi.VolumeCapability{
+		AccessMode: &csi.VolumeCapability_AccessMode{
+			Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_MULTI_WRITER,
+		},
+	}
+
+	tests := []struct {
+		name        string
+		caps        []*csi.VolumeCapability
+		isBlock     bool
+		isMultiNode bool
+	}{{
+		name:        "block/multi-node",
+		caps:        []*csi.VolumeCapability{blockCap, multiNodeCap},
+		isBlock:     true,
+		isMultiNode: true,
+	}, {
+		name:        "block/single-node",
+		caps:        []*csi.VolumeCapability{blockCap, singleNodeCap},
+		isBlock:     true,
+		isMultiNode: false,
+	}, {
+		name:        "filesystem/multi-node",
+		caps:        []*csi.VolumeCapability{fsCap, multiNodeCap},
+		isBlock:     false,
+		isMultiNode: true,
+	}, {
+		name:        "filesystem/single-node",
+		caps:        []*csi.VolumeCapability{fsCap, singleNodeCap},
+		isBlock:     false,
+		isMultiNode: false,
+	}}
+
+	for _, test := range tests {
+		isBlock, isMultiNode := IsBlockMultiNode(test.caps)
+		assert.Equal(t, isBlock, test.isBlock, test.name)
+		assert.Equal(t, isMultiNode, test.isMultiNode, test.name)
+	}
+}

internal/rbd/controllerserver.go

@@ -99,17 +99,8 @@ func (cs *ControllerServer) parseVolCreateRequest(
 	req *csi.CreateVolumeRequest) (*rbdVolume, error) {
 	// TODO (sbezverk) Last check for not exceeding total storage capacity
 
-	isMultiNode := false
-	isBlock := false
-	for _, capability := range req.VolumeCapabilities {
 	// RO modes need to be handled independently (ie right now even if access mode is RO, they'll be RW upon attach)
-		if capability.GetAccessMode().GetMode() == csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER {
+	isBlock, isMultiNode := csicommon.IsBlockMultiNode(req.VolumeCapabilities)
-			isMultiNode = true
-		}
-		if capability.GetBlock() != nil {
-			isBlock = true
-		}
-	}
 
 	// We want to fail early if the user is trying to create a RWX on a non-block type device
 	if isMultiNode && !isBlock {