lib/dispatchcloud/scheduler/run_queue_test.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         "context"
   9         "sync"
  10         "time"
  11
  12         "git.arvados.org/arvados.git/lib/dispatchcloud/test"
  13         "git.arvados.org/arvados.git/lib/dispatchcloud/worker"
  14         "git.arvados.org/arvados.git/sdk/go/arvados"
  15         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  16         check "gopkg.in/check.v1"
  17 )
  18
  19 var (
  20         // arbitrary example container UUIDs
  21         uuids = func() (r []string) {
  22                 for i := 0; i < 16; i++ {
  23                         r = append(r, test.ContainerUUID(i))
  24                 }
  25                 return
  26         }()
  27 )
  28
  29 type stubQuotaError struct {
  30         error
  31 }
  32
  33 func (stubQuotaError) IsQuotaError() bool { return true }
  34
  35 type stubPool struct {
  36         notify    <-chan struct{}
  37         unalloc   map[arvados.InstanceType]int // idle+booting+unknown
  38         idle      map[arvados.InstanceType]int
  39         unknown   map[arvados.InstanceType]int
  40         running   map[string]time.Time
  41         atQuota   bool
  42         canCreate int
  43         creates   []arvados.InstanceType
  44         starts    []string
  45         shutdowns int
  46         sync.Mutex
  47 }
  48
  49 func (p *stubPool) AtQuota() bool               { return p.atQuota }
  50 func (p *stubPool) Subscribe() <-chan struct{}  { return p.notify }
  51 func (p *stubPool) Unsubscribe(<-chan struct{}) {}
  52 func (p *stubPool) Running() map[string]time.Time {
  53         p.Lock()
  54         defer p.Unlock()
  55         r := map[string]time.Time{}
  56         for k, v := range p.running {
  57                 r[k] = v
  58         }
  59         return r
  60 }
  61 func (p *stubPool) Unallocated() map[arvados.InstanceType]int {
  62         p.Lock()
  63         defer p.Unlock()
  64         r := map[arvados.InstanceType]int{}
  65         for it, n := range p.unalloc {
  66                 r[it] = n - p.unknown[it]
  67         }
  68         return r
  69 }
  70 func (p *stubPool) Create(it arvados.InstanceType) bool {
  71         p.Lock()
  72         defer p.Unlock()
  73         p.creates = append(p.creates, it)
  74         if p.canCreate < 1 {
  75                 return false
  76         }
  77         p.canCreate--
  78         p.unalloc[it]++
  79         return true
  80 }
  81 func (p *stubPool) ForgetContainer(uuid string) {
  82 }
  83 func (p *stubPool) KillContainer(uuid, reason string) bool {
  84         p.Lock()
  85         defer p.Unlock()
  86         delete(p.running, uuid)
  87         return true
  88 }
  89 func (p *stubPool) Shutdown(arvados.InstanceType) bool {
  90         p.shutdowns++
  91         return false
  92 }
  93 func (p *stubPool) CountWorkers() map[worker.State]int {
  94         p.Lock()
  95         defer p.Unlock()
  96         return map[worker.State]int{
  97                 worker.StateBooting: len(p.unalloc) - len(p.idle),
  98                 worker.StateIdle:    len(p.idle),
  99                 worker.StateRunning: len(p.running),
 100                 worker.StateUnknown: len(p.unknown),
 101         }
 102 }
 103 func (p *stubPool) StartContainer(it arvados.InstanceType, ctr arvados.Container) bool {
 104         p.Lock()
 105         defer p.Unlock()
 106         p.starts = append(p.starts, ctr.UUID)
 107         if p.idle[it] == 0 {
 108                 return false
 109         }
 110         p.idle[it]--
 111         p.unalloc[it]--
 112         p.running[ctr.UUID] = time.Time{}
 113         return true
 114 }
 115
 116 func chooseType(ctr *arvados.Container) (arvados.InstanceType, error) {
 117         return test.InstanceType(ctr.RuntimeConstraints.VCPUs), nil
 118 }
 119
 120 var _ = check.Suite(&SchedulerSuite{})
 121
 122 type SchedulerSuite struct{}
 123
 124 // Assign priority=4 container to idle node. Create a new instance for
 125 // the priority=3 container. Don't try to start any priority<3
 126 // containers because priority=3 container didn't start
 127 // immediately. Don't try to create any other nodes after the failed
 128 // create.
 129 func (*SchedulerSuite) TestUseIdleWorkers(c *check.C) {
 130         ctx := ctxlog.Context(context.Background(), ctxlog.TestLogger(c))
 131         queue := test.Queue{
 132                 ChooseType: chooseType,
 133                 Containers: []arvados.Container{
 134                         {
 135                                 UUID:     test.ContainerUUID(1),
 136                                 Priority: 1,
 137                                 State:    arvados.ContainerStateLocked,
 138                                 RuntimeConstraints: arvados.RuntimeConstraints{
 139                                         VCPUs: 1,
 140                                         RAM:   1 << 30,
 141                                 },
 142                         },
 143                         {
 144                                 UUID:     test.ContainerUUID(2),
 145                                 Priority: 2,
 146                                 State:    arvados.ContainerStateLocked,
 147                                 RuntimeConstraints: arvados.RuntimeConstraints{
 148                                         VCPUs: 1,
 149                                         RAM:   1 << 30,
 150                                 },
 151                         },
 152                         {
 153                                 UUID:     test.ContainerUUID(3),
 154                                 Priority: 3,
 155                                 State:    arvados.ContainerStateLocked,
 156                                 RuntimeConstraints: arvados.RuntimeConstraints{
 157                                         VCPUs: 1,
 158                                         RAM:   1 << 30,
 159                                 },
 160                         },
 161                         {
 162                                 UUID:     test.ContainerUUID(4),
 163                                 Priority: 4,
 164                                 State:    arvados.ContainerStateLocked,
 165                                 RuntimeConstraints: arvados.RuntimeConstraints{
 166                                         VCPUs: 1,
 167                                         RAM:   1 << 30,
 168                                 },
 169                         },
 170                 },
 171         }
 172         queue.Update()
 173         pool := stubPool{
 174                 unalloc: map[arvados.InstanceType]int{
 175                         test.InstanceType(1): 1,
 176                         test.InstanceType(2): 2,
 177                 },
 178                 idle: map[arvados.InstanceType]int{
 179                         test.InstanceType(1): 1,
 180                         test.InstanceType(2): 2,
 181                 },
 182                 running:   map[string]time.Time{},
 183                 canCreate: 0,
 184         }
 185         New(ctx, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 186         c.Check(pool.creates, check.DeepEquals, []arvados.InstanceType{test.InstanceType(1)})
 187         c.Check(pool.starts, check.DeepEquals, []string{test.ContainerUUID(4)})
 188         c.Check(pool.running, check.HasLen, 1)
 189         for uuid := range pool.running {
 190                 c.Check(uuid, check.Equals, uuids[4])
 191         }
 192 }
 193
 194 // If Create() fails, shutdown some nodes, and don't call Create()
 195 // again.  Don't call Create() at all if AtQuota() is true.
 196 func (*SchedulerSuite) TestShutdownAtQuota(c *check.C) {
 197         ctx := ctxlog.Context(context.Background(), ctxlog.TestLogger(c))
 198         for quota := 0; quota < 2; quota++ {
 199                 c.Logf("quota=%d", quota)
 200                 shouldCreate := []arvados.InstanceType{}
 201                 for i := 0; i < quota; i++ {
 202                         shouldCreate = append(shouldCreate, test.InstanceType(3))
 203                 }
 204                 queue := test.Queue{
 205                         ChooseType: chooseType,
 206                         Containers: []arvados.Container{
 207                                 {
 208                                         UUID:     test.ContainerUUID(2),
 209                                         Priority: 2,
 210                                         State:    arvados.ContainerStateLocked,
 211                                         RuntimeConstraints: arvados.RuntimeConstraints{
 212                                                 VCPUs: 2,
 213                                                 RAM:   2 << 30,
 214                                         },
 215                                 },
 216                                 {
 217                                         UUID:     test.ContainerUUID(3),
 218                                         Priority: 3,
 219                                         State:    arvados.ContainerStateLocked,
 220                                         RuntimeConstraints: arvados.RuntimeConstraints{
 221                                                 VCPUs: 3,
 222                                                 RAM:   3 << 30,
 223                                         },
 224                                 },
 225                         },
 226                 }
 227                 queue.Update()
 228                 pool := stubPool{
 229                         atQuota: quota == 0,
 230                         unalloc: map[arvados.InstanceType]int{
 231                                 test.InstanceType(2): 2,
 232                         },
 233                         idle: map[arvados.InstanceType]int{
 234                                 test.InstanceType(2): 2,
 235                         },
 236                         running:   map[string]time.Time{},
 237                         creates:   []arvados.InstanceType{},
 238                         starts:    []string{},
 239                         canCreate: 0,
 240                 }
 241                 New(ctx, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 242                 c.Check(pool.creates, check.DeepEquals, shouldCreate)
 243                 c.Check(pool.starts, check.DeepEquals, []string{})
 244                 c.Check(pool.shutdowns, check.Not(check.Equals), 0)
 245         }
 246 }
 247
 248 // Start lower-priority containers while waiting for new/existing
 249 // workers to come up for higher-priority containers.
 250 func (*SchedulerSuite) TestStartWhileCreating(c *check.C) {
 251         ctx := ctxlog.Context(context.Background(), ctxlog.TestLogger(c))
 252         pool := stubPool{
 253                 unalloc: map[arvados.InstanceType]int{
 254                         test.InstanceType(1): 2,
 255                         test.InstanceType(2): 2,
 256                 },
 257                 idle: map[arvados.InstanceType]int{
 258                         test.InstanceType(1): 1,
 259                         test.InstanceType(2): 1,
 260                 },
 261                 running:   map[string]time.Time{},
 262                 canCreate: 4,
 263         }
 264         queue := test.Queue{
 265                 ChooseType: chooseType,
 266                 Containers: []arvados.Container{
 267                         {
 268                                 // create a new worker
 269                                 UUID:     test.ContainerUUID(1),
 270                                 Priority: 1,
 271                                 State:    arvados.ContainerStateLocked,
 272                                 RuntimeConstraints: arvados.RuntimeConstraints{
 273                                         VCPUs: 1,
 274                                         RAM:   1 << 30,
 275                                 },
 276                         },
 277                         {
 278                                 // tentatively map to unalloc worker
 279                                 UUID:     test.ContainerUUID(2),
 280                                 Priority: 2,
 281                                 State:    arvados.ContainerStateLocked,
 282                                 RuntimeConstraints: arvados.RuntimeConstraints{
 283                                         VCPUs: 1,
 284                                         RAM:   1 << 30,
 285                                 },
 286                         },
 287                         {
 288                                 // start now on idle worker
 289                                 UUID:     test.ContainerUUID(3),
 290                                 Priority: 3,
 291                                 State:    arvados.ContainerStateLocked,
 292                                 RuntimeConstraints: arvados.RuntimeConstraints{
 293                                         VCPUs: 1,
 294                                         RAM:   1 << 30,
 295                                 },
 296                         },
 297                         {
 298                                 // create a new worker
 299                                 UUID:     test.ContainerUUID(4),
 300                                 Priority: 4,
 301                                 State:    arvados.ContainerStateLocked,
 302                                 RuntimeConstraints: arvados.RuntimeConstraints{
 303                                         VCPUs: 2,
 304                                         RAM:   2 << 30,
 305                                 },
 306                         },
 307                         {
 308                                 // tentatively map to unalloc worker
 309                                 UUID:     test.ContainerUUID(5),
 310                                 Priority: 5,
 311                                 State:    arvados.ContainerStateLocked,
 312                                 RuntimeConstraints: arvados.RuntimeConstraints{
 313                                         VCPUs: 2,
 314                                         RAM:   2 << 30,
 315                                 },
 316                         },
 317                         {
 318                                 // start now on idle worker
 319                                 UUID:     test.ContainerUUID(6),
 320                                 Priority: 6,
 321                                 State:    arvados.ContainerStateLocked,
 322                                 RuntimeConstraints: arvados.RuntimeConstraints{
 323                                         VCPUs: 2,
 324                                         RAM:   2 << 30,
 325                                 },
 326                         },
 327                 },
 328         }
 329         queue.Update()
 330         New(ctx, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 331         c.Check(pool.creates, check.DeepEquals, []arvados.InstanceType{test.InstanceType(2), test.InstanceType(1)})
 332         c.Check(pool.starts, check.DeepEquals, []string{uuids[6], uuids[5], uuids[3], uuids[2]})
 333         running := map[string]bool{}
 334         for uuid, t := range pool.running {
 335                 if t.IsZero() {
 336                         running[uuid] = false
 337                 } else {
 338                         running[uuid] = true
 339                 }
 340         }
 341         c.Check(running, check.DeepEquals, map[string]bool{uuids[3]: false, uuids[6]: false})
 342 }
 343
 344 func (*SchedulerSuite) TestKillNonexistentContainer(c *check.C) {
 345         ctx := ctxlog.Context(context.Background(), ctxlog.TestLogger(c))
 346         pool := stubPool{
 347                 unalloc: map[arvados.InstanceType]int{
 348                         test.InstanceType(2): 0,
 349                 },
 350                 idle: map[arvados.InstanceType]int{
 351                         test.InstanceType(2): 0,
 352                 },
 353                 running: map[string]time.Time{
 354                         test.ContainerUUID(2): time.Time{},
 355                 },
 356         }
 357         queue := test.Queue{
 358                 ChooseType: chooseType,
 359                 Containers: []arvados.Container{
 360                         {
 361                                 // create a new worker
 362                                 UUID:     test.ContainerUUID(1),
 363                                 Priority: 1,
 364                                 State:    arvados.ContainerStateLocked,
 365                                 RuntimeConstraints: arvados.RuntimeConstraints{
 366                                         VCPUs: 1,
 367                                         RAM:   1 << 30,
 368                                 },
 369                         },
 370                 },
 371         }
 372         queue.Update()
 373         sch := New(ctx, &queue, &pool, time.Millisecond, time.Millisecond)
 374         c.Check(pool.running, check.HasLen, 1)
 375         sch.sync()
 376         for deadline := time.Now().Add(time.Second); len(pool.Running()) > 0 && time.Now().Before(deadline); time.Sleep(time.Millisecond) {
 377         }
 378         c.Check(pool.Running(), check.HasLen, 0)
 379 }