csi_hook: Stage/Mount volumes as required

This commit introduces the first stage of volume mounting for an
allocation. The csimanager.VolumeMounter interface manages the blocking
and actual minutia of the CSI implementation allowing this hook to do
the minimal work of volume retrieval and creating mount info.

In the future the `CSIVolume.Get` request should be replaced by
`CSIVolume.Claim(Batch?)` to minimize the number of RPCs and to handle
external triggering of a ControllerPublishVolume request as required.

We also need to ensure that if pre-run hooks fail, we still get a full
unwinding of any publish and staged volumes to ensure that there are no hanging
references to volumes. That is not handled in this commit.

client/allocrunner/alloc_runner_hooks.go

@@ -134,7 +134,7 @@ func (ar *allocRunner) initRunnerHooks(config *clientconfig.Config) error {
 			logger:         hookLogger,
 		}),
 		newConsulSockHook(hookLogger, alloc, ar.allocDir, config.ConsulConfig),
-		newCSIHook(hookLogger, alloc, ar.rpcClient),
+		newCSIHook(hookLogger, alloc, ar.rpcClient, ar.csiManager),
 	}
 
 	return nil

client/allocrunner/csi_hook.go

@@ -1,7 +1,11 @@
 package allocrunner
 
 import (
+	"context"
+	"fmt"
+
 	hclog "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/nomad/client/pluginmanager/csimanager"
 	"github.com/hashicorp/nomad/nomad/structs"
 )
 
@@ -10,9 +14,10 @@ import (
 //
 // It is a noop for allocs that do not depend on CSI Volumes.
 type csiHook struct {
-	alloc     *structs.Allocation
-	logger    hclog.Logger
-	rpcClient RPCer
+	alloc      *structs.Allocation
+	logger     hclog.Logger
+	csimanager csimanager.Manager
+	rpcClient  RPCer
 }
 
 func (c *csiHook) Name() string {
@@ -24,15 +29,74 @@ func (c *csiHook) Prerun() error {
 		return nil
 	}
 
-	// TODO: Volume attachment flow
+	ctx := context.TODO()
+	volumes, err := c.csiVolumesFromAlloc()
+	if err != nil {
+		return err
+	}
+
+	mounts := make(map[string]*csimanager.MountInfo, len(volumes))
+	for alias, volume := range volumes {
+		mounter, err := c.csimanager.MounterForVolume(ctx, volume)
+		if err != nil {
+			return err
+		}
+
+		mountInfo, err := mounter.MountVolume(ctx, volume, c.alloc)
+		if err != nil {
+			return err
+		}
+
+		mounts[alias] = mountInfo
+	}
+
+	// TODO: Propagate mounts back to the tasks.
+
 	return nil
 }
 
-func newCSIHook(logger hclog.Logger, alloc *structs.Allocation, rpcClient RPCer) *csiHook {
+// csiVolumesFromAlloc finds all the CSI Volume requests from the allocation's
+// task group and then fetches them from the Nomad Server, before returning
+// them in the form of map[RequestedAlias]*structs.CSIVolume.
+//
+// If any volume fails to validate then we return an error.
+func (c *csiHook) csiVolumesFromAlloc() (map[string]*structs.CSIVolume, error) {
+	vols := make(map[string]*structs.VolumeRequest)
+	tg := c.alloc.Job.LookupTaskGroup(c.alloc.TaskGroup)
+	for alias, vol := range tg.Volumes {
+		if vol.Type == structs.VolumeTypeCSI {
+			vols[alias] = vol
+		}
+	}
+
+	csiVols := make(map[string]*structs.CSIVolume, len(vols))
+	for alias, request := range vols {
+		req := &structs.CSIVolumeGetRequest{
+			ID: request.Source,
+		}
+		req.Region = c.alloc.Job.Region
+
+		var resp structs.CSIVolumeGetResponse
+		if err := c.rpcClient.RPC("CSIVolume.Get", req, &resp); err != nil {
+			return nil, err
+		}
+
+		if resp.Volume == nil {
+			return nil, fmt.Errorf("Unexpected nil volume returned for ID: %v", request.Source)
+		}
+
+		csiVols[alias] = resp.Volume
+	}
+
+	return csiVols, nil
+}
+
+func newCSIHook(logger hclog.Logger, alloc *structs.Allocation, rpcClient RPCer, csi csimanager.Manager) *csiHook {
 	return &csiHook{
-		alloc:     alloc,
-		logger:    logger.Named("csi_hook"),
-		rpcClient: rpcClient,
+		alloc:      alloc,
+		logger:     logger.Named("csi_hook"),
+		rpcClient:  rpcClient,
+		csimanager: csi,
 	}
 }