lib/lsf/dispatch_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package lsf
   6
   7 import (
   8         "context"
   9         "encoding/json"
  10         "fmt"
  11         "math/rand"
  12         "os/exec"
  13         "strconv"
  14         "sync"
  15         "testing"
  16         "time"
  17
  18         "git.arvados.org/arvados.git/lib/config"
  19         "git.arvados.org/arvados.git/sdk/go/arvados"
  20         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  21         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  22         "github.com/prometheus/client_golang/prometheus"
  23         "gopkg.in/check.v1"
  24 )
  25
  26 func Test(t *testing.T) {
  27         check.TestingT(t)
  28 }
  29
  30 var _ = check.Suite(&suite{})
  31
  32 type suite struct {
  33         disp          *dispatcher
  34         crTooBig      arvados.ContainerRequest
  35         crPending     arvados.ContainerRequest
  36         crCUDARequest arvados.ContainerRequest
  37 }
  38
  39 func (s *suite) TearDownTest(c *check.C) {
  40         arvados.NewClientFromEnv().RequestAndDecode(nil, "POST", "database/reset", nil, nil)
  41 }
  42
  43 func (s *suite) SetUpTest(c *check.C) {
  44         cfg, err := config.NewLoader(nil, ctxlog.TestLogger(c)).Load()
  45         c.Assert(err, check.IsNil)
  46         cluster, err := cfg.GetCluster("")
  47         c.Assert(err, check.IsNil)
  48         cluster.Containers.ReserveExtraRAM = 256 << 20
  49         cluster.Containers.CloudVMs.PollInterval = arvados.Duration(time.Second / 4)
  50         cluster.Containers.MinRetryPeriod = arvados.Duration(time.Second / 4)
  51         cluster.InstanceTypes = arvados.InstanceTypeMap{
  52                 "biggest_available_node": arvados.InstanceType{
  53                         RAM:             100 << 30, // 100 GiB
  54                         VCPUs:           4,
  55                         IncludedScratch: 100 << 30,
  56                         Scratch:         100 << 30,
  57                 }}
  58         s.disp = newHandler(context.Background(), cluster, arvadostest.Dispatch1Token, prometheus.NewRegistry()).(*dispatcher)
  59         s.disp.lsfcli.stubCommand = func(string, ...string) *exec.Cmd {
  60                 return exec.Command("bash", "-c", "echo >&2 unimplemented stub; false")
  61         }
  62         err = arvados.NewClientFromEnv().RequestAndDecode(&s.crTooBig, "POST", "arvados/v1/container_requests", nil, map[string]interface{}{
  63                 "container_request": map[string]interface{}{
  64                         "runtime_constraints": arvados.RuntimeConstraints{
  65                                 RAM:   1000000000000,
  66                                 VCPUs: 1,
  67                         },
  68                         "container_image":     arvadostest.DockerImage112PDH,
  69                         "command":             []string{"sleep", "1"},
  70                         "mounts":              map[string]arvados.Mount{"/mnt/out": {Kind: "tmp", Capacity: 1000}},
  71                         "output_path":         "/mnt/out",
  72                         "state":               arvados.ContainerRequestStateCommitted,
  73                         "priority":            1,
  74                         "container_count_max": 1,
  75                 },
  76         })
  77         c.Assert(err, check.IsNil)
  78
  79         err = arvados.NewClientFromEnv().RequestAndDecode(&s.crPending, "POST", "arvados/v1/container_requests", nil, map[string]interface{}{
  80                 "container_request": map[string]interface{}{
  81                         "runtime_constraints": arvados.RuntimeConstraints{
  82                                 RAM:           100000000,
  83                                 VCPUs:         2,
  84                                 KeepCacheDisk: 8 << 30,
  85                         },
  86                         "container_image":     arvadostest.DockerImage112PDH,
  87                         "command":             []string{"sleep", "1"},
  88                         "mounts":              map[string]arvados.Mount{"/mnt/out": {Kind: "tmp", Capacity: 1000}},
  89                         "output_path":         "/mnt/out",
  90                         "state":               arvados.ContainerRequestStateCommitted,
  91                         "priority":            1,
  92                         "container_count_max": 1,
  93                 },
  94         })
  95         c.Assert(err, check.IsNil)
  96
  97         err = arvados.NewClientFromEnv().RequestAndDecode(&s.crCUDARequest, "POST", "arvados/v1/container_requests", nil, map[string]interface{}{
  98                 "container_request": map[string]interface{}{
  99                         "runtime_constraints": arvados.RuntimeConstraints{
 100                                 RAM:   16000000,
 101                                 VCPUs: 1,
 102                                 CUDA: arvados.CUDARuntimeConstraints{
 103                                         DeviceCount:        1,
 104                                         DriverVersion:      "11.0",
 105                                         HardwareCapability: "8.0",
 106                                 },
 107                         },
 108                         "container_image":     arvadostest.DockerImage112PDH,
 109                         "command":             []string{"sleep", "1"},
 110                         "mounts":              map[string]arvados.Mount{"/mnt/out": {Kind: "tmp", Capacity: 1000}},
 111                         "output_path":         "/mnt/out",
 112                         "state":               arvados.ContainerRequestStateCommitted,
 113                         "priority":            1,
 114                         "container_count_max": 1,
 115                 },
 116         })
 117         c.Assert(err, check.IsNil)
 118
 119 }
 120
 121 type lsfstub struct {
 122         sudoUser  string
 123         errorRate float64
 124 }
 125
 126 func (stub lsfstub) stubCommand(s *suite, c *check.C) func(prog string, args ...string) *exec.Cmd {
 127         mtx := sync.Mutex{}
 128         nextjobid := 100
 129         fakejobq := map[int]string{}
 130         return func(prog string, args ...string) *exec.Cmd {
 131                 c.Logf("stubCommand: %q %q", prog, args)
 132                 if rand.Float64() < stub.errorRate {
 133                         return exec.Command("bash", "-c", "echo >&2 'stub random failure' && false")
 134                 }
 135                 if stub.sudoUser != "" && len(args) > 3 &&
 136                         prog == "sudo" &&
 137                         args[0] == "-E" &&
 138                         args[1] == "-u" &&
 139                         args[2] == stub.sudoUser {
 140                         prog, args = args[3], args[4:]
 141                 }
 142                 switch prog {
 143                 case "bsub":
 144                         defaultArgs := s.disp.Cluster.Containers.LSF.BsubArgumentsList
 145                         if args[5] == s.crCUDARequest.ContainerUUID {
 146                                 c.Assert(len(args), check.Equals, len(defaultArgs)+len(s.disp.Cluster.Containers.LSF.BsubCUDAArguments))
 147                         } else {
 148                                 c.Assert(len(args), check.Equals, len(defaultArgs))
 149                         }
 150                         // %%J must have been rewritten to %J
 151                         c.Check(args[1], check.Equals, "/tmp/crunch-run.%J.out")
 152                         args = args[4:]
 153                         switch args[1] {
 154                         case arvadostest.LockedContainerUUID:
 155                                 c.Check(args, check.DeepEquals, []string{
 156                                         "-J", arvadostest.LockedContainerUUID,
 157                                         "-n", "4",
 158                                         "-D", "11701MB",
 159                                         "-R", "rusage[mem=11701MB:tmp=0MB] span[hosts=1]",
 160                                         "-R", "select[mem>=11701MB]",
 161                                         "-R", "select[tmp>=0MB]",
 162                                         "-R", "select[ncpus>=4]"})
 163                                 mtx.Lock()
 164                                 fakejobq[nextjobid] = args[1]
 165                                 nextjobid++
 166                                 mtx.Unlock()
 167                         case arvadostest.QueuedContainerUUID:
 168                                 c.Check(args, check.DeepEquals, []string{
 169                                         "-J", arvadostest.QueuedContainerUUID,
 170                                         "-n", "4",
 171                                         "-D", "11701MB",
 172                                         "-R", "rusage[mem=11701MB:tmp=45777MB] span[hosts=1]",
 173                                         "-R", "select[mem>=11701MB]",
 174                                         "-R", "select[tmp>=45777MB]",
 175                                         "-R", "select[ncpus>=4]"})
 176                                 mtx.Lock()
 177                                 fakejobq[nextjobid] = args[1]
 178                                 nextjobid++
 179                                 mtx.Unlock()
 180                         case s.crPending.ContainerUUID:
 181                                 c.Check(args, check.DeepEquals, []string{
 182                                         "-J", s.crPending.ContainerUUID,
 183                                         "-n", "2",
 184                                         "-D", "352MB",
 185                                         "-R", "rusage[mem=352MB:tmp=8448MB] span[hosts=1]",
 186                                         "-R", "select[mem>=352MB]",
 187                                         "-R", "select[tmp>=8448MB]",
 188                                         "-R", "select[ncpus>=2]"})
 189                                 mtx.Lock()
 190                                 fakejobq[nextjobid] = args[1]
 191                                 nextjobid++
 192                                 mtx.Unlock()
 193                         case s.crCUDARequest.ContainerUUID:
 194                                 c.Check(args, check.DeepEquals, []string{
 195                                         "-J", s.crCUDARequest.ContainerUUID,
 196                                         "-n", "1",
 197                                         "-D", "528MB",
 198                                         "-R", "rusage[mem=528MB:tmp=256MB] span[hosts=1]",
 199                                         "-R", "select[mem>=528MB]",
 200                                         "-R", "select[tmp>=256MB]",
 201                                         "-R", "select[ncpus>=1]",
 202                                         "-gpu", "num=1"})
 203                                 mtx.Lock()
 204                                 fakejobq[nextjobid] = args[1]
 205                                 nextjobid++
 206                                 mtx.Unlock()
 207                         default:
 208                                 c.Errorf("unexpected uuid passed to bsub: args %q", args)
 209                                 return exec.Command("false")
 210                         }
 211                         return exec.Command("echo", "submitted job")
 212                 case "bjobs":
 213                         c.Check(args, check.DeepEquals, []string{"-u", "all", "-o", "jobid stat job_name pend_reason", "-json"})
 214                         var records []map[string]interface{}
 215                         for jobid, uuid := range fakejobq {
 216                                 stat, reason := "RUN", ""
 217                                 if uuid == s.crPending.ContainerUUID {
 218                                         // The real bjobs output includes a trailing ';' here:
 219                                         stat, reason = "PEND", "There are no suitable hosts for the job;"
 220                                 }
 221                                 records = append(records, map[string]interface{}{
 222                                         "JOBID":       fmt.Sprintf("%d", jobid),
 223                                         "STAT":        stat,
 224                                         "JOB_NAME":    uuid,
 225                                         "PEND_REASON": reason,
 226                                 })
 227                         }
 228                         out, err := json.Marshal(map[string]interface{}{
 229                                 "COMMAND": "bjobs",
 230                                 "JOBS":    len(fakejobq),
 231                                 "RECORDS": records,
 232                         })
 233                         if err != nil {
 234                                 panic(err)
 235                         }
 236                         c.Logf("bjobs out: %s", out)
 237                         return exec.Command("printf", string(out))
 238                 case "bkill":
 239                         killid, _ := strconv.Atoi(args[0])
 240                         if uuid, ok := fakejobq[killid]; !ok {
 241                                 return exec.Command("bash", "-c", fmt.Sprintf("printf >&2 'Job <%d>: No matching job found\n'", killid))
 242                         } else if uuid == "" {
 243                                 return exec.Command("bash", "-c", fmt.Sprintf("printf >&2 'Job <%d>: Job has already finished\n'", killid))
 244                         } else {
 245                                 go func() {
 246                                         time.Sleep(time.Millisecond)
 247                                         mtx.Lock()
 248                                         delete(fakejobq, killid)
 249                                         mtx.Unlock()
 250                                 }()
 251                                 return exec.Command("bash", "-c", fmt.Sprintf("printf 'Job <%d> is being terminated\n'", killid))
 252                         }
 253                 default:
 254                         return exec.Command("bash", "-c", fmt.Sprintf("echo >&2 'stub: command not found: %+q'", prog))
 255                 }
 256         }
 257 }
 258
 259 func (s *suite) TestSubmit(c *check.C) {
 260         s.disp.lsfcli.stubCommand = lsfstub{
 261                 errorRate: 0.1,
 262                 sudoUser:  s.disp.Cluster.Containers.LSF.BsubSudoUser,
 263         }.stubCommand(s, c)
 264         s.disp.Start()
 265
 266         deadline := time.Now().Add(20 * time.Second)
 267         for range time.NewTicker(time.Second).C {
 268                 if time.Now().After(deadline) {
 269                         c.Error("timed out")
 270                         break
 271                 }
 272                 // "crTooBig" should never be submitted to lsf because
 273                 // it is bigger than any configured instance type
 274                 if ent, ok := s.disp.lsfqueue.Lookup(s.crTooBig.ContainerUUID); ok {
 275                         c.Errorf("Lookup(crTooBig) == true, ent = %#v", ent)
 276                         break
 277                 }
 278                 // "queuedcontainer" should be running
 279                 if _, ok := s.disp.lsfqueue.Lookup(arvadostest.QueuedContainerUUID); !ok {
 280                         c.Log("Lookup(queuedcontainer) == false")
 281                         continue
 282                 }
 283                 // "crPending" should be pending
 284                 if ent, ok := s.disp.lsfqueue.Lookup(s.crPending.ContainerUUID); !ok {
 285                         c.Logf("Lookup(crPending) == false", ent)
 286                         continue
 287                 }
 288                 // "lockedcontainer" should be cancelled because it
 289                 // has priority 0 (no matching container requests)
 290                 if ent, ok := s.disp.lsfqueue.Lookup(arvadostest.LockedContainerUUID); ok {
 291                         c.Logf("Lookup(lockedcontainer) == true, ent = %#v", ent)
 292                         continue
 293                 }
 294                 var ctr arvados.Container
 295                 if err := s.disp.arvDispatcher.Arv.Get("containers", arvadostest.LockedContainerUUID, nil, &ctr); err != nil {
 296                         c.Logf("error getting container state for %s: %s", arvadostest.LockedContainerUUID, err)
 297                         continue
 298                 } else if ctr.State != arvados.ContainerStateQueued {
 299                         c.Logf("LockedContainer is not in the LSF queue but its arvados record has not been updated to state==Queued (state is %q)", ctr.State)
 300                         continue
 301                 }
 302
 303                 if err := s.disp.arvDispatcher.Arv.Get("containers", s.crTooBig.ContainerUUID, nil, &ctr); err != nil {
 304                         c.Logf("error getting container state for %s: %s", s.crTooBig.ContainerUUID, err)
 305                         continue
 306                 } else if ctr.State != arvados.ContainerStateCancelled {
 307                         c.Logf("container %s is not in the LSF queue but its arvados record has not been updated to state==Cancelled (state is %q)", s.crTooBig.ContainerUUID, ctr.State)
 308                         continue
 309                 } else {
 310                         c.Check(ctr.RuntimeStatus["error"], check.Equals, "constraints not satisfiable by any configured instance type")
 311                 }
 312                 c.Log("reached desired state")
 313                 break
 314         }
 315 }