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

driver/raw_exec: port existing raw_exec tests and add some testing utilities

Browse source
This commit is contained in:
Nick Ethier 2018-09-26 01:18:03 -04:00 committed by Michael Schurter
parent 8644e8508c
commit 0e3f85222a
7 changed files with 400 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
client/driver/executor/executor.go
View file

@ -358,9 +358,10 @@ func (e *UniversalExecutor) UpdateResources(resources *Resources) error {
func (e *UniversalExecutor) wait() {
defer close(e.processExited)
pid := e.childCmd.Process.Pid
err := e.childCmd.Wait()
if err == nil {
e.exitState = &ProcessState{Pid: 0, ExitCode: 0, Time: time.Now()}
e.exitState = &ProcessState{Pid: pid, ExitCode: 0, Time: time.Now()}
return
}
@ -388,7 +389,7 @@ func (e *UniversalExecutor) wait() {
e.logger.Warn("unexpected Cmd.Wait() error type", "error", err)
}
e.exitState = &ProcessState{Pid: 0, ExitCode: exitCode, Signal: signal, Time: time.Now()}
e.exitState = &ProcessState{Pid: pid, ExitCode: exitCode, Signal: signal, Time: time.Now()}
}
var (

10
helper/testtask/testtask.go
View file

@ -12,6 +12,7 @@ import (
"time"
"github.com/hashicorp/nomad/nomad/structs"
"github.com/hashicorp/nomad/plugins/drivers/base"
)
// Path returns the path to the currently running executable.
@ -38,6 +39,15 @@ func SetTaskEnv(t *structs.Task) {
t.Env["TEST_TASK"] = "execute"
}
// SetTaskConfigEnv configures the environment of t so that Run executes a testtask
// script when called from within t.
func SetTaskConfigEnv(t *base.TaskConfig) {
if t.Env == nil {
t.Env = map[string]string{}
}
t.Env["TEST_TASK"] = "execute"
}
// Run interprets os.Args as a testtask script if the current program was
// launched with an environment configured by SetCmdEnv or SetTaskEnv. It
// returns false if the environment was not set by this package.

2
plugins/drivers/client.go
View file

@ -17,7 +17,7 @@ import (
var _ DriverPlugin = &driverPluginClient{}
type driverPluginClient struct {
base.BasePluginClient
*base.BasePluginClient
client proto.DriverClient
logger hclog.Logger

4
plugins/drivers/driver.go
View file

@ -177,6 +177,10 @@ type ExitResult struct {
Err error
}
func (r *ExitResult) Successful() bool {
return r.ExitCode == 0 && r.Signal == 0 && r.Err == nil
}
type TaskStatus struct {
ID string
Name string

7
plugins/drivers/plugin.go
View file

@ -5,7 +5,9 @@ import (
hclog "github.com/hashicorp/go-hclog"
plugin "github.com/hashicorp/go-plugin"
"github.com/hashicorp/nomad/plugins/drivers/proto"
"github.com/hashicorp/nomad/plugins/base"
baseproto "github.com/hashicorp/nomad/plugins/base/proto"
"github.com/hashicorp/nomad/plugins/drivers/base/proto"
"google.golang.org/grpc"
)
@ -35,6 +37,9 @@ func (p *PluginDriver) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) err
func (p *PluginDriver) GRPCClient(ctx context.Context, broker *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error) {
return &driverPluginClient{
BasePluginClient: &base.BasePluginClient{
Client: baseproto.NewBasePluginClient(c),
},
client: proto.NewDriverClient(c),
logger: p.logger,
}, nil

322
plugins/drivers/raw_exec/driver_test.go
View file

@ -2,18 +2,106 @@ package raw_exec
import (
"context"
"fmt"
"io/ioutil"
"os"
"path/filepath"
"strconv"
"sync"
"syscall"
"testing"
"time"
"github.com/davecgh/go-spew/spew"
ctestutil "github.com/hashicorp/nomad/client/testutil"
"github.com/hashicorp/nomad/helper/testlog"
"github.com/hashicorp/nomad/helper/testtask"
"github.com/hashicorp/nomad/helper/uuid"
basePlug "github.com/hashicorp/nomad/plugins/base"
"github.com/hashicorp/nomad/plugins/drivers/base"
"github.com/hashicorp/nomad/testutil"
"github.com/stretchr/testify/require"
)
func TestMain(m *testing.M) {
if !testtask.Run() {
os.Exit(m.Run())
}
}
func TestRawExecDriver_SetConfig(t *testing.T) {
t.Parallel()
require := require.New(t)
d := NewRawExecDriver(testlog.HCLogger(t))
harness := base.NewDriverHarness(t, d)
// Disable raw exec.
config := &Config{}
var data []byte
require.NoError(basePlug.MsgPackEncode(&data, config))
require.NoError(harness.SetConfig(data))
require.Exactly(config, d.(*RawExecDriver).config)
config.Enabled = true
config.NoCgroups = true
data = []byte{}
require.NoError(basePlug.MsgPackEncode(&data, config))
require.NoError(harness.SetConfig(data))
require.Exactly(config, d.(*RawExecDriver).config)
config.NoCgroups = false
data = []byte{}
require.NoError(basePlug.MsgPackEncode(&data, config))
require.NoError(harness.SetConfig(data))
require.Exactly(config, d.(*RawExecDriver).config)
}
func TestRawExecDriver_Fingerprint(t *testing.T) {
t.Parallel()
require := require.New(t)
d := NewRawExecDriver(testlog.HCLogger(t))
harness := base.NewDriverHarness(t, d)
// Disable raw exec.
config := &Config{}
var data []byte
require.NoError(basePlug.MsgPackEncode(&data, config))
require.NoError(harness.SetConfig(data))
fingerCh, err := harness.Fingerprint(context.Background())
require.NoError(err)
select {
case finger := <-fingerCh:
require.Equal(base.HealthStateUndetected, finger.Health)
require.Empty(finger.Attributes["driver.raw_exec"])
case <-time.After(time.Duration(testutil.TestMultiplier()*5) * time.Second):
require.Fail("timeout receiving fingerprint")
}
// Enable raw exec
config.Enabled = true
data = []byte{}
require.NoError(basePlug.MsgPackEncode(&data, config))
require.NoError(harness.SetConfig(data))
FINGER_LOOP:
for {
select {
case finger := <-fingerCh:
if finger.Health == base.HealthStateHealthy {
break FINGER_LOOP
}
case <-time.After(time.Duration(testutil.TestMultiplier()*5) * time.Second):
require.Fail("timeout receiving fingerprint")
break FINGER_LOOP
}
}
}
func TestRawExecDriver_StartWait(t *testing.T) {
t.Parallel()
require := require.New(t)
@ -28,7 +116,7 @@ func TestRawExecDriver_StartWait(t *testing.T) {
Command: "go",
Args: []string{"version"},
})
cleanup := harness.MkAllocDir(task)
cleanup := harness.MkAllocDir(task, false)
defer cleanup()
handle, err := harness.StartTask(task)
@ -52,25 +140,12 @@ func TestRawExecDriver_StartWaitStop(t *testing.T) {
Name: "test",
}
task.EncodeDriverConfig(&TaskConfig{
Command: "/bin/bash",
Args: []string{"test.sh"},
Command: testtask.Path(),
Args: []string{"sleep", "100s"},
})
cleanup := harness.MkAllocDir(task)
cleanup := harness.MkAllocDir(task, false)
defer cleanup()
testFile := filepath.Join(task.TaskDir().Dir, "test.sh")
testData := []byte(`
at_term() {
echo 'Terminated.'
exit 3
}
trap at_term USR1
while true; do
sleep 1
done
`)
require.NoError(ioutil.WriteFile(testFile, testData, 0777))
handle, err := harness.StartTask(task)
require.NoError(err)
@ -83,17 +158,18 @@ done
go func() {
defer wg.Done()
result := <-ch
require.Equal(3, result.ExitCode)
spew.Dump(result)
require.Equal(2, result.Signal)
waitDone = true
}()
time.Sleep(100 * time.Millisecond)
require.NoError(harness.WaitUntilStarted(task.ID, 1*time.Second))
var stopDone bool
wg.Add(1)
go func() {
defer wg.Done()
err := harness.StopTask(task.ID, 1*time.Second, "SIGUSR1")
err := harness.StopTask(task.ID, 2*time.Second, "SIGINT")
require.NoError(err)
stopDone = true
}()
@ -104,14 +180,6 @@ done
wg.Wait()
}()
time.Sleep(100 * time.Millisecond)
status, err := harness.InspectTask(task.ID)
require.NoError(err)
require.Equal(base.TaskStateRunning, status.State)
require.False(waitDone)
require.False(stopDone)
select {
case <-waitCh:
status, err := harness.InspectTask(task.ID)
@ -132,13 +200,14 @@ func TestRawExecDriver_StartWaitRecoverWaitStop(t *testing.T) {
harness := base.NewDriverHarness(t, d)
task := &base.TaskConfig{
ID: uuid.Generate(),
Name: "test",
Name: "sleep",
}
task.EncodeDriverConfig(&TaskConfig{
Command: "sleep",
Args: []string{"1000"},
Command: testtask.Path(),
Args: []string{"sleep", "100s"},
})
cleanup := harness.MkAllocDir(task)
testtask.SetTaskConfigEnv(task)
cleanup := harness.MkAllocDir(task, false)
defer cleanup()
handle, err := harness.StartTask(task)
@ -183,13 +252,198 @@ func TestRawExecDriver_StartWaitRecoverWaitStop(t *testing.T) {
defer wg.Done()
result := <-ch
require.NoError(result.Err)
require.NotZero(result.ExitCode)
require.Equal(9, result.Signal)
waitDone = true
}()
require.NoError(d.StopTask(task.ID, 0, ""))
require.NoError(d.DestroyTask(task.ID, false))
time.Sleep(300 * time.Millisecond)
require.NoError(d.StopTask(task.ID, 0, "SIGKILL"))
wg.Wait()
require.NoError(d.DestroyTask(task.ID, false))
require.True(waitDone)
}
func TestRawExecDriver_Start_Wait_AllocDir(t *testing.T) {
t.Parallel()
require := require.New(t)
d := NewRawExecDriver(testlog.HCLogger(t))
harness := base.NewDriverHarness(t, d)
task := &base.TaskConfig{
ID: uuid.Generate(),
Name: "sleep",
}
cleanup := harness.MkAllocDir(task, false)
defer cleanup()
exp := []byte("win")
file := "output.txt"
outPath := fmt.Sprintf(`%s/%s`, task.TaskDir().SharedAllocDir, file)
task.EncodeDriverConfig(&TaskConfig{
Command: testtask.Path(),
Args: []string{
"sleep", "1s", "write",
string(exp), outPath,
},
})
testtask.SetTaskConfigEnv(task)
_, err := harness.StartTask(task)
require.NoError(err)
// Task should terminate quickly
waitCh, err := harness.WaitTask(context.Background(), task.ID)
select {
case res := <-waitCh:
require.NoError(res.Err)
require.True(res.Successful())
case <-time.After(time.Duration(testutil.TestMultiplier()*5) * time.Second):
require.Fail("WaitTask timeout")
}
// Check that data was written to the shared alloc directory.
outputFile := filepath.Join(task.TaskDir().SharedAllocDir, file)
act, err := ioutil.ReadFile(outputFile)
require.NoError(err)
require.Exactly(exp, act)
require.NoError(harness.DestroyTask(task.ID, true))
}
// This test creates a process tree such that without cgroups tracking the
// processes cleanup of the children would not be possible. Thus the test
// asserts that the processes get killed properly when using cgroups.
func TestRawExecDriver_Start_Kill_Wait_Cgroup(t *testing.T) {
ctestutil.ExecCompatible(t)
t.Parallel()
require := require.New(t)
pidFile := "pid"
d := NewRawExecDriver(testlog.HCLogger(t))
harness := base.NewDriverHarness(t, d)
task := &base.TaskConfig{
ID: uuid.Generate(),
Name: "sleep",
User: "root",
}
cleanup := harness.MkAllocDir(task, false)
defer cleanup()
task.EncodeDriverConfig(&TaskConfig{
Command: testtask.Path(),
Args: []string{"fork/exec", pidFile, "pgrp", "0", "sleep", "20s"},
})
testtask.SetTaskConfigEnv(task)
_, err := harness.StartTask(task)
require.NoError(err)
// Find the process
var pidData []byte
testutil.WaitForResult(func() (bool, error) {
var err error
pidData, err = ioutil.ReadFile(filepath.Join(task.TaskDir().Dir, pidFile))
if err != nil {
return false, err
}
if len(pidData) == 0 {
return false, fmt.Errorf("pidFile empty")
}
return true, nil
}, func(err error) {
require.NoError(err)
})
pid, err := strconv.Atoi(string(pidData))
require.NoError(err, "failed to read pidData: %s", string(pidData))
// Check the pid is up
process, err := os.FindProcess(pid)
require.NoError(err)
require.NoError(process.Signal(syscall.Signal(0)))
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
time.Sleep(1 * time.Second)
err := harness.StopTask(task.ID, 0, "")
// Can't rely on the ordering between wait and kill on travis...
if !testutil.IsTravis() {
require.NoError(err)
}
}()
// Task should terminate quickly
waitCh, err := harness.WaitTask(context.Background(), task.ID)
require.NoError(err)
select {
case res := <-waitCh:
require.False(res.Successful())
case <-time.After(time.Duration(testutil.TestMultiplier()*5) * time.Second):
require.Fail("WaitTask timeout")
}
testutil.WaitForResult(func() (bool, error) {
if err := process.Signal(syscall.Signal(0)); err == nil {
return false, fmt.Errorf("process should not exist: %v", pid)
}
return true, nil
}, func(err error) {
require.NoError(err)
})
wg.Wait()
require.NoError(harness.DestroyTask(task.ID, true))
}
func TestRawExecDriver_Exec(t *testing.T) {
t.Parallel()
require := require.New(t)
d := NewRawExecDriver(testlog.HCLogger(t))
harness := base.NewDriverHarness(t, d)
task := &base.TaskConfig{
ID: uuid.Generate(),
Name: "sleep",
}
cleanup := harness.MkAllocDir(task, false)
defer cleanup()
task.EncodeDriverConfig(&TaskConfig{
Command: testtask.Path(),
Args: []string{"sleep", "9000s"},
})
testtask.SetTaskConfigEnv(task)
_, err := harness.StartTask(task)
require.NoError(err)
// Exec a command that should work
res, err := harness.ExecTask(task.ID, []string{"/usr/bin/stat", "/tmp"}, 1*time.Second)
require.NoError(err)
require.True(res.ExitResult.Successful())
require.True(len(res.Stdout) > 100)
// Exec a command that should fail
res, err = harness.ExecTask(task.ID, []string{"/usr/bin/stat", "notarealfile123abc"}, 1*time.Second)
require.NoError(err)
require.False(res.ExitResult.Successful())
require.Contains(string(res.Stdout), "No such file or directory")
require.NoError(harness.DestroyTask(task.ID, true))
}

96
plugins/drivers/testing.go
View file

@ -1,17 +1,22 @@
package drivers
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"runtime"
"time"
"github.com/mitchellh/go-testing-interface"
"github.com/stretchr/testify/require"
"golang.org/x/net/context"
hclog "github.com/hashicorp/go-hclog"
plugin "github.com/hashicorp/go-plugin"
"github.com/hashicorp/nomad/client/allocdir"
"github.com/hashicorp/nomad/client/logmon"
"github.com/hashicorp/nomad/helper/testlog"
"github.com/hashicorp/nomad/helper/uuid"
"github.com/hashicorp/nomad/plugins/base"
"github.com/hashicorp/nomad/plugins/shared/hclspec"
)
@ -21,15 +26,22 @@ type DriverHarness struct {
client *plugin.GRPCClient
server *plugin.GRPCServer
t testing.T
lm logmon.LogMon
logger hclog.Logger
}
func NewDriverHarness(t testing.T, d DriverPlugin) *DriverHarness {
client, server := plugin.TestPluginGRPCConn(t, map[string]plugin.Plugin{
logger := testlog.HCLogger(t).Named("driver_harness")
client, server := plugin.TestPluginGRPCConn(t,
map[string]plugin.Plugin{
base.PluginTypeDriver: &PluginDriver{
impl: d,
logger: testlog.HCLogger(t),
logger: logger.Named("driver_plugin"),
},
})
base.PluginTypeBase: &base.PluginBase{Impl: d},
"logmon": logmon.NewPlugin(logmon.NewLogMon(logger.Named("logmon"))),
},
)
raw, err := client.Dispense(base.PluginTypeDriver)
if err != nil {
@ -41,8 +53,15 @@ func NewDriverHarness(t testing.T, d DriverPlugin) *DriverHarness {
client: client,
server: server,
DriverPlugin: dClient,
logger: logger,
}
raw, err = client.Dispense("logmon")
if err != nil {
t.Fatalf("err dispensing plugin: %v", err)
}
h.lm = raw.(logmon.LogMon)
return h
}
@ -52,15 +71,72 @@ func (h *DriverHarness) Kill() {
}
// MkAllocDir creates a tempory directory and allocdir structure.
// If enableLogs is set to true a logmon instance will be started to write logs
// to the LogDir of the task
// A cleanup func is returned and should be defered so as to not leak dirs
// between tests.
func (h *DriverHarness) MkAllocDir(t *TaskConfig) func() {
allocDir, err := ioutil.TempDir("", "nomad_driver_harness-")
func (h *DriverHarness) MkAllocDir(t *TaskConfig, enableLogs bool) func() {
dir, err := ioutil.TempDir("", "nomad_driver_harness-")
require.NoError(h.t, err)
require.NoError(h.t, os.Mkdir(filepath.Join(allocDir, t.Name), os.ModePerm))
require.NoError(h.t, os.MkdirAll(filepath.Join(allocDir, "alloc/logs"), os.ModePerm))
t.AllocDir = allocDir
return func() { os.RemoveAll(allocDir) }
t.AllocDir = dir
allocDir := allocdir.NewAllocDir(h.logger, dir)
require.NoError(h.t, allocDir.Build())
taskDir := allocDir.NewTaskDir(t.Name)
require.NoError(h.t, taskDir.Build(false, nil, 0))
//logmon
if enableLogs {
var stdoutFifo, stderrFifo string
if runtime.GOOS == "windows" {
id := uuid.Generate()[:8]
stdoutFifo = fmt.Sprintf("//./pipe/%s-%s.stdout", t.Name, id)
stderrFifo = fmt.Sprintf("//./pipe/%s-%s.stderr", t.Name, id)
} else {
stdoutFifo = filepath.Join(taskDir.LogDir, fmt.Sprintf(".%s.stdout.fifo", t.Name))
stderrFifo = filepath.Join(taskDir.LogDir, fmt.Sprintf(".%s.stderr.fifo", t.Name))
}
err = h.lm.Start(&logmon.LogConfig{
LogDir: taskDir.LogDir,
StdoutLogFile: fmt.Sprintf("%s.stdout", t.Name),
StderrLogFile: fmt.Sprintf("%s.stderr", t.Name),
StdoutFifo: stdoutFifo,
StderrFifo: stderrFifo,
MaxFiles: 10,
MaxFileSizeMB: 10,
})
require.NoError(h.t, err)
return func() {
h.lm.Stop()
allocDir.Destroy()
}
}
return func() {
allocDir.Destroy()
}
}
// WaitUntilStarted will block until the task for the given ID is in the running
// state or the timeout is reached
func (h *DriverHarness) WaitUntilStarted(taskID string, timeout time.Duration) error {
deadline := time.Now().Add(timeout)
var lastState TaskState
for {
status, err := h.InspectTask(taskID)
if err != nil {
return err
}
if status.State == TaskStateRunning {
return nil
}
lastState = status.State
if time.Now().After(deadline) {
return fmt.Errorf("task never transitioned to running, currently '%s'", lastState)
}
time.Sleep(100 * time.Millisecond)
}
}
// MockDriver is used for testing.