luxolus/open-nomad
2
0
Fork 0
Code Issues 1 Pull requests Packages Releases Wiki Activity

Using a plugin to create the executor

Browse source
This commit is contained in:
Diptanu Choudhury 2016-02-02 13:38:38 -08:00
parent a5ba2153f3
commit 4ed85d0ab2
4 changed files with 298 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

198
client/driver/exec.go
View file

@ -4,17 +4,20 @@ import (
"encoding/json"
"fmt"
"log"
"path/filepath"
"os/exec"
//"path/filepath"
"syscall"
"time"
"github.com/hashicorp/nomad/client/allocdir"
"github.com/hashicorp/go-plugin"
//"github.com/hashicorp/nomad/client/allocdir"
"github.com/hashicorp/nomad/client/config"
"github.com/hashicorp/nomad/client/driver/executor"
"github.com/hashicorp/nomad/client/driver/plugins"
cstructs "github.com/hashicorp/nomad/client/driver/structs"
"github.com/hashicorp/nomad/client/getter"
//"github.com/hashicorp/nomad/client/getter"
"github.com/hashicorp/nomad/nomad/structs"
"github.com/mitchellh/mapstructure"
//"github.com/mitchellh/mapstructure"
)
// ExecDriver fork/execs tasks using as many of the underlying OS's isolation
@ -32,11 +35,13 @@ type ExecDriverConfig struct {
// execHandle is returned from Start/Open as a handle to the PID
type execHandle struct {
cmd executor.Executor
killTimeout time.Duration
logger *log.Logger
waitCh chan *cstructs.WaitResult
doneCh chan struct{}
pluginClient *plugin.Client
executor plugins.Executor
cmd executor.Executor
killTimeout time.Duration
logger *log.Logger
waitCh chan *cstructs.WaitResult
doneCh chan struct{}
}
// NewExecDriver is used to create a new exec driver
@ -63,58 +68,82 @@ func (d *ExecDriver) Periodic() (bool, time.Duration) {
}
func (d *ExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
var driverConfig ExecDriverConfig
if err := mapstructure.WeakDecode(task.Config, &driverConfig); err != nil {
return nil, err
}
// Get the command to be ran
command := driverConfig.Command
if command == "" {
return nil, fmt.Errorf("missing command for exec driver")
// var driverConfig ExecDriverConfig
// if err := mapstructure.WeakDecode(task.Config, &driverConfig); err != nil {
// return nil, err
// }
// // Get the command to be ran
// command := driverConfig.Command
// if command == "" {
// return nil, fmt.Errorf("missing command for exec driver")
// }
//
// // Create a location to download the artifact.
// taskDir, ok := ctx.AllocDir.TaskDirs[d.DriverContext.taskName]
// if !ok {
// return nil, fmt.Errorf("Could not find task directory for task: %v", d.DriverContext.taskName)
// }
//
// // Check if an artificat is specified and attempt to download it
// source, ok := task.Config["artifact_source"]
// if ok && source != "" {
// // Proceed to download an artifact to be executed.
// _, err := getter.GetArtifact(
// filepath.Join(taskDir, allocdir.TaskLocal),
// driverConfig.ArtifactSource,
// driverConfig.Checksum,
// d.logger,
// )
// if err != nil {
// return nil, err
// }
// }
//
// // Setup the command
// execCtx := executor.NewExecutorContext(d.taskEnv)
// cmd := executor.Command(execCtx, command, driverConfig.Args...)
// if err := cmd.Limit(task.Resources); err != nil {
// return nil, fmt.Errorf("failed to constrain resources: %s", err)
// }
//
// // Populate environment variables
// cmd.Command().Env = d.taskEnv.EnvList()
//
// if err := cmd.ConfigureTaskDir(d.taskName, ctx.AllocDir); err != nil {
// return nil, fmt.Errorf("failed to configure task directory: %v", err)
// }
//
// if err := cmd.Start(); err != nil {
// return nil, fmt.Errorf("failed to start command: %v", err)
// }
//
executorClient := plugin.NewClient(&plugin.ClientConfig{
HandshakeConfig: plugins.HandshakeConfig,
Plugins: plugins.PluginMap,
Cmd: exec.Command("/home/diptanuc/Projects/gocode/bin/nomad"),
})
rpcClient, err := executorClient.Client()
if err != nil {
return nil, fmt.Errorf("error creating rpc client for executor plugin: %v", err)
}
// Create a location to download the artifact.
taskDir, ok := ctx.AllocDir.TaskDirs[d.DriverContext.taskName]
if !ok {
return nil, fmt.Errorf("Could not find task directory for task: %v", d.DriverContext.taskName)
raw, err := rpcClient.Dispense("executor")
if err != nil {
return nil, fmt.Errorf("unable to dispense the executor plugin: %v", err)
}
// Check if an artificat is specified and attempt to download it
source, ok := task.Config["artifact_source"]
if ok && source != "" {
// Proceed to download an artifact to be executed.
_, err := getter.GetArtifact(
filepath.Join(taskDir, allocdir.TaskLocal),
driverConfig.ArtifactSource,
driverConfig.Checksum,
d.logger,
)
if err != nil {
return nil, err
}
}
// Setup the command
execCtx := executor.NewExecutorContext(d.taskEnv)
cmd := executor.Command(execCtx, command, driverConfig.Args...)
if err := cmd.Limit(task.Resources); err != nil {
return nil, fmt.Errorf("failed to constrain resources: %s", err)
}
// Populate environment variables
cmd.Command().Env = d.taskEnv.EnvList()
if err := cmd.ConfigureTaskDir(d.taskName, ctx.AllocDir); err != nil {
return nil, fmt.Errorf("failed to configure task directory: %v", err)
}
if err := cmd.Start(); err != nil {
return nil, fmt.Errorf("failed to start command: %v", err)
executorPlugin := raw.(plugins.Executor)
ps, err := executorPlugin.LaunchCmd(exec.Command("/bin/echo", "hello"), &plugins.ExecutorContext{})
if err != nil {
return nil, fmt.Errorf("error starting process via the plugin: %v", err)
}
d.logger.Printf("DIPTANU Started process via plugin: %#v", ps)
// Return a driver handle
h := &execHandle{
cmd: cmd,
pluginClient: executorClient,
executor: executorPlugin,
//cmd: cmd,
killTimeout: d.DriverContext.KillTimeout(task),
logger: d.logger,
doneCh: make(chan struct{}),
@ -125,8 +154,9 @@ func (d *ExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
}
type execId struct {
ExecutorId string
KillTimeout time.Duration
//ExecutorId string
KillTimeout time.Duration
PluginConfig *plugin.ReattachConfig
}
func (d *ExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
@ -136,29 +166,54 @@ func (d *ExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro
}
// Find the process
execCtx := executor.NewExecutorContext(d.taskEnv)
cmd, err := executor.OpenId(execCtx, id.ExecutorId)
// execCtx := executor.NewExecutorContext(d.taskEnv)
// cmd, err := executor.OpenId(execCtx, id.ExecutorId)
// if err != nil {
// return nil, fmt.Errorf("failed to open ID %v: %v", id.ExecutorId, err)
// }
pluginConfig := &plugin.ClientConfig{
HandshakeConfig: plugins.HandshakeConfig,
Plugins: plugins.PluginMap,
Cmd: exec.Command("/home/diptanuc/Projects/gocode/bin/nomad"),
Reattach: id.PluginConfig,
}
executor, client, err := d.executor(pluginConfig)
if err != nil {
return nil, fmt.Errorf("failed to open ID %v: %v", id.ExecutorId, err)
return nil, fmt.Errorf("error connecting to plugin: %v", err)
}
// Return a driver handle
h := &execHandle{
cmd: cmd,
logger: d.logger,
killTimeout: id.KillTimeout,
doneCh: make(chan struct{}),
waitCh: make(chan *cstructs.WaitResult, 1),
pluginClient: client,
executor: executor,
logger: d.logger,
killTimeout: id.KillTimeout,
doneCh: make(chan struct{}),
waitCh: make(chan *cstructs.WaitResult, 1),
}
go h.run()
return h, nil
}
func (d *ExecDriver) executor(config *plugin.ClientConfig) (plugins.Executor, *plugin.Client, error) {
executorClient := plugin.NewClient(config)
rpcClient, err := executorClient.Client()
if err != nil {
return nil, nil, fmt.Errorf("error creating rpc client for executor plugin: %v", err)
}
raw, err := rpcClient.Dispense("executor")
if err != nil {
return nil, nil, fmt.Errorf("unable to dispense the executor plugin: %v", err)
}
executorPlugin := raw.(plugins.Executor)
return executorPlugin, executorClient, nil
}
func (h *execHandle) ID() string {
executorId, _ := h.cmd.ID()
id := execId{
ExecutorId: executorId,
KillTimeout: h.killTimeout,
KillTimeout: h.killTimeout,
PluginConfig: h.pluginClient.ReattachConfig(),
}
data, err := json.Marshal(id)
@ -181,18 +236,19 @@ func (h *execHandle) Update(task *structs.Task) error {
}
func (h *execHandle) Kill() error {
h.cmd.Shutdown()
h.executor.ShutDown()
select {
case <-h.doneCh:
return nil
case <-time.After(h.killTimeout):
return h.cmd.ForceStop()
_, err := h.executor.Exit()
return err
}
}
func (h *execHandle) run() {
res := h.cmd.Wait()
ps, err := h.executor.Wait()
close(h.doneCh)
h.waitCh <- res
h.waitCh <- &cstructs.WaitResult{ExitCode: ps.ExitCode, Signal: 0, Err: err}
close(h.waitCh)
}

104
client/driver/plugins/executor.go Normal file
View file

@ -0,0 +1,104 @@
package plugins
import (
"net/rpc"
"os/exec"
"time"
"github.com/hashicorp/go-plugin"
)
var HandshakeConfig = plugin.HandshakeConfig{
ProtocolVersion: 1,
}
var PluginMap = map[string]plugin.Plugin{
"executor": new(ExecutorPlugin),
}
type ExecutorContext struct {
}
type ProcessState struct {
Pid int
ExitCode int
Time time.Time
}
type Executor interface {
LaunchCmd(cmd *exec.Cmd, ctx *ExecutorContext) (*ProcessState, error)
Wait() (*ProcessState, error)
ShutDown() (*ProcessState, error)
Exit() (*ProcessState, error)
}
type ExecutorRPC struct {
client *rpc.Client
}
type LaunchCmdArgs struct {
Cmd *exec.Cmd
Ctx *ExecutorContext
}
func (e *ExecutorRPC) LaunchCmd(cmd *exec.Cmd, ctx *ExecutorContext) (*ProcessState, error) {
var ps ProcessState
err := e.client.Call("Plugin.LaunchCmd", LaunchCmdArgs{Cmd: cmd, Ctx: ctx}, &ps)
return &ps, err
}
func (e *ExecutorRPC) Wait() (*ProcessState, error) {
var ps ProcessState
err := e.client.Call("Plugin.Wait", new(interface{}), &ps)
return &ps, err
}
func (e *ExecutorRPC) ShutDown() (*ProcessState, error) {
var ps ProcessState
err := e.client.Call("Plugin.ShutDown", new(interface{}), &ps)
return &ps, err
}
func (e *ExecutorRPC) Exit() (*ProcessState, error) {
var ps ProcessState
err := e.client.Call("Plugin.Exit", new(interface{}), &ps)
return &ps, err
}
type ExecutorRPCServer struct {
Impl Executor
}
func (e *ExecutorRPCServer) LaunchCmd(args LaunchCmdArgs, ps *ProcessState) error {
var err error
ps, err = e.Impl.LaunchCmd(args.Cmd, args.Ctx)
return err
}
func (e *ExecutorRPCServer) Wait(args interface{}, ps *ProcessState) error {
var err error
ps, err = e.Impl.Wait()
return err
}
func (e *ExecutorRPCServer) ShutDown(args interface{}, ps *ProcessState) error {
var err error
ps, err = e.Impl.ShutDown()
return err
}
func (e *ExecutorRPCServer) Exit(args interface{}, ps *ProcessState) error {
var err error
ps, err = e.Impl.Exit()
return err
}
type ExecutorPlugin struct{}
func (p *ExecutorPlugin) Server(*plugin.MuxBroker) (interface{}, error) {
return &ExecutorRPCServer{Impl: NewExecutor()}, nil
}
func (p *ExecutorPlugin) Client(b *plugin.MuxBroker, c *rpc.Client) (interface{}, error) {
return &ExecutorRPC{client: c}, nil
}

35
client/driver/plugins/executor_linux.go Normal file
View file

@ -0,0 +1,35 @@
package plugins
import (
"log"
"os/exec"
"time"
)
type LinuxExecutor struct {
cmd *exec.Cmd
ctx *ExecutorContext
log *log.Logger
}
func NewExecutor() Executor {
return &LinuxExecutor{}
}
func (e *LinuxExecutor) LaunchCmd(cmd *exec.Cmd, ctx *ExecutorContext) (*ProcessState, error) {
return &ProcessState{Pid: 5, ExitCode: -1, Time: time.Now()}, nil
}
func (e *LinuxExecutor) Wait() (*ProcessState, error) {
time.Sleep(5 * time.Second)
return &ProcessState{Pid: 0, ExitCode: 0, Time: time.Now()}, nil
}
func (e *LinuxExecutor) Exit() (*ProcessState, error) {
return &ProcessState{Pid: 0, ExitCode: 0, Time: time.Now()}, nil
}
func (e *LinuxExecutor) ShutDown() (*ProcessState, error) {
return &ProcessState{Pid: 0, ExitCode: 0, Time: time.Now()}, nil
}

32
command/executor_plugin.go Normal file
View file

@ -0,0 +1,32 @@
package command
import (
"strings"
"github.com/hashicorp/go-plugin"
"github.com/hashicorp/nomad/client/driver/plugins"
)
type ExecutorPlugin struct {
Meta
}
func (e *ExecutorPlugin) Help() string {
helpText := `
This is a command used by Nomad internally to launch an executor plugin"
`
return strings.TrimSpace(helpText)
}
func (e *ExecutorPlugin) Synopsis() string {
return "internal - launch an executor plugin"
}
func (e *ExecutorPlugin) Run(args []string) int {
plugin.Serve(&plugin.ServeConfig{
HandshakeConfig: plugins.HandshakeConfig,
Plugins: plugins.PluginMap,
})
return 0
}