Prevent scheduling from mutating the jobDb (#2838)

* Make scheduling not mutate the jobDb

Preempted/evicted jobs get mutated as part of the scheduling loop

This causes a bug where preemptible jobs are always retried on the same node:
 - As the nodeSelector is mutated as part of scheduling to point at a specific node

To fix this, we now deepcopy the evicted jobs as part of the scheduling round. Meaning the jobDb objects don't get mutated and prevents these mutations being persisted between scheduling rounds

* Remove comment

* Remove comment

---------

Co-authored-by: owenthomas17 <2owen.thomas@gmail.com>

internal/scheduler/jobdb/job.go

@@ -3,6 +3,7 @@ package jobdb
 import (
 	"time"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/google/uuid"
 	"golang.org/x/exp/maps"
 	v1 "k8s.io/api/core/v1"
@@ -431,6 +432,21 @@ func (job *Job) WithJobSchedulingInfo(jobSchedulingInfo *schedulerobjects.JobSch
 	return j
 }
 
+func (job *Job) DeepCopy() *Job {
+	copiedSchedulingInfo := proto.Clone(job.JobSchedulingInfo()).(*schedulerobjects.JobSchedulingInfo)
+	j := job.WithJobSchedulingInfo(copiedSchedulingInfo)
+
+	j.runsById = maps.Clone(j.runsById)
+	for key, run := range j.runsById {
+		j.runsById[key] = run.DeepCopy()
+	}
+	if j.activeRun != nil {
+		j.activeRun = job.activeRun.DeepCopy()
+	}
+
+	return j
+}
+
 // copyJob makes a copy of the job
 func copyJob(j Job) *Job {
 	return &j

internal/scheduler/jobdb/job_test.go

@@ -290,6 +290,26 @@ func TestJob_TestWithCreated(t *testing.T) {
 	assert.Equal(t, int64(456), newJob.Created())
 }
 
+func TestJob_DeepCopy(t *testing.T) {
+	original := NewJob("test-job", "test-jobset", "test-queue", 2, schedulingInfo, true, 0, false, false, false, 3)
+	original = original.WithUpdatedRun(baseJobRun.DeepCopy())
+	expected := NewJob("test-job", "test-jobset", "test-queue", 2, schedulingInfo, true, 0, false, false, false, 3)
+	expected = expected.WithUpdatedRun(baseJobRun.DeepCopy())
+
+	result := original.DeepCopy()
+	assert.Equal(t, expected, result)
+	assert.Equal(t, expected, original)
+
+	// Modify and confirm original hasn't changed
+	result.activeRun.nodeName = "test"
+	result.runsById[baseJobRun.id].nodeName = "test"
+	result.queue = "test"
+	result.jobSchedulingInfo.Priority = 1
+
+	assert.NotEqual(t, expected, result)
+	assert.Equal(t, expected, original)
+}
+
 func TestJob_TestWithJobSchedulingInfo(t *testing.T) {
 	newSchedInfo := &schedulerobjects.JobSchedulingInfo{
 		ObjectRequirements: []*schedulerobjects.ObjectRequirements{

internal/scheduler/preempting_queue_scheduler.go

@@ -21,6 +21,7 @@ import (
 	schedulerconstraints "github.com/armadaproject/armada/internal/scheduler/constraints"
 	schedulercontext "github.com/armadaproject/armada/internal/scheduler/context"
 	"github.com/armadaproject/armada/internal/scheduler/interfaces"
+	"github.com/armadaproject/armada/internal/scheduler/jobdb"
 	"github.com/armadaproject/armada/internal/scheduler/nodedb"
 )
 
@@ -803,6 +804,13 @@ func (evi *Evictor) Evict(ctx context.Context, it nodedb.NodeIterator) (*Evictor
 		if err != nil {
 			return nil, err
 		}
+
+		for i, evictedJob := range evictedJobs {
+			if dbJob, ok := evictedJob.(*jobdb.Job); ok {
+				evictedJobs[i] = dbJob.DeepCopy()
+			}
+		}
+
 		for _, job := range evictedJobs {
 			evictedJobsById[job.GetId()] = job
 			nodeIdByJobId[job.GetId()] = node.Id