12630: Test docker container config setup.
[arvados.git] / lib / crunchrun / docker.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4 package crunchrun
5
6 import (
7         "fmt"
8         "io"
9         "io/ioutil"
10         "os"
11         "strings"
12         "time"
13
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"
19 )
20
21 // Docker daemon won't let you set a limit less than ~10 MiB
22 const minDockerRAM = int64(16 * 1024 * 1024)
23
24 type dockerExecutor struct {
25         containerUUID    string
26         logf             func(string, ...interface{})
27         watchdogInterval time.Duration
28         dockerclient     *dockerclient.Client
29         containerID      string
30         doneIO           chan struct{}
31         errIO            error
32 }
33
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
40         }
41         return &dockerExecutor{
42                 containerUUID:    containerUUID,
43                 logf:             logf,
44                 watchdogInterval: watchdogInterval,
45                 dockerclient:     client,
46         }, err
47 }
48
49 func (e *dockerExecutor) Runtime() string { return "docker" }
50
51 func (e *dockerExecutor) LoadImage(imageID string, imageTarballPath string, container arvados.Container, arvMountPoint string,
52         containerClient *arvados.Client) error {
53         _, _, err := e.dockerclient.ImageInspectWithRaw(context.TODO(), imageID)
54         if err == nil {
55                 // already loaded
56                 return nil
57         }
58
59         f, err := os.Open(imageTarballPath)
60         if err != nil {
61                 return err
62         }
63         defer f.Close()
64         resp, err := e.dockerclient.ImageLoad(context.TODO(), f, true)
65         if err != nil {
66                 return fmt.Errorf("While loading container image into Docker: %v", err)
67         }
68         defer resp.Body.Close()
69         buf, _ := ioutil.ReadAll(resp.Body)
70         e.logf("loaded image: response %s", buf)
71         return nil
72 }
73
74 func (e *dockerExecutor) config(spec containerSpec) (dockercontainer.Config, dockercontainer.HostConfig) {
75         e.logf("Creating Docker container")
76         cfg := dockercontainer.Config{
77                 Image:        spec.Image,
78                 Cmd:          spec.Command,
79                 WorkingDir:   spec.WorkingDir,
80                 Volumes:      map[string]struct{}{},
81                 OpenStdin:    spec.Stdin != nil,
82                 StdinOnce:    spec.Stdin != nil,
83                 AttachStdin:  spec.Stdin != nil,
84                 AttachStdout: true,
85                 AttachStderr: true,
86         }
87         if cfg.WorkingDir == "." {
88                 cfg.WorkingDir = ""
89         }
90         for k, v := range spec.Env {
91                 cfg.Env = append(cfg.Env, k+"="+v)
92         }
93         if spec.RAM > 0 && spec.RAM < minDockerRAM {
94                 spec.RAM = minDockerRAM
95         }
96         hostCfg := dockercontainer.HostConfig{
97                 LogConfig: dockercontainer.LogConfig{
98                         Type: "none",
99                 },
100                 NetworkMode: dockercontainer.NetworkMode("none"),
101                 Resources: dockercontainer.Resources{
102                         CgroupParent: spec.CgroupParent,
103                         NanoCPUs:     int64(spec.VCPUs) * 1000000000,
104                         Memory:       spec.RAM, // RAM
105                         MemorySwap:   spec.RAM, // RAM+swap
106                         KernelMemory: spec.RAM, // kernel portion
107                 },
108         }
109         if spec.CUDADeviceCount != 0 {
110                 hostCfg.Resources.DeviceRequests = append(hostCfg.Resources.DeviceRequests, dockercontainer.DeviceRequest{
111                         Driver:       "nvidia",
112                         Count:        spec.CUDADeviceCount,
113                         Capabilities: [][]string{[]string{"gpu", "nvidia", "compute"}},
114                 })
115         }
116         for path, mount := range spec.BindMounts {
117                 bind := mount.HostPath + ":" + path
118                 if mount.ReadOnly {
119                         bind += ":ro"
120                 }
121                 hostCfg.Binds = append(hostCfg.Binds, bind)
122         }
123         if spec.EnableNetwork {
124                 hostCfg.NetworkMode = dockercontainer.NetworkMode(spec.NetworkMode)
125         }
126         return cfg, hostCfg
127 }
128
129 func (e *dockerExecutor) Create(spec containerSpec) error {
130         cfg, hostCfg := e.config(spec)
131         created, err := e.dockerclient.ContainerCreate(context.TODO(), &cfg, &hostCfg, nil, e.containerUUID)
132         if err != nil {
133                 return fmt.Errorf("While creating container: %v", err)
134         }
135         e.containerID = created.ID
136         return e.startIO(spec.Stdin, spec.Stdout, spec.Stderr)
137 }
138
139 func (e *dockerExecutor) CgroupID() string {
140         return e.containerID
141 }
142
143 func (e *dockerExecutor) Start() error {
144         return e.dockerclient.ContainerStart(context.TODO(), e.containerID, dockertypes.ContainerStartOptions{})
145 }
146
147 func (e *dockerExecutor) Stop() error {
148         err := e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})
149         if err != nil && strings.Contains(err.Error(), "No such container: "+e.containerID) {
150                 err = nil
151         }
152         return err
153 }
154
155 // Wait for the container to terminate, capture the exit code, and
156 // wait for stdout/stderr logging to finish.
157 func (e *dockerExecutor) Wait(ctx context.Context) (int, error) {
158         ctx, cancel := context.WithCancel(ctx)
159         defer cancel()
160         watchdogErr := make(chan error, 1)
161         go func() {
162                 ticker := time.NewTicker(e.watchdogInterval)
163                 defer ticker.Stop()
164                 for range ticker.C {
165                         dctx, dcancel := context.WithDeadline(ctx, time.Now().Add(e.watchdogInterval))
166                         ctr, err := e.dockerclient.ContainerInspect(dctx, e.containerID)
167                         dcancel()
168                         if ctx.Err() != nil {
169                                 // Either the container already
170                                 // exited, or our caller is trying to
171                                 // kill it.
172                                 return
173                         } else if err != nil {
174                                 e.logf("Error inspecting container: %s", err)
175                                 watchdogErr <- err
176                                 return
177                         } else if ctr.State == nil || !(ctr.State.Running || ctr.State.Status == "created") {
178                                 watchdogErr <- fmt.Errorf("Container is not running: State=%v", ctr.State)
179                                 return
180                         }
181                 }
182         }()
183
184         waitOk, waitErr := e.dockerclient.ContainerWait(ctx, e.containerID, dockercontainer.WaitConditionNotRunning)
185         for {
186                 select {
187                 case waitBody := <-waitOk:
188                         e.logf("Container exited with code: %v", waitBody.StatusCode)
189                         // wait for stdout/stderr to complete
190                         <-e.doneIO
191                         return int(waitBody.StatusCode), nil
192
193                 case err := <-waitErr:
194                         return -1, fmt.Errorf("container wait: %v", err)
195
196                 case <-ctx.Done():
197                         return -1, ctx.Err()
198
199                 case err := <-watchdogErr:
200                         return -1, err
201                 }
202         }
203 }
204
205 func (e *dockerExecutor) startIO(stdin io.Reader, stdout, stderr io.Writer) error {
206         resp, err := e.dockerclient.ContainerAttach(context.TODO(), e.containerID, dockertypes.ContainerAttachOptions{
207                 Stream: true,
208                 Stdin:  stdin != nil,
209                 Stdout: true,
210                 Stderr: true,
211         })
212         if err != nil {
213                 return fmt.Errorf("error attaching container stdin/stdout/stderr streams: %v", err)
214         }
215         var errStdin error
216         if stdin != nil {
217                 go func() {
218                         errStdin = e.handleStdin(stdin, resp.Conn, resp.CloseWrite)
219                 }()
220         }
221         e.doneIO = make(chan struct{})
222         go func() {
223                 e.errIO = e.handleStdoutStderr(stdout, stderr, resp.Reader)
224                 if e.errIO == nil && errStdin != nil {
225                         e.errIO = errStdin
226                 }
227                 close(e.doneIO)
228         }()
229         return nil
230 }
231
232 func (e *dockerExecutor) handleStdin(stdin io.Reader, conn io.Writer, closeConn func() error) error {
233         defer closeConn()
234         _, err := io.Copy(conn, stdin)
235         if err != nil {
236                 return fmt.Errorf("While writing to docker container on stdin: %v", err)
237         }
238         return nil
239 }
240
241 // Handle docker log protocol; see
242 // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
243 func (e *dockerExecutor) handleStdoutStderr(stdout, stderr io.Writer, reader io.Reader) error {
244         header := make([]byte, 8)
245         var err error
246         for err == nil {
247                 _, err = io.ReadAtLeast(reader, header, 8)
248                 if err != nil {
249                         if err == io.EOF {
250                                 err = nil
251                         }
252                         break
253                 }
254                 readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
255                 if header[0] == 1 {
256                         _, err = io.CopyN(stdout, reader, readsize)
257                 } else {
258                         // stderr
259                         _, err = io.CopyN(stderr, reader, readsize)
260                 }
261         }
262         if err != nil {
263                 return fmt.Errorf("error copying stdout/stderr from docker: %v", err)
264         }
265         return nil
266 }
267
268 func (e *dockerExecutor) Close() {
269         e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})
270 }