2017/06/19 追記

エラーハンドリングにバグがあったので修正しました
- channelに何か流したとき、channelから取り出されないと流し込んだ側をブロックするのを忘れていました

- errored <- err
+ go func() {
+   errored <- err
+ }()

前回までで以下のことをやったので

今回は、Goのソースコードが動いているマシンじゃないマシンに対してDockerコンテナの起動を指示する実装をしてみたいなと思いました。

準備

複数のDockerホストを再現するためにdocker-machineを入手しとく
- docker for mac入れたときに入ってくるはずです
ドライバとしてVirtualBoxをインストールしとく
前回までで使っていたclient.NewEnvClientとclient.NewClientの実装の違いを見とく
docker-machine create --driver virtualbox hoge
- docker-machine env hoge で得られるDOCKER_HOSTとかDOCKER_CERT_PATHとかあとで使う
docker-machine create --driver virtualbox fuga
- docker-machine env fuga で得られるDOCKER_HOSTとかDOCKER_CERT_PATHとかあとで使う

動くコード

package main

import (
    "bufio"
    "context"
    "fmt"
    "net/http"
    "path/filepath"

    "github.com/docker/docker/api/types"
    "github.com/docker/docker/api/types/container"
    "github.com/docker/docker/api/types/mount"
    "github.com/docker/docker/api/types/network"
    "github.com/docker/docker/client"
    "github.com/docker/go-connections/tlsconfig"
)

// WorkerMachine is a host on which any container runs,
// This shoule also have file mout source path.
type WorkerMachine struct {
    Host        string
    CertPath    string
    MountSource string
}

// NewClient provides Docker client for **THIS** host machine.
func (machine WorkerMachine) NewClient() (*client.Client, error) {
    options := tlsconfig.Options{
        CAFile:             filepath.Join(machine.CertPath, "ca.pem"),
        CertFile:           filepath.Join(machine.CertPath, "cert.pem"),
        KeyFile:            filepath.Join(machine.CertPath, "key.pem"),
        InsecureSkipVerify: false,
    }
    tlsc, err := tlsconfig.Client(options)
    if err != nil {
        return nil, err
    }
    httpClient := &http.Client{
        Transport: &http.Transport{
            TLSClientConfig: tlsc,
        },
    }
    headers := map[string]string{}
    return client.NewClient(machine.Host, "1.30", httpClient, headers)
}

// This is the definition of docker host machines
// on which any tasks would be executed.
var machines = []WorkerMachine{
    WorkerMachine{
        Host:        "tcp://192.168.99.100:2376",
        CertPath:    "/Users/otiai10/.docker/machine/machines/hoge",
        MountSource: "/Users/otiai10/tmp/hoge",
    },
    WorkerMachine{
        Host:        "tcp://192.168.99.101:2376",
        CertPath:    "/Users/otiai10/.docker/machine/machines/fuga",
        MountSource: "/Users/otiai10/tmp/fuga",
    },
}

// task function defines what to do on that machine.
// Because hijacking stdout of container returns io.Reader
// and also because the first out put would be lost when container starts BEFORE hijacking,
// this function should return channel which can message error,
func task(index int, machine WorkerMachine) (errored chan error) {

    errored = make(chan error)

    c, err := machine.NewClient()
    if err != nil {
        go func() {
            errored <- err
        }()
        return
    }
    defer c.Close()
    ctx := context.Background()

    // Ensure Target Image to exist inside this machine
    r, err := c.ImagePull(ctx, "otiai10/foo", types.ImagePullOptions{})
    if err != nil {
        go func() {
            errored <- err
        }()
        return
    }
    defer r.Close()
    fmt.Printf("[%d][IMAGE PULL][START]", index)
    for scanner := bufio.NewScanner(r); scanner.Scan(); {
        fmt.Printf(".")
    }
    fmt.Printf("\n[%d][IMAGE PULL][FINISH]\n", index)

    // Create container in this machine
    containerconfig := &container.Config{Image: "otiai10/foo"}
    hostconfig := &container.HostConfig{
        Mounts: []mount.Mount{
            mount.Mount{
                Type:   mount.TypeBind,
                Source: machine.MountSource,
                Target: "/var/data",
            },
        },
        AutoRemove: true,
    }
    networkingconfig := &network.NetworkingConfig{}
    body, err := c.ContainerCreate(
        ctx,
        containerconfig, hostconfig, networkingconfig,
        fmt.Sprintf("work-%d", index),
    )
    if err != nil {
        go func() {
            errored <- err
        }()
        return
    }

    // Hijack container's Stdout, and it should be hijacked BEFORE container starts
    hijacked, err := c.ContainerAttach(ctx, body.ID, types.ContainerAttachOptions{
        Stream: true,
        Stdout: true,
    })
    if err != nil {
        go func() {
            errored <- err
        }()
        return
    }
    go func() {
        for scanner := bufio.NewScanner(hijacked.Reader); scanner.Scan(); {
            fmt.Printf("[%d][CONTAINER STDOUT]\t%s\n", index, scanner.Text())
        }
        hijacked.Close()
        fmt.Printf("[%[1]d][COMPLETED] Congrats! `work-%[1]d` is completed without errors!\n", index)
        errored <- nil
    }()

    // OK, now it's time to start the container!
    if err := c.ContainerStart(ctx, body.ID, types.ContainerStartOptions{}); err != nil {
        go func() {
            errored <- err
        }()
        return
    }

    return
}

func main() {
    for i, machine := range machines {
        errored := task(i, machine)
        if err := <-errored; err != nil {
            panic(err)
        }
        close(errored)
    }
}

これを実行すると、

f:id:otiai10:20170607171124p:plain

となり、複数の異なるホストマシンで、異なるContainerのプロセスが、同期的に逐次実行されていることがわかる。

なお、二回目からは、各ホストマシンにimageがすでにあるので

f:id:otiai10:20170607171538p:plain

こういう感じになるし、あえて

% eval $(docker-machine env fuga)
% docker images -a
REPOSITORY          TAG                 IMAGE ID            CREATED             SIZE
otiai10/foo         latest              92af8256ff46        39 minutes ago      193 MB
% docker rmi otiai10/foo
%

このようにfugaの方だけでotiai10/fooimageを消しとくと、

f:id:otiai10:20170607172406p:plain