refactor: consolidate private methods for CSI RPC (#7702)

Follow-up for a method missed in the refactor for #7688. The
`volAndPluginLookup` method is only ever called from the server's `CSI`
RPC and never the `ClientCSI` RPC, so move it into that scope.

nomad/client_csi_endpoint.go

@@ -9,7 +9,6 @@ import (
 	log "github.com/hashicorp/go-hclog"
 	memdb "github.com/hashicorp/go-memdb"
 	cstructs "github.com/hashicorp/nomad/client/structs"
-	"github.com/hashicorp/nomad/nomad/structs"
 )
 
 // ClientCSI is used to forward RPC requests to the targed Nomad client's
@@ -120,34 +119,6 @@ func (a *ClientCSI) NodeDetachVolume(args *cstructs.ClientCSINodeDetachVolumeReq
 
 }
 
-func (srv *Server) volAndPluginLookup(namespace, volID string) (*structs.CSIPlugin, *structs.CSIVolume, error) {
-	state := srv.fsm.State()
-	ws := memdb.NewWatchSet()
-
-	vol, err := state.CSIVolumeByID(ws, namespace, volID)
-	if err != nil {
-		return nil, nil, err
-	}
-	if vol == nil {
-		return nil, nil, fmt.Errorf("volume not found: %s", volID)
-	}
-	if !vol.ControllerRequired {
-		return nil, vol, nil
-	}
-
-	// note: we do this same lookup in CSIVolumeByID but then throw
-	// away the pointer to the plugin rather than attaching it to
-	// the volume so we have to do it again here.
-	plug, err := state.CSIPluginByID(ws, vol.PluginID)
-	if err != nil {
-		return nil, nil, err
-	}
-	if plug == nil {
-		return nil, nil, fmt.Errorf("plugin not found: %s", vol.PluginID)
-	}
-	return plug, vol, nil
-}
-
 // nodeForController validates that the Nomad client node ID for
 // a plugin exists and is new enough to support client RPC. If no node
 // ID is passed, select a random node ID for the controller to load-balance

nomad/csi_endpoint.go

@@ -374,7 +374,7 @@ func (v *CSIVolume) Claim(args *structs.CSIVolumeClaimRequest, reply *structs.CS
 // controllerPublishVolume sends publish request to the CSI controller
 // plugin associated with a volume, if any.
 func (v *CSIVolume) controllerPublishVolume(req *structs.CSIVolumeClaimRequest, resp *structs.CSIVolumeClaimResponse) error {
-	plug, vol, err := v.srv.volAndPluginLookup(req.RequestNamespace(), req.VolumeID)
+	plug, vol, err := v.volAndPluginLookup(req.RequestNamespace(), req.VolumeID)
 	if err != nil {
 		return err
 	}
@@ -431,6 +431,34 @@ func (v *CSIVolume) controllerPublishVolume(req *structs.CSIVolumeClaimRequest,
 	return nil
 }
 
+func (v *CSIVolume) volAndPluginLookup(namespace, volID string) (*structs.CSIPlugin, *structs.CSIVolume, error) {
+	state := v.srv.fsm.State()
+	ws := memdb.NewWatchSet()
+
+	vol, err := state.CSIVolumeByID(ws, namespace, volID)
+	if err != nil {
+		return nil, nil, err
+	}
+	if vol == nil {
+		return nil, nil, fmt.Errorf("volume not found: %s", volID)
+	}
+	if !vol.ControllerRequired {
+		return nil, vol, nil
+	}
+
+	// note: we do this same lookup in CSIVolumeByID but then throw
+	// away the pointer to the plugin rather than attaching it to
+	// the volume so we have to do it again here.
+	plug, err := state.CSIPluginByID(ws, vol.PluginID)
+	if err != nil {
+		return nil, nil, err
+	}
+	if plug == nil {
+		return nil, nil, fmt.Errorf("plugin not found: %s", vol.PluginID)
+	}
+	return plug, vol, nil
+}
+
 // allowCSIMount is called on Job register to check mount permission
 func allowCSIMount(aclObj *acl.ACL, namespace string) bool {
 	return aclObj.AllowPluginRead() &&

nomad/csi_endpoint_test.go

@@ -583,19 +583,20 @@ func TestCSI_RPCVolumeAndPluginLookup(t *testing.T) {
 	require.NoError(t, err)
 
 	// has controller
-	plugin, vol, err := srv.volAndPluginLookup(structs.DefaultNamespace, id0)
+	c := srv.staticEndpoints.CSIVolume
+	plugin, vol, err := c.volAndPluginLookup(structs.DefaultNamespace, id0)
 	require.NotNil(t, plugin)
 	require.NotNil(t, vol)
 	require.NoError(t, err)
 
 	// no controller
-	plugin, vol, err = srv.volAndPluginLookup(structs.DefaultNamespace, id1)
+	plugin, vol, err = c.volAndPluginLookup(structs.DefaultNamespace, id1)
 	require.Nil(t, plugin)
 	require.NotNil(t, vol)
 	require.NoError(t, err)
 
 	// doesn't exist
-	plugin, vol, err = srv.volAndPluginLookup(structs.DefaultNamespace, id2)
+	plugin, vol, err = c.volAndPluginLookup(structs.DefaultNamespace, id2)
 	require.Nil(t, plugin)
 	require.Nil(t, vol)
 	require.EqualError(t, err, fmt.Sprintf("volume not found: %s", id2))