1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
5 // Package dispatch is a helper library for building Arvados container
16 "git.arvados.org/arvados.git/lib/dispatchcloud"
17 "git.arvados.org/arvados.git/sdk/go/arvados"
18 "git.arvados.org/arvados.git/sdk/go/arvadosclient"
19 "github.com/sirupsen/logrus"
23 Queued = arvados.ContainerStateQueued
24 Locked = arvados.ContainerStateLocked
25 Running = arvados.ContainerStateRunning
26 Complete = arvados.ContainerStateComplete
27 Cancelled = arvados.ContainerStateCancelled
30 type Logger interface {
31 Printf(string, ...interface{})
32 Warnf(string, ...interface{})
33 Debugf(string, ...interface{})
37 type Dispatcher struct {
38 Arv *arvadosclient.ArvadosClient
42 // Batch size for container queries
45 // Queue polling frequency
46 PollPeriod time.Duration
48 // Time to wait between successive attempts to run the same container
49 MinRetryPeriod time.Duration
51 // Func that implements the container lifecycle. Must be set
52 // to a non-nil DispatchFunc before calling Run().
53 RunContainer DispatchFunc
55 auth arvados.APIClientAuthorization
57 trackers map[string]*runTracker
61 // A DispatchFunc executes a container (if the container record is
62 // Locked) or resume monitoring an already-running container, and wait
63 // until that container exits.
65 // While the container runs, the DispatchFunc should listen for
66 // updated container records on the provided channel. When the channel
67 // closes, the DispatchFunc should stop the container if it's still
68 // running, and return.
70 // The DispatchFunc should not return until the container is finished.
71 type DispatchFunc func(*Dispatcher, arvados.Container, <-chan arvados.Container) error
73 // Run watches the API server's queue for containers that are either
74 // ready to run and available to lock, or are already locked by this
75 // dispatcher's token. When a new one appears, Run calls RunContainer
76 // in a new goroutine.
77 func (d *Dispatcher) Run(ctx context.Context) error {
79 d.Logger = logrus.StandardLogger()
82 err := d.Arv.Call("GET", "api_client_authorizations", "", "current", nil, &d.auth)
84 return fmt.Errorf("error getting my token UUID: %v", err)
87 d.throttle.hold = d.MinRetryPeriod
89 poll := time.NewTicker(d.PollPeriod)
104 todo := make(map[string]*runTracker)
106 // Make a copy of trackers
107 for uuid, tracker := range d.trackers {
112 // Containers I currently own (Locked/Running)
113 querySuccess := d.checkForUpdates([][]interface{}{
114 {"locked_by_uuid", "=", d.auth.UUID}}, todo)
116 // Containers I should try to dispatch
117 querySuccess = d.checkForUpdates([][]interface{}{
118 {"state", "=", Queued},
119 {"priority", ">", "0"}}, todo) && querySuccess
122 // There was an error in one of the previous queries,
123 // we probably didn't get updates for all the
124 // containers we should have. Don't check them
125 // individually because it may be expensive.
129 // Containers I know about but didn't fall into the
130 // above two categories (probably Complete/Cancelled)
132 for uuid := range todo {
133 missed = append(missed, uuid)
136 for len(missed) > 0 {
138 if len(missed) > 20 {
145 querySuccess = d.checkForUpdates([][]interface{}{
146 {"uuid", "in", batch}}, todo) && querySuccess
150 // There was an error in one of the previous queries, we probably
151 // didn't see all the containers we should have, so don't shut down
152 // the missed containers.
156 // Containers that I know about that didn't show up in any
157 // query should be let go.
158 for uuid, tracker := range todo {
159 d.Logger.Printf("Container %q not returned by any query, stopping tracking.", uuid)
166 // Start a runner in a new goroutine, and send the initial container
167 // record to its updates channel.
168 func (d *Dispatcher) start(c arvados.Container) *runTracker {
169 tracker := &runTracker{
170 updates: make(chan arvados.Container, 1),
175 fallbackState := Queued
176 err := d.RunContainer(d, c, tracker.updates)
178 text := fmt.Sprintf("Error running container %s: %s", c.UUID, err)
179 if err, ok := err.(dispatchcloud.ConstraintsNotSatisfiableError); ok {
180 fallbackState = Cancelled
181 var logBuf bytes.Buffer
182 fmt.Fprintf(&logBuf, "cannot run container %s: %s\n", c.UUID, err)
183 if len(err.AvailableTypes) == 0 {
184 fmt.Fprint(&logBuf, "No instance types are configured.\n")
186 fmt.Fprint(&logBuf, "Available instance types:\n")
187 for _, t := range err.AvailableTypes {
189 "Type %q: %d VCPUs, %d RAM, %d Scratch, %f Price\n",
190 t.Name, t.VCPUs, t.RAM, t.Scratch, t.Price)
193 text = logBuf.String()
195 d.Logger.Printf("%s", text)
196 lr := arvadosclient.Dict{"log": arvadosclient.Dict{
197 "object_uuid": c.UUID,
198 "event_type": "dispatch",
199 "properties": map[string]string{"text": text}}}
200 d.Arv.Create("logs", lr, nil)
202 // If checkListForUpdates() doesn't close the tracker
203 // after 2 queue updates, try to move the container to
204 // the fallback state, which should eventually work
205 // and cause the tracker to close.
207 for upd := range tracker.updates {
209 if upd.State == Locked || upd.State == Running {
210 // Tracker didn't clean up before
211 // returning -- or this is the first
212 // update and it contains stale
213 // information from before
214 // RunContainer() returned.
216 // Avoid generating confusing
217 // logs / API calls in the
221 d.Logger.Printf("container %s state is still %s, changing to %s", c.UUID, upd.State, fallbackState)
222 d.UpdateState(c.UUID, fallbackState)
229 func (d *Dispatcher) checkForUpdates(filters [][]interface{}, todo map[string]*runTracker) bool {
230 var countList arvados.ContainerList
231 params := arvadosclient.Dict{
235 "order": []string{"priority desc"}}
236 err := d.Arv.List("containers", params, &countList)
238 d.Logger.Warnf("error getting count of containers: %q", err)
241 itemsAvailable := countList.ItemsAvailable
242 params = arvadosclient.Dict{
245 "limit": d.BatchSize,
246 "order": []string{"priority desc"}}
249 params["offset"] = offset
251 // This list variable must be a new one declared
252 // inside the loop: otherwise, items in the API
253 // response would get deep-merged into the items
254 // loaded in previous iterations.
255 var list arvados.ContainerList
257 err := d.Arv.List("containers", params, &list)
259 d.Logger.Warnf("error getting list of containers: %q", err)
262 d.checkListForUpdates(list.Items, todo)
263 offset += len(list.Items)
264 if len(list.Items) == 0 || itemsAvailable <= offset {
270 func (d *Dispatcher) checkListForUpdates(containers []arvados.Container, todo map[string]*runTracker) {
273 if d.trackers == nil {
274 d.trackers = make(map[string]*runTracker)
277 for _, c := range containers {
278 tracker, alreadyTracking := d.trackers[c.UUID]
281 if c.LockedByUUID != "" && c.LockedByUUID != d.auth.UUID {
282 d.Logger.Debugf("ignoring %s locked by %s", c.UUID, c.LockedByUUID)
283 } else if alreadyTracking {
285 case Queued, Cancelled, Complete:
286 d.Logger.Debugf("update has %s in state %s, closing tracker", c.UUID, c.State)
288 delete(d.trackers, c.UUID)
289 case Locked, Running:
290 d.Logger.Debugf("update has %s in state %s, updating tracker", c.UUID, c.State)
296 if !d.throttle.Check(c.UUID) {
299 err := d.lock(c.UUID)
301 d.Logger.Warnf("error locking container %s: %s", c.UUID, err)
305 d.trackers[c.UUID] = d.start(c)
306 case Locked, Running:
307 if !d.throttle.Check(c.UUID) {
310 d.trackers[c.UUID] = d.start(c)
311 case Cancelled, Complete:
312 // no-op (we already stopped monitoring)
318 // UpdateState makes an API call to change the state of a container.
319 func (d *Dispatcher) UpdateState(uuid string, state arvados.ContainerState) error {
320 err := d.Arv.Update("containers", uuid,
322 "container": arvadosclient.Dict{"state": state},
325 d.Logger.Warnf("error updating container %s to state %q: %s", uuid, state, err)
330 // Lock makes the lock API call which updates the state of a container to Locked.
331 func (d *Dispatcher) lock(uuid string) error {
332 return d.Arv.Call("POST", "containers", uuid, "lock", nil, nil)
335 // Unlock makes the unlock API call which updates the state of a container to Queued.
336 func (d *Dispatcher) Unlock(uuid string) error {
337 return d.Arv.Call("POST", "containers", uuid, "unlock", nil, nil)
340 // TrackContainer ensures a tracker is running for the given UUID,
341 // regardless of the current state of the container (except: if the
342 // container is locked by a different dispatcher, a tracker will not
343 // be started). If the container is not in Locked or Running state,
344 // the new tracker will close down immediately.
346 // This allows the dispatcher to put its own RunContainer func into a
347 // cleanup phase (for example, to kill local processes created by a
348 // prevous dispatch process that are still running even though the
349 // container state is final) without the risk of having multiple
350 // goroutines monitoring the same UUID.
351 func (d *Dispatcher) TrackContainer(uuid string) error {
352 var cntr arvados.Container
353 err := d.Arv.Call("GET", "containers", uuid, "", nil, &cntr)
357 if cntr.LockedByUUID != "" && cntr.LockedByUUID != d.auth.UUID {
363 if _, alreadyTracking := d.trackers[uuid]; alreadyTracking {
366 if d.trackers == nil {
367 d.trackers = make(map[string]*runTracker)
369 d.trackers[uuid] = d.start(cntr)
371 case Queued, Cancelled, Complete:
372 d.trackers[uuid].close()
377 type runTracker struct {
379 updates chan arvados.Container
383 func (tracker *runTracker) close() {
384 if !tracker.closing {
385 close(tracker.updates)
387 tracker.closing = true
390 func (tracker *runTracker) update(c arvados.Container) {
395 case <-tracker.updates:
396 tracker.logger.Debugf("runner is handling updates slowly, discarded previous update for %s", c.UUID)