Merge pull request #165 from achanda/rkt

Add a driver for rkt

client/driver/driver.go

@@ -19,6 +19,7 @@ var BuiltinDrivers = map[string]Factory{
 	"exec":   NewExecDriver,
 	"java":   NewJavaDriver,
 	"qemu":   NewQemuDriver,
+	"rkt":    NewRktDriver,
 }
 
 // NewDriver is used to instantiate and return a new driver

client/driver/rkt.go

@@ -0,0 +1,214 @@
+package driver
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"regexp"
+	"runtime"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+var (
+	reRktVersion  = regexp.MustCompile("rkt version ([\\d\\.]+).+")
+	reAppcVersion = regexp.MustCompile("appc version ([\\d\\.]+).+")
+)
+
+// RktDriver is a driver for running images via Rkt
+// We attempt to chose sane defaults for now, with more configuration available
+// planned in the future
+type RktDriver struct {
+	DriverContext
+}
+
+// rktHandle is returned from Start/Open as a handle to the PID
+type rktHandle struct {
+	proc   *os.Process
+	name   string
+	logger *log.Logger
+	waitCh chan error
+	doneCh chan struct{}
+}
+
+// rktPID is a struct to map the pid running the process to the vm image on
+// disk
+type rktPID struct {
+	Pid  int
+	Name string
+}
+
+// NewRktDriver is used to create a new exec driver
+func NewRktDriver(ctx *DriverContext) Driver {
+	return &RktDriver{*ctx}
+}
+
+func (d *RktDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
+	// Only enable if we are root when running on non-windows systems.
+	if runtime.GOOS != "windows" && syscall.Geteuid() != 0 {
+		d.logger.Printf("[DEBUG] driver.rkt: must run as root user, disabling")
+		return false, nil
+	}
+
+	outBytes, err := exec.Command("rkt", "version").Output()
+	if err != nil {
+		return false, nil
+	}
+	out := strings.TrimSpace(string(outBytes))
+
+	rktMatches := reRktVersion.FindStringSubmatch(out)
+	appcMatches := reRktVersion.FindStringSubmatch(out)
+	if len(rktMatches) != 2 || len(appcMatches) != 2 {
+		return false, fmt.Errorf("Unable to parse Rkt version string: %#v", rktMatches)
+	}
+
+	node.Attributes["driver.rkt"] = "true"
+	node.Attributes["driver.rkt.version"] = rktMatches[0]
+	node.Attributes["driver.rkt.appc.version"] = appcMatches[1]
+
+	return true, nil
+}
+
+// Run an existing Rkt image.
+func (d *RktDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
+	trust_prefix, ok := task.Config["trust_prefix"]
+	if !ok || trust_prefix == "" {
+		return nil, fmt.Errorf("Missing trust prefix for rkt")
+	}
+
+	// Add the given trust prefix
+	var outBuf, errBuf bytes.Buffer
+	cmd := exec.Command("rkt", "trust", fmt.Sprintf("--prefix=%s", trust_prefix))
+	cmd.Stdout = &outBuf
+	cmd.Stderr = &errBuf
+	d.logger.Printf("[DEBUG] driver.rkt: starting rkt command: %q", cmd.Args)
+	if err := cmd.Run(); err != nil {
+		return nil, fmt.Errorf(
+			"Error running rkt: %s\n\nOutput: %s\n\nError: %s",
+			err, outBuf.String(), errBuf.String())
+	}
+	d.logger.Printf("[DEBUG] driver.rkt: added trust prefix: %q", trust_prefix)
+
+	name, ok := task.Config["name"]
+	if !ok || name == "" {
+		return nil, fmt.Errorf("Missing ACI name for rkt")
+	}
+
+	exec_cmd, ok := task.Config["exec"]
+	if !ok || exec_cmd == "" {
+		d.logger.Printf("[WARN] driver.rkt: could not find a command to execute in the ACI, the default command will be executed")
+	}
+
+	// Run the ACI
+	var aoutBuf, aerrBuf bytes.Buffer
+	run_cmd := []string{
+		"rkt",
+		"run",
+		"--mds-register=false",
+		name,
+	}
+	if exec_cmd != "" {
+		splitted := strings.Fields(exec_cmd)
+		run_cmd = append(run_cmd, "--exec=", splitted[0], "--")
+		run_cmd = append(run_cmd, splitted[1:]...)
+		run_cmd = append(run_cmd, "---")
+	}
+	acmd := exec.Command(run_cmd[0], run_cmd[1:]...)
+	acmd.Stdout = &aoutBuf
+	acmd.Stderr = &aerrBuf
+	d.logger.Printf("[DEBUG] driver:rkt: starting rkt command: %q", acmd.Args)
+	if err := acmd.Start(); err != nil {
+		return nil, fmt.Errorf(
+			"Error running rkt: %s\n\nOutput: %s\n\nError: %s",
+			err, aoutBuf.String(), aerrBuf.String())
+	}
+	d.logger.Printf("[DEBUG] driver.rkt: started ACI: %q", name)
+	h := &rktHandle{
+		proc:   acmd.Process,
+		name:   name,
+		logger: d.logger,
+		doneCh: make(chan struct{}),
+		waitCh: make(chan error, 1),
+	}
+	go h.run()
+	return h, nil
+}
+
+func (d *RktDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
+	// Parse the handle
+	pidBytes := []byte(strings.TrimPrefix(handleID, "Rkt:"))
+	qpid := &rktPID{}
+	if err := json.Unmarshal(pidBytes, qpid); err != nil {
+		return nil, fmt.Errorf("failed to parse Rkt handle '%s': %v", handleID, err)
+	}
+
+	// Find the process
+	proc, err := os.FindProcess(qpid.Pid)
+	if proc == nil || err != nil {
+		return nil, fmt.Errorf("failed to find Rkt PID %d: %v", qpid.Pid, err)
+	}
+
+	// Return a driver handle
+	h := &rktHandle{
+		proc:   proc,
+		name:   qpid.Name,
+		logger: d.logger,
+		doneCh: make(chan struct{}),
+		waitCh: make(chan error, 1),
+	}
+
+	go h.run()
+	return h, nil
+}
+
+func (h *rktHandle) ID() string {
+	// Return a handle to the PID
+	pid := &rktPID{
+		Pid:  h.proc.Pid,
+		Name: h.name,
+	}
+	data, err := json.Marshal(pid)
+	if err != nil {
+		h.logger.Printf("[ERR] driver.rkt: failed to marshal rkt PID to JSON: %s", err)
+	}
+	return fmt.Sprintf("Rkt:%s", string(data))
+}
+
+func (h *rktHandle) WaitCh() chan error {
+	return h.waitCh
+}
+
+func (h *rktHandle) Update(task *structs.Task) error {
+	// Update is not possible
+	return nil
+}
+
+// Kill is used to terminate the task. We send an Interrupt
+// and then provide a 5 second grace period before doing a Kill.
+func (h *rktHandle) Kill() error {
+	h.proc.Signal(os.Interrupt)
+	select {
+	case <-h.doneCh:
+		return nil
+	case <-time.After(5 * time.Second):
+		return h.proc.Kill()
+	}
+}
+
+func (h *rktHandle) run() {
+	ps, err := h.proc.Wait()
+	close(h.doneCh)
+	if err != nil {
+		h.waitCh <- err
+	} else if !ps.Success() {
+		h.waitCh <- fmt.Errorf("task exited with error")
+	}
+	close(h.waitCh)
+}

client/driver/rkt_test.go

@@ -0,0 +1,134 @@
+package driver
+
+import (
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/structs"
+
+	ctestutils "github.com/hashicorp/nomad/client/testutil"
+)
+
+func TestRktDriver_Handle(t *testing.T) {
+	h := &rktHandle{
+		proc:   &os.Process{Pid: 123},
+		name:   "foo",
+		doneCh: make(chan struct{}),
+		waitCh: make(chan error, 1),
+	}
+
+	actual := h.ID()
+	expected := `Rkt:{"Pid":123,"Name":"foo"}`
+	if actual != expected {
+		t.Errorf("Expected `%s`, found `%s`", expected, actual)
+	}
+}
+
+// The fingerprinter test should always pass, even if rkt is not installed.
+func TestRktDriver_Fingerprint(t *testing.T) {
+	ctestutils.RktCompatible(t)
+	d := NewRktDriver(testDriverContext(""))
+	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.rkt"] == "" {
+		t.Fatalf("Missing Rkt driver")
+	}
+	if node.Attributes["driver.rkt.version"] == "" {
+		t.Fatalf("Missing Rkt driver version")
+	}
+	if node.Attributes["driver.rkt.appc.version"] == "" {
+		t.Fatalf("Missing appc version for the Rkt driver")
+	}
+}
+
+func TestRktDriver_Start(t *testing.T) {
+	ctestutils.RktCompatible(t)
+	// TODO: use test server to load from a fixture
+	task := &structs.Task{
+		Name: "etcd",
+		Config: map[string]string{
+			"trust_prefix": "coreos.com/etcd",
+			"name":         "coreos.com/etcd:v2.0.4",
+			"exec":         "/etcd --version",
+		},
+	}
+
+	driverCtx := testDriverContext(task.Name)
+	ctx := testDriverExecContext(task, driverCtx)
+	d := NewRktDriver(driverCtx)
+	defer ctx.AllocDir.Destroy()
+
+	handle, err := d.Start(ctx, task)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if handle == nil {
+		t.Fatalf("missing handle")
+	}
+
+	// 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")
+	}
+
+	// Clean up
+	if err := handle.Kill(); err != nil {
+		fmt.Printf("\nError killing Rkt test: %s", err)
+	}
+}
+
+func TestRktDriver_Start_Wait(t *testing.T) {
+	ctestutils.RktCompatible(t)
+	task := &structs.Task{
+		Name: "etcd",
+		Config: map[string]string{
+			"trust_prefix": "coreos.com/etcd",
+			"name":         "coreos.com/etcd:v2.0.4",
+			"exec":         "/etcd --version",
+		},
+	}
+
+	driverCtx := testDriverContext(task.Name)
+	ctx := testDriverExecContext(task, driverCtx)
+	d := NewRktDriver(driverCtx)
+	defer ctx.AllocDir.Destroy()
+
+	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(5 * time.Second):
+		t.Fatalf("timeout")
+	}
+}

client/testutil/driver_compatible.go

@@ -1,6 +1,7 @@
 package testutil
 
 import (
+	"os/exec"
 	"runtime"
 	"syscall"
 	"testing"
@@ -18,6 +19,17 @@ func QemuCompatible(t *testing.T) {
 	}
 }
 
+func RktCompatible(t *testing.T) {
+	if runtime.GOOS == "windows" || syscall.Geteuid() != 0 {
+		t.Skip("Must be root on non-windows environments to run test")
+	}
+	// else see if rkt exists
+	_, err := exec.Command("rkt", "version").CombinedOutput()
+	if err != nil {
+		t.Skip("Must have rkt installed for rkt specific tests to run")
+	}
+}
+
 func MountCompatible(t *testing.T) {
 	if runtime.GOOS == "windows" {
 		t.Skip("Windows does not support mount")

website/source/docs/drivers/rkt.html.md

@@ -0,0 +1,44 @@
+---
+layout: "docs"
+page_title: "Drivers: Rkt"
+sidebar_current: "docs-drivers-rkt"
+description: |-
+  The Rkt task driver is used to run application containers using Rkt.
+---
+
+# Rkt Driver
+
+Name: `rkt`
+
+The `Rkt` driver provides an interface for using CoreOS Rkt for running
+application containers. Currently, the driver supports launching
+containers.
+
+## Task Configuration
+
+The `Rkt` driver supports the following configuration in the job spec:
+
+* `trust_prefix` - **(Required)** The trust prefix to be passed to rkt. Must be reachable from
+the box running the nomad agent.
+* `name` - **(Required)** Fully qualified name of an image to run using rkt
+* `exec` - **(Optional**) A command to execute on the ACI
+
+## Client Requirements
+
+The `Rkt` driver requires rkt to be installed and in your systems `$PATH`.
+The `trust_prefix` must be accessible by the node running Nomad. This can be an
+internal source, private to your cluster, but it must be reachable by the client
+over HTTP.
+
+## Client Attributes
+
+The `Rkt` driver will set the following client attributes:
+
+* `driver.rkt` - Set to `true` if Rkt is found on the host node. Nomad determines
+this by executing `rkt version` on the host and parsing the output
+* `driver.rkt.version` - Version of `rkt` eg: `0.8.1`
+* `driver.rkt.appc.version` - Version of `appc` that `rkt` is using eg: `0.8.1`
+
+## Resource Isolation
+
+This driver does not support any resource isolation as of now.