volumes: return better error messages for unsupported task drivers (#8030)

When an allocation runs for a task driver that can't support volume mounts,
the mounting will fail in a way that can be hard to understand. With host
volumes this usually means failing silently, whereas with CSI the operator
gets inscrutable internals exposed in the `nomad alloc status`.

This changeset adds a MountConfig field to the task driver Capabilities
response. We validate this when the `csi_hook` or `volume_hook` fires and
return a user-friendly error.

Note that we don't currently have a way to get driver capabilities up to the
server, except through attributes. Validating this when the user initially
submits the jobspec would be even better than what we're doing here (and could
be useful for all our other capabilities), but that's out of scope for this
changeset.

Also note that the MountConfig enum starts with "supports all" in order to
support community plugins in a backwards compatible way, rather than cutting
them off from volume mounting unexpectedly.

client/allocrunner/taskrunner/volume_hook.go

@@ -121,6 +121,18 @@ func (h *volumeHook) prepareHostVolumes(req *interfaces.TaskPrestartRequest, vol
 		return nil, err
 	}
 
+	if len(hostVolumeMounts) > 0 {
+		caps, err := h.runner.DriverCapabilities()
+		if err != nil {
+			return nil, fmt.Errorf("could not validate task driver capabilities: %v", err)
+		}
+		if caps.MountConfigs == drivers.MountConfigSupportNone {
+			return nil, fmt.Errorf(
+				"task driver %q for %q does not support host volumes",
+				h.runner.task.Driver, h.runner.task.Name)
+		}
+	}
+
 	return hostVolumeMounts, nil
 }
 
@@ -167,6 +179,18 @@ func (h *volumeHook) prepareCSIVolumes(req *interfaces.TaskPrestartRequest, volu
 		}
 	}
 
+	if len(mounts) > 0 {
+		caps, err := h.runner.DriverCapabilities()
+		if err != nil {
+			return nil, fmt.Errorf("could not validate task driver capabilities: %v", err)
+		}
+		if caps.MountConfigs == drivers.MountConfigSupportNone {
+			return nil, fmt.Errorf(
+				"task driver %q for %q does not support CSI",
+				h.runner.task.Driver, h.runner.task.Name)
+		}
+	}
+
 	return mounts, nil
 }
 

client/allocrunner/taskrunner/volume_hook_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/nomad/nomad/mock"
 	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/hashicorp/nomad/plugins/drivers"
+	dtu "github.com/hashicorp/nomad/plugins/drivers/testutils"
 	"github.com/stretchr/testify/require"
 )
 
@@ -67,8 +68,11 @@ func TestVolumeHook_PartitionMountsByVolume_Works(t *testing.T) {
 }
 
 func TestVolumeHook_prepareCSIVolumes(t *testing.T) {
+
 	req := &interfaces.TaskPrestartRequest{
 		Task: &structs.Task{
+			Name:   "test",
+			Driver: "mock",
 			VolumeMounts: []*structs.VolumeMount{
 				{
 					Volume:      "foo",
@@ -85,31 +89,71 @@ func TestVolumeHook_prepareCSIVolumes(t *testing.T) {
 		},
 	}
 
-	tr := &TaskRunner{
-		allocHookResources: &cstructs.AllocHookResources{
-			CSIMounts: map[string]*csimanager.MountInfo{
-				"foo": {
-					Source: "/mnt/my-test-volume",
+	cases := []struct {
+		Name          string
+		Driver        drivers.DriverPlugin
+		Expected      []*drivers.MountConfig
+		ExpectedError string
+	}{
+		{
+			Name: "supported driver",
+			Driver: &dtu.MockDriver{
+				CapabilitiesF: func() (*drivers.Capabilities, error) {
+					return &drivers.Capabilities{
+						MountConfigs: drivers.MountConfigSupportAll,
+					}, nil
+				},
+			},
+			Expected: []*drivers.MountConfig{
+				{
+					HostPath: "/mnt/my-test-volume",
+					TaskPath: "/bar",
 				},
 			},
 		},
-	}
-
-	expected := []*drivers.MountConfig{
 		{
-			HostPath: "/mnt/my-test-volume",
-			TaskPath: "/bar",
+			Name: "unsupported driver",
+			Driver: &dtu.MockDriver{
+				CapabilitiesF: func() (*drivers.Capabilities, error) {
+					return &drivers.Capabilities{
+						MountConfigs: drivers.MountConfigSupportNone,
+					}, nil
+				},
+			},
+			ExpectedError: "task driver \"mock\" for \"test\" does not support CSI",
 		},
 	}
 
-	hook := &volumeHook{
-		logger: testlog.HCLogger(t),
-		alloc:  structs.MockAlloc(),
-		runner: tr,
+	for _, tc := range cases {
+		t.Run(tc.Name, func(t *testing.T) {
+
+			tr := &TaskRunner{
+				task:   req.Task,
+				driver: tc.Driver,
+				allocHookResources: &cstructs.AllocHookResources{
+					CSIMounts: map[string]*csimanager.MountInfo{
+						"foo": {
+							Source: "/mnt/my-test-volume",
+						},
+					},
+				},
+			}
+
+			hook := &volumeHook{
+				logger: testlog.HCLogger(t),
+				alloc:  structs.MockAlloc(),
+				runner: tr,
+			}
+			mounts, err := hook.prepareCSIVolumes(req, volumes)
+
+			if tc.ExpectedError != "" {
+				require.EqualError(t, err, tc.ExpectedError)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, tc.Expected, mounts)
+		})
 	}
-	mounts, err := hook.prepareCSIVolumes(req, volumes)
-	require.NoError(t, err)
-	require.Equal(t, expected, mounts)
 }
 
 func TestVolumeHook_Interpolation(t *testing.T) {