Merge pull request #394 from hashicorp/f-qemu-use-exec

Qemu Driver uses Executor

client/driver/args/args.go

@@ -27,7 +27,7 @@ func ParseAndReplace(args string, env map[string]string) ([]string, error) {
 
 	replaced := make([]string, len(parsed))
 	for i, arg := range parsed {
-		replaced[i] = replaceEnv(arg, env)
+		replaced[i] = ReplaceEnv(arg, env)
 	}
 
 	return replaced, nil
@@ -36,7 +36,7 @@ func ParseAndReplace(args string, env map[string]string) ([]string, error) {
 // replaceEnv takes an arg and replaces all occurences of environment variables.
 // If the variable is found in the passed map it is replaced, otherwise the
 // original string is returned.
-func replaceEnv(arg string, env map[string]string) string {
+func ReplaceEnv(arg string, env map[string]string) string {
 	return envRe.ReplaceAllStringFunc(arg, func(arg string) string {
 		stripped := arg[1:]
 		if stripped[0] == '{' {

client/driver/executor/exec_basic.go

@@ -59,14 +59,7 @@ func (e *BasicExecutor) Start() error {
 		return err
 	}
 
-	parsedPath, err := args.ParseAndReplace(e.cmd.Path, envVars.Map())
-	if err != nil {
-		return err
-	} else if len(parsedPath) != 1 {
-		return fmt.Errorf("couldn't properly parse command path: %v", e.cmd.Path)
-	}
-
-	e.cmd.Path = parsedPath[0]
+	e.cmd.Path = args.ReplaceEnv(e.cmd.Path, envVars.Map())
 	combined := strings.Join(e.cmd.Args, " ")
 	parsed, err := args.ParseAndReplace(combined, envVars.Map())
 	if err != nil {

client/driver/executor/exec_linux.go

@@ -165,14 +165,7 @@ func (e *LinuxExecutor) Start() error {
 		return err
 	}
 
-	parsedPath, err := args.ParseAndReplace(e.cmd.Path, envVars.Map())
-	if err != nil {
-		return err
-	} else if len(parsedPath) != 1 {
-		return fmt.Errorf("couldn't properly parse command path: %v", e.cmd.Path)
-	}
-	e.cmd.Path = parsedPath[0]
-
+	e.cmd.Path = args.ReplaceEnv(e.cmd.Path, envVars.Map())
 	combined := strings.Join(e.cmd.Args, " ")
 	parsed, err := args.ParseAndReplace(combined, envVars.Map())
 	if err != nil {

client/driver/qemu.go

@@ -1,11 +1,7 @@
 package driver
 
 import (
-	"bytes"
-	"encoding/json"
 	"fmt"
-	"log"
-	"os"
 	"os/exec"
 	"path/filepath"
 	"regexp"
@@ -16,6 +12,7 @@ import (
 
 	"github.com/hashicorp/nomad/client/allocdir"
 	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/client/driver/executor"
 	"github.com/hashicorp/nomad/client/fingerprint"
 	"github.com/hashicorp/nomad/client/getter"
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -35,19 +32,11 @@ type QemuDriver struct {
 
 // qemuHandle is returned from Start/Open as a handle to the PID
 type qemuHandle struct {
-	proc   *os.Process
-	vmID   string
+	cmd    executor.Executor
 	waitCh chan error
 	doneCh chan struct{}
 }
 
-// qemuPID is a struct to map the pid running the process to the vm image on
-// disk
-type qemuPID struct {
-	Pid  int
-	VmID string
-}
-
 // NewQemuDriver is used to create a new exec driver
 func NewQemuDriver(ctx *DriverContext) Driver {
 	return &QemuDriver{DriverContext: *ctx}
@@ -184,25 +173,25 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
 		)
 	}
 
-	// Start Qemu
-	var outBuf, errBuf bytes.Buffer
-	cmd := exec.Command(args[0], args[1:]...)
-	cmd.Stdout = &outBuf
-	cmd.Stderr = &errBuf
+	// Setup the command
+	cmd := executor.Command(args[0], args[1:]...)
+	if err := cmd.Limit(task.Resources); err != nil {
+		return nil, fmt.Errorf("failed to constrain resources: %s", err)
+	}
+
+	if err := cmd.ConfigureTaskDir(d.taskName, ctx.AllocDir); err != nil {
+		return nil, fmt.Errorf("failed to configure task directory: %v", err)
+	}
 
 	d.logger.Printf("[DEBUG] Starting QemuVM command: %q", strings.Join(args, " "))
 	if err := cmd.Start(); err != nil {
-		return nil, fmt.Errorf(
-			"Error running QEMU: %s\n\nOutput: %s\n\nError: %s",
-			err, outBuf.String(), errBuf.String())
+		return nil, fmt.Errorf("failed to start command: %v", err)
 	}
-
 	d.logger.Printf("[INFO] Started new QemuVM: %s", vmID)
 
 	// Create and Return Handle
 	h := &qemuHandle{
-		proc:   cmd.Process,
-		vmID:   vmPath,
+		cmd:    cmd,
 		doneCh: make(chan struct{}),
 		waitCh: make(chan error, 1),
 	}
@@ -212,42 +201,25 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
 }
 
 func (d *QemuDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
-	// Parse the handle
-	pidBytes := []byte(strings.TrimPrefix(handleID, "QEMU:"))
-	qpid := &qemuPID{}
-	if err := json.Unmarshal(pidBytes, qpid); err != nil {
-		return nil, fmt.Errorf("failed to parse Qemu 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 Qemu PID %d: %v", qpid.Pid, err)
+	cmd, err := executor.OpenId(handleID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open ID %v: %v", handleID, err)
 	}
 
 	// Return a driver handle
-	h := &qemuHandle{
-		proc:   proc,
-		vmID:   qpid.VmID,
+	h := &execHandle{
+		cmd:    cmd,
 		doneCh: make(chan struct{}),
 		waitCh: make(chan error, 1),
 	}
-
 	go h.run()
 	return h, nil
 }
 
 func (h *qemuHandle) ID() string {
-	// Return a handle to the PID
-	pid := &qemuPID{
-		Pid:  h.proc.Pid,
-		VmID: h.vmID,
-	}
-	data, err := json.Marshal(pid)
-	if err != nil {
-		log.Printf("[ERR] failed to marshal Qemu PID to JSON: %s", err)
-	}
-	return fmt.Sprintf("QEMU:%s", string(data))
+	id, _ := h.cmd.ID()
+	return id
 }
 
 func (h *qemuHandle) WaitCh() chan error {
@@ -259,28 +231,23 @@ func (h *qemuHandle) Update(task *structs.Task) error {
 	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.
-//
 // TODO: allow a 'shutdown_command' that can be executed over a ssh connection
 // to the VM
 func (h *qemuHandle) Kill() error {
-	h.proc.Signal(os.Interrupt)
+	h.cmd.Shutdown()
 	select {
 	case <-h.doneCh:
 		return nil
 	case <-time.After(5 * time.Second):
-		return h.proc.Kill()
+		return h.cmd.ForceStop()
 	}
 }
 
 func (h *qemuHandle) run() {
-	ps, err := h.proc.Wait()
+	err := h.cmd.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/qemu_test.go

@@ -2,7 +2,6 @@ package driver
 
 import (
 	"fmt"
-	"os"
 	"testing"
 
 	"github.com/hashicorp/nomad/client/config"
@@ -11,21 +10,6 @@ import (
 	ctestutils "github.com/hashicorp/nomad/client/testutil"
 )
 
-func TestQemuDriver_Handle(t *testing.T) {
-	h := &qemuHandle{
-		proc:   &os.Process{Pid: 123},
-		vmID:   "vmid",
-		doneCh: make(chan struct{}),
-		waitCh: make(chan error, 1),
-	}
-
-	actual := h.ID()
-	expected := `QEMU:{"Pid":123,"VmID":"vmid"}`
-	if actual != expected {
-		t.Errorf("Expected `%s`, found `%s`", expected, actual)
-	}
-}
-
 // The fingerprinter test should always pass, even if QEMU is not installed.
 func TestQemuDriver_Fingerprint(t *testing.T) {
 	ctestutils.QemuCompatible(t)
@@ -48,7 +32,7 @@ func TestQemuDriver_Fingerprint(t *testing.T) {
 	}
 }
 
-func TestQemuDriver_Start(t *testing.T) {
+func TestQemuDriver_StartOpen_Wait(t *testing.T) {
 	ctestutils.QemuCompatible(t)
 	// TODO: use test server to load from a fixture
 	task := &structs.Task{
@@ -60,6 +44,7 @@ func TestQemuDriver_Start(t *testing.T) {
 			"guest_ports":     "22,8080",
 		},
 		Resources: &structs.Resources{
+			CPU:      500,
 			MemoryMB: 512,
 			Networks: []*structs.NetworkResource{
 				&structs.NetworkResource{