From e49f3383b692480090e8c4cd2e409685570f28e9 Mon Sep 17 00:00:00 2001 From: Alex Dadgar Date: Fri, 6 Nov 2015 10:42:49 -0800 Subject: [PATCH] merge --- client/driver/executor/exec_basic.go | 5 ++ client/driver/raw_exec.go | 105 ++++++--------------------- client/driver/raw_exec_test.go | 27 ++----- 3 files changed, 35 insertions(+), 102 deletions(-) diff --git a/client/driver/executor/exec_basic.go b/client/driver/executor/exec_basic.go index a554acfc1..fd0f09049 100644 --- a/client/driver/executor/exec_basic.go +++ b/client/driver/executor/exec_basic.go @@ -7,6 +7,7 @@ import ( "os" "os/exec" "path/filepath" + "runtime" "strings" "github.com/hashicorp/nomad/client/allocdir" @@ -131,6 +132,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) } diff --git a/client/driver/raw_exec.go b/client/driver/raw_exec.go index c53e73a2b..12e99b7f4 100644 --- a/client/driver/raw_exec.go +++ b/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) } diff --git a/client/driver/raw_exec_test.go b/client/driver/raw_exec_test.go index 1dda991c8..053f29337 100644 --- a/client/driver/raw_exec_test.go +++ b/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)