X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b30e81ee46c196476a03bc244fbc4419573ba34e..c22bd85568bdef0358a82bb20e4995477c7d24ac:/services/crunch-run/crunchrun.go diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index a3338ec87f..0793729172 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -1,6 +1,8 @@ package main import ( + "bytes" + "context" "encoding/json" "errors" "flag" @@ -24,7 +26,11 @@ 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" - "github.com/curoverse/dockerclient" + + dockertypes "github.com/docker/docker/api/types" + dockercontainer "github.com/docker/docker/api/types/container" + dockernetwork "github.com/docker/docker/api/types/network" + dockerclient "github.com/docker/docker/client" ) // IArvadosClient is the minimal Arvados API methods used by crunch-run. @@ -33,6 +39,7 @@ type IArvadosClient interface { Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error + CallRaw(method string, resourceType string, uuid string, action string, parameters arvadosclient.Dict) (reader io.ReadCloser, err error) Discovery(key string) (interface{}, error) } @@ -54,14 +61,62 @@ 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 - InspectImage(id string) (*dockerclient.ImageInfo, error) - LoadImage(reader io.Reader) error - CreateContainer(config *dockerclient.ContainerConfig, name string, authConfig *dockerclient.AuthConfig) (string, error) - StartContainer(id string, config *dockerclient.HostConfig) error - AttachContainer(id string, options *dockerclient.AttachOptions) (io.ReadCloser, error) - Wait(id string) <-chan dockerclient.WaitResult - RemoveImage(name string, force bool) ([]*dockerclient.ImageDelete, error) + ContainerAttach(ctx context.Context, container string, options dockertypes.ContainerAttachOptions) (dockertypes.HijackedResponse, error) + ContainerCreate(ctx context.Context, config *dockercontainer.Config, hostConfig *dockercontainer.HostConfig, + networkingConfig *dockernetwork.NetworkingConfig, containerName string) (dockercontainer.ContainerCreateCreatedBody, error) + ContainerStart(ctx context.Context, container string, options dockertypes.ContainerStartOptions) error + ContainerStop(ctx context.Context, container string, timeout *time.Duration) error + ContainerWait(ctx context.Context, container string) (int64, error) + ImageInspectWithRaw(ctx context.Context, image string) (dockertypes.ImageInspect, []byte, error) + ImageLoad(ctx context.Context, input io.Reader, quiet bool) (dockertypes.ImageLoadResponse, error) + 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) (int64, error) { + return proxy.Docker.ContainerWait(ctx, container) +} + +// 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 @@ -71,8 +126,8 @@ type ContainerRunner struct { ArvClient IArvadosClient Kc IKeepClient arvados.Container - dockerclient.ContainerConfig - dockerclient.HostConfig + ContainerConfig dockercontainer.Config + dockercontainer.HostConfig token string ContainerID string ExitCode *int @@ -80,7 +135,7 @@ type ContainerRunner struct { loggingDone chan bool CrunchLog *ThrottledLogger Stdout io.WriteCloser - Stderr *ThrottledLogger + Stderr io.WriteCloser LogCollection *CollectionWriter LogsPDH *string RunArvMount @@ -90,6 +145,7 @@ type ContainerRunner struct { HostOutputDir string CleanupTempDir []string Binds []string + Volumes map[string]struct{} OutputPDH *string SigChan chan os.Signal ArvMountExit chan error @@ -116,6 +172,9 @@ type ContainerRunner struct { cStateLock sync.Mutex cStarted bool // StartContainer() succeeded cCancelled bool // StopContainer() invoked + + enableNetwork string // one of "default" or "always" + networkMode string // passed through to HostConfig.NetworkMode } // SetupSignals sets up signal handling to gracefully terminate the underlying @@ -128,7 +187,6 @@ func (runner *ContainerRunner) SetupSignals() { go func(sig chan os.Signal) { <-sig - log.Print("signal handler calling runner.stop()") runner.stop() signal.Stop(sig) }(runner.SigChan) @@ -143,7 +201,8 @@ func (runner *ContainerRunner) stop() { } runner.cCancelled = true if runner.cStarted { - err := runner.Docker.StopContainer(runner.ContainerID, 10) + timeout := time.Duration(10) + err := runner.Docker.ContainerStop(context.TODO(), runner.ContainerID, &(timeout)) if err != nil { log.Printf("StopContainer failed: %s", err) } @@ -174,7 +233,7 @@ func (runner *ContainerRunner) LoadImage() (err error) { runner.CrunchLog.Printf("Using Docker image id '%s'", imageID) - _, err = runner.Docker.InspectImage(imageID) + _, _, err = runner.Docker.ImageInspectWithRaw(context.TODO(), imageID) if err != nil { runner.CrunchLog.Print("Loading Docker image from keep") @@ -184,10 +243,11 @@ func (runner *ContainerRunner) LoadImage() (err error) { return fmt.Errorf("While creating ManifestFileReader for container image: %v", err) } - err = runner.Docker.LoadImage(readCloser) + response, err := runner.Docker.ImageLoad(context.TODO(), readCloser, false) if err != nil { return fmt.Errorf("While loading container image into Docker: %v", err) } + response.Body.Close() } else { runner.CrunchLog.Print("Docker image is available") } @@ -277,6 +337,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { collectionPaths := []string{} runner.Binds = nil + runner.Volumes = make(map[string]struct{}) needCertMount := true var binds []string @@ -287,10 +348,10 @@ func (runner *ContainerRunner) SetupMounts() (err error) { for _, bind := range binds { mnt := runner.Container.Mounts[bind] - if bind == "stdout" { + if bind == "stdout" || bind == "stderr" { // 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) + return fmt.Errorf("Unsupported mount kind '%s' for %s. Only 'file' is supported.", mnt.Kind, bind) } // Does path start with OutputPath? @@ -299,7 +360,14 @@ func (runner *ContainerRunner) SetupMounts() (err error) { prefix += "/" } if !strings.HasPrefix(mnt.Path, prefix) { - return fmt.Errorf("Stdout path does not start with OutputPath: %s, %s", mnt.Path, prefix) + return fmt.Errorf("%s path does not start with OutputPath: %s, %s", strings.Title(bind), mnt.Path, prefix) + } + } + + if bind == "stdin" { + // Is it a "collection" mount kind? + if mnt.Kind != "collection" && mnt.Kind != "json" { + return fmt.Errorf("Unsupported mount kind '%s' for stdin. Only 'collection' or 'json' are supported.", mnt.Kind) } } @@ -314,7 +382,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } switch { - case mnt.Kind == "collection": + case mnt.Kind == "collection" && bind != "stdin": var src string if mnt.UUID != "" && mnt.PortableDataHash != "" { return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount") @@ -362,24 +430,25 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } collectionPaths = append(collectionPaths, src) - case mnt.Kind == "tmp" && bind == runner.Container.OutputPath: - runner.HostOutputDir, err = runner.MkTempDir("", "") + case mnt.Kind == "tmp": + var tmpdir string + tmpdir, err = runner.MkTempDir("", "") if err != nil { return fmt.Errorf("While creating mount temp dir: %v", err) } - st, staterr := os.Stat(runner.HostOutputDir) + st, staterr := os.Stat(tmpdir) if staterr != nil { return fmt.Errorf("While Stat on temp dir: %v", staterr) } - err = os.Chmod(runner.HostOutputDir, st.Mode()|os.ModeSetgid|0777) + err = os.Chmod(tmpdir, 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) + runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir) + runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", tmpdir, bind)) + if bind == runner.Container.OutputPath { + runner.HostOutputDir = tmpdir + } case mnt.Kind == "json": jsondata, err := json.Marshal(mnt.Content) @@ -533,17 +602,21 @@ func (runner *ContainerRunner) LogNodeInfo() (err error) { }, infoCommand{ label: "Disk Space", - cmd: []string{"df", "-m"}, + cmd: []string{"df", "-m", "/", os.TempDir()}, + }, + infoCommand{ + label: "Disk INodes", + cmd: []string{"df", "-i", "/", os.TempDir()}, }, } // 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:]...).Output() + out, err = exec.Command(command.cmd[0], command.cmd[1:]...).CombinedOutput() if err != nil { - return fmt.Errorf("While running command '%s': %v", - command.cmd[0], err) + return fmt.Errorf("While running command %q: %v", + command.cmd, err) } logger.Println(command.label) for _, line := range strings.Split(string(out), "\n") { @@ -561,24 +634,34 @@ func (runner *ContainerRunner) LogNodeInfo() (err error) { // Get and save the raw JSON container record from the API server func (runner *ContainerRunner) LogContainerRecord() (err error) { w := &ArvLogWriter{ - runner.ArvClient, - runner.Container.UUID, - "container", - runner.LogCollection.Open("container.json"), + ArvClient: runner.ArvClient, + UUID: runner.Container.UUID, + loggingStream: "container", + writeCloser: runner.LogCollection.Open("container.json"), } - logger := log.New(w, "container", 0) - // Convert container record to pretty-printed JSON []byte - rec, err := json.MarshalIndent(runner.Container, "", " ") + // Get Container record JSON from the API Server + reader, err := runner.ArvClient.CallRaw("GET", "containers", runner.Container.UUID, "", nil) if err != nil { - return fmt.Errorf("While converting container record to JSON: %v", err) + return fmt.Errorf("While retrieving container record from the API server: %v", err) } - - // Write JSON record line-by-line - for _, line := range strings.Split(string(rec), "\n") { - logger.Println(line) + defer reader.Close() + // Read the API server response as []byte + json_bytes, err := ioutil.ReadAll(reader) + if err != nil { + return fmt.Errorf("While reading container record API server response: %v", err) + } + // Decode the JSON []byte + var cr map[string]interface{} + if err = json.Unmarshal(json_bytes, &cr); err != nil { + return fmt.Errorf("While decoding the container record JSON response: %v", err) + } + // 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) } - err = w.Close() if err != nil { return fmt.Errorf("While closing container.json log: %v", err) @@ -586,15 +669,44 @@ func (runner *ContainerRunner) LogContainerRecord() (err error) { return nil } -// AttachLogs connects the docker container stdout and stderr logs to the -// Arvados logger which logs to Keep and the API server logs table. +// AttachStreams connects the docker container stdin, stdout and stderr logs +// to the Arvados logger which logs to Keep and the API server logs table. func (runner *ContainerRunner) AttachStreams() (err error) { runner.CrunchLog.Print("Attaching container streams") - var containerReader io.Reader - containerReader, err = runner.Docker.AttachContainer(runner.ContainerID, - &dockerclient.AttachOptions{Stream: true, Stdout: true, Stderr: true}) + // If stdin mount is provided, attach it to the docker container + var stdinRdr keepclient.Reader + var stdinJson []byte + if stdinMnt, ok := runner.Container.Mounts["stdin"]; ok { + if stdinMnt.Kind == "collection" { + var stdinColl arvados.Collection + collId := stdinMnt.UUID + if collId == "" { + collId = stdinMnt.PortableDataHash + } + err = runner.ArvClient.Get("collections", collId, nil, &stdinColl) + if err != nil { + return fmt.Errorf("While getting stding collection: %v", err) + } + + stdinRdr, err = runner.Kc.ManifestFileReader(manifest.Manifest{Text: stdinColl.ManifestText}, stdinMnt.Path) + if os.IsNotExist(err) { + return fmt.Errorf("stdin collection path not found: %v", stdinMnt.Path) + } else if err != nil { + return fmt.Errorf("While getting stdin collection path %v: %v", stdinMnt.Path, err) + } + } else if stdinMnt.Kind == "json" { + stdinJson, err = json.Marshal(stdinMnt.Content) + if err != nil { + return fmt.Errorf("While encoding stdin json data: %v", err) + } + } + } + + stdinUsed := stdinRdr != nil || len(stdinJson) != 0 + response, err := runner.Docker.ContainerAttach(context.TODO(), runner.ContainerID, + dockertypes.ContainerAttachOptions{Stream: true, Stdin: stdinUsed, Stdout: true, Stderr: true}) if err != nil { return fmt.Errorf("While attaching container stdout/stderr streams: %v", err) } @@ -602,37 +714,76 @@ func (runner *ContainerRunner) AttachStreams() (err error) { runner.loggingDone = make(chan bool) 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)) + stdoutFile, err := runner.getStdoutFile(stdoutMnt.Path) if err != nil { - return fmt.Errorf("While creating stdout file: %v", err) + return err } runner.Stdout = stdoutFile } else { runner.Stdout = NewThrottledLogger(runner.NewLogWriter("stdout")) } - runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr")) - go runner.ProcessDockerAttach(containerReader) + if stderrMnt, ok := runner.Container.Mounts["stderr"]; ok { + stderrFile, err := runner.getStdoutFile(stderrMnt.Path) + if err != nil { + return err + } + runner.Stderr = stderrFile + } else { + runner.Stderr = NewThrottledLogger(runner.NewLogWriter("stderr")) + } + + if stdinRdr != nil { + go func() { + _, err := io.Copy(response.Conn, stdinRdr) + if err != nil { + runner.CrunchLog.Print("While writing stdin collection to docker container %q", err) + runner.stop() + } + stdinRdr.Close() + response.CloseWrite() + }() + } else if len(stdinJson) != 0 { + go func() { + _, err := io.Copy(response.Conn, bytes.NewReader(stdinJson)) + if err != nil { + runner.CrunchLog.Print("While writing stdin json to docker container %q", err) + runner.stop() + } + response.CloseWrite() + }() + } + + go runner.ProcessDockerAttach(response.Reader) return nil } +func (runner *ContainerRunner) getStdoutFile(mntPath string) (*os.File, error) { + stdoutPath := mntPath[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 nil, 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 nil, fmt.Errorf("While MkdirAll %q: %v", stdoutPath, err) + } + } + } + stdoutFile, err := os.Create(path.Join(runner.HostOutputDir, stdoutPath)) + if err != nil { + return nil, fmt.Errorf("While creating file %q: %v", stdoutPath, err) + } + + return stdoutFile, nil +} + // CreateContainer creates the docker container. func (runner *ContainerRunner) CreateContainer() error { runner.CrunchLog.Print("Creating Docker container") @@ -645,6 +796,17 @@ func (runner *ContainerRunner) CreateContainer() error { for k, v := range runner.Container.Environment { runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v) } + + runner.ContainerConfig.Volumes = runner.Volumes + + runner.HostConfig = dockercontainer.HostConfig{ + Binds: runner.Binds, + Cgroup: dockercontainer.CgroupSpec(runner.setCgroupParent), + LogConfig: dockercontainer.LogConfig{ + Type: "none", + }, + } + if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI { tok, err := runner.ContainerToken() if err != nil { @@ -655,24 +817,28 @@ 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 + runner.HostConfig.NetworkMode = dockercontainer.NetworkMode(runner.networkMode) } else { - runner.ContainerConfig.NetworkDisabled = true + if runner.enableNetwork == "always" { + runner.HostConfig.NetworkMode = dockercontainer.NetworkMode(runner.networkMode) + } else { + runner.HostConfig.NetworkMode = dockercontainer.NetworkMode("none") + } } - var err error - runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil) + _, stdinUsed := runner.Container.Mounts["stdin"] + runner.ContainerConfig.OpenStdin = stdinUsed + runner.ContainerConfig.StdinOnce = stdinUsed + runner.ContainerConfig.AttachStdin = stdinUsed + runner.ContainerConfig.AttachStdout = true + runner.ContainerConfig.AttachStderr = true + + createdBody, err := runner.Docker.ContainerCreate(context.TODO(), &runner.ContainerConfig, &runner.HostConfig, nil, runner.Container.UUID) 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", - }, - } + runner.ContainerID = createdBody.ID return runner.AttachStreams() } @@ -685,7 +851,8 @@ func (runner *ContainerRunner) StartContainer() error { if runner.cCancelled { return ErrCancelled } - err := runner.Docker.StartContainer(runner.ContainerID, &runner.HostConfig) + err := runner.Docker.ContainerStart(context.TODO(), runner.ContainerID, + dockertypes.ContainerStartOptions{}) if err != nil { return fmt.Errorf("could not start container: %v", err) } @@ -698,21 +865,22 @@ func (runner *ContainerRunner) StartContainer() error { func (runner *ContainerRunner) WaitFinish() error { runner.CrunchLog.Print("Waiting for container to finish") - waitDocker := runner.Docker.Wait(runner.ContainerID) + waitDocker, err := runner.Docker.ContainerWait(context.TODO(), runner.ContainerID) + if err != nil { + return fmt.Errorf("container wait: %v", err) + } + + runner.CrunchLog.Printf("Container exited with code: %v", waitDocker) + code := int(waitDocker) + runner.ExitCode = &code + waitMount := runner.ArvMountExit - for waitDocker != nil { - select { - case err := <-waitMount: - runner.CrunchLog.Printf("arv-mount exited before container finished: %v", err) - waitMount = nil - runner.stop() - case wr := <-waitDocker: - if wr.Error != nil { - return fmt.Errorf("While waiting for container to finish: %v", wr.Error) - } - runner.ExitCode = &wr.ExitCode - waitDocker = nil - } + select { + case err := <-waitMount: + runner.CrunchLog.Printf("arv-mount exited before container finished: %v", err) + waitMount = nil + runner.stop() + default: } // wait for stdout/stderr to complete @@ -899,8 +1067,8 @@ 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.Container.UUID, - "crunch-run", nil}) + runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{ArvClient: runner.ArvClient, + UUID: runner.Container.UUID, loggingStream: "crunch-run", writeCloser: nil}) if runner.LogsPDH != nil { // If we have already assigned something to LogsPDH, @@ -987,7 +1155,8 @@ func (runner *ContainerRunner) IsCancelled() bool { // NewArvLogWriter creates an ArvLogWriter func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser { - return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")} + return &ArvLogWriter{ArvClient: runner.ArvClient, UUID: runner.Container.UUID, loggingStream: name, + writeCloser: runner.LogCollection.Open(name + ".txt")} } // Run the full container lifecycle. @@ -1127,6 +1296,9 @@ func NewContainerRunner(api IArvadosClient, cr.Container.UUID = containerUUID cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run")) cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0) + + loadLogThrottleParams(api) + return cr } @@ -1136,6 +1308,14 @@ func main() { 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") + enableNetwork := flag.String("container-enable-networking", "default", + `Specify if networking should be enabled for container. One of 'default', 'always': + default: only enable networking if container requests it. + always: containers always have networking enabled + `) + networkMode := flag.String("container-network-mode", "default", + `Set networking mode for container. Corresponds to Docker network mode (--net). + `) flag.Parse() containerId := flag.Arg(0) @@ -1157,16 +1337,22 @@ func main() { } kc.Retries = 4 - var docker *dockerclient.DockerClient - docker, err = dockerclient.NewDockerClient("unix:///var/run/docker.sock", nil) + var docker *dockerclient.Client + // API version 1.21 corresponds to Docker 1.9, which is currently the + // minimum version we want to support. + docker, err = dockerclient.NewClient(dockerclient.DefaultDockerHost, "1.21", nil, nil) if err != nil { log.Fatalf("%s: %v", containerId, err) } - cr := NewContainerRunner(api, kc, docker, containerId) + dockerClientProxy := ThinDockerClientProxy{Docker: docker} + + cr := NewContainerRunner(api, kc, dockerClientProxy, containerId) cr.statInterval = *statInterval cr.cgroupRoot = *cgroupRoot cr.expectCgroupParent = *cgroupParent + cr.enableNetwork = *enableNetwork + cr.networkMode = *networkMode if *cgroupParentSubsystem != "" { p := findCgroup(*cgroupParentSubsystem) cr.setCgroupParent = p