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         manifestText := manifest.Manifest{Text: manifestText}.NormalizedText()
 703         err = runner.ArvClient.Create("collections",
 704                 arvadosclient.Dict{
 705                         "collection": arvadosclient.Dict{
 706                                 "trash_at":      time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
 707                                 "name":          "output for " + runner.Container.UUID,
 708                                 "manifest_text": manifestText}},
 709                 &response)
 710         if err != nil {
 711                 return fmt.Errorf("While creating output collection: %v", err)
 712         }
 713         runner.OutputPDH = &response.PortableDataHash
 714         return nil
 715 }
 716
 717 var outputCollections = make(map[string]arvados.Collection)
 718
 719 // Fetch the collection for the mnt.PortableDataHash
 720 // Return the manifest_text fragment corresponding to the specified mnt.Path
 721 //  after making any required updates.
 722 //  Ex:
 723 //    If mnt.Path is not specified,
 724 //      return the entire manifest_text after replacing any "." with bindSuffix
 725 //    If mnt.Path corresponds to one stream,
 726 //      return the manifest_text for that stream after replacing that stream name with bindSuffix
 727 //    Otherwise, check if a filename in any one stream is being sought. Return the manifest_text
 728 //      for that stream after replacing stream name with bindSuffix minus the last word
 729 //      and the file name with last word of the bindSuffix
 730 //  Allowed path examples:
 731 //    "path":"/"
 732 //    "path":"/subdir1"
 733 //    "path":"/subdir1/subdir2"
 734 //    "path":"/subdir/filename" etc
 735 func (runner *ContainerRunner) getCollectionManifestForPath(mnt arvados.Mount, bindSuffix string) (string, error) {
 736         collection := outputCollections[mnt.PortableDataHash]
 737         if collection.PortableDataHash == "" {
 738                 err := runner.ArvClient.Get("collections", mnt.PortableDataHash, nil, &collection)
 739                 if err != nil {
 740                         return "", fmt.Errorf("While getting collection for %v: %v", mnt.PortableDataHash, err)
 741                 }
 742                 outputCollections[mnt.PortableDataHash] = collection
 743         }
 744
 745         if collection.ManifestText == "" {
 746                 runner.CrunchLog.Printf("No manifest text for collection %v", collection.PortableDataHash)
 747                 return "", nil
 748         }
 749
 750         manifest := manifest.Manifest{Text: collection.ManifestText}
 751         manifestText := manifest.ManifestTextForPath(mnt.Path, bindSuffix)
 752         return manifestText, nil
 753 }
 754
 755 func (runner *ContainerRunner) loadDiscoveryVars() {
 756         tl, err := runner.ArvClient.Discovery("defaultTrashLifetime")
 757         if err != nil {
 758                 log.Fatalf("getting defaultTrashLifetime from discovery document: %s", err)
 759         }
 760         runner.trashLifetime = time.Duration(tl.(float64)) * time.Second
 761 }
 762
 763 func (runner *ContainerRunner) CleanupDirs() {
 764         if runner.ArvMount != nil {
 765                 umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
 766                 umnterr := umount.Run()
 767                 if umnterr != nil {
 768                         runner.CrunchLog.Printf("While running fusermount: %v", umnterr)
 769                 }
 770
 771                 mnterr := <-runner.ArvMountExit
 772                 if mnterr != nil {
 773                         runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr)
 774                 }
 775         }
 776
 777         for _, tmpdir := range runner.CleanupTempDir {
 778                 rmerr := os.RemoveAll(tmpdir)
 779                 if rmerr != nil {
 780                         runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr)
 781                 }
 782         }
 783 }
 784
 785 // CommitLogs posts the collection containing the final container logs.
 786 func (runner *ContainerRunner) CommitLogs() error {
 787         runner.CrunchLog.Print(runner.finalState)
 788         runner.CrunchLog.Close()
 789
 790         // Closing CrunchLog above allows it to be committed to Keep at this
 791         // point, but re-open crunch log with ArvClient in case there are any
 792         // other further (such as failing to write the log to Keep!) while
 793         // shutting down
 794         runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
 795                 "crunch-run", nil})
 796
 797         if runner.LogsPDH != nil {
 798                 // If we have already assigned something to LogsPDH,
 799                 // we must be closing the re-opened log, which won't
 800                 // end up getting attached to the container record and
 801                 // therefore doesn't need to be saved as a collection
 802                 // -- it exists only to send logs to other channels.
 803                 return nil
 804         }
 805
 806         mt, err := runner.LogCollection.ManifestText()
 807         if err != nil {
 808                 return fmt.Errorf("While creating log manifest: %v", err)
 809         }
 810
 811         var response arvados.Collection
 812         err = runner.ArvClient.Create("collections",
 813                 arvadosclient.Dict{
 814                         "collection": arvadosclient.Dict{
 815                                 "trash_at":      time.Now().Add(runner.trashLifetime).Format(time.RFC3339),
 816                                 "name":          "logs for " + runner.Container.UUID,
 817                                 "manifest_text": mt}},
 818                 &response)
 819         if err != nil {
 820                 return fmt.Errorf("While creating log collection: %v", err)
 821         }
 822         runner.LogsPDH = &response.PortableDataHash
 823         return nil
 824 }
 825
 826 // UpdateContainerRunning updates the container state to "Running"
 827 func (runner *ContainerRunner) UpdateContainerRunning() error {
 828         runner.CancelLock.Lock()
 829         defer runner.CancelLock.Unlock()
 830         if runner.Cancelled {
 831                 return ErrCancelled
 832         }
 833         return runner.ArvClient.Update("containers", runner.Container.UUID,
 834                 arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil)
 835 }
 836
 837 // ContainerToken returns the api_token the container (and any
 838 // arv-mount processes) are allowed to use.
 839 func (runner *ContainerRunner) ContainerToken() (string, error) {
 840         if runner.token != "" {
 841                 return runner.token, nil
 842         }
 843
 844         var auth arvados.APIClientAuthorization
 845         err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth)
 846         if err != nil {
 847                 return "", err
 848         }
 849         runner.token = auth.APIToken
 850         return runner.token, nil
 851 }
 852
 853 // UpdateContainerComplete updates the container record state on API
 854 // server to "Complete" or "Cancelled"
 855 func (runner *ContainerRunner) UpdateContainerFinal() error {
 856         update := arvadosclient.Dict{}
 857         update["state"] = runner.finalState
 858         if runner.LogsPDH != nil {
 859                 update["log"] = *runner.LogsPDH
 860         }
 861         if runner.finalState == "Complete" {
 862                 if runner.ExitCode != nil {
 863                         update["exit_code"] = *runner.ExitCode
 864                 }
 865                 if runner.OutputPDH != nil {
 866                         update["output"] = *runner.OutputPDH
 867                 }
 868         }
 869         return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil)
 870 }
 871
 872 // IsCancelled returns the value of Cancelled, with goroutine safety.
 873 func (runner *ContainerRunner) IsCancelled() bool {
 874         runner.CancelLock.Lock()
 875         defer runner.CancelLock.Unlock()
 876         return runner.Cancelled
 877 }
 878
 879 // NewArvLogWriter creates an ArvLogWriter
 880 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
 881         return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
 882 }
 883
 884 // Run the full container lifecycle.
 885 func (runner *ContainerRunner) Run() (err error) {
 886         runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID)
 887
 888         hostname, hosterr := os.Hostname()
 889         if hosterr != nil {
 890                 runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr)
 891         } else {
 892                 runner.CrunchLog.Printf("Executing on host '%s'", hostname)
 893         }
 894
 895         // Clean up temporary directories _after_ finalizing
 896         // everything (if we've made any by then)
 897         defer runner.CleanupDirs()
 898
 899         runner.finalState = "Queued"
 900
 901         defer func() {
 902                 // checkErr prints e (unless it's nil) and sets err to
 903                 // e (unless err is already non-nil). Thus, if err
 904                 // hasn't already been assigned when Run() returns,
 905                 // this cleanup func will cause Run() to return the
 906                 // first non-nil error that is passed to checkErr().
 907                 checkErr := func(e error) {
 908                         if e == nil {
 909                                 return
 910                         }
 911                         runner.CrunchLog.Print(e)
 912                         if err == nil {
 913                                 err = e
 914                         }
 915                 }
 916
 917                 // Log the error encountered in Run(), if any
 918                 checkErr(err)
 919
 920                 if runner.finalState == "Queued" {
 921                         runner.CrunchLog.Close()
 922                         runner.UpdateContainerFinal()
 923                         return
 924                 }
 925
 926                 if runner.IsCancelled() {
 927                         runner.finalState = "Cancelled"
 928                         // but don't return yet -- we still want to
 929                         // capture partial output and write logs
 930                 }
 931
 932                 checkErr(runner.CaptureOutput())
 933                 checkErr(runner.CommitLogs())
 934                 checkErr(runner.UpdateContainerFinal())
 935
 936                 // The real log is already closed, but then we opened
 937                 // a new one in case we needed to log anything while
 938                 // finalizing.
 939                 runner.CrunchLog.Close()
 940         }()
 941
 942         err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container)
 943         if err != nil {
 944                 err = fmt.Errorf("While getting container record: %v", err)
 945                 return
 946         }
 947
 948         // setup signal handling
 949         runner.SetupSignals()
 950
 951         // check for and/or load image
 952         err = runner.LoadImage()
 953         if err != nil {
 954                 runner.finalState = "Cancelled"
 955                 err = fmt.Errorf("While loading container image: %v", err)
 956                 return
 957         }
 958
 959         // set up FUSE mount and binds
 960         err = runner.SetupMounts()
 961         if err != nil {
 962                 runner.finalState = "Cancelled"
 963                 err = fmt.Errorf("While setting up mounts: %v", err)
 964                 return
 965         }
 966
 967         err = runner.CreateContainer()
 968         if err != nil {
 969                 return
 970         }
 971
 972         runner.StartCrunchstat()
 973
 974         if runner.IsCancelled() {
 975                 return
 976         }
 977
 978         err = runner.UpdateContainerRunning()
 979         if err != nil {
 980                 return
 981         }
 982         runner.finalState = "Cancelled"
 983
 984         err = runner.StartContainer()
 985         if err != nil {
 986                 return
 987         }
 988
 989         err = runner.WaitFinish()
 990         if err == nil {
 991                 runner.finalState = "Complete"
 992         }
 993         return
 994 }
 995
 996 // NewContainerRunner creates a new container runner.
 997 func NewContainerRunner(api IArvadosClient,
 998         kc IKeepClient,
 999         docker ThinDockerClient,
1000         containerUUID string) *ContainerRunner {
1001
1002         cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
1003         cr.NewLogWriter = cr.NewArvLogWriter
1004         cr.RunArvMount = cr.ArvMountCmd
1005         cr.MkTempDir = ioutil.TempDir
1006         cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
1007         cr.Container.UUID = containerUUID
1008         cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
1009         cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
1010         cr.loadDiscoveryVars()
1011         return cr
1012 }
1013
1014 func main() {
1015         statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
1016         cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
1017         cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
1018         cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
1019         caCertsPath := flag.String("ca-certs", "", "Path to TLS root certificates")
1020         flag.Parse()
1021
1022         containerId := flag.Arg(0)
1023
1024         if *caCertsPath != "" {
1025                 arvadosclient.CertFiles = []string{*caCertsPath}
1026         }
1027
1028         api, err := arvadosclient.MakeArvadosClient()
1029         if err != nil {
1030                 log.Fatalf("%s: %v", containerId, err)
1031         }
1032         api.Retries = 8
1033
1034         var kc *keepclient.KeepClient
1035         kc, err = keepclient.MakeKeepClient(api)
1036         if err != nil {
1037                 log.Fatalf("%s: %v", containerId, err)
1038         }
1039         kc.Retries = 4
1040
1041         var docker *dockerclient.DockerClient
1042         docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil)
1043         if err != nil {
1044                 log.Fatalf("%s: %v", containerId, err)
1045         }
1046
1047         cr := NewContainerRunner(api, kc, docker, containerId)
1048         cr.statInterval = *statInterval
1049         cr.cgroupRoot = *cgroupRoot
1050         cr.expectCgroupParent = *cgroupParent
1051         if *cgroupParentSubsystem != "" {
1052                 p := findCgroup(*cgroupParentSubsystem)
1053                 cr.setCgroupParent = p
1054                 cr.expectCgroupParent = p
1055         }
1056
1057         err = cr.Run()
1058         if err != nil {
1059                 log.Fatalf("%s: %v", containerId, err)
1060         }
1061
1062 }