X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/bc9845761c44beecdd046620694f6d88af0e32fd..d542de20d8617f5823ab8f675c114f78aaf4a924:/services/crunch-run/crunchrun.go?ds=inline diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index 640ac88ca9..ade40c6b03 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -1,26 +1,37 @@ package main import ( + "encoding/json" "errors" "flag" + "fmt" + "git.curoverse.com/arvados.git/lib/crunchstat" + "git.curoverse.com/arvados.git/sdk/go/arvados" "git.curoverse.com/arvados.git/sdk/go/arvadosclient" "git.curoverse.com/arvados.git/sdk/go/keepclient" "git.curoverse.com/arvados.git/sdk/go/manifest" "github.com/curoverse/dockerclient" "io" + "io/ioutil" "log" "os" + "os/exec" "os/signal" + "path" + "path/filepath" "strings" "sync" "syscall" + "time" ) // IArvadosClient is the minimal Arvados API methods used by crunch-run. type IArvadosClient interface { Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error - Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error) + Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error + Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error + Discovery(key string) (interface{}, error) } // ErrCancelled is the error returned when the container is cancelled. @@ -32,31 +43,13 @@ type IKeepClient interface { ManifestFileReader(m manifest.Manifest, filename string) (keepclient.ReadCloserWithLen, error) } -// Mount describes the mount points to create inside the container. -type Mount struct{} - -// Collection record returned by the API server. -type Collection struct { - ManifestText string `json:"manifest_text"` -} - -// ContainerRecord is the container record returned by the API server. -type ContainerRecord struct { - UUID string `json:"uuid"` - Command []string `json:"command"` - ContainerImage string `json:"container_image"` - Cwd string `json:"cwd"` - Environment map[string]string `json:"environment"` - Mounts map[string]Mount `json:"mounts"` - OutputPath string `json:"output_path"` - Priority int `json:"priority"` - RuntimeConstraints map[string]interface{} `json:"runtime_constraints"` - State string `json:"state"` -} - // NewLogWriter is a factory function to create a new log writer. type NewLogWriter func(name string) io.WriteCloser +type RunArvMount func(args []string, tok string) (*exec.Cmd, error) + +type MkTempDir func(string, string) (string, error) + // ThinDockerClient is the minimal Docker client interface used by crunch-run. type ThinDockerClient interface { StopContainer(id string, timeout int) error @@ -64,7 +57,7 @@ type ThinDockerClient interface { LoadImage(reader io.Reader) error CreateContainer(config *dockerclient.ContainerConfig, name string, authConfig *dockerclient.AuthConfig) (string, error) StartContainer(id string, config *dockerclient.HostConfig) error - ContainerLogs(id string, options *dockerclient.LogOptions) (io.ReadCloser, error) + AttachContainer(id string, options *dockerclient.AttachOptions) (io.ReadCloser, error) Wait(id string) <-chan dockerclient.WaitResult RemoveImage(name string, force bool) ([]*dockerclient.ImageDelete, error) } @@ -75,33 +68,63 @@ type ContainerRunner struct { Docker ThinDockerClient ArvClient IArvadosClient Kc IKeepClient - ContainerRecord + arvados.Container dockerclient.ContainerConfig + dockerclient.HostConfig + token string ContainerID string ExitCode *int NewLogWriter loggingDone chan bool CrunchLog *ThrottledLogger - Stdout *ThrottledLogger + Stdout io.WriteCloser Stderr *ThrottledLogger LogCollection *CollectionWriter LogsPDH *string - CancelLock sync.Mutex - Cancelled bool - SigChan chan os.Signal - finalState string + RunArvMount + MkTempDir + ArvMount *exec.Cmd + ArvMountPoint string + HostOutputDir string + CleanupTempDir []string + Binds []string + OutputPDH *string + CancelLock sync.Mutex + Cancelled bool + SigChan chan os.Signal + ArvMountExit chan error + finalState string + trashLifetime time.Duration + + statLogger io.WriteCloser + statReporter *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 + // parent. Note: Ideally we would use the same field for both + // expectCgroupParent and setCgroupParent, and just make it + // default to "docker". However, when using docker < 1.10 with + // systemd, specifying a non-empty cgroup parent (even the + // default value "docker") hits a docker bug + // (https://github.com/docker/docker/issues/17126). Using two + // separate fields makes it possible to use the "expect cgroup + // parent to be X" feature even on sites where the "specify + // cgroup parent" feature breaks. + setCgroupParent string } // SetupSignals sets up signal handling to gracefully terminate the underlying // Docker container and update state when receiving a TERM, INT or QUIT signal. -func (runner *ContainerRunner) SetupSignals() error { +func (runner *ContainerRunner) SetupSignals() { runner.SigChan = make(chan os.Signal, 1) signal.Notify(runner.SigChan, syscall.SIGTERM) signal.Notify(runner.SigChan, syscall.SIGINT) signal.Notify(runner.SigChan, syscall.SIGQUIT) go func(sig <-chan os.Signal) { - for _ = range sig { + for range sig { if !runner.Cancelled { runner.CancelLock.Lock() runner.Cancelled = true @@ -112,8 +135,6 @@ func (runner *ContainerRunner) SetupSignals() error { } } }(runner.SigChan) - - return nil } // LoadImage determines the docker image id from the container record and @@ -121,19 +142,19 @@ func (runner *ContainerRunner) SetupSignals() error { // the image from Keep. func (runner *ContainerRunner) LoadImage() (err error) { - runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.ContainerRecord.ContainerImage) + runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.Container.ContainerImage) - var collection Collection - err = runner.ArvClient.Get("collections", runner.ContainerRecord.ContainerImage, nil, &collection) + var collection arvados.Collection + err = runner.ArvClient.Get("collections", runner.Container.ContainerImage, nil, &collection) if err != nil { - return err + return fmt.Errorf("While getting container image collection: %v", err) } manifest := manifest.Manifest{Text: collection.ManifestText} var img, imageID string for ms := range manifest.StreamIter() { img = ms.FileStreamSegments[0].Name if !strings.HasSuffix(img, ".tar") { - return errors.New("First file in the collection does not end in .tar") + return fmt.Errorf("First file in the container image collection does not end in .tar") } imageID = img[:len(img)-4] } @@ -147,12 +168,12 @@ func (runner *ContainerRunner) LoadImage() (err error) { var readCloser io.ReadCloser readCloser, err = runner.Kc.ManifestFileReader(manifest, img) if err != nil { - return err + return fmt.Errorf("While creating ManifestFileReader for container image: %v", err) } err = runner.Docker.LoadImage(readCloser) if err != nil { - return err + return fmt.Errorf("While loading container image into Docker: %v", err) } } else { runner.CrunchLog.Print("Docker image is available") @@ -163,85 +184,486 @@ func (runner *ContainerRunner) LoadImage() (err error) { return nil } -// StartContainer creates the container and runs it. -func (runner *ContainerRunner) StartContainer() (err error) { - runner.CrunchLog.Print("Creating Docker container") +func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string, token string) (c *exec.Cmd, err error) { + c = exec.Command("arv-mount", arvMountCmd...) - runner.CancelLock.Lock() - defer runner.CancelLock.Unlock() + // Copy our environment, but override ARVADOS_API_TOKEN with + // the container auth token. + c.Env = nil + for _, s := range os.Environ() { + if !strings.HasPrefix(s, "ARVADOS_API_TOKEN=") { + c.Env = append(c.Env, s) + } + } + c.Env = append(c.Env, "ARVADOS_API_TOKEN="+token) - if runner.Cancelled { - return ErrCancelled + nt := NewThrottledLogger(runner.NewLogWriter("arv-mount")) + c.Stdout = nt + c.Stderr = nt + + err = c.Start() + if err != nil { + return nil, err } - runner.ContainerConfig.Cmd = runner.ContainerRecord.Command - if runner.ContainerRecord.Cwd != "." { - runner.ContainerConfig.WorkingDir = runner.ContainerRecord.Cwd + statReadme := make(chan bool) + runner.ArvMountExit = make(chan error) + + keepStatting := true + go func() { + for keepStatting { + time.Sleep(100 * time.Millisecond) + _, err = os.Stat(fmt.Sprintf("%s/by_id/README", runner.ArvMountPoint)) + if err == nil { + keepStatting = false + statReadme <- true + } + } + close(statReadme) + }() + + go func() { + runner.ArvMountExit <- c.Wait() + close(runner.ArvMountExit) + }() + + select { + case <-statReadme: + break + case err := <-runner.ArvMountExit: + runner.ArvMount = nil + keepStatting = false + return nil, err } - for k, v := range runner.ContainerRecord.Environment { - runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v) + + return c, nil +} + +func (runner *ContainerRunner) SetupMounts() (err error) { + runner.ArvMountPoint, err = runner.MkTempDir("", "keep") + if err != nil { + return fmt.Errorf("While creating keep mount temp dir: %v", err) } - runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil) + + runner.CleanupTempDir = append(runner.CleanupTempDir, runner.ArvMountPoint) + + pdhOnly := true + tmpcount := 0 + arvMountCmd := []string{"--foreground", "--allow-other", "--read-write"} + + if runner.Container.RuntimeConstraints.KeepCacheRAM > 0 { + arvMountCmd = append(arvMountCmd, "--file-cache", fmt.Sprintf("%d", runner.Container.RuntimeConstraints.KeepCacheRAM)) + } + + collectionPaths := []string{} + runner.Binds = nil + + for bind, mnt := range runner.Container.Mounts { + if bind == "stdout" { + // Is it a "file" mount kind? + if mnt.Kind != "file" { + return fmt.Errorf("Unsupported mount kind '%s' for stdout. Only 'file' is supported.", mnt.Kind) + } + + // Does path start with OutputPath? + prefix := runner.Container.OutputPath + if !strings.HasSuffix(prefix, "/") { + prefix += "/" + } + if !strings.HasPrefix(mnt.Path, prefix) { + return fmt.Errorf("Stdout path does not start with OutputPath: %s, %s", mnt.Path, prefix) + } + } + + switch { + case mnt.Kind == "collection": + var src string + if mnt.UUID != "" && mnt.PortableDataHash != "" { + return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount") + } + if mnt.UUID != "" { + if mnt.Writable { + return fmt.Errorf("Writing to existing collections currently not permitted.") + } + pdhOnly = false + src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID) + } else if mnt.PortableDataHash != "" { + if mnt.Writable { + return fmt.Errorf("Can never write to a collection specified by portable data hash") + } + src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash) + } else { + src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount) + arvMountCmd = append(arvMountCmd, "--mount-tmp") + arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%d", tmpcount)) + tmpcount += 1 + } + if mnt.Writable { + if bind == runner.Container.OutputPath { + runner.HostOutputDir = src + } + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind)) + } else { + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind)) + } + collectionPaths = append(collectionPaths, src) + + case mnt.Kind == "tmp" && bind == runner.Container.OutputPath: + runner.HostOutputDir, err = runner.MkTempDir("", "") + if err != nil { + return fmt.Errorf("While creating mount temp dir: %v", err) + } + st, staterr := os.Stat(runner.HostOutputDir) + if staterr != nil { + return fmt.Errorf("While Stat on temp dir: %v", staterr) + } + err = os.Chmod(runner.HostOutputDir, st.Mode()|os.ModeSetgid|0777) + if staterr != nil { + return fmt.Errorf("While Chmod temp dir: %v", err) + } + runner.CleanupTempDir = append(runner.CleanupTempDir, runner.HostOutputDir) + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind)) + + case mnt.Kind == "tmp": + runner.Binds = append(runner.Binds, bind) + + case mnt.Kind == "json": + jsondata, err := json.Marshal(mnt.Content) + if err != nil { + return fmt.Errorf("encoding json data: %v", err) + } + // 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("", "") + 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) + if err != nil { + return fmt.Errorf("writing temp file: %v", err) + } + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind)) + } + } + + if runner.HostOutputDir == "" { + return fmt.Errorf("Output path does not correspond to a writable mount point") + } + + if pdhOnly { + arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id") + } else { + arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id") + } + arvMountCmd = append(arvMountCmd, runner.ArvMountPoint) + + token, err := runner.ContainerToken() if err != nil { - return + return fmt.Errorf("could not get container token: %s", err) } - hostConfig := &dockerclient.HostConfig{} - runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID) - err = runner.Docker.StartContainer(runner.ContainerID, hostConfig) + runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token) if err != nil { - return + return fmt.Errorf("While trying to start arv-mount: %v", err) + } + + for _, p := range collectionPaths { + _, err = os.Stat(p) + if err != nil { + return fmt.Errorf("While checking that input files exist: %v", 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 + + 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) + } + } + + 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) + } + + closeerr = runner.Stderr.Close() + if closeerr != nil { + runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr) + } + + if runner.statReporter != nil { + runner.statReporter.Stop() + closeerr = runner.statLogger.Close() + if closeerr != nil { + runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr) + } + } + + runner.loggingDone <- true + close(runner.loggingDone) + return + } + } +} + +func (runner *ContainerRunner) StartCrunchstat() { + runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat")) + runner.statReporter = &crunchstat.Reporter{ + CID: runner.ContainerID, + Logger: log.New(runner.statLogger, "", 0), + CgroupParent: runner.expectCgroupParent, + CgroupRoot: runner.cgroupRoot, + PollPeriod: runner.statInterval, + } + runner.statReporter.Start() +} + // AttachLogs connects the docker container stdout and stderr logs to the // Arvados logger which logs to Keep and the API server logs table. -func (runner *ContainerRunner) AttachLogs() (err error) { +func (runner *ContainerRunner) AttachStreams() (err error) { - runner.CrunchLog.Print("Attaching container logs") + runner.CrunchLog.Print("Attaching container streams") - var stderrReader, stdoutReader io.Reader - stderrReader, err = runner.Docker.ContainerLogs(runner.ContainerID, &dockerclient.LogOptions{Follow: true, Stderr: true}) + var containerReader io.Reader + containerReader, err = runner.Docker.AttachContainer(runner.ContainerID, + &dockerclient.AttachOptions{Stream: true, Stdout: true, Stderr: true}) if err != nil { - return - } - stdoutReader, err = runner.Docker.ContainerLogs(runner.ContainerID, &dockerclient.LogOptions{Follow: true, Stdout: true}) - if err != nil { - return + return fmt.Errorf("While attaching container stdout/stderr streams: %v", err) } runner.loggingDone = make(chan bool) - runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout")) + if stdoutMnt, ok := runner.Container.Mounts["stdout"]; ok { + stdoutPath := stdoutMnt.Path[len(runner.Container.OutputPath):] + index := strings.LastIndex(stdoutPath, "/") + if index > 0 { + subdirs := stdoutPath[:index] + if subdirs != "" { + st, err := os.Stat(runner.HostOutputDir) + if err != nil { + return fmt.Errorf("While Stat on temp dir: %v", err) + } + stdoutPath := path.Join(runner.HostOutputDir, subdirs) + err = os.MkdirAll(stdoutPath, st.Mode()|os.ModeSetgid|0777) + if err != nil { + return fmt.Errorf("While MkdirAll %q: %v", stdoutPath, err) + } + } + } + stdoutFile, err := os.Create(path.Join(runner.HostOutputDir, stdoutPath)) + if err != nil { + return fmt.Errorf("While creating stdout file: %v", err) + } + runner.Stdout = stdoutFile + } else { + runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout")) + } runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr")) - go ReadWriteLines(stdoutReader, runner.Stdout, runner.loggingDone) - go ReadWriteLines(stderrReader, runner.Stderr, runner.loggingDone) + go runner.ProcessDockerAttach(containerReader) + + return nil +} + +// CreateContainer creates the docker container. +func (runner *ContainerRunner) CreateContainer() error { + runner.CrunchLog.Print("Creating Docker container") + + runner.ContainerConfig.Cmd = runner.Container.Command + if runner.Container.Cwd != "." { + runner.ContainerConfig.WorkingDir = runner.Container.Cwd + } + + for k, v := range runner.Container.Environment { + runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v) + } + if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI { + tok, err := runner.ContainerToken() + if err != nil { + return err + } + runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, + "ARVADOS_API_TOKEN="+tok, + "ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"), + "ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"), + ) + runner.ContainerConfig.NetworkDisabled = false + } else { + runner.ContainerConfig.NetworkDisabled = true + } + + var err error + runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil) + if err != nil { + return fmt.Errorf("While creating container: %v", err) + } + + runner.HostConfig = dockerclient.HostConfig{ + Binds: runner.Binds, + CgroupParent: runner.setCgroupParent, + LogConfig: dockerclient.LogConfig{ + Type: "none", + }, + } + + return runner.AttachStreams() +} + +// StartContainer starts the docker container created by CreateContainer. +func (runner *ContainerRunner) StartContainer() error { + runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID) + err := runner.Docker.StartContainer(runner.ContainerID, &runner.HostConfig) + if err != nil { + return fmt.Errorf("could not start container: %v", err) + } return nil } // WaitFinish waits for the container to terminate, capture the exit code, and // close the stdout/stderr logging. func (runner *ContainerRunner) WaitFinish() error { + runner.CrunchLog.Print("Waiting for container to finish") + result := runner.Docker.Wait(runner.ContainerID) wr := <-result if wr.Error != nil { - return wr.Error + return fmt.Errorf("While waiting for container to finish: %v", wr.Error) } runner.ExitCode = &wr.ExitCode - // drain stdout/stderr - <-runner.loggingDone + // wait for stdout/stderr to complete <-runner.loggingDone - runner.Stdout.Close() - runner.Stderr.Close() + return nil +} + +// 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. + err := runner.ArvClient.Get("containers", runner.Container.UUID, + nil, &runner.Container) + if err != nil { + return err + } + if runner.Container.Output != "" { + // Container output is already set. + runner.OutputPDH = &runner.Container.Output + return nil + } + } + + if runner.HostOutputDir == "" { + return nil + } + + _, err := os.Stat(runner.HostOutputDir) + if err != nil { + return fmt.Errorf("While checking host output path: %v", err) + } + + var manifestText string + + collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir) + _, err = os.Stat(collectionMetafile) + if err != nil { + // Regular directory + cw := CollectionWriter{runner.Kc, nil, sync.Mutex{}} + manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger) + if err != nil { + return fmt.Errorf("While uploading output files: %v", err) + } + } else { + // FUSE mount directory + file, openerr := os.Open(collectionMetafile) + if openerr != nil { + return fmt.Errorf("While opening FUSE metafile: %v", err) + } + defer file.Close() + + var rec arvados.Collection + err = json.NewDecoder(file).Decode(&rec) + if err != nil { + return fmt.Errorf("While reading FUSE metafile: %v", err) + } + manifestText = rec.ManifestText + } + var response arvados.Collection + err = runner.ArvClient.Create("collections", + arvadosclient.Dict{ + "collection": arvadosclient.Dict{ + "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339), + "name": "output for " + runner.Container.UUID, + "manifest_text": manifestText}}, + &response) + if err != nil { + return fmt.Errorf("While creating output collection: %v", err) + } + runner.OutputPDH = &response.PortableDataHash return nil } +func (runner *ContainerRunner) loadDiscoveryVars() { + tl, err := runner.ArvClient.Discovery("defaultTrashLifetime") + if err != nil { + log.Fatalf("getting defaultTrashLifetime from discovery document: %s", err) + } + runner.trashLifetime = time.Duration(tl.(float64)) * time.Second +} + +func (runner *ContainerRunner) CleanupDirs() { + if runner.ArvMount != nil { + umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint) + umnterr := umount.Run() + if umnterr != nil { + runner.CrunchLog.Printf("While running fusermount: %v", umnterr) + } + + mnterr := <-runner.ArvMountExit + if mnterr != nil { + runner.CrunchLog.Printf("Arv-mount exit error: %v", mnterr) + } + } + + for _, tmpdir := range runner.CleanupTempDir { + rmerr := os.RemoveAll(tmpdir) + if rmerr != nil { + runner.CrunchLog.Printf("While cleaning up temporary directory %s: %v", tmpdir, rmerr) + } + } +} + // CommitLogs posts the collection containing the final container logs. func (runner *ContainerRunner) CommitLogs() error { runner.CrunchLog.Print(runner.finalState) @@ -251,183 +673,264 @@ func (runner *ContainerRunner) CommitLogs() error { // point, but re-open crunch log with ArvClient in case there are any // other further (such as failing to write the log to Keep!) while // shutting down - runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.ContainerRecord.UUID, + runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID, "crunch-run", nil}) + if runner.LogsPDH != nil { + // If we have already assigned something to LogsPDH, + // we must be closing the re-opened log, which won't + // end up getting attached to the container record and + // therefore doesn't need to be saved as a collection + // -- it exists only to send logs to other channels. + return nil + } + mt, err := runner.LogCollection.ManifestText() if err != nil { - return err + return fmt.Errorf("While creating log manifest: %v", err) } - response := make(map[string]string) + var response arvados.Collection err = runner.ArvClient.Create("collections", - arvadosclient.Dict{"name": "logs for " + runner.ContainerRecord.UUID, - "manifest_text": mt}, - response) + arvadosclient.Dict{ + "collection": arvadosclient.Dict{ + "expires_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339), + "name": "logs for " + runner.Container.UUID, + "manifest_text": mt}}, + &response) if err != nil { - return err + return fmt.Errorf("While creating log collection: %v", err) } - - runner.LogsPDH = new(string) - *runner.LogsPDH = response["portable_data_hash"] - + runner.LogsPDH = &response.PortableDataHash return nil } -// UpdateContainerRecordRunning updates the container state to "Running" -func (runner *ContainerRunner) UpdateContainerRecordRunning() error { - update := arvadosclient.Dict{"state": "Running"} - return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID, update, nil) +// UpdateContainerRunning updates the container state to "Running" +func (runner *ContainerRunner) UpdateContainerRunning() error { + runner.CancelLock.Lock() + defer runner.CancelLock.Unlock() + if runner.Cancelled { + return ErrCancelled + } + return runner.ArvClient.Update("containers", runner.Container.UUID, + arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil) } -// UpdateContainerRecordComplete updates the container record state on API -// server to "Complete" or "Cancelled" -func (runner *ContainerRunner) UpdateContainerRecordComplete() error { - update := arvadosclient.Dict{} - if runner.LogsPDH != nil { - update["log"] = *runner.LogsPDH +// ContainerToken returns the api_token the container (and any +// arv-mount processes) are allowed to use. +func (runner *ContainerRunner) ContainerToken() (string, error) { + if runner.token != "" { + return runner.token, nil } - if runner.ExitCode != nil { - update["exit_code"] = *runner.ExitCode + + var auth arvados.APIClientAuthorization + err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth) + if err != nil { + return "", err } + runner.token = auth.APIToken + return runner.token, nil +} +// UpdateContainerComplete updates the container record state on API +// server to "Complete" or "Cancelled" +func (runner *ContainerRunner) UpdateContainerFinal() error { + update := arvadosclient.Dict{} update["state"] = runner.finalState + if runner.finalState == "Complete" { + if runner.LogsPDH != nil { + update["log"] = *runner.LogsPDH + } + if runner.ExitCode != nil { + update["exit_code"] = *runner.ExitCode + } + if runner.OutputPDH != nil { + update["output"] = *runner.OutputPDH + } + } + return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil) +} - return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID, update, nil) +// IsCancelled returns the value of Cancelled, with goroutine safety. +func (runner *ContainerRunner) IsCancelled() bool { + runner.CancelLock.Lock() + defer runner.CancelLock.Unlock() + return runner.Cancelled } // NewArvLogWriter creates an ArvLogWriter func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser { - return &ArvLogWriter{runner.ArvClient, runner.ContainerRecord.UUID, name, runner.LogCollection.Open(name + ".txt")} + return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")} } // Run the full container lifecycle. -func (runner *ContainerRunner) Run(containerUUID string) (err error) { - runner.CrunchLog.Printf("Executing container '%s'", containerUUID) +func (runner *ContainerRunner) Run() (err error) { + runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID) + + hostname, hosterr := os.Hostname() + if hosterr != nil { + runner.CrunchLog.Printf("Error getting hostname '%v'", hosterr) + } else { + runner.CrunchLog.Printf("Executing on host '%s'", hostname) + } + + // Clean up temporary directories _after_ finalizing + // everything (if we've made any by then) + defer runner.CleanupDirs() - var runerr, waiterr error + runner.finalState = "Queued" defer func() { - if err != nil { - runner.CrunchLog.Print(err) + // checkErr prints e (unless it's nil) and sets err to + // e (unless err is already non-nil). Thus, if err + // hasn't already been assigned when Run() returns, + // this cleanup func will cause Run() to return the + // first non-nil error that is passed to checkErr(). + checkErr := func(e error) { + if e == nil { + return + } + runner.CrunchLog.Print(e) + if err == nil { + err = e + } } - if runner.Cancelled { - runner.finalState = "Cancelled" - } else { - runner.finalState = "Complete" - } + // Log the error encountered in Run(), if any + checkErr(err) - // (6) write logs - logerr := runner.CommitLogs() - if logerr != nil { - runner.CrunchLog.Print(logerr) + if runner.finalState == "Queued" { + runner.UpdateContainerFinal() + return } - // (7) update container record with results - updateerr := runner.UpdateContainerRecordComplete() - if updateerr != nil { - runner.CrunchLog.Print(updateerr) + if runner.IsCancelled() { + runner.finalState = "Cancelled" + // but don't return yet -- we still want to + // capture partial output and write logs } - runner.CrunchLog.Close() + checkErr(runner.CaptureOutput()) + checkErr(runner.CommitLogs()) + checkErr(runner.UpdateContainerFinal()) - if err == nil { - if runerr != nil { - err = runerr - } else if waiterr != nil { - err = runerr - } else if logerr != nil { - err = logerr - } else if updateerr != nil { - err = updateerr - } - } + // The real log is already closed, but then we opened + // a new one in case we needed to log anything while + // finalizing. + runner.CrunchLog.Close() }() - err = runner.ArvClient.Get("containers", containerUUID, nil, &runner.ContainerRecord) + err = runner.ArvClient.Get("containers", runner.Container.UUID, nil, &runner.Container) if err != nil { + err = fmt.Errorf("While getting container record: %v", err) return } - // (0) setup signal handling - err = runner.SetupSignals() + // setup signal handling + runner.SetupSignals() + + // check for and/or load image + err = runner.LoadImage() if err != nil { + err = fmt.Errorf("While loading container image: %v", err) return } - // (1) check for and/or load image - err = runner.LoadImage() + // set up FUSE mount and binds + err = runner.SetupMounts() if err != nil { + err = fmt.Errorf("While setting up mounts: %v", err) return } - // (2) start container - err = runner.StartContainer() + err = runner.CreateContainer() if err != nil { - if err == ErrCancelled { - err = nil - } return } - // (3) update container record state - err = runner.UpdateContainerRecordRunning() - if err != nil { - runner.CrunchLog.Print(err) + runner.StartCrunchstat() + + if runner.IsCancelled() { + return } - // (4) attach container logs - runerr = runner.AttachLogs() - if runerr != nil { - runner.CrunchLog.Print(runerr) + err = runner.UpdateContainerRunning() + if err != nil { + return } + runner.finalState = "Cancelled" - // (5) wait for container to finish - waiterr = runner.WaitFinish() + err = runner.StartContainer() + if err != nil { + return + } + err = runner.WaitFinish() + if err == nil { + runner.finalState = "Complete" + } return } // NewContainerRunner creates a new container runner. func NewContainerRunner(api IArvadosClient, kc IKeepClient, - docker ThinDockerClient) *ContainerRunner { + docker ThinDockerClient, + containerUUID string) *ContainerRunner { cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker} cr.NewLogWriter = cr.NewArvLogWriter - cr.LogCollection = &CollectionWriter{kc, nil} + cr.RunArvMount = cr.ArvMountCmd + cr.MkTempDir = ioutil.TempDir + cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}} + cr.Container.UUID = containerUUID cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run")) + cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0) + cr.loadDiscoveryVars() return cr } func main() { + statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting") + cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree") + cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)") + cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container") flag.Parse() + containerId := flag.Arg(0) + api, err := arvadosclient.MakeArvadosClient() if err != nil { - log.Fatal(err) + log.Fatalf("%s: %v", containerId, err) } api.Retries = 8 var kc *keepclient.KeepClient - kc, err = keepclient.MakeKeepClient(&api) + kc, err = keepclient.MakeKeepClient(api) if err != nil { - log.Fatal(err) + log.Fatalf("%s: %v", containerId, err) } kc.Retries = 4 var docker *dockerclient.DockerClient docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil) if err != nil { - log.Fatal(err) + log.Fatalf("%s: %v", containerId, err) } - cr := NewContainerRunner(api, kc, docker) + cr := NewContainerRunner(api, kc, docker, containerId) + cr.statInterval = *statInterval + cr.cgroupRoot = *cgroupRoot + cr.expectCgroupParent = *cgroupParent + if *cgroupParentSubsystem != "" { + p := findCgroup(*cgroupParentSubsystem) + cr.setCgroupParent = p + cr.expectCgroupParent = p + } - err = cr.Run(flag.Arg(0)) + err = cr.Run() if err != nil { - log.Fatal(err) + log.Fatalf("%s: %v", containerId, err) } }