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.ContainerStateQueued,
 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.ContainerStateQueued,
 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.ContainerStateQueued,
 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.ContainerStateQueued,
 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: 1,
 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), test.ContainerUUID(3)})
 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                 shouldCreate := []arvados.InstanceType{}
 173                 for i := 1; i < 1+quota; i++ {
 174                         shouldCreate = append(shouldCreate, test.InstanceType(i))
 175                 }
 176                 queue := test.Queue{
 177                         ChooseType: func(ctr *arvados.Container) (arvados.InstanceType, error) {
 178                                 return test.InstanceType(ctr.RuntimeConstraints.VCPUs), nil
 179                         },
 180                         Containers: []arvados.Container{
 181                                 {
 182                                         UUID:     test.ContainerUUID(1),
 183                                         Priority: 1,
 184                                         State:    arvados.ContainerStateQueued,
 185                                         RuntimeConstraints: arvados.RuntimeConstraints{
 186                                                 VCPUs: 1,
 187                                                 RAM:   1 << 30,
 188                                         },
 189                                 },
 190                         },
 191                 }
 192                 queue.Update()
 193                 pool := stubPool{
 194                         atQuota: quota == 0,
 195                         unalloc: map[arvados.InstanceType]int{
 196                                 test.InstanceType(2): 2,
 197                         },
 198                         idle: map[arvados.InstanceType]int{
 199                                 test.InstanceType(2): 2,
 200                         },
 201                         running:   map[string]time.Time{},
 202                         creates:   []arvados.InstanceType{},
 203                         starts:    []string{},
 204                         canCreate: 0,
 205                 }
 206                 New(logger, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 207                 c.Check(pool.creates, check.DeepEquals, shouldCreate)
 208                 c.Check(pool.starts, check.DeepEquals, []string{})
 209                 c.Check(pool.shutdowns, check.Not(check.Equals), 0)
 210         }
 211 }
 212
 213 // Start lower-priority containers while waiting for new/existing
 214 // workers to come up for higher-priority containers.
 215 func (*SchedulerSuite) TestStartWhileCreating(c *check.C) {
 216         pool := stubPool{
 217                 unalloc: map[arvados.InstanceType]int{
 218                         test.InstanceType(1): 1,
 219                         test.InstanceType(2): 1,
 220                 },
 221                 idle: map[arvados.InstanceType]int{
 222                         test.InstanceType(1): 1,
 223                         test.InstanceType(2): 1,
 224                 },
 225                 running:   map[string]time.Time{},
 226                 canCreate: 2,
 227         }
 228         queue := test.Queue{
 229                 ChooseType: func(ctr *arvados.Container) (arvados.InstanceType, error) {
 230                         return test.InstanceType(ctr.RuntimeConstraints.VCPUs), nil
 231                 },
 232                 Containers: []arvados.Container{
 233                         {
 234                                 // create a new worker
 235                                 UUID:     test.ContainerUUID(1),
 236                                 Priority: 1,
 237                                 State:    arvados.ContainerStateQueued,
 238                                 RuntimeConstraints: arvados.RuntimeConstraints{
 239                                         VCPUs: 1,
 240                                         RAM:   1 << 30,
 241                                 },
 242                         },
 243                         {
 244                                 // tentatively map to unalloc worker
 245                                 UUID:     test.ContainerUUID(2),
 246                                 Priority: 2,
 247                                 State:    arvados.ContainerStateQueued,
 248                                 RuntimeConstraints: arvados.RuntimeConstraints{
 249                                         VCPUs: 1,
 250                                         RAM:   1 << 30,
 251                                 },
 252                         },
 253                         {
 254                                 // start now on idle worker
 255                                 UUID:     test.ContainerUUID(3),
 256                                 Priority: 3,
 257                                 State:    arvados.ContainerStateQueued,
 258                                 RuntimeConstraints: arvados.RuntimeConstraints{
 259                                         VCPUs: 1,
 260                                         RAM:   1 << 30,
 261                                 },
 262                         },
 263                         {
 264                                 // create a new worker
 265                                 UUID:     test.ContainerUUID(4),
 266                                 Priority: 4,
 267                                 State:    arvados.ContainerStateQueued,
 268                                 RuntimeConstraints: arvados.RuntimeConstraints{
 269                                         VCPUs: 2,
 270                                         RAM:   2 << 30,
 271                                 },
 272                         },
 273                         {
 274                                 // tentatively map to unalloc worker
 275                                 UUID:     test.ContainerUUID(5),
 276                                 Priority: 5,
 277                                 State:    arvados.ContainerStateQueued,
 278                                 RuntimeConstraints: arvados.RuntimeConstraints{
 279                                         VCPUs: 2,
 280                                         RAM:   2 << 30,
 281                                 },
 282                         },
 283                         {
 284                                 // start now on idle worker
 285                                 UUID:     test.ContainerUUID(6),
 286                                 Priority: 6,
 287                                 State:    arvados.ContainerStateQueued,
 288                                 RuntimeConstraints: arvados.RuntimeConstraints{
 289                                         VCPUs: 2,
 290                                         RAM:   2 << 30,
 291                                 },
 292                         },
 293                 },
 294         }
 295         queue.Update()
 296         New(logger, &queue, &pool, time.Millisecond, time.Millisecond).runQueue()
 297         c.Check(pool.creates, check.DeepEquals, []arvados.InstanceType{test.InstanceType(2), test.InstanceType(1)})
 298         c.Check(pool.starts, check.DeepEquals, []string{uuids[6], uuids[5], uuids[3], uuids[2]})
 299         running := map[string]bool{}
 300         for uuid, t := range pool.running {
 301                 if t.IsZero() {
 302                         running[uuid] = false
 303                 } else {
 304                         running[uuid] = true
 305                 }
 306         }
 307         c.Check(running, check.DeepEquals, map[string]bool{uuids[3]: false, uuids[6]: false})
 308 }