CSI: create volume capability validation

client/pluginmanager/csimanager/fingerprint.go

@@ -122,10 +122,18 @@ func (p *pluginFingerprinter) buildBasicFingerprint(ctx context.Context) (*struc
 }
 
 func applyCapabilitySetToControllerInfo(cs *csi.ControllerCapabilitySet, info *structs.CSIControllerInfo) {
-	info.SupportsReadOnlyAttach = cs.HasPublishReadonly
+	info.SupportsCreateDelete = cs.HasCreateDeleteVolume
 	info.SupportsAttachDetach = cs.HasPublishUnpublishVolume
 	info.SupportsListVolumes = cs.HasListVolumes
+	info.SupportsGetCapacity = cs.HasGetCapacity
+	info.SupportsCreateDeleteSnapshot = cs.HasCreateDeleteSnapshot
+	info.SupportsListSnapshots = cs.HasListSnapshots
+	info.SupportsClone = cs.HasCloneVolume
+	info.SupportsReadOnlyAttach = cs.HasPublishReadonly
+	info.SupportsExpand = cs.HasExpandVolume
 	info.SupportsListVolumesAttachedNodes = cs.HasListVolumesPublishedNodes
+	info.SupportsCondition = cs.HasVolumeCondition
+	info.SupportsGet = cs.HasGetVolume
 }
 
 func (p *pluginFingerprinter) buildControllerFingerprint(ctx context.Context, base *structs.CSIInfo) (*structs.CSIInfo, error) {

nomad/csi_endpoint.go

@@ -441,9 +441,9 @@ func (v *CSIVolume) controllerPublishVolume(req *structs.CSIVolumeClaimRequest,
 		return fmt.Errorf("%s: %s", structs.ErrUnknownAllocationPrefix, req.AllocationID)
 	}
 
-	// if no plugin was returned then controller validation is not required.
-	// Here we can return nil.
-	if plug == nil {
+	// if no plugin was returned or the plugin doesn't attach volumes, then
+	// controller validation is not required and we can safely return nil.
+	if plug == nil || !plug.HasControllerCapability(structs.CSIControllerSupportsAttachDetach) {
 		return nil
 	}
 
@@ -679,11 +679,21 @@ func (v *CSIVolume) controllerUnpublishVolume(vol *structs.CSIVolume, claim *str
 		return nil
 	}
 
+	state := v.srv.fsm.State()
+	ws := memdb.NewWatchSet()
+
+	plugin, err := state.CSIPluginByID(ws, vol.PluginID)
+	if err != nil || plugin == nil {
+		return fmt.Errorf("could not query plugin: %v", err)
+	}
+	if !plugin.HasControllerCapability(structs.CSIControllerSupportsAttachDetach) {
+		return nil
+	}
+
 	// we only send a controller detach if a Nomad client no longer has
 	// any claim to the volume, so we need to check the status of claimed
 	// allocations
-	state := v.srv.fsm.State()
-	vol, err := state.CSIVolumeDenormalize(memdb.NewWatchSet(), vol)
+	vol, err = state.CSIVolumeDenormalize(ws, vol)
 	if err != nil {
 		return err
 	}
@@ -834,9 +844,8 @@ func (v *CSIVolume) Create(args *structs.CSIVolumeCreateRequest, reply *structs.
 		if !plugin.ControllerRequired {
 			return fmt.Errorf("plugin has no controller")
 		}
-
-		if err := v.controllerValidateVolume(regArgs, vol, plugin); err != nil {
-			return err
+		if !plugin.HasControllerCapability(structs.CSIControllerSupportsCreateDelete) {
+			return fmt.Errorf("plugin does not support creating volumes")
 		}
 
 		validatedVols = append(validatedVols, validated{vol, plugin})

nomad/csi_endpoint_test.go

@@ -708,6 +708,7 @@ func TestCSIVolumeEndpoint_Create(t *testing.T) {
 			Healthy:  true,
 			ControllerInfo: &structs.CSIControllerInfo{
 				SupportsAttachDetach: true,
+				SupportsCreateDelete: true,
 			},
 			RequiresControllerPlugin: true,
 		},

nomad/structs/csi.go

@@ -846,14 +846,67 @@ func (p *CSIPlugin) Copy() *CSIPlugin {
 	return out
 }
 
+type CSIControllerCapability int
+
+const (
+	CSIControllerSupportsCreateDelete CSIControllerCapability = iota
+	CSIControllerSupportsAttachDetach
+	CSIControllerSupportsListVolumes
+	CSIControllerSupportsGetCapacity
+	CSIControllerSupportsCreateDeleteSnapshot
+	CSIControllerSupportsListSnapshots
+	CSIControllerSupportsClone
+	CSIControllerSupportsReadOnlyAttach
+	CSIControllerSupportsExpand
+	CSIControllerSupportsListVolumesAttachedNodes
+	CSIControllerSupportsCondition
+	CSIControllerSupportsGet
+)
+
+func (p *CSIPlugin) HasControllerCapability(cap CSIControllerCapability) bool {
+	if len(p.Controllers) < 1 {
+		return false
+	}
+	// we're picking the first controller because they should be uniform
+	// across the same version of the plugin
+	for _, c := range p.Controllers {
+		switch cap {
+		case CSIControllerSupportsCreateDelete:
+			return c.ControllerInfo.SupportsCreateDelete
+		case CSIControllerSupportsAttachDetach:
+			return c.ControllerInfo.SupportsAttachDetach
+		case CSIControllerSupportsListVolumes:
+			return c.ControllerInfo.SupportsListVolumes
+		case CSIControllerSupportsGetCapacity:
+			return c.ControllerInfo.SupportsGetCapacity
+		case CSIControllerSupportsCreateDeleteSnapshot:
+			return c.ControllerInfo.SupportsCreateDeleteSnapshot
+		case CSIControllerSupportsListSnapshots:
+			return c.ControllerInfo.SupportsListSnapshots
+		case CSIControllerSupportsClone:
+			return c.ControllerInfo.SupportsClone
+		case CSIControllerSupportsReadOnlyAttach:
+			return c.ControllerInfo.SupportsReadOnlyAttach
+		case CSIControllerSupportsExpand:
+			return c.ControllerInfo.SupportsExpand
+		case CSIControllerSupportsListVolumesAttachedNodes:
+			return c.ControllerInfo.SupportsListVolumesAttachedNodes
+		case CSIControllerSupportsCondition:
+			return c.ControllerInfo.SupportsCondition
+		case CSIControllerSupportsGet:
+			return c.ControllerInfo.SupportsGet
+		default:
+			return false
+		}
+	}
+	return false
+}
+
 // AddPlugin adds a single plugin running on the node. Called from state.NodeUpdate in a
 // transaction
 func (p *CSIPlugin) AddPlugin(nodeID string, info *CSIInfo) error {
 	if info.ControllerInfo != nil {
-		p.ControllerRequired = info.RequiresControllerPlugin &&
-			(info.ControllerInfo.SupportsAttachDetach ||
-				info.ControllerInfo.SupportsReadOnlyAttach)
-
+		p.ControllerRequired = info.RequiresControllerPlugin
 		prev, ok := p.Controllers[nodeID]
 		if ok {
 			if prev == nil {

nomad/structs/node.go

@@ -112,23 +112,50 @@ func (n *CSINodeInfo) Copy() *CSINodeInfo {
 // CSIControllerInfo is the fingerprinted data from a CSI Plugin that is specific to
 // the Controller API.
 type CSIControllerInfo struct {
-	// SupportsReadOnlyAttach is set to true when the controller returns the
-	// ATTACH_READONLY capability.
-	SupportsReadOnlyAttach bool
 
-	// SupportsPublishVolume is true when the controller implements the methods
-	// required to attach and detach volumes. If this is false Nomad should skip
-	// the controller attachment flow.
+	// SupportsCreateDelete indicates plugin support for CREATE_DELETE_VOLUME
+	SupportsCreateDelete bool
+
+	// SupportsPublishVolume is true when the controller implements the
+	// methods required to attach and detach volumes. If this is false Nomad
+	// should skip the controller attachment flow.
 	SupportsAttachDetach bool
 
-	// SupportsListVolumes is true when the controller implements the ListVolumes
-	// RPC. NOTE: This does not guaruntee that attached nodes will be returned
-	// unless SupportsListVolumesAttachedNodes is also true.
+	// SupportsListVolumes is true when the controller implements the
+	// ListVolumes RPC. NOTE: This does not guarantee that attached nodes will
+	// be returned unless SupportsListVolumesAttachedNodes is also true.
 	SupportsListVolumes bool
 
-	// SupportsListVolumesAttachedNodes indicates whether the plugin will return
-	// attached nodes data when making ListVolume RPCs
+	// SupportsGetCapacity indicates plugin support for GET_CAPACITY
+	SupportsGetCapacity bool
+
+	// SupportsCreateDeleteSnapshot indicates plugin support for
+	// CREATE_DELETE_SNAPSHOT
+	SupportsCreateDeleteSnapshot bool
+
+	// SupportsListSnapshots indicates plugin support for LIST_SNAPSHOTS
+	SupportsListSnapshots bool
+
+	// SupportsClone indicates plugin support for CLONE_VOLUME
+	SupportsClone bool
+
+	// SupportsReadOnlyAttach is set to true when the controller returns the
+	// ATTACH_READONLY capability.
+	SupportsReadOnlyAttach bool
+
+	// SupportsExpand indicates plugin support for EXPAND_VOLUME
+	SupportsExpand bool
+
+	// SupportsListVolumesAttachedNodes indicates whether the plugin will
+	// return attached nodes data when making ListVolume RPCs (plugin support
+	// for LIST_VOLUMES_PUBLISHED_NODES)
 	SupportsListVolumesAttachedNodes bool
+
+	// SupportsCondition indicates plugin support for VOLUME_CONDITION
+	SupportsCondition bool
+
+	// SupportsGet indicates plugin support for GET_VOLUME
+	SupportsGet bool
 }
 
 func (c *CSIControllerInfo) Copy() *CSIControllerInfo {

plugins/csi/client_test.go

@@ -222,7 +222,7 @@ func TestClient_RPC_ControllerGetCapabilities(t *testing.T) {
 					{
 						Type: &csipbv1.ControllerServiceCapability_Rpc{
 							Rpc: &csipbv1.ControllerServiceCapability_RPC{
-								Type: csipbv1.ControllerServiceCapability_RPC_GET_CAPACITY,
+								Type: csipbv1.ControllerServiceCapability_RPC_UNKNOWN,
 							},
 						},
 					},

plugins/csi/plugin.go

@@ -280,10 +280,18 @@ func NewPluginCapabilitySet(capabilities *csipbv1.GetPluginCapabilitiesResponse)
 }
 
 type ControllerCapabilitySet struct {
+	HasCreateDeleteVolume        bool
 	HasPublishUnpublishVolume    bool
-	HasPublishReadonly           bool
 	HasListVolumes               bool
+	HasGetCapacity               bool
+	HasCreateDeleteSnapshot      bool
+	HasListSnapshots             bool
+	HasCloneVolume               bool
+	HasPublishReadonly           bool
+	HasExpandVolume              bool
 	HasListVolumesPublishedNodes bool
+	HasVolumeCondition           bool
+	HasGetVolume                 bool
 }
 
 func NewControllerCapabilitySet(resp *csipbv1.ControllerGetCapabilitiesResponse) *ControllerCapabilitySet {
@@ -293,14 +301,30 @@ func NewControllerCapabilitySet(resp *csipbv1.ControllerGetCapabilitiesResponse)
 	for _, pcap := range pluginCapabilities {
 		if c := pcap.GetRpc(); c != nil {
 			switch c.Type {
+			case csipbv1.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME:
+				cs.HasCreateDeleteVolume = true
 			case csipbv1.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME:
 				cs.HasPublishUnpublishVolume = true
-			case csipbv1.ControllerServiceCapability_RPC_PUBLISH_READONLY:
-				cs.HasPublishReadonly = true
 			case csipbv1.ControllerServiceCapability_RPC_LIST_VOLUMES:
 				cs.HasListVolumes = true
+			case csipbv1.ControllerServiceCapability_RPC_GET_CAPACITY:
+				cs.HasGetCapacity = true
+			case csipbv1.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT:
+				cs.HasCreateDeleteSnapshot = true
+			case csipbv1.ControllerServiceCapability_RPC_LIST_SNAPSHOTS:
+				cs.HasListSnapshots = true
+			case csipbv1.ControllerServiceCapability_RPC_CLONE_VOLUME:
+				cs.HasCloneVolume = true
+			case csipbv1.ControllerServiceCapability_RPC_PUBLISH_READONLY:
+				cs.HasPublishReadonly = true
+			case csipbv1.ControllerServiceCapability_RPC_EXPAND_VOLUME:
+				cs.HasExpandVolume = true
 			case csipbv1.ControllerServiceCapability_RPC_LIST_VOLUMES_PUBLISHED_NODES:
 				cs.HasListVolumesPublishedNodes = true
+			case csipbv1.ControllerServiceCapability_RPC_VOLUME_CONDITION:
+				cs.HasVolumeCondition = true
+			case csipbv1.ControllerServiceCapability_RPC_GET_VOLUME:
+				cs.HasGetVolume = true
 			default:
 				continue
 			}