X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/0c888bcc93b559339c8abbce784bdcc44746bca2..3194c1b24ffe6fff5fcb2f620ca6ee43741e3462:/services/crunch-run/crunchrun.go diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index 0980ba1150..1bd49569d3 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" @@ -19,6 +18,7 @@ import ( "os/signal" "path" "path/filepath" + "regexp" "runtime" "runtime/pprof" "sort" @@ -32,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" @@ -81,53 +82,6 @@ type ThinDockerClient interface { 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 { @@ -230,12 +184,15 @@ func (runner *ContainerRunner) stopSignals() { } } -var errorBlacklist = []string{"Cannot connect to the Docker daemon"} +var errorBlacklist = []string{ + "(?ms).*[Cc]annot connect to the Docker daemon.*", + "(?ms).*oci runtime error.*starting container process.*container init.*mounting.*to rootfs.*no such file or directory.*", +} var brokenNodeHook *string = flag.String("broken-node-hook", "", "Script to run if node is detected to be broken (for example, Docker daemon is not running)") func (runner *ContainerRunner) checkBrokenNode(goterr error) bool { for _, d := range errorBlacklist { - if strings.Index(goterr.Error(), d) != -1 { + if m, e := regexp.MatchString(d, goterr.Error()); m && e == nil { runner.CrunchLog.Printf("Error suggests node is unable to run containers: %v", goterr) if *brokenNodeHook == "" { runner.CrunchLog.Printf("No broken node hook provided, cannot mark node as broken.") @@ -386,6 +343,11 @@ func (runner *ContainerRunner) SetupMounts() (err error) { return fmt.Errorf("While creating keep mount temp dir: %v", err) } + token, err := runner.ContainerToken() + if err != nil { + return fmt.Errorf("could not get container token: %s", err) + } + pdhOnly := true tmpcount := 0 arvMountCmd := []string{ @@ -534,6 +496,18 @@ func (runner *ContainerRunner) SetupMounts() (err error) { return fmt.Errorf("writing temp file: %v", err) } runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind)) + + case mnt.Kind == "git_tree": + tmpdir, err := runner.MkTempDir("", "") + 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 + } + runner.Binds = append(runner.Binds, tmpdir+":"+bind+":ro") } } @@ -558,11 +532,6 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } arvMountCmd = append(arvMountCmd, runner.ArvMountPoint) - token, err := runner.ContainerToken() - if err != nil { - return fmt.Errorf("could not get container token: %s", err) - } - runner.ArvMount, err = runner.RunArvMount(arvMountCmd, token) if err != nil { return fmt.Errorf("While trying to start arv-mount: %v", err) @@ -644,11 +613,13 @@ type infoCommand struct { cmd []string } -// LogNodeInfo gathers node information and store it on the log for debugging -// purposes. -func (runner *ContainerRunner) LogNodeInfo() (err error) { +// LogHostInfo logs info about the current host, for debugging and +// accounting purposes. Although it's logged as "node-info", this is +// about the environment where crunch-run is actually running, which +// might differ from what's described in the node record (see +// LogNodeRecord). +func (runner *ContainerRunner) LogHostInfo() (err error) { w := runner.NewLogWriter("node-info") - logger := log.New(w, "node-info", 0) commands := []infoCommand{ { @@ -674,17 +645,17 @@ func (runner *ContainerRunner) LogNodeInfo() (err error) { } // Run commands with informational output to be logged. - var out []byte for _, command := range commands { - out, err = exec.Command(command.cmd[0], command.cmd[1:]...).CombinedOutput() - if err != nil { - return fmt.Errorf("While running command %q: %v", - command.cmd, err) - } - logger.Println(command.label) - for _, line := range strings.Split(string(out), "\n") { - logger.Println(" ", line) + fmt.Fprintln(w, command.label) + cmd := exec.Command(command.cmd[0], command.cmd[1:]...) + cmd.Stdout = w + cmd.Stderr = w + if err := cmd.Run(); err != nil { + err = fmt.Errorf("While running command %q: %v", command.cmd, err) + fmt.Fprintln(w, err) + return err } + fmt.Fprintln(w, "") } err = w.Close() @@ -695,38 +666,71 @@ func (runner *ContainerRunner) LogNodeInfo() (err error) { } // LogContainerRecord gets and saves the raw JSON container record from the API server -func (runner *ContainerRunner) LogContainerRecord() (err error) { +func (runner *ContainerRunner) LogContainerRecord() error { + logged, err := runner.logAPIResponse("container", "containers", map[string]interface{}{"filters": [][]string{{"uuid", "=", runner.Container.UUID}}}, nil) + if !logged && err == nil { + err = fmt.Errorf("error: no container record found for %s", runner.Container.UUID) + } + return err +} + +// LogNodeRecord logs arvados#node record corresponding to the current host. +func (runner *ContainerRunner) LogNodeRecord() error { + hostname := os.Getenv("SLURMD_NODENAME") + if hostname == "" { + hostname, _ = os.Hostname() + } + _, err := runner.logAPIResponse("node", "nodes", map[string]interface{}{"filters": [][]string{{"hostname", "=", hostname}}}, func(resp interface{}) { + // The "info" field has admin-only info when obtained + // with a privileged token, and should not be logged. + node, ok := resp.(map[string]interface{}) + if ok { + delete(node, "info") + } + }) + return err +} + +func (runner *ContainerRunner) logAPIResponse(label, path string, params map[string]interface{}, munge func(interface{})) (logged bool, err error) { w := &ArvLogWriter{ ArvClient: runner.ArvClient, UUID: runner.Container.UUID, - loggingStream: "container", - writeCloser: runner.LogCollection.Open("container.json"), + loggingStream: label, + writeCloser: runner.LogCollection.Open(label + ".json"), } - // Get Container record JSON from the API Server - reader, err := runner.ArvClient.CallRaw("GET", "containers", runner.Container.UUID, "", nil) + reader, err := runner.ArvClient.CallRaw("GET", path, "", "", arvadosclient.Dict(params)) if err != nil { - return fmt.Errorf("While retrieving container record from the API server: %v", err) + return false, fmt.Errorf("error getting %s record: %v", label, err) } defer reader.Close() dec := json.NewDecoder(reader) dec.UseNumber() - var cr map[string]interface{} - if err = dec.Decode(&cr); err != nil { - return fmt.Errorf("While decoding the container record JSON response: %v", err) + var resp map[string]interface{} + if err = dec.Decode(&resp); err != nil { + return false, fmt.Errorf("error decoding %s list response: %v", label, err) + } + items, ok := resp["items"].([]interface{}) + if !ok { + return false, fmt.Errorf("error decoding %s list response: no \"items\" key in API list response", label) + } else if len(items) < 1 { + return false, nil + } + if munge != nil { + munge(items[0]) } // Re-encode it using indentation to improve readability enc := json.NewEncoder(w) enc.SetIndent("", " ") - if err = enc.Encode(cr); err != nil { - return fmt.Errorf("While logging the JSON container record: %v", err) + if err = enc.Encode(items[0]); err != nil { + return false, fmt.Errorf("error logging %s record: %v", label, err) } err = w.Close() if err != nil { - return fmt.Errorf("While closing container.json log: %v", err) + return false, fmt.Errorf("error closing %s.json in log collection: %v", label, err) } - return nil + return true, nil } // AttachStreams connects the docker container stdin, stdout and stderr logs @@ -917,7 +921,7 @@ func (runner *ContainerRunner) StartContainer() error { dockertypes.ContainerStartOptions{}) if err != nil { var advice string - if strings.Contains(err.Error(), "no such file or directory") { + if m, e := regexp.MatchString("(?ms).*(exec|System error).*(no such file or directory|file not found).*", err.Error()); m && e == nil { advice = fmt.Sprintf("\nPossible causes: command %q is missing, the interpreter given in #! is missing, or script has Windows line endings.", runner.Container.Command[0]) } return fmt.Errorf("could not start container: %v%s", err, advice) @@ -1035,14 +1039,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. @@ -1050,9 +1074,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) @@ -1073,7 +1104,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 } @@ -1537,13 +1568,14 @@ func (runner *ContainerRunner) Run() (err error) { if err != nil { return } - - // Gather and record node information - err = runner.LogNodeInfo() + err = runner.LogHostInfo() + if err != nil { + return + } + err = runner.LogNodeRecord() if err != nil { return } - // Save container.json record on log collection err = runner.LogContainerRecord() if err != nil { return @@ -1660,10 +1692,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)