Repository: armadaproject/armada
Pull request #3107 (2023-11-14 15:41:17)

Updated scheduler metrics


Patch:
diff --git a/internal/armada/server/submit_from_log.go b/internal/armada/server/submit_from_log.go
index 1d358320..c32060bf 100644
--- a/internal/armada/server/submit_from_log.go
+++ b/internal/armada/server/submit_from_log.go
@@ -19,6 +19,7 @@ import (
 	"github.com/armadaproject/armada/internal/common/eventutil"
 	"github.com/armadaproject/armada/internal/common/logging"
 	"github.com/armadaproject/armada/internal/common/schedulers"
+	armadaslices "github.com/armadaproject/armada/internal/common/slices"
 	"github.com/armadaproject/armada/internal/common/util"
 	"github.com/armadaproject/armada/pkg/api"
 	"github.com/armadaproject/armada/pkg/armadaevents"
@@ -355,7 +356,7 @@ func (srv *SubmitFromLog) SubmitJobs(
 	// We can't report job failure on error here, since the job failure message bundles the job struct.
 	// Hence, if an error occurs here, the job disappears from the point of view of the user.
 	// However, this code path is exercised when jobs are submitted to the log so errors should be rare.
-	es = util.Filter(es, func(e *armadaevents.SubmitJob) bool {
+	es = armadaslices.Filter(es, func(e *armadaevents.SubmitJob) bool {
 		return !e.IsDuplicate
 	})
 	jobs, err := eventutil.ApiJobsFromLogSubmitJobs(userId, groups, queueName, jobSetName, time.Now(), es)
diff --git a/internal/common/math/math.go b/internal/common/math/math.go
new file mode 100644
index 00000000..5553dce8
--- /dev/null
+++ b/internal/common/math/math.go
@@ -0,0 +1,12 @@
+package math
+
+import "golang.org/x/exp/constraints"
+
+// Max returns a if a > b and b otherwise.
+func Max[T constraints.Ordered](a, b T) T {
+	if a > b {
+		return a
+	} else {
+		return b
+	}
+}
diff --git a/internal/common/stringinterner/stringintern.go b/internal/common/stringinterner/stringintern.go
index df9290ba..e6fab776 100644
--- a/internal/common/stringinterner/stringintern.go
+++ b/internal/common/stringinterner/stringintern.go
@@ -14,12 +14,12 @@ type StringInterner struct {
 }
 
 // New return a new *StringInterner backed by a LRU of the given size.
-func New(cacheSize uint32) (*StringInterner, error) {
+func New(cacheSize uint32) *StringInterner {
 	lru, err := lru.New(int(cacheSize))
 	if err != nil {
-		return nil, errors.WithStack(err)
+		panic(errors.WithStack(err).Error())
 	}
-	return &StringInterner{lru: lru}, nil
+	return &StringInterner{lru: lru}
 }
 
 // Intern ensures the string is cached and returns the cached string
diff --git a/internal/common/util/list.go b/internal/common/util/list.go
index 8111a183..10416199 100644
--- a/internal/common/util/list.go
+++ b/internal/common/util/list.go
@@ -36,19 +36,6 @@ func DeepCopyListUint32(list []uint32) []uint32 {
 	return result
 }
 
-func Filter[S ~[]E, E any](s S, predicate func(e E) bool) S {
-	if s == nil {
-		return nil
-	}
-	out := make(S, 0, len(s))
-	for _, e := range s {
-		if predicate(e) {
-			out = append(out, e)
-		}
-	}
-	return out
-}
-
 func Concat[T any](slices ...[]T) []T {
 	total := 0
 	for _, s := range slices {
diff --git a/internal/scheduler/common.go b/internal/scheduler/common.go
index 02982ee3..1b89dddf 100644
--- a/internal/scheduler/common.go
+++ b/internal/scheduler/common.go
@@ -10,86 +10,10 @@ import (
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	armadamaps "github.com/armadaproject/armada/internal/common/maps"
 	armadaslices "github.com/armadaproject/armada/internal/common/slices"
-	schedulercontext "github.com/armadaproject/armada/internal/scheduler/context"
 	"github.com/armadaproject/armada/internal/scheduler/interfaces"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
 )
 
-// SchedulerResult is returned by Rescheduler.Schedule().
-type SchedulerResult struct {
-	// Whether the scheduler failed to create a result for some reason
-	EmptyResult bool
-	// Running jobs that should be preempted.
-	PreemptedJobs []interfaces.LegacySchedulerJob
-	// Queued jobs that should be scheduled.
-	ScheduledJobs []interfaces.LegacySchedulerJob
-	// Queued jobs that could not be scheduled.
-	// This is used to fail jobs that could not schedule above `minimumGangCardinality`.
-	FailedJobs []interfaces.LegacySchedulerJob
-	// For each preempted job, maps the job id to the id of the node on which the job was running.
-	// For each scheduled job, maps the job id to the id of the node on which the job should be scheduled.
-	NodeIdByJobId map[string]string
-	// The Scheduling Context. Being passed up for metrics decisions made in scheduler.go and scheduler_metrics.go.
-	// Passing a pointer as the structure is enormous
-	SchedulingContexts []*schedulercontext.SchedulingContext
-}
-
-func NewSchedulerResultForTest[S ~[]T, T interfaces.LegacySchedulerJob](
-	preemptedJobs S,
-	scheduledJobs S,
-	failedJobs S,
-	nodeIdByJobId map[string]string,
-) *SchedulerResult {
-	castPreemptedJobs := make([]interfaces.LegacySchedulerJob, len(preemptedJobs))
-	for i, job := range preemptedJobs {
-		castPreemptedJobs[i] = job
-	}
-	castScheduledJobs := make([]interfaces.LegacySchedulerJob, len(scheduledJobs))
-	for i, job := range scheduledJobs {
-		castScheduledJobs[i] = job
-	}
-	castFailedJobs := make([]interfaces.LegacySchedulerJob, len(failedJobs))
-	for i, job := range failedJobs {
-		castFailedJobs[i] = job
-	}
-	return &SchedulerResult{
-		PreemptedJobs: castPreemptedJobs,
-		ScheduledJobs: castScheduledJobs,
-		NodeIdByJobId: nodeIdByJobId,
-		FailedJobs:    castFailedJobs,
-	}
-}
-
-// PreemptedJobsFromSchedulerResult returns the slice of preempted jobs in the result,
-// cast to type T.
-func PreemptedJobsFromSchedulerResult[T interfaces.LegacySchedulerJob](sr *SchedulerResult) []T {
-	rv := make([]T, len(sr.PreemptedJobs))
-	for i, job := range sr.PreemptedJobs {
-		rv[i] = job.(T)
-	}
-	return rv
-}
-
-// ScheduledJobsFromScheduleResult returns the slice of scheduled jobs in the result,
-// cast to type T.
-func ScheduledJobsFromSchedulerResult[T interfaces.LegacySchedulerJob](sr *SchedulerResult) []T {
-	rv := make([]T, len(sr.ScheduledJobs))
-	for i, job := range sr.ScheduledJobs {
-		rv[i] = job.(T)
-	}
-	return rv
-}
-
-// FailedJobsFromScheduleResult returns the slice of scheduled jobs in the result,
-// cast to type T.
-func FailedJobsFromSchedulerResult[T interfaces.LegacySchedulerJob](sr *SchedulerResult) []T {
-	rv := make([]T, len(sr.FailedJobs))
-	for i, job := range sr.FailedJobs {
-		rv[i] = job.(T)
-	}
-	return rv
-}
-
 // JobsSummary returns a string giving an overview of the provided jobs meant for logging.
 // For example: "affected queues [A, B]; resources {A: {cpu: 1}, B: {cpu: 2}}; jobs [jobAId, jobBId]".
 func JobsSummary(jobs []interfaces.LegacySchedulerJob) string {
diff --git a/internal/scheduler/configuration/configuration.go b/internal/scheduler/configuration/configuration.go
index 8df4477b..964498dd 100644
--- a/internal/scheduler/configuration/configuration.go
+++ b/internal/scheduler/configuration/configuration.go
@@ -3,6 +3,8 @@ package configuration
 import (
 	"time"
 
+	v1 "k8s.io/api/core/v1"
+
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	authconfig "github.com/armadaproject/armada/internal/common/auth/configuration"
 	"github.com/armadaproject/armada/internal/common/config"
@@ -27,6 +29,9 @@ type Configuration struct {
 	Leader LeaderConfig
 	// Configuration controlling metrics
 	Metrics configuration.MetricsConfig
+	// Configuration for new scheduler metrics.
+	// Due to replace metrics configured via the above entry.
+	SchedulerMetrics MetricsConfig
 	// Scheduler configuration (this is shared with the old scheduler)
 	Scheduling configuration.SchedulingConfig
 	Auth       authconfig.AuthConfig
@@ -52,6 +57,39 @@ type Configuration struct {
 	PulsarSendTimeout time.Duration `validate:"required"`
 }
 
+type MetricsConfig struct {
+	// If true, disable metric collection and publishing.
+	Disabled bool
+	// The scheduler exports job failure counters labelled by (queue, cluster, node, reason0, reason1, ..., reasonN).
+	//
+	// Each reason label has a regex associated with it and the label value is set to "1"
+	// if this regex matches the error message associated with the job failure.
+	// For example, if TrackedErrorRegexes contains an entry, "isCudaError": "/CUDA/",
+	// then job failure metrics will have a label "isCudaError" with value either 0 or 1.
+	TrackedErrorRegexByLabel map[string]string
+	// Metrics are exported for these resources.
+	TrackedResourceNames []v1.ResourceName
+	// Controls the cycle time metrics.
+	// TODO(albin): Not used yet.
+	CycleTimeConfig PrometheusSummaryConfig
+}
+
+// PrometheusSummaryConfig contains the relevant config for a prometheus.Summary.
+type PrometheusSummaryConfig struct {
+	// Objectives defines the quantile rank estimates with their respective
+	// absolute error. If Objectives[q] = e, then the value reported for q
+	// will be the φ-quantile value for some φ between q-e and q+e.  The
+	// default value is an empty map, resulting in a summary without
+	// quantiles.
+	Objectives map[float64]float64
+
+	// MaxAge defines the duration for which an observation stays relevant
+	// for the summary. Only applies to pre-calculated quantiles, does not
+	// apply to _sum and _count. Must be positive. The default value is
+	// DefMaxAge.
+	MaxAge time.Duration
+}
+
 type LeaderConfig struct {
 	// Valid modes are "standalone" or "kubernetes"
 	Mode string `validate:"required"`
diff --git a/internal/scheduler/database/job_repository.go b/internal/scheduler/database/job_repository.go
index ebc08d03..b30e46eb 100644
--- a/internal/scheduler/database/job_repository.go
+++ b/internal/scheduler/database/job_repository.go
@@ -2,6 +2,7 @@ package database
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
@@ -37,8 +38,8 @@ type JobRepository interface {
 	// These updates are guaranteed to be consistent with each other
 	FetchJobUpdates(ctx *armadacontext.Context, jobSerial int64, jobRunSerial int64) ([]Job, []Run, error)
 
-	// FetchJobRunErrors returns all armadaevents.JobRunErrors for the provided job run ids.  The returned map is
-	// keyed by job run id.  Any dbRuns which don't have errors wil be absent from the map.
+	// FetchJobRunErrors returns all armadaevents.JobRunErrors for the provided job run ids. The returned map is
+	// keyed by job run id. Any dbRuns which don't have errors wil be absent from the map.
 	FetchJobRunErrors(ctx *armadacontext.Context, runIds []uuid.UUID) (map[uuid.UUID]*armadaevents.Error, error)
 
 	// CountReceivedPartitions returns a count of the number of partition messages present in the database corresponding
@@ -129,6 +130,9 @@ func (r *PostgresJobRepository) FetchJobUpdates(ctx *armadacontext.Context, jobS
 	var updatedJobs []Job = nil
 	var updatedRuns []Run = nil
 
+	start := time.Now()
+	defer ctx.Infof("received %d updated jobs and %d updated job runs from postgres in %s", len(updatedJobs), len(updatedRuns), time.Since(start))
+
 	// Use a RepeatableRead transaction here so that we get consistency between jobs and dbRuns
 	err := pgx.BeginTxFunc(ctx, r.db, pgx.TxOptions{
 		IsoLevel:       pgx.RepeatableRead,
diff --git a/internal/scheduler/jobdb/comparison.go b/internal/scheduler/jobdb/comparison.go
index 83d54346..c7278cf1 100644
--- a/internal/scheduler/jobdb/comparison.go
+++ b/internal/scheduler/jobdb/comparison.go
@@ -123,7 +123,7 @@ func SchedulingOrderCompare(job, other *Job) int {
 	}
 
 	// Tie-break by jobId, which must be unique.
-	// This ensure there is a total order between jobs, i.e., no jobs are equal from an ordering point of view.
+	// This ensures there is a total order between jobs, i.e., no jobs are equal from an ordering point of view.
 	if job.id < other.id {
 		return -1
 	} else if job.id > other.id {
diff --git a/internal/scheduler/jobdb/job_run.go b/internal/scheduler/jobdb/job_run.go
index f951f4f6..a9faee39 100644
--- a/internal/scheduler/jobdb/job_run.go
+++ b/internal/scheduler/jobdb/job_run.go
@@ -55,7 +55,7 @@ func MinimalRun(id uuid.UUID, creationTime int64) *JobRun {
 }
 
 // CreateRun creates a new scheduler job run from a database job run
-func CreateRun(
+func (jobDb *JobDb) CreateRun(
 	id uuid.UUID,
 	jobId string,
 	creationTime int64,
@@ -73,9 +73,9 @@ func CreateRun(
 		id:           id,
 		jobId:        jobId,
 		created:      creationTime,
-		executor:     executor,
-		nodeId:       nodeId,
-		nodeName:     nodeName,
+		executor:     jobDb.stringInterner.Intern(executor),
+		nodeId:       jobDb.stringInterner.Intern(nodeId),
+		nodeName:     jobDb.stringInterner.Intern(nodeName),
 		running:      running,
 		succeeded:    succeeded,
 		failed:       failed,
@@ -90,7 +90,7 @@ func (run *JobRun) Id() uuid.UUID {
 	return run.id
 }
 
-// Id returns the id of the job this run is associated with.
+// JobId returns the id of the job this run is associated with.
 func (run *JobRun) JobId() string {
 	return run.jobId
 }
@@ -105,7 +105,7 @@ func (run *JobRun) NodeId() string {
 	return run.nodeId
 }
 
-// NodeId returns the name of the node to which the JobRun is assigned.
+// NodeName returns the name of the node to which the JobRun is assigned.
 func (run *JobRun) NodeName() string {
 	return run.nodeName
 }
diff --git a/internal/scheduler/jobdb/jobdb.go b/internal/scheduler/jobdb/jobdb.go
index 7f010559..1ab68a59 100644
--- a/internal/scheduler/jobdb/jobdb.go
+++ b/internal/scheduler/jobdb/jobdb.go
@@ -9,6 +9,7 @@ import (
 	"github.com/pkg/errors"
 	"golang.org/x/exp/maps"
 
+	"github.com/armadaproject/armada/internal/common/stringinterner"
 	"github.com/armadaproject/armada/internal/common/types"
 	"github.com/armadaproject/armada/internal/scheduler/interfaces"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
@@ -29,15 +30,18 @@ type JobDb struct {
 	// Priority class assigned to jobs with a priorityClassName not in jobDb.priorityClasses.
 	defaultPriorityClass   types.PriorityClass
 	schedulingKeyGenerator *schedulerobjects.SchedulingKeyGenerator
-	copyMutex              sync.Mutex
-	writerMutex            sync.Mutex
+	// We intern strings to save memory.
+	stringInterner *stringinterner.StringInterner
+	copyMutex      sync.Mutex
+	writerMutex    sync.Mutex
 }
 
-func NewJobDb(priorityClasses map[string]types.PriorityClass, defaultPriorityClassName string) *JobDb {
+func NewJobDb(priorityClasses map[string]types.PriorityClass, defaultPriorityClassName string, stringInternerCacheSize uint32) *JobDb {
 	return NewJobDbWithSchedulingKeyGenerator(
 		priorityClasses,
 		defaultPriorityClassName,
 		schedulerobjects.NewSchedulingKeyGenerator(),
+		stringInternerCacheSize,
 	)
 }
 
@@ -45,10 +49,11 @@ func NewJobDbWithSchedulingKeyGenerator(
 	priorityClasses map[string]types.PriorityClass,
 	defaultPriorityClassName string,
 	skg *schedulerobjects.SchedulingKeyGenerator,
+	stringInternerCacheSize uint32,
 ) *JobDb {
 	defaultPriorityClass, ok := priorityClasses[defaultPriorityClassName]
 	if !ok {
-		// TODO: Return an error instead.
+		// TODO(albin): Return an error instead.
 		panic(fmt.Sprintf("unknown default priority class %s", defaultPriorityClassName))
 	}
 	return &JobDb{
@@ -59,6 +64,7 @@ func NewJobDbWithSchedulingKeyGenerator(
 		priorityClasses:        priorityClasses,
 		defaultPriorityClass:   defaultPriorityClass,
 		schedulingKeyGenerator: skg,
+		stringInterner:         stringinterner.New(stringInternerCacheSize),
 	}
 }
 
@@ -83,14 +89,14 @@ func (jobDb *JobDb) NewJob(
 	}
 	job := &Job{
 		id:                      jobId,
-		queue:                   queue,
-		jobSet:                  jobSet,
+		queue:                   jobDb.stringInterner.Intern(queue),
+		jobSet:                  jobDb.stringInterner.Intern(jobSet),
 		priority:                priority,
 		queued:                  queued,
 		queuedVersion:           queuedVersion,
 		requestedPriority:       priority,
 		submittedTime:           created,
-		jobSchedulingInfo:       schedulingInfo,
+		jobSchedulingInfo:       jobDb.internJobSchedulingInfoStrings(schedulingInfo),
 		priorityClass:           priorityClass,
 		cancelRequested:         cancelRequested,
 		cancelByJobSetRequested: cancelByJobSetRequested,
@@ -102,6 +108,22 @@ func (jobDb *JobDb) NewJob(
 	return job
 }
 
+func (jobDb *JobDb) internJobSchedulingInfoStrings(info *schedulerobjects.JobSchedulingInfo) *schedulerobjects.JobSchedulingInfo {
+	for _, requirement := range info.ObjectRequirements {
+		if podRequirement := requirement.GetPodRequirements(); podRequirement != nil {
+			for k, v := range podRequirement.Annotations {
+				podRequirement.Annotations[jobDb.stringInterner.Intern(k)] = jobDb.stringInterner.Intern(v)
+			}
+
+			for k, v := range podRequirement.NodeSelector {
+				podRequirement.NodeSelector[jobDb.stringInterner.Intern(k)] = jobDb.stringInterner.Intern(v)
+			}
+			podRequirement.PreemptionPolicy = jobDb.stringInterner.Intern(podRequirement.PreemptionPolicy)
+		}
+	}
+	return info
+}
+
 // ReadTxn returns a read-only transaction.
 // Multiple read-only transactions can access the db concurrently
 func (jobDb *JobDb) ReadTxn() *Txn {
diff --git a/internal/scheduler/jobdb/reconciliation.go b/internal/scheduler/jobdb/reconciliation.go
new file mode 100644
index 00000000..795b192a
--- /dev/null
+++ b/internal/scheduler/jobdb/reconciliation.go
@@ -0,0 +1,237 @@
+package jobdb
+
+import (
+	"github.com/gogo/protobuf/proto"
+	"github.com/pkg/errors"
+
+	armadamath "github.com/armadaproject/armada/internal/common/math"
+	armadaslices "github.com/armadaproject/armada/internal/common/slices"
+	"github.com/armadaproject/armada/internal/scheduler/database"
+	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
+	"github.com/armadaproject/armada/pkg/api"
+)
+
+// JobStateTransitions captures the process of updating a job.
+// It bundles the updated job with booleans indicating which state transitions were applied to produce it.
+// These are cumulative in the sense that a job with transitions queued -> scheduled -> queued -> running -> failed
+// will have the fields queued, scheduled, running, and failed set to true.
+type JobStateTransitions struct {
+	Job *Job
+
+	Queued    bool
+	Scheduled bool
+	Pending   bool
+	Running   bool
+	Cancelled bool
+	Preempted bool
+	Failed    bool
+	Succeeded bool
+}
+
+// applyRunStateTransitions applies the state transitions of a run to that of the associated job.
+func (jst JobStateTransitions) applyRunStateTransitions(rst RunStateTransitions) JobStateTransitions {
+	jst.Queued = jst.Queued || rst.Returned
+	jst.Scheduled = jst.Scheduled || rst.Scheduled
+	jst.Pending = jst.Pending || rst.Pending
+	jst.Running = jst.Running || rst.Running
+	jst.Cancelled = jst.Cancelled || rst.Cancelled
+	jst.Preempted = jst.Preempted || rst.Preempted
+	jst.Failed = jst.Failed || rst.Failed
+	jst.Succeeded = jst.Succeeded || rst.Succeeded
+	return jst
+}
+
+// RunStateTransitions captures the process of updating a run.
+// It works in the same way as JobStateTransitions does for jobs.
+type RunStateTransitions struct {
+	JobRun *JobRun
+
+	Scheduled bool
+	Returned  bool
+	Pending   bool
+	Running   bool
+	Cancelled bool
+	Preempted bool
+	Failed    bool
+	Succeeded bool
+}
+
+// ReconcileDifferences reconciles any differences between jobs stored in the jobDb with those provided to this function
+// and returns the updated jobs together with a summary of the state transitions applied to those jobs.
+func (jobDb *JobDb) ReconcileDifferences(txn *Txn, jobRepoJobs []database.Job, jobRepoRuns []database.Run) ([]JobStateTransitions, error) {
+	// Map jobs for which a run was updated to nil and jobs updated directly to the updated job.
+	jobRepoJobsById := make(map[string]*database.Job, armadamath.Max(len(jobRepoJobs), len(jobRepoRuns)))
+	for _, jobRepoRun := range jobRepoRuns {
+		jobRepoJobsById[jobRepoRun.JobID] = nil
+	}
+	for _, jobRepoJob := range jobRepoJobs {
+		jobRepoJob := jobRepoJob
+		jobRepoJobsById[jobRepoJob.JobID] = &jobRepoJob
+	}
+
+	// Group updated runs by the id of the job they're associated with.
+	jobRepoRunsById := armadaslices.MapAndGroupByFuncs(
+		jobRepoRuns,
+		func(jobRepoRun database.Run) string { return jobRepoRun.JobID },
+		func(jobRepoRun database.Run) *database.Run { return &jobRepoRun },
+	)
+
+	jsts := make([]JobStateTransitions, 0, len(jobRepoJobsById))
+	for jobId, jobRepoJob := range jobRepoJobsById {
+		if jst, err := jobDb.reconcileJobDifferences(
+			txn.GetById(jobId),     // Existing job in the jobDb.
+			jobRepoJob,             // New or updated job from the jobRepo.
+			jobRepoRunsById[jobId], // New or updated runs associated with this job from the jobRepo.
+		); err != nil {
+			return nil, err
+		} else {
+			jsts = append(jsts, jst)
+		}
+	}
+	return jsts, nil
+}
+
+// reconcileJobDifferences takes as its arguments for some job id
+// - the job currently stored in the jobDb, or nil, if there is no such job,
+// - the job stored in the job repository, or nil if there is no such job,
+// - a slice composed of the runs associated with the job stored in the job repository,
+// and returns a new jobdb.Job produced by reconciling any differences between the input jobs
+// along with a summary of the state transitions applied to the job.
+//
+// TODO(albin): Pending, running, and preempted are not supported yet.
+func (jobDb *JobDb) reconcileJobDifferences(job *Job, jobRepoJob *database.Job, jobRepoRuns []*database.Run) (jst JobStateTransitions, err error) {
+	defer func() { jst.Job = job }()
+	if job == nil && jobRepoJob == nil {
+		return
+	} else if job == nil && jobRepoJob != nil {
+		if job, err = jobDb.schedulerJobFromDatabaseJob(jobRepoJob); err != nil {
+			return
+		}
+		jst.Queued = true
+	} else if job != nil && jobRepoJob == nil {
+		// No direct updates to the job; just process any updated runs below.
+	} else if job != nil && jobRepoJob != nil {
+		if jobRepoJob.CancelRequested && !job.CancelRequested() {
+			job = job.WithCancelRequested(true)
+		}
+		if jobRepoJob.CancelByJobsetRequested && !job.CancelByJobsetRequested() {
+			job = job.WithCancelByJobsetRequested(true)
+		}
+		if jobRepoJob.Cancelled && !job.Cancelled() {
+			job = job.WithCancelled(true)
+		}
+		if jobRepoJob.Succeeded && !job.Succeeded() {
+			job = job.WithSucceeded(true)
+		}
+		if jobRepoJob.Failed && !job.Failed() {
+			job = job.WithFailed(true)
+		}
+		if uint32(jobRepoJob.Priority) != job.RequestedPriority() {
+			job = job.WithRequestedPriority(uint32(jobRepoJob.Priority))
+		}
+		if uint32(jobRepoJob.SchedulingInfoVersion) > job.JobSchedulingInfo().Version {
+			schedulingInfo := &schedulerobjects.JobSchedulingInfo{}
+			if err = proto.Unmarshal(jobRepoJob.SchedulingInfo, schedulingInfo); err != nil {
+				err = errors.Wrapf(err, "error unmarshalling scheduling info for job %s", jobRepoJob.JobID)
+				return
+			}
+			job = job.WithJobSchedulingInfo(schedulingInfo)
+		}
+		if jobRepoJob.QueuedVersion > job.QueuedVersion() {
+			job = job.WithQueuedVersion(jobRepoJob.QueuedVersion)
+			job = job.WithQueued(jobRepoJob.Queued)
+		}
+	}
+
+	// Reconcile run state transitions.
+	for _, jobRepoRun := range jobRepoRuns {
+		rst := jobDb.reconcileRunDifferences(job.RunById(jobRepoRun.RunID), jobRepoRun)
+		jst = jst.applyRunStateTransitions(rst)
+		job = job.WithUpdatedRun(rst.JobRun)
+	}
+
+	return
+}
+
+// TODO(albin): Preempted is not supported.
+func (jobDb *JobDb) reconcileRunDifferences(jobRun *JobRun, jobRepoRun *database.Run) (rst RunStateTransitions) {
+	defer func() { rst.JobRun = jobRun }()
+	if jobRun == nil && jobRepoRun == nil {
+		return
+	} else if jobRun == nil && jobRepoRun != nil {
+		jobRun = jobDb.schedulerRunFromDatabaseRun(jobRepoRun)
+		rst.Returned = jobRepoRun.Returned
+		rst.Pending = jobRepoRun.PendingTimestamp != nil
+		rst.Running = jobRepoRun.Running
+		rst.Cancelled = jobRepoRun.Cancelled
+		rst.Failed = jobRepoRun.Failed
+		rst.Succeeded = jobRepoRun.Succeeded
+	} else if jobRun != nil && jobRepoRun == nil {
+		return
+	} else if jobRun != nil && jobRepoRun != nil {
+		if jobRepoRun.Running && !jobRun.Running() {
+			jobRun = jobRun.WithRunning(true)
+			rst.Running = true
+		}
+		if jobRepoRun.Succeeded && !jobRun.Succeeded() {
+			jobRun = jobRun.WithSucceeded(true)
+			rst.Succeeded = true
+		}
+		if jobRepoRun.Failed && !jobRun.Failed() {
+			jobRun = jobRun.WithFailed(true)
+			rst.Failed = true
+		}
+		if jobRepoRun.Cancelled && !jobRun.Cancelled() {
+			jobRun = jobRun.WithCancelled(true)
+			rst.Cancelled = true
+		}
+		if jobRepoRun.Returned && !jobRun.Returned() {
+			jobRun = jobRun.WithReturned(true)
+			rst.Returned = true
+		}
+		if jobRepoRun.RunAttempted && !jobRun.RunAttempted() {
+			jobRun = jobRun.WithAttempted(true)
+		}
+	}
+	return
+}
+
+// schedulerJobFromDatabaseJob creates a new scheduler job from a database job.
+func (jobDb *JobDb) schedulerJobFromDatabaseJob(dbJob *database.Job) (*Job, error) {
+	schedulingInfo := &schedulerobjects.JobSchedulingInfo{}
+	if err := proto.Unmarshal(dbJob.SchedulingInfo, schedulingInfo); err != nil {
+		return nil, errors.Wrapf(err, "error unmarshalling scheduling info for job %s", dbJob.JobID)
+	}
+	return jobDb.NewJob(
+		dbJob.JobID,
+		dbJob.JobSet,
+		dbJob.Queue,
+		uint32(dbJob.Priority),
+		schedulingInfo,
+		dbJob.Queued,
+		dbJob.QueuedVersion,
+		dbJob.CancelRequested,
+		dbJob.CancelByJobsetRequested,
+		dbJob.Cancelled,
+		dbJob.Submitted,
+	), nil
+}
+
+// schedulerRunFromDatabaseRun creates a new scheduler job run from a database job run
+func (jobDb *JobDb) schedulerRunFromDatabaseRun(dbRun *database.Run) *JobRun {
+	nodeId := api.NodeIdFromExecutorAndNodeName(dbRun.Executor, dbRun.Node)
+	return jobDb.CreateRun(
+		dbRun.RunID,
+		dbRun.JobID,
+		dbRun.Created,
+		dbRun.Executor,
+		nodeId,
+		dbRun.Node,
+		dbRun.Running,
+		dbRun.Succeeded,
+		dbRun.Failed,
+		dbRun.Cancelled,
+		dbRun.Returned,
+		dbRun.RunAttempted,
+	)
+}
diff --git a/internal/scheduler/metrics/metrics.go b/internal/scheduler/metrics/metrics.go
new file mode 100644
index 00000000..f576f6f9
--- /dev/null
+++ b/internal/scheduler/metrics/metrics.go
@@ -0,0 +1,378 @@
+package metrics
+
+import (
+	"regexp"
+	"sync"
+
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/exp/maps"
+	"golang.org/x/exp/slices"
+
+	"github.com/armadaproject/armada/internal/common/armadacontext"
+	"github.com/armadaproject/armada/internal/scheduler/configuration"
+	"github.com/armadaproject/armada/internal/scheduler/jobdb"
+	"github.com/armadaproject/armada/pkg/armadaevents"
+)
+
+const (
+	namespace = "armada"
+	subsystem = "scheduler"
+
+	jobsResourceLabel = "jobs"
+
+	trackedErrorRegexMatches      = "1"
+	trackedErrorRegexDoesNotMatch = "0"
+
+	unknown          = "unknown"
+	podUnschedulable = "podUnschedulable"
+	leaseExpired     = "leaseExpired"
+	podError         = "podError"
+	podLeaseReturned = "podLeaseReturned"
+	podTerminated    = "podTerminated"
+)
+
+type Metrics struct {
+	config configuration.MetricsConfig
+
+	// Buffer used to avoid allocations when updating metrics.
+	// Protected by a mutex.
+	buffer []string
+	mu     sync.Mutex
+
+	// For disabling metrics at runtime, e.g., if not leader.
+	disabled bool
+
+	// Labels of tracked errors. Stored here to ensure consistent ordering.
+	trackedErrorLabels  []string
+	trackedErrorRegexes []*regexp.Regexp
+
+	// Job metrics.
+	queued    *prometheus.CounterVec
+	scheduled *prometheus.CounterVec
+	preempted *prometheus.CounterVec
+	failed    *prometheus.CounterVec
+	cancelled *prometheus.CounterVec
+	succeeded *prometheus.CounterVec
+}
+
+func New(config configuration.MetricsConfig) (*Metrics, error) {
+	trackedErrorLabels := maps.Keys(config.TrackedErrorRegexByLabel)
+	slices.Sort(trackedErrorLabels)
+	trackedErrorRegexes := make([]*regexp.Regexp, len(trackedErrorLabels))
+	for i, trackedErrorLabel := range trackedErrorLabels {
+		r, err := regexp.Compile(config.TrackedErrorRegexByLabel[trackedErrorLabel])
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+		trackedErrorRegexes[i] = r
+	}
+
+	inactiveJobLabels := []string{"queue", "resource"}
+	activeJobLabels := []string{"queue", "cluster", "node", "resource"}
+	failedJobLabels := append(append(
+		[]string{"queue", "cluster", "node", "errorType"},
+		trackedErrorLabels...,
+	), "resource")
+
+	return &Metrics{
+		config: config,
+
+		trackedErrorLabels:  trackedErrorLabels,
+		trackedErrorRegexes: trackedErrorRegexes,
+
+		buffer: make([]string, 0, len(failedJobLabels)),
+
+		queued: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "queued_total",
+				Help:      "Queued jobs.",
+			},
+			inactiveJobLabels,
+		),
+		scheduled: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "scheduled_total",
+				Help:      "Scheduled jobs.",
+			},
+			activeJobLabels,
+		),
+		preempted: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "preempted_total",
+				Help:      "Preempted jobs.",
+			},
+			activeJobLabels,
+		),
+		cancelled: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "cancelled_total",
+				Help:      "Cancelled jobs.",
+			},
+			activeJobLabels,
+		),
+		failed: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "failed_total",
+				Help:      "Failed jobs.",
+			},
+			failedJobLabels,
+		),
+		succeeded: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Subsystem: subsystem,
+				Name:      "succeeded_total",
+				Help:      "Successful jobs.",
+			},
+			activeJobLabels,
+		),
+	}, nil
+}
+
+func (m *Metrics) Disable() {
+	if m != nil {
+		m.disabled = true
+	}
+}
+
+func (m *Metrics) Enable() {
+	if m != nil {
+		m.disabled = false
+	}
+}
+
+func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
+	if m == nil || m.config.Disabled || m.disabled {
+		return
+	}
+	// TODO(albin): Only these metrics are expected to work for now.
+	m.queued.Describe(ch)
+	m.scheduled.Describe(ch)
+	m.preempted.Describe(ch)
+	m.failed.Describe(ch)
+	m.succeeded.Describe(ch)
+}
+
+// Collect and then reset all metrics.
+// Resetting ensures we do not build up a large number of counters over time.
+func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
+	if m == nil || m.config.Disabled || m.disabled {
+		return
+	}
+	// TODO(albin): Only these metrics are expected to work for now.
+	m.queued.Collect(ch)
+	m.queued.Reset()
+	m.scheduled.Collect(ch)
+	m.scheduled.Reset()
+	m.preempted.Collect(ch)
+	m.preempted.Reset()
+	m.failed.Collect(ch)
+	m.failed.Reset()
+	m.succeeded.Collect(ch)
+	m.succeeded.Reset()
+}
+
+func (m *Metrics) UpdateMany(
+	ctx *armadacontext.Context,
+	jsts []jobdb.JobStateTransitions,
+	jobRunErrorsByRunId map[uuid.UUID]*armadaevents.Error,
+) error {
+	if m == nil || m.config.Disabled || m.disabled {
+		return nil
+	}
+	for _, jst := range jsts {
+		if err := m.Update(ctx, jst, jobRunErrorsByRunId); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *Metrics) Update(
+	ctx *armadacontext.Context,
+	jst jobdb.JobStateTransitions,
+	jobRunErrorsByRunId map[uuid.UUID]*armadaevents.Error,
+) error {
+	if m == nil || m.config.Disabled || m.disabled {
+		return nil
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if jst.Queued {
+		if err := m.updateQueued(m.buffer[0:0], jst.Job); err != nil {
+			return err
+		}
+	}
+	if jst.Scheduled {
+		if err := m.updateLeased(m.buffer[0:0], jst.Job); err != nil {
+			return err
+		}
+	}
+	if jst.Preempted {
+		if err := m.updatePreempted(m.buffer[0:0], jst.Job); err != nil {
+			return err
+		}
+	}
+	if jst.Cancelled {
+		if err := m.updateCancelled(m.buffer[0:0], jst.Job); err != nil {
+			return err
+		}
+	}
+	if jst.Failed {
+		if err := m.updateFailed(ctx, m.buffer[0:0], jst.Job, jobRunErrorsByRunId); err != nil {
+			return err
+		}
+	}
+	if jst.Succeeded {
+		if err := m.updateSucceeded(m.buffer[0:0], jst.Job); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *Metrics) updateQueued(labels []string, job *jobdb.Job) error {
+	labels = append(labels, job.GetQueue())
+	if err := m.updateCounterVecFromJob(m.queued, labels, job); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *Metrics) updateLeased(labels []string, job *jobdb.Job) error {
+	executor, nodeName := executorAndNodeNameFromRun(job.LatestRun())
+	labels = append(labels, job.GetQueue())
+	labels = append(labels, executor)
+	labels = append(labels, nodeName)
+	if err := m.updateCounterVecFromJob(m.scheduled, labels, job); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *Metrics) updatePreempted(labels []string, job *jobdb.Job) error {
+	executor, nodeName := executorAndNodeNameFromRun(job.LatestRun())
+	labels = append(labels, job.GetQueue())
+	labels = append(labels, executor)
+	labels = append(labels, nodeName)
+	if err := m.updateCounterVecFromJob(m.preempted, labels, job); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *Metrics) updateCancelled(labels []string, job *jobdb.Job) error {
+	executor, nodeName := executorAndNodeNameFromRun(job.LatestRun())
+	labels = append(labels, job.GetQueue())
+	labels = append(labels, executor)
+	labels = append(labels, nodeName)
+	if err := m.updateCounterVecFromJob(m.cancelled, labels, job); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *Metrics) updateFailed(ctx *armadacontext.Context, labels []string, job *jobdb.Job, jobRunErrorsByRunId map[uuid.UUID]*armadaevents.Error) error {
+	run := job.LatestRun()
+	executor, nodeName := executorAndNodeNameFromRun(run)
+	name, message := errorTypeAndMessageFromError(ctx, jobRunErrorsByRunId[run.Id()])
+
+	labels = append(labels, job.GetQueue())
+	labels = append(labels, executor)
+	labels = append(labels, nodeName)
+	labels = append(labels, name)
+
+	for _, r := range m.trackedErrorRegexes {
+		if r.MatchString(message) {
+			labels = append(labels, trackedErrorRegexMatches)
+		} else {
+			labels = append(labels, trackedErrorRegexDoesNotMatch)
+		}
+	}
+
+	if err := m.updateCounterVecFromJob(m.failed, labels, job); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *Metrics) updateSucceeded(labels []string, job *jobdb.Job) error {
+	executor, nodeName := executorAndNodeNameFromRun(job.LatestRun())
+	labels = append(labels, job.GetQueue())
+	labels = append(labels, executor)
+	labels = append(labels, nodeName)
+	if err := m.updateCounterVecFromJob(m.succeeded, labels, job); err != nil {
+		return err
+	}
+	return nil
+}
+
+func executorAndNodeNameFromRun(run *jobdb.JobRun) (string, string) {
+	if run == nil {
+		// This case covers, e.g., jobs failing that have never been scheduled.
+		return "", ""
+	}
+	return run.Executor(), run.NodeName()
+}
+
+func errorTypeAndMessageFromError(ctx *armadacontext.Context, err *armadaevents.Error) (string, string) {
+	if err == nil {
+		return unknown, ""
+	}
+	// The following errors relate to job run failures.
+	// We do not process JobRunPreemptedError as there is separate metric for preemption.
+	switch reason := err.Reason.(type) {
+	case *armadaevents.Error_PodUnschedulable:
+		return podUnschedulable, reason.PodUnschedulable.Message
+	case *armadaevents.Error_LeaseExpired:
+		return leaseExpired, ""
+	case *armadaevents.Error_PodError:
+		return podError, reason.PodError.Message
+	case *armadaevents.Error_PodLeaseReturned:
+		return podLeaseReturned, reason.PodLeaseReturned.Message
+	case *armadaevents.Error_PodTerminated:
+		return podTerminated, reason.PodTerminated.Message
+	default:
+		ctx.Warnf("omitting name and message for unknown error type %T", err.Reason)
+		return unknown, ""
+	}
+}
+
+// updateCounterVecFromJob is a helper method to increment vector counters.
+func (m *Metrics) updateCounterVecFromJob(vec *prometheus.CounterVec, labels []string, job *jobdb.Job) error {
+	// Number of jobs.
+	i := len(labels)
+	labels = append(labels, jobsResourceLabel)
+	if c, err := vec.GetMetricWithLabelValues(labels...); err != nil {
+		return err
+	} else {
+		c.Add(1)
+	}
+
+	// Total resource requests of jobs.
+	requests := job.GetResourceRequirements().Requests
+	for _, resourceName := range m.config.TrackedResourceNames {
+		labels[i] = string(resourceName)
+		if c, err := vec.GetMetricWithLabelValues(labels...); err != nil {
+			return err
+		} else {
+			q := requests[resourceName]
+			v := float64(q.MilliValue()) / 1000
+			c.Add(v)
+		}
+	}
+
+	return nil
+}
diff --git a/internal/scheduler/result.go b/internal/scheduler/result.go
new file mode 100644
index 00000000..f577c8a5
--- /dev/null
+++ b/internal/scheduler/result.go
@@ -0,0 +1,51 @@
+package scheduler
+
+import (
+	schedulercontext "github.com/armadaproject/armada/internal/scheduler/context"
+	"github.com/armadaproject/armada/internal/scheduler/interfaces"
+)
+
+// SchedulerResult is returned by Rescheduler.Schedule().
+type SchedulerResult struct {
+	// Running jobs that should be preempted.
+	PreemptedJobs []interfaces.LegacySchedulerJob
+	// Queued jobs that should be scheduled.
+	ScheduledJobs []interfaces.LegacySchedulerJob
+	// Queued jobs that could not be scheduled.
+	// This is used to fail jobs that could not schedule above `minimumGangCardinality`.
+	FailedJobs []interfaces.LegacySchedulerJob
+	// For each preempted job, maps the job id to the id of the node on which the job was running.
+	// For each scheduled job, maps the job id to the id of the node on which the job should be scheduled.
+	NodeIdByJobId map[string]string
+	// Each result may bundle the result of several scheduling decisions.
+	// These are the corresponding scheduling contexts.
+	// TODO: This doesn't seem like the right approach.
+	SchedulingContexts []*schedulercontext.SchedulingContext
+}
+
+// PreemptedJobsFromSchedulerResult returns the slice of preempted jobs in the result cast to type T.
+func PreemptedJobsFromSchedulerResult[T interfaces.LegacySchedulerJob](sr *SchedulerResult) []T {
+	rv := make([]T, len(sr.PreemptedJobs))
+	for i, job := range sr.PreemptedJobs {
+		rv[i] = job.(T)
+	}
+	return rv
+}
+
+// ScheduledJobsFromSchedulerResult returns the slice of scheduled jobs in the result cast to type T.
+func ScheduledJobsFromSchedulerResult[T interfaces.LegacySchedulerJob](sr *SchedulerResult) []T {
+	rv := make([]T, len(sr.ScheduledJobs))
+	for i, job := range sr.ScheduledJobs {
+		rv[i] = job.(T)
+	}
+	return rv
+}
+
+// FailedJobsFromSchedulerResult returns the slice of scheduled jobs in the result cast to type T.
+func FailedJobsFromSchedulerResult[T interfaces.LegacySchedulerJob](sr *SchedulerResult) []T {
+	rv := make([]T, len(sr.FailedJobs))
+	for i, job := range sr.FailedJobs {
+		rv[i] = job.(T)
+	}
+	return rv
+}
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index ca28e262..b35be5be 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -8,19 +8,18 @@ import (
 	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	"github.com/renstrom/shortuuid"
-	"golang.org/x/exp/maps"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/clock"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/armadacontext"
 	"github.com/armadaproject/armada/internal/common/logging"
-	"github.com/armadaproject/armada/internal/common/stringinterner"
+	"github.com/armadaproject/armada/internal/common/util"
 	"github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
 	"github.com/armadaproject/armada/internal/scheduler/kubernetesobjects/affinity"
+	"github.com/armadaproject/armada/internal/scheduler/metrics"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
-	"github.com/armadaproject/armada/pkg/api"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 )
 
@@ -42,8 +41,6 @@ type Scheduler struct {
 	schedulingAlgo SchedulingAlgo
 	// Tells us if we are leader. Only the leader may schedule jobs.
 	leaderController LeaderController
-	// We intern strings to save memory.
-	stringInterner *stringinterner.StringInterner
 	// This is used to check if jobs are still schedulable.
 	// Useful when we are adding node anti-affinities.
 	submitChecker SubmitScheduleChecker
@@ -75,6 +72,8 @@ type Scheduler struct {
 	onCycleCompleted func()
 	// metrics set for the scheduler.
 	metrics *SchedulerMetrics
+	// New scheduler metrics due to replace the above.
+	schedulerMetrics *metrics.Metrics
 }
 
 func NewScheduler(
@@ -84,14 +83,14 @@ func NewScheduler(
 	schedulingAlgo SchedulingAlgo,
 	leaderController LeaderController,
 	publisher Publisher,
-	stringInterner *stringinterner.StringInterner,
 	submitChecker SubmitScheduleChecker,
 	cyclePeriod time.Duration,
 	schedulePeriod time.Duration,
 	executorTimeout time.Duration,
 	maxAttemptedRuns uint,
 	nodeIdLabel string,
-	schedulerMetrics *SchedulerMetrics,
+	metrics *SchedulerMetrics,
+	schedulerMetrics *metrics.Metrics,
 ) (*Scheduler, error) {
 	return &Scheduler{
 		jobRepository:              jobRepository,
@@ -99,7 +98,6 @@ func NewScheduler(
 		schedulingAlgo:             schedulingAlgo,
 		leaderController:           leaderController,
 		publisher:                  publisher,
-		stringInterner:             stringInterner,
 		submitChecker:              submitChecker,
 		jobDb:                      jobDb,
 		clock:                      clock.RealClock{},
@@ -111,7 +109,8 @@ func NewScheduler(
 		nodeIdLabel:                nodeIdLabel,
 		jobsSerial:                 -1,
 		runsSerial:                 -1,
-		metrics:                    schedulerMetrics,
+		metrics:                    metrics,
+		schedulerMetrics:           schedulerMetrics,
 	}, nil
 }
 
@@ -196,18 +195,28 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error {
 // cycle is a single iteration of the main scheduling loop.
 // If updateAll is true, we generate events from all jobs in the jobDb.
 // Otherwise, we only generate events from jobs updated since the last cycle.
-func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToken LeaderToken, shouldSchedule bool) (overallSchedulerResult SchedulerResult, err error) {
-	overallSchedulerResult = SchedulerResult{EmptyResult: true}
+func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToken LeaderToken, shouldSchedule bool) (SchedulerResult, error) {
+	// TODO: Consider returning a slice of these instead.
+	overallSchedulerResult := SchedulerResult{}
 
 	// Update job state.
-	updatedJobs, err := s.syncState(ctx)
+	updatedJobs, jsts, jobRepoRunErrorsByRunId, err := s.syncState(ctx)
 	if err != nil {
-		return
+		return overallSchedulerResult, err
 	}
 
 	// Only the leader may make decisions; exit if not leader.
+	// Only export metrics if leader.
 	if !s.leaderController.ValidateToken(leaderToken) {
-		return
+		s.schedulerMetrics.Disable()
+		return overallSchedulerResult, err
+	} else {
+		s.schedulerMetrics.Enable()
+	}
+
+	// Update metrics.
+	if err := s.schedulerMetrics.UpdateMany(ctx, jsts, jobRepoRunErrorsByRunId); err != nil {
+		return overallSchedulerResult, err
 	}
 
 	// If we've been asked to generate messages for all jobs, do so.
@@ -219,22 +228,22 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 	}
 
 	// Generate any events that came out of synchronising the db state.
-	events, err := s.generateUpdateMessages(ctx, updatedJobs, txn)
+	events, err := s.generateUpdateMessages(ctx, txn, updatedJobs, jobRepoRunErrorsByRunId)
 	if err != nil {
-		return
+		return overallSchedulerResult, err
 	}
 
 	// Expire any jobs running on clusters that haven't heartbeated within the configured deadline.
 	expirationEvents, err := s.expireJobsIfNecessary(ctx, txn)
 	if err != nil {
-		return
+		return overallSchedulerResult, err
 	}
 	events = append(events, expirationEvents...)
 
 	// Request cancel for any jobs that exceed queueTtl
 	queueTtlCancelEvents, err := s.cancelQueuedJobsIfExpired(txn)
 	if err != nil {
-		return
+		return overallSchedulerResult, err
 	}
 	events = append(events, queueTtlCancelEvents...)
 
@@ -243,13 +252,13 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 		var result *SchedulerResult
 		result, err = s.schedulingAlgo.Schedule(ctx, txn)
 		if err != nil {
-			return
+			return overallSchedulerResult, err
 		}
 
 		var resultEvents []*armadaevents.EventSequence
 		resultEvents, err = s.eventsFromSchedulerResult(result)
 		if err != nil {
-			return
+			return overallSchedulerResult, err
 		}
 		events = append(events, resultEvents...)
 		s.previousSchedulingRoundEnd = s.clock.Now()
@@ -263,95 +272,118 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 	}
 	start := s.clock.Now()
 	if err = s.publisher.PublishMessages(ctx, events, isLeader); err != nil {
-		return
+		return overallSchedulerResult, err
 	}
 	ctx.Infof("published %d events to pulsar in %s", len(events), s.clock.Since(start))
 	txn.Commit()
-	return
-}
 
-// syncState updates jobs in jobDb to match state in postgres and returns all updated jobs.
-func (s *Scheduler) syncState(ctx *armadacontext.Context) ([]*jobdb.Job, error) {
-	start := s.clock.Now()
-	updatedJobs, updatedRuns, err := s.jobRepository.FetchJobUpdates(ctx, s.jobsSerial, s.runsSerial)
-	if err != nil {
-		return nil, err
+	// Update metrics based on overallSchedulerResult.
+	if err := s.updateMetricsFromSchedulerResult(ctx, overallSchedulerResult); err != nil {
+		return overallSchedulerResult, err
 	}
-	ctx.Infof("received %d updated jobs and %d updated job runs in %s", len(updatedJobs), len(updatedRuns), s.clock.Since(start))
 
-	txn := s.jobDb.WriteTxn()
-	defer txn.Abort()
+	return overallSchedulerResult, nil
+}
 
-	// Process jobs.
-	jobsToDelete := make([]string, 0, len(updatedJobs))
-	jobsToUpdateById := make(map[string]*jobdb.Job, len(updatedJobs))
-	for _, dbJob := range updatedJobs {
-		if dbJob.InTerminalState() {
-			// Scheduler has sent a terminal message; we can safely remove the job.
-			jobsToDelete = append(jobsToDelete, dbJob.JobID)
-			continue
+func (s *Scheduler) updateMetricsFromSchedulerResult(ctx *armadacontext.Context, overallSchedulerResult SchedulerResult) error {
+	for _, job := range overallSchedulerResult.ScheduledJobs {
+		if err := s.schedulerMetrics.Update(
+			ctx,
+			jobdb.JobStateTransitions{
+				Job:       job.(*jobdb.Job),
+				Scheduled: true,
+			},
+			nil,
+		); err != nil {
+			return err
 		}
-
-		// Try and retrieve the job from the jobDb. If it doesn't exist then create it.
-		job := txn.GetById(dbJob.JobID)
-		if job == nil {
-			job, err = s.schedulerJobFromDatabaseJob(&dbJob)
-			if err != nil {
-				return nil, err
-			}
-		} else {
-			// make the scheduler job look like the db job.
-			job, err = updateSchedulerJob(job, &dbJob)
-			if err != nil {
-				return nil, err
-			}
+	}
+	for _, job := range overallSchedulerResult.PreemptedJobs {
+		if err := s.schedulerMetrics.Update(
+			ctx,
+			jobdb.JobStateTransitions{
+				Job:       job.(*jobdb.Job),
+				Preempted: true,
+			},
+			nil,
+		); err != nil {
+			return err
 		}
-		jobsToUpdateById[job.Id()] = job
 	}
+	for _, job := range overallSchedulerResult.FailedJobs {
+		if err := s.schedulerMetrics.Update(
+			ctx,
+			jobdb.JobStateTransitions{
+				Job:    job.(*jobdb.Job),
+				Failed: true,
+			},
+			nil,
+		); err != nil {
+			return err
+		}
+	}
+	return nil
+}
 
-	// Process runs.
-	for _, dbRun := range updatedRuns {
-		jobId := dbRun.JobID
+// syncState updates jobs in jobDb to match state in postgres and returns all updated jobs.
+func (s *Scheduler) syncState(ctx *armadacontext.Context) ([]*jobdb.Job, []jobdb.JobStateTransitions, map[uuid.UUID]*armadaevents.Error, error) {
+	txn := s.jobDb.WriteTxn()
+	defer txn.Abort()
 
-		// Retrieve the job, look first in the list of updates, then in the jobDb.
-		job, present := jobsToUpdateById[jobId]
-		if !present {
-			job = txn.GetById(jobId)
+	// Load new and updated jobs from the jobRepo.
+	updatedJobs, updatedRuns, err := s.jobRepository.FetchJobUpdates(ctx, s.jobsSerial, s.runsSerial)
+	if err != nil {
+		return nil, nil, nil, err
+	}
 
-			// If the job is nil or terminal at this point then it cannot be active.
-			// In this case we can ignore the run.
-			if job == nil || job.InTerminalState() {
-				ctx.Debugf("job %s is not active; ignoring update for run %s", jobId, dbRun.RunID)
-				continue
-			}
-		}
+	// Load any error associated with updated runs.
+	jobRunIds := util.Map(updatedRuns, func(jobRepoRun database.Run) uuid.UUID { return jobRepoRun.RunID })
+	jobRepoRunErrorsByRunId, err := s.jobRepository.FetchJobRunErrors(ctx, jobRunIds)
+	if err != nil {
+		return nil, nil, nil, err
+	}
 
-		run := job.RunById(dbRun.RunID)
-		if run == nil {
-			run = s.createSchedulerRun(&dbRun)
-		} else {
-			// make the scheduler job look like the db job
-			run = updateSchedulerRun(run, &dbRun)
+	// Reconcile any differences between the updated jobs and runs.
+	jsts, err := s.jobDb.ReconcileDifferences(txn, updatedJobs, updatedRuns)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	// Upsert updated jobs (including associated runs).
+	jobDbJobs := make([]*jobdb.Job, 0, len(jsts))
+	for _, jst := range jsts {
+		if jst.Job != nil {
+			// We receive nil jobs from jobDb.ReconcileDifferences if a run is updated after the associated job is deleted.
+			// These nil job must be sorted out.
+			jobDbJobs = append(jobDbJobs, jst.Job)
 		}
-		job = job.WithUpdatedRun(run)
-		jobsToUpdateById[jobId] = job
+	}
+	if err := txn.Upsert(jobDbJobs); err != nil {
+		return nil, nil, nil, err
 	}
 
-	jobsToUpdate := maps.Values(jobsToUpdateById)
-	if err := txn.Upsert(jobsToUpdate); err != nil {
-		return nil, err
+	// Delete jobs in a terminal state.
+	idsOfJobsToDelete := make([]string, 0)
+	for _, jobDbJob := range jobDbJobs {
+		if jobDbJob.InTerminalState() {
+			idsOfJobsToDelete = append(idsOfJobsToDelete, jobDbJob.Id())
+		}
 	}
-	if err := txn.BatchDelete(jobsToDelete); err != nil {
-		return nil, err
+	if err := txn.BatchDelete(idsOfJobsToDelete); err != nil {
+		return nil, nil, nil, err
 	}
+
 	txn.Commit()
+
+	// Update serial to include these updates.
 	if len(updatedJobs) > 0 {
 		s.jobsSerial = updatedJobs[len(updatedJobs)-1].Serial
 	}
 	if len(updatedRuns) > 0 {
 		s.runsSerial = updatedRuns[len(updatedRuns)-1].Serial
 	}
-	return jobsToUpdate, nil
+
+	return jobDbJobs, jsts, jobRepoRunErrorsByRunId, nil
 }
 
 func (s *Scheduler) createSchedulingInfoWithNodeAntiAffinityForAttemptedRuns(job *jobdb.Job) (*schedulerobjects.JobSchedulingInfo, error) {
@@ -535,20 +567,8 @@ func AppendEventSequencesFromUnschedulableJobs(eventSequences []*armadaevents.Ev
 
 // generateUpdateMessages generates EventSequences representing the state changes on updated jobs
 // If there are no state changes then an empty slice will be returned
-func (s *Scheduler) generateUpdateMessages(ctx *armadacontext.Context, updatedJobs []*jobdb.Job, txn *jobdb.Txn) ([]*armadaevents.EventSequence, error) {
-	failedRunIds := make([]uuid.UUID, 0, len(updatedJobs))
-	for _, job := range updatedJobs {
-		run := job.LatestRun()
-		if run != nil && run.Failed() {
-			failedRunIds = append(failedRunIds, run.Id())
-		}
-	}
-	jobRunErrors, err := s.jobRepository.FetchJobRunErrors(ctx, failedRunIds)
-	if err != nil {
-		return nil, err
-	}
-
-	// Generate any events that came out of synchronising the db state
+func (s *Scheduler) generateUpdateMessages(ctx *armadacontext.Context, txn *jobdb.Txn, updatedJobs []*jobdb.Job, jobRunErrors map[uuid.UUID]*armadaevents.Error) ([]*armadaevents.EventSequence, error) {
+	// Generate any events that came out of synchronising the db state.
 	var events []*armadaevents.EventSequence
 	for _, job := range updatedJobs {
 		jobEvents, err := s.generateUpdateMessagesFromJob(job, jobRunErrors, txn)
@@ -567,7 +587,7 @@ func (s *Scheduler) generateUpdateMessages(ctx *armadacontext.Context, updatedJo
 func (s *Scheduler) generateUpdateMessagesFromJob(job *jobdb.Job, jobRunErrors map[uuid.UUID]*armadaevents.Error, txn *jobdb.Txn) (*armadaevents.EventSequence, error) {
 	var events []*armadaevents.EventSequence_Event
 
-	// Is the job already in a terminal state?  If so then don't send any more messages
+	// Is the job already in a terminal state? If so then don't send any more messages
 	if job.InTerminalState() {
 		return nil, nil
 	}
@@ -634,7 +654,7 @@ func (s *Scheduler) generateUpdateMessagesFromJob(job *jobdb.Job, jobRunErrors m
 					if schedulable {
 						job = jobWithAntiAffinity
 					} else {
-						// If job is not schedulable with anti-affinity added. Do not requeue it and let it fail
+						// If job is not schedulable with anti-affinity added. Do not requeue it and let it fail.
 						requeueJob = false
 					}
 				}
@@ -884,11 +904,11 @@ func (s *Scheduler) initialise(ctx *armadacontext.Context) error {
 		case <-ctx.Done():
 			return nil
 		default:
-			if _, err := s.syncState(ctx); err != nil {
+			if _, _, _, err := s.syncState(ctx); err != nil {
 				logging.WithStacktrace(ctx, err).Error("failed to initialise; trying again in 1 second")
 				time.Sleep(1 * time.Second)
 			} else {
-				// Initialisation succeeded.
+				ctx.Info("initialisation succeeded")
 				return nil
 			}
 		}
@@ -941,115 +961,3 @@ func (s *Scheduler) ensureDbUpToDate(ctx *armadacontext.Context, pollInterval ti
 		}
 	}
 }
-
-// schedulerJobFromDatabaseJob creates a new scheduler job from a database job.
-func (s *Scheduler) schedulerJobFromDatabaseJob(dbJob *database.Job) (*jobdb.Job, error) {
-	schedulingInfo := &schedulerobjects.JobSchedulingInfo{}
-	err := proto.Unmarshal(dbJob.SchedulingInfo, schedulingInfo)
-	if err != nil {
-		return nil, errors.Wrapf(err, "error unmarshalling scheduling info for job %s", dbJob.JobID)
-	}
-	s.internJobSchedulingInfoStrings(schedulingInfo)
-	return s.jobDb.NewJob(
-		dbJob.JobID,
-		s.stringInterner.Intern(dbJob.JobSet),
-		s.stringInterner.Intern(dbJob.Queue),
-		uint32(dbJob.Priority),
-		schedulingInfo,
-		dbJob.Queued,
-		dbJob.QueuedVersion,
-		dbJob.CancelRequested,
-		dbJob.CancelByJobsetRequested,
-		dbJob.Cancelled,
-		dbJob.Submitted,
-	), nil
-}
-
-// createSchedulerRun creates a new scheduler job run from a database job run
-func (s *Scheduler) createSchedulerRun(dbRun *database.Run) *jobdb.JobRun {
-	nodeId := api.NodeIdFromExecutorAndNodeName(dbRun.Executor, dbRun.Node)
-	return jobdb.CreateRun(
-		dbRun.RunID,
-		dbRun.JobID,
-		dbRun.Created,
-		s.stringInterner.Intern(dbRun.Executor),
-		s.stringInterner.Intern(nodeId),
-		s.stringInterner.Intern(dbRun.Node),
-		dbRun.Running,
-		dbRun.Succeeded,
-		dbRun.Failed,
-		dbRun.Cancelled,
-		dbRun.Returned,
-		dbRun.RunAttempted,
-	)
-}
-
-func (s *Scheduler) internJobSchedulingInfoStrings(info *schedulerobjects.JobSchedulingInfo) {
-	for _, requirement := range info.ObjectRequirements {
-		if podRequirement := requirement.GetPodRequirements(); podRequirement != nil {
-			for k, v := range podRequirement.Annotations {
-				podRequirement.Annotations[s.stringInterner.Intern(k)] = s.stringInterner.Intern(v)
-			}
-
-			for k, v := range podRequirement.NodeSelector {
-				podRequirement.NodeSelector[s.stringInterner.Intern(k)] = s.stringInterner.Intern(v)
-			}
-			podRequirement.PreemptionPolicy = s.stringInterner.Intern(podRequirement.PreemptionPolicy)
-		}
-	}
-}
-
-// updateSchedulerRun updates the scheduler job run (in-place) to match the database job run
-func updateSchedulerRun(run *jobdb.JobRun, dbRun *database.Run) *jobdb.JobRun {
-	if dbRun.Succeeded && !run.Succeeded() {
-		run = run.WithSucceeded(true)
-	}
-	if dbRun.Failed && !run.Failed() {
-		run = run.WithFailed(true)
-	}
-	if dbRun.Cancelled && !run.Cancelled() {
-		run = run.WithCancelled(true)
-	}
-	if dbRun.Returned && !run.Returned() {
-		run = run.WithReturned(true)
-	}
-	if dbRun.RunAttempted && !run.RunAttempted() {
-		run = run.WithAttempted(true)
-	}
-	return run
-}
-
-// updateSchedulerJob updates the scheduler job in-place to match the database job.
-func updateSchedulerJob(job *jobdb.Job, dbJob *database.Job) (*jobdb.Job, error) {
-	if dbJob.CancelRequested && !job.CancelRequested() {
-		job = job.WithCancelRequested(true)
-	}
-	if dbJob.CancelByJobsetRequested && !job.CancelByJobsetRequested() {
-		job = job.WithCancelByJobsetRequested(true)
-	}
-	if dbJob.Cancelled && !job.Cancelled() {
-		job = job.WithCancelled(true)
-	}
-	if dbJob.Succeeded && !job.Succeeded() {
-		job = job.WithSucceeded(true)
-	}
-	if dbJob.Failed && !job.Failed() {
-		job = job.WithFailed(true)
-	}
-	if uint32(dbJob.Priority) != job.RequestedPriority() {
-		job = job.WithRequestedPriority(uint32(dbJob.Priority))
-	}
-	if uint32(dbJob.SchedulingInfoVersion) > job.JobSchedulingInfo().Version {
-		schedulingInfo := &schedulerobjects.JobSchedulingInfo{}
-		err := proto.Unmarshal(dbJob.SchedulingInfo, schedulingInfo)
-		if err != nil {
-			return nil, errors.Wrapf(err, "error unmarshalling scheduling info for job %s", dbJob.JobID)
-		}
-		job = job.WithJobSchedulingInfo(schedulingInfo)
-	}
-	if dbJob.QueuedVersion > job.QueuedVersion() {
-		job = job.WithQueuedVersion(dbJob.QueuedVersion)
-		job = job.WithQueued(dbJob.Queued)
-	}
-	return job, nil
-}
diff --git a/internal/scheduler/scheduler_metrics.go b/internal/scheduler/scheduler_metrics.go
index 3ba197eb..5964f73a 100644
--- a/internal/scheduler/scheduler_metrics.go
+++ b/internal/scheduler/scheduler_metrics.go
@@ -158,10 +158,6 @@ func (metrics *SchedulerMetrics) ReportReconcileCycleTime(cycleTime time.Duratio
 }
 
 func (metrics *SchedulerMetrics) ReportSchedulerResult(ctx *armadacontext.Context, result SchedulerResult) {
-	if result.EmptyResult {
-		return // TODO: Add logging or maybe place to add failure metric?
-	}
-
 	// Report the total scheduled jobs (possibly we can get these out of contexts?)
 	metrics.reportScheduledJobs(ctx, result.ScheduledJobs)
 	metrics.reportPreemptedJobs(ctx, result.PreemptedJobs)
@@ -173,11 +169,17 @@ func (metrics *SchedulerMetrics) ReportSchedulerResult(ctx *armadacontext.Contex
 }
 
 func (metrics *SchedulerMetrics) reportScheduledJobs(ctx *armadacontext.Context, scheduledJobs []interfaces.LegacySchedulerJob) {
+	if len(scheduledJobs) == 0 {
+		return
+	}
 	jobAggregates := aggregateJobs(scheduledJobs)
 	observeJobAggregates(ctx, metrics.scheduledJobsPerQueue, jobAggregates)
 }
 
 func (metrics *SchedulerMetrics) reportPreemptedJobs(ctx *armadacontext.Context, preemptedJobs []interfaces.LegacySchedulerJob) {
+	if len(preemptedJobs) == 0 {
+		return
+	}
 	jobAggregates := aggregateJobs(preemptedJobs)
 	observeJobAggregates(ctx, metrics.preemptedJobsPerQueue, jobAggregates)
 }
diff --git a/internal/scheduler/schedulerapp.go b/internal/scheduler/schedulerapp.go
index 1093c5b7..57aa48ad 100644
--- a/internal/scheduler/schedulerapp.go
+++ b/internal/scheduler/schedulerapp.go
@@ -28,10 +28,10 @@ import (
 	"github.com/armadaproject/armada/internal/common/profiling"
 	"github.com/armadaproject/armada/internal/common/pulsarutils"
 	"github.com/armadaproject/armada/internal/common/serve"
-	"github.com/armadaproject/armada/internal/common/stringinterner"
 	schedulerconfig "github.com/armadaproject/armada/internal/scheduler/configuration"
 	"github.com/armadaproject/armada/internal/scheduler/database"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
+	"github.com/armadaproject/armada/internal/scheduler/metrics"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
 	"github.com/armadaproject/armada/pkg/executorapi"
 )
@@ -40,17 +40,17 @@ import (
 func Run(config schedulerconfig.Configuration) error {
 	g, ctx := armadacontext.ErrGroup(app.CreateContextWithShutdown())
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Profiling
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	pprofServer := profiling.SetupPprofHttpServer(config.PprofPort)
 	g.Go(func() error {
 		return serve.ListenAndServe(ctx, pprofServer)
 	})
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Health Checks
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	mux := http.NewServeMux()
 
 	startupCompleteCheck := health.NewStartupCompleteChecker()
@@ -64,9 +64,9 @@ func Run(config schedulerconfig.Configuration) error {
 	// we add all services to a slice and start them together at the end of this function.
 	var services []func() error
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Database setup (postgres and redis)
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	ctx.Infof("Setting up database connections")
 	db, err := dbcommon.OpenPgxPool(config.Postgres)
 	if err != nil {
@@ -88,9 +88,9 @@ func Run(config schedulerconfig.Configuration) error {
 	queueRepository := database.NewLegacyQueueRepository(redisClient)
 	legacyExecutorRepository := database.NewRedisExecutorRepository(redisClient, "pulsar")
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Pulsar
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	ctx.Infof("Setting up Pulsar connectivity")
 	pulsarClient, err := pulsarutils.NewPulsarClient(&config.Pulsar)
 	if err != nil {
@@ -108,18 +108,18 @@ func Run(config schedulerconfig.Configuration) error {
 		return errors.WithMessage(err, "error creating pulsar publisher")
 	}
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Leader Election
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	leaderController, err := createLeaderController(ctx, config.Leader)
 	if err != nil {
 		return errors.WithMessage(err, "error creating leader controller")
 	}
 	services = append(services, func() error { return leaderController.Run(ctx) })
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Executor Api
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	ctx.Infof("Setting up executor api")
 	apiProducer, err := pulsarClient.CreateProducer(pulsar.ProducerOptions{
 		Name:             fmt.Sprintf("armada-executor-api-%s", uuid.NewString()),
@@ -163,14 +163,10 @@ func Run(config schedulerconfig.Configuration) error {
 	})
 	services = append(services, grpcCommon.CreateShutdownHandler(ctx, 5*time.Second, grpcServer))
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Scheduling
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	ctx.Infof("setting up scheduling loop")
-	stringInterner, err := stringinterner.New(config.InternedStringsCacheSize)
-	if err != nil {
-		return errors.WithMessage(err, "error creating string interner")
-	}
 
 	submitChecker := NewSubmitChecker(
 		30*time.Minute,
@@ -206,7 +202,15 @@ func Run(config schedulerconfig.Configuration) error {
 	jobDb := jobdb.NewJobDb(
 		config.Scheduling.Preemption.PriorityClasses,
 		config.Scheduling.Preemption.DefaultPriorityClass,
+		config.InternedStringsCacheSize,
 	)
+	schedulerMetrics, err := metrics.New(config.SchedulerMetrics)
+	if err != nil {
+		return err
+	}
+	if err := prometheus.Register(schedulerMetrics); err != nil {
+		return errors.WithStack(err)
+	}
 	scheduler, err := NewScheduler(
 		jobDb,
 		jobRepository,
@@ -214,7 +218,6 @@ func Run(config schedulerconfig.Configuration) error {
 		schedulingAlgo,
 		leaderController,
 		pulsarPublisher,
-		stringInterner,
 		submitChecker,
 		config.CyclePeriod,
 		config.SchedulePeriod,
@@ -222,15 +225,16 @@ func Run(config schedulerconfig.Configuration) error {
 		config.Scheduling.MaxRetries+1,
 		config.Scheduling.Preemption.NodeIdLabel,
 		NewSchedulerMetrics(config.Metrics.Metrics),
+		schedulerMetrics,
 	)
 	if err != nil {
 		return errors.WithMessage(err, "error creating scheduler")
 	}
 	services = append(services, func() error { return scheduler.Run(ctx) })
 
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	// Metrics
-	//////////////////////////////////////////////////////////////////////////
+	// ////////////////////////////////////////////////////////////////////////
 	poolAssigner, err := NewPoolAssigner(config.Scheduling.ExecutorTimeout, config.Scheduling, executorRepository)
 	if err != nil {
 		return errors.WithMessage(err, "error creating pool assigner")
@@ -240,8 +244,11 @@ func Run(config schedulerconfig.Configuration) error {
 		queueRepository,
 		executorRepository,
 		poolAssigner,
-		config.Metrics.RefreshInterval)
-	prometheus.MustRegister(metricsCollector)
+		config.Metrics.RefreshInterval,
+	)
+	if err := prometheus.Register(metricsCollector); err != nil {
+		return errors.WithStack(err)
+	}
 	services = append(services, func() error { return metricsCollector.Run(ctx) })
 	shutdownMetricServer := common.ServeMetrics(config.Metrics.Port)
 	defer shutdownMetricServer()
diff --git a/internal/scheduler/simulator/simulator.go b/internal/scheduler/simulator/simulator.go
index e8da3010..99393d39 100644
--- a/internal/scheduler/simulator/simulator.go
+++ b/internal/scheduler/simulator/simulator.go
@@ -100,6 +100,7 @@ func NewSimulator(clusterSpec *ClusterSpec, workloadSpec *WorkloadSpec, scheduli
 	jobDb := jobdb.NewJobDb(
 		schedulingConfig.Preemption.PriorityClasses,
 		schedulingConfig.Preemption.DefaultPriorityClass,
+		1024,
 	)
 	randomSeed := workloadSpec.RandomSeed
 	if randomSeed == 0 {


Test patch:
diff --git a/internal/common/math/math_test.go b/internal/common/math/math_test.go
new file mode 100644
index 00000000..6a1402ef
--- /dev/null
+++ b/internal/common/math/math_test.go
@@ -0,0 +1,16 @@
+package math
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMax(t *testing.T) {
+	assert.Equal(t, 1, Max(0, 1))
+	assert.Equal(t, 1, Max(1, 0))
+	assert.Equal(t, 1.0, Max(0.0, 1.0))
+	assert.Equal(t, 1.0, Max(1.0, 0.0))
+	assert.Equal(t, 0, Max(0, -1))
+	assert.Equal(t, 0, Max(-1, 0))
+}
diff --git a/internal/common/stringinterner/stringintern_test.go b/internal/common/stringinterner/stringintern_test.go
index 7a450aba..72a81a83 100644
--- a/internal/common/stringinterner/stringintern_test.go
+++ b/internal/common/stringinterner/stringintern_test.go
@@ -7,7 +7,6 @@ import (
 	"unsafe"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -19,9 +18,7 @@ var (
 
 // Test that repeated calls to intern for strings with identical values but different backing arrays result in deduplication
 func TestStringInterner_TestIntern(t *testing.T) {
-	interner, err := New(defaultCapacity)
-	require.NoError(t, err)
-
+	interner := New(defaultCapacity)
 	for i := 0; i < len(cloned1); i++ {
 		s1 := cloned1[i]
 		s2 := cloned2[i]
@@ -42,8 +39,7 @@ func TestStringInterner_TestIntern(t *testing.T) {
 
 // Test that we don't store more strings than capacity
 func TestStringInterner_TestCapacity(t *testing.T) {
-	interner, err := New(defaultCapacity)
-	require.NoError(t, err)
+	interner := New(defaultCapacity)
 
 	// intern a string and assert we return the same string
 	retrieved1 := interner.Intern(cloned1[0])
diff --git a/internal/common/util/list_test.go b/internal/common/util/list_test.go
index f50e3bc7..39066948 100644
--- a/internal/common/util/list_test.go
+++ b/internal/common/util/list_test.go
@@ -7,50 +7,6 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestFilter(t *testing.T) {
-	includeOver5 := func(val int) bool { return val > 5 }
-	input := []int{1, 3, 5, 7, 9}
-	expectedOutput := []int{7, 9}
-
-	output := Filter(input, includeOver5)
-	assert.Equal(t, expectedOutput, output)
-}
-
-func TestFilter_ExcludeAllFilter(t *testing.T) {
-	excludeAll := func(val int) bool { return false }
-	input := []int{1, 3, 5, 7, 9}
-	expectedOutput := []int{}
-	output := Filter(input, excludeAll)
-	assert.Equal(t, expectedOutput, output)
-}
-
-func TestFilter_IncludeAllFilter(t *testing.T) {
-	includeAll := func(val int) bool { return true }
-	input := []int{1, 3, 5, 7, 9}
-	expectedOutput := input
-
-	output := Filter(input, includeAll)
-	assert.Equal(t, expectedOutput, output)
-}
-
-func TestFilter_Empty(t *testing.T) {
-	includeAll := func(val int) bool { return true }
-	input := []int{}
-	expectedOutput := input
-
-	output := Filter(input, includeAll)
-	assert.Equal(t, expectedOutput, output)
-}
-
-func TestFilter_Nil(t *testing.T) {
-	includeAll := func(val int) bool { return true }
-	var input []int = nil
-	expectedOutput := input
-
-	output := Filter(input, includeAll)
-	assert.Equal(t, expectedOutput, output)
-}
-
 func TestConcat_Empty(t *testing.T) {
 	output := Concat([]int{})
 	assert.Equal(t, []int{}, output)
diff --git a/internal/scheduler/jobdb/job_run_test.go b/internal/scheduler/jobdb/job_run_test.go
index ca71c5d8..282914ce 100644
--- a/internal/scheduler/jobdb/job_run_test.go
+++ b/internal/scheduler/jobdb/job_run_test.go
@@ -5,9 +5,38 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
+
+	"github.com/armadaproject/armada/internal/common/types"
+	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
+)
+
+const (
+	PriorityClass0               = "priority-0"
+	PriorityClass1               = "priority-1"
+	PriorityClass2               = "priority-2"
+	PriorityClass2NonPreemptible = "priority-2-non-preemptible"
+	PriorityClass3               = "priority-3"
+)
+
+var (
+	TestPriorityClasses = map[string]types.PriorityClass{
+		PriorityClass0:               {Priority: 0, Preemptible: true},
+		PriorityClass1:               {Priority: 1, Preemptible: true},
+		PriorityClass2:               {Priority: 2, Preemptible: true},
+		PriorityClass2NonPreemptible: {Priority: 2, Preemptible: false},
+		PriorityClass3:               {Priority: 3, Preemptible: false},
+	}
+	TestDefaultPriorityClass = PriorityClass3
+	SchedulingKeyGenerator   = schedulerobjects.NewSchedulingKeyGeneratorWithKey(make([]byte, 32))
+	jobDb                    = NewJobDbWithSchedulingKeyGenerator(
+		TestPriorityClasses,
+		TestDefaultPriorityClass,
+		SchedulingKeyGenerator,
+		1024,
+	)
 )
 
-var baseJobRun = CreateRun(
+var baseJobRun = jobDb.CreateRun(
 	uuid.New(),
 	uuid.NewString(),
 	5,
@@ -67,7 +96,7 @@ func TestJobRun_TestRunAttempted(t *testing.T) {
 }
 
 func TestDeepCopy(t *testing.T) {
-	run := CreateRun(
+	run := jobDb.CreateRun(
 		uuid.New(),
 		"job id",
 		1,
@@ -81,7 +110,7 @@ func TestDeepCopy(t *testing.T) {
 		true,
 		true,
 	)
-	expected := CreateRun(
+	expected := jobDb.CreateRun(
 		run.id,
 		"job id",
 		1,
diff --git a/internal/scheduler/jobdb/job_test.go b/internal/scheduler/jobdb/job_test.go
index 373c2167..d62b2e1b 100644
--- a/internal/scheduler/jobdb/job_test.go
+++ b/internal/scheduler/jobdb/job_test.go
@@ -7,7 +7,6 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 
-	"github.com/armadaproject/armada/internal/common/types"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
 )
 
@@ -25,15 +24,6 @@ var jobSchedulingInfo = &schedulerobjects.JobSchedulingInfo{
 	},
 }
 
-// Used for creating jobs.
-var jobDb = NewJobDb(
-	map[string]types.PriorityClass{
-		"foo": {},
-		"bar": {},
-	},
-	"foo",
-)
-
 var baseJob = jobDb.NewJob(
 	"test-job",
 	"test-jobSet",
diff --git a/internal/scheduler/jobdb/jobdb_test.go b/internal/scheduler/jobdb/jobdb_test.go
index ac4e7e61..03fcedc9 100644
--- a/internal/scheduler/jobdb/jobdb_test.go
+++ b/internal/scheduler/jobdb/jobdb_test.go
@@ -27,6 +27,7 @@ func NewTestJobDb() *JobDb {
 			"bar": {},
 		},
 		"foo",
+		1024,
 	)
 }
 
diff --git a/internal/scheduler/preempting_queue_scheduler_test.go b/internal/scheduler/preempting_queue_scheduler_test.go
index 2dab76d3..987eb63f 100644
--- a/internal/scheduler/preempting_queue_scheduler_test.go
+++ b/internal/scheduler/preempting_queue_scheduler_test.go
@@ -50,7 +50,7 @@ func TestEvictOversubscribed(t *testing.T) {
 	err = nodeDb.CreateAndInsertWithJobDbJobsWithTxn(nodeDbTxn, jobs, node)
 	require.NoError(t, err)
 
-	jobDb := jobdb.NewJobDb(config.Preemption.PriorityClasses, config.Preemption.DefaultPriorityClass)
+	jobDb := jobdb.NewJobDb(config.Preemption.PriorityClasses, config.Preemption.DefaultPriorityClass, 1024)
 	jobDbTxn := jobDb.WriteTxn()
 	err = jobDbTxn.Upsert(jobs)
 	require.NoError(t, err)
@@ -1501,7 +1501,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 				priorities = append(priorities, pc.Priority)
 			}
 
-			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.Preemption.PriorityClasses, tc.SchedulingConfig.Preemption.DefaultPriorityClass)
+			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.Preemption.PriorityClasses, tc.SchedulingConfig.Preemption.DefaultPriorityClass, 1024)
 			jobDbTxn := jobDb.WriteTxn()
 
 			// Accounting across scheduling rounds.
@@ -1897,7 +1897,7 @@ func BenchmarkPreemptingQueueScheduler(b *testing.B) {
 			}
 			txn.Commit()
 
-			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.Preemption.PriorityClasses, tc.SchedulingConfig.Preemption.DefaultPriorityClass)
+			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.Preemption.PriorityClasses, tc.SchedulingConfig.Preemption.DefaultPriorityClass, 1024)
 			jobDbTxn := jobDb.WriteTxn()
 			var queuedJobs []*jobdb.Job
 			for _, jobs := range jobsByQueue {
diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go
index 8a652563..357b47c6 100644
--- a/internal/scheduler/scheduler_test.go
+++ b/internal/scheduler/scheduler_test.go
@@ -16,9 +16,9 @@ import (
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/armadacontext"
 	protoutil "github.com/armadaproject/armada/internal/common/proto"
-	"github.com/armadaproject/armada/internal/common/stringinterner"
 	"github.com/armadaproject/armada/internal/common/util"
 	"github.com/armadaproject/armada/internal/scheduler/database"
+	"github.com/armadaproject/armada/internal/scheduler/interfaces"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
 	"github.com/armadaproject/armada/internal/scheduler/kubernetesobjects/affinity"
 	"github.com/armadaproject/armada/internal/scheduler/schedulerobjects"
@@ -181,7 +181,7 @@ var (
 		false,
 		false,
 		1).WithUpdatedRun(
-		jobdb.CreateRun(
+		testfixtures.JobDb.CreateRun(
 			uuid.New(),
 			requeuedJobId,
 			time.Now().Unix(),
@@ -251,6 +251,33 @@ func TestScheduler_TestCycle(t *testing.T) {
 			expectedLeased:        []string{queuedJob.Id()},
 			expectedQueuedVersion: queuedJob.QueuedVersion() + 1,
 		},
+		"Lease two jobs from an update": {
+			jobUpdates: []database.Job{
+				{
+					JobID:                 "01h3w2wtdchtc80hgyp782shrv",
+					JobSet:                "testJobSet",
+					Queue:                 "testQueue",
+					Queued:                true,
+					QueuedVersion:         1,
+					SchedulingInfo:        schedulingInfoBytes,
+					SchedulingInfoVersion: int32(schedulingInfo.Version),
+					Serial:                1,
+				},
+				{
+					JobID:                 "01h434g4hxww2pknb2q1nfmfph",
+					JobSet:                "testJobSet",
+					Queue:                 "testQueue",
+					Queued:                true,
+					QueuedVersion:         1,
+					SchedulingInfo:        schedulingInfoBytes,
+					SchedulingInfoVersion: int32(schedulingInfo.Version),
+					Serial:                1,
+				},
+			},
+			expectedJobRunLeased:  []string{"01h3w2wtdchtc80hgyp782shrv", "01h434g4hxww2pknb2q1nfmfph"},
+			expectedLeased:        []string{"01h3w2wtdchtc80hgyp782shrv", "01h434g4hxww2pknb2q1nfmfph"},
+			expectedQueuedVersion: queuedJob.QueuedVersion() + 1,
+		},
 		"Nothing leased": {
 			initialJobs:           []*jobdb.Job{queuedJob},
 			expectedQueued:        []string{queuedJob.Id()},
@@ -604,8 +631,6 @@ func TestScheduler_TestCycle(t *testing.T) {
 				shouldError:    tc.scheduleError,
 			}
 			publisher := &testPublisher{shouldError: tc.publishError}
-			stringInterner, err := stringinterner.New(100)
-			require.NoError(t, err)
 			submitChecker := &testSubmitChecker{checkSuccess: !tc.submitCheckerFailure}
 
 			heartbeatTime := testClock.Now()
@@ -622,7 +647,6 @@ func TestScheduler_TestCycle(t *testing.T) {
 				schedulingAlgo,
 				NewStandaloneLeaderController(),
 				publisher,
-				stringInterner,
 				submitChecker,
 				1*time.Second,
 				5*time.Second,
@@ -630,6 +654,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 				maxNumberOfAttempts,
 				nodeIdLabel,
 				schedulerMetrics,
+				nil,
 			)
 			require.NoError(t, err)
 
@@ -779,9 +804,6 @@ func TestRun(t *testing.T) {
 	clusterRepo := &testExecutorRepository{}
 	leaderController := NewStandaloneLeaderController()
 	submitChecker := &testSubmitChecker{checkSuccess: true}
-	stringInterner, err := stringinterner.New(100)
-	require.NoError(t, err)
-
 	sched, err := NewScheduler(
 		testfixtures.NewJobDb(),
 		&jobRepo,
@@ -789,14 +811,15 @@ func TestRun(t *testing.T) {
 		schedulingAlgo,
 		leaderController,
 		publisher,
-		stringInterner,
 		submitChecker,
 		1*time.Second,
 		15*time.Second,
 		1*time.Hour,
 		maxNumberOfAttempts,
 		nodeIdLabel,
-		schedulerMetrics)
+		schedulerMetrics,
+		nil,
+	)
 	require.NoError(t, err)
 
 	sched.clock = testClock
@@ -908,7 +931,7 @@ func TestScheduler_TestSyncState(t *testing.T) {
 			},
 			expectedUpdatedJobs: []*jobdb.Job{
 				queuedJob.WithUpdatedRun(
-					jobdb.CreateRun(
+					testfixtures.JobDb.CreateRun(
 						uuid.UUID{},
 						queuedJob.Id(),
 						123,
@@ -948,8 +971,10 @@ func TestScheduler_TestSyncState(t *testing.T) {
 					Succeeded: true,
 				},
 			},
-			expectedUpdatedJobs: []*jobdb.Job{leasedJob.WithUpdatedRun(leasedJob.LatestRun().WithSucceeded(true))},
-			expectedJobDbIds:    []string{},
+			expectedUpdatedJobs: []*jobdb.Job{leasedJob.
+				WithUpdatedRun(leasedJob.LatestRun().WithSucceeded(true)).
+				WithSucceeded(true)},
+			expectedJobDbIds: []string{},
 		},
 		"job requeued": {
 			initialJobs: []*jobdb.Job{leasedJob},
@@ -990,9 +1015,6 @@ func TestScheduler_TestSyncState(t *testing.T) {
 			publisher := &testPublisher{}
 			clusterRepo := &testExecutorRepository{}
 			leaderController := NewStandaloneLeaderController()
-			stringInterner, err := stringinterner.New(100)
-			require.NoError(t, err)
-
 			sched, err := NewScheduler(
 				testfixtures.NewJobDb(),
 				jobRepo,
@@ -1000,7 +1022,6 @@ func TestScheduler_TestSyncState(t *testing.T) {
 				schedulingAlgo,
 				leaderController,
 				publisher,
-				stringInterner,
 				nil,
 				1*time.Second,
 				5*time.Second,
@@ -1008,6 +1029,7 @@ func TestScheduler_TestSyncState(t *testing.T) {
 				maxNumberOfAttempts,
 				nodeIdLabel,
 				schedulerMetrics,
+				nil,
 			)
 			require.NoError(t, err)
 
@@ -1021,7 +1043,7 @@ func TestScheduler_TestSyncState(t *testing.T) {
 			require.NoError(t, err)
 			txn.Commit()
 
-			updatedJobs, err := sched.syncState(ctx)
+			updatedJobs, _, _, err := sched.syncState(ctx)
 			require.NoError(t, err)
 
 			assert.Equal(t, tc.expectedUpdatedJobs, updatedJobs)
@@ -1183,6 +1205,32 @@ func (t *testSchedulingAlgo) Schedule(ctx *armadacontext.Context, txn *jobdb.Txn
 	return NewSchedulerResultForTest(preemptedJobs, scheduledJobs, failedJobs, nil), nil
 }
 
+func NewSchedulerResultForTest[S ~[]T, T interfaces.LegacySchedulerJob](
+	preemptedJobs S,
+	scheduledJobs S,
+	failedJobs S,
+	nodeIdByJobId map[string]string,
+) *SchedulerResult {
+	castPreemptedJobs := make([]interfaces.LegacySchedulerJob, len(preemptedJobs))
+	for i, job := range preemptedJobs {
+		castPreemptedJobs[i] = job
+	}
+	castScheduledJobs := make([]interfaces.LegacySchedulerJob, len(scheduledJobs))
+	for i, job := range scheduledJobs {
+		castScheduledJobs[i] = job
+	}
+	castFailedJobs := make([]interfaces.LegacySchedulerJob, len(failedJobs))
+	for i, job := range failedJobs {
+		castFailedJobs[i] = job
+	}
+	return &SchedulerResult{
+		PreemptedJobs: castPreemptedJobs,
+		ScheduledJobs: castScheduledJobs,
+		NodeIdByJobId: nodeIdByJobId,
+		FailedJobs:    castFailedJobs,
+	}
+}
+
 type testPublisher struct {
 	events      []*armadaevents.EventSequence
 	shouldError bool
diff --git a/internal/scheduler/testfixtures/testfixtures.go b/internal/scheduler/testfixtures/testfixtures.go
index e84ea325..407b4218 100644
--- a/internal/scheduler/testfixtures/testfixtures.go
+++ b/internal/scheduler/testfixtures/testfixtures.go
@@ -79,6 +79,7 @@ func NewJobDb() *jobdb.JobDb {
 		TestPriorityClasses,
 		TestDefaultPriorityClass,
 		SchedulingKeyGenerator,
+		1024,
 	)
 }