1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
14 "git.arvados.org/arvados.git/sdk/go/arvados"
15 dockertypes "github.com/docker/docker/api/types"
16 dockercontainer "github.com/docker/docker/api/types/container"
17 dockerclient "github.com/docker/docker/client"
18 "golang.org/x/net/context"
21 // Docker daemon won't let you set a limit less than ~10 MiB
22 const minDockerRAM = int64(16 * 1024 * 1024)
24 type dockerExecutor struct {
26 logf func(string, ...interface{})
27 watchdogInterval time.Duration
28 dockerclient *dockerclient.Client
34 func newDockerExecutor(containerUUID string, logf func(string, ...interface{}), watchdogInterval time.Duration) (*dockerExecutor, error) {
35 // API version 1.21 corresponds to Docker 1.9, which is
36 // currently the minimum version we want to support.
37 client, err := dockerclient.NewClient(dockerclient.DefaultDockerHost, "1.21", nil, nil)
38 if watchdogInterval < 1 {
39 watchdogInterval = time.Minute
41 return &dockerExecutor{
42 containerUUID: containerUUID,
44 watchdogInterval: watchdogInterval,
49 func (e *dockerExecutor) Runtime() string {
50 v, _ := e.dockerclient.ServerVersion(context.Background())
52 for _, cv := range v.Components {
56 info += cv.Name + " " + cv.Version
59 info = "(unknown version)"
61 return "docker " + info
64 func (e *dockerExecutor) LoadImage(imageID string, imageTarballPath string, container arvados.Container, arvMountPoint string,
65 containerClient *arvados.Client) error {
66 _, _, err := e.dockerclient.ImageInspectWithRaw(context.TODO(), imageID)
72 f, err := os.Open(imageTarballPath)
77 resp, err := e.dockerclient.ImageLoad(context.TODO(), f, true)
79 return fmt.Errorf("While loading container image into Docker: %v", err)
81 defer resp.Body.Close()
82 buf, _ := ioutil.ReadAll(resp.Body)
83 e.logf("loaded image: response %s", buf)
87 func (e *dockerExecutor) config(spec containerSpec) (dockercontainer.Config, dockercontainer.HostConfig) {
88 e.logf("Creating Docker container")
89 cfg := dockercontainer.Config{
92 WorkingDir: spec.WorkingDir,
93 Volumes: map[string]struct{}{},
94 OpenStdin: spec.Stdin != nil,
95 StdinOnce: spec.Stdin != nil,
96 AttachStdin: spec.Stdin != nil,
100 if cfg.WorkingDir == "." {
103 for k, v := range spec.Env {
104 cfg.Env = append(cfg.Env, k+"="+v)
106 if spec.RAM > 0 && spec.RAM < minDockerRAM {
107 spec.RAM = minDockerRAM
109 hostCfg := dockercontainer.HostConfig{
110 LogConfig: dockercontainer.LogConfig{
113 NetworkMode: dockercontainer.NetworkMode("none"),
114 Resources: dockercontainer.Resources{
115 CgroupParent: spec.CgroupParent,
116 NanoCPUs: int64(spec.VCPUs) * 1000000000,
117 Memory: spec.RAM, // RAM
118 MemorySwap: spec.RAM, // RAM+swap
119 KernelMemory: spec.RAM, // kernel portion
122 if spec.CUDADeviceCount != 0 {
123 var deviceIds []string
124 if cudaVisibleDevices := os.Getenv("CUDA_VISIBLE_DEVICES"); cudaVisibleDevices != "" {
125 // If a resource manager such as slurm or LSF told
126 // us to select specific devices we need to propagate that.
127 deviceIds = strings.Split(cudaVisibleDevices, ",")
130 deviceCount := spec.CUDADeviceCount
131 if len(deviceIds) > 0 {
132 // Docker won't accept both non-empty
133 // DeviceIDs and a non-zero Count
135 // (it turns out "Count" is a dumb fallback
136 // that just allocates device 0, 1, 2, ...,
141 // Capabilities are confusing. The driver has generic
142 // capabilities "gpu" and "nvidia" but then there's
143 // additional capabilities "compute" and "utility"
144 // that are passed to nvidia-container-cli.
146 // "compute" means include the CUDA libraries and
147 // "utility" means include the CUDA utility programs
148 // (like nvidia-smi).
150 // https://github.com/moby/moby/blob/7b9275c0da707b030e62c96b679a976f31f929d3/daemon/nvidia_linux.go#L37
151 // https://github.com/containerd/containerd/blob/main/contrib/nvidia/nvidia.go
152 hostCfg.Resources.DeviceRequests = append(hostCfg.Resources.DeviceRequests, dockercontainer.DeviceRequest{
155 DeviceIDs: deviceIds,
156 Capabilities: [][]string{[]string{"gpu", "nvidia", "compute", "utility"}},
159 for path, mount := range spec.BindMounts {
160 bind := mount.HostPath + ":" + path
164 hostCfg.Binds = append(hostCfg.Binds, bind)
166 if spec.EnableNetwork {
167 hostCfg.NetworkMode = dockercontainer.NetworkMode(spec.NetworkMode)
172 func (e *dockerExecutor) Create(spec containerSpec) error {
173 cfg, hostCfg := e.config(spec)
174 created, err := e.dockerclient.ContainerCreate(context.TODO(), &cfg, &hostCfg, nil, e.containerUUID)
176 return fmt.Errorf("While creating container: %v", err)
178 e.containerID = created.ID
179 return e.startIO(spec.Stdin, spec.Stdout, spec.Stderr)
182 func (e *dockerExecutor) CgroupID() string {
186 func (e *dockerExecutor) Start() error {
187 return e.dockerclient.ContainerStart(context.TODO(), e.containerID, dockertypes.ContainerStartOptions{})
190 func (e *dockerExecutor) Stop() error {
191 err := e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})
192 if err != nil && strings.Contains(err.Error(), "No such container: "+e.containerID) {
198 // Wait for the container to terminate, capture the exit code, and
199 // wait for stdout/stderr logging to finish.
200 func (e *dockerExecutor) Wait(ctx context.Context) (int, error) {
201 ctx, cancel := context.WithCancel(ctx)
203 watchdogErr := make(chan error, 1)
205 ticker := time.NewTicker(e.watchdogInterval)
208 dctx, dcancel := context.WithDeadline(ctx, time.Now().Add(e.watchdogInterval))
209 ctr, err := e.dockerclient.ContainerInspect(dctx, e.containerID)
211 if ctx.Err() != nil {
212 // Either the container already
213 // exited, or our caller is trying to
216 } else if err != nil {
217 e.logf("Error inspecting container: %s", err)
220 } else if ctr.State == nil || !(ctr.State.Running || ctr.State.Status == "created") {
221 watchdogErr <- fmt.Errorf("Container is not running: State=%v", ctr.State)
227 waitOk, waitErr := e.dockerclient.ContainerWait(ctx, e.containerID, dockercontainer.WaitConditionNotRunning)
230 case waitBody := <-waitOk:
231 // wait for stdout/stderr to complete
233 return int(waitBody.StatusCode), nil
235 case err := <-waitErr:
236 return -1, fmt.Errorf("container wait: %v", err)
241 case err := <-watchdogErr:
247 func (e *dockerExecutor) startIO(stdin io.Reader, stdout, stderr io.Writer) error {
248 resp, err := e.dockerclient.ContainerAttach(context.TODO(), e.containerID, dockertypes.ContainerAttachOptions{
255 return fmt.Errorf("error attaching container stdin/stdout/stderr streams: %v", err)
260 errStdin = e.handleStdin(stdin, resp.Conn, resp.CloseWrite)
263 e.doneIO = make(chan struct{})
265 e.errIO = e.handleStdoutStderr(stdout, stderr, resp.Reader)
266 if e.errIO == nil && errStdin != nil {
274 func (e *dockerExecutor) handleStdin(stdin io.Reader, conn io.Writer, closeConn func() error) error {
276 _, err := io.Copy(conn, stdin)
278 return fmt.Errorf("While writing to docker container on stdin: %v", err)
283 // Handle docker log protocol; see
284 // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
285 func (e *dockerExecutor) handleStdoutStderr(stdout, stderr io.Writer, reader io.Reader) error {
286 header := make([]byte, 8)
289 _, err = io.ReadAtLeast(reader, header, 8)
296 readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
298 _, err = io.CopyN(stdout, reader, readsize)
301 _, err = io.CopyN(stderr, reader, readsize)
305 return fmt.Errorf("error copying stdout/stderr from docker: %v", err)
310 func (e *dockerExecutor) Close() {
311 e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})