CSI: get plugin ID for creating snapshot from volume, not args

.changelog/12195.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+csi: Fixed a bug where creating snapshots required a plugin ID instead of falling back to the volume's plugin ID
+```

nomad/csi_endpoint.go

@@ -1095,29 +1095,34 @@ func (v *CSIVolume) CreateSnapshot(args *structs.CSISnapshotCreateRequest, reply
 			return fmt.Errorf("snapshot cannot be nil")
 		}
 
-		plugin, err := state.CSIPluginByID(nil, snap.PluginID)
+		vol, err := state.CSIVolumeByID(nil, args.RequestNamespace(), snap.SourceVolumeID)
 		if err != nil {
-			multierror.Append(&mErr,
-				fmt.Errorf("error querying plugin %q: %v", snap.PluginID, err))
+			multierror.Append(&mErr, fmt.Errorf("error querying volume %q: %v", snap.SourceVolumeID, err))
 			continue
 		}
-		if plugin == nil {
-			multierror.Append(&mErr, fmt.Errorf("no such plugin %q", snap.PluginID))
+		if vol == nil {
+			multierror.Append(&mErr, fmt.Errorf("no such volume %q", snap.SourceVolumeID))
 			continue
 		}
-		if !plugin.HasControllerCapability(structs.CSIControllerSupportsCreateDeleteSnapshot) {
-			multierror.Append(&mErr,
-				fmt.Errorf("plugin %q does not support snapshot", snap.PluginID))
-			continue
+
+		pluginID := snap.PluginID
+		if pluginID == "" {
+			pluginID = vol.PluginID
 		}
 
-		vol, err := state.CSIVolumeByID(nil, args.RequestNamespace(), snap.SourceVolumeID)
+		plugin, err := state.CSIPluginByID(nil, pluginID)
 		if err != nil {
-			multierror.Append(&mErr, fmt.Errorf("error querying volume %q: %v", snap.SourceVolumeID, err))
+			multierror.Append(&mErr,
+				fmt.Errorf("error querying plugin %q: %v", pluginID, err))
 			continue
 		}
-		if vol == nil {
-			multierror.Append(&mErr, fmt.Errorf("no such volume %q", snap.SourceVolumeID))
+		if plugin == nil {
+			multierror.Append(&mErr, fmt.Errorf("no such plugin %q", pluginID))
+			continue
+		}
+		if !plugin.HasControllerCapability(structs.CSIControllerSupportsCreateDeleteSnapshot) {
+			multierror.Append(&mErr,
+				fmt.Errorf("plugin %q does not support snapshot", pluginID))
 			continue
 		}
 
@@ -1127,7 +1132,7 @@ func (v *CSIVolume) CreateSnapshot(args *structs.CSISnapshotCreateRequest, reply
 			Secrets:                vol.Secrets,
 			Parameters:             snap.Parameters,
 		}
-		cReq.PluginID = plugin.ID
+		cReq.PluginID = pluginID
 		cResp := &cstructs.ClientCSIControllerCreateSnapshotResponse{}
 		err = v.srv.RPC(method, cReq, cResp)
 		if err != nil {

nomad/csi_endpoint_test.go

@@ -1199,7 +1199,6 @@ func TestCSIVolumeEndpoint_CreateSnapshot(t *testing.T) {
 			SourceVolumeID: "test-volume0",
 			Secrets:        structs.CSISecrets{"mysecret": "secretvalue"},
 			Parameters:     map[string]string{"myparam": "paramvalue"},
-			PluginID:       "minnie",
 		}},
 		WriteRequest: structs.WriteRequest{
 			Region:    "global",