services/crunch-run/crunchrun.go

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