services/crunch-run/crunchrun.go

   1 package main
   2
   3 import (
   4         "encoding/json"
   5         "errors"
   6         "flag"
   7         "fmt"
   8         "git.curoverse.com/arvados.git/lib/crunchstat"
   9         "git.curoverse.com/arvados.git/sdk/go/arvados"
  10         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  11         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  12         "git.curoverse.com/arvados.git/sdk/go/manifest"
  13         "github.com/curoverse/dockerclient"
  14         "io"
  15         "io/ioutil"
  16         "log"
  17         "os"
  18         "os/exec"
  19         "os/signal"
  20         "path"
  21         "path/filepath"
  22         "sort"
  23         "strings"
  24         "sync"
  25         "syscall"
  26         "time"
  27 )
  28
  29 // IArvadosClient is the minimal Arvados API methods used by crunch-run.
  30 type IArvadosClient interface {
  31         Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
  32         Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
  33         Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
  34         Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error
  35         Discovery(key string) (interface{}, error)
  36 }
  37
  38 // ErrCancelled is the error returned when the container is cancelled.
  39 var ErrCancelled = errors.New("Cancelled")
  40
  41 // IKeepClient is the minimal Keep API methods used by crunch-run.
  42 type IKeepClient interface {
  43         PutHB(hash string, buf []byte) (string, int, error)
  44         ManifestFileReader(m manifest.Manifest, filename string) (keepclient.Reader, error)
  45 }
  46
  47 // NewLogWriter is a factory function to create a new log writer.
  48 type NewLogWriter func(name string) io.WriteCloser
  49
  50 type RunArvMount func(args []string, tok string) (*exec.Cmd, error)
  51
  52 type MkTempDir func(string, string) (string, error)
  53
  54 // ThinDockerClient is the minimal Docker client interface used by crunch-run.
  55 type ThinDockerClient interface {
  56         StopContainer(id string, timeout int) error
  57         InspectImage(id string) (*dockerclient.ImageInfo, error)
  58         LoadImage(reader io.Reader) error
  59         CreateContainer(config *dockerclient.ContainerConfig, name string, authConfig *dockerclient.AuthConfig) (string, error)
  60         StartContainer(id string, config *dockerclient.HostConfig) error
  61         AttachContainer(id string, options *dockerclient.AttachOptions) (io.ReadCloser, error)
  62         Wait(id string) <-chan dockerclient.WaitResult
  63         RemoveImage(name string, force bool) ([]*dockerclient.ImageDelete, error)
  64 }
  65
  66 // ContainerRunner is the main stateful struct used for a single execution of a
  67 // container.
  68 type ContainerRunner struct {
  69         Docker    ThinDockerClient
  70         ArvClient IArvadosClient
  71         Kc        IKeepClient
  72         arvados.Container
  73         dockerclient.ContainerConfig
  74         dockerclient.HostConfig
  75         token       string
  76         ContainerID string
  77         ExitCode    *int
  78         NewLogWriter
  79         loggingDone   chan bool
  80         CrunchLog     *ThrottledLogger
  81         Stdout        io.WriteCloser
  82         Stderr        *ThrottledLogger
  83         LogCollection *CollectionWriter
  84         LogsPDH       *string
  85         RunArvMount
  86         MkTempDir
  87         ArvMount       *exec.Cmd
  88         ArvMountPoint  string
  89         HostOutputDir  string
  90         CleanupTempDir []string
  91         Binds          []string
  92         OutputPDH      *string
  93         CancelLock     sync.Mutex
  94         Cancelled      bool
  95         SigChan        chan os.Signal
  96         ArvMountExit   chan error
  97         finalState     string
  98         trashLifetime  time.Duration
  99
 100         statLogger   io.WriteCloser
 101         statReporter *crunchstat.Reporter
 102         statInterval time.Duration
 103         cgroupRoot   string
 104         // What we expect the container's cgroup parent to be.
 105         expectCgroupParent string
 106         // What we tell docker to use as the container's cgroup
 107         // parent. Note: Ideally we would use the same field for both
 108         // expectCgroupParent and setCgroupParent, and just make it
 109         // default to "docker". However, when using docker < 1.10 with
 110         // systemd, specifying a non-empty cgroup parent (even the
 111         // default value "docker") hits a docker bug
 112         // (https://github.com/docker/docker/issues/17126). Using two
 113         // separate fields makes it possible to use the "expect cgroup
 114         // parent to be X" feature even on sites where the "specify
 115         // cgroup parent" feature breaks.
 116         setCgroupParent string
 117 }
 118
 119 // SetupSignals sets up signal handling to gracefully terminate the underlying
 120 // Docker container and update state when receiving a TERM, INT or QUIT signal.
 121 func (runner *ContainerRunner) SetupSignals() {
 122         runner.SigChan = make(chan os.Signal, 1)
 123         signal.Notify(runner.SigChan, syscall.SIGTERM)
 124         signal.Notify(runner.SigChan, syscall.SIGINT)
 125         signal.Notify(runner.SigChan, syscall.SIGQUIT)
 126
 127         go func(sig <-chan os.Signal) {
 128                 for range sig {
 129                         if !runner.Cancelled {
 130                                 runner.CancelLock.Lock()
 131                                 runner.Cancelled = true
 132                                 if runner.ContainerID != "" {
 133                                         runner.Docker.StopContainer(runner.ContainerID, 10)
 134                                 }
 135                                 runner.CancelLock.Unlock()
 136                         }
 137                 }
 138         }(runner.SigChan)
 139 }
 140
 141 // LoadImage determines the docker image id from the container record and
 142 // checks if it is available in the local Docker image store.  If not, it loads
 143 // the image from Keep.
 144 func (runner *ContainerRunner) LoadImage() (err error) {
 145
 146         runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.Container.ContainerImage)
 147
 148         var collection arvados.Collection
 149         err = runner.ArvClient.Get("collections", runner.Container.ContainerImage, nil, &collection)
 150         if err != nil {
 151                 return fmt.Errorf("While getting container image collection: %v", err)
 152         }
 153         manifest := manifest.Manifest{Text: collection.ManifestText}
 154         var img, imageID string
 155         for ms := range manifest.StreamIter() {
 156                 img = ms.FileStreamSegments[0].Name
 157                 if !strings.HasSuffix(img, ".tar") {
 158                         return fmt.Errorf("First file in the container image collection does not end in .tar")
 159                 }
 160                 imageID = img[:len(img)-4]
 161         }
 162
 163         runner.CrunchLog.Printf("Using Docker image id '%s'", imageID)
 164
 165         _, err = runner.Docker.InspectImage(imageID)
 166         if err != nil {
 167                 runner.CrunchLog.Print("Loading Docker image from keep")
 168
 169                 var readCloser io.ReadCloser
 170                 readCloser, err = runner.Kc.ManifestFileReader(manifest, img)
 171                 if err != nil {
 172                         return fmt.Errorf("While creating ManifestFileReader for container image: %v", err)
 173                 }
 174
 175                 err = runner.Docker.LoadImage(readCloser)
 176                 if err != nil {
 177                         return fmt.Errorf("While loading container image into Docker: %v", err)
 178                 }
 179         } else {
 180                 runner.CrunchLog.Print("Docker image is available")
 181         }
 182
 183         runner.ContainerConfig.Image = imageID
 184
 185         return nil
 186 }
 187
 188 func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string, token string) (c *exec.Cmd, err error) {
 189         c = exec.Command("arv-mount", arvMountCmd...)
 190
 191         // Copy our environment, but override ARVADOS_API_TOKEN with
 192         // the container auth token.
 193         c.Env = nil
 194         for _, s := range os.Environ() {
 195                 if !strings.HasPrefix(s, "ARVADOS_API_TOKEN=") {
 196                         c.Env = append(c.Env, s)
 197                 }
 198         }
 199         c.Env = append(c.Env, "ARVADOS_API_TOKEN="+token)
 200
 201         nt := NewThrottledLogger(runner.NewLogWriter("arv-mount"))
 202         c.Stdout = nt
 203         c.Stderr = nt
 204
 205         err = c.Start()
 206         if err != nil {
 207                 return nil, err
 208         }
 209
 210         statReadme := make(chan bool)
 211         runner.ArvMountExit = make(chan error)
 212
 213         keepStatting := true
 214         go func() {
 215                 for keepStatting {
 216                         time.Sleep(100 * time.Millisecond)
 217                         _, err = os.Stat(fmt.Sprintf("%s/by_id/README", runner.ArvMountPoint))
 218                         if err == nil {
 219                                 keepStatting = false
 220                                 statReadme <- true
 221                         }
 222                 }
 223                 close(statReadme)
 224         }()
 225
 226         go func() {
 227                 runner.ArvMountExit <- c.Wait()
 228                 close(runner.ArvMountExit)
 229         }()
 230
 231         select {
 232         case <-statReadme:
 233                 break
 234         case err := <-runner.ArvMountExit:
 235                 runner.ArvMount = nil
 236                 keepStatting = false
 237                 return nil, err
 238         }
 239
 240         return c, nil
 241 }
 242
 243 func (runner *ContainerRunner) SetupArvMountPoint(prefix string) (err error) {
 244         if runner.ArvMountPoint == "" {
 245                 runner.ArvMountPoint, err = runner.MkTempDir("", prefix)
 246         }
 247         return
 248 }
 249
 250 func (runner *ContainerRunner) SetupMounts() (err error) {
 251         err = runner.SetupArvMountPoint("keep")
 252         if err != nil {
 253                 return fmt.Errorf("While creating keep mount temp dir: %v", err)
 254         }
 255
 256         runner.CleanupTempDir = append(runner.CleanupTempDir, runner.ArvMountPoint)
 257
 258         pdhOnly := true
 259         tmpcount := 0
 260         arvMountCmd := []string{"--foreground", "--allow-other", "--read-write"}
 261
 262         if runner.Container.RuntimeConstraints.KeepCacheRAM > 0 {
 263                 arvMountCmd = append(arvMountCmd, "--file-cache", fmt.Sprintf("%d", runner.Container.RuntimeConstraints.KeepCacheRAM))
 264         }
 265
 266         collectionPaths := []string{}
 267         runner.Binds = nil
 268         needCertMount := true
 269
 270         var binds []string
 271         for bind, _ := range runner.Container.Mounts {
 272                 binds = append(binds, bind)
 273         }
 274         sort.Strings(binds)
 275
 276         for _, bind := range binds {
 277                 mnt := runner.Container.Mounts[bind]
 278                 if bind == "stdout" {
 279                         // Is it a "file" mount kind?
 280                         if mnt.Kind != "file" {
 281                                 return fmt.Errorf("Unsupported mount kind '%s' for stdout. Only 'file' is supported.", mnt.Kind)
 282                         }
 283
 284                         // Does path start with OutputPath?
 285                         prefix := runner.Container.OutputPath
 286                         if !strings.HasSuffix(prefix, "/") {
 287                                 prefix += "/"
 288                         }
 289                         if !strings.HasPrefix(mnt.Path, prefix) {
 290                                 return fmt.Errorf("Stdout path does not start with OutputPath: %s, %s", mnt.Path, prefix)
 291                         }
 292                 }
 293
 294                 if bind == "/etc/arvados/ca-certificates.crt" {
 295                         needCertMount = false
 296                 }
 297
 298                 if strings.HasPrefix(bind, runner.Container.OutputPath+"/") && bind != runner.Container.OutputPath+"/" {
 299                         if mnt.Kind != "collection" {
 300                                 return fmt.Errorf("Only mount points of kind 'collection' are supported underneath the output_path: %v", bind)
 301                         }
 302                 }
 303
 304                 switch {
 305                 case mnt.Kind == "collection":
 306                         var src string
 307                         if mnt.UUID != "" && mnt.PortableDataHash != "" {
 308                                 return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount")
 309                         }
 310                         if mnt.UUID != "" {
 311                                 if mnt.Writable {
 312                                         return fmt.Errorf("Writing to existing collections currently not permitted.")
 313                                 }
 314                                 pdhOnly = false
 315                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID)
 316                         } else if mnt.PortableDataHash != "" {
 317                                 if mnt.Writable {
 318                                         return fmt.Errorf("Can never write to a collection specified by portable data hash")
 319                                 }
 320                                 src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
 321                         } else {
 322                                 src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount)
 323                                 arvMountCmd = append(arvMountCmd, "--mount-tmp")
 324                                 arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%d", tmpcount))
 325                                 tmpcount += 1
 326                         }
 327                         if mnt.Writable {
 328                                 if bind == runner.Container.OutputPath {
 329                                         runner.HostOutputDir = src
 330                                 } else if strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
 331                                         return fmt.Errorf("Writable mount points are not permitted underneath the output_path: %v", bind)
 332                                 }
 333                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
 334                         } else {
 335                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
 336                         }
 337                         collectionPaths = append(collectionPaths, src)
 338
 339                 case mnt.Kind == "tmp" && bind == runner.Container.OutputPath:
 340                         runner.HostOutputDir, err = runner.MkTempDir("", "")
 341                         if err != nil {
 342                                 return fmt.Errorf("While creating mount temp dir: %v", err)
 343                         }
 344                         st, staterr := os.Stat(runner.HostOutputDir)
 345                         if staterr != nil {
 346                                 return fmt.Errorf("While Stat on temp dir: %v", staterr)
 347                         }
 348                         err = os.Chmod(runner.HostOutputDir, st.Mode()|os.ModeSetgid|0777)
 349                         if staterr != nil {
 350                                 return fmt.Errorf("While Chmod temp dir: %v", err)
 351                         }
 352                         runner.CleanupTempDir = append(runner.CleanupTempDir, runner.HostOutputDir)
 353                         runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind))
 354
 355                 case mnt.Kind == "tmp":
 356                         runner.Binds = append(runner.Binds, bind)
 357
 358                 case mnt.Kind == "json":
 359                         jsondata, err := json.Marshal(mnt.Content)
 360                         if err != nil {
 361                                 return fmt.Errorf("encoding json data: %v", err)
 362                         }
 363                         // Create a tempdir with a single file
 364                         // (instead of just a tempfile): this way we
 365                         // can ensure the file is world-readable
 366                         // inside the container, without having to
 367                         // make it world-readable on the docker host.
 368                         tmpdir, err := runner.MkTempDir("", "")
 369                         if err != nil {
 370                                 return fmt.Errorf("creating temp dir: %v", err)
 371                         }
 372                         runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
 373                         tmpfn := filepath.Join(tmpdir, "mountdata.json")
 374                         err = ioutil.WriteFile(tmpfn, jsondata, 0644)
 375                         if err != nil {
 376                                 return fmt.Errorf("writing temp file: %v", err)
 377                         }
 378                         runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind))
 379                 }
 380         }
 381
 382         if runner.HostOutputDir == "" {
 383                 return fmt.Errorf("Output path does not correspond to a writable mount point")
 384         }
 385
 386         if wantAPI := runner.Container.RuntimeConstraints.API; needCertMount && wantAPI != nil && *wantAPI {
 387                 for _, certfile := range arvadosclient.CertFiles {
 388                         _, err := os.Stat(certfile)
 389                         if err == nil {
 390                                 runner.Binds = append(runner.Binds, fmt.Sprintf("%s:/etc/arvados/ca-certificates.crt:ro", certfile))
 391                                 break
 392                         }
 393                 }
 394         }
 395
 396         if pdhOnly {
 397                 arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
 398         } else {
 399                 arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id")
 400         }
 401         arvMountCmd = append(arvMountCmd, runner.ArvMountPoint)
 402
 403         token, err := runner.ContainerToken()
 404         if err != nil {
 405                 return fmt.Errorf("could not get container token: %s", err)
 406         }
 407
 408         runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token)
 409         if err != nil {
 410                 return fmt.Errorf("While trying to start arv-mount: %v", err)
 411         }
 412
 413         for _, p := range collectionPaths {
 414                 _, err = os.Stat(p)
 415                 if err != nil {
 416                         return fmt.Errorf("While checking that input files exist: %v", err)
 417                 }
 418         }
 419
 420         return nil
 421 }
 422
 423 func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
 424         // Handle docker log protocol
 425         // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
 426
 427         header := make([]byte, 8)
 428         for {
 429                 _, readerr := io.ReadAtLeast(containerReader, header, 8)
 430
 431                 if readerr == nil {
 432                         readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
 433                         if header[0] == 1 {
 434                                 // stdout
 435                                 _, readerr = io.CopyN(runner.Stdout, containerReader, readsize)
 436                         } else {
 437                                 // stderr
 438                                 _, readerr = io.CopyN(runner.Stderr, containerReader, readsize)
 439                         }
 440                 }
 441
 442                 if readerr != nil {
 443                         if readerr != io.EOF {
 444                                 runner.CrunchLog.Printf("While reading docker logs: %v", readerr)
 445                         }
 446
 447                         closeerr := runner.Stdout.Close()
 448                         if closeerr != nil {
 449                                 runner.CrunchLog.Printf("While closing stdout logs: %v", closeerr)
 450                         }
 451
 452                         closeerr = runner.Stderr.Close()
 453                         if closeerr != nil {
 454                                 runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr)
 455                         }
 456
 457                         if runner.statReporter != nil {
 458                                 runner.statReporter.Stop()
 459                                 closeerr = runner.statLogger.Close()
 460                                 if closeerr != nil {
 461                                         runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr)
 462                                 }
 463                         }
 464
 465                         runner.loggingDone <- true
 466                         close(runner.loggingDone)
 467                         return
 468                 }
 469         }
 470 }
 471
 472 func (runner *ContainerRunner) StartCrunchstat() {
 473         runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat"))
 474         runner.statReporter = &crunchstat.Reporter{
 475                 CID:          runner.ContainerID,
 476                 Logger:       log.New(runner.statLogger, "", 0),
 477                 CgroupParent: runner.expectCgroupParent,
 478                 CgroupRoot:   runner.cgroupRoot,
 479                 PollPeriod:   runner.statInterval,
 480         }
 481         runner.statReporter.Start()
 482 }
 483
 484 // AttachLogs connects the docker container stdout and stderr logs to the
 485 // Arvados logger which logs to Keep and the API server logs table.
 486 func (runner *ContainerRunner) AttachStreams() (err error) {
 487
 488         runner.CrunchLog.Print("Attaching container streams")
 489
 490         var containerReader io.Reader
 491         containerReader, err = runner.Docker.AttachContainer(runner.ContainerID,
 492                 &dockerclient.AttachOptions{Stream: true, Stdout: true, Stderr: true})
 493         if err != nil {
 494                 return fmt.Errorf("While attaching container stdout/stderr streams: %v", err)
 495         }
 496
 497         runner.loggingDone = make(chan bool)
 498
 499         if stdoutMnt, ok := runner.Container.Mounts["stdout"]; ok {
 500                 stdoutPath := stdoutMnt.Path[len(runner.Container.OutputPath):]
 501                 index := strings.LastIndex(stdoutPath, "/")
 502                 if index > 0 {
 503                         subdirs := stdoutPath[:index]
 504                         if subdirs != "" {
 505                                 st, err := os.Stat(runner.HostOutputDir)
 506                                 if err != nil {
 507                                         return fmt.Errorf("While Stat on temp dir: %v", err)
 508                                 }
 509                                 stdoutPath := path.Join(runner.HostOutputDir, subdirs)
 510                                 err = os.MkdirAll(stdoutPath, st.Mode()|os.ModeSetgid|0777)
 511                                 if err != nil {
 512                                         return fmt.Errorf("While MkdirAll %q: %v", stdoutPath, err)
 513                                 }
 514                         }
 515                 }
 516                 stdoutFile, err := os.Create(path.Join(runner.HostOutputDir, stdoutPath))
 517                 if err != nil {
 518                         return fmt.Errorf("While creating stdout file: %v", err)
 519                 }
 520                 runner.Stdout = stdoutFile
 521         } else {
 522                 runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout"))
 523         }
 524         runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr"))
 525
 526         go runner.ProcessDockerAttach(containerReader)
 527
 528         return nil
 529 }
 530
 531 // CreateContainer creates the docker container.
 532 func (runner *ContainerRunner) CreateContainer() error {
 533         runner.CrunchLog.Print("Creating Docker container")
 534
 535         runner.ContainerConfig.Cmd = runner.Container.Command
 536         if runner.Container.Cwd != "." {
 537                 runner.ContainerConfig.WorkingDir = runner.Container.Cwd
 538         }
 539
 540         for k, v := range runner.Container.Environment {
 541                 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
 542         }
 543         if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
 544                 tok, err := runner.ContainerToken()
 545                 if err != nil {
 546                         return err
 547                 }
 548                 runner.ContainerConfig.Env = append(runner.ContainerConfig.Env,
 549                         "ARVADOS_API_TOKEN="+tok,
 550                         "ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"),
 551                         "ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"),
 552                 )
 553                 runner.ContainerConfig.NetworkDisabled = false
 554         } else {
 555                 runner.ContainerConfig.NetworkDisabled = true
 556         }
 557
 558         var err error
 559         runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil)
 560         if err != nil {
 561                 return fmt.Errorf("While creating container: %v", err)
 562         }
 563
 564         runner.HostConfig = dockerclient.HostConfig{
 565                 Binds:        runner.Binds,
 566                 CgroupParent: runner.setCgroupParent,
 567                 LogConfig: dockerclient.LogConfig{
 568                         Type: "none",
 569                 },
 570         }
 571
 572         return runner.AttachStreams()
 573 }
 574
 575 // StartContainer starts the docker container created by CreateContainer.
 576 func (runner *ContainerRunner) StartContainer() error {
 577         runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID)
 578         err := runner.Docker.StartContainer(runner.ContainerID, &runner.HostConfig)
 579         if err != nil {
 580                 return fmt.Errorf("could not start container: %v", err)
 581         }
 582         return nil
 583 }
 584
 585 // WaitFinish waits for the container to terminate, capture the exit code, and
 586 // close the stdout/stderr logging.
 587 func (runner *ContainerRunner) WaitFinish() error {
 588         runner.CrunchLog.Print("Waiting for container to finish")
 589
 590         result := runner.Docker.Wait(runner.ContainerID)
 591         wr := <-result
 592         if wr.Error != nil {
 593                 return fmt.Errorf("While waiting for container to finish: %v", wr.Error)
 594         }
 595         runner.ExitCode = &wr.ExitCode
 596
 597         // wait for stdout/stderr to complete
 598         <-runner.loggingDone
 599
 600         return nil
 601 }
 602
 603 // HandleOutput sets the output, unmounts the FUSE mount, and deletes temporary directories
 604 func (runner *ContainerRunner) CaptureOutput() error {
 605         if runner.finalState != "Complete" {
 606                 return nil
 607         }
 608
 609         if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
 610                 // Output may have been set directly by the container, so
 611                 // refresh the container record to check.
 612                 err := runner.ArvClient.Get("containers", runner.Container.UUID,
 613                         nil, &runner.Container)
 614                 if err != nil {
 615                         return err
 616                 }
 617                 if runner.Container.Output != "" {
 618                         // Container output is already set.
 619                         runner.OutputPDH = &runner.Container.Output
 620                         return nil
 621                 }
 622         }
 623
 624         if runner.HostOutputDir == "" {
 625                 return nil
 626         }
 627
 628         _, err := os.Stat(runner.HostOutputDir)
 629         if err != nil {
 630                 return fmt.Errorf("While checking host output path: %v", err)
 631         }
 632
 633         var manifestText string
 634
 635         collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
 636         _, err = os.Stat(collectionMetafile)
 637         if err != nil {
 638                 // Regular directory
 639                 cw := CollectionWriter{runner.Kc, nil, sync.Mutex{}}
 640                 manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger)
 641                 if err != nil {
 642                         return fmt.Errorf("While uploading output files: %v", err)
 643                 }
 644         } else {
 645                 // FUSE mount directory
 646                 file, openerr := os.Open(collectionMetafile)
 647                 if openerr != nil {
 648                         return fmt.Errorf("While opening FUSE metafile: %v", err)
 649                 }
 650                 defer file.Close()
 651
 652                 var rec arvados.Collection
 653                 err = json.NewDecoder(file).Decode(&rec)
 654                 if err != nil {
 655                         return fmt.Errorf("While reading FUSE metafile: %v", err)
 656                 }
 657                 manifestText = rec.ManifestText
 658         }
 659
 660         // Pre-populate output from the configured mount points
 661         var binds []string
 662         for bind, _ := range runner.Container.Mounts {
 663                 binds = append(binds, bind)
 664         }
 665         sort.Strings(binds)
 666
 667         for _, bind := range binds {
 668                 mnt := runner.Container.Mounts[bind]
 669
 670                 bindSuffix := strings.TrimPrefix(bind, runner.Container.OutputPath)
 671
 672                 if bindSuffix == bind || len(bindSuffix) <= 0 {
 673                         // either does not start with OutputPath or is OutputPath itself
 674                         continue
 675                 }
 676
 677                 if strings.HasPrefix(bindSuffix, "/") == false {
 678                         bindSuffix = "/" + bindSuffix
 679                 }
 680
 681                 if mnt.ExcludeFromOutput == true {
 682                         continue
 683                 }
 684
 685                 idx := strings.Index(mnt.PortableDataHash, "/")
 686                 if idx > 0 {
 687                         mnt.Path = mnt.PortableDataHash[idx:]
 688                         mnt.PortableDataHash = mnt.PortableDataHash[0:idx]
 689                 }
 690
 691                 // append to manifest_text
 692                 m, err := runner.getCollectionManifestForPath(mnt, bindSuffix)
 693                 if err != nil {
 694                         return err
 695                 }
 696
 697                 manifestText = manifestText + m
 698         }
 699
 700         // Save output
 701         var response arvados.Collection
 702         err = runner.ArvClient.Create("collections",
 703                 arvadosclient.Dict{
 704                         "collection": arvadosclient.Dict{
 705                                 "trash_at":      time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
 706                                 "name":          "output for " + runner.Container.UUID,
 707                                 "manifest_text": manifestText}},
 708                 &response)
 709         if err != nil {
 710                 return fmt.Errorf("While creating output collection: %v", err)
 711         }
 712         runner.OutputPDH = &response.PortableDataHash
 713         return nil
 714 }
 715
 716 var outputCollections = make(map[string]arvados.Collection)
 717
 718 // Fetch the collection for the mnt.PortableDataHash
 719 // Return the manifest_text fragment corresponding to the specified mnt.Path
 720 //  after making any required updates.
 721 //  Ex:
 722 //    If mnt.Path is not specified,
 723 //      return the entire manifest_text after replacing any "." with bindSuffix
 724 //    If mnt.Path corresponds to one stream,
 725 //      return the manifest_text for that stream after replacing that stream name with bindSuffix
 726 //    Otherwise, check if a filename in any one stream is being sought. Return the manifest_text
 727 //      for that stream after replacing stream name with bindSuffix minus the last word
 728 //      and the file name with last word of the bindSuffix
 729 //  Allowed path examples:
 730 //    "path":"/"
 731 //    "path":"/subdir1"
 732 //    "path":"/subdir1/subdir2"
 733 //    "path":"/subdir/filename" etc
 734 func (runner *ContainerRunner) getCollectionManifestForPath(mnt arvados.Mount, bindSuffix string) (string, error) {
 735         collection := outputCollections[mnt.PortableDataHash]
 736         if collection.PortableDataHash == "" {
 737                 err := runner.ArvClient.Get("collections", mnt.PortableDataHash, nil, &collection)
 738                 if err != nil {
 739                         return "", fmt.Errorf("While getting collection for %v: %v", mnt.PortableDataHash, err)
 740                 }
 741                 outputCollections[mnt.PortableDataHash] = collection
 742         }
 743
 744         manifest := manifest.Manifest{Text: collection.ManifestText}
 745
 746         manifestText := ""
 747         if mnt.Path == "" || mnt.Path == "/" {
 748                 // no path specified; return the entire manifest text after making adjustments
 749                 manifestText = collection.ManifestText
 750                 manifestText = strings.Replace(manifestText, "./", "."+bindSuffix+"/", -1)
 751                 manifestText = strings.Replace(manifestText, ". ", "."+bindSuffix+" ", -1)
 752         } else {
 753                 // either a single stream or file from a stream is being sought
 754                 bindIdx := strings.LastIndex(bindSuffix, "/")
 755                 var bindSubdir, bindFileName string
 756                 if bindIdx >= 0 {
 757                         bindSubdir = "." + bindSuffix[0:bindIdx]
 758                         bindFileName = bindSuffix[bindIdx+1:]
 759                 }
 760                 mntPath := mnt.Path
 761                 if strings.HasSuffix(mntPath, "/") {
 762                         mntPath = mntPath[0 : len(mntPath)-1]
 763                 }
 764                 pathIdx := strings.LastIndex(mntPath, "/")
 765                 var pathSubdir, pathFileName string
 766                 if pathIdx >= 0 {
 767                         pathSubdir = "." + mntPath[0:pathIdx]
 768                         pathFileName = mntPath[pathIdx+1:]
 769                 }
 770                 streams := strings.Split(collection.ManifestText, "\n")
 771                 for _, stream := range streams {
 772                         tokens := strings.Split(stream, " ")
 773                         if tokens[0] == "."+mntPath {
 774                                 // path refers to this complete stream
 775                                 adjustedStream := strings.Replace(stream, "."+mntPath, "."+bindSuffix, -1)
 776                                 manifestText = adjustedStream + "\n"
 777                                 break
 778                         } else {
 779                                 // look for a matching file in this stream
 780                                 fs := manifest.FileSegmentForPath(mntPath[1:])
 781                                 if fs != "" {
 782                                         manifestText = strings.Replace(fs, ":"+pathFileName, ":"+bindFileName, -1)
 783                                         manifestText = strings.Replace(manifestText, pathSubdir, bindSubdir, -1)
 784                                         manifestText += "\n"
 785                                         break
 786                                 }
 787                         }
 788                 }
 789         }
 790
 791         if manifestText == "" {
 792                 runner.CrunchLog.Printf("No manifest segment found for bind '%v' with path '%v'", bindSuffix, mnt.Path)
 793         }
 794
 795         return manifestText, nil
 796 }
 797
 798 func (runner *ContainerRunner) loadDiscoveryVars() {
 799         tl, err := runner.ArvClient.Discovery("defaultTrashLifetime")
 800         if err != nil {
 801                 log.Fatalf("getting defaultTrashLifetime from discovery document: %s", err)
 802         }
 803         runner.trashLifetime = time.Duration(tl.(float64)) * time.Second
 804 }
 805
 806 func (runner *ContainerRunner) CleanupDirs() {
 807         if runner.ArvMount != nil {
 808                 umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
 809                 umnterr := umount.Run()
 810                 if umnterr != nil {
 811                         runner.CrunchLog.Printf("While running fusermount: %v", umnterr)
 812                 }
 813
 814                 mnterr := <-runner.ArvMountExit
 815                 if mnterr != nil {
 816                         runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
 817                 }
 818         }
 819
 820         for _, tmpdir := range runner.CleanupTempDir {
 821                 rmerr := os.RemoveAll(tmpdir)
 822                 if rmerr != nil {
 823                         runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
 824                 }
 825         }
 826 }
 827
 828 // CommitLogs posts the collection containing the final container logs.
 829 func (runner *ContainerRunner) CommitLogs() error {
 830         runner.CrunchLog.Print(runner.finalState)
 831         runner.CrunchLog.Close()
 832
 833         // Closing CrunchLog above allows it to be committed to Keep at this
 834         // point, but re-open crunch log with ArvClient in case there are any
 835         // other further (such as failing to write the log to Keep!) while
 836         // shutting down
 837         runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
 838                 "crunch-run", nil})
 839
 840         if runner.LogsPDH != nil {
 841                 // If we have already assigned something to LogsPDH,
 842                 // we must be closing the re-opened log, which won't
 843                 // end up getting attached to the container record and
 844                 // therefore doesn't need to be saved as a collection
 845                 // -- it exists only to send logs to other channels.
 846                 return nil
 847         }
 848
 849         mt, err := runner.LogCollection.ManifestText()
 850         if err != nil {
 851                 return fmt.Errorf("While creating log manifest: %v", err)
 852         }
 853
 854         var response arvados.Collection
 855         err = runner.ArvClient.Create("collections",
 856                 arvadosclient.Dict{
 857                         "collection": arvadosclient.Dict{
 858                                 "trash_at":      time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
 859                                 "name":          "logs for " + runner.Container.UUID,
 860                                 "manifest_text": mt}},
 861                 &response)
 862         if err != nil {
 863                 return fmt.Errorf("While creating log collection: %v", err)
 864         }
 865         runner.LogsPDH = &response.PortableDataHash
 866         return nil
 867 }
 868
 869 // UpdateContainerRunning updates the container state to "Running"
 870 func (runner *ContainerRunner) UpdateContainerRunning() error {
 871         runner.CancelLock.Lock()
 872         defer runner.CancelLock.Unlock()
 873         if runner.Cancelled {
 874                 return ErrCancelled
 875         }
 876         return runner.ArvClient.Update("containers", runner.Container.UUID,
 877                 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
 878 }
 879
 880 // ContainerToken returns the api_token the container (and any
 881 // arv-mount processes) are allowed to use.
 882 func (runner *ContainerRunner) ContainerToken() (string, error) {
 883         if runner.token != "" {
 884                 return runner.token, nil
 885         }
 886
 887         var auth arvados.APIClientAuthorization
 888         err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
 889         if err != nil {
 890                 return "", err
 891         }
 892         runner.token = auth.APIToken
 893         return runner.token, nil
 894 }
 895
 896 // UpdateContainerComplete updates the container record state on API
 897 // server to "Complete" or "Cancelled"
 898 func (runner *ContainerRunner) UpdateContainerFinal() error {
 899         update := arvadosclient.Dict{}
 900         update["state"] = runner.finalState
 901         if runner.LogsPDH != nil {
 902                 update["log"] = *runner.LogsPDH
 903         }
 904         if runner.finalState == "Complete" {
 905                 if runner.ExitCode != nil {
 906                         update["exit_code"] = *runner.ExitCode
 907                 }
 908                 if runner.OutputPDH != nil {
 909                         update["output"] = *runner.OutputPDH
 910                 }
 911         }
 912         return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
 913 }
 914
 915 // IsCancelled returns the value of Cancelled, with goroutine safety.
 916 func (runner *ContainerRunner) IsCancelled() bool {
 917         runner.CancelLock.Lock()
 918         defer runner.CancelLock.Unlock()
 919         return runner.Cancelled
 920 }
 921
 922 // NewArvLogWriter creates an ArvLogWriter
 923 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
 924         return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
 925 }
 926
 927 // Run the full container lifecycle.
 928 func (runner *ContainerRunner) Run() (err error) {
 929         runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
 930
 931         hostname, hosterr := os.Hostname()
 932         if hosterr != nil {
 933                 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
 934         } else {
 935                 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
 936         }
 937
 938         // Clean up temporary directories _after_ finalizing
 939         // everything (if we've made any by then)
 940         defer runner.CleanupDirs()
 941
 942         runner.finalState = "Queued"
 943
 944         defer func() {
 945                 // checkErr prints e (unless it's nil) and sets err to
 946                 // e (unless err is already non-nil). Thus, if err
 947                 // hasn't already been assigned when Run() returns,
 948                 // this cleanup func will cause Run() to return the
 949                 // first non-nil error that is passed to checkErr().
 950                 checkErr := func(e error) {
 951                         if e == nil {
 952                                 return
 953                         }
 954                         runner.CrunchLog.Print(e)
 955                         if err == nil {
 956                                 err = e
 957                         }
 958                 }
 959
 960                 // Log the error encountered in Run(), if any
 961                 checkErr(err)
 962
 963                 if runner.finalState == "Queued" {
 964                         runner.CrunchLog.Close()
 965                         runner.UpdateContainerFinal()
 966                         return
 967                 }
 968
 969                 if runner.IsCancelled() {
 970                         runner.finalState = "Cancelled"
 971                         // but don't return yet -- we still want to
 972                         // capture partial output and write logs
 973                 }
 974
 975                 checkErr(runner.CaptureOutput())
 976                 checkErr(runner.CommitLogs())
 977                 checkErr(runner.UpdateContainerFinal())
 978
 979                 // The real log is already closed, but then we opened
 980                 // a new one in case we needed to log anything while
 981                 // finalizing.
 982                 runner.CrunchLog.Close()
 983         }()
 984
 985         err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container)
 986         if err != nil {
 987                 err = fmt.Errorf("While getting container record: %v", err)
 988                 return
 989         }
 990
 991         // setup signal handling
 992         runner.SetupSignals()
 993
 994         // check for and/or load image
 995         err = runner.LoadImage()
 996         if err != nil {
 997                 runner.finalState = "Cancelled"
 998                 err = fmt.Errorf("While loading container image: %v", err)
 999                 return
1000         }
1001
1002         // set up FUSE mount and binds
1003         err = runner.SetupMounts()
1004         if err != nil {
1005                 runner.finalState = "Cancelled"
1006                 err = fmt.Errorf("While setting up mounts: %v", err)
1007                 return
1008         }
1009
1010         err = runner.CreateContainer()
1011         if err != nil {
1012                 return
1013         }
1014
1015         runner.StartCrunchstat()
1016
1017         if runner.IsCancelled() {
1018                 return
1019         }
1020
1021         err = runner.UpdateContainerRunning()
1022         if err != nil {
1023                 return
1024         }
1025         runner.finalState = "Cancelled"
1026
1027         err = runner.StartContainer()
1028         if err != nil {
1029                 return
1030         }
1031
1032         err = runner.WaitFinish()
1033         if err == nil {
1034                 runner.finalState = "Complete"
1035         }
1036         return
1037 }
1038
1039 // NewContainerRunner creates a new container runner.
1040 func NewContainerRunner(api IArvadosClient,
1041         kc IKeepClient,
1042         docker ThinDockerClient,
1043         containerUUID string) *ContainerRunner {
1044
1045         cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
1046         cr.NewLogWriter = cr.NewArvLogWriter
1047         cr.RunArvMount = cr.ArvMountCmd
1048         cr.MkTempDir = ioutil.TempDir
1049         cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
1050         cr.Container.UUID = containerUUID
1051         cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
1052         cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
1053         cr.loadDiscoveryVars()
1054         return cr
1055 }
1056
1057 func main() {
1058         statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
1059         cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
1060         cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
1061         cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
1062         caCertsPath := flag.String("ca-certs", "", "Path to TLS root certificates")
1063         flag.Parse()
1064
1065         containerId := flag.Arg(0)
1066
1067         if *caCertsPath != "" {
1068                 arvadosclient.CertFiles = []string{*caCertsPath}
1069         }
1070
1071         api, err := arvadosclient.MakeArvadosClient()
1072         if err != nil {
1073                 log.Fatalf("%s: %v", containerId, err)
1074         }
1075         api.Retries = 8
1076
1077         var kc *keepclient.KeepClient
1078         kc, err = keepclient.MakeKeepClient(api)
1079         if err != nil {
1080                 log.Fatalf("%s: %v", containerId, err)
1081         }
1082         kc.Retries = 4
1083
1084         var docker *dockerclient.DockerClient
1085         docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil)
1086         if err != nil {
1087                 log.Fatalf("%s: %v", containerId, err)
1088         }
1089
1090         cr := NewContainerRunner(api, kc, docker, containerId)
1091         cr.statInterval = *statInterval
1092         cr.cgroupRoot = *cgroupRoot
1093         cr.expectCgroupParent = *cgroupParent
1094         if *cgroupParentSubsystem != "" {
1095                 p := findCgroup(*cgroupParentSubsystem)
1096                 cr.setCgroupParent = p
1097                 cr.expectCgroupParent = p
1098         }
1099
1100         err = cr.Run()
1101         if err != nil {
1102                 log.Fatalf("%s: %v", containerId, err)
1103         }
1104
1105 }