armadaproject · jparraga-stackav · May 22, 2025 · May 28, 2025 · May 29, 2025 · jparraga-stackav
diff --git a/internal/common/ingest/testfixtures/event.go b/internal/common/ingest/testfixtures/event.go
@@ -48,6 +48,7 @@ const (
 )
 
 var (
+	RunIndex                   = int64(0)
 	PartitionMarkerGroupIdUuid = uuid.MustParse(PartitionMarkerGroupId)
 	PriorityClassName          = "test-priority"
 	Groups                     = []string{"group1", "group2"}

diff --git a/internal/scheduler/api.go b/internal/scheduler/api.go
@@ -170,11 +170,16 @@ func (srv *ExecutorApi) LeaseJobRuns(stream executorapi.ExecutorApi_LeaseJobRuns
 			}
 		}
 
+		var runIndex uint32 = 0
+		if lease.RunIndex != nil {
+			runIndex = uint32(*lease.RunIndex)
+		}
+
 		err := stream.Send(&executorapi.LeaseStreamMessage{
 			Event: &executorapi.LeaseStreamMessage_Lease{
 				Lease: &executorapi.JobRunLease{
 					JobRunId:    lease.RunID,
-					JobRunIndex: lease.RunIndex,
+					JobRunIndex: runIndex,
 					Queue:       lease.Queue,
 					Jobset:      lease.JobSet,
 					User:        lease.UserID,

diff --git a/internal/scheduler/database/job_repository.go b/internal/scheduler/database/job_repository.go
@@ -24,7 +24,7 @@ type hasSerial interface {
 
 type JobRunLease struct {
 	RunID                  string
-	RunIndex               uint32
+	RunIndex               *int64
 	Queue                  string
 	Pool                   string
 	JobSet                 string

diff --git a/internal/scheduler/database/migrations/022_add_job_run_index.sql b/internal/scheduler/database/migrations/022_add_job_run_index.sql
@@ -1 +1 @@
-ALTER TABLE runs ADD COLUMN IF NOT EXISTS run_index bigint NOT NULL DEFAULT 0;
+ALTER TABLE runs ADD COLUMN IF NOT EXISTS run_index bigint;
diff --git a/internal/scheduler/database/models.go b/internal/scheduler/database/models.go
diff --git a/internal/scheduler/jobdb/reconciliation.go b/internal/scheduler/jobdb/reconciliation.go
@@ -306,9 +306,13 @@ func (jobDb *JobDb) schedulerJobFromDatabaseJob(dbJob *database.Job) (*Job, erro
 // 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)
+	var runIndex uint32 = 0
+	if dbRun.RunIndex != nil {
+		runIndex = uint32(*dbRun.RunIndex)
+	}
 	return jobDb.CreateRun(
 		dbRun.RunID,
-		uint32(dbRun.RunIndex),
+		runIndex,
 		dbRun.JobID,
 		dbRun.Created,
 		dbRun.Executor,

diff --git a/internal/scheduleringester/instructions.go b/internal/scheduleringester/instructions.go
@@ -186,6 +186,7 @@ func (c *JobSetEventsInstructionConverter) handleJobRunLeased(jobRunLeased *arma
 	if jobRunLeased.HasScheduledAtPriority {
 		scheduledAtPriority = &jobRunLeased.ScheduledAtPriority
 	}
+	runIndex := int64(jobRunLeased.RunIndex)
 	PodRequirementsOverlay, err := proto.Marshal(jobRunLeased.GetPodRequirementsOverlay())
 	if err != nil {
 		return nil, errors.WithStack(err)
@@ -195,7 +196,7 @@ func (c *JobSetEventsInstructionConverter) handleJobRunLeased(jobRunLeased *arma
 			Queue: meta.queue,
 			DbRun: &schedulerdb.Run{
 				RunID:                  runId,
-				RunIndex:               int64(jobRunLeased.RunIndex),
+				RunIndex:               &runIndex,
 				JobID:                  jobRunLeased.JobId,
 				Created:                eventTime.UnixNano(),
 				JobSet:                 meta.jobset,

diff --git a/internal/scheduleringester/instructions_test.go b/internal/scheduleringester/instructions_test.go
@@ -54,6 +54,7 @@ func TestConvertEventSequence(t *testing.T) {
 			expected: []DbOperation{
 				InsertRuns{f.RunId: &JobRunDetails{Queue: f.Queue, DbRun: &schedulerdb.Run{
 					RunID:                  f.RunId,
+					RunIndex:               &f.RunIndex,
 					JobID:                  f.JobId,
 					JobSet:                 f.JobsetName,
 					Queue:                  f.Queue,
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		ALTER TABLE runs ADD COLUMN IF NOT EXISTS run_index bigint NOT NULL DEFAULT 0;
		ALTER TABLE runs ADD COLUMN IF NOT EXISTS run_index bigint;
Copy link Author jparraga-stackav May 29, 2025 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. During testing we found that the old scheduler ingester was trying to write rows with a null value which blocked ingestion from occurring. In order to make this smoother we've made this null-able and then handle null-able values in the app layer.