X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d10c79192b333191796d949841ec792e61a6006c..8d61948fecccfc60db2f18ba4daf7c01ddf3d3c8:/services/crunch-run/crunchrun.go diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index f55834566d..0b4eb2bcbf 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -5,6 +5,8 @@ import ( "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" @@ -16,6 +18,7 @@ import ( "os/exec" "os/signal" "path" + "path/filepath" "strings" "sync" "syscall" @@ -39,47 +42,6 @@ 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 { - Kind string `json:"kind"` - Writable bool `json:"writable"` - PortableDataHash string `json:"portable_data_hash"` - UUID string `json:"uuid"` - DeviceType string `json:"device_type"` - Path string `json:"path"` -} - -// Collection record returned by the API server. -type CollectionRecord struct { - ManifestText string `json:"manifest_text"` - PortableDataHash string `json:"portable_data_hash"` -} - -type RuntimeConstraints struct { - API *bool -} - -// 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 RuntimeConstraints `json:"runtime_constraints"` - State string `json:"state"` - Output string `json:"output"` -} - -// APIClientAuthorization is an arvados#api_client_authorization resource. -type APIClientAuthorization struct { - UUID string `json:"uuid"` - APIToken string `json:"api_token"` -} - // NewLogWriter is a factory function to create a new log writer. type NewLogWriter func(name string) io.WriteCloser @@ -105,7 +67,7 @@ type ContainerRunner struct { Docker ThinDockerClient ArvClient IArvadosClient Kc IKeepClient - ContainerRecord + arvados.Container dockerclient.ContainerConfig dockerclient.HostConfig token string @@ -131,6 +93,24 @@ type ContainerRunner struct { SigChan chan os.Signal ArvMountExit chan error finalState string + + 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 @@ -142,7 +122,7 @@ func (runner *ContainerRunner) SetupSignals() { 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 @@ -160,10 +140,10 @@ func (runner *ContainerRunner) SetupSignals() { // 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 CollectionRecord - 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 fmt.Errorf("While getting container image collection: %v", err) } @@ -271,7 +251,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { collectionPaths := []string{} runner.Binds = nil - for bind, mnt := range runner.ContainerRecord.Mounts { + for bind, mnt := range runner.Container.Mounts { if bind == "stdout" { // Is it a "file" mount kind? if mnt.Kind != "file" { @@ -279,7 +259,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } // Does path start with OutputPath? - prefix := runner.ContainerRecord.OutputPath + prefix := runner.Container.OutputPath if !strings.HasSuffix(prefix, "/") { prefix += "/" } @@ -288,7 +268,8 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } } - if mnt.Kind == "collection" { + 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") @@ -311,7 +292,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { tmpcount += 1 } if mnt.Writable { - if bind == runner.ContainerRecord.OutputPath { + if bind == runner.Container.OutputPath { runner.HostOutputDir = src } runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind)) @@ -319,25 +300,47 @@ func (runner *ContainerRunner) SetupMounts() (err error) { runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind)) } collectionPaths = append(collectionPaths, src) - } else if mnt.Kind == "tmp" { - if bind == runner.ContainerRecord.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)) - } else { - runner.Binds = append(runner.Binds, bind) + + 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)) } } @@ -406,6 +409,14 @@ func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) { 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 @@ -413,6 +424,18 @@ func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) { } } +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) AttachStreams() (err error) { @@ -428,8 +451,8 @@ func (runner *ContainerRunner) AttachStreams() (err error) { runner.loggingDone = make(chan bool) - if stdoutMnt, ok := runner.ContainerRecord.Mounts["stdout"]; ok { - stdoutPath := stdoutMnt.Path[len(runner.ContainerRecord.OutputPath):] + 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] @@ -464,15 +487,15 @@ func (runner *ContainerRunner) AttachStreams() (err error) { func (runner *ContainerRunner) CreateContainer() error { runner.CrunchLog.Print("Creating Docker container") - runner.ContainerConfig.Cmd = runner.ContainerRecord.Command - if runner.ContainerRecord.Cwd != "." { - runner.ContainerConfig.WorkingDir = runner.ContainerRecord.Cwd + runner.ContainerConfig.Cmd = runner.Container.Command + if runner.Container.Cwd != "." { + runner.ContainerConfig.WorkingDir = runner.Container.Cwd } - for k, v := range runner.ContainerRecord.Environment { + for k, v := range runner.Container.Environment { runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v) } - if wantAPI := runner.ContainerRecord.RuntimeConstraints.API; wantAPI != nil && *wantAPI { + if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI { tok, err := runner.ContainerToken() if err != nil { return err @@ -482,18 +505,24 @@ func (runner *ContainerRunner) CreateContainer() error { "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 } - 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, - LogConfig: dockerclient.LogConfig{Type: "none"}} + runner.HostConfig = dockerclient.HostConfig{ + Binds: runner.Binds, + CgroupParent: runner.setCgroupParent, + LogConfig: dockerclient.LogConfig{ + Type: "none", + }, + } return runner.AttachStreams() } @@ -560,7 +589,7 @@ func (runner *ContainerRunner) CaptureOutput() error { } defer file.Close() - rec := CollectionRecord{} + var rec arvados.Collection err = json.NewDecoder(file).Decode(&rec) if err != nil { return fmt.Errorf("While reading FUSE metafile: %v", err) @@ -568,7 +597,7 @@ func (runner *ContainerRunner) CaptureOutput() error { manifestText = rec.ManifestText } - var response CollectionRecord + var response arvados.Collection err = runner.ArvClient.Create("collections", arvadosclient.Dict{ "collection": arvadosclient.Dict{ @@ -615,7 +644,7 @@ 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 { @@ -632,11 +661,11 @@ func (runner *ContainerRunner) CommitLogs() error { return fmt.Errorf("While creating log manifest: %v", err) } - var response CollectionRecord + var response arvados.Collection err = runner.ArvClient.Create("collections", arvadosclient.Dict{ "collection": arvadosclient.Dict{ - "name": "logs for " + runner.ContainerRecord.UUID, + "name": "logs for " + runner.Container.UUID, "manifest_text": mt}}, &response) if err != nil { @@ -648,14 +677,14 @@ func (runner *ContainerRunner) CommitLogs() error { return nil } -// UpdateContainerRecordRunning updates the container state to "Running" -func (runner *ContainerRunner) UpdateContainerRecordRunning() error { +// 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.ContainerRecord.UUID, + return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": arvadosclient.Dict{"state": "Running"}}, nil) } @@ -666,8 +695,8 @@ func (runner *ContainerRunner) ContainerToken() (string, error) { return runner.token, nil } - var auth APIClientAuthorization - err := runner.ArvClient.Call("GET", "containers", runner.ContainerRecord.UUID, "auth", nil, &auth) + var auth arvados.APIClientAuthorization + err := runner.ArvClient.Call("GET", "containers", runner.Container.UUID, "auth", nil, &auth) if err != nil { return "", err } @@ -675,9 +704,9 @@ func (runner *ContainerRunner) ContainerToken() (string, error) { return runner.token, nil } -// UpdateContainerRecordComplete updates the container record state on API +// UpdateContainerComplete updates the container record state on API // server to "Complete" or "Cancelled" -func (runner *ContainerRunner) UpdateContainerRecordFinal() error { +func (runner *ContainerRunner) UpdateContainerFinal() error { update := arvadosclient.Dict{} update["state"] = runner.finalState if runner.finalState == "Complete" { @@ -691,7 +720,7 @@ func (runner *ContainerRunner) UpdateContainerRecordFinal() error { update["output"] = *runner.OutputPDH } } - return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID, arvadosclient.Dict{"container": update}, nil) + return runner.ArvClient.Update("containers", runner.Container.UUID, arvadosclient.Dict{"container": update}, nil) } // IsCancelled returns the value of Cancelled, with goroutine safety. @@ -703,12 +732,12 @@ func (runner *ContainerRunner) IsCancelled() bool { // 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() (err error) { - runner.CrunchLog.Printf("Executing container '%s'", runner.ContainerRecord.UUID) + runner.CrunchLog.Printf("Executing container '%s'", runner.Container.UUID) hostname, hosterr := os.Hostname() if hosterr != nil { @@ -743,7 +772,7 @@ func (runner *ContainerRunner) Run() (err error) { checkErr(err) if runner.finalState == "Queued" { - runner.UpdateContainerRecordFinal() + runner.UpdateContainerFinal() return } @@ -755,7 +784,7 @@ func (runner *ContainerRunner) Run() (err error) { checkErr(runner.CaptureOutput()) checkErr(runner.CommitLogs()) - checkErr(runner.UpdateContainerRecordFinal()) + checkErr(runner.UpdateContainerFinal()) // The real log is already closed, but then we opened // a new one in case we needed to log anything while @@ -763,7 +792,7 @@ func (runner *ContainerRunner) Run() (err error) { runner.CrunchLog.Close() }() - err = runner.ArvClient.Get("containers", runner.ContainerRecord.UUID, 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 @@ -791,11 +820,13 @@ func (runner *ContainerRunner) Run() (err error) { return } + runner.StartCrunchstat() + if runner.IsCancelled() { return } - err = runner.UpdateContainerRecordRunning() + err = runner.UpdateContainerRunning() if err != nil { return } @@ -824,13 +855,17 @@ func NewContainerRunner(api IArvadosClient, cr.RunArvMount = cr.ArvMountCmd cr.MkTempDir = ioutil.TempDir cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}} - cr.ContainerRecord.UUID = containerUUID + cr.Container.UUID = containerUUID cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run")) cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0) 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) @@ -855,6 +890,14 @@ func main() { } 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() if err != nil {