X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/c3cbac378fb45e3bf996a5d691cd3f205dfb3f90..471270bc6ec9453fdd4c1faf97b65a8291543c6f:/services/crunch-run/crunchrun.go diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index 0b4eb2bcbf..876df03d2c 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -19,6 +19,7 @@ import ( "os/signal" "path" "path/filepath" + "sort" "strings" "sync" "syscall" @@ -29,8 +30,9 @@ import ( 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) - Call(method, resourceType, uuid, action 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. @@ -39,7 +41,7 @@ var ErrCancelled = errors.New("Cancelled") // IKeepClient is the minimal Keep API methods used by crunch-run. type IKeepClient interface { PutHB(hash string, buf []byte) (string, int, error) - ManifestFileReader(m manifest.Manifest, filename string) (keepclient.ReadCloserWithLen, error) + ManifestFileReader(m manifest.Manifest, filename string) (keepclient.Reader, error) } // NewLogWriter is a factory function to create a new log writer. @@ -93,6 +95,7 @@ type ContainerRunner struct { SigChan chan os.Signal ArvMountExit chan error finalState string + trashLifetime time.Duration statLogger io.WriteCloser statReporter *crunchstat.Reporter @@ -237,8 +240,15 @@ func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string, token string) ( return c, nil } +func (runner *ContainerRunner) SetupArvMountPoint(prefix string) (err error) { + if runner.ArvMountPoint == "" { + runner.ArvMountPoint, err = runner.MkTempDir("", prefix) + } + return +} + func (runner *ContainerRunner) SetupMounts() (err error) { - runner.ArvMountPoint, err = runner.MkTempDir("", "keep") + err = runner.SetupArvMountPoint("keep") if err != nil { return fmt.Errorf("While creating keep mount temp dir: %v", err) } @@ -248,10 +258,23 @@ func (runner *ContainerRunner) SetupMounts() (err error) { 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 + needCertMount := true - for bind, mnt := range runner.Container.Mounts { + var binds []string + for bind, _ := range runner.Container.Mounts { + binds = append(binds, bind) + } + sort.Strings(binds) + + for _, bind := range binds { + mnt := runner.Container.Mounts[bind] if bind == "stdout" { // Is it a "file" mount kind? if mnt.Kind != "file" { @@ -268,6 +291,16 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } } + if bind == "/etc/arvados/ca-certificates.crt" { + needCertMount = false + } + + 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) + } + } + switch { case mnt.Kind == "collection": var src string @@ -294,6 +327,8 @@ func (runner *ContainerRunner) SetupMounts() (err error) { if mnt.Writable { if bind == runner.Container.OutputPath { runner.HostOutputDir = src + } else if strings.HasPrefix(bind, runner.Container.OutputPath+"/") { + return fmt.Errorf("Writable mount points are not permitted underneath the output_path: %v", bind) } runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind)) } else { @@ -348,6 +383,16 @@ func (runner *ContainerRunner) SetupMounts() (err error) { return fmt.Errorf("Output path does not correspond to a writable mount point") } + if wantAPI := runner.Container.RuntimeConstraints.API; needCertMount && wantAPI != nil && *wantAPI { + for _, certfile := range arvadosclient.CertFiles { + _, err := os.Stat(certfile) + if err == nil { + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:/etc/arvados/ca-certificates.crt:ro", certfile)) + break + } + } + } + if pdhOnly { arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id") } else { @@ -561,6 +606,21 @@ func (runner *ContainerRunner) CaptureOutput() error { 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 } @@ -597,20 +657,150 @@ func (runner *ContainerRunner) CaptureOutput() error { manifestText = rec.ManifestText } + // Pre-populate output from the configured mount points + var binds []string + for bind, _ := range runner.Container.Mounts { + binds = append(binds, bind) + } + sort.Strings(binds) + + for _, bind := range binds { + mnt := runner.Container.Mounts[bind] + + bindSuffix := strings.TrimPrefix(bind, runner.Container.OutputPath) + + if bindSuffix == bind || len(bindSuffix) <= 0 { + // either does not start with OutputPath or is OutputPath itself + continue + } + + if strings.HasPrefix(bindSuffix, "/") == false { + bindSuffix = "/" + bindSuffix + } + + if mnt.ExcludeFromOutput == true { + continue + } + + idx := strings.Index(mnt.PortableDataHash, "/") + if idx > 0 { + mnt.Path = mnt.PortableDataHash[idx:] + mnt.PortableDataHash = mnt.PortableDataHash[0:idx] + } + + // append to manifest_text + m, err := runner.getCollectionManifestForPath(mnt, bindSuffix) + if err != nil { + return err + } + + manifestText = manifestText + m + } + + // Save output var response arvados.Collection err = runner.ArvClient.Create("collections", arvadosclient.Dict{ "collection": arvadosclient.Dict{ + "trash_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 +} - runner.OutputPDH = new(string) - *runner.OutputPDH = response.PortableDataHash +var outputCollections = make(map[string]arvados.Collection) + +// Fetch the collection for the mnt.PortableDataHash +// Return the manifest_text fragment corresponding to the specified mnt.Path +// after making any required updates. +// Ex: +// If mnt.Path is not specified, +// return the entire manifest_text after replacing any "." with bindSuffix +// If mnt.Path corresponds to one stream, +// return the manifest_text for that stream after replacing that stream name with bindSuffix +// Otherwise, check if a filename in any one stream is being sought. Return the manifest_text +// for that stream after replacing stream name with bindSuffix minus the last word +// and the file name with last word of the bindSuffix +// Allowed path examples: +// "path":"/" +// "path":"/subdir1" +// "path":"/subdir1/subdir2" +// "path":"/subdir/filename" etc +func (runner *ContainerRunner) getCollectionManifestForPath(mnt arvados.Mount, bindSuffix string) (string, error) { + collection := outputCollections[mnt.PortableDataHash] + if collection.PortableDataHash == "" { + err := runner.ArvClient.Get("collections", mnt.PortableDataHash, nil, &collection) + if err != nil { + return "", fmt.Errorf("While getting collection for %v: %v", mnt.PortableDataHash, err) + } + outputCollections[mnt.PortableDataHash] = collection + } - return nil + manifest := manifest.Manifest{Text: collection.ManifestText} + + manifestText := "" + if mnt.Path == "" || mnt.Path == "/" { + // no path specified; return the entire manifest text after making adjustments + manifestText = collection.ManifestText + manifestText = strings.Replace(manifestText, "./", "."+bindSuffix+"/", -1) + manifestText = strings.Replace(manifestText, ". ", "."+bindSuffix+" ", -1) + } else { + // either a single stream or file from a stream is being sought + bindIdx := strings.LastIndex(bindSuffix, "/") + var bindSubdir, bindFileName string + if bindIdx >= 0 { + bindSubdir = "." + bindSuffix[0:bindIdx] + bindFileName = bindSuffix[bindIdx+1:] + } + mntPath := mnt.Path + if strings.HasSuffix(mntPath, "/") { + mntPath = mntPath[0 : len(mntPath)-1] + } + pathIdx := strings.LastIndex(mntPath, "/") + var pathSubdir, pathFileName string + if pathIdx >= 0 { + pathSubdir = "." + mntPath[0:pathIdx] + pathFileName = mntPath[pathIdx+1:] + } + streams := strings.Split(collection.ManifestText, "\n") + for _, stream := range streams { + tokens := strings.Split(stream, " ") + if tokens[0] == "."+mntPath { + // path refers to this complete stream + adjustedStream := strings.Replace(stream, "."+mntPath, "."+bindSuffix, -1) + manifestText = adjustedStream + "\n" + break + } else { + // look for a matching file in this stream + fs := manifest.FileSegmentForPath(mntPath[1:]) + if fs != "" { + manifestText = strings.Replace(fs, ":"+pathFileName, ":"+bindFileName, -1) + manifestText = strings.Replace(manifestText, pathSubdir, bindSubdir, -1) + manifestText += "\n" + break + } + } + } + } + + if manifestText == "" { + runner.CrunchLog.Printf("No manifest segment found for bind '%v' with path '%v'", bindSuffix, mnt.Path) + } + + return manifestText, 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() { @@ -665,15 +855,14 @@ func (runner *ContainerRunner) CommitLogs() error { err = runner.ArvClient.Create("collections", arvadosclient.Dict{ "collection": arvadosclient.Dict{ + "trash_at": time.Now().Add(runner.trashLifetime).Format(time.RFC3339), "name": "logs for " + runner.Container.UUID, "manifest_text": mt}}, &response) if err != nil { return fmt.Errorf("While creating log collection: %v", err) } - runner.LogsPDH = &response.PortableDataHash - return nil } @@ -709,10 +898,10 @@ func (runner *ContainerRunner) ContainerToken() (string, error) { func (runner *ContainerRunner) UpdateContainerFinal() error { update := arvadosclient.Dict{} update["state"] = runner.finalState + if runner.LogsPDH != nil { + update["log"] = *runner.LogsPDH + } if runner.finalState == "Complete" { - if runner.LogsPDH != nil { - update["log"] = *runner.LogsPDH - } if runner.ExitCode != nil { update["exit_code"] = *runner.ExitCode } @@ -772,6 +961,7 @@ func (runner *ContainerRunner) Run() (err error) { checkErr(err) if runner.finalState == "Queued" { + runner.CrunchLog.Close() runner.UpdateContainerFinal() return } @@ -804,6 +994,7 @@ func (runner *ContainerRunner) Run() (err error) { // check for and/or load image err = runner.LoadImage() if err != nil { + runner.finalState = "Cancelled" err = fmt.Errorf("While loading container image: %v", err) return } @@ -811,6 +1002,7 @@ func (runner *ContainerRunner) Run() (err error) { // set up FUSE mount and binds err = runner.SetupMounts() if err != nil { + runner.finalState = "Cancelled" err = fmt.Errorf("While setting up mounts: %v", err) return } @@ -858,6 +1050,7 @@ func NewContainerRunner(api IArvadosClient, cr.Container.UUID = containerUUID cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run")) cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0) + cr.loadDiscoveryVars() return cr } @@ -866,10 +1059,15 @@ func main() { 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") + caCertsPath := flag.String("ca-certs", "", "Path to TLS root certificates") flag.Parse() containerId := flag.Arg(0) + if *caCertsPath != "" { + arvadosclient.CertFiles = []string{*caCertsPath} + } + api, err := arvadosclient.MakeArvadosClient() if err != nil { log.Fatalf("%s: %v", containerId, err) @@ -877,7 +1075,7 @@ func main() { api.Retries = 8 var kc *keepclient.KeepClient - kc, err = keepclient.MakeKeepClient(&api) + kc, err = keepclient.MakeKeepClient(api) if err != nil { log.Fatalf("%s: %v", containerId, err) }