services/crunch-run/crunchrun.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "bytes"
   9         "encoding/json"
  10         "errors"
  11         "flag"
  12         "fmt"
  13         "io"
  14         "io/ioutil"
  15         "log"
  16         "os"
  17         "os/exec"
  18         "os/signal"
  19         "path"
  20         "path/filepath"
  21         "regexp"
  22         "runtime"
  23         "runtime/pprof"
  24         "sort"
  25         "strings"
  26         "sync"
  27         "syscall"
  28         "time"
  29
  30         "git.curoverse.com/arvados.git/lib/crunchstat"
  31         "git.curoverse.com/arvados.git/sdk/go/arvados"
  32         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  33         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  34         "git.curoverse.com/arvados.git/sdk/go/manifest"
  35         "golang.org/x/net/context"
  36
  37         dockertypes "github.com/docker/docker/api/types"
  38         dockercontainer "github.com/docker/docker/api/types/container"
  39         dockernetwork "github.com/docker/docker/api/types/network"
  40         dockerclient "github.com/docker/docker/client"
  41 )
  42
  43 var version = "dev"
  44
  45 // IArvadosClient is the minimal Arvados API methods used by crunch-run.
  46 type IArvadosClient interface {
  47         Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
  48         Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
  49         Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
  50         Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error
  51         CallRaw(method string, resourceType string, uuid string, action string, parameters arvadosclient.Dict) (reader io.ReadCloser, err error)
  52         Discovery(key string) (interface{}, error)
  53 }
  54
  55 // ErrCancelled is the error returned when the container is cancelled.
  56 var ErrCancelled = errors.New("Cancelled")
  57
  58 // IKeepClient is the minimal Keep API methods used by crunch-run.
  59 type IKeepClient interface {
  60         PutHB(hash string, buf []byte) (string, int, error)
  61         ManifestFileReader(m manifest.Manifest, filename string) (arvados.File, error)
  62         ClearBlockCache()
  63 }
  64
  65 // NewLogWriter is a factory function to create a new log writer.
  66 type NewLogWriter func(name string) io.WriteCloser
  67
  68 type RunArvMount func(args []string, tok string) (*exec.Cmd, error)
  69
  70 type MkTempDir func(string, string) (string, error)
  71
  72 // ThinDockerClient is the minimal Docker client interface used by crunch-run.
  73 type ThinDockerClient interface {
  74         ContainerAttach(ctx context.Context, container string, options dockertypes.ContainerAttachOptions) (dockertypes.HijackedResponse, error)
  75         ContainerCreate(ctx context.Context, config *dockercontainer.Config, hostConfig *dockercontainer.HostConfig,
  76                 networkingConfig *dockernetwork.NetworkingConfig, containerName string) (dockercontainer.ContainerCreateCreatedBody, error)
  77         ContainerStart(ctx context.Context, container string, options dockertypes.ContainerStartOptions) error
  78         ContainerRemove(ctx context.Context, container string, options dockertypes.ContainerRemoveOptions) error
  79         ContainerWait(ctx context.Context, container string, condition dockercontainer.WaitCondition) (<-chan dockercontainer.ContainerWaitOKBody, <-chan error)
  80         ImageInspectWithRaw(ctx context.Context, image string) (dockertypes.ImageInspect, []byte, error)
  81         ImageLoad(ctx context.Context, input io.Reader, quiet bool) (dockertypes.ImageLoadResponse, error)
  82         ImageRemove(ctx context.Context, image string, options dockertypes.ImageRemoveOptions) ([]dockertypes.ImageDeleteResponseItem, error)
  83 }
  84
  85 // ContainerRunner is the main stateful struct used for a single execution of a
  86 // container.
  87 type ContainerRunner struct {
  88         Docker    ThinDockerClient
  89         ArvClient IArvadosClient
  90         Kc        IKeepClient
  91         arvados.Container
  92         ContainerConfig dockercontainer.Config
  93         dockercontainer.HostConfig
  94         token       string
  95         ContainerID string
  96         ExitCode    *int
  97         NewLogWriter
  98         loggingDone   chan bool
  99         CrunchLog     *ThrottledLogger
 100         Stdout        io.WriteCloser
 101         Stderr        io.WriteCloser
 102         LogCollection *CollectionWriter
 103         LogsPDH       *string
 104         RunArvMount
 105         MkTempDir
 106         ArvMount       *exec.Cmd
 107         ArvMountPoint  string
 108         HostOutputDir  string
 109         CleanupTempDir []string
 110         Binds          []string
 111         Volumes        map[string]struct{}
 112         OutputPDH      *string
 113         SigChan        chan os.Signal
 114         ArvMountExit   chan error
 115         finalState     string
 116
 117         statLogger   io.WriteCloser
 118         statReporter *crunchstat.Reporter
 119         statInterval time.Duration
 120         cgroupRoot   string
 121         // What we expect the container's cgroup parent to be.
 122         expectCgroupParent string
 123         // What we tell docker to use as the container's cgroup
 124         // parent. Note: Ideally we would use the same field for both
 125         // expectCgroupParent and setCgroupParent, and just make it
 126         // default to "docker". However, when using docker < 1.10 with
 127         // systemd, specifying a non-empty cgroup parent (even the
 128         // default value "docker") hits a docker bug
 129         // (https://github.com/docker/docker/issues/17126). Using two
 130         // separate fields makes it possible to use the "expect cgroup
 131         // parent to be X" feature even on sites where the "specify
 132         // cgroup parent" feature breaks.
 133         setCgroupParent string
 134
 135         cStateLock sync.Mutex
 136         cCancelled bool // StopContainer() invoked
 137
 138         enableNetwork string // one of "default" or "always"
 139         networkMode   string // passed through to HostConfig.NetworkMode
 140         arvMountLog   *ThrottledLogger
 141 }
 142
 143 // setupSignals sets up signal handling to gracefully terminate the underlying
 144 // Docker container and update state when receiving a TERM, INT or QUIT signal.
 145 func (runner *ContainerRunner) setupSignals() {
 146         runner.SigChan = make(chan os.Signal, 1)
 147         signal.Notify(runner.SigChan, syscall.SIGTERM)
 148         signal.Notify(runner.SigChan, syscall.SIGINT)
 149         signal.Notify(runner.SigChan, syscall.SIGQUIT)
 150
 151         go func(sig chan os.Signal) {
 152                 for s := range sig {
 153                         runner.CrunchLog.Printf("caught signal: %v", s)
 154                         runner.stop()
 155                 }
 156         }(runner.SigChan)
 157 }
 158
 159 // stop the underlying Docker container.
 160 func (runner *ContainerRunner) stop() {
 161         runner.cStateLock.Lock()
 162         defer runner.cStateLock.Unlock()
 163         if runner.ContainerID == "" {
 164                 return
 165         }
 166         runner.cCancelled = true
 167         runner.CrunchLog.Printf("removing container")
 168         err := runner.Docker.ContainerRemove(context.TODO(), runner.ContainerID, dockertypes.ContainerRemoveOptions{Force: true})
 169         if err != nil {
 170                 runner.CrunchLog.Printf("error removing container: %s", err)
 171         }
 172 }
 173
 174 func (runner *ContainerRunner) stopSignals() {
 175         if runner.SigChan != nil {
 176                 signal.Stop(runner.SigChan)
 177         }
 178 }
 179
 180 var errorBlacklist = []string{
 181         "(?ms).*[Cc]annot connect to the Docker daemon.*",
 182         "(?ms).*oci runtime error.*starting container process.*container init.*mounting.*to rootfs.*no such file or directory.*",
 183 }
 184 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)")
 185
 186 func (runner *ContainerRunner) checkBrokenNode(goterr error) bool {
 187         for _, d := range errorBlacklist {
 188                 if m, e := regexp.MatchString(d, goterr.Error()); m && e == nil {
 189                         runner.CrunchLog.Printf("Error suggests node is unable to run containers: %v", goterr)
 190                         if *brokenNodeHook == "" {
 191                                 runner.CrunchLog.Printf("No broken node hook provided, cannot mark node as broken.")
 192                         } else {
 193                                 runner.CrunchLog.Printf("Running broken node hook %q", *brokenNodeHook)
 194                                 // run killme script
 195                                 c := exec.Command(*brokenNodeHook)
 196                                 c.Stdout = runner.CrunchLog
 197                                 c.Stderr = runner.CrunchLog
 198                                 err := c.Run()
 199                                 if err != nil {
 200                                         runner.CrunchLog.Printf("Error running broken node hook: %v", err)
 201                                 }
 202                         }
 203                         return true
 204                 }
 205         }
 206         return false
 207 }
 208
 209 // LoadImage determines the docker image id from the container record and
 210 // checks if it is available in the local Docker image store.  If not, it loads
 211 // the image from Keep.
 212 func (runner *ContainerRunner) LoadImage() (err error) {
 213
 214         runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.Container.ContainerImage)
 215
 216         var collection arvados.Collection
 217         err = runner.ArvClient.Get("collections", runner.Container.ContainerImage, nil, &collection)
 218         if err != nil {
 219                 return fmt.Errorf("While getting container image collection: %v", err)
 220         }
 221         manifest := manifest.Manifest{Text: collection.ManifestText}
 222         var img, imageID string
 223         for ms := range manifest.StreamIter() {
 224                 img = ms.FileStreamSegments[0].Name
 225                 if !strings.HasSuffix(img, ".tar") {
 226                         return fmt.Errorf("First file in the container image collection does not end in .tar")
 227                 }
 228                 imageID = img[:len(img)-4]
 229         }
 230
 231         runner.CrunchLog.Printf("Using Docker image id '%s'", imageID)
 232
 233         _, _, err = runner.Docker.ImageInspectWithRaw(context.TODO(), imageID)
 234         if err != nil {
 235                 runner.CrunchLog.Print("Loading Docker image from keep")
 236
 237                 var readCloser io.ReadCloser
 238                 readCloser, err = runner.Kc.ManifestFileReader(manifest, img)
 239                 if err != nil {
 240                         return fmt.Errorf("While creating ManifestFileReader for container image: %v", err)
 241                 }
 242
 243                 response, err := runner.Docker.ImageLoad(context.TODO(), readCloser, true)
 244                 if err != nil {
 245                         return fmt.Errorf("While loading container image into Docker: %v", err)
 246                 }
 247
 248                 defer response.Body.Close()
 249                 rbody, err := ioutil.ReadAll(response.Body)
 250                 if err != nil {
 251                         return fmt.Errorf("Reading response to image load: %v", err)
 252                 }
 253                 runner.CrunchLog.Printf("Docker response: %s", rbody)
 254         } else {
 255                 runner.CrunchLog.Print("Docker image is available")
 256         }
 257
 258         runner.ContainerConfig.Image = imageID
 259
 260         runner.Kc.ClearBlockCache()
 261
 262         return nil
 263 }
 264
 265 func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string, token string) (c *exec.Cmd, err error) {
 266         c = exec.Command("arv-mount", arvMountCmd...)
 267
 268         // Copy our environment, but override ARVADOS_API_TOKEN with
 269         // the container auth token.
 270         c.Env = nil
 271         for _, s := range os.Environ() {
 272                 if !strings.HasPrefix(s, "ARVADOS_API_TOKEN=") {
 273                         c.Env = append(c.Env, s)
 274                 }
 275         }
 276         c.Env = append(c.Env, "ARVADOS_API_TOKEN="+token)
 277
 278         runner.arvMountLog = NewThrottledLogger(runner.NewLogWriter("arv-mount"))
 279         c.Stdout = runner.arvMountLog
 280         c.Stderr = runner.arvMountLog
 281
 282         runner.CrunchLog.Printf("Running %v", c.Args)
 283
 284         err = c.Start()
 285         if err != nil {
 286                 return nil, err
 287         }
 288
 289         statReadme := make(chan bool)
 290         runner.ArvMountExit = make(chan error)
 291
 292         keepStatting := true
 293         go func() {
 294                 for keepStatting {
 295                         time.Sleep(100 * time.Millisecond)
 296                         _, err = os.Stat(fmt.Sprintf("%s/by_id/README", runner.ArvMountPoint))
 297                         if err == nil {
 298                                 keepStatting = false
 299                                 statReadme <- true
 300                         }
 301                 }
 302                 close(statReadme)
 303         }()
 304
 305         go func() {
 306                 mnterr := c.Wait()
 307                 if mnterr != nil {
 308                         runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
 309                 }
 310                 runner.ArvMountExit <- mnterr
 311                 close(runner.ArvMountExit)
 312         }()
 313
 314         select {
 315         case <-statReadme:
 316                 break
 317         case err := <-runner.ArvMountExit:
 318                 runner.ArvMount = nil
 319                 keepStatting = false
 320                 return nil, err
 321         }
 322
 323         return c, nil
 324 }
 325
 326 func (runner *ContainerRunner) SetupArvMountPoint(prefix string) (err error) {
 327         if runner.ArvMountPoint == "" {
 328                 runner.ArvMountPoint, err = runner.MkTempDir("", prefix)
 329         }
 330         return
 331 }
 332
 333 func copyfile(src string, dst string) (err error) {
 334         srcfile, err := os.Open(src)
 335         if err != nil {
 336                 return
 337         }
 338
 339         os.MkdirAll(path.Dir(dst), 0770)
 340
 341         dstfile, err := os.Create(dst)
 342         if err != nil {
 343                 return
 344         }
 345         _, err = io.Copy(srcfile, dstfile)
 346         if err != nil {
 347                 return
 348         }
 349
 350         err = srcfile.Close()
 351         err2 := dstfile.Close()
 352
 353         if err != nil {
 354                 return
 355         }
 356
 357         if err2 != nil {
 358                 return err2
 359         }
 360
 361         return nil
 362 }
 363
 364 func (runner *ContainerRunner) SetupMounts() (err error) {
 365         err = runner.SetupArvMountPoint("keep")
 366         if err != nil {
 367                 return fmt.Errorf("While creating keep mount temp dir: %v", err)
 368         }
 369
 370         token, err := runner.ContainerToken()
 371         if err != nil {
 372                 return fmt.Errorf("could not get container token: %s", err)
 373         }
 374
 375         pdhOnly := true
 376         tmpcount := 0
 377         arvMountCmd := []string{
 378                 "--foreground",
 379                 "--allow-other",
 380                 "--read-write",
 381                 fmt.Sprintf("--crunchstat-interval=%v", runner.statInterval.Seconds())}
 382
 383         if runner.Container.RuntimeConstraints.KeepCacheRAM > 0 {
 384                 arvMountCmd = append(arvMountCmd, "--file-cache", fmt.Sprintf("%d", runner.Container.RuntimeConstraints.KeepCacheRAM))
 385         }
 386
 387         collectionPaths := []string{}
 388         runner.Binds = nil
 389         runner.Volumes = make(map[string]struct{})
 390         needCertMount := true
 391         type copyFile struct {
 392                 src  string
 393                 bind string
 394         }
 395         var copyFiles []copyFile
 396
 397         var binds []string
 398         for bind := range runner.Container.Mounts {
 399                 binds = append(binds, bind)
 400         }
 401         sort.Strings(binds)
 402
 403         for _, bind := range binds {
 404                 mnt := runner.Container.Mounts[bind]
 405                 if bind == "stdout" || bind == "stderr" {
 406                         // Is it a "file" mount kind?
 407                         if mnt.Kind != "file" {
 408                                 return fmt.Errorf("Unsupported mount kind '%s' for %s. Only 'file' is supported.", mnt.Kind, bind)
 409                         }
 410
 411                         // Does path start with OutputPath?
 412                         prefix := runner.Container.OutputPath
 413                         if !strings.HasSuffix(prefix, "/") {
 414                                 prefix += "/"
 415                         }
 416                         if !strings.HasPrefix(mnt.Path, prefix) {
 417                                 return fmt.Errorf("%s path does not start with OutputPath: %s, %s", strings.Title(bind), mnt.Path, prefix)
 418                         }
 419                 }
 420
 421                 if bind == "stdin" {
 422                         // Is it a "collection" mount kind?
 423                         if mnt.Kind != "collection" && mnt.Kind != "json" {
 424                                 return fmt.Errorf("Unsupported mount kind '%s' for stdin. Only 'collection' or 'json' are supported.", mnt.Kind)
 425                         }
 426                 }
 427
 428                 if bind == "/etc/arvados/ca-certificates.crt" {
 429                         needCertMount = false
 430                 }
 431
 432                 if strings.HasPrefix(bind, runner.Container.OutputPath+"/") && bind != runner.Container.OutputPath+"/" {
 433                         if mnt.Kind != "collection" {
 434                                 return fmt.Errorf("Only mount points of kind 'collection' are supported underneath the output_path: %v", bind)
 435                         }
 436                 }
 437
 438                 switch {
 439                 case mnt.Kind == "collection" && bind != "stdin":
 440                         var src string
 441                         if mnt.UUID != "" && mnt.PortableDataHash != "" {
 442                                 return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount")
 443                         }
 444                         if mnt.UUID != "" {
 445                                 if mnt.Writable {
 446                                         return fmt.Errorf("Writing to existing collections currently not permitted.")
 447                                 }
 448                                 pdhOnly = false
 449                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID)
 450                         } else if mnt.PortableDataHash != "" {
 451                                 if mnt.Writable && !strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
 452                                         return fmt.Errorf("Can never write to a collection specified by portable data hash")
 453                                 }
 454                                 idx := strings.Index(mnt.PortableDataHash, "/")
 455                                 if idx > 0 {
 456                                         mnt.Path = path.Clean(mnt.PortableDataHash[idx:])
 457                                         mnt.PortableDataHash = mnt.PortableDataHash[0:idx]
 458                                         runner.Container.Mounts[bind] = mnt
 459                                 }
 460                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
 461                                 if mnt.Path != "" && mnt.Path != "." {
 462                                         if strings.HasPrefix(mnt.Path, "./") {
 463                                                 mnt.Path = mnt.Path[2:]
 464                                         } else if strings.HasPrefix(mnt.Path, "/") {
 465                                                 mnt.Path = mnt.Path[1:]
 466                                         }
 467                                         src += "/" + mnt.Path
 468                                 }
 469                         } else {
 470                                 src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount)
 471                                 arvMountCmd = append(arvMountCmd, "--mount-tmp")
 472                                 arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%d", tmpcount))
 473                                 tmpcount += 1
 474                         }
 475                         if mnt.Writable {
 476                                 if bind == runner.Container.OutputPath {
 477                                         runner.HostOutputDir = src
 478                                         runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
 479                                 } else if strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
 480                                         copyFiles = append(copyFiles, copyFile{src, runner.HostOutputDir + bind[len(runner.Container.OutputPath):]})
 481                                 } else {
 482                                         runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
 483                                 }
 484                         } else {
 485                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
 486                         }
 487                         collectionPaths = append(collectionPaths, src)
 488
 489                 case mnt.Kind == "tmp":
 490                         var tmpdir string
 491                         tmpdir, err = runner.MkTempDir("", "")
 492                         if err != nil {
 493                                 return fmt.Errorf("While creating mount temp dir: %v", err)
 494                         }
 495                         st, staterr := os.Stat(tmpdir)
 496                         if staterr != nil {
 497                                 return fmt.Errorf("While Stat on temp dir: %v", staterr)
 498                         }
 499                         err = os.Chmod(tmpdir, st.Mode()|os.ModeSetgid|0777)
 500                         if staterr != nil {
 501                                 return fmt.Errorf("While Chmod temp dir: %v", err)
 502                         }
 503                         runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
 504                         runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", tmpdir, bind))
 505                         if bind == runner.Container.OutputPath {
 506                                 runner.HostOutputDir = tmpdir
 507                         }
 508
 509                 case mnt.Kind == "json":
 510                         jsondata, err := json.Marshal(mnt.Content)
 511                         if err != nil {
 512                                 return fmt.Errorf("encoding json data: %v", err)
 513                         }
 514                         // Create a tempdir with a single file
 515                         // (instead of just a tempfile): this way we
 516                         // can ensure the file is world-readable
 517                         // inside the container, without having to
 518                         // make it world-readable on the docker host.
 519                         tmpdir, err := runner.MkTempDir("", "")
 520                         if err != nil {
 521                                 return fmt.Errorf("creating temp dir: %v", err)
 522                         }
 523                         runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
 524                         tmpfn := filepath.Join(tmpdir, "mountdata.json")
 525                         err = ioutil.WriteFile(tmpfn, jsondata, 0644)
 526                         if err != nil {
 527                                 return fmt.Errorf("writing temp file: %v", err)
 528                         }
 529                         runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind))
 530
 531                 case mnt.Kind == "git_tree":
 532                         tmpdir, err := runner.MkTempDir("", "")
 533                         if err != nil {
 534                                 return fmt.Errorf("creating temp dir: %v", err)
 535                         }
 536                         runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
 537                         err = gitMount(mnt).extractTree(runner.ArvClient, tmpdir, token)
 538                         if err != nil {
 539                                 return err
 540                         }
 541                         runner.Binds = append(runner.Binds, tmpdir+":"+bind+":ro")
 542                 }
 543         }
 544
 545         if runner.HostOutputDir == "" {
 546                 return fmt.Errorf("Output path does not correspond to a writable mount point")
 547         }
 548
 549         if wantAPI := runner.Container.RuntimeConstraints.API; needCertMount && wantAPI != nil && *wantAPI {
 550                 for _, certfile := range arvadosclient.CertFiles {
 551                         _, err := os.Stat(certfile)
 552                         if err == nil {
 553                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:/etc/arvados/ca-certificates.crt:ro", certfile))
 554                                 break
 555                         }
 556                 }
 557         }
 558
 559         if pdhOnly {
 560                 arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
 561         } else {
 562                 arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id")
 563         }
 564         arvMountCmd = append(arvMountCmd, runner.ArvMountPoint)
 565
 566         runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token)
 567         if err != nil {
 568                 return fmt.Errorf("While trying to start arv-mount: %v", err)
 569         }
 570
 571         for _, p := range collectionPaths {
 572                 _, err = os.Stat(p)
 573                 if err != nil {
 574                         return fmt.Errorf("While checking that input files exist: %v", err)
 575                 }
 576         }
 577
 578         for _, cp := range copyFiles {
 579                 dir, err := os.Stat(cp.src)
 580                 if err == nil {
 581                         if dir.IsDir() {
 582                                 err = filepath.Walk(cp.src, func(walkpath string, walkinfo os.FileInfo, walkerr error) error {
 583                                         return copyfile(walkpath, path.Join(cp.bind, walkpath))
 584                                 })
 585                         } else {
 586                                 err = copyfile(cp.src, cp.bind)
 587                         }
 588                 }
 589                 if err != nil {
 590                         return fmt.Errorf("While staging writable files: %v", err)
 591                 }
 592         }
 593
 594         return nil
 595 }
 596
 597 func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
 598         // Handle docker log protocol
 599         // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
 600
 601         header := make([]byte, 8)
 602         for {
 603                 _, readerr := io.ReadAtLeast(containerReader, header, 8)
 604
 605                 if readerr == nil {
 606                         readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
 607                         if header[0] == 1 {
 608                                 // stdout
 609                                 _, readerr = io.CopyN(runner.Stdout, containerReader, readsize)
 610                         } else {
 611                                 // stderr
 612                                 _, readerr = io.CopyN(runner.Stderr, containerReader, readsize)
 613                         }
 614                 }
 615
 616                 if readerr != nil {
 617                         if readerr != io.EOF {
 618                                 runner.CrunchLog.Printf("While reading docker logs: %v", readerr)
 619                         }
 620
 621                         closeerr := runner.Stdout.Close()
 622                         if closeerr != nil {
 623                                 runner.CrunchLog.Printf("While closing stdout logs: %v", closeerr)
 624                         }
 625
 626                         closeerr = runner.Stderr.Close()
 627                         if closeerr != nil {
 628                                 runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr)
 629                         }
 630
 631                         if runner.statReporter != nil {
 632                                 runner.statReporter.Stop()
 633                                 closeerr = runner.statLogger.Close()
 634                                 if closeerr != nil {
 635                                         runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr)
 636                                 }
 637                         }
 638
 639                         close(runner.loggingDone)
 640                         return
 641                 }
 642         }
 643 }
 644
 645 func (runner *ContainerRunner) StartCrunchstat() {
 646         runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat"))
 647         runner.statReporter = &crunchstat.Reporter{
 648                 CID:          runner.ContainerID,
 649                 Logger:       log.New(runner.statLogger, "", 0),
 650                 CgroupParent: runner.expectCgroupParent,
 651                 CgroupRoot:   runner.cgroupRoot,
 652                 PollPeriod:   runner.statInterval,
 653         }
 654         runner.statReporter.Start()
 655 }
 656
 657 type infoCommand struct {
 658         label string
 659         cmd   []string
 660 }
 661
 662 // LogHostInfo logs info about the current host, for debugging and
 663 // accounting purposes. Although it's logged as "node-info", this is
 664 // about the environment where crunch-run is actually running, which
 665 // might differ from what's described in the node record (see
 666 // LogNodeRecord).
 667 func (runner *ContainerRunner) LogHostInfo() (err error) {
 668         w := runner.NewLogWriter("node-info")
 669
 670         commands := []infoCommand{
 671                 {
 672                         label: "Host Information",
 673                         cmd:   []string{"uname", "-a"},
 674                 },
 675                 {
 676                         label: "CPU Information",
 677                         cmd:   []string{"cat", "/proc/cpuinfo"},
 678                 },
 679                 {
 680                         label: "Memory Information",
 681                         cmd:   []string{"cat", "/proc/meminfo"},
 682                 },
 683                 {
 684                         label: "Disk Space",
 685                         cmd:   []string{"df", "-m", "/", os.TempDir()},
 686                 },
 687                 {
 688                         label: "Disk INodes",
 689                         cmd:   []string{"df", "-i", "/", os.TempDir()},
 690                 },
 691         }
 692
 693         // Run commands with informational output to be logged.
 694         for _, command := range commands {
 695                 fmt.Fprintln(w, command.label)
 696                 cmd := exec.Command(command.cmd[0], command.cmd[1:]...)
 697                 cmd.Stdout = w
 698                 cmd.Stderr = w
 699                 if err := cmd.Run(); err != nil {
 700                         err = fmt.Errorf("While running command %q: %v", command.cmd, err)
 701                         fmt.Fprintln(w, err)
 702                         return err
 703                 }
 704                 fmt.Fprintln(w, "")
 705         }
 706
 707         err = w.Close()
 708         if err != nil {
 709                 return fmt.Errorf("While closing node-info logs: %v", err)
 710         }
 711         return nil
 712 }
 713
 714 // LogContainerRecord gets and saves the raw JSON container record from the API server
 715 func (runner *ContainerRunner) LogContainerRecord() error {
 716         logged, err := runner.logAPIResponse("container", "containers", map[string]interface{}{"filters": [][]string{{"uuid", "=", runner.Container.UUID}}}, nil)
 717         if !logged && err == nil {
 718                 err = fmt.Errorf("error: no container record found for %s", runner.Container.UUID)
 719         }
 720         return err
 721 }
 722
 723 // LogNodeRecord logs arvados#node record corresponding to the current host.
 724 func (runner *ContainerRunner) LogNodeRecord() error {
 725         hostname := os.Getenv("SLURMD_NODENAME")
 726         if hostname == "" {
 727                 hostname, _ = os.Hostname()
 728         }
 729         _, err := runner.logAPIResponse("node", "nodes", map[string]interface{}{"filters": [][]string{{"hostname", "=", hostname}}}, func(resp interface{}) {
 730                 // The "info" field has admin-only info when obtained
 731                 // with a privileged token, and should not be logged.
 732                 node, ok := resp.(map[string]interface{})
 733                 if ok {
 734                         delete(node, "info")
 735                 }
 736         })
 737         return err
 738 }
 739
 740 func (runner *ContainerRunner) logAPIResponse(label, path string, params map[string]interface{}, munge func(interface{})) (logged bool, err error) {
 741         w := &ArvLogWriter{
 742                 ArvClient:     runner.ArvClient,
 743                 UUID:          runner.Container.UUID,
 744                 loggingStream: label,
 745                 writeCloser:   runner.LogCollection.Open(label + ".json"),
 746         }
 747
 748         reader, err := runner.ArvClient.CallRaw("GET", path, "", "", arvadosclient.Dict(params))
 749         if err != nil {
 750                 return false, fmt.Errorf("error getting %s record: %v", label, err)
 751         }
 752         defer reader.Close()
 753
 754         dec := json.NewDecoder(reader)
 755         dec.UseNumber()
 756         var resp map[string]interface{}
 757         if err = dec.Decode(&resp); err != nil {
 758                 return false, fmt.Errorf("error decoding %s list response: %v", label, err)
 759         }
 760         items, ok := resp["items"].([]interface{})
 761         if !ok {
 762                 return false, fmt.Errorf("error decoding %s list response: no \"items\" key in API list response", label)
 763         } else if len(items) < 1 {
 764                 return false, nil
 765         }
 766         if munge != nil {
 767                 munge(items[0])
 768         }
 769         // Re-encode it using indentation to improve readability
 770         enc := json.NewEncoder(w)
 771         enc.SetIndent("", "    ")
 772         if err = enc.Encode(items[0]); err != nil {
 773                 return false, fmt.Errorf("error logging %s record: %v", label, err)
 774         }
 775         err = w.Close()
 776         if err != nil {
 777                 return false, fmt.Errorf("error closing %s.json in log collection: %v", label, err)
 778         }
 779         return true, nil
 780 }
 781
 782 // AttachStreams connects the docker container stdin, stdout and stderr logs
 783 // to the Arvados logger which logs to Keep and the API server logs table.
 784 func (runner *ContainerRunner) AttachStreams() (err error) {
 785
 786         runner.CrunchLog.Print("Attaching container streams")
 787
 788         // If stdin mount is provided, attach it to the docker container
 789         var stdinRdr arvados.File
 790         var stdinJson []byte
 791         if stdinMnt, ok := runner.Container.Mounts["stdin"]; ok {
 792                 if stdinMnt.Kind == "collection" {
 793                         var stdinColl arvados.Collection
 794                         collId := stdinMnt.UUID
 795                         if collId == "" {
 796                                 collId = stdinMnt.PortableDataHash
 797                         }
 798                         err = runner.ArvClient.Get("collections", collId, nil, &stdinColl)
 799                         if err != nil {
 800                                 return fmt.Errorf("While getting stding collection: %v", err)
 801                         }
 802
 803                         stdinRdr, err = runner.Kc.ManifestFileReader(manifest.Manifest{Text: stdinColl.ManifestText}, stdinMnt.Path)
 804                         if os.IsNotExist(err) {
 805                                 return fmt.Errorf("stdin collection path not found: %v", stdinMnt.Path)
 806                         } else if err != nil {
 807                                 return fmt.Errorf("While getting stdin collection path %v: %v", stdinMnt.Path, err)
 808                         }
 809                 } else if stdinMnt.Kind == "json" {
 810                         stdinJson, err = json.Marshal(stdinMnt.Content)
 811                         if err != nil {
 812                                 return fmt.Errorf("While encoding stdin json data: %v", err)
 813                         }
 814                 }
 815         }
 816
 817         stdinUsed := stdinRdr != nil || len(stdinJson) != 0
 818         response, err := runner.Docker.ContainerAttach(context.TODO(), runner.ContainerID,
 819                 dockertypes.ContainerAttachOptions{Stream: true, Stdin: stdinUsed, Stdout: true, Stderr: true})
 820         if err != nil {
 821                 return fmt.Errorf("While attaching container stdout/stderr streams: %v", err)
 822         }
 823
 824         runner.loggingDone = make(chan bool)
 825
 826         if stdoutMnt, ok := runner.Container.Mounts["stdout"]; ok {
 827                 stdoutFile, err := runner.getStdoutFile(stdoutMnt.Path)
 828                 if err != nil {
 829                         return err
 830                 }
 831                 runner.Stdout = stdoutFile
 832         } else {
 833                 runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout"))
 834         }
 835
 836         if stderrMnt, ok := runner.Container.Mounts["stderr"]; ok {
 837                 stderrFile, err := runner.getStdoutFile(stderrMnt.Path)
 838                 if err != nil {
 839                         return err
 840                 }
 841                 runner.Stderr = stderrFile
 842         } else {
 843                 runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr"))
 844         }
 845
 846         if stdinRdr != nil {
 847                 go func() {
 848                         _, err := io.Copy(response.Conn, stdinRdr)
 849                         if err != nil {
 850                                 runner.CrunchLog.Print("While writing stdin collection to docker container %q", err)
 851                                 runner.stop()
 852                         }
 853                         stdinRdr.Close()
 854                         response.CloseWrite()
 855                 }()
 856         } else if len(stdinJson) != 0 {
 857                 go func() {
 858                         _, err := io.Copy(response.Conn, bytes.NewReader(stdinJson))
 859                         if err != nil {
 860                                 runner.CrunchLog.Print("While writing stdin json to docker container %q", err)
 861                                 runner.stop()
 862                         }
 863                         response.CloseWrite()
 864                 }()
 865         }
 866
 867         go runner.ProcessDockerAttach(response.Reader)
 868
 869         return 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() error {
 899         runner.CrunchLog.Print("Creating Docker container")
 900
 901         runner.ContainerConfig.Cmd = runner.Container.Command
 902         if runner.Container.Cwd != "." {
 903                 runner.ContainerConfig.WorkingDir = runner.Container.Cwd
 904         }
 905
 906         for k, v := range runner.Container.Environment {
 907                 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
 908         }
 909
 910         runner.ContainerConfig.Volumes = runner.Volumes
 911
 912         runner.HostConfig = dockercontainer.HostConfig{
 913                 Binds: runner.Binds,
 914                 LogConfig: dockercontainer.LogConfig{
 915                         Type: "none",
 916                 },
 917                 Resources: dockercontainer.Resources{
 918                         CgroupParent: runner.setCgroupParent,
 919                 },
 920         }
 921
 922         if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
 923                 tok, err := runner.ContainerToken()
 924                 if err != nil {
 925                         return err
 926                 }
 927                 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env,
 928                         "ARVADOS_API_TOKEN="+tok,
 929                         "ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"),
 930                         "ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"),
 931                 )
 932                 runner.HostConfig.NetworkMode = dockercontainer.NetworkMode(runner.networkMode)
 933         } else {
 934                 if runner.enableNetwork == "always" {
 935                         runner.HostConfig.NetworkMode = dockercontainer.NetworkMode(runner.networkMode)
 936                 } else {
 937                         runner.HostConfig.NetworkMode = dockercontainer.NetworkMode("none")
 938                 }
 939         }
 940
 941         _, stdinUsed := runner.Container.Mounts["stdin"]
 942         runner.ContainerConfig.OpenStdin = stdinUsed
 943         runner.ContainerConfig.StdinOnce = stdinUsed
 944         runner.ContainerConfig.AttachStdin = stdinUsed
 945         runner.ContainerConfig.AttachStdout = true
 946         runner.ContainerConfig.AttachStderr = true
 947
 948         createdBody, err := runner.Docker.ContainerCreate(context.TODO(), &runner.ContainerConfig, &runner.HostConfig, nil, runner.Container.UUID)
 949         if err != nil {
 950                 return fmt.Errorf("While creating container: %v", err)
 951         }
 952
 953         runner.ContainerID = createdBody.ID
 954
 955         return runner.AttachStreams()
 956 }
 957
 958 // StartContainer starts the docker container created by CreateContainer.
 959 func (runner *ContainerRunner) StartContainer() error {
 960         runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID)
 961         runner.cStateLock.Lock()
 962         defer runner.cStateLock.Unlock()
 963         if runner.cCancelled {
 964                 return ErrCancelled
 965         }
 966         err := runner.Docker.ContainerStart(context.TODO(), runner.ContainerID,
 967                 dockertypes.ContainerStartOptions{})
 968         if err != nil {
 969                 var advice string
 970                 if m, e := regexp.MatchString("(?ms).*(exec|System error).*(no such file or directory|file not found).*", err.Error()); m && e == nil {
 971                         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])
 972                 }
 973                 return fmt.Errorf("could not start container: %v%s", err, advice)
 974         }
 975         return nil
 976 }
 977
 978 // WaitFinish waits for the container to terminate, capture the exit code, and
 979 // close the stdout/stderr logging.
 980 func (runner *ContainerRunner) WaitFinish() error {
 981         runner.CrunchLog.Print("Waiting for container to finish")
 982
 983         waitOk, waitErr := runner.Docker.ContainerWait(context.TODO(), runner.ContainerID, dockercontainer.WaitConditionNotRunning)
 984         arvMountExit := runner.ArvMountExit
 985         for {
 986                 select {
 987                 case waitBody := <-waitOk:
 988                         runner.CrunchLog.Printf("Container exited with code: %v", waitBody.StatusCode)
 989                         code := int(waitBody.StatusCode)
 990                         runner.ExitCode = &code
 991
 992                         // wait for stdout/stderr to complete
 993                         <-runner.loggingDone
 994                         return nil
 995
 996                 case err := <-waitErr:
 997                         return fmt.Errorf("container wait: %v", err)
 998
 999                 case <-arvMountExit:
1000                         runner.CrunchLog.Printf("arv-mount exited while container is still running.  Stopping container.")
1001                         runner.stop()
1002                         // arvMountExit will always be ready now that
1003                         // it's closed, but that doesn't interest us.
1004                         arvMountExit = nil
1005                 }
1006         }
1007 }
1008
1009 var ErrNotInOutputDir = fmt.Errorf("Must point to path within the output directory")
1010
1011 func (runner *ContainerRunner) derefOutputSymlink(path string, startinfo os.FileInfo) (tgt string, readlinktgt string, info os.FileInfo, err error) {
1012         // Follow symlinks if necessary
1013         info = startinfo
1014         tgt = path
1015         readlinktgt = ""
1016         nextlink := path
1017         for followed := 0; info.Mode()&os.ModeSymlink != 0; followed++ {
1018                 if followed >= limitFollowSymlinks {
1019                         // Got stuck in a loop or just a pathological number of links, give up.
1020                         err = fmt.Errorf("Followed more than %v symlinks from path %q", limitFollowSymlinks, path)
1021                         return
1022                 }
1023
1024                 readlinktgt, err = os.Readlink(nextlink)
1025                 if err != nil {
1026                         return
1027                 }
1028
1029                 tgt = readlinktgt
1030                 if !strings.HasPrefix(tgt, "/") {
1031                         // Relative symlink, resolve it to host path
1032                         tgt = filepath.Join(filepath.Dir(path), tgt)
1033                 }
1034                 if strings.HasPrefix(tgt, runner.Container.OutputPath+"/") && !strings.HasPrefix(tgt, runner.HostOutputDir+"/") {
1035                         // Absolute symlink to container output path, adjust it to host output path.
1036                         tgt = filepath.Join(runner.HostOutputDir, tgt[len(runner.Container.OutputPath):])
1037                 }
1038                 if !strings.HasPrefix(tgt, runner.HostOutputDir+"/") {
1039                         // After dereferencing, symlink target must either be
1040                         // within output directory, or must point to a
1041                         // collection mount.
1042                         err = ErrNotInOutputDir
1043                         return
1044                 }
1045
1046                 info, err = os.Lstat(tgt)
1047                 if err != nil {
1048                         // tgt
1049                         err = fmt.Errorf("Symlink in output %q points to invalid location %q: %v",
1050                                 path[len(runner.HostOutputDir):], readlinktgt, err)
1051                         return
1052                 }
1053
1054                 nextlink = tgt
1055         }
1056
1057         return
1058 }
1059
1060 var limitFollowSymlinks = 10
1061
1062 // UploadFile uploads files within the output directory, with special handling
1063 // for symlinks. If the symlink leads to a keep mount, copy the manifest text
1064 // from the keep mount into the output manifestText.  Ensure that whether
1065 // symlinks are relative or absolute, every symlink target (even targets that
1066 // are symlinks themselves) must point to a path in either the output directory
1067 // or a collection mount.
1068 //
1069 // Assumes initial value of "path" is absolute, and located within runner.HostOutputDir.
1070 func (runner *ContainerRunner) UploadOutputFile(
1071         path string,
1072         info os.FileInfo,
1073         infoerr error,
1074         binds []string,
1075         walkUpload *WalkUpload,
1076         relocateFrom string,
1077         relocateTo string,
1078         followed int) (manifestText string, err error) {
1079
1080         if infoerr != nil {
1081                 return "", infoerr
1082         }
1083
1084         if info.Mode().IsDir() {
1085                 // if empty, need to create a .keep file
1086                 dir, direrr := os.Open(path)
1087                 if direrr != nil {
1088                         return "", direrr
1089                 }
1090                 defer dir.Close()
1091                 names, eof := dir.Readdirnames(1)
1092                 if len(names) == 0 && eof == io.EOF && path != runner.HostOutputDir {
1093                         containerPath := runner.OutputPath + path[len(runner.HostOutputDir):]
1094                         for _, bind := range binds {
1095                                 mnt := runner.Container.Mounts[bind]
1096                                 // Check if there is a bind for this
1097                                 // directory, in which case assume we don't need .keep
1098                                 if (containerPath == bind || strings.HasPrefix(containerPath, bind+"/")) && mnt.PortableDataHash != "d41d8cd98f00b204e9800998ecf8427e+0" {
1099                                         return
1100                                 }
1101                         }
1102                         outputSuffix := path[len(runner.HostOutputDir)+1:]
1103                         return fmt.Sprintf("./%v d41d8cd98f00b204e9800998ecf8427e+0 0:0:.keep\n", outputSuffix), nil
1104                 }
1105                 return
1106         }
1107
1108         if followed >= limitFollowSymlinks {
1109                 // Got stuck in a loop or just a pathological number of
1110                 // directory links, give up.
1111                 err = fmt.Errorf("Followed more than %v symlinks from path %q", limitFollowSymlinks, path)
1112                 return
1113         }
1114
1115         // "path" is the actual path we are visiting
1116         // "tgt" is the target of "path" (a non-symlink) after following symlinks
1117         // "relocated" is the path in the output manifest where the file should be placed,
1118         // but has HostOutputDir as a prefix.
1119
1120         // The destination path in the output manifest may need to be
1121         // logically relocated to some other path in order to appear
1122         // in the correct location as a result of following a symlink.
1123         // Remove the relocateFrom prefix and replace it with
1124         // relocateTo.
1125         relocated := relocateTo + path[len(relocateFrom):]
1126
1127         tgt, readlinktgt, info, derefErr := runner.derefOutputSymlink(path, info)
1128         if derefErr != nil && derefErr != ErrNotInOutputDir {
1129                 return "", derefErr
1130         }
1131
1132         // go through mounts and try reverse map to collection reference
1133         for _, bind := range binds {
1134                 mnt := runner.Container.Mounts[bind]
1135                 if tgt == bind || strings.HasPrefix(tgt, bind+"/") {
1136                         // get path relative to bind
1137                         targetSuffix := tgt[len(bind):]
1138
1139                         // Copy mount and adjust the path to add path relative to the bind
1140                         adjustedMount := mnt
1141                         adjustedMount.Path = filepath.Join(adjustedMount.Path, targetSuffix)
1142
1143                         // Terminates in this keep mount, so add the
1144                         // manifest text at appropriate location.
1145                         outputSuffix := relocated[len(runner.HostOutputDir):]
1146                         manifestText, err = runner.getCollectionManifestForPath(adjustedMount, outputSuffix)
1147                         return
1148                 }
1149         }
1150
1151         // If target is not a collection mount, it must be located within the
1152         // output directory, otherwise it is an error.
1153         if derefErr == ErrNotInOutputDir {
1154                 err = fmt.Errorf("Symlink in output %q points to invalid location %q, must point to path within the output directory.",
1155                         path[len(runner.HostOutputDir):], readlinktgt)
1156                 return
1157         }
1158
1159         if info.Mode().IsRegular() {
1160                 return "", walkUpload.UploadFile(relocated, tgt)
1161         }
1162
1163         if info.Mode().IsDir() {
1164                 // Symlink leads to directory.  Walk() doesn't follow
1165                 // directory symlinks, so we walk the target directory
1166                 // instead.  Within the walk, file paths are relocated
1167                 // so they appear under the original symlink path.
1168                 err = filepath.Walk(tgt, func(walkpath string, walkinfo os.FileInfo, walkerr error) error {
1169                         var m string
1170                         m, walkerr = runner.UploadOutputFile(walkpath, walkinfo, walkerr,
1171                                 binds, walkUpload, tgt, relocated, followed+1)
1172                         if walkerr == nil {
1173                                 manifestText = manifestText + m
1174                         }
1175                         return walkerr
1176                 })
1177                 return
1178         }
1179
1180         return
1181 }
1182
1183 // HandleOutput sets the output, unmounts the FUSE mount, and deletes temporary directories
1184 func (runner *ContainerRunner) CaptureOutput() error {
1185         if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
1186                 // Output may have been set directly by the container, so
1187                 // refresh the container record to check.
1188                 err := runner.ArvClient.Get("containers", runner.Container.UUID,
1189                         nil, &runner.Container)
1190                 if err != nil {
1191                         return err
1192                 }
1193                 if runner.Container.Output != "" {
1194                         // Container output is already set.
1195                         runner.OutputPDH = &runner.Container.Output
1196                         return nil
1197                 }
1198         }
1199
1200         if runner.HostOutputDir == "" {
1201                 return nil
1202         }
1203
1204         _, err := os.Stat(runner.HostOutputDir)
1205         if err != nil {
1206                 return fmt.Errorf("While checking host output path: %v", err)
1207         }
1208
1209         // Pre-populate output from the configured mount points
1210         var binds []string
1211         for bind, mnt := range runner.Container.Mounts {
1212                 if mnt.Kind == "collection" {
1213                         binds = append(binds, bind)
1214                 }
1215         }
1216         sort.Strings(binds)
1217
1218         var manifestText string
1219
1220         collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
1221         _, err = os.Stat(collectionMetafile)
1222         if err != nil {
1223                 // Regular directory
1224
1225                 cw := CollectionWriter{0, runner.Kc, nil, nil, sync.Mutex{}}
1226                 walkUpload := cw.BeginUpload(runner.HostOutputDir, runner.CrunchLog.Logger)
1227
1228                 var m string
1229                 err = filepath.Walk(runner.HostOutputDir, func(path string, info os.FileInfo, err error) error {
1230                         m, err = runner.UploadOutputFile(path, info, err, binds, walkUpload, "", "", 0)
1231                         if err == nil {
1232                                 manifestText = manifestText + m
1233                         }
1234                         return err
1235                 })
1236
1237                 cw.EndUpload(walkUpload)
1238
1239                 if err != nil {
1240                         return fmt.Errorf("While uploading output files: %v", err)
1241                 }
1242
1243                 m, err = cw.ManifestText()
1244                 manifestText = manifestText + m
1245                 if err != nil {
1246                         return fmt.Errorf("While uploading output files: %v", err)
1247                 }
1248         } else {
1249                 // FUSE mount directory
1250                 file, openerr := os.Open(collectionMetafile)
1251                 if openerr != nil {
1252                         return fmt.Errorf("While opening FUSE metafile: %v", err)
1253                 }
1254                 defer file.Close()
1255
1256                 var rec arvados.Collection
1257                 err = json.NewDecoder(file).Decode(&rec)
1258                 if err != nil {
1259                         return fmt.Errorf("While reading FUSE metafile: %v", err)
1260                 }
1261                 manifestText = rec.ManifestText
1262         }
1263
1264         for _, bind := range binds {
1265                 mnt := runner.Container.Mounts[bind]
1266
1267                 bindSuffix := strings.TrimPrefix(bind, runner.Container.OutputPath)
1268
1269                 if bindSuffix == bind || len(bindSuffix) <= 0 {
1270                         // either does not start with OutputPath or is OutputPath itself
1271                         continue
1272                 }
1273
1274                 if mnt.ExcludeFromOutput == true {
1275                         continue
1276                 }
1277
1278                 // append to manifest_text
1279                 m, err := runner.getCollectionManifestForPath(mnt, bindSuffix)
1280                 if err != nil {
1281                         return err
1282                 }
1283
1284                 manifestText = manifestText + m
1285         }
1286
1287         // Save output
1288         var response arvados.Collection
1289         manifest := manifest.Manifest{Text: manifestText}
1290         manifestText = manifest.Extract(".", ".").Text
1291         err = runner.ArvClient.Create("collections",
1292                 arvadosclient.Dict{
1293                         "ensure_unique_name": true,
1294                         "collection": arvadosclient.Dict{
1295                                 "is_trashed":    true,
1296                                 "name":          "output for " + runner.Container.UUID,
1297                                 "manifest_text": manifestText}},
1298                 &response)
1299         if err != nil {
1300                 return fmt.Errorf("While creating output collection: %v", err)
1301         }
1302         runner.OutputPDH = &response.PortableDataHash
1303         return nil
1304 }
1305
1306 var outputCollections = make(map[string]arvados.Collection)
1307
1308 // Fetch the collection for the mnt.PortableDataHash
1309 // Return the manifest_text fragment corresponding to the specified mnt.Path
1310 //  after making any required updates.
1311 //  Ex:
1312 //    If mnt.Path is not specified,
1313 //      return the entire manifest_text after replacing any "." with bindSuffix
1314 //    If mnt.Path corresponds to one stream,
1315 //      return the manifest_text for that stream after replacing that stream name with bindSuffix
1316 //    Otherwise, check if a filename in any one stream is being sought. Return the manifest_text
1317 //      for that stream after replacing stream name with bindSuffix minus the last word
1318 //      and the file name with last word of the bindSuffix
1319 //  Allowed path examples:
1320 //    "path":"/"
1321 //    "path":"/subdir1"
1322 //    "path":"/subdir1/subdir2"
1323 //    "path":"/subdir/filename" etc
1324 func (runner *ContainerRunner) getCollectionManifestForPath(mnt arvados.Mount, bindSuffix string) (string, error) {
1325         collection := outputCollections[mnt.PortableDataHash]
1326         if collection.PortableDataHash == "" {
1327                 err := runner.ArvClient.Get("collections", mnt.PortableDataHash, nil, &collection)
1328                 if err != nil {
1329                         return "", fmt.Errorf("While getting collection for %v: %v", mnt.PortableDataHash, err)
1330                 }
1331                 outputCollections[mnt.PortableDataHash] = collection
1332         }
1333
1334         if collection.ManifestText == "" {
1335                 runner.CrunchLog.Printf("No manifest text for collection %v", collection.PortableDataHash)
1336                 return "", nil
1337         }
1338
1339         mft := manifest.Manifest{Text: collection.ManifestText}
1340         extracted := mft.Extract(mnt.Path, bindSuffix)
1341         if extracted.Err != nil {
1342                 return "", fmt.Errorf("Error parsing manifest for %v: %v", mnt.PortableDataHash, extracted.Err.Error())
1343         }
1344         return extracted.Text, nil
1345 }
1346
1347 func (runner *ContainerRunner) CleanupDirs() {
1348         if runner.ArvMount != nil {
1349                 var delay int64 = 8
1350                 umount := exec.Command("arv-mount", fmt.Sprintf("--unmount-timeout=%d", delay), "--unmount", runner.ArvMountPoint)
1351                 umount.Stdout = runner.CrunchLog
1352                 umount.Stderr = runner.CrunchLog
1353                 runner.CrunchLog.Printf("Running %v", umount.Args)
1354                 umnterr := umount.Start()
1355
1356                 if umnterr != nil {
1357                         runner.CrunchLog.Printf("Error unmounting: %v", umnterr)
1358                 } else {
1359                         // If arv-mount --unmount gets stuck for any reason, we
1360                         // don't want to wait for it forever.  Do Wait() in a goroutine
1361                         // so it doesn't block crunch-run.
1362                         umountExit := make(chan error)
1363                         go func() {
1364                                 mnterr := umount.Wait()
1365                                 if mnterr != nil {
1366                                         runner.CrunchLog.Printf("Error unmounting: %v", mnterr)
1367                                 }
1368                                 umountExit <- mnterr
1369                         }()
1370
1371                         for again := true; again; {
1372                                 again = false
1373                                 select {
1374                                 case <-umountExit:
1375                                         umount = nil
1376                                         again = true
1377                                 case <-runner.ArvMountExit:
1378                                         break
1379                                 case <-time.After(time.Duration((delay + 1) * int64(time.Second))):
1380                                         runner.CrunchLog.Printf("Timed out waiting for unmount")
1381                                         if umount != nil {
1382                                                 umount.Process.Kill()
1383                                         }
1384                                         runner.ArvMount.Process.Kill()
1385                                 }
1386                         }
1387                 }
1388         }
1389
1390         if runner.ArvMountPoint != "" {
1391                 if rmerr := os.Remove(runner.ArvMountPoint); rmerr != nil {
1392                         runner.CrunchLog.Printf("While cleaning up arv-mount directory %s: %v", runner.ArvMountPoint, rmerr)
1393                 }
1394         }
1395
1396         for _, tmpdir := range runner.CleanupTempDir {
1397                 if rmerr := os.RemoveAll(tmpdir); rmerr != nil {
1398                         runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
1399                 }
1400         }
1401 }
1402
1403 // CommitLogs posts the collection containing the final container logs.
1404 func (runner *ContainerRunner) CommitLogs() error {
1405         runner.CrunchLog.Print(runner.finalState)
1406
1407         if runner.arvMountLog != nil {
1408                 runner.arvMountLog.Close()
1409         }
1410         runner.CrunchLog.Close()
1411
1412         // Closing CrunchLog above allows them to be committed to Keep at this
1413         // point, but re-open crunch log with ArvClient in case there are any
1414         // other further errors (such as failing to write the log to Keep!)
1415         // while shutting down
1416         runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{ArvClient: runner.ArvClient,
1417                 UUID: runner.Container.UUID, loggingStream: "crunch-run", writeCloser: nil})
1418         runner.CrunchLog.Immediate = log.New(os.Stderr, runner.Container.UUID+" ", 0)
1419
1420         if runner.LogsPDH != nil {
1421                 // If we have already assigned something to LogsPDH,
1422                 // we must be closing the re-opened log, which won't
1423                 // end up getting attached to the container record and
1424                 // therefore doesn't need to be saved as a collection
1425                 // -- it exists only to send logs to other channels.
1426                 return nil
1427         }
1428
1429         mt, err := runner.LogCollection.ManifestText()
1430         if err != nil {
1431                 return fmt.Errorf("While creating log manifest: %v", err)
1432         }
1433
1434         var response arvados.Collection
1435         err = runner.ArvClient.Create("collections",
1436                 arvadosclient.Dict{
1437                         "ensure_unique_name": true,
1438                         "collection": arvadosclient.Dict{
1439                                 "is_trashed":    true,
1440                                 "name":          "logs for " + runner.Container.UUID,
1441                                 "manifest_text": mt}},
1442                 &response)
1443         if err != nil {
1444                 return fmt.Errorf("While creating log collection: %v", err)
1445         }
1446         runner.LogsPDH = &response.PortableDataHash
1447         return nil
1448 }
1449
1450 // UpdateContainerRunning updates the container state to "Running"
1451 func (runner *ContainerRunner) UpdateContainerRunning() error {
1452         runner.cStateLock.Lock()
1453         defer runner.cStateLock.Unlock()
1454         if runner.cCancelled {
1455                 return ErrCancelled
1456         }
1457         return runner.ArvClient.Update("containers", runner.Container.UUID,
1458                 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
1459 }
1460
1461 // ContainerToken returns the api_token the container (and any
1462 // arv-mount processes) are allowed to use.
1463 func (runner *ContainerRunner) ContainerToken() (string, error) {
1464         if runner.token != "" {
1465                 return runner.token, nil
1466         }
1467
1468         var auth arvados.APIClientAuthorization
1469         err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
1470         if err != nil {
1471                 return "", err
1472         }
1473         runner.token = auth.APIToken
1474         return runner.token, nil
1475 }
1476
1477 // UpdateContainerComplete updates the container record state on API
1478 // server to "Complete" or "Cancelled"
1479 func (runner *ContainerRunner) UpdateContainerFinal() error {
1480         update := arvadosclient.Dict{}
1481         update["state"] = runner.finalState
1482         if runner.LogsPDH != nil {
1483                 update["log"] = *runner.LogsPDH
1484         }
1485         if runner.finalState == "Complete" {
1486                 if runner.ExitCode != nil {
1487                         update["exit_code"] = *runner.ExitCode
1488                 }
1489                 if runner.OutputPDH != nil {
1490                         update["output"] = *runner.OutputPDH
1491                 }
1492         }
1493         return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
1494 }
1495
1496 // IsCancelled returns the value of Cancelled, with goroutine safety.
1497 func (runner *ContainerRunner) IsCancelled() bool {
1498         runner.cStateLock.Lock()
1499         defer runner.cStateLock.Unlock()
1500         return runner.cCancelled
1501 }
1502
1503 // NewArvLogWriter creates an ArvLogWriter
1504 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
1505         return &ArvLogWriter{
1506                 ArvClient:     runner.ArvClient,
1507                 UUID:          runner.Container.UUID,
1508                 loggingStream: name,
1509                 writeCloser:   runner.LogCollection.Open(name + ".txt")}
1510 }
1511
1512 // Run the full container lifecycle.
1513 func (runner *ContainerRunner) Run() (err error) {
1514         runner.CrunchLog.Printf("crunch-run %s started", version)
1515         runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
1516
1517         hostname, hosterr := os.Hostname()
1518         if hosterr != nil {
1519                 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
1520         } else {
1521                 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
1522         }
1523
1524         runner.finalState = "Queued"
1525
1526         defer func() {
1527                 runner.stopSignals()
1528                 runner.CleanupDirs()
1529
1530                 runner.CrunchLog.Printf("crunch-run finished")
1531                 runner.CrunchLog.Close()
1532         }()
1533
1534         defer func() {
1535                 // checkErr prints e (unless it's nil) and sets err to
1536                 // e (unless err is already non-nil). Thus, if err
1537                 // hasn't already been assigned when Run() returns,
1538                 // this cleanup func will cause Run() to return the
1539                 // first non-nil error that is passed to checkErr().
1540                 checkErr := func(e error) {
1541                         if e == nil {
1542                                 return
1543                         }
1544                         runner.CrunchLog.Print(e)
1545                         if err == nil {
1546                                 err = e
1547                         }
1548                         if runner.finalState == "Complete" {
1549                                 // There was an error in the finalization.
1550                                 runner.finalState = "Cancelled"
1551                         }
1552                 }
1553
1554                 // Log the error encountered in Run(), if any
1555                 checkErr(err)
1556
1557                 if runner.finalState == "Queued" {
1558                         runner.UpdateContainerFinal()
1559                         return
1560                 }
1561
1562                 if runner.IsCancelled() {
1563                         runner.finalState = "Cancelled"
1564                         // but don't return yet -- we still want to
1565                         // capture partial output and write logs
1566                 }
1567
1568                 checkErr(runner.CaptureOutput())
1569                 checkErr(runner.CommitLogs())
1570                 checkErr(runner.UpdateContainerFinal())
1571         }()
1572
1573         err = runner.fetchContainerRecord()
1574         if err != nil {
1575                 return
1576         }
1577
1578         // setup signal handling
1579         runner.setupSignals()
1580
1581         // check for and/or load image
1582         err = runner.LoadImage()
1583         if err != nil {
1584                 if !runner.checkBrokenNode(err) {
1585                         // Failed to load image but not due to a "broken node"
1586                         // condition, probably user error.
1587                         runner.finalState = "Cancelled"
1588                 }
1589                 err = fmt.Errorf("While loading container image: %v", err)
1590                 return
1591         }
1592
1593         // set up FUSE mount and binds
1594         err = runner.SetupMounts()
1595         if err != nil {
1596                 runner.finalState = "Cancelled"
1597                 err = fmt.Errorf("While setting up mounts: %v", err)
1598                 return
1599         }
1600
1601         err = runner.CreateContainer()
1602         if err != nil {
1603                 return
1604         }
1605         err = runner.LogHostInfo()
1606         if err != nil {
1607                 return
1608         }
1609         err = runner.LogNodeRecord()
1610         if err != nil {
1611                 return
1612         }
1613         err = runner.LogContainerRecord()
1614         if err != nil {
1615                 return
1616         }
1617
1618         if runner.IsCancelled() {
1619                 return
1620         }
1621
1622         err = runner.UpdateContainerRunning()
1623         if err != nil {
1624                 return
1625         }
1626         runner.finalState = "Cancelled"
1627
1628         runner.StartCrunchstat()
1629
1630         err = runner.StartContainer()
1631         if err != nil {
1632                 runner.checkBrokenNode(err)
1633                 return
1634         }
1635
1636         err = runner.WaitFinish()
1637         if err == nil && !runner.IsCancelled() {
1638                 runner.finalState = "Complete"
1639         }
1640         return
1641 }
1642
1643 // Fetch the current container record (uuid = runner.Container.UUID)
1644 // into runner.Container.
1645 func (runner *ContainerRunner) fetchContainerRecord() error {
1646         reader, err := runner.ArvClient.CallRaw("GET", "containers", runner.Container.UUID, "", nil)
1647         if err != nil {
1648                 return fmt.Errorf("error fetching container record: %v", err)
1649         }
1650         defer reader.Close()
1651
1652         dec := json.NewDecoder(reader)
1653         dec.UseNumber()
1654         err = dec.Decode(&runner.Container)
1655         if err != nil {
1656                 return fmt.Errorf("error decoding container record: %v", err)
1657         }
1658         return nil
1659 }
1660
1661 // NewContainerRunner creates a new container runner.
1662 func NewContainerRunner(api IArvadosClient,
1663         kc IKeepClient,
1664         docker ThinDockerClient,
1665         containerUUID string) *ContainerRunner {
1666
1667         cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
1668         cr.NewLogWriter = cr.NewArvLogWriter
1669         cr.RunArvMount = cr.ArvMountCmd
1670         cr.MkTempDir = ioutil.TempDir
1671         cr.LogCollection = &CollectionWriter{0, kc, nil, nil, sync.Mutex{}}
1672         cr.Container.UUID = containerUUID
1673         cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
1674         cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
1675
1676         loadLogThrottleParams(api)
1677
1678         return cr
1679 }
1680
1681 func main() {
1682         statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
1683         cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
1684         cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
1685         cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
1686         caCertsPath := flag.String("ca-certs", "", "Path to TLS root certificates")
1687         enableNetwork := flag.String("container-enable-networking", "default",
1688                 `Specify if networking should be enabled for container.  One of 'default', 'always':
1689         default: only enable networking if container requests it.
1690         always:  containers always have networking enabled
1691         `)
1692         networkMode := flag.String("container-network-mode", "default",
1693                 `Set networking mode for container.  Corresponds to Docker network mode (--net).
1694         `)
1695         memprofile := flag.String("memprofile", "", "write memory profile to `file` after running container")
1696         getVersion := flag.Bool("version", false, "Print version information and exit.")
1697         flag.Parse()
1698
1699         // Print version information if requested
1700         if *getVersion {
1701                 fmt.Printf("crunch-run %s\n", version)
1702                 return
1703         }
1704
1705         log.Printf("crunch-run %s started", version)
1706
1707         containerId := flag.Arg(0)
1708
1709         if *caCertsPath != "" {
1710                 arvadosclient.CertFiles = []string{*caCertsPath}
1711         }
1712
1713         api, err := arvadosclient.MakeArvadosClient()
1714         if err != nil {
1715                 log.Fatalf("%s: %v", containerId, err)
1716         }
1717         api.Retries = 8
1718
1719         kc, kcerr := keepclient.MakeKeepClient(api)
1720         if kcerr != nil {
1721                 log.Fatalf("%s: %v", containerId, kcerr)
1722         }
1723         kc.BlockCache = &keepclient.BlockCache{MaxBlocks: 2}
1724         kc.Retries = 4
1725
1726         // API version 1.21 corresponds to Docker 1.9, which is currently the
1727         // minimum version we want to support.
1728         docker, dockererr := dockerclient.NewClient(dockerclient.DefaultDockerHost, "1.21", nil, nil)
1729
1730         cr := NewContainerRunner(api, kc, docker, containerId)
1731         if dockererr != nil {
1732                 cr.CrunchLog.Printf("%s: %v", containerId, dockererr)
1733                 cr.checkBrokenNode(dockererr)
1734                 cr.CrunchLog.Close()
1735                 os.Exit(1)
1736         }
1737
1738         cr.statInterval = *statInterval
1739         cr.cgroupRoot = *cgroupRoot
1740         cr.expectCgroupParent = *cgroupParent
1741         cr.enableNetwork = *enableNetwork
1742         cr.networkMode = *networkMode
1743         if *cgroupParentSubsystem != "" {
1744                 p := findCgroup(*cgroupParentSubsystem)
1745                 cr.setCgroupParent = p
1746                 cr.expectCgroupParent = p
1747         }
1748
1749         runerr := cr.Run()
1750
1751         if *memprofile != "" {
1752                 f, err := os.Create(*memprofile)
1753                 if err != nil {
1754                         log.Printf("could not create memory profile: ", err)
1755                 }
1756                 runtime.GC() // get up-to-date statistics
1757                 if err := pprof.WriteHeapProfile(f); err != nil {
1758                         log.Printf("could not write memory profile: ", err)
1759                 }
1760                 closeerr := f.Close()
1761                 if closeerr != nil {
1762                         log.Printf("closing memprofile file: ", err)
1763                 }
1764         }
1765
1766         if runerr != nil {
1767                 log.Fatalf("%s: %v", containerId, runerr)
1768         }
1769 }