lib/crunchrun/crunchrun.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         "context"
  10         "encoding/json"
  11         "errors"
  12         "flag"
  13         "fmt"
  14         "io"
  15         "io/ioutil"
  16         "log"
  17         "net"
  18         "net/http"
  19         "os"
  20         "os/exec"
  21         "os/signal"
  22         "path"
  23         "path/filepath"
  24         "regexp"
  25         "runtime"
  26         "runtime/pprof"
  27         "sort"
  28         "strings"
  29         "sync"
  30         "syscall"
  31         "time"
  32
  33         "git.arvados.org/arvados.git/lib/cmd"
  34         "git.arvados.org/arvados.git/lib/crunchstat"
  35         "git.arvados.org/arvados.git/sdk/go/arvados"
  36         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  37         "git.arvados.org/arvados.git/sdk/go/keepclient"
  38         "git.arvados.org/arvados.git/sdk/go/manifest"
  39 )
  40
  41 type command struct{}
  42
  43 var Command = command{}
  44
  45 // ConfigData contains environment variables and (when needed) cluster
  46 // configuration, passed from dispatchcloud to crunch-run on stdin.
  47 type ConfigData struct {
  48         Env         map[string]string
  49         KeepBuffers int
  50         Cluster     *arvados.Cluster
  51 }
  52
  53 // IArvadosClient is the minimal Arvados API methods used by crunch-run.
  54 type IArvadosClient interface {
  55         Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
  56         Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
  57         Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
  58         Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error
  59         CallRaw(method string, resourceType string, uuid string, action string, parameters arvadosclient.Dict) (reader io.ReadCloser, err error)
  60         Discovery(key string) (interface{}, error)
  61 }
  62
  63 // ErrCancelled is the error returned when the container is cancelled.
  64 var ErrCancelled = errors.New("Cancelled")
  65
  66 // IKeepClient is the minimal Keep API methods used by crunch-run.
  67 type IKeepClient interface {
  68         BlockWrite(context.Context, arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error)
  69         ReadAt(locator string, p []byte, off int) (int, error)
  70         ManifestFileReader(m manifest.Manifest, filename string) (arvados.File, error)
  71         LocalLocator(locator string) (string, error)
  72         ClearBlockCache()
  73         SetStorageClasses(sc []string)
  74 }
  75
  76 // NewLogWriter is a factory function to create a new log writer.
  77 type NewLogWriter func(name string) (io.WriteCloser, error)
  78
  79 type RunArvMount func(cmdline []string, tok string) (*exec.Cmd, error)
  80
  81 type MkTempDir func(string, string) (string, error)
  82
  83 type PsProcess interface {
  84         CmdlineSlice() ([]string, error)
  85 }
  86
  87 // ContainerRunner is the main stateful struct used for a single execution of a
  88 // container.
  89 type ContainerRunner struct {
  90         executor       containerExecutor
  91         executorStdin  io.Closer
  92         executorStdout io.Closer
  93         executorStderr io.Closer
  94
  95         // Dispatcher client is initialized with the Dispatcher token.
  96         // This is a privileged token used to manage container status
  97         // and logs.
  98         //
  99         // We have both dispatcherClient and DispatcherArvClient
 100         // because there are two different incompatible Arvados Go
 101         // SDKs and we have to use both (hopefully this gets fixed in
 102         // #14467)
 103         dispatcherClient     *arvados.Client
 104         DispatcherArvClient  IArvadosClient
 105         DispatcherKeepClient IKeepClient
 106
 107         // Container client is initialized with the Container token
 108         // This token controls the permissions of the container, and
 109         // must be used for operations such as reading collections.
 110         //
 111         // Same comment as above applies to
 112         // containerClient/ContainerArvClient.
 113         containerClient     *arvados.Client
 114         ContainerArvClient  IArvadosClient
 115         ContainerKeepClient IKeepClient
 116
 117         Container     arvados.Container
 118         token         string
 119         ExitCode      *int
 120         NewLogWriter  NewLogWriter
 121         CrunchLog     *ThrottledLogger
 122         logUUID       string
 123         logMtx        sync.Mutex
 124         LogCollection arvados.CollectionFileSystem
 125         LogsPDH       *string
 126         RunArvMount   RunArvMount
 127         MkTempDir     MkTempDir
 128         ArvMount      *exec.Cmd
 129         ArvMountPoint string
 130         HostOutputDir string
 131         Volumes       map[string]struct{}
 132         OutputPDH     *string
 133         SigChan       chan os.Signal
 134         ArvMountExit  chan error
 135         SecretMounts  map[string]arvados.Mount
 136         MkArvClient   func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error)
 137         finalState    string
 138         parentTemp    string
 139
 140         statLogger       io.WriteCloser
 141         statReporter     *crunchstat.Reporter
 142         hoststatLogger   io.WriteCloser
 143         hoststatReporter *crunchstat.Reporter
 144         statInterval     time.Duration
 145         cgroupRoot       string
 146         // What we expect the container's cgroup parent to be.
 147         expectCgroupParent string
 148         // What we tell docker to use as the container's cgroup
 149         // parent. Note: Ideally we would use the same field for both
 150         // expectCgroupParent and setCgroupParent, and just make it
 151         // default to "docker". However, when using docker < 1.10 with
 152         // systemd, specifying a non-empty cgroup parent (even the
 153         // default value "docker") hits a docker bug
 154         // (https://github.com/docker/docker/issues/17126). Using two
 155         // separate fields makes it possible to use the "expect cgroup
 156         // parent to be X" feature even on sites where the "specify
 157         // cgroup parent" feature breaks.
 158         setCgroupParent string
 159
 160         cStateLock sync.Mutex
 161         cCancelled bool // StopContainer() invoked
 162
 163         enableMemoryLimit bool
 164         enableNetwork     string // one of "default" or "always"
 165         networkMode       string // "none", "host", or "" -- passed through to executor
 166         arvMountLog       *ThrottledLogger
 167
 168         containerWatchdogInterval time.Duration
 169
 170         gateway Gateway
 171 }
 172
 173 // setupSignals sets up signal handling to gracefully terminate the
 174 // underlying container and update state when receiving a TERM, INT or
 175 // QUIT signal.
 176 func (runner *ContainerRunner) setupSignals() {
 177         runner.SigChan = make(chan os.Signal, 1)
 178         signal.Notify(runner.SigChan, syscall.SIGTERM)
 179         signal.Notify(runner.SigChan, syscall.SIGINT)
 180         signal.Notify(runner.SigChan, syscall.SIGQUIT)
 181
 182         go func(sig chan os.Signal) {
 183                 for s := range sig {
 184                         runner.stop(s)
 185                 }
 186         }(runner.SigChan)
 187 }
 188
 189 // stop the underlying container.
 190 func (runner *ContainerRunner) stop(sig os.Signal) {
 191         runner.cStateLock.Lock()
 192         defer runner.cStateLock.Unlock()
 193         if sig != nil {
 194                 runner.CrunchLog.Printf("caught signal: %v", sig)
 195         }
 196         runner.cCancelled = true
 197         runner.CrunchLog.Printf("stopping container")
 198         err := runner.executor.Stop()
 199         if err != nil {
 200                 runner.CrunchLog.Printf("error stopping container: %s", err)
 201         }
 202 }
 203
 204 var errorBlacklist = []string{
 205         "(?ms).*[Cc]annot connect to the Docker daemon.*",
 206         "(?ms).*oci runtime error.*starting container process.*container init.*mounting.*to rootfs.*no such file or directory.*",
 207         "(?ms).*grpc: the connection is unavailable.*",
 208 }
 209 var brokenNodeHook *string = flag.String("broken-node-hook", "", "Script to run if node is detected to be broken (for example, Docker daemon is not running)")
 210
 211 func (runner *ContainerRunner) runBrokenNodeHook() {
 212         if *brokenNodeHook == "" {
 213                 path := filepath.Join(lockdir, brokenfile)
 214                 runner.CrunchLog.Printf("Writing %s to mark node as broken", path)
 215                 f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0700)
 216                 if err != nil {
 217                         runner.CrunchLog.Printf("Error writing %s: %s", path, err)
 218                         return
 219                 }
 220                 f.Close()
 221         } else {
 222                 runner.CrunchLog.Printf("Running broken node hook %q", *brokenNodeHook)
 223                 // run killme script
 224                 c := exec.Command(*brokenNodeHook)
 225                 c.Stdout = runner.CrunchLog
 226                 c.Stderr = runner.CrunchLog
 227                 err := c.Run()
 228                 if err != nil {
 229                         runner.CrunchLog.Printf("Error running broken node hook: %v", err)
 230                 }
 231         }
 232 }
 233
 234 func (runner *ContainerRunner) checkBrokenNode(goterr error) bool {
 235         for _, d := range errorBlacklist {
 236                 if m, e := regexp.MatchString(d, goterr.Error()); m && e == nil {
 237                         runner.CrunchLog.Printf("Error suggests node is unable to run containers: %v", goterr)
 238                         runner.runBrokenNodeHook()
 239                         return true
 240                 }
 241         }
 242         return false
 243 }
 244
 245 // LoadImage determines the docker image id from the container record and
 246 // checks if it is available in the local Docker image store.  If not, it loads
 247 // the image from Keep.
 248 func (runner *ContainerRunner) LoadImage() (string, error) {
 249         runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.Container.ContainerImage)
 250
 251         d, err := os.Open(runner.ArvMountPoint + "/by_id/" + runner.Container.ContainerImage)
 252         if err != nil {
 253                 return "", err
 254         }
 255         defer d.Close()
 256         allfiles, err := d.Readdirnames(-1)
 257         if err != nil {
 258                 return "", err
 259         }
 260         var tarfiles []string
 261         for _, fnm := range allfiles {
 262                 if strings.HasSuffix(fnm, ".tar") {
 263                         tarfiles = append(tarfiles, fnm)
 264                 }
 265         }
 266         if len(tarfiles) == 0 {
 267                 return "", fmt.Errorf("image collection does not include a .tar image file")
 268         }
 269         if len(tarfiles) > 1 {
 270                 return "", fmt.Errorf("cannot choose from multiple tar files in image collection: %v", tarfiles)
 271         }
 272         imageID := tarfiles[0][:len(tarfiles[0])-4]
 273         imageTarballPath := runner.ArvMountPoint + "/by_id/" + runner.Container.ContainerImage + "/" + imageID + ".tar"
 274         runner.CrunchLog.Printf("Using Docker image id %q", imageID)
 275
 276         runner.CrunchLog.Print("Loading Docker image from keep")
 277         err = runner.executor.LoadImage(imageID, imageTarballPath, runner.Container, runner.ArvMountPoint,
 278                 runner.containerClient)
 279         if err != nil {
 280                 return "", err
 281         }
 282
 283         return imageID, nil
 284 }
 285
 286 func (runner *ContainerRunner) ArvMountCmd(cmdline []string, token string) (c *exec.Cmd, err error) {
 287         c = exec.Command(cmdline[0], cmdline[1:]...)
 288
 289         // Copy our environment, but override ARVADOS_API_TOKEN with
 290         // the container auth token.
 291         c.Env = nil
 292         for _, s := range os.Environ() {
 293                 if !strings.HasPrefix(s, "ARVADOS_API_TOKEN=") {
 294                         c.Env = append(c.Env, s)
 295                 }
 296         }
 297         c.Env = append(c.Env, "ARVADOS_API_TOKEN="+token)
 298
 299         w, err := runner.NewLogWriter("arv-mount")
 300         if err != nil {
 301                 return nil, err
 302         }
 303         runner.arvMountLog = NewThrottledLogger(w)
 304         scanner := logScanner{
 305                 Patterns: []string{
 306                         "Keep write error",
 307                         "Block not found error",
 308                         "Unhandled exception during FUSE operation",
 309                 },
 310                 ReportFunc: runner.reportArvMountWarning,
 311         }
 312         c.Stdout = runner.arvMountLog
 313         c.Stderr = io.MultiWriter(runner.arvMountLog, os.Stderr, &scanner)
 314
 315         runner.CrunchLog.Printf("Running %v", c.Args)
 316
 317         err = c.Start()
 318         if err != nil {
 319                 return nil, err
 320         }
 321
 322         statReadme := make(chan bool)
 323         runner.ArvMountExit = make(chan error)
 324
 325         keepStatting := true
 326         go func() {
 327                 for keepStatting {
 328                         time.Sleep(100 * time.Millisecond)
 329                         _, err = os.Stat(fmt.Sprintf("%s/by_id/README", runner.ArvMountPoint))
 330                         if err == nil {
 331                                 keepStatting = false
 332                                 statReadme <- true
 333                         }
 334                 }
 335                 close(statReadme)
 336         }()
 337
 338         go func() {
 339                 mnterr := c.Wait()
 340                 if mnterr != nil {
 341                         runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
 342                 }
 343                 runner.ArvMountExit <- mnterr
 344                 close(runner.ArvMountExit)
 345         }()
 346
 347         select {
 348         case <-statReadme:
 349                 break
 350         case err := <-runner.ArvMountExit:
 351                 runner.ArvMount = nil
 352                 keepStatting = false
 353                 return nil, err
 354         }
 355
 356         return c, nil
 357 }
 358
 359 func (runner *ContainerRunner) SetupArvMountPoint(prefix string) (err error) {
 360         if runner.ArvMountPoint == "" {
 361                 runner.ArvMountPoint, err = runner.MkTempDir(runner.parentTemp, prefix)
 362         }
 363         return
 364 }
 365
 366 func copyfile(src string, dst string) (err error) {
 367         srcfile, err := os.Open(src)
 368         if err != nil {
 369                 return
 370         }
 371
 372         os.MkdirAll(path.Dir(dst), 0777)
 373
 374         dstfile, err := os.Create(dst)
 375         if err != nil {
 376                 return
 377         }
 378         _, err = io.Copy(dstfile, srcfile)
 379         if err != nil {
 380                 return
 381         }
 382
 383         err = srcfile.Close()
 384         err2 := dstfile.Close()
 385
 386         if err != nil {
 387                 return
 388         }
 389
 390         if err2 != nil {
 391                 return err2
 392         }
 393
 394         return nil
 395 }
 396
 397 func (runner *ContainerRunner) SetupMounts() (map[string]bindmount, error) {
 398         bindmounts := map[string]bindmount{}
 399         err := runner.SetupArvMountPoint("keep")
 400         if err != nil {
 401                 return nil, fmt.Errorf("While creating keep mount temp dir: %v", err)
 402         }
 403
 404         token, err := runner.ContainerToken()
 405         if err != nil {
 406                 return nil, fmt.Errorf("could not get container token: %s", err)
 407         }
 408         runner.CrunchLog.Printf("container token %q", token)
 409
 410         pdhOnly := true
 411         tmpcount := 0
 412         arvMountCmd := []string{
 413                 "arv-mount",
 414                 "--foreground",
 415                 "--allow-other",
 416                 "--read-write",
 417                 "--storage-classes", strings.Join(runner.Container.OutputStorageClasses, ","),
 418                 fmt.Sprintf("--crunchstat-interval=%v", runner.statInterval.Seconds())}
 419
 420         if runner.Container.RuntimeConstraints.KeepCacheRAM > 0 {
 421                 arvMountCmd = append(arvMountCmd, "--file-cache", fmt.Sprintf("%d", runner.Container.RuntimeConstraints.KeepCacheRAM))
 422         }
 423
 424         collectionPaths := []string{}
 425         needCertMount := true
 426         type copyFile struct {
 427                 src  string
 428                 bind string
 429         }
 430         var copyFiles []copyFile
 431
 432         var binds []string
 433         for bind := range runner.Container.Mounts {
 434                 binds = append(binds, bind)
 435         }
 436         for bind := range runner.SecretMounts {
 437                 if _, ok := runner.Container.Mounts[bind]; ok {
 438                         return nil, fmt.Errorf("secret mount %q conflicts with regular mount", bind)
 439                 }
 440                 if runner.SecretMounts[bind].Kind != "json" &&
 441                         runner.SecretMounts[bind].Kind != "text" {
 442                         return nil, fmt.Errorf("secret mount %q type is %q but only 'json' and 'text' are permitted",
 443                                 bind, runner.SecretMounts[bind].Kind)
 444                 }
 445                 binds = append(binds, bind)
 446         }
 447         sort.Strings(binds)
 448
 449         for _, bind := range binds {
 450                 mnt, ok := runner.Container.Mounts[bind]
 451                 if !ok {
 452                         mnt = runner.SecretMounts[bind]
 453                 }
 454                 if bind == "stdout" || bind == "stderr" {
 455                         // Is it a "file" mount kind?
 456                         if mnt.Kind != "file" {
 457                                 return nil, fmt.Errorf("unsupported mount kind '%s' for %s: only 'file' is supported", mnt.Kind, bind)
 458                         }
 459
 460                         // Does path start with OutputPath?
 461                         prefix := runner.Container.OutputPath
 462                         if !strings.HasSuffix(prefix, "/") {
 463                                 prefix += "/"
 464                         }
 465                         if !strings.HasPrefix(mnt.Path, prefix) {
 466                                 return nil, fmt.Errorf("%s path does not start with OutputPath: %s, %s", strings.Title(bind), mnt.Path, prefix)
 467                         }
 468                 }
 469
 470                 if bind == "stdin" {
 471                         // Is it a "collection" mount kind?
 472                         if mnt.Kind != "collection" && mnt.Kind != "json" {
 473                                 return nil, fmt.Errorf("unsupported mount kind '%s' for stdin: only 'collection' and 'json' are supported", mnt.Kind)
 474                         }
 475                 }
 476
 477                 if bind == "/etc/arvados/ca-certificates.crt" {
 478                         needCertMount = false
 479                 }
 480
 481                 if strings.HasPrefix(bind, runner.Container.OutputPath+"/") && bind != runner.Container.OutputPath+"/" {
 482                         if mnt.Kind != "collection" && mnt.Kind != "text" && mnt.Kind != "json" {
 483                                 return nil, fmt.Errorf("only mount points of kind 'collection', 'text' or 'json' are supported underneath the output_path for %q, was %q", bind, mnt.Kind)
 484                         }
 485                 }
 486
 487                 switch {
 488                 case mnt.Kind == "collection" && bind != "stdin":
 489                         var src string
 490                         if mnt.UUID != "" && mnt.PortableDataHash != "" {
 491                                 return nil, fmt.Errorf("cannot specify both 'uuid' and 'portable_data_hash' for a collection mount")
 492                         }
 493                         if mnt.UUID != "" {
 494                                 if mnt.Writable {
 495                                         return nil, fmt.Errorf("writing to existing collections currently not permitted")
 496                                 }
 497                                 pdhOnly = false
 498                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID)
 499                         } else if mnt.PortableDataHash != "" {
 500                                 if mnt.Writable && !strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
 501                                         return nil, fmt.Errorf("can never write to a collection specified by portable data hash")
 502                                 }
 503                                 idx := strings.Index(mnt.PortableDataHash, "/")
 504                                 if idx > 0 {
 505                                         mnt.Path = path.Clean(mnt.PortableDataHash[idx:])
 506                                         mnt.PortableDataHash = mnt.PortableDataHash[0:idx]
 507                                         runner.Container.Mounts[bind] = mnt
 508                                 }
 509                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
 510                                 if mnt.Path != "" && mnt.Path != "." {
 511                                         if strings.HasPrefix(mnt.Path, "./") {
 512                                                 mnt.Path = mnt.Path[2:]
 513                                         } else if strings.HasPrefix(mnt.Path, "/") {
 514                                                 mnt.Path = mnt.Path[1:]
 515                                         }
 516                                         src += "/" + mnt.Path
 517                                 }
 518                         } else {
 519                                 src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount)
 520                                 arvMountCmd = append(arvMountCmd, "--mount-tmp")
 521                                 arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%d", tmpcount))
 522                                 tmpcount++
 523                         }
 524                         if mnt.Writable {
 525                                 if bind == runner.Container.OutputPath {
 526                                         runner.HostOutputDir = src
 527                                         bindmounts[bind] = bindmount{HostPath: src}
 528                                 } else if strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
 529                                         copyFiles = append(copyFiles, copyFile{src, runner.HostOutputDir + bind[len(runner.Container.OutputPath):]})
 530                                 } else {
 531                                         bindmounts[bind] = bindmount{HostPath: src}
 532                                 }
 533                         } else {
 534                                 bindmounts[bind] = bindmount{HostPath: src, ReadOnly: true}
 535                         }
 536                         collectionPaths = append(collectionPaths, src)
 537
 538                 case mnt.Kind == "tmp":
 539                         var tmpdir string
 540                         tmpdir, err = runner.MkTempDir(runner.parentTemp, "tmp")
 541                         if err != nil {
 542                                 return nil, fmt.Errorf("while creating mount temp dir: %v", err)
 543                         }
 544                         st, staterr := os.Stat(tmpdir)
 545                         if staterr != nil {
 546                                 return nil, fmt.Errorf("while Stat on temp dir: %v", staterr)
 547                         }
 548                         err = os.Chmod(tmpdir, st.Mode()|os.ModeSetgid|0777)
 549                         if staterr != nil {
 550                                 return nil, fmt.Errorf("while Chmod temp dir: %v", err)
 551                         }
 552                         bindmounts[bind] = bindmount{HostPath: tmpdir}
 553                         if bind == runner.Container.OutputPath {
 554                                 runner.HostOutputDir = tmpdir
 555                         }
 556
 557                 case mnt.Kind == "json" || mnt.Kind == "text":
 558                         var filedata []byte
 559                         if mnt.Kind == "json" {
 560                                 filedata, err = json.Marshal(mnt.Content)
 561                                 if err != nil {
 562                                         return nil, fmt.Errorf("encoding json data: %v", err)
 563                                 }
 564                         } else {
 565                                 text, ok := mnt.Content.(string)
 566                                 if !ok {
 567                                         return nil, fmt.Errorf("content for mount %q must be a string", bind)
 568                                 }
 569                                 filedata = []byte(text)
 570                         }
 571
 572                         tmpdir, err := runner.MkTempDir(runner.parentTemp, mnt.Kind)
 573                         if err != nil {
 574                                 return nil, fmt.Errorf("creating temp dir: %v", err)
 575                         }
 576                         tmpfn := filepath.Join(tmpdir, "mountdata."+mnt.Kind)
 577                         err = ioutil.WriteFile(tmpfn, filedata, 0444)
 578                         if err != nil {
 579                                 return nil, fmt.Errorf("writing temp file: %v", err)
 580                         }
 581                         if strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
 582                                 copyFiles = append(copyFiles, copyFile{tmpfn, runner.HostOutputDir + bind[len(runner.Container.OutputPath):]})
 583                         } else {
 584                                 bindmounts[bind] = bindmount{HostPath: tmpfn, ReadOnly: true}
 585                         }
 586
 587                 case mnt.Kind == "git_tree":
 588                         tmpdir, err := runner.MkTempDir(runner.parentTemp, "git_tree")
 589                         if err != nil {
 590                                 return nil, fmt.Errorf("creating temp dir: %v", err)
 591                         }
 592                         err = gitMount(mnt).extractTree(runner.ContainerArvClient, tmpdir, token)
 593                         if err != nil {
 594                                 return nil, err
 595                         }
 596                         bindmounts[bind] = bindmount{HostPath: tmpdir, ReadOnly: true}
 597                 }
 598         }
 599
 600         if runner.HostOutputDir == "" {
 601                 return nil, fmt.Errorf("output path does not correspond to a writable mount point")
 602         }
 603
 604         if needCertMount && runner.Container.RuntimeConstraints.API {
 605                 for _, certfile := range arvadosclient.CertFiles {
 606                         _, err := os.Stat(certfile)
 607                         if err == nil {
 608                                 bindmounts["/etc/arvados/ca-certificates.crt"] = bindmount{HostPath: certfile, ReadOnly: true}
 609                                 break
 610                         }
 611                 }
 612         }
 613
 614         if pdhOnly {
 615                 arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
 616         } else {
 617                 arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id")
 618         }
 619         arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_uuid")
 620         arvMountCmd = append(arvMountCmd, runner.ArvMountPoint)
 621
 622         runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token)
 623         if err != nil {
 624                 return nil, fmt.Errorf("while trying to start arv-mount: %v", err)
 625         }
 626
 627         for _, p := range collectionPaths {
 628                 _, err = os.Stat(p)
 629                 if err != nil {
 630                         return nil, fmt.Errorf("while checking that input files exist: %v", err)
 631                 }
 632         }
 633
 634         for _, cp := range copyFiles {
 635                 st, err := os.Stat(cp.src)
 636                 if err != nil {
 637                         return nil, fmt.Errorf("while staging writable file from %q to %q: %v", cp.src, cp.bind, err)
 638                 }
 639                 if st.IsDir() {
 640                         err = filepath.Walk(cp.src, func(walkpath string, walkinfo os.FileInfo, walkerr error) error {
 641                                 if walkerr != nil {
 642                                         return walkerr
 643                                 }
 644                                 target := path.Join(cp.bind, walkpath[len(cp.src):])
 645                                 if walkinfo.Mode().IsRegular() {
 646                                         copyerr := copyfile(walkpath, target)
 647                                         if copyerr != nil {
 648                                                 return copyerr
 649                                         }
 650                                         return os.Chmod(target, walkinfo.Mode()|0777)
 651                                 } else if walkinfo.Mode().IsDir() {
 652                                         mkerr := os.MkdirAll(target, 0777)
 653                                         if mkerr != nil {
 654                                                 return mkerr
 655                                         }
 656                                         return os.Chmod(target, walkinfo.Mode()|os.ModeSetgid|0777)
 657                                 } else {
 658                                         return fmt.Errorf("source %q is not a regular file or directory", cp.src)
 659                                 }
 660                         })
 661                 } else if st.Mode().IsRegular() {
 662                         err = copyfile(cp.src, cp.bind)
 663                         if err == nil {
 664                                 err = os.Chmod(cp.bind, st.Mode()|0777)
 665                         }
 666                 }
 667                 if err != nil {
 668                         return nil, fmt.Errorf("while staging writable file from %q to %q: %v", cp.src, cp.bind, err)
 669                 }
 670         }
 671
 672         return bindmounts, nil
 673 }
 674
 675 func (runner *ContainerRunner) stopHoststat() error {
 676         if runner.hoststatReporter == nil {
 677                 return nil
 678         }
 679         runner.hoststatReporter.Stop()
 680         err := runner.hoststatLogger.Close()
 681         if err != nil {
 682                 return fmt.Errorf("error closing hoststat logs: %v", err)
 683         }
 684         return nil
 685 }
 686
 687 func (runner *ContainerRunner) startHoststat() error {
 688         w, err := runner.NewLogWriter("hoststat")
 689         if err != nil {
 690                 return err
 691         }
 692         runner.hoststatLogger = NewThrottledLogger(w)
 693         runner.hoststatReporter = &crunchstat.Reporter{
 694                 Logger:     log.New(runner.hoststatLogger, "", 0),
 695                 CgroupRoot: runner.cgroupRoot,
 696                 PollPeriod: runner.statInterval,
 697         }
 698         runner.hoststatReporter.Start()
 699         return nil
 700 }
 701
 702 func (runner *ContainerRunner) startCrunchstat() error {
 703         w, err := runner.NewLogWriter("crunchstat")
 704         if err != nil {
 705                 return err
 706         }
 707         runner.statLogger = NewThrottledLogger(w)
 708         runner.statReporter = &crunchstat.Reporter{
 709                 CID:          runner.executor.CgroupID(),
 710                 Logger:       log.New(runner.statLogger, "", 0),
 711                 CgroupParent: runner.expectCgroupParent,
 712                 CgroupRoot:   runner.cgroupRoot,
 713                 PollPeriod:   runner.statInterval,
 714                 TempDir:      runner.parentTemp,
 715         }
 716         runner.statReporter.Start()
 717         return nil
 718 }
 719
 720 type infoCommand struct {
 721         label string
 722         cmd   []string
 723 }
 724
 725 // LogHostInfo logs info about the current host, for debugging and
 726 // accounting purposes. Although it's logged as "node-info", this is
 727 // about the environment where crunch-run is actually running, which
 728 // might differ from what's described in the node record (see
 729 // LogNodeRecord).
 730 func (runner *ContainerRunner) LogHostInfo() (err error) {
 731         w, err := runner.NewLogWriter("node-info")
 732         if err != nil {
 733                 return
 734         }
 735
 736         commands := []infoCommand{
 737                 {
 738                         label: "Host Information",
 739                         cmd:   []string{"uname", "-a"},
 740                 },
 741                 {
 742                         label: "CPU Information",
 743                         cmd:   []string{"cat", "/proc/cpuinfo"},
 744                 },
 745                 {
 746                         label: "Memory Information",
 747                         cmd:   []string{"cat", "/proc/meminfo"},
 748                 },
 749                 {
 750                         label: "Disk Space",
 751                         cmd:   []string{"df", "-m", "/", os.TempDir()},
 752                 },
 753                 {
 754                         label: "Disk INodes",
 755                         cmd:   []string{"df", "-i", "/", os.TempDir()},
 756                 },
 757         }
 758
 759         // Run commands with informational output to be logged.
 760         for _, command := range commands {
 761                 fmt.Fprintln(w, command.label)
 762                 cmd := exec.Command(command.cmd[0], command.cmd[1:]...)
 763                 cmd.Stdout = w
 764                 cmd.Stderr = w
 765                 if err := cmd.Run(); err != nil {
 766                         err = fmt.Errorf("While running command %q: %v", command.cmd, err)
 767                         fmt.Fprintln(w, err)
 768                         return err
 769                 }
 770                 fmt.Fprintln(w, "")
 771         }
 772
 773         err = w.Close()
 774         if err != nil {
 775                 return fmt.Errorf("While closing node-info logs: %v", err)
 776         }
 777         return nil
 778 }
 779
 780 // LogContainerRecord gets and saves the raw JSON container record from the API server
 781 func (runner *ContainerRunner) LogContainerRecord() error {
 782         logged, err := runner.logAPIResponse("container", "containers", map[string]interface{}{"filters": [][]string{{"uuid", "=", runner.Container.UUID}}}, nil)
 783         if !logged && err == nil {
 784                 err = fmt.Errorf("error: no container record found for %s", runner.Container.UUID)
 785         }
 786         return err
 787 }
 788
 789 // LogNodeRecord logs the current host's InstanceType config entry (or
 790 // the arvados#node record, if running via crunch-dispatch-slurm).
 791 func (runner *ContainerRunner) LogNodeRecord() error {
 792         if it := os.Getenv("InstanceType"); it != "" {
 793                 // Dispatched via arvados-dispatch-cloud. Save
 794                 // InstanceType config fragment received from
 795                 // dispatcher on stdin.
 796                 w, err := runner.LogCollection.OpenFile("node.json", os.O_CREATE|os.O_WRONLY, 0666)
 797                 if err != nil {
 798                         return err
 799                 }
 800                 defer w.Close()
 801                 _, err = io.WriteString(w, it)
 802                 if err != nil {
 803                         return err
 804                 }
 805                 return w.Close()
 806         }
 807         // Dispatched via crunch-dispatch-slurm. Look up
 808         // apiserver's node record corresponding to
 809         // $SLURMD_NODENAME.
 810         hostname := os.Getenv("SLURMD_NODENAME")
 811         if hostname == "" {
 812                 hostname, _ = os.Hostname()
 813         }
 814         _, err := runner.logAPIResponse("node", "nodes", map[string]interface{}{"filters": [][]string{{"hostname", "=", hostname}}}, func(resp interface{}) {
 815                 // The "info" field has admin-only info when
 816                 // obtained with a privileged token, and
 817                 // should not be logged.
 818                 node, ok := resp.(map[string]interface{})
 819                 if ok {
 820                         delete(node, "info")
 821                 }
 822         })
 823         return err
 824 }
 825
 826 func (runner *ContainerRunner) logAPIResponse(label, path string, params map[string]interface{}, munge func(interface{})) (logged bool, err error) {
 827         writer, err := runner.LogCollection.OpenFile(label+".json", os.O_CREATE|os.O_WRONLY, 0666)
 828         if err != nil {
 829                 return false, err
 830         }
 831         w := &ArvLogWriter{
 832                 ArvClient:     runner.DispatcherArvClient,
 833                 UUID:          runner.Container.UUID,
 834                 loggingStream: label,
 835                 writeCloser:   writer,
 836         }
 837
 838         reader, err := runner.DispatcherArvClient.CallRaw("GET", path, "", "", arvadosclient.Dict(params))
 839         if err != nil {
 840                 return false, fmt.Errorf("error getting %s record: %v", label, err)
 841         }
 842         defer reader.Close()
 843
 844         dec := json.NewDecoder(reader)
 845         dec.UseNumber()
 846         var resp map[string]interface{}
 847         if err = dec.Decode(&resp); err != nil {
 848                 return false, fmt.Errorf("error decoding %s list response: %v", label, err)
 849         }
 850         items, ok := resp["items"].([]interface{})
 851         if !ok {
 852                 return false, fmt.Errorf("error decoding %s list response: no \"items\" key in API list response", label)
 853         } else if len(items) < 1 {
 854                 return false, nil
 855         }
 856         if munge != nil {
 857                 munge(items[0])
 858         }
 859         // Re-encode it using indentation to improve readability
 860         enc := json.NewEncoder(w)
 861         enc.SetIndent("", "    ")
 862         if err = enc.Encode(items[0]); err != nil {
 863                 return false, fmt.Errorf("error logging %s record: %v", label, err)
 864         }
 865         err = w.Close()
 866         if err != nil {
 867                 return false, fmt.Errorf("error closing %s.json in log collection: %v", label, err)
 868         }
 869         return true, nil
 870 }
 871
 872 func (runner *ContainerRunner) getStdoutFile(mntPath string) (*os.File, error) {
 873         stdoutPath := mntPath[len(runner.Container.OutputPath):]
 874         index := strings.LastIndex(stdoutPath, "/")
 875         if index > 0 {
 876                 subdirs := stdoutPath[:index]
 877                 if subdirs != "" {
 878                         st, err := os.Stat(runner.HostOutputDir)
 879                         if err != nil {
 880                                 return nil, fmt.Errorf("While Stat on temp dir: %v", err)
 881                         }
 882                         stdoutPath := filepath.Join(runner.HostOutputDir, subdirs)
 883                         err = os.MkdirAll(stdoutPath, st.Mode()|os.ModeSetgid|0777)
 884                         if err != nil {
 885                                 return nil, fmt.Errorf("While MkdirAll %q: %v", stdoutPath, err)
 886                         }
 887                 }
 888         }
 889         stdoutFile, err := os.Create(filepath.Join(runner.HostOutputDir, stdoutPath))
 890         if err != nil {
 891                 return nil, fmt.Errorf("While creating file %q: %v", stdoutPath, err)
 892         }
 893
 894         return stdoutFile, nil
 895 }
 896
 897 // CreateContainer creates the docker container.
 898 func (runner *ContainerRunner) CreateContainer(imageID string, bindmounts map[string]bindmount) error {
 899         var stdin io.ReadCloser = ioutil.NopCloser(bytes.NewReader(nil))
 900         if mnt, ok := runner.Container.Mounts["stdin"]; ok {
 901                 switch mnt.Kind {
 902                 case "collection":
 903                         var collID string
 904                         if mnt.UUID != "" {
 905                                 collID = mnt.UUID
 906                         } else {
 907                                 collID = mnt.PortableDataHash
 908                         }
 909                         path := runner.ArvMountPoint + "/by_id/" + collID + "/" + mnt.Path
 910                         f, err := os.Open(path)
 911                         if err != nil {
 912                                 return err
 913                         }
 914                         stdin = f
 915                 case "json":
 916                         j, err := json.Marshal(mnt.Content)
 917                         if err != nil {
 918                                 return fmt.Errorf("error encoding stdin json data: %v", err)
 919                         }
 920                         stdin = ioutil.NopCloser(bytes.NewReader(j))
 921                 default:
 922                         return fmt.Errorf("stdin mount has unsupported kind %q", mnt.Kind)
 923                 }
 924         }
 925
 926         var stdout, stderr io.WriteCloser
 927         if mnt, ok := runner.Container.Mounts["stdout"]; ok {
 928                 f, err := runner.getStdoutFile(mnt.Path)
 929                 if err != nil {
 930                         return err
 931                 }
 932                 stdout = f
 933         } else if w, err := runner.NewLogWriter("stdout"); err != nil {
 934                 return err
 935         } else {
 936                 stdout = NewThrottledLogger(w)
 937         }
 938
 939         if mnt, ok := runner.Container.Mounts["stderr"]; ok {
 940                 f, err := runner.getStdoutFile(mnt.Path)
 941                 if err != nil {
 942                         return err
 943                 }
 944                 stderr = f
 945         } else if w, err := runner.NewLogWriter("stderr"); err != nil {
 946                 return err
 947         } else {
 948                 stderr = NewThrottledLogger(w)
 949         }
 950
 951         env := runner.Container.Environment
 952         enableNetwork := runner.enableNetwork == "always"
 953         if runner.Container.RuntimeConstraints.API {
 954                 enableNetwork = true
 955                 tok, err := runner.ContainerToken()
 956                 if err != nil {
 957                         return err
 958                 }
 959                 env = map[string]string{}
 960                 for k, v := range runner.Container.Environment {
 961                         env[k] = v
 962                 }
 963                 env["ARVADOS_API_TOKEN"] = tok
 964                 env["ARVADOS_API_HOST"] = os.Getenv("ARVADOS_API_HOST")
 965                 env["ARVADOS_API_HOST_INSECURE"] = os.Getenv("ARVADOS_API_HOST_INSECURE")
 966         }
 967         workdir := runner.Container.Cwd
 968         if workdir == "." {
 969                 // both "" and "." mean default
 970                 workdir = ""
 971         }
 972         ram := runner.Container.RuntimeConstraints.RAM
 973         if !runner.enableMemoryLimit {
 974                 ram = 0
 975         }
 976         runner.executorStdin = stdin
 977         runner.executorStdout = stdout
 978         runner.executorStderr = stderr
 979         return runner.executor.Create(containerSpec{
 980                 Image:         imageID,
 981                 VCPUs:         runner.Container.RuntimeConstraints.VCPUs,
 982                 RAM:           ram,
 983                 WorkingDir:    workdir,
 984                 Env:           env,
 985                 BindMounts:    bindmounts,
 986                 Command:       runner.Container.Command,
 987                 EnableNetwork: enableNetwork,
 988                 NetworkMode:   runner.networkMode,
 989                 CgroupParent:  runner.setCgroupParent,
 990                 Stdin:         stdin,
 991                 Stdout:        stdout,
 992                 Stderr:        stderr,
 993         })
 994 }
 995
 996 // StartContainer starts the docker container created by CreateContainer.
 997 func (runner *ContainerRunner) StartContainer() error {
 998         runner.CrunchLog.Printf("Starting container")
 999         runner.cStateLock.Lock()
1000         defer runner.cStateLock.Unlock()
1001         if runner.cCancelled {
1002                 return ErrCancelled
1003         }
1004         err := runner.executor.Start()
1005         if err != nil {
1006                 var advice string
1007                 if m, e := regexp.MatchString("(?ms).*(exec|System error).*(no such file or directory|file not found).*", err.Error()); m && e == nil {
1008                         advice = fmt.Sprintf("\nPossible causes: command %q is missing, the interpreter given in #! is missing, or script has Windows line endings.", runner.Container.Command[0])
1009                 }
1010                 return fmt.Errorf("could not start container: %v%s", err, advice)
1011         }
1012         return nil
1013 }
1014
1015 // WaitFinish waits for the container to terminate, capture the exit code, and
1016 // close the stdout/stderr logging.
1017 func (runner *ContainerRunner) WaitFinish() error {
1018         runner.CrunchLog.Print("Waiting for container to finish")
1019         var timeout <-chan time.Time
1020         if s := runner.Container.SchedulingParameters.MaxRunTime; s > 0 {
1021                 timeout = time.After(time.Duration(s) * time.Second)
1022         }
1023         ctx, cancel := context.WithCancel(context.Background())
1024         defer cancel()
1025         go func() {
1026                 select {
1027                 case <-timeout:
1028                         runner.CrunchLog.Printf("maximum run time exceeded. Stopping container.")
1029                         runner.stop(nil)
1030                 case <-runner.ArvMountExit:
1031                         runner.CrunchLog.Printf("arv-mount exited while container is still running. Stopping container.")
1032                         runner.stop(nil)
1033                 case <-ctx.Done():
1034                 }
1035         }()
1036         exitcode, err := runner.executor.Wait(ctx)
1037         if err != nil {
1038                 runner.checkBrokenNode(err)
1039                 return err
1040         }
1041         runner.ExitCode = &exitcode
1042
1043         var returnErr error
1044         if err = runner.executorStdin.Close(); err != nil {
1045                 err = fmt.Errorf("error closing container stdin: %s", err)
1046                 runner.CrunchLog.Printf("%s", err)
1047                 returnErr = err
1048         }
1049         if err = runner.executorStdout.Close(); err != nil {
1050                 err = fmt.Errorf("error closing container stdout: %s", err)
1051                 runner.CrunchLog.Printf("%s", err)
1052                 if returnErr == nil {
1053                         returnErr = err
1054                 }
1055         }
1056         if err = runner.executorStderr.Close(); err != nil {
1057                 err = fmt.Errorf("error closing container stderr: %s", err)
1058                 runner.CrunchLog.Printf("%s", err)
1059                 if returnErr == nil {
1060                         returnErr = err
1061                 }
1062         }
1063
1064         if runner.statReporter != nil {
1065                 runner.statReporter.Stop()
1066                 err = runner.statLogger.Close()
1067                 if err != nil {
1068                         runner.CrunchLog.Printf("error closing crunchstat logs: %v", err)
1069                 }
1070         }
1071         return returnErr
1072 }
1073
1074 func (runner *ContainerRunner) updateLogs() {
1075         ticker := time.NewTicker(crunchLogUpdatePeriod / 360)
1076         defer ticker.Stop()
1077
1078         sigusr1 := make(chan os.Signal, 1)
1079         signal.Notify(sigusr1, syscall.SIGUSR1)
1080         defer signal.Stop(sigusr1)
1081
1082         saveAtTime := time.Now().Add(crunchLogUpdatePeriod)
1083         saveAtSize := crunchLogUpdateSize
1084         var savedSize int64
1085         for {
1086                 select {
1087                 case <-ticker.C:
1088                 case <-sigusr1:
1089                         saveAtTime = time.Now()
1090                 }
1091                 runner.logMtx.Lock()
1092                 done := runner.LogsPDH != nil
1093                 runner.logMtx.Unlock()
1094                 if done {
1095                         return
1096                 }
1097                 size := runner.LogCollection.Size()
1098                 if size == savedSize || (time.Now().Before(saveAtTime) && size < saveAtSize) {
1099                         continue
1100                 }
1101                 saveAtTime = time.Now().Add(crunchLogUpdatePeriod)
1102                 saveAtSize = runner.LogCollection.Size() + crunchLogUpdateSize
1103                 saved, err := runner.saveLogCollection(false)
1104                 if err != nil {
1105                         runner.CrunchLog.Printf("error updating log collection: %s", err)
1106                         continue
1107                 }
1108
1109                 var updated arvados.Container
1110                 err = runner.DispatcherArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{
1111                         "container": arvadosclient.Dict{"log": saved.PortableDataHash},
1112                 }, &updated)
1113                 if err != nil {
1114                         runner.CrunchLog.Printf("error updating container log to %s: %s", saved.PortableDataHash, err)
1115                         continue
1116                 }
1117
1118                 savedSize = size
1119         }
1120 }
1121
1122 func (runner *ContainerRunner) reportArvMountWarning(pattern, text string) {
1123         var updated arvados.Container
1124         err := runner.DispatcherArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{
1125                 "container": arvadosclient.Dict{
1126                         "runtime_status": arvadosclient.Dict{
1127                                 "warning":       "arv-mount: " + pattern,
1128                                 "warningDetail": text,
1129                         },
1130                 },
1131         }, &updated)
1132         if err != nil {
1133                 runner.CrunchLog.Printf("error updating container runtime_status: %s", err)
1134         }
1135 }
1136
1137 // CaptureOutput saves data from the container's output directory if
1138 // needed, and updates the container output accordingly.
1139 func (runner *ContainerRunner) CaptureOutput(bindmounts map[string]bindmount) error {
1140         if runner.Container.RuntimeConstraints.API {
1141                 // Output may have been set directly by the container, so
1142                 // refresh the container record to check.
1143                 err := runner.DispatcherArvClient.Get("containers", runner.Container.UUID,
1144                         nil, &runner.Container)
1145                 if err != nil {
1146                         return err
1147                 }
1148                 if runner.Container.Output != "" {
1149                         // Container output is already set.
1150                         runner.OutputPDH = &runner.Container.Output
1151                         return nil
1152                 }
1153         }
1154
1155         txt, err := (&copier{
1156                 client:        runner.containerClient,
1157                 arvClient:     runner.ContainerArvClient,
1158                 keepClient:    runner.ContainerKeepClient,
1159                 hostOutputDir: runner.HostOutputDir,
1160                 ctrOutputDir:  runner.Container.OutputPath,
1161                 bindmounts:    bindmounts,
1162                 mounts:        runner.Container.Mounts,
1163                 secretMounts:  runner.SecretMounts,
1164                 logger:        runner.CrunchLog,
1165         }).Copy()
1166         if err != nil {
1167                 return err
1168         }
1169         if n := len(regexp.MustCompile(` [0-9a-f]+\+\S*\+R`).FindAllStringIndex(txt, -1)); n > 0 {
1170                 runner.CrunchLog.Printf("Copying %d data blocks from remote input collections...", n)
1171                 fs, err := (&arvados.Collection{ManifestText: txt}).FileSystem(runner.containerClient, runner.ContainerKeepClient)
1172                 if err != nil {
1173                         return err
1174                 }
1175                 txt, err = fs.MarshalManifest(".")
1176                 if err != nil {
1177                         return err
1178                 }
1179         }
1180         var resp arvados.Collection
1181         err = runner.ContainerArvClient.Create("collections", arvadosclient.Dict{
1182                 "ensure_unique_name": true,
1183                 "collection": arvadosclient.Dict{
1184                         "is_trashed":    true,
1185                         "name":          "output for " + runner.Container.UUID,
1186                         "manifest_text": txt,
1187                 },
1188         }, &resp)
1189         if err != nil {
1190                 return fmt.Errorf("error creating output collection: %v", err)
1191         }
1192         runner.OutputPDH = &resp.PortableDataHash
1193         return nil
1194 }
1195
1196 func (runner *ContainerRunner) CleanupDirs() {
1197         if runner.ArvMount != nil {
1198                 var delay int64 = 8
1199                 umount := exec.Command("arv-mount", fmt.Sprintf("--unmount-timeout=%d", delay), "--unmount", runner.ArvMountPoint)
1200                 umount.Stdout = runner.CrunchLog
1201                 umount.Stderr = runner.CrunchLog
1202                 runner.CrunchLog.Printf("Running %v", umount.Args)
1203                 umnterr := umount.Start()
1204
1205                 if umnterr != nil {
1206                         runner.CrunchLog.Printf("Error unmounting: %v", umnterr)
1207                         runner.ArvMount.Process.Kill()
1208                 } else {
1209                         // If arv-mount --unmount gets stuck for any reason, we
1210                         // don't want to wait for it forever.  Do Wait() in a goroutine
1211                         // so it doesn't block crunch-run.
1212                         umountExit := make(chan error)
1213                         go func() {
1214                                 mnterr := umount.Wait()
1215                                 if mnterr != nil {
1216                                         runner.CrunchLog.Printf("Error unmounting: %v", mnterr)
1217                                 }
1218                                 umountExit <- mnterr
1219                         }()
1220
1221                         for again := true; again; {
1222                                 again = false
1223                                 select {
1224                                 case <-umountExit:
1225                                         umount = nil
1226                                         again = true
1227                                 case <-runner.ArvMountExit:
1228                                         break
1229                                 case <-time.After(time.Duration((delay + 1) * int64(time.Second))):
1230                                         runner.CrunchLog.Printf("Timed out waiting for unmount")
1231                                         if umount != nil {
1232                                                 umount.Process.Kill()
1233                                         }
1234                                         runner.ArvMount.Process.Kill()
1235                                 }
1236                         }
1237                 }
1238                 runner.ArvMount = nil
1239         }
1240
1241         if runner.ArvMountPoint != "" {
1242                 if rmerr := os.Remove(runner.ArvMountPoint); rmerr != nil {
1243                         runner.CrunchLog.Printf("While cleaning up arv-mount directory %s: %v", runner.ArvMountPoint, rmerr)
1244                 }
1245                 runner.ArvMountPoint = ""
1246         }
1247
1248         if rmerr := os.RemoveAll(runner.parentTemp); rmerr != nil {
1249                 runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", runner.parentTemp, rmerr)
1250         }
1251 }
1252
1253 // CommitLogs posts the collection containing the final container logs.
1254 func (runner *ContainerRunner) CommitLogs() error {
1255         func() {
1256                 // Hold cStateLock to prevent races on CrunchLog (e.g., stop()).
1257                 runner.cStateLock.Lock()
1258                 defer runner.cStateLock.Unlock()
1259
1260                 runner.CrunchLog.Print(runner.finalState)
1261
1262                 if runner.arvMountLog != nil {
1263                         runner.arvMountLog.Close()
1264                 }
1265                 runner.CrunchLog.Close()
1266
1267                 // Closing CrunchLog above allows them to be committed to Keep at this
1268                 // point, but re-open crunch log with ArvClient in case there are any
1269                 // other further errors (such as failing to write the log to Keep!)
1270                 // while shutting down
1271                 runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{
1272                         ArvClient:     runner.DispatcherArvClient,
1273                         UUID:          runner.Container.UUID,
1274                         loggingStream: "crunch-run",
1275                         writeCloser:   nil,
1276                 })
1277                 runner.CrunchLog.Immediate = log.New(os.Stderr, runner.Container.UUID+" ", 0)
1278         }()
1279
1280         if runner.LogsPDH != nil {
1281                 // If we have already assigned something to LogsPDH,
1282                 // we must be closing the re-opened log, which won't
1283                 // end up getting attached to the container record and
1284                 // therefore doesn't need to be saved as a collection
1285                 // -- it exists only to send logs to other channels.
1286                 return nil
1287         }
1288         saved, err := runner.saveLogCollection(true)
1289         if err != nil {
1290                 return fmt.Errorf("error saving log collection: %s", err)
1291         }
1292         runner.logMtx.Lock()
1293         defer runner.logMtx.Unlock()
1294         runner.LogsPDH = &saved.PortableDataHash
1295         return nil
1296 }
1297
1298 func (runner *ContainerRunner) saveLogCollection(final bool) (response arvados.Collection, err error) {
1299         runner.logMtx.Lock()
1300         defer runner.logMtx.Unlock()
1301         if runner.LogsPDH != nil {
1302                 // Already finalized.
1303                 return
1304         }
1305         updates := arvadosclient.Dict{
1306                 "name": "logs for " + runner.Container.UUID,
1307         }
1308         mt, err1 := runner.LogCollection.MarshalManifest(".")
1309         if err1 == nil {
1310                 // Only send updated manifest text if there was no
1311                 // error.
1312                 updates["manifest_text"] = mt
1313         }
1314
1315         // Even if flushing the manifest had an error, we still want
1316         // to update the log record, if possible, to push the trash_at
1317         // and delete_at times into the future.  Details on bug
1318         // #17293.
1319         if final {
1320                 updates["is_trashed"] = true
1321         } else {
1322                 exp := time.Now().Add(crunchLogUpdatePeriod * 24)
1323                 updates["trash_at"] = exp
1324                 updates["delete_at"] = exp
1325         }
1326         reqBody := arvadosclient.Dict{"collection": updates}
1327         var err2 error
1328         if runner.logUUID == "" {
1329                 reqBody["ensure_unique_name"] = true
1330                 err2 = runner.DispatcherArvClient.Create("collections", reqBody, &response)
1331         } else {
1332                 err2 = runner.DispatcherArvClient.Update("collections", runner.logUUID, reqBody, &response)
1333         }
1334         if err2 == nil {
1335                 runner.logUUID = response.UUID
1336         }
1337
1338         if err1 != nil || err2 != nil {
1339                 err = fmt.Errorf("error recording logs: %q, %q", err1, err2)
1340         }
1341         return
1342 }
1343
1344 // UpdateContainerRunning updates the container state to "Running"
1345 func (runner *ContainerRunner) UpdateContainerRunning() error {
1346         runner.cStateLock.Lock()
1347         defer runner.cStateLock.Unlock()
1348         if runner.cCancelled {
1349                 return ErrCancelled
1350         }
1351         return runner.DispatcherArvClient.Update("containers", runner.Container.UUID,
1352                 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running", "gateway_address": runner.gateway.Address}}, nil)
1353 }
1354
1355 // ContainerToken returns the api_token the container (and any
1356 // arv-mount processes) are allowed to use.
1357 func (runner *ContainerRunner) ContainerToken() (string, error) {
1358         if runner.token != "" {
1359                 return runner.token, nil
1360         }
1361
1362         var auth arvados.APIClientAuthorization
1363         err := runner.DispatcherArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
1364         if err != nil {
1365                 return "", err
1366         }
1367         runner.token = fmt.Sprintf("v2/%s/%s/%s", auth.UUID, auth.APIToken, runner.Container.UUID)
1368         return runner.token, nil
1369 }
1370
1371 // UpdateContainerFinal updates the container record state on API
1372 // server to "Complete" or "Cancelled"
1373 func (runner *ContainerRunner) UpdateContainerFinal() error {
1374         update := arvadosclient.Dict{}
1375         update["state"] = runner.finalState
1376         if runner.LogsPDH != nil {
1377                 update["log"] = *runner.LogsPDH
1378         }
1379         if runner.finalState == "Complete" {
1380                 if runner.ExitCode != nil {
1381                         update["exit_code"] = *runner.ExitCode
1382                 }
1383                 if runner.OutputPDH != nil {
1384                         update["output"] = *runner.OutputPDH
1385                 }
1386         }
1387         return runner.DispatcherArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
1388 }
1389
1390 // IsCancelled returns the value of Cancelled, with goroutine safety.
1391 func (runner *ContainerRunner) IsCancelled() bool {
1392         runner.cStateLock.Lock()
1393         defer runner.cStateLock.Unlock()
1394         return runner.cCancelled
1395 }
1396
1397 // NewArvLogWriter creates an ArvLogWriter
1398 func (runner *ContainerRunner) NewArvLogWriter(name string) (io.WriteCloser, error) {
1399         writer, err := runner.LogCollection.OpenFile(name+".txt", os.O_CREATE|os.O_WRONLY, 0666)
1400         if err != nil {
1401                 return nil, err
1402         }
1403         return &ArvLogWriter{
1404                 ArvClient:     runner.DispatcherArvClient,
1405                 UUID:          runner.Container.UUID,
1406                 loggingStream: name,
1407                 writeCloser:   writer,
1408         }, nil
1409 }
1410
1411 // Run the full container lifecycle.
1412 func (runner *ContainerRunner) Run() (err error) {
1413         runner.CrunchLog.Printf("crunch-run %s started", cmd.Version.String())
1414         runner.CrunchLog.Printf("Executing container '%s' using %s runtime", runner.Container.UUID, runner.executor.Runtime())
1415
1416         hostname, hosterr := os.Hostname()
1417         if hosterr != nil {
1418                 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
1419         } else {
1420                 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
1421         }
1422
1423         runner.finalState = "Queued"
1424
1425         defer func() {
1426                 runner.CleanupDirs()
1427
1428                 runner.CrunchLog.Printf("crunch-run finished")
1429                 runner.CrunchLog.Close()
1430         }()
1431
1432         err = runner.fetchContainerRecord()
1433         if err != nil {
1434                 return
1435         }
1436         if runner.Container.State != "Locked" {
1437                 return fmt.Errorf("dispatch error detected: container %q has state %q", runner.Container.UUID, runner.Container.State)
1438         }
1439
1440         var bindmounts map[string]bindmount
1441         defer func() {
1442                 // checkErr prints e (unless it's nil) and sets err to
1443                 // e (unless err is already non-nil). Thus, if err
1444                 // hasn't already been assigned when Run() returns,
1445                 // this cleanup func will cause Run() to return the
1446                 // first non-nil error that is passed to checkErr().
1447                 checkErr := func(errorIn string, e error) {
1448                         if e == nil {
1449                                 return
1450                         }
1451                         runner.CrunchLog.Printf("error in %s: %v", errorIn, e)
1452                         if err == nil {
1453                                 err = e
1454                         }
1455                         if runner.finalState == "Complete" {
1456                                 // There was an error in the finalization.
1457                                 runner.finalState = "Cancelled"
1458                         }
1459                 }
1460
1461                 // Log the error encountered in Run(), if any
1462                 checkErr("Run", err)
1463
1464                 if runner.finalState == "Queued" {
1465                         runner.UpdateContainerFinal()
1466                         return
1467                 }
1468
1469                 if runner.IsCancelled() {
1470                         runner.finalState = "Cancelled"
1471                         // but don't return yet -- we still want to
1472                         // capture partial output and write logs
1473                 }
1474
1475                 if bindmounts != nil {
1476                         checkErr("CaptureOutput", runner.CaptureOutput(bindmounts))
1477                 }
1478                 checkErr("stopHoststat", runner.stopHoststat())
1479                 checkErr("CommitLogs", runner.CommitLogs())
1480                 runner.CleanupDirs()
1481                 checkErr("UpdateContainerFinal", runner.UpdateContainerFinal())
1482         }()
1483
1484         runner.setupSignals()
1485         err = runner.startHoststat()
1486         if err != nil {
1487                 return
1488         }
1489
1490         // set up FUSE mount and binds
1491         bindmounts, err = runner.SetupMounts()
1492         if err != nil {
1493                 runner.finalState = "Cancelled"
1494                 err = fmt.Errorf("While setting up mounts: %v", err)
1495                 return
1496         }
1497
1498         // check for and/or load image
1499         imageID, err := runner.LoadImage()
1500         if err != nil {
1501                 if !runner.checkBrokenNode(err) {
1502                         // Failed to load image but not due to a "broken node"
1503                         // condition, probably user error.
1504                         runner.finalState = "Cancelled"
1505                 }
1506                 err = fmt.Errorf("While loading container image: %v", err)
1507                 return
1508         }
1509
1510         err = runner.CreateContainer(imageID, bindmounts)
1511         if err != nil {
1512                 return
1513         }
1514         err = runner.LogHostInfo()
1515         if err != nil {
1516                 return
1517         }
1518         err = runner.LogNodeRecord()
1519         if err != nil {
1520                 return
1521         }
1522         err = runner.LogContainerRecord()
1523         if err != nil {
1524                 return
1525         }
1526
1527         if runner.IsCancelled() {
1528                 return
1529         }
1530
1531         err = runner.UpdateContainerRunning()
1532         if err != nil {
1533                 return
1534         }
1535         runner.finalState = "Cancelled"
1536
1537         err = runner.startCrunchstat()
1538         if err != nil {
1539                 return
1540         }
1541
1542         err = runner.StartContainer()
1543         if err != nil {
1544                 runner.checkBrokenNode(err)
1545                 return
1546         }
1547
1548         err = runner.WaitFinish()
1549         if err == nil && !runner.IsCancelled() {
1550                 runner.finalState = "Complete"
1551         }
1552         return
1553 }
1554
1555 // Fetch the current container record (uuid = runner.Container.UUID)
1556 // into runner.Container.
1557 func (runner *ContainerRunner) fetchContainerRecord() error {
1558         reader, err := runner.DispatcherArvClient.CallRaw("GET", "containers", runner.Container.UUID, "", nil)
1559         if err != nil {
1560                 return fmt.Errorf("error fetching container record: %v", err)
1561         }
1562         defer reader.Close()
1563
1564         dec := json.NewDecoder(reader)
1565         dec.UseNumber()
1566         err = dec.Decode(&runner.Container)
1567         if err != nil {
1568                 return fmt.Errorf("error decoding container record: %v", err)
1569         }
1570
1571         var sm struct {
1572                 SecretMounts map[string]arvados.Mount `json:"secret_mounts"`
1573         }
1574
1575         containerToken, err := runner.ContainerToken()
1576         if err != nil {
1577                 return fmt.Errorf("error getting container token: %v", err)
1578         }
1579
1580         runner.ContainerArvClient, runner.ContainerKeepClient,
1581                 runner.containerClient, err = runner.MkArvClient(containerToken)
1582         if err != nil {
1583                 return fmt.Errorf("error creating container API client: %v", err)
1584         }
1585
1586         runner.ContainerKeepClient.SetStorageClasses(runner.Container.OutputStorageClasses)
1587         runner.DispatcherKeepClient.SetStorageClasses(runner.Container.OutputStorageClasses)
1588
1589         err = runner.ContainerArvClient.Call("GET", "containers", runner.Container.UUID, "secret_mounts", nil, &sm)
1590         if err != nil {
1591                 if apierr, ok := err.(arvadosclient.APIServerError); !ok || apierr.HttpStatusCode != 404 {
1592                         return fmt.Errorf("error fetching secret_mounts: %v", err)
1593                 }
1594                 // ok && apierr.HttpStatusCode == 404, which means
1595                 // secret_mounts isn't supported by this API server.
1596         }
1597         runner.SecretMounts = sm.SecretMounts
1598
1599         return nil
1600 }
1601
1602 // NewContainerRunner creates a new container runner.
1603 func NewContainerRunner(dispatcherClient *arvados.Client,
1604         dispatcherArvClient IArvadosClient,
1605         dispatcherKeepClient IKeepClient,
1606         containerUUID string) (*ContainerRunner, error) {
1607
1608         cr := &ContainerRunner{
1609                 dispatcherClient:     dispatcherClient,
1610                 DispatcherArvClient:  dispatcherArvClient,
1611                 DispatcherKeepClient: dispatcherKeepClient,
1612         }
1613         cr.NewLogWriter = cr.NewArvLogWriter
1614         cr.RunArvMount = cr.ArvMountCmd
1615         cr.MkTempDir = ioutil.TempDir
1616         cr.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
1617                 cl, err := arvadosclient.MakeArvadosClient()
1618                 if err != nil {
1619                         return nil, nil, nil, err
1620                 }
1621                 cl.ApiToken = token
1622                 kc, err := keepclient.MakeKeepClient(cl)
1623                 if err != nil {
1624                         return nil, nil, nil, err
1625                 }
1626                 c2 := arvados.NewClientFromEnv()
1627                 c2.AuthToken = token
1628                 return cl, kc, c2, nil
1629         }
1630         var err error
1631         cr.LogCollection, err = (&arvados.Collection{}).FileSystem(cr.dispatcherClient, cr.DispatcherKeepClient)
1632         if err != nil {
1633                 return nil, err
1634         }
1635         cr.Container.UUID = containerUUID
1636         w, err := cr.NewLogWriter("crunch-run")
1637         if err != nil {
1638                 return nil, err
1639         }
1640         cr.CrunchLog = NewThrottledLogger(w)
1641         cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
1642
1643         loadLogThrottleParams(dispatcherArvClient)
1644         go cr.updateLogs()
1645
1646         return cr, nil
1647 }
1648
1649 func (command) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
1650         flags := flag.NewFlagSet(prog, flag.ContinueOnError)
1651         statInterval := flags.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
1652         cgroupRoot := flags.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
1653         cgroupParent := flags.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
1654         cgroupParentSubsystem := flags.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
1655         caCertsPath := flags.String("ca-certs", "", "Path to TLS root certificates")
1656         detach := flags.Bool("detach", false, "Detach from parent process and run in the background")
1657         stdinConfig := flags.Bool("stdin-config", false, "Load config and environment variables from JSON message on stdin")
1658         sleep := flags.Duration("sleep", 0, "Delay before starting (testing use only)")
1659         kill := flags.Int("kill", -1, "Send signal to an existing crunch-run process for given UUID")
1660         list := flags.Bool("list", false, "List UUIDs of existing crunch-run processes")
1661         enableMemoryLimit := flags.Bool("enable-memory-limit", true, "tell container runtime to limit container's memory usage")
1662         enableNetwork := flags.String("container-enable-networking", "default", "enable networking \"always\" (for all containers) or \"default\" (for containers that request it)")
1663         networkMode := flags.String("container-network-mode", "default", `Docker network mode for container (use any argument valid for docker --net)`)
1664         memprofile := flags.String("memprofile", "", "write memory profile to `file` after running container")
1665         runtimeEngine := flags.String("runtime-engine", "docker", "container runtime: docker or singularity")
1666         flags.Duration("check-containerd", 0, "Ignored. Exists for compatibility with older versions.")
1667
1668         ignoreDetachFlag := false
1669         if len(args) > 0 && args[0] == "-no-detach" {
1670                 // This process was invoked by a parent process, which
1671                 // has passed along its own arguments, including
1672                 // -detach, after the leading -no-detach flag.  Strip
1673                 // the leading -no-detach flag (it's not recognized by
1674                 // flags.Parse()) and ignore the -detach flag that
1675                 // comes later.
1676                 args = args[1:]
1677                 ignoreDetachFlag = true
1678         }
1679
1680         if err := flags.Parse(args); err == flag.ErrHelp {
1681                 return 0
1682         } else if err != nil {
1683                 log.Print(err)
1684                 return 1
1685         }
1686
1687         containerUUID := flags.Arg(0)
1688
1689         switch {
1690         case *detach && !ignoreDetachFlag:
1691                 return Detach(containerUUID, prog, args, os.Stdin, os.Stdout, os.Stderr)
1692         case *kill >= 0:
1693                 return KillProcess(containerUUID, syscall.Signal(*kill), os.Stdout, os.Stderr)
1694         case *list:
1695                 return ListProcesses(os.Stdout, os.Stderr)
1696         }
1697
1698         if len(containerUUID) != 27 {
1699                 log.Printf("usage: %s [options] UUID", prog)
1700                 return 1
1701         }
1702
1703         var conf ConfigData
1704         if *stdinConfig {
1705                 err := json.NewDecoder(os.Stdin).Decode(&conf)
1706                 if err != nil {
1707                         log.Print(err)
1708                         return 1
1709                 }
1710                 for k, v := range conf.Env {
1711                         err = os.Setenv(k, v)
1712                         if err != nil {
1713                                 log.Printf("setenv(%q): %s", k, err)
1714                                 return 1
1715                         }
1716                 }
1717                 if conf.Cluster != nil {
1718                         // ClusterID is missing from the JSON
1719                         // representation, but we need it to generate
1720                         // a valid config file for keepstore, so we
1721                         // fill it using the container UUID prefix.
1722                         conf.Cluster.ClusterID = containerUUID[:5]
1723                 }
1724         }
1725
1726         log.Printf("crunch-run %s started", cmd.Version.String())
1727         time.Sleep(*sleep)
1728
1729         if *caCertsPath != "" {
1730                 arvadosclient.CertFiles = []string{*caCertsPath}
1731         }
1732
1733         var keepstoreLog bufThenWrite
1734         keepstore, err := startLocalKeepstore(conf, io.MultiWriter(&keepstoreLog, stderr))
1735         if err != nil {
1736                 log.Print(err)
1737                 return 1
1738         }
1739         if keepstore != nil {
1740                 defer keepstore.Process.Kill()
1741         }
1742
1743         api, err := arvadosclient.MakeArvadosClient()
1744         if err != nil {
1745                 log.Printf("%s: %v", containerUUID, err)
1746                 return 1
1747         }
1748         api.Retries = 8
1749
1750         kc, kcerr := keepclient.MakeKeepClient(api)
1751         if kcerr != nil {
1752                 log.Printf("%s: %v", containerUUID, kcerr)
1753                 return 1
1754         }
1755         kc.BlockCache = &keepclient.BlockCache{MaxBlocks: 2}
1756         kc.Retries = 4
1757
1758         cr, err := NewContainerRunner(arvados.NewClientFromEnv(), api, kc, containerUUID)
1759         if err != nil {
1760                 log.Print(err)
1761                 return 1
1762         }
1763
1764         if keepstore != nil {
1765                 w, err := cr.NewLogWriter("keepstore")
1766                 if err != nil {
1767                         log.Print(err)
1768                         return 1
1769                 }
1770                 err = keepstoreLog.SetWriter(NewThrottledLogger(w))
1771                 if err != nil {
1772                         log.Print(err)
1773                         return 1
1774                 }
1775         }
1776
1777         switch *runtimeEngine {
1778         case "docker":
1779                 cr.executor, err = newDockerExecutor(containerUUID, cr.CrunchLog.Printf, cr.containerWatchdogInterval)
1780         case "singularity":
1781                 cr.executor, err = newSingularityExecutor(cr.CrunchLog.Printf)
1782         default:
1783                 cr.CrunchLog.Printf("%s: unsupported RuntimeEngine %q", containerUUID, *runtimeEngine)
1784                 cr.CrunchLog.Close()
1785                 return 1
1786         }
1787         if err != nil {
1788                 cr.CrunchLog.Printf("%s: %v", containerUUID, err)
1789                 cr.checkBrokenNode(err)
1790                 cr.CrunchLog.Close()
1791                 return 1
1792         }
1793         defer cr.executor.Close()
1794
1795         gwAuthSecret := os.Getenv("GatewayAuthSecret")
1796         os.Unsetenv("GatewayAuthSecret")
1797         if gwAuthSecret == "" {
1798                 // not safe to run a gateway service without an auth
1799                 // secret
1800                 cr.CrunchLog.Printf("Not starting a gateway server (GatewayAuthSecret was not provided by dispatcher)")
1801         } else if gwListen := os.Getenv("GatewayAddress"); gwListen == "" {
1802                 // dispatcher did not tell us which external IP
1803                 // address to advertise --> no gateway service
1804                 cr.CrunchLog.Printf("Not starting a gateway server (GatewayAddress was not provided by dispatcher)")
1805         } else if de, ok := cr.executor.(*dockerExecutor); ok {
1806                 cr.gateway = Gateway{
1807                         Address:            gwListen,
1808                         AuthSecret:         gwAuthSecret,
1809                         ContainerUUID:      containerUUID,
1810                         DockerContainerID:  &de.containerID,
1811                         Log:                cr.CrunchLog,
1812                         ContainerIPAddress: dockerContainerIPAddress(&de.containerID),
1813                 }
1814                 err = cr.gateway.Start()
1815                 if err != nil {
1816                         log.Printf("error starting gateway server: %s", err)
1817                         return 1
1818                 }
1819         }
1820
1821         parentTemp, tmperr := cr.MkTempDir("", "crunch-run."+containerUUID+".")
1822         if tmperr != nil {
1823                 log.Printf("%s: %v", containerUUID, tmperr)
1824                 return 1
1825         }
1826
1827         cr.parentTemp = parentTemp
1828         cr.statInterval = *statInterval
1829         cr.cgroupRoot = *cgroupRoot
1830         cr.expectCgroupParent = *cgroupParent
1831         cr.enableMemoryLimit = *enableMemoryLimit
1832         cr.enableNetwork = *enableNetwork
1833         cr.networkMode = *networkMode
1834         if *cgroupParentSubsystem != "" {
1835                 p := findCgroup(*cgroupParentSubsystem)
1836                 cr.setCgroupParent = p
1837                 cr.expectCgroupParent = p
1838         }
1839
1840         runerr := cr.Run()
1841
1842         if *memprofile != "" {
1843                 f, err := os.Create(*memprofile)
1844                 if err != nil {
1845                         log.Printf("could not create memory profile: %s", err)
1846                 }
1847                 runtime.GC() // get up-to-date statistics
1848                 if err := pprof.WriteHeapProfile(f); err != nil {
1849                         log.Printf("could not write memory profile: %s", err)
1850                 }
1851                 closeerr := f.Close()
1852                 if closeerr != nil {
1853                         log.Printf("closing memprofile file: %s", err)
1854                 }
1855         }
1856
1857         if runerr != nil {
1858                 log.Printf("%s: %v", containerUUID, runerr)
1859                 return 1
1860         }
1861         return 0
1862 }
1863
1864 func startLocalKeepstore(configData ConfigData, logbuf io.Writer) (*exec.Cmd, error) {
1865         if configData.Cluster == nil || configData.KeepBuffers < 1 {
1866                 return nil, nil
1867         }
1868
1869         // Rather than have an alternate way to tell keepstore how
1870         // many buffers to use when starting it this way, we just
1871         // modify the cluster configuration that we feed it on stdin.
1872         configData.Cluster.API.MaxKeepBlobBuffers = configData.KeepBuffers
1873
1874         ln, err := net.Listen("tcp", "localhost:0")
1875         if err != nil {
1876                 return nil, err
1877         }
1878         _, port, err := net.SplitHostPort(ln.Addr().String())
1879         if err != nil {
1880                 ln.Close()
1881                 return nil, err
1882         }
1883         ln.Close()
1884         url := "http://localhost:" + port
1885
1886         fmt.Fprintf(logbuf, "starting keepstore on %s\n", url)
1887
1888         var confJSON bytes.Buffer
1889         err = json.NewEncoder(&confJSON).Encode(arvados.Config{
1890                 Clusters: map[string]arvados.Cluster{
1891                         configData.Cluster.ClusterID: *configData.Cluster,
1892                 },
1893         })
1894         if err != nil {
1895                 return nil, err
1896         }
1897         cmd := exec.Command("/proc/self/exe", "keepstore", "-config=-")
1898         cmd.Stdin = &confJSON
1899         cmd.Stdout = logbuf
1900         cmd.Stderr = logbuf
1901         cmd.Env = []string{
1902                 "GOGC=10",
1903                 "ARVADOS_SERVICE_INTERNAL_URL=" + url,
1904         }
1905         err = cmd.Start()
1906         if err != nil {
1907                 return nil, fmt.Errorf("error starting keepstore process: %w", err)
1908         }
1909         ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
1910         defer cancel()
1911         poll := time.NewTicker(time.Second / 10)
1912         defer poll.Stop()
1913         client := http.Client{}
1914         for range poll.C {
1915                 testReq, err := http.NewRequestWithContext(ctx, "GET", url, nil)
1916                 if err != nil {
1917                         return nil, err
1918                 }
1919                 resp, err := client.Do(testReq)
1920                 if err == nil {
1921                         // Success -- don't need to check the
1922                         // response, we just need to know it's
1923                         // accepting requests.
1924                         resp.Body.Close()
1925                         break
1926                 }
1927                 if ctx.Err() != nil {
1928                         return nil, fmt.Errorf("timed out waiting for new keepstore process to accept a request")
1929                 }
1930         }
1931         os.Setenv("ARVADOS_KEEP_SERVICES", url)
1932         return cmd, nil
1933 }