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()
51 supervisorFraction float64 // maximum fraction of "supervisor" containers (these are containers who's main job is to launch other containers, e.g. workflow runners)
52 maxInstances int // maximum number of instances the pool will bring up (0 = unlimited)
54 mContainersAllocatedNotStarted prometheus.Gauge
55 mContainersNotAllocatedOverQuota prometheus.Gauge
56 mLongestWaitTimeSinceQueue prometheus.Gauge
57 mLast503Time prometheus.Gauge
58 mMaxContainerConcurrency prometheus.Gauge
61 // New returns a new unstarted Scheduler.
63 // Any given queue and pool should not be used by more than one
64 // scheduler at a time.
65 func New(ctx context.Context, client *arvados.Client, queue ContainerQueue, pool WorkerPool, reg *prometheus.Registry, staleLockTimeout, queueUpdateInterval time.Duration, maxInstances int, supervisorFraction float64) *Scheduler {
67 logger: ctxlog.FromContext(ctx),
72 staleLockTimeout: staleLockTimeout,
73 queueUpdateInterval: queueUpdateInterval,
74 wakeup: time.NewTimer(time.Second),
75 stop: make(chan struct{}),
76 stopped: make(chan struct{}),
77 uuidOp: map[string]string{},
78 maxConcurrency: maxInstances, // initial value -- will be dynamically adjusted
79 supervisorFraction: supervisorFraction,
80 maxInstances: maxInstances,
82 sch.registerMetrics(reg)
86 func (sch *Scheduler) registerMetrics(reg *prometheus.Registry) {
88 reg = prometheus.NewRegistry()
90 sch.mContainersAllocatedNotStarted = prometheus.NewGauge(prometheus.GaugeOpts{
92 Subsystem: "dispatchcloud",
93 Name: "containers_allocated_not_started",
94 Help: "Number of containers allocated to a worker but not started yet (worker is booting).",
96 reg.MustRegister(sch.mContainersAllocatedNotStarted)
97 sch.mContainersNotAllocatedOverQuota = prometheus.NewGauge(prometheus.GaugeOpts{
99 Subsystem: "dispatchcloud",
100 Name: "containers_not_allocated_over_quota",
101 Help: "Number of containers not allocated to a worker because the system has hit a quota.",
103 reg.MustRegister(sch.mContainersNotAllocatedOverQuota)
104 sch.mLongestWaitTimeSinceQueue = prometheus.NewGauge(prometheus.GaugeOpts{
105 Namespace: "arvados",
106 Subsystem: "dispatchcloud",
107 Name: "containers_longest_wait_time_seconds",
108 Help: "Current longest wait time of any container since queuing, and before the start of crunch-run.",
110 reg.MustRegister(sch.mLongestWaitTimeSinceQueue)
111 sch.mLast503Time = prometheus.NewGauge(prometheus.GaugeOpts{
112 Namespace: "arvados",
113 Subsystem: "dispatchcloud",
114 Name: "last_503_time",
115 Help: "Time of most recent 503 error received from API.",
117 reg.MustRegister(sch.mLast503Time)
118 sch.mMaxContainerConcurrency = prometheus.NewGauge(prometheus.GaugeOpts{
119 Namespace: "arvados",
120 Subsystem: "dispatchcloud",
121 Name: "max_concurrent_containers",
122 Help: "Dynamically assigned limit on number of containers scheduled concurrency, set after receiving 503 errors from API.",
124 reg.MustRegister(sch.mMaxContainerConcurrency)
127 func (sch *Scheduler) updateMetrics() {
128 earliest := time.Time{}
129 entries, _ := sch.queue.Entries()
130 running := sch.pool.Running()
131 for _, ent := range entries {
132 if ent.Container.Priority > 0 &&
133 (ent.Container.State == arvados.ContainerStateQueued || ent.Container.State == arvados.ContainerStateLocked) {
134 // Exclude containers that are preparing to run the payload (i.e.
135 // ContainerStateLocked and running on a worker, most likely loading the
137 if _, ok := running[ent.Container.UUID]; !ok {
138 if ent.Container.CreatedAt.Before(earliest) || earliest.IsZero() {
139 earliest = ent.Container.CreatedAt
144 if !earliest.IsZero() {
145 sch.mLongestWaitTimeSinceQueue.Set(time.Since(earliest).Seconds())
147 sch.mLongestWaitTimeSinceQueue.Set(0)
151 // Start starts the scheduler.
152 func (sch *Scheduler) Start() {
153 go sch.runOnce.Do(sch.run)
156 // Stop stops the scheduler. No other method should be called after
158 func (sch *Scheduler) Stop() {
163 func (sch *Scheduler) run() {
164 defer close(sch.stopped)
166 // Ensure the queue is fetched once before attempting anything.
167 for err := sch.queue.Update(); err != nil; err = sch.queue.Update() {
168 sch.logger.Errorf("error updating queue: %s", err)
169 d := sch.queueUpdateInterval / 10
173 sch.logger.Infof("waiting %s before retry", d)
177 // Keep the queue up to date.
180 starttime := time.Now()
181 err := sch.queue.Update()
183 sch.logger.Errorf("error updating queue: %s", err)
185 // If the previous update took a long time,
186 // that probably means the server is
187 // overloaded, so wait that long before doing
188 // another. Otherwise, wait for the configured
190 delay := time.Since(starttime)
191 if delay < sch.queueUpdateInterval {
192 delay = sch.queueUpdateInterval
199 sch.logger.Infof("FixStaleLocks starting.")
201 sch.logger.Infof("FixStaleLocks finished (%s), starting scheduling.", time.Since(t0))
203 poolNotify := sch.pool.Subscribe()
204 defer sch.pool.Unsubscribe(poolNotify)
206 queueNotify := sch.queue.Subscribe()
207 defer sch.queue.Unsubscribe(queueNotify)