Add test cases for waiting on children

Also, make the test use files just like in the non-test case.

drivers/shared/executor/executor_linux_test.go

@@ -39,7 +39,7 @@ var libcontainerFactory = executorFactory{
 // chroot. Use testExecutorContext if you don't need a chroot.
 //
 // The caller is responsible for calling AllocDir.Destroy() to cleanup.
-func testExecutorCommandWithChroot(t *testing.T) (*ExecCommand, *allocdir.AllocDir) {
+func testExecutorCommandWithChroot(t *testing.T) *testExecCmd {
 	chrootEnv := map[string]string{
 		"/etc/ld.so.cache":  "/etc/ld.so.cache",
 		"/etc/ld.so.conf":   "/etc/ld.so.conf",
@@ -74,9 +74,13 @@ func testExecutorCommandWithChroot(t *testing.T) (*ExecCommand, *allocdir.AllocD
 			NomadResources: alloc.AllocatedResources.Tasks[task.Name],
 		},
 	}
-	configureTLogging(cmd)
 
-	return cmd, allocDir
+	testCmd := &testExecCmd{
+		command:  cmd,
+		allocDir: allocDir,
+	}
+	configureTLogging(t, testCmd)
+	return testCmd
 }
 
 func TestExecutor_IsolationAndConstraints(t *testing.T) {
@@ -84,7 +88,8 @@ func TestExecutor_IsolationAndConstraints(t *testing.T) {
 	require := require.New(t)
 	testutil.ExecCompatible(t)
 
-	execCmd, allocDir := testExecutorCommandWithChroot(t)
+	testExecCmd := testExecutorCommandWithChroot(t)
+	execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 	execCmd.Cmd = "/bin/ls"
 	execCmd.Args = []string{"-F", "/", "/etc/"}
 	defer allocDir.Destroy()
@@ -146,8 +151,7 @@ ld.so.cache
 ld.so.conf
 ld.so.conf.d/`
 	tu.WaitForResult(func() (bool, error) {
-		outWriter, _ := execCmd.GetWriters()
-		output := outWriter.(*bufferCloser).String()
+		output := testExecCmd.stdout.String()
 		act := strings.TrimSpace(string(output))
 		if act != expected {
 			return false, fmt.Errorf("Command output incorrectly: want %v; got %v", expected, act)
@@ -161,7 +165,8 @@ func TestExecutor_ClientCleanup(t *testing.T) {
 	testutil.ExecCompatible(t)
 	require := require.New(t)
 
-	execCmd, allocDir := testExecutorCommandWithChroot(t)
+	testExecCmd := testExecutorCommandWithChroot(t)
+	execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 	defer allocDir.Destroy()
 
 	executor := NewExecutorWithIsolation(testlog.HCLogger(t))
@@ -193,11 +198,10 @@ func TestExecutor_ClientCleanup(t *testing.T) {
 		require.Fail("timeout waiting for exec to shutdown")
 	}
 
-	outWriter, _ := execCmd.GetWriters()
-	output := outWriter.(*bufferCloser).String()
+	output := testExecCmd.stdout.String()
 	require.NotZero(len(output))
 	time.Sleep(2 * time.Second)
-	output1 := outWriter.(*bufferCloser).String()
+	output1 := testExecCmd.stdout.String()
 	require.Equal(len(output), len(output1))
 }
 

drivers/shared/executor/executor_test.go

@@ -4,11 +4,13 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"runtime"
 	"strings"
+	"sync"
 	"syscall"
 	"testing"
 	"time"
@@ -42,10 +44,19 @@ func init() {
 	executorFactories["UniversalExecutor"] = universalFactory
 }
 
+type testExecCmd struct {
+	command  *ExecCommand
+	allocDir *allocdir.AllocDir
+
+	stdout         *bytes.Buffer
+	stderr         *bytes.Buffer
+	outputCopyDone *sync.WaitGroup
+}
+
 // testExecutorContext returns an ExecutorContext and AllocDir.
 //
 // The caller is responsible for calling AllocDir.Destroy() to cleanup.
-func testExecutorCommand(t *testing.T) (*ExecCommand, *allocdir.AllocDir) {
+func testExecutorCommand(t *testing.T) *testExecCmd {
 	alloc := mock.Alloc()
 	task := alloc.Job.TaskGroups[0].Tasks[0]
 	taskEnv := taskenv.NewBuilder(mock.Node(), alloc, task, "global").Build()
@@ -78,18 +89,40 @@ func testExecutorCommand(t *testing.T) (*ExecCommand, *allocdir.AllocDir) {
 		},
 	}
 
-	configureTLogging(cmd)
-	return cmd, allocDir
+	testCmd := &testExecCmd{
+		command:  cmd,
+		allocDir: allocDir,
+	}
+	configureTLogging(t, testCmd)
+	return testCmd
 }
 
-type bufferCloser struct {
-	bytes.Buffer
-}
+func configureTLogging(t *testing.T, testcmd *testExecCmd) {
+	var stdout, stderr bytes.Buffer
+	var copyDone sync.WaitGroup
 
-func (_ *bufferCloser) Close() error { return nil }
+	stdoutPr, stdoutPw, err := os.Pipe()
+	require.NoError(t, err)
 
-func configureTLogging(cmd *ExecCommand) (stdout bufferCloser, stderr bufferCloser) {
-	cmd.SetWriters(&stdout, &stderr)
+	stderrPr, stderrPw, err := os.Pipe()
+	require.NoError(t, err)
+
+	copyDone.Add(2)
+	go func() {
+		defer copyDone.Done()
+		io.Copy(&stdout, stdoutPr)
+	}()
+	go func() {
+		defer copyDone.Done()
+		io.Copy(&stderr, stderrPr)
+	}()
+
+	testcmd.stdout = &stdout
+	testcmd.stderr = &stderr
+	testcmd.outputCopyDone = &copyDone
+
+	testcmd.command.stdout = stdoutPw
+	testcmd.command.stderr = stderrPw
 	return
 }
 
@@ -99,7 +132,8 @@ func TestExecutor_Start_Invalid(pt *testing.T) {
 	for name, factory := range executorFactories {
 		pt.Run(name, func(t *testing.T) {
 			require := require.New(t)
-			execCmd, allocDir := testExecutorCommand(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 			execCmd.Cmd = invalid
 			execCmd.Args = []string{"1"}
 			factory.configureExecCmd(t, execCmd)
@@ -118,7 +152,8 @@ func TestExecutor_Start_Wait_Failure_Code(pt *testing.T) {
 	for name, factory := range executorFactories {
 		pt.Run(name, func(t *testing.T) {
 			require := require.New(t)
-			execCmd, allocDir := testExecutorCommand(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 			execCmd.Cmd = "/bin/date"
 			execCmd.Args = []string{"fail"}
 			factory.configureExecCmd(t, execCmd)
@@ -141,7 +176,8 @@ func TestExecutor_Start_Wait(pt *testing.T) {
 	for name, factory := range executorFactories {
 		pt.Run(name, func(t *testing.T) {
 			require := require.New(t)
-			execCmd, allocDir := testExecutorCommand(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 			execCmd.Cmd = "/bin/echo"
 			execCmd.Args = []string{"hello world"}
 			factory.configureExecCmd(t, execCmd)
@@ -160,9 +196,7 @@ func TestExecutor_Start_Wait(pt *testing.T) {
 
 			expected := "hello world"
 			tu.WaitForResult(func() (bool, error) {
-				outWriter, _ := execCmd.GetWriters()
-				output := outWriter.(*bufferCloser).String()
-				act := strings.TrimSpace(string(output))
+				act := strings.TrimSpace(string(testExecCmd.stdout.String()))
 				if expected != act {
 					return false, fmt.Errorf("expected: '%s' actual: '%s'", expected, act)
 				}
@@ -174,12 +208,52 @@ func TestExecutor_Start_Wait(pt *testing.T) {
 	}
 }
 
+func TestExecutor_Start_Wait_Children(pt *testing.T) {
+	pt.Parallel()
+	for name, factory := range executorFactories {
+		pt.Run(name, func(t *testing.T) {
+			require := require.New(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
+			execCmd.Cmd = "/bin/sh"
+			execCmd.Args = []string{"-c", "(sleep 30 > /dev/null & ) ; exec sleep 1"}
+			factory.configureExecCmd(t, execCmd)
+
+			defer allocDir.Destroy()
+			executor := factory.new(testlog.HCLogger(t))
+			defer executor.Shutdown("SIGKILL", 0)
+
+			ps, err := executor.Launch(execCmd)
+			require.NoError(err)
+			require.NotZero(ps.Pid)
+
+			ch := make(chan error)
+
+			go func() {
+				ps, err = executor.Wait(context.Background())
+				t.Logf("Processe completed with %#v error: %#v", ps, err)
+				ch <- err
+			}()
+
+			timeout := 7 * time.Second
+			select {
+			case <-ch:
+				require.NoError(err)
+				//good
+			case <-time.After(timeout):
+				require.Fail(fmt.Sprintf("process is running after timeout: %v", timeout))
+			}
+		})
+	}
+}
+
 func TestExecutor_WaitExitSignal(pt *testing.T) {
 	pt.Parallel()
 	for name, factory := range executorFactories {
 		pt.Run(name, func(t *testing.T) {
 			require := require.New(t)
-			execCmd, allocDir := testExecutorCommand(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 			execCmd.Cmd = "/bin/sleep"
 			execCmd.Args = []string{"10000"}
 			execCmd.ResourceLimits = true
@@ -233,7 +307,8 @@ func TestExecutor_Start_Kill(pt *testing.T) {
 	for name, factory := range executorFactories {
 		pt.Run(name, func(t *testing.T) {
 			require := require.New(t)
-			execCmd, allocDir := testExecutorCommand(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 			execCmd.Cmd = "/bin/sleep"
 			execCmd.Args = []string{"10"}
 			factory.configureExecCmd(t, execCmd)
@@ -249,8 +324,7 @@ func TestExecutor_Start_Kill(pt *testing.T) {
 			require.NoError(executor.Shutdown("SIGINT", 100*time.Millisecond))
 
 			time.Sleep(time.Duration(tu.TestMultiplier()*2) * time.Second)
-			outWriter, _ := execCmd.GetWriters()
-			output := outWriter.(*bufferCloser).String()
+			output := testExecCmd.stdout.String()
 			expected := ""
 			act := strings.TrimSpace(string(output))
 			if act != expected {
@@ -263,7 +337,8 @@ func TestExecutor_Start_Kill(pt *testing.T) {
 func TestExecutor_Shutdown_Exit(t *testing.T) {
 	require := require.New(t)
 	t.Parallel()
-	execCmd, allocDir := testExecutorCommand(t)
+	testExecCmd := testExecutorCommand(t)
+	execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 	execCmd.Cmd = "/bin/sleep"
 	execCmd.Args = []string{"100"}
 	cfg := &ExecutorConfig{
@@ -400,7 +475,8 @@ func TestExecutor_Start_Kill_Immediately_NoGrace(pt *testing.T) {
 	for name, factory := range executorFactories {
 		pt.Run(name, func(t *testing.T) {
 			require := require.New(t)
-			execCmd, allocDir := testExecutorCommand(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 			execCmd.Cmd = "/bin/sleep"
 			execCmd.Args = []string{"100"}
 			factory.configureExecCmd(t, execCmd)
@@ -435,7 +511,8 @@ func TestExecutor_Start_Kill_Immediately_WithGrace(pt *testing.T) {
 	for name, factory := range executorFactories {
 		pt.Run(name, func(t *testing.T) {
 			require := require.New(t)
-			execCmd, allocDir := testExecutorCommand(t)
+			testExecCmd := testExecutorCommand(t)
+			execCmd, allocDir := testExecCmd.command, testExecCmd.allocDir
 			execCmd.Cmd = "/bin/sleep"
 			execCmd.Args = []string{"100"}
 			factory.configureExecCmd(t, execCmd)