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) Create(spec containerSpec) error {
  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
 127         created, err := e.dockerclient.ContainerCreate(context.TODO(), &cfg, &hostCfg, nil, e.containerUUID)
 128         if err != nil {
 129                 return fmt.Errorf("While creating container: %v", err)
 130         }
 131         e.containerID = created.ID
 132         return e.startIO(spec.Stdin, spec.Stdout, spec.Stderr)
 133 }
 134
 135 func (e *dockerExecutor) CgroupID() string {
 136         return e.containerID
 137 }
 138
 139 func (e *dockerExecutor) Start() error {
 140         return e.dockerclient.ContainerStart(context.TODO(), e.containerID, dockertypes.ContainerStartOptions{})
 141 }
 142
 143 func (e *dockerExecutor) Stop() error {
 144         err := e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})
 145         if err != nil && strings.Contains(err.Error(), "No such container: "+e.containerID) {
 146                 err = nil
 147         }
 148         return err
 149 }
 150
 151 // Wait for the container to terminate, capture the exit code, and
 152 // wait for stdout/stderr logging to finish.
 153 func (e *dockerExecutor) Wait(ctx context.Context) (int, error) {
 154         ctx, cancel := context.WithCancel(ctx)
 155         defer cancel()
 156         watchdogErr := make(chan error, 1)
 157         go func() {
 158                 ticker := time.NewTicker(e.watchdogInterval)
 159                 defer ticker.Stop()
 160                 for range ticker.C {
 161                         dctx, dcancel := context.WithDeadline(ctx, time.Now().Add(e.watchdogInterval))
 162                         ctr, err := e.dockerclient.ContainerInspect(dctx, e.containerID)
 163                         dcancel()
 164                         if ctx.Err() != nil {
 165                                 // Either the container already
 166                                 // exited, or our caller is trying to
 167                                 // kill it.
 168                                 return
 169                         } else if err != nil {
 170                                 e.logf("Error inspecting container: %s", err)
 171                                 watchdogErr <- err
 172                                 return
 173                         } else if ctr.State == nil || !(ctr.State.Running || ctr.State.Status == "created") {
 174                                 watchdogErr <- fmt.Errorf("Container is not running: State=%v", ctr.State)
 175                                 return
 176                         }
 177                 }
 178         }()
 179
 180         waitOk, waitErr := e.dockerclient.ContainerWait(ctx, e.containerID, dockercontainer.WaitConditionNotRunning)
 181         for {
 182                 select {
 183                 case waitBody := <-waitOk:
 184                         e.logf("Container exited with code: %v", waitBody.StatusCode)
 185                         // wait for stdout/stderr to complete
 186                         <-e.doneIO
 187                         return int(waitBody.StatusCode), nil
 188
 189                 case err := <-waitErr:
 190                         return -1, fmt.Errorf("container wait: %v", err)
 191
 192                 case <-ctx.Done():
 193                         return -1, ctx.Err()
 194
 195                 case err := <-watchdogErr:
 196                         return -1, err
 197                 }
 198         }
 199 }
 200
 201 func (e *dockerExecutor) startIO(stdin io.Reader, stdout, stderr io.Writer) error {
 202         resp, err := e.dockerclient.ContainerAttach(context.TODO(), e.containerID, dockertypes.ContainerAttachOptions{
 203                 Stream: true,
 204                 Stdin:  stdin != nil,
 205                 Stdout: true,
 206                 Stderr: true,
 207         })
 208         if err != nil {
 209                 return fmt.Errorf("error attaching container stdin/stdout/stderr streams: %v", err)
 210         }
 211         var errStdin error
 212         if stdin != nil {
 213                 go func() {
 214                         errStdin = e.handleStdin(stdin, resp.Conn, resp.CloseWrite)
 215                 }()
 216         }
 217         e.doneIO = make(chan struct{})
 218         go func() {
 219                 e.errIO = e.handleStdoutStderr(stdout, stderr, resp.Reader)
 220                 if e.errIO == nil && errStdin != nil {
 221                         e.errIO = errStdin
 222                 }
 223                 close(e.doneIO)
 224         }()
 225         return nil
 226 }
 227
 228 func (e *dockerExecutor) handleStdin(stdin io.Reader, conn io.Writer, closeConn func() error) error {
 229         defer closeConn()
 230         _, err := io.Copy(conn, stdin)
 231         if err != nil {
 232                 return fmt.Errorf("While writing to docker container on stdin: %v", err)
 233         }
 234         return nil
 235 }
 236
 237 // Handle docker log protocol; see
 238 // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
 239 func (e *dockerExecutor) handleStdoutStderr(stdout, stderr io.Writer, reader io.Reader) error {
 240         header := make([]byte, 8)
 241         var err error
 242         for err == nil {
 243                 _, err = io.ReadAtLeast(reader, header, 8)
 244                 if err != nil {
 245                         if err == io.EOF {
 246                                 err = nil
 247                         }
 248                         break
 249                 }
 250                 readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
 251                 if header[0] == 1 {
 252                         _, err = io.CopyN(stdout, reader, readsize)
 253                 } else {
 254                         // stderr
 255                         _, err = io.CopyN(stderr, reader, readsize)
 256                 }
 257         }
 258         if err != nil {
 259                 return fmt.Errorf("error copying stdout/stderr from docker: %v", err)
 260         }
 261         return nil
 262 }
 263
 264 func (e *dockerExecutor) Close() {
 265         e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})
 266 }