X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e8e9262a0f5e1e5908d338ac1655a2cdfaecf23a..f9a1468065776ee71eb43c45e6813bfae69fe0e0:/services/crunch-run/crunchrun.go diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go index 80ca27f68d..6406c17420 100644 --- a/services/crunch-run/crunchrun.go +++ b/services/crunch-run/crunchrun.go @@ -1,6 +1,7 @@ package main import ( + "bytes" "context" "encoding/json" "errors" @@ -60,15 +61,62 @@ type MkTempDir func(string, string) (string, error) // ThinDockerClient is the minimal Docker client interface used by crunch-run. type ThinDockerClient interface { - 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) + 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 - ContainerAttach(ctx context.Context, container string, options dockertypes.ContainerAttachOptions) (dockertypes.HijackedResponse, 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 @@ -195,10 +243,10 @@ func (runner *ContainerRunner) LoadImage() (err error) { } response, err := runner.Docker.ImageLoad(context.TODO(), readCloser, false) - response.Body.Close() 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") } @@ -314,6 +362,13 @@ func (runner *ContainerRunner) SetupMounts() (err error) { } } + 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) + } + } + if bind == "/etc/arvados/ca-certificates.crt" { needCertMount = false } @@ -325,7 +380,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") @@ -616,8 +671,38 @@ func (runner *ContainerRunner) AttachStreams() (err error) { runner.CrunchLog.Print("Attaching container streams") + // 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, Stdout: true, Stderr: true}) + dockertypes.ContainerAttachOptions{Stream: true, Stdin: stdinUsed, Stdout: true, Stderr: true}) if err != nil { return fmt.Errorf("While attaching container stdout/stderr streams: %v", err) } @@ -651,6 +736,27 @@ func (runner *ContainerRunner) AttachStreams() (err error) { } 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 @@ -669,8 +775,7 @@ func (runner *ContainerRunner) CreateContainer() error { runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v) } - runner.ContainerID = createdBody.ID - runner.HostConfig = dockerclient.HostConfig{ + runner.HostConfig = dockercontainer.HostConfig{ Binds: runner.Binds, Cgroup: dockercontainer.CgroupSpec(runner.setCgroupParent), LogConfig: dockercontainer.LogConfig{ @@ -688,20 +793,29 @@ func (runner *ContainerRunner) CreateContainer() error { "ARVADOS_API_HOST="+os.Getenv("ARVADOS_API_HOST"), "ARVADOS_API_HOST_INSECURE="+os.Getenv("ARVADOS_API_HOST_INSECURE"), ) - runner.HostConfig.NetworkMode = runner.networkMode + runner.HostConfig.NetworkMode = dockercontainer.NetworkMode(runner.networkMode) } else { if runner.enableNetwork == "always" { - runner.HostConfig.NetworkMode = runner.networkMode + runner.HostConfig.NetworkMode = dockercontainer.NetworkMode(runner.networkMode) } else { - runner.HostConfig.NetworkMode = "none" + runner.HostConfig.NetworkMode = dockercontainer.NetworkMode("none") } } - createdBody, err := runner.Docker.ContainerCreate(context.TODO(), &runner.ContainerConfig, nil, 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.ContainerID = createdBody.ID + return runner.AttachStreams() } @@ -732,11 +846,9 @@ func (runner *ContainerRunner) WaitFinish() error { return fmt.Errorf("container wait: %v", err) } - if waitDocker != 0 { // what is the acceptable waitDocker code? - runner.CrunchLog.Printf("container wait API status code: %v", waitDocker) - code := int(waitDocker) - runner.ExitCode = &code - } + runner.CrunchLog.Printf("Container exited with code: %v", waitDocker) + code := int(waitDocker) + runner.ExitCode = &code waitMount := runner.ArvMountExit select { @@ -1205,7 +1317,9 @@ func main() { 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