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