wip

Author: d80tb7

.run/Scheduler.run.xml

@@ -160,6 +160,13 @@ func ApiJobFromLogSubmitJob(ownerId string, groups []string, queueName string, j
 		podSpecs = k8sPodSpecs
 	}
 
+	var priceInfo *api.ExperimentalPriceInfo
+	if e.PriceInfo != nil {
+		priceInfo = &api.ExperimentalPriceInfo{
+			BidPrice: e.PriceInfo.BidPrice,
+		}
+	}
+
 	return &api.Job{
 		Id:       e.JobId,
 		ClientId: e.DeduplicationId,
@@ -172,7 +179,7 @@ func ApiJobFromLogSubmitJob(ownerId string, groups []string, queueName string, j
 
 		K8SIngress: k8sIngresses,
 		K8SService: k8sServices,
-		Price:      e.Price,
+		PriceInfo:  priceInfo,
 		Priority:   float64(e.Priority),
 
 		PodSpec:                        podSpec,

internal/common/eventutil/eventutil.go

@@ -0,0 +1 @@
+ALTER TABLE jobs ADD COLUMN bid_price double precision NOT NULL DEFAULT 0;

internal/scheduler/database/migrations/018_add_price.sql

@@ -260,7 +260,7 @@ func (jobDb *JobDb) schedulerJobFromDatabaseJob(dbJob *database.Job) (*Job, erro
 		dbJob.JobSet,
 		dbJob.Queue,
 		uint32(dbJob.Priority),
-		dbJob.Price,
+		dbJob.BidPrice,
 		schedulingInfo,
 		dbJob.Queued,
 		dbJob.QueuedVersion,

internal/scheduler/database/migrations/018_add_validated.sql

@@ -320,7 +320,7 @@ func (sch *PreemptingQueueScheduler) evict(ctx *armadacontext.Context, evictor *
 	if err := sch.nodeDb.Reset(); err != nil {
 		return nil, nil, err
 	}
-	if err := addEvictedJobsToNodeDb(ctx, sch.schedulingContext, sch.nodeDb, inMemoryJobRepo, sch.marketDriven); err != nil {
+	if err := sch.addEvictedJobsToNodeDb(ctx, inMemoryJobRepo); err != nil {
 		return nil, nil, err
 	}
 	return result, inMemoryJobRepo, nil
@@ -492,7 +492,8 @@ func (q MinimalQueue) GetWeight() float64 {
 
 // addEvictedJobsToNodeDb adds evicted jobs to the NodeDb.
 // Needed to enable the nodeDb accounting for these when preempting.
-func addEvictedJobsToNodeDb(_ *armadacontext.Context, sctx *schedulercontext.SchedulingContext, nodeDb *nodedb.NodeDb, inMemoryJobRepo *InMemoryJobRepository, marketDriven bool) error {
+func (sch *PreemptingQueueScheduler) addEvictedJobsToNodeDb(_ *armadacontext.Context, inMemoryJobRepo *InMemoryJobRepository) error {
+	sctx := sch.schedulingContext
 	gangItByQueue := make(map[string]*QueuedGangIterator)
 	for _, qctx := range sch.schedulingContext.QueueSchedulingContexts {
 		gangItByQueue[qctx.Queue] = NewQueuedGangIterator(
@@ -502,11 +503,11 @@ func addEvictedJobsToNodeDb(_ *armadacontext.Context, sctx *schedulercontext.Sch
 			false,
 		)
 	}
-	qr := NewMinimalQueueRepositoryFromSchedulingContext(sctx)
+	qr := NewMinimalQueueRepositoryFromSchedulingContext(sch.schedulingContext)
 	var candidateGangIterator CandidateGangIterator
 	var err error
-	if marketDriven {
-		candidateGangIterator, err = NewCostBasedCandidateGangIterator(sctx.Pool, sctx, sctx.FairnessCostProvider, gangItByQueue, false)
+	if sch.marketDriven {
+		candidateGangIterator, err = NewCostBasedCandidateGangIterator(sctx.Pool, sctx, sctx.FairnessCostProvider, gangItByQueue, false, sch.preferLargeJobOrdering)
 		if err != nil {
 			return err
 		}
@@ -516,8 +517,7 @@ func addEvictedJobsToNodeDb(_ *armadacontext.Context, sctx *schedulercontext.Sch
 			return err
 		}
 	}
-
-	txn := nodeDb.Txn(true)
+	txn := sch.nodeDb.Txn(true)
 	defer txn.Abort()
 	i := 0
 	for {

internal/scheduler/database/models.go

@@ -68,7 +68,7 @@ func NewQueueScheduler(
 			return nil, err
 		}
 	} else {
-		candidateGangIterator, err = NewCostBasedCandidateGangIterator(sctx.Pool, sctx, sctx.FairnessCostProvider, gangIteratorsByQueue, considerPriorityClassPriority)
+		candidateGangIterator, err = NewCostBasedCandidateGangIterator(sctx.Pool, sctx, sctx.FairnessCostProvider, gangIteratorsByQueue, considerPriorityClassPriority, prioritiseLargerJobs)
 		if err != nil {
 			return nil, err
 		}

internal/scheduler/database/query.sql.go

@@ -788,6 +788,15 @@ func SwaggerJsonTemplate() string {
 		"        }\n" +
 		"      }\n" +
 		"    },\n" +
+		"    \"apiExperimentalPriceInfo\": {\n" +
+		"      \"type\": \"object\",\n" +
+		"      \"properties\": {\n" +
+		"        \"bidPrice\": {\n" +
+		"          \"type\": \"number\",\n" +
+		"          \"format\": \"double\"\n" +
+		"        }\n" +
+		"      }\n" +
+		"    },\n" +
 		"    \"apiIngressConfig\": {\n" +
 		"      \"type\": \"object\",\n" +
 		"      \"properties\": {\n" +
@@ -892,9 +901,8 @@ func SwaggerJsonTemplate() string {
 		"            \"$ref\": \"#/definitions/v1PodSpec\"\n" +
 		"          }\n" +
 		"        },\n" +
-		"        \"price\": {\n" +
-		"          \"type\": \"number\",\n" +
-		"          \"format\": \"double\"\n" +
+		"        \"priceInfo\": {\n" +
+		"          \"$ref\": \"#/definitions/apiExperimentalPriceInfo\"\n" +
 		"        },\n" +
 		"        \"priority\": {\n" +
 		"          \"type\": \"number\",\n" +