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         "os/exec"
  12         "strings"
  13         "sync/atomic"
  14         "time"
  15
  16         "git.arvados.org/arvados.git/sdk/go/arvados"
  17         dockertypes "github.com/docker/docker/api/types"
  18         dockercontainer "github.com/docker/docker/api/types/container"
  19         dockerclient "github.com/docker/docker/client"
  20         "golang.org/x/net/context"
  21 )
  22
  23 // Docker daemon won't let you set a limit less than ~10 MiB
  24 const minDockerRAM = int64(16 * 1024 * 1024)
  25
  26 type dockerExecutor struct {
  27         containerUUID    string
  28         logf             func(string, ...interface{})
  29         watchdogInterval time.Duration
  30         dockerclient     *dockerclient.Client
  31         containerID      string
  32         savedIPAddress   atomic.Value
  33         doneIO           chan struct{}
  34         errIO            error
  35 }
  36
  37 func newDockerExecutor(containerUUID string, logf func(string, ...interface{}), watchdogInterval time.Duration) (*dockerExecutor, error) {
  38         // API version 1.21 corresponds to Docker 1.9, which is
  39         // currently the minimum version we want to support.
  40         client, err := dockerclient.NewClient(dockerclient.DefaultDockerHost, "1.21", nil, nil)
  41         if watchdogInterval < 1 {
  42                 watchdogInterval = time.Minute
  43         }
  44         return &dockerExecutor{
  45                 containerUUID:    containerUUID,
  46                 logf:             logf,
  47                 watchdogInterval: watchdogInterval,
  48                 dockerclient:     client,
  49         }, err
  50 }
  51
  52 func (e *dockerExecutor) Runtime() string { return "docker" }
  53
  54 func (e *dockerExecutor) LoadImage(imageID string, imageTarballPath string, container arvados.Container, arvMountPoint string,
  55         containerClient *arvados.Client) error {
  56         _, _, err := e.dockerclient.ImageInspectWithRaw(context.TODO(), imageID)
  57         if err == nil {
  58                 // already loaded
  59                 return nil
  60         }
  61
  62         f, err := os.Open(imageTarballPath)
  63         if err != nil {
  64                 return err
  65         }
  66         defer f.Close()
  67         resp, err := e.dockerclient.ImageLoad(context.TODO(), f, true)
  68         if err != nil {
  69                 return fmt.Errorf("While loading container image into Docker: %v", err)
  70         }
  71         defer resp.Body.Close()
  72         buf, _ := ioutil.ReadAll(resp.Body)
  73         e.logf("loaded image: response %s", buf)
  74         return nil
  75 }
  76
  77 func (e *dockerExecutor) config(spec containerSpec) (dockercontainer.Config, dockercontainer.HostConfig) {
  78         e.logf("Creating Docker container")
  79         cfg := dockercontainer.Config{
  80                 Image:        spec.Image,
  81                 Cmd:          spec.Command,
  82                 WorkingDir:   spec.WorkingDir,
  83                 Volumes:      map[string]struct{}{},
  84                 OpenStdin:    spec.Stdin != nil,
  85                 StdinOnce:    spec.Stdin != nil,
  86                 AttachStdin:  spec.Stdin != nil,
  87                 AttachStdout: true,
  88                 AttachStderr: true,
  89         }
  90         if cfg.WorkingDir == "." {
  91                 cfg.WorkingDir = ""
  92         }
  93         for k, v := range spec.Env {
  94                 cfg.Env = append(cfg.Env, k+"="+v)
  95         }
  96         if spec.RAM > 0 && spec.RAM < minDockerRAM {
  97                 spec.RAM = minDockerRAM
  98         }
  99         hostCfg := dockercontainer.HostConfig{
 100                 LogConfig: dockercontainer.LogConfig{
 101                         Type: "none",
 102                 },
 103                 NetworkMode: dockercontainer.NetworkMode("none"),
 104                 Resources: dockercontainer.Resources{
 105                         CgroupParent: spec.CgroupParent,
 106                         NanoCPUs:     int64(spec.VCPUs) * 1000000000,
 107                         Memory:       spec.RAM, // RAM
 108                         MemorySwap:   spec.RAM, // RAM+swap
 109                         KernelMemory: spec.RAM, // kernel portion
 110                 },
 111         }
 112         if spec.CUDADeviceCount != 0 {
 113                 var deviceIds []string
 114                 if cudaVisibleDevices := os.Getenv("CUDA_VISIBLE_DEVICES"); cudaVisibleDevices != "" {
 115                         // If a resource manager such as slurm or LSF told
 116                         // us to select specific devices we need to propagate that.
 117                         deviceIds = strings.Split(cudaVisibleDevices, ",")
 118                 }
 119
 120                 deviceCount := spec.CUDADeviceCount
 121                 if len(deviceIds) > 0 {
 122                         // Docker won't accept both non-empty
 123                         // DeviceIDs and a non-zero Count
 124                         //
 125                         // (it turns out "Count" is a dumb fallback
 126                         // that just allocates device 0, 1, 2, ...,
 127                         // Count-1)
 128                         deviceCount = 0
 129                 }
 130
 131                 // Capabilities are confusing.  The driver has generic
 132                 // capabilities "gpu" and "nvidia" but then there's
 133                 // additional capabilities "compute" and "utility"
 134                 // that are passed to nvidia-container-cli.
 135                 //
 136                 // "compute" means include the CUDA libraries and
 137                 // "utility" means include the CUDA utility programs
 138                 // (like nvidia-smi).
 139                 //
 140                 // https://github.com/moby/moby/blob/7b9275c0da707b030e62c96b679a976f31f929d3/daemon/nvidia_linux.go#L37
 141                 // https://github.com/containerd/containerd/blob/main/contrib/nvidia/nvidia.go
 142                 hostCfg.Resources.DeviceRequests = append(hostCfg.Resources.DeviceRequests, dockercontainer.DeviceRequest{
 143                         Driver:       "nvidia",
 144                         Count:        deviceCount,
 145                         DeviceIDs:    deviceIds,
 146                         Capabilities: [][]string{[]string{"gpu", "nvidia", "compute", "utility"}},
 147                 })
 148         }
 149         for path, mount := range spec.BindMounts {
 150                 bind := mount.HostPath + ":" + path
 151                 if mount.ReadOnly {
 152                         bind += ":ro"
 153                 }
 154                 hostCfg.Binds = append(hostCfg.Binds, bind)
 155         }
 156         if spec.EnableNetwork {
 157                 hostCfg.NetworkMode = dockercontainer.NetworkMode(spec.NetworkMode)
 158         }
 159         return cfg, hostCfg
 160 }
 161
 162 func (e *dockerExecutor) Create(spec containerSpec) error {
 163         cfg, hostCfg := e.config(spec)
 164         created, err := e.dockerclient.ContainerCreate(context.TODO(), &cfg, &hostCfg, nil, e.containerUUID)
 165         if err != nil {
 166                 return fmt.Errorf("While creating container: %v", err)
 167         }
 168         e.containerID = created.ID
 169         return e.startIO(spec.Stdin, spec.Stdout, spec.Stderr)
 170 }
 171
 172 func (e *dockerExecutor) CgroupID() string {
 173         return e.containerID
 174 }
 175
 176 func (e *dockerExecutor) Start() error {
 177         return e.dockerclient.ContainerStart(context.TODO(), e.containerID, dockertypes.ContainerStartOptions{})
 178 }
 179
 180 func (e *dockerExecutor) Stop() error {
 181         err := e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})
 182         if err != nil && strings.Contains(err.Error(), "No such container: "+e.containerID) {
 183                 err = nil
 184         }
 185         return err
 186 }
 187
 188 // Wait for the container to terminate, capture the exit code, and
 189 // wait for stdout/stderr logging to finish.
 190 func (e *dockerExecutor) Wait(ctx context.Context) (int, error) {
 191         ctx, cancel := context.WithCancel(ctx)
 192         defer cancel()
 193         watchdogErr := make(chan error, 1)
 194         go func() {
 195                 ticker := time.NewTicker(e.watchdogInterval)
 196                 defer ticker.Stop()
 197                 for range ticker.C {
 198                         dctx, dcancel := context.WithDeadline(ctx, time.Now().Add(e.watchdogInterval))
 199                         ctr, err := e.dockerclient.ContainerInspect(dctx, e.containerID)
 200                         dcancel()
 201                         if ctx.Err() != nil {
 202                                 // Either the container already
 203                                 // exited, or our caller is trying to
 204                                 // kill it.
 205                                 return
 206                         } else if err != nil {
 207                                 e.logf("Error inspecting container: %s", err)
 208                                 watchdogErr <- err
 209                                 return
 210                         } else if ctr.State == nil || !(ctr.State.Running || ctr.State.Status == "created") {
 211                                 watchdogErr <- fmt.Errorf("Container is not running: State=%v", ctr.State)
 212                                 return
 213                         }
 214                 }
 215         }()
 216
 217         waitOk, waitErr := e.dockerclient.ContainerWait(ctx, e.containerID, dockercontainer.WaitConditionNotRunning)
 218         for {
 219                 select {
 220                 case waitBody := <-waitOk:
 221                         // wait for stdout/stderr to complete
 222                         <-e.doneIO
 223                         return int(waitBody.StatusCode), nil
 224
 225                 case err := <-waitErr:
 226                         return -1, fmt.Errorf("container wait: %v", err)
 227
 228                 case <-ctx.Done():
 229                         return -1, ctx.Err()
 230
 231                 case err := <-watchdogErr:
 232                         return -1, err
 233                 }
 234         }
 235 }
 236
 237 func (e *dockerExecutor) startIO(stdin io.Reader, stdout, stderr io.Writer) error {
 238         resp, err := e.dockerclient.ContainerAttach(context.TODO(), e.containerID, dockertypes.ContainerAttachOptions{
 239                 Stream: true,
 240                 Stdin:  stdin != nil,
 241                 Stdout: true,
 242                 Stderr: true,
 243         })
 244         if err != nil {
 245                 return fmt.Errorf("error attaching container stdin/stdout/stderr streams: %v", err)
 246         }
 247         var errStdin error
 248         if stdin != nil {
 249                 go func() {
 250                         errStdin = e.handleStdin(stdin, resp.Conn, resp.CloseWrite)
 251                 }()
 252         }
 253         e.doneIO = make(chan struct{})
 254         go func() {
 255                 e.errIO = e.handleStdoutStderr(stdout, stderr, resp.Reader)
 256                 if e.errIO == nil && errStdin != nil {
 257                         e.errIO = errStdin
 258                 }
 259                 close(e.doneIO)
 260         }()
 261         return nil
 262 }
 263
 264 func (e *dockerExecutor) handleStdin(stdin io.Reader, conn io.Writer, closeConn func() error) error {
 265         defer closeConn()
 266         _, err := io.Copy(conn, stdin)
 267         if err != nil {
 268                 return fmt.Errorf("While writing to docker container on stdin: %v", err)
 269         }
 270         return nil
 271 }
 272
 273 // Handle docker log protocol; see
 274 // https://docs.docker.com/engine/reference/api/docker_remote_api_v1.15/#attach-to-a-container
 275 func (e *dockerExecutor) handleStdoutStderr(stdout, stderr io.Writer, reader io.Reader) error {
 276         header := make([]byte, 8)
 277         var err error
 278         for err == nil {
 279                 _, err = io.ReadAtLeast(reader, header, 8)
 280                 if err != nil {
 281                         if err == io.EOF {
 282                                 err = nil
 283                         }
 284                         break
 285                 }
 286                 readsize := int64(header[7]) | (int64(header[6]) << 8) | (int64(header[5]) << 16) | (int64(header[4]) << 24)
 287                 if header[0] == 1 {
 288                         _, err = io.CopyN(stdout, reader, readsize)
 289                 } else {
 290                         // stderr
 291                         _, err = io.CopyN(stderr, reader, readsize)
 292                 }
 293         }
 294         if err != nil {
 295                 return fmt.Errorf("error copying stdout/stderr from docker: %v", err)
 296         }
 297         return nil
 298 }
 299
 300 func (e *dockerExecutor) Close() {
 301         e.dockerclient.ContainerRemove(context.TODO(), e.containerID, dockertypes.ContainerRemoveOptions{Force: true})
 302 }
 303
 304 func (e *dockerExecutor) InjectCommand(ctx context.Context, detachKeys, username string, usingTTY bool, injectcmd []string) (*exec.Cmd, error) {
 305         cmd := exec.CommandContext(ctx, "docker", "exec", "-i", "--detach-keys="+detachKeys, "--user="+username)
 306         if usingTTY {
 307                 cmd.Args = append(cmd.Args, "-t")
 308         }
 309         cmd.Args = append(cmd.Args, e.containerID)
 310         cmd.Args = append(cmd.Args, injectcmd...)
 311         return cmd, nil
 312 }
 313
 314 func (e *dockerExecutor) IPAddress() (string, error) {
 315         if ip, ok := e.savedIPAddress.Load().(*string); ok {
 316                 return *ip, nil
 317         }
 318         ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Minute))
 319         defer cancel()
 320         ctr, err := e.dockerclient.ContainerInspect(ctx, e.containerID)
 321         if err != nil {
 322                 return "", fmt.Errorf("cannot get docker container info: %s", err)
 323         }
 324         ip := ctr.NetworkSettings.IPAddress
 325         if ip == "" {
 326                 // TODO: try to enable networking if it wasn't
 327                 // already enabled when the container was
 328                 // created.
 329                 return "", fmt.Errorf("container has no IP address")
 330         }
 331         e.savedIPAddress.Store(&ip)
 332         return ip, nil
 333 }