dynamic host volumes: account for other claims in capability check

scheduler/feasible.go

@@ -192,6 +192,11 @@ func (h *HostVolumeChecker) hasVolumes(n *structs.Node) bool {
 		return true
 	}
 
+	proposed, err := h.ctx.ProposedAllocs(n.ID)
+	if err != nil {
+		return false // only hit this on state store invariant failure
+	}
+
 	for _, req := range h.volumeReqs {
 		volCfg, ok := n.HostVolumes[req.Source]
 		if !ok {
@@ -207,18 +212,12 @@ func (h *HostVolumeChecker) hasVolumes(n *structs.Node) bool {
 				// raft entry completes
 				return false
 			}
-			if vol.State != structs.HostVolumeStateReady {
-				return false
-			}
-			var capOk bool
-			for _, cap := range vol.RequestedCapabilities {
-				if req.AccessMode == structs.CSIVolumeAccessMode(cap.AccessMode) &&
-					req.AttachmentMode == structs.CSIVolumeAttachmentMode(cap.AttachmentMode) {
-					capOk = true
-					break
-				}
-			}
-			if !capOk {
+			if !h.hostVolumeIsAvailable(vol,
+				structs.HostVolumeAccessMode(req.AccessMode),
+				structs.HostVolumeAttachmentMode(req.AttachmentMode),
+				req.ReadOnly,
+				proposed,
+			) {
 				return false
 			}
 
@@ -242,6 +241,84 @@ func (h *HostVolumeChecker) hasVolumes(n *structs.Node) bool {
 	return true
 }
 
+// hostVolumeIsAvailable determines if a dynamic host volume is available for a request
+func (h *HostVolumeChecker) hostVolumeIsAvailable(
+	vol *structs.HostVolume,
+	reqAccess structs.HostVolumeAccessMode,
+	reqAttach structs.HostVolumeAttachmentMode,
+	readOnly bool,
+	proposed []*structs.Allocation) bool {
+
+	if vol.State != structs.HostVolumeStateReady {
+		return false
+	}
+
+	// pick a default capability based on the read-only flag. this happens here
+	// in the scheduler rather than job submit because we don't know whether a
+	// host volume is dynamic or not until we try to schedule it (ex. the same
+	// name could be static on one node and dynamic on another)
+	if reqAccess == structs.HostVolumeAccessModeUnknown {
+		if readOnly {
+			reqAccess = structs.HostVolumeAccessModeSingleNodeReader
+		} else {
+			reqAccess = structs.HostVolumeAccessModeSingleNodeWriter
+		}
+	}
+	if reqAttach == structs.HostVolumeAttachmentModeUnknown {
+		reqAttach = structs.HostVolumeAttachmentModeFilesystem
+	}
+
+	// check that the volume has the requested capability at all
+	var capOk bool
+	for _, cap := range vol.RequestedCapabilities {
+		if reqAccess == cap.AccessMode &&
+			reqAttach == cap.AttachmentMode {
+			capOk = true
+			break
+		}
+	}
+	if !capOk {
+		return false
+	}
+
+	switch reqAccess {
+	case structs.HostVolumeAccessModeSingleNodeReader:
+		return readOnly
+	case structs.HostVolumeAccessModeSingleNodeWriter:
+		return !readOnly
+	case structs.HostVolumeAccessModeSingleNodeSingleWriter:
+		// examine all proposed allocs on the node, including those that might
+		// not have yet been persisted. they have nil pointers to their Job, so
+		// we have to go back to the state store to get them
+		seen := map[structs.NamespacedID]struct{}{}
+		for _, alloc := range proposed {
+			if _, ok := seen[alloc.JobNamespacedID()]; ok {
+				// all allocs for the same job will have the same read-only flag
+				// and capabilities, so we only need to check a given job once
+				continue
+			}
+			seen[alloc.JobNamespacedID()] = struct{}{}
+			job, err := h.ctx.State().JobByID(nil, alloc.Namespace, alloc.JobID)
+			if err != nil {
+				return false
+			}
+			tg := job.LookupTaskGroup(alloc.TaskGroup)
+			for _, req := range tg.Volumes {
+				if req.Type == structs.VolumeTypeHost && req.Source == vol.Name {
+					if !req.ReadOnly {
+						return false
+					}
+				}
+			}
+		}
+
+	case structs.HostVolumeAccessModeSingleNodeMultiWriter:
+		// no contraint
+	}
+
+	return true
+}
+
 type CSIVolumeChecker struct {
 	ctx       Context
 	namespace string

scheduler/feasible_test.go

@@ -91,7 +91,7 @@ func TestRandomIterator(t *testing.T) {
 	}
 }
 
-func TestHostVolumeChecker(t *testing.T) {
+func TestHostVolumeChecker_Static(t *testing.T) {
 	ci.Parallel(t)
 
 	_, ctx := testContext(t)
@@ -184,7 +184,7 @@ func TestHostVolumeChecker(t *testing.T) {
 	}
 }
 
-func TestHostVolumeChecker_ReadOnly(t *testing.T) {
+func TestHostVolumeChecker_Dynamic(t *testing.T) {
 	ci.Parallel(t)
 
 	store, ctx := testContext(t)
@@ -284,6 +284,7 @@ func TestHostVolumeChecker_ReadOnly(t *testing.T) {
 		"foo": {
 			Type:           "host",
 			Source:         "foo",
+			ReadOnly:       true,
 			AccessMode:     structs.CSIVolumeAccessModeSingleNodeReader,
 			AttachmentMode: structs.CSIVolumeAttachmentModeFilesystem,
 		},
@@ -475,6 +476,152 @@ func TestHostVolumeChecker_Sticky(t *testing.T) {
 	}
 }
 
+// TestDynamicHostVolumeIsAvailable provides fine-grained coverage of the
+// dynamicHostVolumeIsAvailable function
+func TestDynamicHostVolumeIsAvailable(t *testing.T) {
+
+	store, ctx := testContext(t)
+
+	allCaps := []*structs.HostVolumeCapability{}
+
+	for _, accessMode := range []structs.HostVolumeAccessMode{
+		structs.HostVolumeAccessModeSingleNodeReader,
+		structs.HostVolumeAccessModeSingleNodeWriter,
+		structs.HostVolumeAccessModeSingleNodeSingleWriter,
+		structs.HostVolumeAccessModeSingleNodeMultiWriter,
+	} {
+		for _, attachMode := range []structs.HostVolumeAttachmentMode{
+			structs.HostVolumeAttachmentModeFilesystem,
+			structs.HostVolumeAttachmentModeBlockDevice,
+		} {
+			allCaps = append(allCaps, &structs.HostVolumeCapability{
+				AttachmentMode: attachMode,
+				AccessMode:     accessMode,
+			})
+		}
+	}
+
+	jobReader, jobWriter := mock.Job(), mock.Job()
+	jobReader.TaskGroups[0].Volumes = map[string]*structs.VolumeRequest{
+		"example": {
+			Type:     structs.VolumeTypeHost,
+			Source:   "example",
+			ReadOnly: true,
+		},
+	}
+	jobWriter.TaskGroups[0].Volumes = map[string]*structs.VolumeRequest{
+		"example": {
+			Type:   structs.VolumeTypeHost,
+			Source: "example",
+		},
+	}
+	index, _ := store.LatestIndex()
+	index++
+	must.NoError(t, store.UpsertJob(structs.MsgTypeTestSetup, index, nil, jobReader))
+	index++
+	must.NoError(t, store.UpsertJob(structs.MsgTypeTestSetup, index, nil, jobWriter))
+
+	allocReader0, allocReader1 := mock.Alloc(), mock.Alloc()
+	allocReader0.JobID = jobReader.ID
+	allocReader1.JobID = jobReader.ID
+
+	allocWriter0, allocWriter1 := mock.Alloc(), mock.Alloc()
+	allocWriter0.JobID = jobWriter.ID
+	allocWriter1.JobID = jobWriter.ID
+
+	index++
+	must.NoError(t, store.UpsertAllocs(structs.MsgTypeTestSetup, index,
+		[]*structs.Allocation{allocReader0, allocReader1, allocWriter0, allocWriter1}))
+
+	testCases := []struct {
+		name        string
+		hasProposed []*structs.Allocation
+		hasCaps     []*structs.HostVolumeCapability
+		wantAccess  structs.HostVolumeAccessMode
+		wantAttach  structs.HostVolumeAttachmentMode
+		readOnly    bool
+		expect      bool
+	}{
+		{
+			name: "enforce attachment mode",
+			hasCaps: []*structs.HostVolumeCapability{{
+				AttachmentMode: structs.HostVolumeAttachmentModeBlockDevice,
+				AccessMode:     structs.HostVolumeAccessModeSingleNodeSingleWriter,
+			}},
+			wantAttach: structs.HostVolumeAttachmentModeFilesystem,
+			wantAccess: structs.HostVolumeAccessModeSingleNodeSingleWriter,
+			expect:     false,
+		},
+		{
+			name:        "enforce read only",
+			hasProposed: []*structs.Allocation{allocReader0, allocReader1},
+			wantAttach:  structs.HostVolumeAttachmentModeFilesystem,
+			wantAccess:  structs.HostVolumeAccessModeSingleNodeReader,
+			expect:      false,
+		},
+		{
+			name:        "enforce read only ok",
+			hasProposed: []*structs.Allocation{allocReader0, allocReader1},
+			wantAttach:  structs.HostVolumeAttachmentModeFilesystem,
+			wantAccess:  structs.HostVolumeAccessModeSingleNodeReader,
+			readOnly:    true,
+			expect:      true,
+		},
+		{
+			name:        "enforce single writer",
+			hasProposed: []*structs.Allocation{allocReader0, allocReader1, allocWriter0},
+			wantAttach:  structs.HostVolumeAttachmentModeFilesystem,
+			wantAccess:  structs.HostVolumeAccessModeSingleNodeSingleWriter,
+			expect:      false,
+		},
+		{
+			name:        "enforce single writer ok",
+			hasProposed: []*structs.Allocation{allocReader0, allocReader1},
+			wantAttach:  structs.HostVolumeAttachmentModeFilesystem,
+			wantAccess:  structs.HostVolumeAccessModeSingleNodeSingleWriter,
+			expect:      true,
+		},
+		{
+			name:        "multi writer is always ok",
+			hasProposed: []*structs.Allocation{allocReader0, allocWriter0, allocWriter1},
+			wantAttach:  structs.HostVolumeAttachmentModeFilesystem,
+			wantAccess:  structs.HostVolumeAccessModeSingleNodeMultiWriter,
+			expect:      true,
+		},
+		{
+			name:   "default capabilities ok",
+			expect: true,
+		},
+		{
+			name:     "default capabilities fail",
+			readOnly: true,
+			hasCaps: []*structs.HostVolumeCapability{{
+				AttachmentMode: structs.HostVolumeAttachmentModeBlockDevice,
+				AccessMode:     structs.HostVolumeAccessModeSingleNodeSingleWriter,
+			}},
+			expect: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			vol := &structs.HostVolume{
+				Name:  "example",
+				State: structs.HostVolumeStateReady,
+			}
+			if len(tc.hasCaps) > 0 {
+				vol.RequestedCapabilities = tc.hasCaps
+			} else {
+				vol.RequestedCapabilities = allCaps
+			}
+			checker := NewHostVolumeChecker(ctx)
+			must.Eq(t, tc.expect, checker.hostVolumeIsAvailable(
+				vol, tc.wantAccess, tc.wantAttach, tc.readOnly, tc.hasProposed))
+		})
+	}
+
+}
+
 func TestCSIVolumeChecker(t *testing.T) {
 	ci.Parallel(t)
 	state, ctx := testContext(t)

scheduler/generic_sched_test.go

@@ -218,7 +218,7 @@ func TestServiceSched_JobRegister_StickyAllocs(t *testing.T) {
 	}
 }
 
-func TestServiceSched_JobRegister_StickyVolumes(t *testing.T) {
+func TestServiceSched_JobRegister_StickyHostVolumes(t *testing.T) {
 	ci.Parallel(t)
 
 	h := NewHarness(t)