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         "errors"
   9         "time"
  10
  11         "git.curoverse.com/arvados.git/lib/dispatchcloud/test"
  12         "git.curoverse.com/arvados.git/lib/dispatchcloud/worker"
  13         "git.curoverse.com/arvados.git/sdk/go/arvados"
  14         "github.com/Sirupsen/logrus"
  15         check "gopkg.in/check.v1"
  16 )
  17
  18 var (
  19         logger = logrus.StandardLogger()
  20
  21         // arbitrary example container UUIDs
  22         uuids = func() (r []string) {
  23                 for i := 0; i < 16; i++ {
  24                         r = append(r, test.ContainerUUID(i))
  25                 }
  26                 return
  27         }()
  28 )
  29
  30 type stubQuotaError struct {
  31         error
  32 }
  33
  34 func (stubQuotaError) IsQuotaError() bool { return true }
  35
  36 type stubPool struct {
  37         notify    <-chan struct{}
  38         unalloc   map[arvados.InstanceType]int // idle+booting+unknown
  39         idle      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 }
  47
  48 func (p *stubPool) AtQuota() bool                 { return p.atQuota }
  49 func (p *stubPool) Subscribe() <-chan struct{}    { return p.notify }
  50 func (p *stubPool) Unsubscribe(<-chan struct{})   {}
  51 func (p *stubPool) Running() map[string]time.Time { return p.running }
  52 func (p *stubPool) Unallocated() map[arvados.InstanceType]int {
  53         r := map[arvados.InstanceType]int{}
  54         for it, n := range p.unalloc {
  55                 r[it] = n
  56         }
  57         return r
  58 }
  59 func (p *stubPool) Create(it arvados.InstanceType) error {
  60         p.creates = append(p.creates, it)
  61         if p.canCreate < 1 {
  62                 return stubQuotaError{errors.New("quota")}
  63         }
  64         p.canCreate--
  65         p.unalloc[it]++
  66         return nil
  67 }
  68 func (p *stubPool) KillContainer(uuid string) {
  69         p.running[uuid] = time.Now()
  70 }
  71 func (p *stubPool) Shutdown(arvados.InstanceType) bool {
  72         p.shutdowns++
  73         return false
  74 }
  75 func (p *stubPool) Workers() map[worker.State]int {
  76         return map[worker.State]int{
  77                 worker.StateBooting: len(p.unalloc) - len(p.idle),
  78                 worker.StateIdle:    len(p.idle),
  79                 worker.StateRunning: len(p.running),
  80         }
  81 }
  82 func (p *stubPool) StartContainer(it arvados.InstanceType, ctr arvados.Container) bool {
  83         p.starts = append(p.starts, ctr.UUID)
  84         if p.idle[it] == 0 {
  85                 return false
  86         }
  87         p.idle[it]--
  88         p.unalloc[it]--
  89         p.running[ctr.UUID] = time.Time{}
  90         return true
  91 }
  92
  93 var _ = check.Suite(&SchedulerSuite{})
  94
  95 type SchedulerSuite struct{}
  96
  97 // Assign priority=4 container to idle node. Create a new instance for
  98 // the priority=3 container. Don't try to start any priority<3
  99 // containers because priority=3 container didn't start
 100 // immediately. Don't try to create any other nodes after the failed
 101 // create.
 102 func (*SchedulerSuite) TestUseIdleWorkers(c *check.C) {
 103         queue := test.Queue{
 104                 ChooseType: func(ctr *arvados.Container) (arvados.InstanceType, error) {
 105                         return test.InstanceType(ctr.RuntimeConstraints.VCPUs), nil
 106                 },
 107                 Containers: []arvados.Container{
 108                         {
 109                                 UUID:     test.ContainerUUID(1),
 110                                 Priority: 1,
 111                                 State:    arvados.ContainerStateLocked,
 112                                 RuntimeConstraints: arvados.RuntimeConstraints{
 113                                         VCPUs: 1,
 114                                         RAM:   1 << 30,
 115                                 },
 116                         },
 117                         {
 118                                 UUID:     test.ContainerUUID(2),
 119                                 Priority: 2,
 120                                 State:    arvados.ContainerStateLocked,
 121                                 RuntimeConstraints: arvados.RuntimeConstraints{
 122                                         VCPUs: 1,
 123                                         RAM:   1 << 30,
 124                                 },
 125                         },
 126                         {
 127                                 UUID:     test.ContainerUUID(3),
 128                                 Priority: 3,
 129                                 State:    arvados.ContainerStateLocked,
 130                                 RuntimeConstraints: arvados.RuntimeConstraints{
 131                                         VCPUs: 1,
 132                                         RAM:   1 << 30,
 133                                 },
 134                         },
 135                         {
 136                                 UUID:     test.ContainerUUID(4),
 137                                 Priority: 4,
 138                                 State:    arvados.ContainerStateLocked,
 139                                 RuntimeConstraints: arvados.RuntimeConstraints{
 140                                         VCPUs: 1,
 141                                         RAM:   1 << 30,
 142                                 },
 143                         },
 144                 },
 145         }
 146         queue.Update()
 147         pool := stubPool{
 148                 unalloc: map[arvados.InstanceType]int{
 149                         test.InstanceType(1): 1,
 150                         test.InstanceType(2): 2,
 151                 },
 152                 idle: map[arvados.InstanceType]int{
 153                         test.InstanceType(1): 1,
 154                         test.InstanceType(2): 2,
 155                 },
 156                 running:   map[string]time.Time{},
 157                 canCreate: 0,
 158         }
 159         New(logger, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 160         c.Check(pool.creates, check.DeepEquals, []arvados.InstanceType{test.InstanceType(1)})
 161         c.Check(pool.starts, check.DeepEquals, []string{test.ContainerUUID(4)})
 162         c.Check(pool.running, check.HasLen, 1)
 163         for uuid := range pool.running {
 164                 c.Check(uuid, check.Equals, uuids[4])
 165         }
 166 }
 167
 168 // Shutdown some nodes if Create() fails -- and without even calling
 169 // Create(), if AtQuota() is true.
 170 func (*SchedulerSuite) TestShutdownAtQuota(c *check.C) {
 171         for quota := 0; quota < 2; quota++ {
 172                 c.Logf("quota=%d", quota)
 173                 shouldCreate := []arvados.InstanceType{}
 174                 for i := 0; i < quota; i++ {
 175                         shouldCreate = append(shouldCreate, test.InstanceType(1))
 176                 }
 177                 queue := test.Queue{
 178                         ChooseType: func(ctr *arvados.Container) (arvados.InstanceType, error) {
 179                                 return test.InstanceType(ctr.RuntimeConstraints.VCPUs), nil
 180                         },
 181                         Containers: []arvados.Container{
 182                                 {
 183                                         UUID:     test.ContainerUUID(1),
 184                                         Priority: 1,
 185                                         State:    arvados.ContainerStateLocked,
 186                                         RuntimeConstraints: arvados.RuntimeConstraints{
 187                                                 VCPUs: 1,
 188                                                 RAM:   1 << 30,
 189                                         },
 190                                 },
 191                         },
 192                 }
 193                 queue.Update()
 194                 pool := stubPool{
 195                         atQuota: quota == 0,
 196                         unalloc: map[arvados.InstanceType]int{
 197                                 test.InstanceType(2): 2,
 198                         },
 199                         idle: map[arvados.InstanceType]int{
 200                                 test.InstanceType(2): 2,
 201                         },
 202                         running:   map[string]time.Time{},
 203                         creates:   []arvados.InstanceType{},
 204                         starts:    []string{},
 205                         canCreate: 0,
 206                 }
 207                 New(logger, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 208                 c.Check(pool.creates, check.DeepEquals, shouldCreate)
 209                 c.Check(pool.starts, check.DeepEquals, []string{})
 210                 c.Check(pool.shutdowns, check.Not(check.Equals), 0)
 211         }
 212 }
 213
 214 // Start lower-priority containers while waiting for new/existing
 215 // workers to come up for higher-priority containers.
 216 func (*SchedulerSuite) TestStartWhileCreating(c *check.C) {
 217         pool := stubPool{
 218                 unalloc: map[arvados.InstanceType]int{
 219                         test.InstanceType(1): 1,
 220                         test.InstanceType(2): 1,
 221                 },
 222                 idle: map[arvados.InstanceType]int{
 223                         test.InstanceType(1): 1,
 224                         test.InstanceType(2): 1,
 225                 },
 226                 running:   map[string]time.Time{},
 227                 canCreate: 4,
 228         }
 229         queue := test.Queue{
 230                 ChooseType: func(ctr *arvados.Container) (arvados.InstanceType, error) {
 231                         return test.InstanceType(ctr.RuntimeConstraints.VCPUs), nil
 232                 },
 233                 Containers: []arvados.Container{
 234                         {
 235                                 // create a new worker
 236                                 UUID:     test.ContainerUUID(1),
 237                                 Priority: 1,
 238                                 State:    arvados.ContainerStateLocked,
 239                                 RuntimeConstraints: arvados.RuntimeConstraints{
 240                                         VCPUs: 1,
 241                                         RAM:   1 << 30,
 242                                 },
 243                         },
 244                         {
 245                                 // tentatively map to unalloc worker
 246                                 UUID:     test.ContainerUUID(2),
 247                                 Priority: 2,
 248                                 State:    arvados.ContainerStateLocked,
 249                                 RuntimeConstraints: arvados.RuntimeConstraints{
 250                                         VCPUs: 1,
 251                                         RAM:   1 << 30,
 252                                 },
 253                         },
 254                         {
 255                                 // start now on idle worker
 256                                 UUID:     test.ContainerUUID(3),
 257                                 Priority: 3,
 258                                 State:    arvados.ContainerStateLocked,
 259                                 RuntimeConstraints: arvados.RuntimeConstraints{
 260                                         VCPUs: 1,
 261                                         RAM:   1 << 30,
 262                                 },
 263                         },
 264                         {
 265                                 // create a new worker
 266                                 UUID:     test.ContainerUUID(4),
 267                                 Priority: 4,
 268                                 State:    arvados.ContainerStateLocked,
 269                                 RuntimeConstraints: arvados.RuntimeConstraints{
 270                                         VCPUs: 2,
 271                                         RAM:   2 << 30,
 272                                 },
 273                         },
 274                         {
 275                                 // tentatively map to unalloc worker
 276                                 UUID:     test.ContainerUUID(5),
 277                                 Priority: 5,
 278                                 State:    arvados.ContainerStateLocked,
 279                                 RuntimeConstraints: arvados.RuntimeConstraints{
 280                                         VCPUs: 2,
 281                                         RAM:   2 << 30,
 282                                 },
 283                         },
 284                         {
 285                                 // start now on idle worker
 286                                 UUID:     test.ContainerUUID(6),
 287                                 Priority: 6,
 288                                 State:    arvados.ContainerStateLocked,
 289                                 RuntimeConstraints: arvados.RuntimeConstraints{
 290                                         VCPUs: 2,
 291                                         RAM:   2 << 30,
 292                                 },
 293                         },
 294                 },
 295         }
 296         queue.Update()
 297         New(logger, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 298         c.Check(pool.creates, check.DeepEquals, []arvados.InstanceType{test.InstanceType(2), test.InstanceType(1)})
 299         c.Check(pool.starts, check.DeepEquals, []string{uuids[6], uuids[5], uuids[3], uuids[2]})
 300         running := map[string]bool{}
 301         for uuid, t := range pool.running {
 302                 if t.IsZero() {
 303                         running[uuid] = false
 304                 } else {
 305                         running[uuid] = true
 306                 }
 307         }
 308         c.Check(running, check.DeepEquals, map[string]bool{uuids[3]: false, uuids[6]: false})
 309 }