lib/crunchrun/integration_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package crunchrun
   6
   7 import (
   8         "bytes"
   9         "encoding/json"
  10         "fmt"
  11         "io"
  12         "io/ioutil"
  13         "os"
  14         "os/exec"
  15         "strings"
  16
  17         "git.arvados.org/arvados.git/lib/config"
  18         "git.arvados.org/arvados.git/sdk/go/arvados"
  19         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  20         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  21         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  22         "git.arvados.org/arvados.git/sdk/go/keepclient"
  23         . "gopkg.in/check.v1"
  24 )
  25
  26 var _ = Suite(&integrationSuite{})
  27
  28 type integrationSuite struct {
  29         engine string
  30         image  arvados.Collection
  31         input  arvados.Collection
  32         stdin  bytes.Buffer
  33         stdout bytes.Buffer
  34         stderr bytes.Buffer
  35         args   []string
  36         cr     arvados.ContainerRequest
  37         client *arvados.Client
  38         ac     *arvadosclient.ArvadosClient
  39         kc     *keepclient.KeepClient
  40
  41         logCollection    arvados.Collection
  42         outputCollection arvados.Collection
  43         logFiles         map[string]string // filename => contents
  44 }
  45
  46 func (s *integrationSuite) SetUpSuite(c *C) {
  47         _, err := exec.LookPath("docker")
  48         if err != nil {
  49                 c.Skip("looks like docker is not installed")
  50         }
  51
  52         arvadostest.StartKeep(2, true)
  53
  54         out, err := exec.Command("docker", "load", "--input", arvadostest.BusyboxDockerImage(c)).CombinedOutput()
  55         c.Log(string(out))
  56         c.Assert(err, IsNil)
  57         out, err = exec.Command("arv-keepdocker", "--no-resume", "busybox:uclibc").Output()
  58         imageUUID := strings.TrimSpace(string(out))
  59         c.Logf("image uuid %s", imageUUID)
  60         if !c.Check(err, IsNil) {
  61                 if err, ok := err.(*exec.ExitError); ok {
  62                         c.Logf("%s", err.Stderr)
  63                 }
  64                 c.Fail()
  65         }
  66         err = arvados.NewClientFromEnv().RequestAndDecode(&s.image, "GET", "arvados/v1/collections/"+imageUUID, nil, nil)
  67         c.Assert(err, IsNil)
  68         c.Logf("image pdh %s", s.image.PortableDataHash)
  69
  70         s.client = arvados.NewClientFromEnv()
  71         s.ac, err = arvadosclient.New(s.client)
  72         c.Assert(err, IsNil)
  73         s.kc = keepclient.New(s.ac)
  74         fs, err := s.input.FileSystem(s.client, s.kc)
  75         c.Assert(err, IsNil)
  76         f, err := fs.OpenFile("inputfile", os.O_CREATE|os.O_WRONLY, 0755)
  77         c.Assert(err, IsNil)
  78         _, err = f.Write([]byte("inputdata"))
  79         c.Assert(err, IsNil)
  80         err = f.Close()
  81         c.Assert(err, IsNil)
  82         s.input.ManifestText, err = fs.MarshalManifest(".")
  83         c.Assert(err, IsNil)
  84         err = s.client.RequestAndDecode(&s.input, "POST", "arvados/v1/collections", nil, map[string]interface{}{
  85                 "ensure_unique_name": true,
  86                 "collection": map[string]interface{}{
  87                         "manifest_text": s.input.ManifestText,
  88                 },
  89         })
  90         c.Assert(err, IsNil)
  91         c.Logf("input pdh %s", s.input.PortableDataHash)
  92 }
  93
  94 func (s *integrationSuite) TearDownSuite(c *C) {
  95         os.Unsetenv("ARVADOS_KEEP_SERVICES")
  96         if s.client == nil {
  97                 // didn't set up
  98                 return
  99         }
 100         err := s.client.RequestAndDecode(nil, "POST", "database/reset", nil, nil)
 101         c.Check(err, IsNil)
 102 }
 103
 104 func (s *integrationSuite) SetUpTest(c *C) {
 105         os.Unsetenv("ARVADOS_KEEP_SERVICES")
 106         s.engine = "docker"
 107         s.args = nil
 108         s.stdin = bytes.Buffer{}
 109         s.stdout = bytes.Buffer{}
 110         s.stderr = bytes.Buffer{}
 111         s.logCollection = arvados.Collection{}
 112         s.outputCollection = arvados.Collection{}
 113         s.logFiles = map[string]string{}
 114         s.cr = arvados.ContainerRequest{
 115                 Priority:       1,
 116                 State:          "Committed",
 117                 OutputPath:     "/mnt/out",
 118                 ContainerImage: s.image.PortableDataHash,
 119                 Mounts: map[string]arvados.Mount{
 120                         "/mnt/json": {
 121                                 Kind: "json",
 122                                 Content: []interface{}{
 123                                         "foo",
 124                                         map[string]string{"foo": "bar"},
 125                                         nil,
 126                                 },
 127                         },
 128                         "/mnt/in": {
 129                                 Kind:             "collection",
 130                                 PortableDataHash: s.input.PortableDataHash,
 131                         },
 132                         "/mnt/out": {
 133                                 Kind:     "tmp",
 134                                 Capacity: 1000,
 135                         },
 136                 },
 137                 RuntimeConstraints: arvados.RuntimeConstraints{
 138                         RAM:   128000000,
 139                         VCPUs: 1,
 140                         API:   true,
 141                 },
 142         }
 143 }
 144
 145 func (s *integrationSuite) setup(c *C) {
 146         err := s.client.RequestAndDecode(&s.cr, "POST", "arvados/v1/container_requests", nil, map[string]interface{}{"container_request": map[string]interface{}{
 147                 "priority":            s.cr.Priority,
 148                 "state":               s.cr.State,
 149                 "command":             s.cr.Command,
 150                 "output_path":         s.cr.OutputPath,
 151                 "output_glob":         s.cr.OutputGlob,
 152                 "container_image":     s.cr.ContainerImage,
 153                 "mounts":              s.cr.Mounts,
 154                 "runtime_constraints": s.cr.RuntimeConstraints,
 155                 "use_existing":        false,
 156         }})
 157         c.Assert(err, IsNil)
 158         c.Assert(s.cr.ContainerUUID, Not(Equals), "")
 159         err = s.client.RequestAndDecode(nil, "POST", "arvados/v1/containers/"+s.cr.ContainerUUID+"/lock", nil, nil)
 160         c.Assert(err, IsNil)
 161 }
 162
 163 func (s *integrationSuite) TestRunTrivialContainerWithDocker(c *C) {
 164         s.engine = "docker"
 165         s.testRunTrivialContainer(c)
 166         c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*Using container runtime: docker Engine \d+\.\d+.*`)
 167 }
 168
 169 func (s *integrationSuite) TestRunTrivialContainerWithSingularity(c *C) {
 170         s.engine = "singularity"
 171         s.testRunTrivialContainer(c)
 172         c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*Using container runtime: singularity.* version 3\.\d+.*`)
 173 }
 174
 175 func (s *integrationSuite) TestRunTrivialContainerWithLocalKeepstore(c *C) {
 176         for _, trial := range []struct {
 177                 logConfig           string
 178                 matchGetReq         Checker
 179                 matchPutReq         Checker
 180                 matchStartupMessage Checker
 181         }{
 182                 {"none", Not(Matches), Not(Matches), Not(Matches)},
 183                 {"all", Matches, Matches, Matches},
 184                 {"errors", Not(Matches), Not(Matches), Matches},
 185         } {
 186                 c.Logf("=== testing with Containers.LocalKeepLogsToContainerLog: %q", trial.logConfig)
 187                 s.SetUpTest(c)
 188
 189                 cfg, err := config.NewLoader(nil, ctxlog.TestLogger(c)).Load()
 190                 c.Assert(err, IsNil)
 191                 cluster, err := cfg.GetCluster("")
 192                 c.Assert(err, IsNil)
 193                 for uuid, volume := range cluster.Volumes {
 194                         volume.AccessViaHosts = nil
 195                         volume.Replication = 2
 196                         cluster.Volumes[uuid] = volume
 197
 198                         var v struct {
 199                                 Root string
 200                         }
 201                         err = json.Unmarshal(volume.DriverParameters, &v)
 202                         c.Assert(err, IsNil)
 203                         err = os.Mkdir(v.Root, 0777)
 204                         if !os.IsExist(err) {
 205                                 c.Assert(err, IsNil)
 206                         }
 207                 }
 208                 cluster.Containers.LocalKeepLogsToContainerLog = trial.logConfig
 209
 210                 s.stdin.Reset()
 211                 err = json.NewEncoder(&s.stdin).Encode(ConfigData{
 212                         Env:         nil,
 213                         KeepBuffers: 1,
 214                         Cluster:     cluster,
 215                 })
 216                 c.Assert(err, IsNil)
 217
 218                 s.engine = "docker"
 219                 s.testRunTrivialContainer(c)
 220
 221                 log, logExists := s.logFiles["keepstore.txt"]
 222                 if trial.logConfig == "none" {
 223                         c.Check(logExists, Equals, false)
 224                 } else {
 225                         c.Check(log, Matches, `(?ms).*not running trash worker.*`)
 226                         c.Check(log, Matches, `(?ms).*not running trash emptier.*`)
 227                         c.Check(log, trial.matchGetReq, `(?ms).*"reqMethod":"GET".*`)
 228                         c.Check(log, trial.matchPutReq, `(?ms).*"reqMethod":"PUT".*,"reqPath":"0e3bcff26d51c895a60ea0d4585e134d".*`)
 229                 }
 230
 231                 c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*using local keepstore process .* at http://[\d\.]{7,}:\d+.*`)
 232                 c.Check(s.logFiles["crunch-run.txt"], Not(Matches), `(?ms).* at http://127\..*`)
 233                 c.Check(s.logFiles["crunch-run.txt"], Not(Matches), `(?ms).* at http://169\.254\..*`)
 234                 c.Check(s.logFiles["stderr.txt"], Matches, `(?ms).*ARVADOS_KEEP_SERVICES=http://[\d\.]{7,}:\d+\n.*`)
 235         }
 236 }
 237
 238 func (s *integrationSuite) TestRunTrivialContainerWithNoLocalKeepstore(c *C) {
 239         // Check that (1) config is loaded from $ARVADOS_CONFIG when
 240         // not provided on stdin and (2) if a local keepstore is not
 241         // started, crunch-run.txt explains why not.
 242         s.SetUpTest(c)
 243         s.stdin.Reset()
 244         s.testRunTrivialContainer(c)
 245         c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*not starting a local keepstore process because KeepBuffers=0 in config\n.*`)
 246
 247         s.SetUpTest(c)
 248         s.args = []string{"-config", c.MkDir() + "/config.yaml"}
 249         s.stdin.Reset()
 250         buf, err := ioutil.ReadFile(os.Getenv("ARVADOS_CONFIG"))
 251         c.Assert(err, IsNil)
 252         err = ioutil.WriteFile(s.args[1], bytes.Replace(buf, []byte("LocalKeepBlobBuffersPerVCPU: 0"), []byte("LocalKeepBlobBuffersPerVCPU: 1"), -1), 0666)
 253         c.Assert(err, IsNil)
 254         s.testRunTrivialContainer(c)
 255         c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*not starting a local keepstore process because a volume \(zzzzz-nyw5e-00000000000000\d\) uses AccessViaHosts\n.*`)
 256
 257         // Check that config read errors are logged
 258         s.SetUpTest(c)
 259         s.args = []string{"-config", c.MkDir() + "/config-error.yaml"}
 260         s.stdin.Reset()
 261         s.testRunTrivialContainer(c)
 262         c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*could not load config file \Q`+s.args[1]+`\E:.* no such file or directory\n.*`)
 263
 264         s.SetUpTest(c)
 265         s.args = []string{"-config", c.MkDir() + "/config-unreadable.yaml"}
 266         s.stdin.Reset()
 267         err = ioutil.WriteFile(s.args[1], []byte{}, 0)
 268         c.Check(err, IsNil)
 269         s.testRunTrivialContainer(c)
 270         c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*could not load config file \Q`+s.args[1]+`\E:.* permission denied\n.*`)
 271
 272         s.SetUpTest(c)
 273         s.stdin.Reset()
 274         s.testRunTrivialContainer(c)
 275         c.Check(s.logFiles["crunch-run.txt"], Matches, `(?ms).*loaded config file \Q`+os.Getenv("ARVADOS_CONFIG")+`\E\n.*`)
 276 }
 277
 278 func (s *integrationSuite) TestRunTrivialContainerWithOutputGlob(c *C) {
 279         s.cr.OutputGlob = []string{"js?n"}
 280         s.testRunTrivialContainer(c)
 281         fs, err := s.outputCollection.FileSystem(s.client, s.kc)
 282         c.Assert(err, IsNil)
 283         _, err = fs.Stat("json")
 284         c.Check(err, IsNil)
 285         _, err = fs.Stat("inputfile")
 286         c.Check(err, Equals, os.ErrNotExist)
 287         _, err = fs.Stat("emptydir")
 288         c.Check(err, Equals, os.ErrNotExist)
 289 }
 290
 291 func (s *integrationSuite) testRunTrivialContainer(c *C) {
 292         if err := exec.Command("which", s.engine).Run(); err != nil {
 293                 c.Skip(fmt.Sprintf("%s: %s", s.engine, err))
 294         }
 295         s.cr.Command = []string{"sh", "-c", "env >&2 && cat /mnt/in/inputfile >/mnt/out/inputfile && cat /mnt/json >/mnt/out/json && ! touch /mnt/in/shouldbereadonly && mkdir /mnt/out/emptydir"}
 296         s.setup(c)
 297
 298         args := []string{
 299                 "-runtime-engine=" + s.engine,
 300                 "-enable-memory-limit=false",
 301         }
 302         if s.stdin.Len() > 0 {
 303                 args = append(args, "-stdin-config=true")
 304         }
 305         args = append(args, s.args...)
 306         args = append(args, s.cr.ContainerUUID)
 307         code := command{}.RunCommand("crunch-run", args, &s.stdin, io.MultiWriter(&s.stdout, os.Stderr), io.MultiWriter(&s.stderr, os.Stderr))
 308         c.Logf("\n===== stdout =====\n%s", s.stdout.String())
 309         c.Logf("\n===== stderr =====\n%s", s.stderr.String())
 310         c.Check(code, Equals, 0)
 311         err := s.client.RequestAndDecode(&s.cr, "GET", "arvados/v1/container_requests/"+s.cr.UUID, nil, nil)
 312         c.Assert(err, IsNil)
 313         c.Logf("Finished container request: %#v", s.cr)
 314
 315         var log arvados.Collection
 316         err = s.client.RequestAndDecode(&log, "GET", "arvados/v1/collections/"+s.cr.LogUUID, nil, nil)
 317         c.Assert(err, IsNil)
 318         fs, err := log.FileSystem(s.client, s.kc)
 319         c.Assert(err, IsNil)
 320         if d, err := fs.Open("/"); c.Check(err, IsNil) {
 321                 fis, err := d.Readdir(-1)
 322                 c.Assert(err, IsNil)
 323                 for _, fi := range fis {
 324                         if fi.IsDir() {
 325                                 continue
 326                         }
 327                         f, err := fs.Open(fi.Name())
 328                         c.Assert(err, IsNil)
 329                         buf, err := ioutil.ReadAll(f)
 330                         c.Assert(err, IsNil)
 331                         c.Logf("\n===== %s =====\n%s", fi.Name(), buf)
 332                         s.logFiles[fi.Name()] = string(buf)
 333                 }
 334         }
 335         s.logCollection = log
 336
 337         var output arvados.Collection
 338         err = s.client.RequestAndDecode(&output, "GET", "arvados/v1/collections/"+s.cr.OutputUUID, nil, nil)
 339         c.Assert(err, IsNil)
 340         s.outputCollection = output
 341
 342         if len(s.cr.OutputGlob) == 0 {
 343                 fs, err = output.FileSystem(s.client, s.kc)
 344                 c.Assert(err, IsNil)
 345                 if f, err := fs.Open("inputfile"); c.Check(err, IsNil) {
 346                         defer f.Close()
 347                         buf, err := ioutil.ReadAll(f)
 348                         c.Check(err, IsNil)
 349                         c.Check(string(buf), Equals, "inputdata")
 350                 }
 351                 if f, err := fs.Open("json"); c.Check(err, IsNil) {
 352                         defer f.Close()
 353                         buf, err := ioutil.ReadAll(f)
 354                         c.Check(err, IsNil)
 355                         c.Check(string(buf), Equals, `["foo",{"foo":"bar"},null]`)
 356                 }
 357                 if fi, err := fs.Stat("emptydir"); c.Check(err, IsNil) {
 358                         c.Check(fi.IsDir(), Equals, true)
 359                 }
 360                 if d, err := fs.Open("emptydir"); c.Check(err, IsNil) {
 361                         defer d.Close()
 362                         fis, err := d.Readdir(-1)
 363                         c.Assert(err, IsNil)
 364                         // crunch-run still saves a ".keep" file to preserve
 365                         // empty dirs even though that shouldn't be
 366                         // necessary. Ideally we would do:
 367                         // c.Check(fis, HasLen, 0)
 368                         for _, fi := range fis {
 369                                 c.Check(fi.Name(), Equals, ".keep")
 370                         }
 371                 }
 372         }
 373 }