X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/453f922b16ffa246b32700ce06bf16a1cdffce60..80081626101a2193d7b916fe488903b66777576e:/services/crunch-run/crunchrun.go?ds=sidebyside diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index 6a118a6743..53815cbe1c 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -6,7 +6,6 @@ package main import ( "bytes" - "context" "encoding/json" "errors" "flag" @@ -33,6 +32,7 @@ import ( "git.curoverse.com/arvados.git/sdk/go/arvadosclient" "git.curoverse.com/arvados.git/sdk/go/keepclient" "git.curoverse.com/arvados.git/sdk/go/manifest" + "golang.org/x/net/context" dockertypes "github.com/docker/docker/api/types" dockercontainer "github.com/docker/docker/api/types/container" @@ -75,60 +75,13 @@ type ThinDockerClient interface { ContainerCreate(ctx context.Context, config *dockercontainer.Config, hostConfig *dockercontainer.HostConfig, networkingConfig *dockernetwork.NetworkingConfig, containerName string) (dockercontainer.ContainerCreateCreatedBody, error) ContainerStart(ctx context.Context, container string, options dockertypes.ContainerStartOptions) error - ContainerStop(ctx context.Context, container string, timeout *time.Duration) error + ContainerRemove(ctx context.Context, container string, options dockertypes.ContainerRemoveOptions) error ContainerWait(ctx context.Context, container string, condition dockercontainer.WaitCondition) (<-chan dockercontainer.ContainerWaitOKBody, <-chan error) ImageInspectWithRaw(ctx context.Context, image string) (dockertypes.ImageInspect, []byte, error) ImageLoad(ctx context.Context, input io.Reader, quiet bool) (dockertypes.ImageLoadResponse, error) ImageRemove(ctx context.Context, image string, options dockertypes.ImageRemoveOptions) ([]dockertypes.ImageDeleteResponseItem, error) } -// ThinDockerClientProxy is a proxy implementation of ThinDockerClient -// that executes the docker requests on dockerclient.Client -type ThinDockerClientProxy struct { - Docker *dockerclient.Client -} - -// ContainerAttach invokes dockerclient.Client.ContainerAttach -func (proxy ThinDockerClientProxy) ContainerAttach(ctx context.Context, container string, options dockertypes.ContainerAttachOptions) (dockertypes.HijackedResponse, error) { - return proxy.Docker.ContainerAttach(ctx, container, options) -} - -// ContainerCreate invokes dockerclient.Client.ContainerCreate -func (proxy ThinDockerClientProxy) ContainerCreate(ctx context.Context, config *dockercontainer.Config, hostConfig *dockercontainer.HostConfig, - networkingConfig *dockernetwork.NetworkingConfig, containerName string) (dockercontainer.ContainerCreateCreatedBody, error) { - return proxy.Docker.ContainerCreate(ctx, config, hostConfig, networkingConfig, containerName) -} - -// ContainerStart invokes dockerclient.Client.ContainerStart -func (proxy ThinDockerClientProxy) ContainerStart(ctx context.Context, container string, options dockertypes.ContainerStartOptions) error { - return proxy.Docker.ContainerStart(ctx, container, options) -} - -// ContainerStop invokes dockerclient.Client.ContainerStop -func (proxy ThinDockerClientProxy) ContainerStop(ctx context.Context, container string, timeout *time.Duration) error { - return proxy.Docker.ContainerStop(ctx, container, timeout) -} - -// ContainerWait invokes dockerclient.Client.ContainerWait -func (proxy ThinDockerClientProxy) ContainerWait(ctx context.Context, container string, condition dockercontainer.WaitCondition) (<-chan dockercontainer.ContainerWaitOKBody, <-chan error) { - return proxy.Docker.ContainerWait(ctx, container, condition) -} - -// ImageInspectWithRaw invokes dockerclient.Client.ImageInspectWithRaw -func (proxy ThinDockerClientProxy) ImageInspectWithRaw(ctx context.Context, image string) (dockertypes.ImageInspect, []byte, error) { - return proxy.Docker.ImageInspectWithRaw(ctx, image) -} - -// ImageLoad invokes dockerclient.Client.ImageLoad -func (proxy ThinDockerClientProxy) ImageLoad(ctx context.Context, input io.Reader, quiet bool) (dockertypes.ImageLoadResponse, error) { - return proxy.Docker.ImageLoad(ctx, input, quiet) -} - -// ImageRemove invokes dockerclient.Client.ImageRemove -func (proxy ThinDockerClientProxy) ImageRemove(ctx context.Context, image string, options dockertypes.ImageRemoveOptions) ([]dockertypes.ImageDeleteResponseItem, error) { - return proxy.Docker.ImageRemove(ctx, image, options) -} - // ContainerRunner is the main stateful struct used for a single execution of a // container. type ContainerRunner struct { @@ -150,21 +103,25 @@ type ContainerRunner struct { LogsPDH *string RunArvMount MkTempDir - ArvMount *exec.Cmd - ArvMountPoint string - HostOutputDir string - CleanupTempDir []string - Binds []string - Volumes map[string]struct{} - OutputPDH *string - SigChan chan os.Signal - ArvMountExit chan error - finalState string - - statLogger io.WriteCloser - statReporter *crunchstat.Reporter - statInterval time.Duration - cgroupRoot string + ArvMount *exec.Cmd + ArvMountPoint string + HostOutputDir string + Binds []string + Volumes map[string]struct{} + OutputPDH *string + SigChan chan os.Signal + ArvMountExit chan error + SecretMounts map[string]arvados.Mount + MkArvClient func(token string) (IArvadosClient, error) + finalState string + parentTemp string + + statLogger io.WriteCloser + statReporter *crunchstat.Reporter + hoststatLogger io.WriteCloser + hoststatReporter *crunchstat.Reporter + statInterval time.Duration + cgroupRoot string // What we expect the container's cgroup parent to be. expectCgroupParent string // What we tell docker to use as the container's cgroup @@ -180,7 +137,6 @@ type ContainerRunner struct { setCgroupParent string cStateLock sync.Mutex - cStarted bool // StartContainer() succeeded cCancelled bool // StopContainer() invoked enableNetwork string // one of "default" or "always" @@ -197,37 +153,27 @@ func (runner *ContainerRunner) setupSignals() { signal.Notify(runner.SigChan, syscall.SIGQUIT) go func(sig chan os.Signal) { - s := <-sig - if s != nil { - runner.CrunchLog.Printf("Caught signal %v", s) + for s := range sig { + runner.stop(s) } - runner.stop() }(runner.SigChan) } // stop the underlying Docker container. -func (runner *ContainerRunner) stop() { +func (runner *ContainerRunner) stop(sig os.Signal) { runner.cStateLock.Lock() defer runner.cStateLock.Unlock() - if runner.cCancelled { + if sig != nil { + runner.CrunchLog.Printf("caught signal: %v", sig) + } + if runner.ContainerID == "" { return } runner.cCancelled = true - if runner.cStarted { - timeout := time.Duration(10) - err := runner.Docker.ContainerStop(context.TODO(), runner.ContainerID, &(timeout)) - if err != nil { - runner.CrunchLog.Printf("StopContainer failed: %s", err) - } - // Suppress multiple calls to stop() - runner.cStarted = false - } -} - -func (runner *ContainerRunner) stopSignals() { - if runner.SigChan != nil { - signal.Stop(runner.SigChan) - close(runner.SigChan) + runner.CrunchLog.Printf("removing container") + err := runner.Docker.ContainerRemove(context.TODO(), runner.ContainerID, dockertypes.ContainerRemoveOptions{Force: true}) + if err != nil { + runner.CrunchLog.Printf("error removing container: %s", err) } } @@ -379,11 +325,42 @@ func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string, token string) ( func (runner *ContainerRunner) SetupArvMountPoint(prefix string) (err error) { if runner.ArvMountPoint == "" { - runner.ArvMountPoint, err = runner.MkTempDir("", prefix) + runner.ArvMountPoint, err = runner.MkTempDir(runner.parentTemp, prefix) } return } +func copyfile(src string, dst string) (err error) { + srcfile, err := os.Open(src) + if err != nil { + return + } + + os.MkdirAll(path.Dir(dst), 0777) + + dstfile, err := os.Create(dst) + if err != nil { + return + } + _, err = io.Copy(dstfile, srcfile) + if err != nil { + return + } + + err = srcfile.Close() + err2 := dstfile.Close() + + if err != nil { + return + } + + if err2 != nil { + return err2 + } + + return nil +} + func (runner *ContainerRunner) SetupMounts() (err error) { err = runner.SetupArvMountPoint("keep") if err != nil { @@ -411,15 +388,34 @@ func (runner *ContainerRunner) SetupMounts() (err error) { runner.Binds = nil runner.Volumes = make(map[string]struct{}) needCertMount := true + type copyFile struct { + src string + bind string + } + var copyFiles []copyFile var binds []string for bind := range runner.Container.Mounts { binds = append(binds, bind) } + for bind := range runner.SecretMounts { + if _, ok := runner.Container.Mounts[bind]; ok { + return fmt.Errorf("Secret mount %q conflicts with regular mount", bind) + } + if runner.SecretMounts[bind].Kind != "json" && + runner.SecretMounts[bind].Kind != "text" { + return fmt.Errorf("Secret mount %q type is %q but only 'json' and 'text' are permitted.", + bind, runner.SecretMounts[bind].Kind) + } + binds = append(binds, bind) + } sort.Strings(binds) for _, bind := range binds { - mnt := runner.Container.Mounts[bind] + mnt, ok := runner.Container.Mounts[bind] + if !ok { + mnt = runner.SecretMounts[bind] + } if bind == "stdout" || bind == "stderr" { // Is it a "file" mount kind? if mnt.Kind != "file" { @@ -448,8 +444,8 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } if strings.HasPrefix(bind, runner.Container.OutputPath+"/") && bind != runner.Container.OutputPath+"/" { - if mnt.Kind != "collection" { - return fmt.Errorf("Only mount points of kind 'collection' are supported underneath the output_path: %v", bind) + if mnt.Kind != "collection" && mnt.Kind != "text" && mnt.Kind != "json" { + return fmt.Errorf("Only mount points of kind 'collection', 'text' or 'json' are supported underneath the output_path for %q, was %q", bind, mnt.Kind) } } @@ -466,7 +462,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { pdhOnly = false src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID) } else if mnt.PortableDataHash != "" { - if mnt.Writable { + if mnt.Writable && !strings.HasPrefix(bind, runner.Container.OutputPath+"/") { return fmt.Errorf("Can never write to a collection specified by portable data hash") } idx := strings.Index(mnt.PortableDataHash, "/") @@ -493,10 +489,12 @@ func (runner *ContainerRunner) SetupMounts() (err error) { if mnt.Writable { if bind == runner.Container.OutputPath { runner.HostOutputDir = src + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind)) } else if strings.HasPrefix(bind, runner.Container.OutputPath+"/") { - return fmt.Errorf("Writable mount points are not permitted underneath the output_path: %v", bind) + copyFiles = append(copyFiles, copyFile{src, runner.HostOutputDir + bind[len(runner.Container.OutputPath):]}) + } else { + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind)) } - runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind)) } else { runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind)) } @@ -504,7 +502,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { case mnt.Kind == "tmp": var tmpdir string - tmpdir, err = runner.MkTempDir("", "") + tmpdir, err = runner.MkTempDir(runner.parentTemp, "tmp") if err != nil { return fmt.Errorf("While creating mount temp dir: %v", err) } @@ -516,40 +514,46 @@ func (runner *ContainerRunner) SetupMounts() (err error) { if staterr != nil { return fmt.Errorf("While Chmod temp dir: %v", err) } - runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir) runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", tmpdir, bind)) if bind == runner.Container.OutputPath { runner.HostOutputDir = tmpdir } - case mnt.Kind == "json": - jsondata, err := json.Marshal(mnt.Content) - if err != nil { - return fmt.Errorf("encoding json data: %v", err) + case mnt.Kind == "json" || mnt.Kind == "text": + var filedata []byte + if mnt.Kind == "json" { + filedata, err = json.Marshal(mnt.Content) + if err != nil { + return fmt.Errorf("encoding json data: %v", err) + } + } else { + text, ok := mnt.Content.(string) + if !ok { + return fmt.Errorf("content for mount %q must be a string", bind) + } + filedata = []byte(text) } - // Create a tempdir with a single file - // (instead of just a tempfile): this way we - // can ensure the file is world-readable - // inside the container, without having to - // make it world-readable on the docker host. - tmpdir, err := runner.MkTempDir("", "") + + tmpdir, err := runner.MkTempDir(runner.parentTemp, mnt.Kind) if err != nil { return fmt.Errorf("creating temp dir: %v", err) } - runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir) - tmpfn := filepath.Join(tmpdir, "mountdata.json") - err = ioutil.WriteFile(tmpfn, jsondata, 0644) + tmpfn := filepath.Join(tmpdir, "mountdata."+mnt.Kind) + err = ioutil.WriteFile(tmpfn, filedata, 0444) if err != nil { return fmt.Errorf("writing temp file: %v", err) } - runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind)) + if strings.HasPrefix(bind, runner.Container.OutputPath+"/") { + copyFiles = append(copyFiles, copyFile{tmpfn, runner.HostOutputDir + bind[len(runner.Container.OutputPath):]}) + } else { + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind)) + } case mnt.Kind == "git_tree": - tmpdir, err := runner.MkTempDir("", "") + tmpdir, err := runner.MkTempDir(runner.parentTemp, "git_tree") if err != nil { return fmt.Errorf("creating temp dir: %v", err) } - runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir) err = gitMount(mnt).extractTree(runner.ArvClient, tmpdir, token) if err != nil { return err @@ -591,59 +595,118 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } } + for _, cp := range copyFiles { + st, err := os.Stat(cp.src) + if err != nil { + return fmt.Errorf("While staging writable file from %q to %q: %v", cp.src, cp.bind, err) + } + if st.IsDir() { + err = filepath.Walk(cp.src, func(walkpath string, walkinfo os.FileInfo, walkerr error) error { + if walkerr != nil { + return walkerr + } + target := path.Join(cp.bind, walkpath[len(cp.src):]) + if walkinfo.Mode().IsRegular() { + copyerr := copyfile(walkpath, target) + if copyerr != nil { + return copyerr + } + return os.Chmod(target, walkinfo.Mode()|0777) + } else if walkinfo.Mode().IsDir() { + mkerr := os.MkdirAll(target, 0777) + if mkerr != nil { + return mkerr + } + return os.Chmod(target, walkinfo.Mode()|os.ModeSetgid|0777) + } else { + return fmt.Errorf("Source %q is not a regular file or directory", cp.src) + } + }) + } else if st.Mode().IsRegular() { + err = copyfile(cp.src, cp.bind) + if err == nil { + err = os.Chmod(cp.bind, st.Mode()|0777) + } + } + if err != nil { + return fmt.Errorf("While staging writable file from %q to %q: %v", cp.src, cp.bind, err) + } + } + return nil } func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) { // Handle docker log protocol // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container + defer close(runner.loggingDone) header := make([]byte, 8) - for { - _, readerr := io.ReadAtLeast(containerReader, header, 8) - - if readerr == nil { - readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24) - if header[0] == 1 { - // stdout - _, readerr = io.CopyN(runner.Stdout, containerReader, readsize) - } else { - // stderr - _, readerr = io.CopyN(runner.Stderr, containerReader, readsize) + var err error + for err == nil { + _, err = io.ReadAtLeast(containerReader, header, 8) + if err != nil { + if err == io.EOF { + err = nil } + break } + readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24) + if header[0] == 1 { + // stdout + _, err = io.CopyN(runner.Stdout, containerReader, readsize) + } else { + // stderr + _, err = io.CopyN(runner.Stderr, containerReader, readsize) + } + } - if readerr != nil { - if readerr != io.EOF { - runner.CrunchLog.Printf("While reading docker logs: %v", readerr) - } - - closeerr := runner.Stdout.Close() - if closeerr != nil { - runner.CrunchLog.Printf("While closing stdout logs: %v", closeerr) - } + if err != nil { + runner.CrunchLog.Printf("error reading docker logs: %v", err) + } - closeerr = runner.Stderr.Close() - if closeerr != nil { - runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr) - } + err = runner.Stdout.Close() + if err != nil { + runner.CrunchLog.Printf("error closing stdout logs: %v", err) + } - if runner.statReporter != nil { - runner.statReporter.Stop() - closeerr = runner.statLogger.Close() - if closeerr != nil { - runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr) - } - } + err = runner.Stderr.Close() + if err != nil { + runner.CrunchLog.Printf("error closing stderr logs: %v", err) + } - runner.loggingDone <- true - close(runner.loggingDone) - return + if runner.statReporter != nil { + runner.statReporter.Stop() + err = runner.statLogger.Close() + if err != nil { + runner.CrunchLog.Printf("error closing crunchstat logs: %v", err) } } } -func (runner *ContainerRunner) StartCrunchstat() { +func (runner *ContainerRunner) stopHoststat() error { + if runner.hoststatReporter == nil { + return nil + } + runner.hoststatReporter.Stop() + err := runner.hoststatLogger.Close() + if err != nil { + return fmt.Errorf("error closing hoststat logs: %v", err) + } + return nil +} + +func (runner *ContainerRunner) startHoststat() { + runner.hoststatLogger = NewThrottledLogger(runner.NewLogWriter("hoststat")) + runner.hoststatReporter = &crunchstat.Reporter{ + Logger: log.New(runner.hoststatLogger, "", 0), + CgroupRoot: runner.cgroupRoot, + PollPeriod: runner.statInterval, + } + runner.hoststatReporter.Start() +} + +func (runner *ContainerRunner) startCrunchstat() { runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat")) runner.statReporter = &crunchstat.Reporter{ CID: runner.ContainerID, @@ -849,7 +912,7 @@ func (runner *ContainerRunner) AttachStreams() (err error) { _, err := io.Copy(response.Conn, stdinRdr) if err != nil { runner.CrunchLog.Print("While writing stdin collection to docker container %q", err) - runner.stop() + runner.stop(nil) } stdinRdr.Close() response.CloseWrite() @@ -859,7 +922,7 @@ func (runner *ContainerRunner) AttachStreams() (err error) { _, err := io.Copy(response.Conn, bytes.NewReader(stdinJson)) if err != nil { runner.CrunchLog.Print("While writing stdin json to docker container %q", err) - runner.stop() + runner.stop(nil) } response.CloseWrite() }() @@ -910,6 +973,7 @@ func (runner *ContainerRunner) CreateContainer() error { runner.ContainerConfig.Volumes = runner.Volumes + maxRAM := int64(runner.Container.RuntimeConstraints.RAM) runner.HostConfig = dockercontainer.HostConfig{ Binds: runner.Binds, LogConfig: dockercontainer.LogConfig{ @@ -917,6 +981,10 @@ func (runner *ContainerRunner) CreateContainer() error { }, Resources: dockercontainer.Resources{ CgroupParent: runner.setCgroupParent, + NanoCPUs: int64(runner.Container.RuntimeConstraints.VCPUs) * 1000000000, + Memory: maxRAM, // RAM + MemorySwap: maxRAM, // RAM+swap + KernelMemory: maxRAM, // kernel portion }, } @@ -973,46 +1041,38 @@ func (runner *ContainerRunner) StartContainer() error { } return fmt.Errorf("could not start container: %v%s", err, advice) } - runner.cStarted = true return nil } // WaitFinish waits for the container to terminate, capture the exit code, and // close the stdout/stderr logging. -func (runner *ContainerRunner) WaitFinish() (err error) { +func (runner *ContainerRunner) WaitFinish() error { runner.CrunchLog.Print("Waiting for container to finish") - waitOk, waitErr := runner.Docker.ContainerWait(context.TODO(), runner.ContainerID, "not-running") + waitOk, waitErr := runner.Docker.ContainerWait(context.TODO(), runner.ContainerID, dockercontainer.WaitConditionNotRunning) + arvMountExit := runner.ArvMountExit + for { + select { + case waitBody := <-waitOk: + runner.CrunchLog.Printf("Container exited with code: %v", waitBody.StatusCode) + code := int(waitBody.StatusCode) + runner.ExitCode = &code + + // wait for stdout/stderr to complete + <-runner.loggingDone + return nil + + case err := <-waitErr: + return fmt.Errorf("container wait: %v", err) - go func() { - <-runner.ArvMountExit - if runner.cStarted { + case <-arvMountExit: runner.CrunchLog.Printf("arv-mount exited while container is still running. Stopping container.") - runner.stop() + runner.stop(nil) + // arvMountExit will always be ready now that + // it's closed, but that doesn't interest us. + arvMountExit = nil } - }() - - var waitBody dockercontainer.ContainerWaitOKBody - select { - case waitBody = <-waitOk: - case err = <-waitErr: } - - // Container isn't running any more - runner.cStarted = false - - if err != nil { - return fmt.Errorf("container wait: %v", err) - } - - runner.CrunchLog.Printf("Container exited with code: %v", waitBody.StatusCode) - code := int(waitBody.StatusCode) - runner.ExitCode = &code - - // wait for stdout/stderr to complete - <-runner.loggingDone - - return nil } var ErrNotInOutputDir = fmt.Errorf("Must point to path within the output directory") @@ -1086,14 +1146,34 @@ func (runner *ContainerRunner) UploadOutputFile( relocateTo string, followed int) (manifestText string, err error) { - if info.Mode().IsDir() { - return - } - if infoerr != nil { return "", infoerr } + if info.Mode().IsDir() { + // if empty, need to create a .keep file + dir, direrr := os.Open(path) + if direrr != nil { + return "", direrr + } + defer dir.Close() + names, eof := dir.Readdirnames(1) + if len(names) == 0 && eof == io.EOF && path != runner.HostOutputDir { + containerPath := runner.OutputPath + path[len(runner.HostOutputDir):] + for _, bind := range binds { + mnt := runner.Container.Mounts[bind] + // Check if there is a bind for this + // directory, in which case assume we don't need .keep + if (containerPath == bind || strings.HasPrefix(containerPath, bind+"/")) && mnt.PortableDataHash != "d41d8cd98f00b204e9800998ecf8427e+0" { + return + } + } + outputSuffix := path[len(runner.HostOutputDir)+1:] + return fmt.Sprintf("./%v d41d8cd98f00b204e9800998ecf8427e+0 0:0:.keep\n", outputSuffix), nil + } + return + } + if followed >= limitFollowSymlinks { // Got stuck in a loop or just a pathological number of // directory links, give up. @@ -1101,9 +1181,16 @@ func (runner *ContainerRunner) UploadOutputFile( return } - // When following symlinks, the source path may need to be logically - // relocated to some other path within the output collection. Remove - // the relocateFrom prefix and replace it with relocateTo. + // "path" is the actual path we are visiting + // "tgt" is the target of "path" (a non-symlink) after following symlinks + // "relocated" is the path in the output manifest where the file should be placed, + // but has HostOutputDir as a prefix. + + // The destination path in the output manifest may need to be + // logically relocated to some other path in order to appear + // in the correct location as a result of following a symlink. + // Remove the relocateFrom prefix and replace it with + // relocateTo. relocated := relocateTo + path[len(relocateFrom):] tgt, readlinktgt, info, derefErr := runner.derefOutputSymlink(path, info) @@ -1114,7 +1201,7 @@ func (runner *ContainerRunner) UploadOutputFile( // go through mounts and try reverse map to collection reference for _, bind := range binds { mnt := runner.Container.Mounts[bind] - if tgt == bind || strings.HasPrefix(tgt, bind+"/") { + if (tgt == bind || strings.HasPrefix(tgt, bind+"/")) && !mnt.Writable { // get path relative to bind targetSuffix := tgt[len(bind):] @@ -1124,7 +1211,7 @@ func (runner *ContainerRunner) UploadOutputFile( // Terminates in this keep mount, so add the // manifest text at appropriate location. - outputSuffix := path[len(runner.HostOutputDir):] + outputSuffix := relocated[len(runner.HostOutputDir):] manifestText, err = runner.getCollectionManifestForPath(adjustedMount, outputSuffix) return } @@ -1164,10 +1251,6 @@ func (runner *ContainerRunner) UploadOutputFile( // HandleOutput sets the output, unmounts the FUSE mount, and deletes temporary directories func (runner *ContainerRunner) CaptureOutput() error { - if runner.finalState != "Complete" { - return nil - } - if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI { // Output may have been set directly by the container, so // refresh the container record to check. @@ -1201,6 +1284,16 @@ func (runner *ContainerRunner) CaptureOutput() error { } sort.Strings(binds) + // Delete secret mounts so they don't get saved to the output collection. + for bind := range runner.SecretMounts { + if strings.HasPrefix(bind, runner.Container.OutputPath+"/") { + err = os.Remove(runner.HostOutputDir + bind[len(runner.Container.OutputPath):]) + if err != nil { + return fmt.Errorf("Unable to remove secret mount: %v", err) + } + } + } + var manifestText string collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir) @@ -1257,7 +1350,7 @@ func (runner *ContainerRunner) CaptureOutput() error { continue } - if mnt.ExcludeFromOutput == true { + if mnt.ExcludeFromOutput == true || mnt.Writable { continue } @@ -1379,29 +1472,33 @@ func (runner *ContainerRunner) CleanupDirs() { } } - for _, tmpdir := range runner.CleanupTempDir { - if rmerr := os.RemoveAll(tmpdir); rmerr != nil { - runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr) - } + if rmerr := os.RemoveAll(runner.parentTemp); rmerr != nil { + runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", runner.parentTemp, rmerr) } } // CommitLogs posts the collection containing the final container logs. func (runner *ContainerRunner) CommitLogs() error { - runner.CrunchLog.Print(runner.finalState) + func() { + // Hold cStateLock to prevent races on CrunchLog (e.g., stop()). + runner.cStateLock.Lock() + defer runner.cStateLock.Unlock() - if runner.arvMountLog != nil { - runner.arvMountLog.Close() - } - runner.CrunchLog.Close() + runner.CrunchLog.Print(runner.finalState) + + if runner.arvMountLog != nil { + runner.arvMountLog.Close() + } + runner.CrunchLog.Close() - // Closing CrunchLog above allows them to be committed to Keep at this - // point, but re-open crunch log with ArvClient in case there are any - // other further errors (such as failing to write the log to Keep!) - // while shutting down - runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{ArvClient: runner.ArvClient, - UUID: runner.Container.UUID, loggingStream: "crunch-run", writeCloser: nil}) - runner.CrunchLog.Immediate = log.New(os.Stderr, runner.Container.UUID+" ", 0) + // Closing CrunchLog above allows them to be committed to Keep at this + // point, but re-open crunch log with ArvClient in case there are any + // other further errors (such as failing to write the log to Keep!) + // while shutting down + runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{ArvClient: runner.ArvClient, + UUID: runner.Container.UUID, loggingStream: "crunch-run", writeCloser: nil}) + runner.CrunchLog.Immediate = log.New(os.Stderr, runner.Container.UUID+" ", 0) + }() if runner.LogsPDH != nil { // If we have already assigned something to LogsPDH, @@ -1510,7 +1607,6 @@ func (runner *ContainerRunner) Run() (err error) { runner.finalState = "Queued" defer func() { - runner.stopSignals() runner.CleanupDirs() runner.CrunchLog.Printf("crunch-run finished") @@ -1552,6 +1648,7 @@ func (runner *ContainerRunner) Run() (err error) { } checkErr(runner.CaptureOutput()) + checkErr(runner.stopHoststat()) checkErr(runner.CommitLogs()) checkErr(runner.UpdateContainerFinal()) }() @@ -1560,9 +1657,8 @@ func (runner *ContainerRunner) Run() (err error) { if err != nil { return } - - // setup signal handling runner.setupSignals() + runner.startHoststat() // check for and/or load image err = runner.LoadImage() @@ -1611,7 +1707,7 @@ func (runner *ContainerRunner) Run() (err error) { } runner.finalState = "Cancelled" - runner.StartCrunchstat() + runner.startCrunchstat() err = runner.StartContainer() if err != nil { @@ -1620,7 +1716,7 @@ func (runner *ContainerRunner) Run() (err error) { } err = runner.WaitFinish() - if err == nil { + if err == nil && !runner.IsCancelled() { runner.finalState = "Complete" } return @@ -1641,6 +1737,31 @@ func (runner *ContainerRunner) fetchContainerRecord() error { if err != nil { return fmt.Errorf("error decoding container record: %v", err) } + + var sm struct { + SecretMounts map[string]arvados.Mount `json:"secret_mounts"` + } + + containerToken, err := runner.ContainerToken() + if err != nil { + return fmt.Errorf("error getting container token: %v", err) + } + + containerClient, err := runner.MkArvClient(containerToken) + if err != nil { + return fmt.Errorf("error creating container API client: %v", err) + } + + err = containerClient.Call("GET", "containers", runner.Container.UUID, "secret_mounts", nil, &sm) + if err != nil { + if apierr, ok := err.(arvadosclient.APIServerError); !ok || apierr.HttpStatusCode != 404 { + return fmt.Errorf("error fetching secret_mounts: %v", err) + } + // ok && apierr.HttpStatusCode == 404, which means + // secret_mounts isn't supported by this API server. + } + runner.SecretMounts = sm.SecretMounts + return nil } @@ -1654,6 +1775,14 @@ func NewContainerRunner(api IArvadosClient, cr.NewLogWriter = cr.NewArvLogWriter cr.RunArvMount = cr.ArvMountCmd cr.MkTempDir = ioutil.TempDir + cr.MkArvClient = func(token string) (IArvadosClient, error) { + cl, err := arvadosclient.MakeArvadosClient() + if err != nil { + return nil, err + } + cl.ApiToken = token + return cl, nil + } cr.LogCollection = &CollectionWriter{0, kc, nil, nil, sync.Mutex{}} cr.Container.UUID = containerUUID cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run")) @@ -1712,10 +1841,8 @@ func main() { // API version 1.21 corresponds to Docker 1.9, which is currently the // minimum version we want to support. docker, dockererr := dockerclient.NewClient(dockerclient.DefaultDockerHost, "1.21", nil, nil) - dockerClientProxy := ThinDockerClientProxy{Docker: docker} - - cr := NewContainerRunner(api, kc, dockerClientProxy, containerId) + cr := NewContainerRunner(api, kc, docker, containerId) if dockererr != nil { cr.CrunchLog.Printf("%s: %v", containerId, dockererr) cr.checkBrokenNode(dockererr) @@ -1723,6 +1850,12 @@ func main() { os.Exit(1) } + parentTemp, tmperr := cr.MkTempDir("", "crunch-run."+containerId+".") + if tmperr != nil { + log.Fatalf("%s: %v", containerId, tmperr) + } + + cr.parentTemp = parentTemp cr.statInterval = *statInterval cr.cgroupRoot = *cgroupRoot cr.expectCgroupParent = *cgroupParent