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

Added executorconfig

Browse source
This commit is contained in:
Diptanu Choudhury 2017-01-12 11:50:49 -08:00
parent dffc0acfc9
commit 6809a4b104
13 changed files with 97 additions and 154 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
client/driver/docker.go
View file

@ -6,7 +6,6 @@ import (
"log"
"net"
"os"
"os/exec"
"path/filepath"
"regexp"
"runtime"
@ -26,7 +25,6 @@ import (
"github.com/hashicorp/nomad/client/driver/executor"
dstructs "github.com/hashicorp/nomad/client/driver/structs"
cstructs "github.com/hashicorp/nomad/client/structs"
"github.com/hashicorp/nomad/helper/discover"
"github.com/hashicorp/nomad/helper/fields"
shelpers "github.com/hashicorp/nomad/helper/stats"
"github.com/hashicorp/nomad/nomad/structs"
@ -387,17 +385,14 @@ func (d *DockerDriver) Prestart(ctx *ExecContext, task *structs.Task) error {
}
func (d *DockerDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
bin, err := discover.NomadExecutable()
if err != nil {
return nil, fmt.Errorf("unable to find the nomad binary: %v", err)
}
pluginLogFile := filepath.Join(ctx.TaskDir.Dir, "executor.out")
pluginConfig := &plugin.ClientConfig{
Cmd: exec.Command(bin, "executor", pluginLogFile, ctx.LogLevel),
executorConfig := &dstructs.ExecutorConfig{
LogFile: pluginLogFile,
LogLevel: d.config.LogLevel,
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutor(d.config.LogOutput, d.config, executorConfig)
if err != nil {
return nil, err
}
@ -1119,7 +1114,7 @@ func (d *DockerDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, er
if !found {
return nil, fmt.Errorf("Failed to find container %s", pid.ContainerID)
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutorWithConfig(pluginConfig, d.config.LogOutput)
if err != nil {
d.logger.Printf("[INFO] driver.docker: couldn't re-attach to the plugin process: %v", err)
d.logger.Printf("[DEBUG] driver.docker: stopping container %q", pid.ContainerID)

10
client/driver/driver.go
View file

@ -152,17 +152,13 @@ type ExecContext struct {
// Alloc ID
AllocID string
// LogLevel is the level of the logs to putout
LogLevel string
}
// NewExecContext is used to create a new execution context
func NewExecContext(td *allocdir.TaskDir, allocID string, logLevel string) *ExecContext {
func NewExecContext(td *allocdir.TaskDir, allocID string) *ExecContext {
return &ExecContext{
TaskDir: td,
AllocID: allocID,
LogLevel: logLevel,
TaskDir: td,
AllocID: allocID,
}
}

16
client/driver/exec.go
View file

@ -5,7 +5,6 @@ import (
"fmt"
"log"
"os"
"os/exec"
"path/filepath"
"time"
@ -15,7 +14,6 @@ import (
"github.com/hashicorp/nomad/client/driver/executor"
dstructs "github.com/hashicorp/nomad/client/driver/structs"
cstructs "github.com/hashicorp/nomad/client/structs"
"github.com/hashicorp/nomad/helper/discover"
"github.com/hashicorp/nomad/helper/fields"
"github.com/hashicorp/nomad/nomad/structs"
"github.com/mitchellh/mapstructure"
@ -110,16 +108,12 @@ func (d *ExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
return nil, err
}
bin, err := discover.NomadExecutable()
if err != nil {
return nil, fmt.Errorf("unable to find the nomad binary: %v", err)
}
pluginLogFile := filepath.Join(ctx.TaskDir.Dir, "executor.out")
pluginConfig := &plugin.ClientConfig{
Cmd: exec.Command(bin, "executor", pluginLogFile, ctx.LogLevel),
executorConfig := &dstructs.ExecutorConfig{
LogFile: pluginLogFile,
LogLevel: d.config.LogLevel,
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutor(d.config.LogOutput, d.config, executorConfig)
if err != nil {
return nil, err
}
@ -191,7 +185,7 @@ func (d *ExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro
pluginConfig := &plugin.ClientConfig{
Reattach: id.PluginConfig.PluginConfig(),
}
exec, client, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, client, err := createExecutorWithConfig(pluginConfig, d.config.LogOutput)
if err != nil {
merrs := new(multierror.Error)
merrs.Errors = append(merrs.Errors, err)

15
client/driver/java.go
View file

@ -22,7 +22,6 @@ import (
dstructs "github.com/hashicorp/nomad/client/driver/structs"
"github.com/hashicorp/nomad/client/fingerprint"
cstructs "github.com/hashicorp/nomad/client/structs"
"github.com/hashicorp/nomad/helper/discover"
"github.com/hashicorp/nomad/helper/fields"
"github.com/hashicorp/nomad/nomad/structs"
)
@ -191,17 +190,13 @@ func (d *JavaDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
args = append(args, driverConfig.Args...)
}
bin, err := discover.NomadExecutable()
if err != nil {
return nil, fmt.Errorf("unable to find the nomad binary: %v", err)
}
pluginLogFile := filepath.Join(ctx.TaskDir.Dir, "executor.out")
pluginConfig := &plugin.ClientConfig{
Cmd: exec.Command(bin, "executor", pluginLogFile, ctx.LogLevel),
executorConfig := &dstructs.ExecutorConfig{
LogFile: pluginLogFile,
LogLevel: d.config.LogLevel,
}
execIntf, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
execIntf, pluginClient, err := createExecutor(d.config.LogOutput, d.config, executorConfig)
if err != nil {
return nil, err
}
@ -285,7 +280,7 @@ func (d *JavaDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro
pluginConfig := &plugin.ClientConfig{
Reattach: id.PluginConfig.PluginConfig(),
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutorWithConfig(pluginConfig, d.config.LogOutput)
if err != nil {
merrs := new(multierror.Error)
merrs.Errors = append(merrs.Errors, err)

10
client/driver/plugins.go
View file

@ -2,7 +2,6 @@ package driver
import (
"io"
"io/ioutil"
"log"
"net"
"strings"
@ -22,16 +21,13 @@ func GetPluginMap(w io.Writer, logLevel string) map[string]plugin.Plugin {
filter := &logutils.LevelFilter{
Levels: []logutils.LogLevel{"TRACE", "DEBUG", "INFO", "WARN", "ERR"},
MinLevel: logutils.LogLevel(strings.ToUpper(logLevel)),
Writer: ioutil.Discard,
Writer: w,
}
e.logger = log.New(filter, "", log.LstdFlags)
e.logger = log.New(filter, "", log.LstdFlags|log.Lmicroseconds)
s := new(SyslogCollectorPlugin)
s.logger = log.New(w, "", log.LstdFlags)
return map[string]plugin.Plugin{
"executor": e,
"syslogcollector": s,
"executor": e,
}
}

15
client/driver/qemu.go
View file

@ -18,7 +18,6 @@ import (
dstructs "github.com/hashicorp/nomad/client/driver/structs"
"github.com/hashicorp/nomad/client/fingerprint"
cstructs "github.com/hashicorp/nomad/client/structs"
"github.com/hashicorp/nomad/helper/discover"
"github.com/hashicorp/nomad/helper/fields"
"github.com/hashicorp/nomad/nomad/structs"
"github.com/mitchellh/mapstructure"
@ -233,17 +232,13 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle,
}
d.logger.Printf("[DEBUG] Starting QemuVM command: %q", strings.Join(args, " "))
bin, err := discover.NomadExecutable()
if err != nil {
return nil, fmt.Errorf("unable to find the nomad binary: %v", err)
}
pluginLogFile := filepath.Join(ctx.TaskDir.Dir, "executor.out")
pluginConfig := &plugin.ClientConfig{
Cmd: exec.Command(bin, "executor", pluginLogFile, ctx.LogLevel),
executorConfig := &dstructs.ExecutorConfig{
LogFile: pluginLogFile,
LogLevel: d.config.LogLevel,
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutor(d.config.LogOutput, d.config, executorConfig)
if err != nil {
return nil, err
}
@ -311,7 +306,7 @@ func (d *QemuDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro
Reattach: id.PluginConfig.PluginConfig(),
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutorWithConfig(pluginConfig, d.config.LogOutput)
if err != nil {
d.logger.Println("[ERR] driver.qemu: error connecting to plugin so destroying plugin pid and user pid")
if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil {

15
client/driver/raw_exec.go
View file

@ -5,7 +5,6 @@ import (
"fmt"
"log"
"os"
"os/exec"
"path/filepath"
"time"
@ -15,7 +14,6 @@ import (
dstructs "github.com/hashicorp/nomad/client/driver/structs"
"github.com/hashicorp/nomad/client/fingerprint"
cstructs "github.com/hashicorp/nomad/client/structs"
"github.com/hashicorp/nomad/helper/discover"
"github.com/hashicorp/nomad/helper/fields"
"github.com/hashicorp/nomad/nomad/structs"
"github.com/mitchellh/mapstructure"
@ -124,16 +122,13 @@ func (d *RawExecDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandl
return nil, err
}
bin, err := discover.NomadExecutable()
if err != nil {
return nil, fmt.Errorf("unable to find the nomad binary: %v", err)
}
pluginLogFile := filepath.Join(ctx.TaskDir.Dir, "executor.out")
pluginConfig := &plugin.ClientConfig{
Cmd: exec.Command(bin, "executor", pluginLogFile, ctx.LogLevel),
executorConfig := &dstructs.ExecutorConfig{
LogFile: pluginLogFile,
LogLevel: d.config.LogLevel,
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutor(d.config.LogOutput, d.config, executorConfig)
if err != nil {
return nil, err
}
@ -199,7 +194,7 @@ func (d *RawExecDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, e
pluginConfig := &plugin.ClientConfig{
Reattach: id.PluginConfig.PluginConfig(),
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutorWithConfig(pluginConfig, d.config.LogOutput)
if err != nil {
d.logger.Println("[ERR] driver.raw_exec: error connecting to plugin so destroying plugin pid and user pid")
if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil {

15
client/driver/rkt.go
View file

@ -23,7 +23,6 @@ import (
"github.com/hashicorp/nomad/client/driver/executor"
dstructs "github.com/hashicorp/nomad/client/driver/structs"
cstructs "github.com/hashicorp/nomad/client/structs"
"github.com/hashicorp/nomad/helper/discover"
"github.com/hashicorp/nomad/helper/fields"
"github.com/hashicorp/nomad/nomad/structs"
"github.com/mitchellh/mapstructure"
@ -394,17 +393,13 @@ func (d *RktDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, e
filter := strings.Split(d.config.ReadDefault("env.blacklist", config.DefaultEnvBlacklist), ",")
d.taskEnv.AppendHostEnvvars(filter)
bin, err := discover.NomadExecutable()
if err != nil {
return nil, fmt.Errorf("unable to find the nomad binary: %v", err)
}
pluginLogFile := filepath.Join(ctx.TaskDir.Dir, fmt.Sprintf("%s-executor.out", task.Name))
pluginConfig := &plugin.ClientConfig{
Cmd: exec.Command(bin, "executor", pluginLogFile, ctx.LogLevel),
executorConfig := &dstructs.ExecutorConfig{
LogFile: pluginLogFile,
LogLevel: d.config.LogLevel,
}
execIntf, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
execIntf, pluginClient, err := createExecutor(d.config.LogOutput, d.config, executorConfig)
if err != nil {
return nil, err
}
@ -467,7 +462,7 @@ func (d *RktDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error
pluginConfig := &plugin.ClientConfig{
Reattach: id.PluginConfig.PluginConfig(),
}
exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
exec, pluginClient, err := createExecutorWithConfig(pluginConfig, d.config.LogOutput)
if err != nil {
d.logger.Println("[ERROR] driver.rkt: error connecting to plugin so destroying plugin pid and user pid")
if e := destroyPlugin(id.PluginConfig.Pid, id.ExecutorPid); e != nil {

10
client/driver/structs/structs.go
View file

@ -55,3 +55,13 @@ type CheckResult struct {
// Err is the error that a check returned
Err error
}
// ExecutorConfig is the config that Nomad passes to the executor
type ExecutorConfig struct {
// LogFile is the file to which Executor logs
LogFile string
// LogLevel is the level of the logs to putout
LogLevel string
}

69
client/driver/syslog_plugin.go
View file

@ -1,69 +0,0 @@
package driver
import (
"log"
"net/rpc"
"github.com/hashicorp/go-plugin"
"github.com/hashicorp/nomad/client/driver/logging"
"github.com/hashicorp/nomad/nomad/structs"
)
type SyslogCollectorRPC struct {
client *rpc.Client
}
type LaunchCollectorArgs struct {
Ctx *logging.LogCollectorContext
}
func (e *SyslogCollectorRPC) LaunchCollector(ctx *logging.LogCollectorContext) (*logging.SyslogCollectorState, error) {
var ss *logging.SyslogCollectorState
err := e.client.Call("Plugin.LaunchCollector", LaunchCollectorArgs{Ctx: ctx}, &ss)
return ss, err
}
func (e *SyslogCollectorRPC) Exit() error {
return e.client.Call("Plugin.Exit", new(interface{}), new(interface{}))
}
func (e *SyslogCollectorRPC) UpdateLogConfig(logConfig *structs.LogConfig) error {
return e.client.Call("Plugin.UpdateLogConfig", logConfig, new(interface{}))
}
type SyslogCollectorRPCServer struct {
Impl logging.LogCollector
}
func (s *SyslogCollectorRPCServer) LaunchCollector(args LaunchCollectorArgs,
resp *logging.SyslogCollectorState) error {
ss, err := s.Impl.LaunchCollector(args.Ctx)
if ss != nil {
*resp = *ss
}
return err
}
func (s *SyslogCollectorRPCServer) Exit(args interface{}, resp *interface{}) error {
return s.Impl.Exit()
}
func (s *SyslogCollectorRPCServer) UpdateLogConfig(logConfig *structs.LogConfig, resp *interface{}) error {
return s.Impl.UpdateLogConfig(logConfig)
}
type SyslogCollectorPlugin struct {
logger *log.Logger
Impl *SyslogCollectorRPCServer
}
func (p *SyslogCollectorPlugin) Server(*plugin.MuxBroker) (interface{}, error) {
if p.Impl == nil {
p.Impl = &SyslogCollectorRPCServer{Impl: logging.NewSyslogCollector(p.logger)}
}
return p.Impl, nil
}
func (p *SyslogCollectorPlugin) Client(b *plugin.MuxBroker, c *rpc.Client) (interface{}, error) {
return &SyslogCollectorRPC{client: c}, nil
}

40
client/driver/utils.go
View file

@ -1,6 +1,7 @@
package driver
import (
"encoding/json"
"fmt"
"io"
"os"
@ -14,13 +15,27 @@ import (
"github.com/hashicorp/nomad/client/config"
"github.com/hashicorp/nomad/client/driver/executor"
cstructs "github.com/hashicorp/nomad/client/driver/structs"
"github.com/hashicorp/nomad/helper/discover"
"github.com/hashicorp/nomad/nomad/structs"
)
// createExecutor launches an executor plugin and returns an instance of the
// Executor interface
func createExecutor(config *plugin.ClientConfig, w io.Writer,
clientConfig *config.Config) (executor.Executor, *plugin.Client, error) {
func createExecutor(w io.Writer, clientConfig *config.Config,
executorConfig *cstructs.ExecutorConfig) (executor.Executor, *plugin.Client, error) {
c, err := json.Marshal(executorConfig)
if err != nil {
return nil, nil, fmt.Errorf("unable to create executor config: %v", err)
}
bin, err := discover.NomadExecutable()
if err != nil {
return nil, nil, fmt.Errorf("unable to find the nomad binary: %v", err)
}
config := &plugin.ClientConfig{
Cmd: exec.Command(bin, "executor", string(c)),
}
config.HandshakeConfig = HandshakeConfig
config.Plugins = GetPluginMap(w, clientConfig.LogLevel)
config.MaxPort = clientConfig.ClientMaxPort
@ -46,6 +61,27 @@ func createExecutor(config *plugin.ClientConfig, w io.Writer,
return executorPlugin, executorClient, nil
}
func createExecutorWithConfig(config *plugin.ClientConfig, w io.Writer) (executor.Executor, *plugin.Client, error) {
config.HandshakeConfig = HandshakeConfig
// Setting this to DEBUG since the log level at the executor server process
// is already set, and this effects only the executor client.
config.Plugins = GetPluginMap(w, "DEBUG")
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.(executor.Executor)
return executorPlugin, executorClient, nil
}
func consulContext(clientConfig *config.Config, containerID string) *executor.ConsulContext {
return &executor.ConsulContext{
ConsulConfig: clientConfig.ConsulConfig,

4
client/task_runner.go
View file

@ -267,7 +267,7 @@ func (r *TaskRunner) RestoreState() error {
return err
}
ctx := driver.NewExecContext(r.taskDir, r.alloc.ID, r.config.LogLevel)
ctx := driver.NewExecContext(r.taskDir, r.alloc.ID)
handle, err := d.Open(ctx, snap.HandleID)
// In the case it fails, we relaunch the task in the Run() method.
@ -1094,7 +1094,7 @@ func (r *TaskRunner) startTask() error {
}
// Run prestart
ctx := driver.NewExecContext(r.taskDir, r.alloc.ID, r.config.LogLevel)
ctx := driver.NewExecContext(r.taskDir, r.alloc.ID)
if err := drv.Prestart(ctx, r.task); err != nil {
wrapped := fmt.Errorf("failed to initialize task %q for alloc %q: %v",
r.task.Name, r.alloc.ID, err)

17
command/executor_plugin.go
View file

@ -1,12 +1,14 @@
package command
import (
"encoding/json"
"os"
"strings"
"github.com/hashicorp/go-plugin"
"github.com/hashicorp/nomad/client/driver"
dstructs "github.com/hashicorp/nomad/client/driver/structs"
)
type ExecutorPluginCommand struct {
@ -25,20 +27,23 @@ func (e *ExecutorPluginCommand) Synopsis() string {
}
func (e *ExecutorPluginCommand) Run(args []string) int {
if len(args) != 2 {
e.Ui.Error("log output file and log level are not provided")
if len(args) != 1 {
e.Ui.Error("json configuration not provided")
return 1
}
logFileName := args[0]
logLevel := args[1]
stdo, err := os.OpenFile(logFileName, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666)
config := args[0]
var executorConfig dstructs.ExecutorConfig
if err := json.Unmarshal([]byte(config), &executorConfig); err != nil {
return 1
}
stdo, err := os.OpenFile(executorConfig.LogFile, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666)
if err != nil {
e.Ui.Error(err.Error())
return 1
}
plugin.Serve(&plugin.ServeConfig{
HandshakeConfig: driver.HandshakeConfig,
Plugins: driver.GetPluginMap(stdo, logLevel),
Plugins: driver.GetPluginMap(stdo, executorConfig.LogLevel),
})
return 0
}