16811: Add a test that system users/groups can't be deleted.
[arvados.git] / lib / dispatchcloud / scheduler / run_queue.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package scheduler
6
7 import (
8         "sort"
9         "time"
10
11         "git.arvados.org/arvados.git/lib/dispatchcloud/container"
12         "git.arvados.org/arvados.git/sdk/go/arvados"
13         "github.com/sirupsen/logrus"
14 )
15
16 func (sch *Scheduler) runQueue() {
17         unsorted, _ := sch.queue.Entries()
18         sorted := make([]container.QueueEnt, 0, len(unsorted))
19         for _, ent := range unsorted {
20                 sorted = append(sorted, ent)
21         }
22         sort.Slice(sorted, func(i, j int) bool {
23                 return sorted[i].Container.Priority > sorted[j].Container.Priority
24         })
25
26         running := sch.pool.Running()
27         unalloc := sch.pool.Unallocated()
28
29         sch.logger.WithFields(logrus.Fields{
30                 "Containers": len(sorted),
31                 "Processes":  len(running),
32         }).Debug("runQueue")
33
34         dontstart := map[arvados.InstanceType]bool{}
35         var overquota []container.QueueEnt // entries that are unmappable because of worker pool quota
36
37 tryrun:
38         for i, ctr := range sorted {
39                 ctr, it := ctr.Container, ctr.InstanceType
40                 logger := sch.logger.WithFields(logrus.Fields{
41                         "ContainerUUID": ctr.UUID,
42                         "InstanceType":  it.Name,
43                 })
44                 if _, running := running[ctr.UUID]; running || ctr.Priority < 1 {
45                         continue
46                 }
47                 switch ctr.State {
48                 case arvados.ContainerStateQueued:
49                         if unalloc[it] < 1 && sch.pool.AtQuota() {
50                                 logger.Debug("not locking: AtQuota and no unalloc workers")
51                                 overquota = sorted[i:]
52                                 break tryrun
53                         }
54                         if sch.pool.KillContainer(ctr.UUID, "about to lock") {
55                                 logger.Info("not locking: crunch-run process from previous attempt has not exited")
56                                 continue
57                         }
58                         go sch.lockContainer(logger, ctr.UUID)
59                         unalloc[it]--
60                 case arvados.ContainerStateLocked:
61                         if unalloc[it] > 0 {
62                                 unalloc[it]--
63                         } else if sch.pool.AtQuota() {
64                                 // Don't let lower-priority containers
65                                 // starve this one by using keeping
66                                 // idle workers alive on different
67                                 // instance types.
68                                 logger.Debug("unlocking: AtQuota and no unalloc workers")
69                                 sch.queue.Unlock(ctr.UUID)
70                                 overquota = sorted[i:]
71                                 break tryrun
72                         } else if logger.Info("creating new instance"); sch.pool.Create(it) {
73                                 // Success. (Note pool.Create works
74                                 // asynchronously and does its own
75                                 // logging, so we don't need to.)
76                         } else {
77                                 // Failed despite not being at quota,
78                                 // e.g., cloud ops throttled.  TODO:
79                                 // avoid getting starved here if
80                                 // instances of a specific type always
81                                 // fail.
82                                 continue
83                         }
84
85                         if dontstart[it] {
86                                 // We already tried & failed to start
87                                 // a higher-priority container on the
88                                 // same instance type. Don't let this
89                                 // one sneak in ahead of it.
90                         } else if sch.pool.KillContainer(ctr.UUID, "about to start") {
91                                 logger.Info("not restarting yet: crunch-run process from previous attempt has not exited")
92                         } else if sch.pool.StartContainer(it, ctr) {
93                                 // Success.
94                         } else {
95                                 dontstart[it] = true
96                         }
97                 }
98         }
99
100         if len(overquota) > 0 {
101                 // Unlock any containers that are unmappable while
102                 // we're at quota.
103                 for _, ctr := range overquota {
104                         ctr := ctr.Container
105                         if ctr.State == arvados.ContainerStateLocked {
106                                 logger := sch.logger.WithField("ContainerUUID", ctr.UUID)
107                                 logger.Debug("unlock because pool capacity is used by higher priority containers")
108                                 err := sch.queue.Unlock(ctr.UUID)
109                                 if err != nil {
110                                         logger.WithError(err).Warn("error unlocking")
111                                 }
112                         }
113                 }
114                 // Shut down idle workers that didn't get any
115                 // containers mapped onto them before we hit quota.
116                 for it, n := range unalloc {
117                         if n < 1 {
118                                 continue
119                         }
120                         sch.pool.Shutdown(it)
121                 }
122         }
123 }
124
125 // Lock the given container. Should be called in a new goroutine.
126 func (sch *Scheduler) lockContainer(logger logrus.FieldLogger, uuid string) {
127         if !sch.uuidLock(uuid, "lock") {
128                 return
129         }
130         defer sch.uuidUnlock(uuid)
131         if ctr, ok := sch.queue.Get(uuid); !ok || ctr.State != arvados.ContainerStateQueued {
132                 // This happens if the container has been cancelled or
133                 // locked since runQueue called sch.queue.Entries(),
134                 // possibly by a lockContainer() call from a previous
135                 // runQueue iteration. In any case, we will respond
136                 // appropriately on the next runQueue iteration, which
137                 // will have already been triggered by the queue
138                 // update.
139                 logger.WithField("State", ctr.State).Debug("container no longer queued by the time we decided to lock it, doing nothing")
140                 return
141         }
142         err := sch.queue.Lock(uuid)
143         if err != nil {
144                 logger.WithError(err).Warn("error locking container")
145                 return
146         }
147         logger.Debug("lock succeeded")
148         ctr, ok := sch.queue.Get(uuid)
149         if !ok {
150                 logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
151         } else if ctr.State != arvados.ContainerStateLocked {
152                 logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
153         }
154 }
155
156 // Acquire a non-blocking lock for specified UUID, returning true if
157 // successful.  The op argument is used only for debug logs.
158 //
159 // If the lock is not available, uuidLock arranges to wake up the
160 // scheduler after a short delay, so it can retry whatever operation
161 // is trying to get the lock (if that operation is still worth doing).
162 //
163 // This mechanism helps avoid spamming the controller/database with
164 // concurrent updates for any single container, even when the
165 // scheduler loop is running frequently.
166 func (sch *Scheduler) uuidLock(uuid, op string) bool {
167         sch.mtx.Lock()
168         defer sch.mtx.Unlock()
169         logger := sch.logger.WithFields(logrus.Fields{
170                 "ContainerUUID": uuid,
171                 "Op":            op,
172         })
173         if op, locked := sch.uuidOp[uuid]; locked {
174                 logger.Debugf("uuidLock not available, Op=%s in progress", op)
175                 // Make sure the scheduler loop wakes up to retry.
176                 sch.wakeup.Reset(time.Second / 4)
177                 return false
178         }
179         logger.Debug("uuidLock acquired")
180         sch.uuidOp[uuid] = op
181         return true
182 }
183
184 func (sch *Scheduler) uuidUnlock(uuid string) {
185         sch.mtx.Lock()
186         defer sch.mtx.Unlock()
187         delete(sch.uuidOp, uuid)
188 }