scheduler/csi: fix early return when multiple volumes are requested

When multiple CSI volumes are requested, the feasibility check could return
early for read/write volumes with free claims, even if a later volume in the
request was not feasible for any other reason (including not existing at
all). This can result in random failure to fail feasibility checking,
depending on how the map of volumes was being ordered at runtime.

Remove the early return from the feasibility check. Add a test to verify that
missing volumes in the map will cause a failure; this test will not catch a
regression every test run because of the random map ordering, but any failure
will be caught over the course of several CI runs.

scheduler/feasible.go

@@ -312,15 +312,15 @@ func (c *CSIVolumeChecker) hasPlugins(n *structs.Node) (bool, string) {
 			if !vol.WriteSchedulable() {
 				return false, fmt.Sprintf(FilterConstraintCSIVolumeNoWriteTemplate, vol.ID)
 			}
-			if vol.WriteFreeClaims() {
-				return true, ""
-			}
-
-			// Check the blocking allocations to see if they belong to this job
-			for id := range vol.WriteAllocs {
-				a, err := c.ctx.State().AllocByID(ws, id)
-				if err != nil || a == nil || a.Namespace != c.namespace || a.JobID != c.jobID {
-					return false, fmt.Sprintf(FilterConstraintCSIVolumeInUseTemplate, vol.ID)
+			if !vol.WriteFreeClaims() {
+				// Check the blocking allocations to see if they belong to this job
+				for id := range vol.WriteAllocs {
+					a, err := c.ctx.State().AllocByID(ws, id)
+					if err != nil || a == nil ||
+						a.Namespace != c.namespace || a.JobID != c.jobID {
+						return false, fmt.Sprintf(
+							FilterConstraintCSIVolumeInUseTemplate, vol.ID)
+					}
 				}
 			}
 		}

scheduler/feasible_test.go

@@ -395,6 +395,23 @@ func TestCSIVolumeChecker(t *testing.T) {
 			t.Fatalf("case(%d) failed: got %v; want %v", i, act, c.Result)
 		}
 	}
+
+	// add a missing volume
+	volumes["missing"] = &structs.VolumeRequest{
+		Type:   "csi",
+		Name:   "bar",
+		Source: "does-not-exist",
+	}
+
+	checker = NewCSIVolumeChecker(ctx)
+	checker.SetNamespace(structs.DefaultNamespace)
+
+	for _, node := range nodes {
+		checker.SetVolumes(volumes)
+		act := checker.Feasible(node)
+		require.False(t, act, "request with missing volume should never be feasible")
+	}
+
 }
 
 func TestNetworkChecker(t *testing.T) {