batch queue: add queue initialization to server (#27953)

Author: mismithhisler

nomad/fsm.go

@@ -18,6 +18,7 @@ import (
 	"github.com/hashicorp/go-msgpack/v2/codec"
 	"github.com/hashicorp/nomad/helper/pointer"
 	"github.com/hashicorp/nomad/helper/uuid"
+	"github.com/hashicorp/nomad/nomad/queues"
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/hashicorp/nomad/scheduler"
@@ -132,6 +133,7 @@ type SnapshotRestorers map[SnapshotType]SnapshotRestorer
 // this outside the Server to avoid exposing this outside the package.
 type nomadFSM struct {
 	evalBroker         *EvalBroker
+	batchQueue         queues.Queue
 	blockedEvals       *BlockedEvals
 	periodicDispatcher *PeriodicDispatch
 	encrypter          *Encrypter
@@ -170,6 +172,9 @@ type FSMConfig struct {
 	// EvalBroker is the evaluation broker evaluations should be added to
 	EvalBroker *EvalBroker
 
+	// BatchQueue is the configured queue for batch job registrations
+	BatchQueue queues.Queue
+
 	// Periodic is the periodic job dispatcher that periodic jobs should be
 	// added/removed from
 	Periodic *PeriodicDispatch
@@ -215,6 +220,7 @@ func NewFSM(config *FSMConfig) (*nomadFSM, error) {
 
 	fsm := &nomadFSM{
 		evalBroker:          config.EvalBroker,
+		batchQueue:          config.BatchQueue,
 		periodicDispatcher:  config.Periodic,
 		blockedEvals:        config.Blocked,
 		encrypter:           config.Encrypter,
@@ -972,7 +978,11 @@ func (n *nomadFSM) handleUpsertedEval(eval *structs.Evaluation) {
 	}
 
 	if eval.ShouldEnqueue() {
-		n.evalBroker.Enqueue(eval)
+		if eval.Type == structs.JobTypeBatch && eval.TriggeredBy == structs.EvalTriggerJobRegister {
+			n.batchQueue.Enqueue(eval)
+		} else {
+			n.evalBroker.Enqueue(eval)
+		}
 	} else if eval.ShouldBlock() {
 		n.blockedEvals.Block(eval)
 	} else if eval.Status == structs.EvalStatusComplete &&

nomad/leader.go

@@ -409,6 +409,8 @@ func (s *Server) establishLeadership(stopCh chan struct{}) error {
 	// the operator.
 	restoreEvals := s.handleEvalBrokerStateChange(schedulerConfig)
 
+	s.batchJobQueue.SetEnabled(true, s.State())
+
 	// Enable the deployment watcher, since we are now the leader
 	s.deploymentWatcher.SetEnabled(true, s.State())
 
@@ -1447,6 +1449,8 @@ func (s *Server) revokeLeadership() error {
 	// Disable the deployment watcher as it is only useful as a leader.
 	s.deploymentWatcher.SetEnabled(false, nil)
 
+	s.batchJobQueue.SetEnabled(false, nil)
+
 	// Disable the node drainer
 	s.nodeDrainer.SetEnabled(false, nil)
 

nomad/queues/batch_job_queue.go nomad/queues/dynamic_priority_queue.gonomad/queues/batch_job_queue.go renamed to nomad/queues/dynamic_priority_queue.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/go-hclog"
@@ -24,7 +25,7 @@ type DynamicPriorityQueue struct {
 	// tenants is used to keep track of cluster usage for this queue.
 	// When workloads are placed or the  configured interval is passed,
 	// cluster usage is updated for the workloads of each tenant.
-	tenants map[TenantID]Tenant
+	tenants map[TenantID]*Tenant
 
 	// queue is the main datastructure that contains all pending workloads
 	//
@@ -49,25 +50,27 @@ type DynamicPriorityQueue struct {
 	// totalUsage is the sum of all tenant usages
 	totalUsage int
 
+	tenantType structs.BatchQueueTenant
+
+	metadataKey string
+
 	// conf contains user configurations for tuning the behavior of the queue
-	conf *DynamicPriorityConfig
+	conf *structs.DynamicQueueConfig
 
 	// evalBroker is the injected broker for passing an evaluation
 	// on to be scheduled by Nomad
-	evalBroker Queue
+	evalBroker Broker
+
+	// enabled tracks whether the server running the batch job queue is the leader
+	// so should process evaluations
+	enabled atomic.Bool
 
 	// state is the in-memory state store used for both reconciling tenant
 	// workload usages, and polling submitted evaluations for placement
 	state  *state.StateStore
 	logger hclog.Logger
 }
 
-type DynamicPriorityConfig struct {
-	TenantType   string
-	MetadataKey  string
-	CalcInterval time.Duration
-}
-
 type Tenant struct {
 	tid       TenantID
 	workloads map[string]*Workload
@@ -87,33 +90,46 @@ func (w *Workload) calculatePriority(_ int64) {
 	// unimplemented
 }
 
-func NewDynamicPriorityQueue(state *state.StateStore, broker Queue, conf *DynamicPriorityConfig, logger hclog.Logger) *DynamicPriorityQueue {
+func NewDynamicPriorityQueue(broker Broker, qconf *structs.BatchQueue, conf *structs.DynamicQueueConfig, logger hclog.Logger) *DynamicPriorityQueue {
 	return &DynamicPriorityQueue{
-		tenants:    map[TenantID]Tenant{},
-		queue:      WorkloadQueue{},
-		state:      state,
-		enqueueCh:  make(chan *Workload, 8096),
-		evalBroker: broker,
-		qMux:       sync.Mutex{},
-		qNotify:    make(chan struct{}, 1),
-		conf:       conf,
-		logger:     logger.Named("Dynamic Priority Queue"),
+		tenants:     make(map[TenantID]*Tenant),
+		queue:       WorkloadQueue{},
+		enqueueCh:   make(chan *Workload, 8192),
+		evalBroker:  broker,
+		qMux:        sync.Mutex{},
+		qNotify:     make(chan struct{}, 1),
+		tenantType:  qconf.TenantType,
+		metadataKey: qconf.MetadataKey,
+		conf:        conf,
+		logger:      logger.Named("Dynamic Priority Queue"),
+		totalUsage:  0,
 	}
 }
 
-func (d *DynamicPriorityQueue) Start(ctx context.Context) {
-	// rebuild internal state from statestore, unimplemented
-
+func (d *DynamicPriorityQueue) Start(ctx context.Context) error {
 	go d.runProducer(ctx)
 	go d.runConsumer(ctx)
+
+	return nil
+}
+
+func (d *DynamicPriorityQueue) SetEnabled(val bool, state *state.StateStore) {
+	// rebuild internal state from statestore, unimplemented
+	d.state = state
+	d.enabled.Store(val)
 }
 
 // Enqueue is the method used to put evaluations on the queue.
 // It generates a workload with an empty priority, appends it
 // to an internal channel to be processed and added to the actual
 // heap container.
 func (d *DynamicPriorityQueue) Enqueue(e *structs.Evaluation) {
+	if !d.enabled.Load() {
+		return
+	}
+
 	w := d.generateWorkload(e)
+
 	// in the event of an empty workload, just pass eval to eval broker
 	if w == nil {
 		d.evalBroker.Enqueue(e)
@@ -143,6 +159,10 @@ func (d *DynamicPriorityQueue) runProducer(ctx context.Context) {
 			default:
 			}
 		case <-time.After(d.conf.CalcInterval):
+			if !d.enabled.Load() {
+				continue
+			}
+
 			d.qMux.Lock()
 			d.calculatePriorities(time.Now().UnixNano())
 			heap.Init(&d.queue)
@@ -199,11 +219,11 @@ func (d *DynamicPriorityQueue) generateWorkload(e *structs.Evaluation) *Workload
 	}
 
 	tid := ""
-	switch d.conf.TenantType {
+	switch d.tenantType {
 	case "namespace":
 		tid = job.Namespace
 	case "metadata":
-		tenantID, ok := job.Meta[d.conf.MetadataKey]
+		tenantID, ok := job.Meta[d.metadataKey]
 		if !ok {
 			return nil
 		}

nomad/queues/batch_job_queue_test.go …ad/queues/dynamic_priority_queue_test.gonomad/queues/batch_job_queue_test.go renamed to nomad/queues/dynamic_priority_queue_test.go nomad/queues/batch_job_queue_test.go renamed to nomad/queues/dynamic_priority_queue_test.go

@@ -21,7 +21,8 @@ func TestWaitForPlacement(t *testing.T) {
 
 	t.Run("returns if eval complete", func(t *testing.T) {
 		ss := state.TestStateStore(t)
-		testQueue := NewDynamicPriorityQueue(ss, nil, &DynamicPriorityConfig{}, hclog.New(hclog.DefaultOptions))
+		testQueue := NewDynamicPriorityQueue(nil, &structs.BatchQueue{}, &structs.DynamicQueueConfig{}, hclog.New(hclog.DefaultOptions))
+		testQueue.SetEnabled(true, ss)
 
 		testEval := mock.Eval()
 		ss.UpsertEvals(structs.MsgTypeTestSetup, 0, []*structs.Evaluation{testEval})
@@ -43,7 +44,8 @@ func TestWaitForPlacement(t *testing.T) {
 
 	t.Run("continues watching blocked evals", func(t *testing.T) {
 		ss := state.TestStateStore(t)
-		testQueue := NewDynamicPriorityQueue(ss, nil, &DynamicPriorityConfig{}, hclog.New(hclog.DefaultOptions))
+		testQueue := NewDynamicPriorityQueue(nil, &structs.BatchQueue{}, &structs.DynamicQueueConfig{}, hclog.New(hclog.DefaultOptions))
+		testQueue.SetEnabled(true, ss)
 
 		testEval := mock.Eval()
 		blocked := mock.Eval()
@@ -88,7 +90,8 @@ func TestWaitForPlacement(t *testing.T) {
 
 	t.Run("continues watching next evals after eval failure", func(t *testing.T) {
 		ss := state.TestStateStore(t)
-		testQueue := NewDynamicPriorityQueue(ss, nil, &DynamicPriorityConfig{}, hclog.New(hclog.DefaultOptions))
+		testQueue := NewDynamicPriorityQueue(nil, &structs.BatchQueue{}, &structs.DynamicQueueConfig{}, hclog.New(hclog.DefaultOptions))
+		testQueue.SetEnabled(true, ss)
 
 		testEval := mock.Eval()
 		next := mock.Eval()

nomad/queues/interface.go

@@ -3,8 +3,20 @@
 
 package queues
 
-import "github.com/hashicorp/nomad/nomad/structs"
+import (
+	"context"
+
+	"github.com/hashicorp/nomad/nomad/state"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
 
 type Queue interface {
 	Enqueue(*structs.Evaluation)
+	Start(context.Context) error
+	SetEnabled(bool, *state.StateStore)
+}
+
+// Broker is the interface for an evaluation broker
+type Broker interface {
+	Enqueue(*structs.Evaluation)
 }

nomad/queues/passthrough_queue.go

@@ -0,0 +1,28 @@
+// Copyright IBM Corp. 2015, 2026
+// SPDX-License-Identifier: BUSL-1.1
+
+package queues
+
+import (
+	"context"
+
+	"github.com/hashicorp/nomad/nomad/state"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+type PassthroughQueue struct {
+	broker Broker
+}
+
+func NewPassthroughQueue(b Broker) *PassthroughQueue {
+	return &PassthroughQueue{
+		broker: b,
+	}
+}
+
+// Start is a noop for the passthrough implementation
+func (p *PassthroughQueue) Start(context.Context) error { return nil }
+
+func (p *PassthroughQueue) Enqueue(e *structs.Evaluation) { p.broker.Enqueue(e) }
+
+func (p *PassthroughQueue) SetEnabled(bool, *state.StateStore) {}

nomad/queues/queue.go

@@ -0,0 +1,22 @@
+// Copyright IBM Corp. 2015, 2026
+// SPDX-License-Identifier: BUSL-1.1
+
+package queues
+
+import (
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+func NewQueue(sconf *structs.BatchQueue, broker Broker, logger hclog.Logger) (Queue, error) {
+	switch sconf.Type {
+	case structs.BatchQueueTypeDynamic:
+		qconf := &structs.DynamicQueueConfig{}
+		if err := structs.DecodeBatchQueueConf(sconf.Config, qconf); err != nil {
+			return nil, err
+		}
+		return NewDynamicPriorityQueue(broker, sconf, qconf, logger), nil
+	default:
+		return NewPassthroughQueue(broker), nil
+	}
+}

nomad/server.go

@@ -47,6 +47,7 @@ import (
 	"github.com/hashicorp/nomad/nomad/drainer"
 	"github.com/hashicorp/nomad/nomad/lock"
 	"github.com/hashicorp/nomad/nomad/peers"
+	"github.com/hashicorp/nomad/nomad/queues"
 	"github.com/hashicorp/nomad/nomad/reporting"
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -214,6 +215,10 @@ type Server struct {
 	// that are waiting to be brokered to a sub-scheduler
 	evalBroker *EvalBroker
 
+	// batchJobQueue is the interface for enqueuing job
+	// register evaluations on a queue implementation
+	batchJobQueue queues.Queue
+
 	// brokerLock is used to synchronise the alteration of the blockedEvals and
 	// evalBroker enabled state. These two subsystems change state when
 	// leadership changes or when the user modifies the setting via the
@@ -482,13 +487,31 @@ func NewServer(config *Config, consulCatalog consul.CatalogAPI, consulConfigFunc
 		Encrypter:      s.encrypter,
 	})
 
+	// Creates the batch job queue
+	s.batchJobQueue, err = queues.NewQueue(
+		&s.config.DefaultSchedulerConfig.BatchQueue,
+		evalBroker,
+		logger,
+	)
+	if err != nil {
+		s.Shutdown()
+		s.logger.Error("failed to create batch job queue", "error", err)
+		return nil, fmt.Errorf("Failed to create batch jo queue: %v", err)
+	}
+
 	// Initialize the Raft server
 	if err := s.setupRaft(); err != nil {
 		s.Shutdown()
 		s.logger.Error("failed to start Raft", "error", err)
 		return nil, fmt.Errorf("Failed to start Raft: %v", err)
 	}
 
+	if err := s.batchJobQueue.Start(s.shutdownCtx); err != nil {
+		s.Shutdown()
+		s.logger.Error("failed to start batch job queue", "error", err)
+		return nil, fmt.Errorf("Failed to start batcj job queue: %v", err)
+	}
+
 	// Initialize the wan Serf
 	s.serf, err = s.setupSerf(config.SerfConfig, s.eventCh, serfSnapshot)
 	if err != nil {
@@ -1356,6 +1379,7 @@ func (s *Server) setupRaft() error {
 	// Create the FSM
 	fsmConfig := &FSMConfig{
 		EvalBroker:         s.evalBroker,
+		BatchQueue:         s.batchJobQueue,
 		Periodic:           s.periodicDispatcher,
 		Blocked:            s.blockedEvals,
 		Encrypter:          s.encrypter,