1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
5 // Package scheduler uses a resizable worker pool to execute
6 // containers in priority order.
14 "git.arvados.org/arvados.git/sdk/go/arvados"
15 "git.arvados.org/arvados.git/sdk/go/ctxlog"
16 "github.com/prometheus/client_golang/prometheus"
17 "github.com/sirupsen/logrus"
20 // A Scheduler maps queued containers onto unallocated workers in
21 // priority order, creating new workers if needed. It locks containers
22 // that can be mapped onto existing/pending workers, and starts them
25 // A Scheduler unlocks any containers that are locked but can't be
26 // mapped. (For example, this happens when the cloud provider reaches
27 // quota/capacity and a previously mappable container's priority is
28 // surpassed by a newer container.)
30 // If it encounters errors while creating new workers, a Scheduler
31 // shuts down idle workers, in case they are consuming quota.
32 type Scheduler struct {
33 logger logrus.FieldLogger
34 client *arvados.Client
37 reg *prometheus.Registry
38 staleLockTimeout time.Duration
39 queueUpdateInterval time.Duration
41 uuidOp map[string]string // operation in progress: "lock", "cancel", ...
49 last503time time.Time // last time API responded 503
50 maxConcurrency int // dynamic container limit (0 = unlimited), see runQueue()
52 mContainersAllocatedNotStarted prometheus.Gauge
53 mContainersNotAllocatedOverQuota prometheus.Gauge
54 mLongestWaitTimeSinceQueue prometheus.Gauge
57 // New returns a new unstarted Scheduler.
59 // Any given queue and pool should not be used by more than one
60 // scheduler at a time.
61 func New(ctx context.Context, client *arvados.Client, queue ContainerQueue, pool WorkerPool, reg *prometheus.Registry, staleLockTimeout, queueUpdateInterval time.Duration) *Scheduler {
63 logger: ctxlog.FromContext(ctx),
68 staleLockTimeout: staleLockTimeout,
69 queueUpdateInterval: queueUpdateInterval,
70 wakeup: time.NewTimer(time.Second),
71 stop: make(chan struct{}),
72 stopped: make(chan struct{}),
73 uuidOp: map[string]string{},
75 sch.registerMetrics(reg)
79 func (sch *Scheduler) registerMetrics(reg *prometheus.Registry) {
81 reg = prometheus.NewRegistry()
83 sch.mContainersAllocatedNotStarted = prometheus.NewGauge(prometheus.GaugeOpts{
85 Subsystem: "dispatchcloud",
86 Name: "containers_allocated_not_started",
87 Help: "Number of containers allocated to a worker but not started yet (worker is booting).",
89 reg.MustRegister(sch.mContainersAllocatedNotStarted)
90 sch.mContainersNotAllocatedOverQuota = prometheus.NewGauge(prometheus.GaugeOpts{
92 Subsystem: "dispatchcloud",
93 Name: "containers_not_allocated_over_quota",
94 Help: "Number of containers not allocated to a worker because the system has hit a quota.",
96 reg.MustRegister(sch.mContainersNotAllocatedOverQuota)
97 sch.mLongestWaitTimeSinceQueue = prometheus.NewGauge(prometheus.GaugeOpts{
99 Subsystem: "dispatchcloud",
100 Name: "containers_longest_wait_time_seconds",
101 Help: "Current longest wait time of any container since queuing, and before the start of crunch-run.",
103 reg.MustRegister(sch.mLongestWaitTimeSinceQueue)
106 func (sch *Scheduler) updateMetrics() {
107 earliest := time.Time{}
108 entries, _ := sch.queue.Entries()
109 running := sch.pool.Running()
110 for _, ent := range entries {
111 if ent.Container.Priority > 0 &&
112 (ent.Container.State == arvados.ContainerStateQueued || ent.Container.State == arvados.ContainerStateLocked) {
113 // Exclude containers that are preparing to run the payload (i.e.
114 // ContainerStateLocked and running on a worker, most likely loading the
116 if _, ok := running[ent.Container.UUID]; !ok {
117 if ent.Container.CreatedAt.Before(earliest) || earliest.IsZero() {
118 earliest = ent.Container.CreatedAt
123 if !earliest.IsZero() {
124 sch.mLongestWaitTimeSinceQueue.Set(time.Since(earliest).Seconds())
126 sch.mLongestWaitTimeSinceQueue.Set(0)
130 // Start starts the scheduler.
131 func (sch *Scheduler) Start() {
132 go sch.runOnce.Do(sch.run)
135 // Stop stops the scheduler. No other method should be called after
137 func (sch *Scheduler) Stop() {
142 func (sch *Scheduler) run() {
143 defer close(sch.stopped)
145 // Ensure the queue is fetched once before attempting anything.
146 for err := sch.queue.Update(); err != nil; err = sch.queue.Update() {
147 sch.logger.Errorf("error updating queue: %s", err)
148 d := sch.queueUpdateInterval / 10
152 sch.logger.Infof("waiting %s before retry", d)
156 // Keep the queue up to date.
157 poll := time.NewTicker(sch.queueUpdateInterval)
161 err := sch.queue.Update()
163 sch.logger.Errorf("error updating queue: %s", err)
169 sch.logger.Infof("FixStaleLocks starting.")
171 sch.logger.Infof("FixStaleLocks finished (%s), starting scheduling.", time.Since(t0))
173 poolNotify := sch.pool.Subscribe()
174 defer sch.pool.Unsubscribe(poolNotify)
176 queueNotify := sch.queue.Subscribe()
177 defer sch.queue.Unsubscribe(queueNotify)