diff --git a/client/driver/exec.go b/client/driver/exec.go index 5cfb0fcac..429280a36 100644 --- a/client/driver/exec.go +++ b/client/driver/exec.go @@ -102,7 +102,7 @@ func (d *ExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, if err != nil { return nil, fmt.Errorf("unable to find the nomad binary: %v", err) } - pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, "plugin.out") + pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, fmt.Sprintf("%s-plugin.out", task.Name)) pluginConfig := &plugin.ClientConfig{ Cmd: exec.Command(bin, "executor", pluginLogFile), } @@ -125,7 +125,7 @@ func (d *ExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, pluginClient.Kill() return nil, fmt.Errorf("error starting process via the plugin: %v", err) } - d.logger.Printf("started process via plugin with pid: %v", ps.Pid) + d.logger.Printf("[INFO] driver.exec: started process via plugin with pid: %v", ps.Pid) // Return a driver handle h := &execHandle{ @@ -153,15 +153,14 @@ func (d *ExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro return nil, fmt.Errorf("Failed to parse handle '%s': %v", handleID, err) } - reattachConfig := id.PluginConfig.PluginConfig() pluginConfig := &plugin.ClientConfig{ - Reattach: reattachConfig, + Reattach: id.PluginConfig.PluginConfig(), } executor, client, err := createExecutor(pluginConfig, d.config.LogOutput) if err != nil { - d.logger.Println("[ERROR] error connecting to plugin so destroying plugin pid and user pid") + d.logger.Println("[ERROR] driver.exec: error connecting to plugin so destroying plugin pid and user pid") if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil { - d.logger.Printf("[ERROR] error destroying plugin and userpid: %v", e) + d.logger.Printf("[ERROR] driver.exec: error destroying plugin and userpid: %v", e) } return nil, fmt.Errorf("error connecting to plugin: %v", err) } diff --git a/client/driver/executor/executor.go b/client/driver/executor/executor.go index e568f3696..136e2208d 100644 --- a/client/driver/executor/executor.go +++ b/client/driver/executor/executor.go @@ -7,10 +7,12 @@ import ( "os/exec" "path/filepath" "runtime" + "strings" "sync" "syscall" "time" + "github.com/hashicorp/go-multierror" cgroupConfig "github.com/opencontainers/runc/libcontainer/configs" "github.com/hashicorp/nomad/client/allocdir" @@ -18,34 +20,35 @@ import ( "github.com/hashicorp/nomad/nomad/structs" ) -// ExecutorContext is a wrapper to hold context to configure the command user -// wants to run +// ExecutorContext holds context to configure the command user +// wants to run and isolate it type ExecutorContext struct { - TaskEnv *env.TaskEnvironment - AllocDir *allocdir.AllocDir - TaskName string - TaskResources *structs.Resources - FSIsolation bool - ResourceLimits bool - UnprivilegedUser bool + TaskEnv *env.TaskEnvironment //TaskEnv holds information about the environment of a Task + AllocDir *allocdir.AllocDir //AllocDir is the handle to do operations on the alloc dir of the Task + TaskName string // TaskName is the name of the Task + TaskResources *structs.Resources // TaskResources are the resource constraints for the Task + FSIsolation bool // FSIsolation is a flag for drivers to impose file system isolation on certain platforms + ResourceLimits bool // ResourceLimits is a flag for drivers to impose resource contraints on a Task on certain platforms + UnprivilegedUser bool // UnprivilegedUser is a flag for drivers to make the process run as nobody } -// ExecCommand is a wrapper to hold the user command +// ExecCommand holds the user command and args. It's a lightweight replacement +// of exec.Cmd for serialization purposes. type ExecCommand struct { Cmd string Args []string } -// ProcessState holds information about the state of -// a user process +// ProcessState holds information about the state of a user process. type ProcessState struct { Pid int ExitCode int + Signal int Time time.Time } // Executor is the interface which allows a driver to launch and supervise -// a process user wants to run +// a process type Executor interface { LaunchCmd(command *ExecCommand, ctx *ExecutorContext) (*ProcessState, error) Wait() (*ProcessState, error) @@ -77,7 +80,7 @@ func NewExecutor(logger *log.Logger) Executor { // LaunchCmd launches a process and returns it's state. It also configures an // applies isolation on certain platforms. func (e *UniversalExecutor) LaunchCmd(command *ExecCommand, ctx *ExecutorContext) (*ProcessState, error) { - e.logger.Printf("[INFO] executor: launching command %v", command.Cmd) + e.logger.Printf("[DEBUG] executor: launching command %v %v", command.Cmd, strings.Join(command.Args, "")) e.ctx = ctx @@ -86,11 +89,14 @@ func (e *UniversalExecutor) LaunchCmd(command *ExecCommand, ctx *ExecutorContext return nil, err } - // confiuguring the chroot + // configuring the chroot if err := e.configureIsolation(); err != nil { return nil, err } + // entering the plugin process in cgroup + e.applyLimits(os.Getpid()) + // setting the user of the process if e.ctx.UnprivilegedUser { if err := e.runAs("nobody"); err != nil { @@ -114,6 +120,7 @@ func (e *UniversalExecutor) LaunchCmd(command *ExecCommand, ctx *ExecutorContext e.cmd.Stderr = stde // setting the env, path and args for the command + e.ctx.TaskEnv.Build() e.cmd.Env = ctx.TaskEnv.EnvList() e.cmd.Path = ctx.TaskEnv.ReplaceEnv(command.Cmd) e.cmd.Args = append([]string{e.cmd.Path}, ctx.TaskEnv.ParseAndReplace(command.Args)...) @@ -129,10 +136,6 @@ func (e *UniversalExecutor) LaunchCmd(command *ExecCommand, ctx *ExecutorContext return nil, fmt.Errorf("error starting command: %v", err) } - // entering the user process in the cgroup - e.applyLimits(e.cmd.Process.Pid) - // entering the plugin process in cgroup - e.applyLimits(os.Getpid()) go e.wait() return &ProcessState{Pid: e.cmd.Process.Pid, ExitCode: -1, Time: time.Now()}, nil } @@ -168,24 +171,28 @@ func (e *UniversalExecutor) wait() { // Exit cleans up the alloc directory, destroys cgroups and kills the user // process func (e *UniversalExecutor) Exit() error { - e.logger.Printf("[INFO] Exiting plugin for task %q", e.ctx.TaskName) - if e.cmd.Process == nil { - return fmt.Errorf("executor.exit error: no process found") - } - proc, err := os.FindProcess(e.cmd.Process.Pid) - if err != nil { - return fmt.Errorf("failied to find user process %v: %v", e.cmd.Process.Pid, err) - } - if err = proc.Kill(); err != nil { - e.logger.Printf("[DEBUG] executor.exit error: %v", err) + var merr multierror.Error + if e.cmd.Process != nil { + proc, err := os.FindProcess(e.cmd.Process.Pid) + if err != nil { + e.logger.Printf("[ERROR] can't find process with pid: %v, err: %v", e.cmd.Process.Pid, err) + } + if err := proc.Kill(); err != nil { + e.logger.Printf("[ERROR] can't kill process with pid: %v, err: %v", e.cmd.Process.Pid, err) + } } + if e.ctx.FSIsolation { - e.removeChrootMounts() + if err := e.removeChrootMounts(); err != nil { + merr.Errors = append(merr.Errors, err) + } } if e.ctx.ResourceLimits { - e.destroyCgroup() + if err := e.destroyCgroup(); err != nil { + merr.Errors = append(merr.Errors, err) + } } - return nil + return merr.ErrorOrNil() } // Shutdown sends an interrupt signal to the user process diff --git a/client/driver/executor/executor_linux.go b/client/driver/executor/executor_linux.go index 2e6b282c7..9c653eade 100644 --- a/client/driver/executor/executor_linux.go +++ b/client/driver/executor/executor_linux.go @@ -213,8 +213,7 @@ func (e *UniversalExecutor) removeChrootMounts() error { } } - // Unmount - // proc. + // Unmount proc. proc := filepath.Join(e.taskDir, "proc") if e.pathExists(proc) { if err := syscall.Unmount(proc, 0); err != nil { diff --git a/client/driver/executor/executor_test.go b/client/driver/executor/executor_test.go index 2b7e281ac..d70d652b1 100644 --- a/client/driver/executor/executor_test.go +++ b/client/driver/executor/executor_test.go @@ -57,17 +57,18 @@ func TestExecutor_Start_Invalid(t *testing.T) { invalid := "/bin/foobar" execCmd := ExecCommand{Cmd: invalid, Args: []string{"1"}} ctx := testExecutorContext(t) + defer ctx.AllocDir.Destroy() executor := NewExecutor(log.New(os.Stdout, "", log.LstdFlags)) _, err := executor.LaunchCmd(&execCmd, ctx) if err == nil { t.Fatalf("Expected error") } - defer ctx.AllocDir.Destroy() } func TestExecutor_Start_Wait_Failure_Code(t *testing.T) { execCmd := ExecCommand{Cmd: "/bin/sleep", Args: []string{"fail"}} ctx := testExecutorContext(t) + defer ctx.AllocDir.Destroy() executor := NewExecutor(log.New(os.Stdout, "", log.LstdFlags)) ps, _ := executor.LaunchCmd(&execCmd, ctx) if ps.Pid == 0 { @@ -77,12 +78,12 @@ func TestExecutor_Start_Wait_Failure_Code(t *testing.T) { if ps.ExitCode < 1 { t.Fatalf("expected exit code to be non zero, actual: %v", ps.ExitCode) } - defer ctx.AllocDir.Destroy() } func TestExecutor_Start_Wait(t *testing.T) { execCmd := ExecCommand{Cmd: "/bin/echo", Args: []string{"hello world"}} ctx := testExecutorContext(t) + defer ctx.AllocDir.Destroy() executor := NewExecutor(log.New(os.Stdout, "", log.LstdFlags)) ps, err := executor.LaunchCmd(&execCmd, ctx) if err != nil { @@ -95,7 +96,6 @@ func TestExecutor_Start_Wait(t *testing.T) { if err != nil { t.Fatalf("error in waiting for command: %v", err) } - defer ctx.AllocDir.Destroy() task := "web" taskDir, ok := ctx.AllocDir.TaskDirs[task] @@ -120,6 +120,7 @@ func TestExecutor_Start_Wait(t *testing.T) { func TestExecutor_Start_Kill(t *testing.T) { execCmd := ExecCommand{Cmd: "/bin/sleep", Args: []string{"10 && hello world"}} ctx := testExecutorContext(t) + defer ctx.AllocDir.Destroy() executor := NewExecutor(log.New(os.Stdout, "", log.LstdFlags)) ps, err := executor.LaunchCmd(&execCmd, ctx) if err != nil { @@ -132,7 +133,6 @@ func TestExecutor_Start_Kill(t *testing.T) { if err != nil { t.Fatalf("error in waiting for command: %v", err) } - defer ctx.AllocDir.Destroy() task := "web" taskDir, ok := ctx.AllocDir.TaskDirs[task] diff --git a/client/driver/java.go b/client/driver/java.go index 494916324..1fdac01db 100644 --- a/client/driver/java.go +++ b/client/driver/java.go @@ -148,7 +148,7 @@ func (d *JavaDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, return nil, fmt.Errorf("unable to find the nomad binary: %v", err) } - pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, "plugin.out") + pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, fmt.Sprintf("%s-plugin.out", task.Name)) pluginConfig := &plugin.ClientConfig{ Cmd: exec.Command(bin, "executor", pluginLogFile), } @@ -168,7 +168,7 @@ func (d *JavaDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, pluginClient.Kill() return nil, fmt.Errorf("error starting process via the plugin: %v", err) } - d.logger.Printf("[INFO] started process with pid: %v", ps.Pid) + d.logger.Printf("[INFO] driver.java: started process with pid: %v", ps.Pid) // Return a driver handle h := &javaHandle{ @@ -197,15 +197,14 @@ func (d *JavaDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro return nil, fmt.Errorf("Failed to parse handle '%s': %v", handleID, err) } - reattachConfig := id.PluginConfig.PluginConfig() pluginConfig := &plugin.ClientConfig{ - Reattach: reattachConfig, + Reattach: id.PluginConfig.PluginConfig(), } executor, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput) if err != nil { - d.logger.Println("[ERROR] error connecting to plugin so destroying plugin pid and user pid") + d.logger.Println("[ERROR] driver.java: error connecting to plugin so destroying plugin pid and user pid") if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil { - d.logger.Printf("[ERROR] error destroying plugin and userpid: %v", e) + d.logger.Printf("[ERROR] driver.java: error destroying plugin and userpid: %v", e) } return nil, fmt.Errorf("error connecting to plugin: %v", err) } diff --git a/client/driver/qemu.go b/client/driver/qemu.go index f1bed809a..640c416a9 100644 --- a/client/driver/qemu.go +++ b/client/driver/qemu.go @@ -192,7 +192,7 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, return nil, fmt.Errorf("unable to find the nomad binary: %v", err) } - pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, "plugin.out") + pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, fmt.Sprintf("%s-plugin.out", task.Name)) pluginConfig := &plugin.ClientConfig{ Cmd: exec.Command(bin, "executor", pluginLogFile), } @@ -212,7 +212,6 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, pluginClient.Kill() return nil, fmt.Errorf("error starting process via the plugin: %v", err) } - d.logger.Printf("[INFO] started process with pid: %v", ps.Pid) d.logger.Printf("[INFO] Started new QemuVM: %s", vmID) // Create and Return Handle @@ -242,16 +241,15 @@ func (d *QemuDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro return nil, fmt.Errorf("Failed to parse handle '%s': %v", handleID, err) } - reattachConfig := id.PluginConfig.PluginConfig() pluginConfig := &plugin.ClientConfig{ - Reattach: reattachConfig, + Reattach: id.PluginConfig.PluginConfig(), } executor, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput) if err != nil { - d.logger.Println("[ERROR] error connecting to plugin so destroying plugin pid and user pid") + d.logger.Println("[ERROR] driver.qemu: error connecting to plugin so destroying plugin pid and user pid") if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil { - d.logger.Printf("[ERROR] error destroying plugin and userpid: %v", e) + d.logger.Printf("[ERROR] driver.qemu: error destroying plugin and userpid: %v", e) } return nil, fmt.Errorf("error connecting to plugin: %v", err) } diff --git a/client/driver/raw_exec.go b/client/driver/raw_exec.go index 784e7292e..99fbadbab 100644 --- a/client/driver/raw_exec.go +++ b/client/driver/raw_exec.go @@ -98,7 +98,7 @@ func (d *RawExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandl if err != nil { return nil, fmt.Errorf("unable to find the nomad binary: %v", err) } - pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, "plugin.out") + pluginLogFile := filepath.Join(ctx.AllocDir.AllocDir, fmt.Sprintf("%s-plugin.out", task.Name)) pluginConfig := &plugin.ClientConfig{ Cmd: exec.Command(bin, "executor", pluginLogFile), } @@ -118,7 +118,7 @@ func (d *RawExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandl pluginClient.Kill() return nil, fmt.Errorf("error starting process via the plugin: %v", err) } - d.logger.Printf("[INFO] started process with pid: %v", ps.Pid) + d.logger.Printf("[INFO] driver.raw_exec: started process with pid: %v", ps.Pid) // Return a driver handle h := &rawExecHandle{ @@ -151,9 +151,9 @@ func (d *RawExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, e } executor, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput) if err != nil { - d.logger.Println("[ERROR] error connecting to plugin so destroying plugin pid and user pid") + d.logger.Println("[ERROR] driver.raw_exec: error connecting to plugin so destroying plugin pid and user pid") if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil { - d.logger.Printf("[ERROR] error destroying plugin and userpid: %v", e) + d.logger.Printf("[ERROR] driver.raw_exec: error destroying plugin and userpid: %v", e) } return nil, fmt.Errorf("error connecting to plugin: %v", err) } diff --git a/client/driver/utils.go b/client/driver/utils.go index b42c4a74a..e8935c80e 100644 --- a/client/driver/utils.go +++ b/client/driver/utils.go @@ -10,6 +10,8 @@ import ( "github.com/hashicorp/nomad/client/driver/executor" ) +// createExecutor launches an executor plugin and returns an instance of the +// Executor interface func createExecutor(config *plugin.ClientConfig, w io.Writer) (executor.Executor, *plugin.Client, error) { config.HandshakeConfig = HandshakeConfig config.Plugins = GetPluginMap(w) @@ -27,6 +29,7 @@ func createExecutor(config *plugin.ClientConfig, w io.Writer) (executor.Executor return executorPlugin, executorClient, nil } +// killProcess kills a process with the given pid func killProcess(pid int) error { proc, err := os.FindProcess(pid) if err != nil { @@ -35,6 +38,8 @@ func killProcess(pid int) error { return proc.Kill() } +// destroyPlugin kills the plugin with the given pid and also kills the user +// process func destroyPlugin(pluginPid int, userPid int) error { var merr error if err := killProcess(pluginPid); err != nil {