csi: provide CSI_ENDPOINT env var to plugins

The CSI specification says:
> The CO SHALL provide the listen-address for the Plugin by way of the
`CSI_ENDPOINT` environment variable.

Note that plugins without filesystem isolation won't have the plugin
dir bind-mounted to their alloc dir, but we can provide a path to the
socket anyways.

.changelog/12050.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+csi: provide `CSI_ENDPOINT` environment variable to plugin tasks
+```

client/allocrunner/taskrunner/plugin_supervisor_hook.go

@@ -34,6 +34,8 @@ type csiPluginSupervisorHook struct {
 	runner     *TaskRunner
 	mountPoint string
 
+	caps *drivers.Capabilities
+
 	// eventEmitter is used to emit events to the task
 	eventEmitter ti.EventEmitter
 
@@ -61,7 +63,7 @@ var _ interfaces.TaskPoststartHook = &csiPluginSupervisorHook{}
 // with the catalog and to ensure any mounts are cleaned up.
 var _ interfaces.TaskStopHook = &csiPluginSupervisorHook{}
 
-func newCSIPluginSupervisorHook(csiRootDir string, eventEmitter ti.EventEmitter, runner *TaskRunner, logger hclog.Logger) *csiPluginSupervisorHook {
+func newCSIPluginSupervisorHook(csiRootDir string, eventEmitter ti.EventEmitter, runner *TaskRunner, caps *drivers.Capabilities, logger hclog.Logger) *csiPluginSupervisorHook {
 	task := runner.Task()
 
 	// The Plugin directory will look something like this:
@@ -82,6 +84,7 @@ func newCSIPluginSupervisorHook(csiRootDir string, eventEmitter ti.EventEmitter,
 		logger:           logger,
 		task:             task,
 		mountPoint:       pluginRoot,
+		caps:             caps,
 		shutdownCtx:      shutdownCtx,
 		shutdownCancelFn: cancelFn,
 		eventEmitter:     eventEmitter,
@@ -119,6 +122,20 @@ func (h *csiPluginSupervisorHook) Prestart(ctx context.Context,
 		Readonly: false,
 	}
 
+	switch h.caps.FSIsolation {
+	case drivers.FSIsolationNone:
+		// Plugin tasks with no filesystem isolation won't have the
+		// plugin dir bind-mounted to their alloc dir, but we can
+		// provide them the path to the socket. These Nomad-only
+		// plugins will need to be aware of the csi directory layout
+		// in the client data dir
+		resp.Env = map[string]string{
+			"CSI_ENDPOINT": filepath.Join(h.mountPoint, "csi.sock")}
+	default:
+		resp.Env = map[string]string{
+			"CSI_ENDPOINT": filepath.Join(h.task.CSIPluginConfig.MountDir, "csi.sock")}
+	}
+
 	mounts := ensureMountpointInserted(h.runner.hookResources.getMounts(), configMount)
 	mounts = ensureMountpointInserted(mounts, devMount)
 

client/allocrunner/taskrunner/task_runner_hooks.go

@@ -72,7 +72,7 @@ func (tr *TaskRunner) initHooks() {
 
 	// If the task has a CSI stanza, add the hook.
 	if task.CSIPluginConfig != nil {
-		tr.runnerHooks = append(tr.runnerHooks, newCSIPluginSupervisorHook(filepath.Join(tr.clientConfig.StateDir, "csi"), tr, tr, hookLogger))
+		tr.runnerHooks = append(tr.runnerHooks, newCSIPluginSupervisorHook(filepath.Join(tr.clientConfig.StateDir, "csi"), tr, tr, tr.driverCapabilities, hookLogger))
 	}
 
 	// If Vault is enabled, add the hook