services/crunch-run/crunchrun.go

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