Merge pull request #392 from hashicorp/f-raw-exec-use-exec

RawExec driver uses exec_basic

client/driver/executor/exec_basic.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"runtime"
 	"strings"
 
 	"github.com/hashicorp/nomad/client/allocdir"
@@ -124,6 +125,10 @@ func (e *BasicExecutor) Shutdown() error {
 		return fmt.Errorf("Failed to find user processes %v: %v", e.spawn.UserPid, err)
 	}
 
+	if runtime.GOOS == "windows" {
+		return proc.Kill()
+	}
+
 	return proc.Signal(os.Interrupt)
 }
 

client/driver/raw_exec.go

@@ -2,17 +2,13 @@ package driver
 
 import (
 	"fmt"
-	"os"
-	"os/exec"
 	"path/filepath"
-	"runtime"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/hashicorp/nomad/client/allocdir"
 	"github.com/hashicorp/nomad/client/config"
-	"github.com/hashicorp/nomad/client/driver/args"
+	"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"
@@ -21,10 +17,6 @@ import (
 const (
 	// The option that enables this driver in the Config.Options map.
 	rawExecConfigOption = "driver.raw_exec.enable"
-
-	// Null files to use as stdin.
-	unixNull    = "/dev/null"
-	windowsNull = "nul"
 )
 
 // The RawExecDriver is a privileged version of the exec driver. It provides no
@@ -37,7 +29,7 @@ type RawExecDriver struct {
 
 // rawExecHandle is returned from Start/Open as a handle to the PID
 type rawExecHandle struct {
-	proc   *os.Process
+	cmd    executor.Executor
 	waitCh chan error
 	doneCh chan struct{}
 }
@@ -70,7 +62,6 @@ func (d *RawExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandl
 	if !ok {
 		return nil, fmt.Errorf("Could not find task directory for task: %v", d.DriverContext.taskName)
 	}
-	taskLocal := filepath.Join(taskDir, allocdir.TaskLocal)
 
 	// Get the command to be ran
 	command, ok := task.Config["command"]
@@ -96,65 +87,33 @@ func (d *RawExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandl
 	// Get the environment variables.
 	envVars := TaskEnvironmentVariables(ctx, task)
 
-	// expand NOMAD_TASK_DIR
-	parsedPath, err := args.ParseAndReplace(command, envVars.Map())
-	if err != nil {
-		return nil, fmt.Errorf("failure to parse arguments in command path: %v", command)
-	} else if len(parsedPath) != 1 {
-		return nil, fmt.Errorf("couldn't properly parse command path: %v", command)
-	}
-
-	cm := parsedPath[0]
-
 	// Look for arguments
-	var cmdArgs []string
+	var args []string
 	if argRaw, ok := task.Config["args"]; ok {
-		parsed, err := args.ParseAndReplace(argRaw, envVars.Map())
-		if err != nil {
-			return nil, err
-		}
-		cmdArgs = append(cmdArgs, parsed...)
+		args = append(args, argRaw)
 	}
 
 	// Setup the command
-	cmd := exec.Command(cm, cmdArgs...)
-	cmd.Dir = taskDir
-	cmd.Env = envVars.List()
-
-	// Capture the stdout/stderr and redirect stdin to /dev/null
-	stdoutFilename := filepath.Join(taskLocal, fmt.Sprintf("%s.stdout", taskName))
-	stderrFilename := filepath.Join(taskLocal, fmt.Sprintf("%s.stderr", taskName))
-	stdinFilename := unixNull
-	if runtime.GOOS == "windows" {
-		stdinFilename = windowsNull
+	cmd := executor.NewBasicExecutor()
+	executor.SetCommand(cmd, command, args)
+	if err := cmd.Limit(task.Resources); err != nil {
+		return nil, fmt.Errorf("failed to constrain resources: %s", err)
 	}
 
-	stdo, err := os.OpenFile(stdoutFilename, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666)
-	if err != nil {
-		return nil, fmt.Errorf("Error opening file to redirect stdout: %v", err)
-	}
+	// Populate environment variables
+	cmd.Command().Env = envVars.List()
 
-	stde, err := os.OpenFile(stderrFilename, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666)
-	if err != nil {
-		return nil, fmt.Errorf("Error opening file to redirect stderr: %v", err)
+	if err := cmd.ConfigureTaskDir(d.taskName, ctx.AllocDir); err != nil {
+		return nil, fmt.Errorf("failed to configure task directory: %v", err)
 	}
 
-	stdi, err := os.OpenFile(stdinFilename, os.O_CREATE|os.O_RDONLY, 0666)
-	if err != nil {
-		return nil, fmt.Errorf("Error opening file to redirect stdin: %v", err)
-	}
-
-	cmd.Stdout = stdo
-	cmd.Stderr = stde
-	cmd.Stdin = stdi
-
 	if err := cmd.Start(); err != nil {
 		return nil, fmt.Errorf("failed to start command: %v", err)
 	}
 
 	// Return a driver handle
-	h := &rawExecHandle{
-		proc:   cmd.Process,
+	h := &execHandle{
+		cmd:    cmd,
 		doneCh: make(chan struct{}),
 		waitCh: make(chan error, 1),
 	}
@@ -163,22 +122,15 @@ func (d *RawExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandl
 }
 
 func (d *RawExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
-	// Split the handle
-	pidStr := strings.TrimPrefix(handleID, "PID:")
-	pid, err := strconv.Atoi(pidStr)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse handle '%s': %v", handleID, err)
-	}
-
 	// Find the process
-	proc, err := os.FindProcess(pid)
-	if proc == nil || err != nil {
-		return nil, fmt.Errorf("failed to find PID %d: %v", pid, err)
+	cmd := executor.NewBasicExecutor()
+	if err := cmd.Open(handleID); err != nil {
+		return nil, fmt.Errorf("failed to open ID %v: %v", handleID, err)
 	}
 
 	// Return a driver handle
-	h := &rawExecHandle{
-		proc:   proc,
+	h := &execHandle{
+		cmd:    cmd,
 		doneCh: make(chan struct{}),
 		waitCh: make(chan error, 1),
 	}
@@ -187,8 +139,8 @@ func (d *RawExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, e
 }
 
 func (h *rawExecHandle) ID() string {
-	// Return a handle to the PID
-	return fmt.Sprintf("PID:%d", h.proc.Pid)
+	id, _ := h.cmd.ID()
+	return id
 }
 
 func (h *rawExecHandle) WaitCh() chan error {
@@ -200,30 +152,21 @@ func (h *rawExecHandle) 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 on supported
-// OS's, otherwise we kill immediately.
 func (h *rawExecHandle) Kill() error {
-	if runtime.GOOS == "windows" {
-		return h.proc.Kill()
-	}
-
-	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 *rawExecHandle) 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/raw_exec_test.go

@@ -55,6 +55,7 @@ func TestRawExecDriver_StartOpen_Wait(t *testing.T) {
 			"command": "/bin/sleep",
 			"args":    "1",
 		},
+		Resources: basicResources,
 	}
 	driverCtx := testDriverContext(task.Name)
 	ctx := testDriverExecContext(task, driverCtx)
@@ -84,13 +85,6 @@ func TestRawExecDriver_StartOpen_Wait(t *testing.T) {
 	case <-time.After(2 * time.Second):
 		t.Fatalf("timeout")
 	}
-
-	// Check they are both tracking the same PID.
-	pid1 := handle.(*rawExecHandle).proc.Pid
-	pid2 := handle2.(*rawExecHandle).proc.Pid
-	if pid1 != pid2 {
-		t.Fatalf("tracking incorrect Pid; %v != %v", pid1, pid2)
-	}
 }
 
 func TestRawExecDriver_Start_Artifact_basic(t *testing.T) {
@@ -111,6 +105,7 @@ func TestRawExecDriver_Start_Artifact_basic(t *testing.T) {
 			"command":         filepath.Join("$NOMAD_TASK_DIR", file),
 			"checksum":        checksum,
 		},
+		Resources: basicResources,
 	}
 	driverCtx := testDriverContext(task.Name)
 	ctx := testDriverExecContext(task, driverCtx)
@@ -140,13 +135,6 @@ func TestRawExecDriver_Start_Artifact_basic(t *testing.T) {
 	case <-time.After(5 * time.Second):
 		t.Fatalf("timeout")
 	}
-
-	// Check they are both tracking the same PID.
-	pid1 := handle.(*rawExecHandle).proc.Pid
-	pid2 := handle2.(*rawExecHandle).proc.Pid
-	if pid1 != pid2 {
-		t.Fatalf("tracking incorrect Pid; %v != %v", pid1, pid2)
-	}
 }
 
 func TestRawExecDriver_Start_Artifact_expanded(t *testing.T) {
@@ -165,6 +153,7 @@ func TestRawExecDriver_Start_Artifact_expanded(t *testing.T) {
 			"command":         "/bin/bash",
 			"args":            fmt.Sprintf("-c '/bin/sleep 1 && %s'", filepath.Join("$NOMAD_TASK_DIR", file)),
 		},
+		Resources: basicResources,
 	}
 	driverCtx := testDriverContext(task.Name)
 	ctx := testDriverExecContext(task, driverCtx)
@@ -194,13 +183,6 @@ func TestRawExecDriver_Start_Artifact_expanded(t *testing.T) {
 	case <-time.After(5 * time.Second):
 		t.Fatalf("timeout")
 	}
-
-	// Check they are both tracking the same PID.
-	pid1 := handle.(*rawExecHandle).proc.Pid
-	pid2 := handle2.(*rawExecHandle).proc.Pid
-	if pid1 != pid2 {
-		t.Fatalf("tracking incorrect Pid; %v != %v", pid1, pid2)
-	}
 }
 
 func TestRawExecDriver_Start_Wait(t *testing.T) {
@@ -210,6 +192,7 @@ func TestRawExecDriver_Start_Wait(t *testing.T) {
 			"command": "/bin/sleep",
 			"args":    "1",
 		},
+		Resources: basicResources,
 	}
 
 	driverCtx := testDriverContext(task.Name)
@@ -251,6 +234,7 @@ func TestRawExecDriver_Start_Wait_AllocDir(t *testing.T) {
 			"command": "/bin/bash",
 			"args":    fmt.Sprintf(`-c "sleep 1; echo -n %s > $%s/%s"`, string(exp), environment.AllocDir, file),
 		},
+		Resources: basicResources,
 	}
 
 	driverCtx := testDriverContext(task.Name)
@@ -295,6 +279,7 @@ func TestRawExecDriver_Start_Kill_Wait(t *testing.T) {
 			"command": "/bin/sleep",
 			"args":    "1",
 		},
+		Resources: basicResources,
 	}
 
 	driverCtx := testDriverContext(task.Name)