scheduler: volume updates should always be destructive

.changelog/13008.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+volumes: Fixed a bug where additions, updates, or removals of host volumes or CSI volumes were not treated as destructive updates
+```

scheduler/util.go

@@ -524,6 +524,12 @@ func tasksUpdated(jobA, jobB *structs.Job, taskGroup string) bool {
 		return true
 	}
 
+	// Check if volumes are updated (no task driver can support
+	// altering mounts in-place)
+	if !reflect.DeepEqual(a.Volumes, b.Volumes) {
+		return true
+	}
+
 	// Check each task
 	for _, at := range a.Tasks {
 		bt := b.LookupTask(at.Name)
@@ -554,6 +560,9 @@ func tasksUpdated(jobA, jobB *structs.Job, taskGroup string) bool {
 		if !reflect.DeepEqual(at.CSIPluginConfig, bt.CSIPluginConfig) {
 			return true
 		}
+		if !reflect.DeepEqual(at.VolumeMounts, bt.VolumeMounts) {
+			return true
+		}
 
 		// Check the metadata
 		if !reflect.DeepEqual(

scheduler/util_test.go

@@ -811,6 +811,32 @@ func TestTasksUpdated(t *testing.T) {
 	// Compare changed Template wait configs
 	j23.TaskGroups[0].Tasks[0].Templates[0].Wait.Max = helper.TimeToPtr(10 * time.Second)
 	require.True(t, tasksUpdated(j22, j23, name))
+
+	// Add a volume
+	j24 := mock.Job()
+	j25 := j24.Copy()
+	j25.TaskGroups[0].Volumes = map[string]*structs.VolumeRequest{
+		"myvolume": {
+			Name:   "myvolume",
+			Type:   "csi",
+			Source: "test-volume[0]",
+		}}
+	require.True(t, tasksUpdated(j24, j25, name))
+
+	// Alter a volume
+	j26 := j25.Copy()
+	j26.TaskGroups[0].Volumes["myvolume"].ReadOnly = true
+	require.True(t, tasksUpdated(j25, j26, name))
+
+	// Alter a CSI plugin
+	j27 := mock.Job()
+	j27.TaskGroups[0].Tasks[0].CSIPluginConfig = &structs.TaskCSIPluginConfig{
+		ID:   "myplugin",
+		Type: "node",
+	}
+	j28 := j27.Copy()
+	j28.TaskGroups[0].Tasks[0].CSIPluginConfig.Type = "monolith"
+	require.True(t, tasksUpdated(j27, j28, name))
 }
 
 func TestTasksUpdated_connectServiceUpdated(t *testing.T) {