X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/6b2f232c3d73a023d64112c609a28c8ff9cc27de..f9e3d32c92920a806d50548dbb9b6c0aab7d8c54:/services/crunch-run/crunchrun.go diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index a1b246c841..2fd002907c 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -5,6 +5,7 @@ import ( "errors" "flag" "fmt" + "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" @@ -39,43 +40,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"` -} - -// 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"` - 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 @@ -101,8 +65,10 @@ type ContainerRunner struct { Docker ThinDockerClient ArvClient IArvadosClient Kc IKeepClient - ContainerRecord + arvados.Container dockerclient.ContainerConfig + dockerclient.HostConfig + token string ContainerID string ExitCode *int NewLogWriter @@ -136,7 +102,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 @@ -154,10 +120,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) } @@ -265,7 +231,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" { @@ -273,7 +239,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 += "/" } @@ -305,7 +271,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)) @@ -314,7 +280,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } collectionPaths = append(collectionPaths, src) } else if mnt.Kind == "tmp" { - if bind == runner.ContainerRecord.OutputPath { + if bind == runner.Container.OutputPath { runner.HostOutputDir, err = runner.MkTempDir("", "") if err != nil { return fmt.Errorf("While creating mount temp dir: %v", err) @@ -422,8 +388,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] @@ -454,43 +420,52 @@ func (runner *ContainerRunner) AttachStreams() (err error) { return nil } -// StartContainer creates the container and runs it. -func (runner *ContainerRunner) StartContainer() (err error) { +// CreateContainer creates the docker container. +func (runner *ContainerRunner) CreateContainer() error { runner.CrunchLog.Print("Creating Docker container") - runner.CancelLock.Lock() - defer runner.CancelLock.Unlock() - - if runner.Cancelled { - return ErrCancelled + runner.ContainerConfig.Cmd = runner.Container.Command + if runner.Container.Cwd != "." { + runner.ContainerConfig.WorkingDir = runner.Container.Cwd } - runner.ContainerConfig.Cmd = runner.ContainerRecord.Command - if runner.ContainerRecord.Cwd != "." { - runner.ContainerConfig.WorkingDir = runner.ContainerRecord.Cwd - } - for k, v := range runner.ContainerRecord.Environment { + for k, v := range runner.Container.Environment { runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v) } - runner.ContainerConfig.NetworkDisabled = true + 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) } - hostConfig := &dockerclient.HostConfig{Binds: runner.Binds, + + runner.HostConfig = dockerclient.HostConfig{Binds: runner.Binds, LogConfig: dockerclient.LogConfig{Type: "none"}} - err = runner.AttachStreams() - if err != nil { - return err - } + 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, hostConfig) + err := runner.Docker.StartContainer(runner.ContainerID, &runner.HostConfig) if err != nil { - return fmt.Errorf("While starting container: %v", err) + return fmt.Errorf("could not start container: %v", err) } - return nil } @@ -546,7 +521,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) @@ -554,7 +529,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{ @@ -601,72 +576,100 @@ 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 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 { return fmt.Errorf("While creating log collection: %v", err) } - runner.LogsPDH = new(string) - *runner.LogsPDH = response.PortableDataHash + runner.LogsPDH = &response.PortableDataHash return nil } -// UpdateContainerRecordRunning updates the container state to "Running" -func (runner *ContainerRunner) UpdateContainerRecordRunning() error { - return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID, +// 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) } // ContainerToken returns the api_token the container (and any // arv-mount processes) are allowed to use. func (runner *ContainerRunner) ContainerToken() (string, error) { - var auth APIClientAuthorization - err := runner.ArvClient.Call("GET", "containers", runner.ContainerRecord.UUID, "auth", nil, &auth) - return auth.APIToken, err + if runner.token != "" { + return runner.token, nil + } + + 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 } -// UpdateContainerRecordComplete updates the container record state on API +// UpdateContainerComplete updates the container record state on API // server to "Complete" or "Cancelled" -func (runner *ContainerRunner) UpdateContainerRecordComplete() error { +func (runner *ContainerRunner) UpdateContainerFinal() error { update := arvadosclient.Dict{} - 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 - } - 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, arvadosclient.Dict{"container": 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() (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 { @@ -675,93 +678,99 @@ func (runner *ContainerRunner) Run() (err error) { runner.CrunchLog.Printf("Executing on host '%s'", hostname) } - var runerr, waiterr error + // Clean up temporary directories _after_ finalizing + // everything (if we've made any by then) + defer runner.CleanupDirs() + + 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) capture output - outputerr := runner.CaptureOutput() - if outputerr != nil { - runner.CrunchLog.Print(outputerr) + if runner.finalState == "Queued" { + runner.UpdateContainerFinal() + return } - // (7) clean up temporary directories - runner.CleanupDirs() - - // (8) write logs - logerr := runner.CommitLogs() - if logerr != nil { - runner.CrunchLog.Print(logerr) + if runner.IsCancelled() { + runner.finalState = "Cancelled" + // but don't return yet -- we still want to + // capture partial output and write logs } - // (9) update container record with results - updateerr := runner.UpdateContainerRecordComplete() - if updateerr != nil { - runner.CrunchLog.Print(updateerr) - } + checkErr(runner.CaptureOutput()) + checkErr(runner.CommitLogs()) + checkErr(runner.UpdateContainerFinal()) + // 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() - - if err == nil { - if runerr != nil { - err = runerr - } else if waiterr != nil { - err = waiterr - } else if logerr != nil { - err = logerr - } else if updateerr != nil { - err = updateerr - } - } }() - 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 { - return fmt.Errorf("While getting container record: %v", err) + err = fmt.Errorf("While getting container record: %v", err) + return } - // (1) setup signal handling + // setup signal handling runner.SetupSignals() - // (2) check for and/or load image + // check for and/or load image err = runner.LoadImage() if err != nil { - return fmt.Errorf("While loading container image: %v", err) + err = fmt.Errorf("While loading container image: %v", err) + return } - // (3) set up FUSE mount and binds + // set up FUSE mount and binds err = runner.SetupMounts() if err != nil { - return fmt.Errorf("While setting up mounts: %v", err) + err = fmt.Errorf("While setting up mounts: %v", err) + return } - // (3) create and start container - err = runner.StartContainer() + err = runner.CreateContainer() if err != nil { - if err == ErrCancelled { - err = nil - } return } - // (4) update container record state - err = runner.UpdateContainerRecordRunning() + if runner.IsCancelled() { + return + } + + err = runner.UpdateContainerRunning() if err != nil { - runner.CrunchLog.Print(err) + 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 } @@ -776,7 +785,7 @@ 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