Merge pull request #16 from hashicorp/f-docker

Docker Driver

client/driver/docker.go

@@ -0,0 +1,240 @@
+package driver
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"os/exec"
+	"regexp"
+	"strings"
+
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+var (
+	reDockerVersion = regexp.MustCompile("Docker version ([\\d\\.]+),.+")
+	reDockerSha     = regexp.MustCompile("^[a-f0-9]{64}$")
+)
+
+type DockerDriver struct {
+	logger *log.Logger
+}
+
+type dockerPID struct {
+	ImageID     string
+	ContainerID string
+}
+
+type dockerHandle struct {
+	logger      *log.Logger
+	imageID     string
+	containerID string
+	waitCh      chan error
+	doneCh      chan struct{}
+}
+
+func NewDockerDriver(logger *log.Logger) Driver {
+	d := &DockerDriver{
+		logger: logger,
+	}
+	return d
+}
+
+func (d *DockerDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
+	outBytes, err := exec.Command("docker", "-v").Output()
+	out := strings.TrimSpace(string(outBytes))
+	if err != nil {
+		return false, nil
+	}
+
+	matches := reDockerVersion.FindStringSubmatch(out)
+	if len(matches) != 2 {
+		return false, fmt.Errorf("Unable to parse docker version string: %#v", matches)
+	}
+
+	node.Attributes["driver.docker"] = "true"
+	node.Attributes["driver.docker.version"] = matches[1]
+
+	return true, nil
+}
+
+func (d *DockerDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
+	// Get the image from config
+	image, ok := task.Config["image"]
+	if !ok || image == "" {
+		return nil, fmt.Errorf("Image not specified")
+	}
+
+	// Download the image
+	pull, err := exec.Command("docker", "pull", image).CombinedOutput()
+	if err != nil {
+		d.logger.Printf("[ERROR] driver.docker %s", pull)
+		return nil, fmt.Errorf("Failed to pull `%s`: %s", image, err)
+	}
+	d.logger.Printf("[DEBUG] driver.docker docker pull %s:\n%s", image, pull)
+
+	// Get the image ID (sha256). We need to keep track of this in case another
+	// process pulls down a newer version of the image.
+	imageIDBytes, err := exec.Command("docker", "images", "-q", "--no-trunc", image).CombinedOutput()
+	imageID := strings.TrimSpace(string(imageIDBytes))
+	if err != nil || imageID == "" {
+		d.logger.Printf("[ERROR] driver.docker %s", imageID)
+		return nil, fmt.Errorf("Failed to determine image id for `%s`: %s", image, err)
+	}
+	if !reDockerSha.MatchString(imageID) {
+		return nil, fmt.Errorf("Image id not in expected format (sha256); found %s", imageID)
+	}
+	d.logger.Printf("[DEBUG] driver.docker using image %s", imageID)
+	d.logger.Printf("[INFO] driver.docker downloaded image %s as %s", image, imageID)
+
+	// Create a container
+	containerIDBytes, err := exec.Command("docker", "create", imageID).CombinedOutput()
+	containerID := strings.TrimSpace(string(containerIDBytes))
+	if err != nil {
+		d.logger.Printf("[ERROR] driver.docker %s", containerID)
+		return nil, fmt.Errorf("Failed to create container from image %s", image)
+	}
+	if !reDockerSha.MatchString(containerID) {
+		return nil, fmt.Errorf("Container id not in expected format (sha256); found %s", containerID)
+	}
+	d.logger.Printf("[INFO] driver.docker created container %s", containerID)
+
+	// Start the container. The output is containerID again so don't need to
+	// validate it. Also, the docker daemon is responsible for running this so
+	// start will return immediately. We'll use the containerID with the docker
+	// watch command to track its status.
+	//
+	// Note: at some point we will want to use docker run instead, to set extra
+	// options and such. You should use docker run -d and still use wait to
+	// check whether the process is available so we are able to re-wait if the
+	// nomad process is restarted. Also, you will need to parse the containerID
+	// out of the run command output since run combines pull, create and start
+	// into a single command.
+	startBytes, err := exec.Command("docker", "start", containerID).CombinedOutput()
+	if err != nil {
+		d.logger.Printf("[ERROR] driver.docker %s", strings.TrimSpace(string(startBytes)))
+		return nil, fmt.Errorf("Failed to start container %s", containerID)
+	}
+	d.logger.Printf("[INFO] driver.docker started container %s", containerID)
+
+	// Return a driver handle
+	h := &dockerHandle{
+		imageID:     imageID,
+		containerID: containerID,
+		doneCh:      make(chan struct{}),
+		waitCh:      make(chan error, 1),
+	}
+	go h.run()
+	return h, nil
+}
+
+func (d *DockerDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
+	// Split the handle
+	pidBytes := []byte(strings.TrimPrefix(handleID, "DOCKER:"))
+	pid := &dockerPID{}
+	err := json.Unmarshal(pidBytes, pid)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to parse handle '%s': %v", handleID, err)
+	}
+	log.Printf("[INFO] driver.docker Re-attaching to docker process: %s", handleID)
+
+	// Look for a running container with this ID
+	// docker ps does not return an exit code if there are no matching processes
+	// so we have to read the output and compare it to our known containerID
+	psBytes, err := exec.Command("docker", "ps", "-q", "--no-trunc",
+		fmt.Sprintf("-f=id=%s", pid.ContainerID)).Output()
+	ps := strings.TrimSpace(string(psBytes))
+	if err != nil {
+		return nil, fmt.Errorf("Failed to find container %s: %v", pid.ContainerID, err)
+	} else if ps != pid.ContainerID {
+		return nil, fmt.Errorf("Container ID does not match; expected %s found %s", pid.ContainerID, ps)
+	}
+
+	// Return a driver handle
+	h := &dockerHandle{
+		imageID:     pid.ImageID,
+		containerID: pid.ContainerID,
+		doneCh:      make(chan struct{}),
+		waitCh:      make(chan error, 1),
+	}
+	go h.run()
+	return h, nil
+}
+
+func (h *dockerHandle) ID() string {
+	// Return a handle to the PID
+	pid := dockerPID{
+		ImageID:     h.imageID,
+		ContainerID: h.containerID,
+	}
+	data, err := json.Marshal(pid)
+	if err != nil {
+		log.Printf("[ERROR] failed to marshal docker PID to JSON: %s", err)
+	}
+	return fmt.Sprintf("DOCKER:%s", string(data))
+}
+
+func (h *dockerHandle) WaitCh() chan error {
+	return h.waitCh
+}
+
+func (h *dockerHandle) Update(task *structs.Task) error {
+	// Update is not possible
+	return nil
+}
+
+// Kill is used to terminate the task. This uses docker stop -t 5
+func (h *dockerHandle) Kill() error {
+	// Stop the container
+	stop, err := exec.Command("docker", "stop", "-t", "5", h.containerID).CombinedOutput()
+	if err != nil {
+		log.Printf("[ERROR] driver.docker %s", stop)
+		return fmt.Errorf("Failed to stop container %s: %s", h.containerID, err)
+	}
+	log.Printf("[INFO] driver.docker stopped container %s", h.containerID)
+
+	// Cleanup container
+	rmContainer, err := exec.Command("docker", "rm", h.containerID).CombinedOutput()
+	if err != nil {
+		log.Printf("[ERROR] driver.docker %s", rmContainer)
+		return fmt.Errorf("Failed to remove container %s: %s", h.containerID, err)
+	}
+	log.Printf("[INFO] driver.docker removed container %s", h.containerID)
+
+	// Cleanup image. This operation may fail if the image is in use by another
+	// job. That is OK. Will we log a message but continue.
+	_, err = exec.Command("docker", "rmi", h.imageID).CombinedOutput()
+	if err != nil {
+		log.Printf("[WARN] driver.docker failed to remove image %s; it may still be in use", h.imageID)
+	} else {
+		log.Printf("[INFO] driver.docker removed image %s", h.imageID)
+	}
+	return nil
+}
+
+func (h *dockerHandle) run() {
+	// Wait for it...
+	waitBytes, err := exec.Command("docker", "wait", h.containerID).Output()
+	if err != nil {
+		h.logger.Printf("[ERROR] driver.docker unable to wait for %s; container already terminated", h.containerID)
+	}
+	wait := strings.TrimSpace(string(waitBytes))
+
+	// If the container failed, try to get the last 10 lines of logs for our
+	// error message.
+	if wait != "0" {
+		var logsBytes []byte
+		logsBytes, err = exec.Command("docker", "logs", "--tail=10", h.containerID).Output()
+		logs := string(logsBytes)
+		if err == nil {
+			err = fmt.Errorf("%s", logs)
+		}
+	}
+
+	close(h.doneCh)
+	if err != nil {
+		h.waitCh <- err
+	}
+	close(h.waitCh)
+}

client/driver/docker_test.go

@@ -0,0 +1,152 @@
+package driver
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+var dockerLocated bool = true
+
+func TestDockerDriver_Handle(t *testing.T) {
+	h := &dockerHandle{
+		imageID:     "imageid",
+		containerID: "containerid",
+		doneCh:      make(chan struct{}),
+		waitCh:      make(chan error, 1),
+	}
+
+	actual := h.ID()
+	expected := `DOCKER:{"ImageID":"imageid","ContainerID":"containerid"}`
+	if actual != expected {
+		t.Errorf("Expected `%s`, found `%s`", expected, actual)
+	}
+}
+
+func TestDockerDriver_Fingerprint(t *testing.T) {
+	d := NewDockerDriver(testLogger())
+	node := &structs.Node{
+		Attributes: make(map[string]string),
+	}
+	apply, err := d.Fingerprint(&config.Config{}, node)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if !apply {
+		t.Fatalf("should apply")
+	}
+	if node.Attributes["driver.docker"] == "" {
+		dockerLocated = false
+		t.Fatalf("Docker not found. The remainder of the docker tests will be skipped.")
+	}
+	t.Logf("Found docker version %s", node.Attributes["driver.docker.version"])
+}
+
+func TestDockerDriver_StartOpen_Wait(t *testing.T) {
+	if !dockerLocated {
+		t.SkipNow()
+	}
+	ctx := NewExecContext()
+	d := NewDockerDriver(testLogger())
+
+	task := &structs.Task{
+		Config: map[string]string{
+			"image": "cbednarski/python-demo",
+		},
+	}
+	handle, err := d.Start(ctx, task)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if handle == nil {
+		t.Fatalf("missing handle")
+	}
+	defer handle.Kill()
+
+	// Attempt to open
+	handle2, err := d.Open(ctx, handle.ID())
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if handle2 == nil {
+		t.Fatalf("missing handle")
+	}
+}
+
+func TestDockerDriver_Start_Wait(t *testing.T) {
+	if !dockerLocated {
+		t.SkipNow()
+	}
+	ctx := NewExecContext()
+	d := NewDockerDriver(testLogger())
+
+	task := &structs.Task{
+		Config: map[string]string{
+			"image": "cbednarski/python-demo",
+		},
+	}
+	handle, err := d.Start(ctx, task)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if handle == nil {
+		t.Fatalf("missing handle")
+	}
+	defer handle.Kill()
+
+	// Update should be a no-op
+	err = handle.Update(task)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	select {
+	case err := <-handle.WaitCh():
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatalf("timeout")
+	}
+}
+
+func TestDockerDriver_Start_Kill_Wait(t *testing.T) {
+	if !dockerLocated {
+		t.SkipNow()
+	}
+	ctx := NewExecContext()
+	d := NewDockerDriver(testLogger())
+
+	task := &structs.Task{
+		Config: map[string]string{
+			"image": "cbednarski/python-demo",
+		},
+	}
+	handle, err := d.Start(ctx, task)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if handle == nil {
+		t.Fatalf("missing handle")
+	}
+	defer handle.Kill()
+
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		err := handle.Kill()
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	select {
+	case err := <-handle.WaitCh():
+		if err == nil {
+			t.Fatalf("should err: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatalf("timeout")
+	}
+}