runservice drivers: define ExecConfig AttachStdin

internal/services/executor/driver/docker.go

@@ -395,7 +395,7 @@ func (dp *DockerPod) Exec(ctx context.Context, execConfig *ExecConfig) (Containe
 		Env:          makeEnvSlice(execConfig.Env),
 		Tty:          execConfig.Tty,
 		WorkingDir:   execConfig.WorkingDir,
-		AttachStdin:  true,
+		AttachStdin:  execConfig.AttachStdin,
 		AttachStdout: execConfig.Stdout != nil,
 		AttachStderr: execConfig.Stderr != nil,
 		User:         execConfig.User,
@@ -451,11 +451,18 @@ func (e *DockerContainerExec) Wait(ctx context.Context) (int, error) {
 	// ignore error, we'll use the exit code of the exec
 	<-e.endCh
 
+	var exitCode int
+	for {
 		resp, err := e.client.ContainerExecInspect(ctx, e.execID)
 		if err != nil {
 			return -1, err
 		}
-	exitCode := resp.ExitCode
+		if !resp.Running {
+			exitCode = resp.ExitCode
+			break
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
 
 	e.hresp.Close()
 

internal/services/executor/driver/docker_test.go

@@ -137,7 +137,6 @@ func TestDockerPod(t *testing.T) {
 			t.Fatalf("unexpected err: %v", err)
 		}
 
-		ce.Stdin().Close()
 		code, err := ce.Wait(ctx)
 		if err != nil {
 			t.Fatalf("unexpected err: %v", err)
@@ -180,7 +179,6 @@ func TestDockerPod(t *testing.T) {
 			t.Fatalf("unexpected err: %v", err)
 		}
 
-		ce.Stdin().Close()
 		code, err := ce.Wait(ctx)
 		if err != nil {
 			t.Fatalf("unexpected err: %v", err)

internal/services/executor/driver/driver.go

@@ -100,6 +100,7 @@ type ExecConfig struct {
 	Env         map[string]string
 	WorkingDir  string
 	User        string
+	AttachStdin bool
 	Stdout      io.Writer
 	Stderr      io.Writer
 	Tty         bool

internal/services/executor/driver/k8s.go

@@ -670,7 +670,7 @@ func (p *K8sPod) Exec(ctx context.Context, execConfig *ExecConfig) (ContainerExe
 		VersionedParams(&corev1.PodExecOptions{
 			Container: mainContainerName,
 			Command:   cmd,
-			Stdin:     true,
+			Stdin:     execConfig.AttachStdin,
 			Stdout:    execConfig.Stdout != nil,
 			Stderr:    execConfig.Stderr != nil,
 			TTY:       execConfig.Tty,
@@ -683,9 +683,14 @@ func (p *K8sPod) Exec(ctx context.Context, execConfig *ExecConfig) (ContainerExe
 
 	reader, writer := io.Pipe()
 
+	var stdin io.Reader
+	if execConfig.AttachStdin {
+		stdin = reader
+	}
+
 	go func() {
 		err := exec.Stream(remotecommand.StreamOptions{
-			Stdin:  reader,
+			Stdin:  stdin,
 			Stdout: execConfig.Stdout,
 			Stderr: execConfig.Stderr,
 			Tty:    execConfig.Tty,

internal/services/executor/driver/k8s_test.go

@@ -81,14 +81,15 @@ func TestK8sPod(t *testing.T) {
 		}
 		defer pod.Remove(ctx)
 
+		var buf bytes.Buffer
 		ce, err := pod.Exec(ctx, &ExecConfig{
 			Cmd:    []string{"ls"},
+			Stdout: &buf,
 		})
 		if err != nil {
 			t.Fatalf("unexpected err: %v", err)
 		}
 
-		ce.Stdin().Close()
 		code, err := ce.Wait(ctx)
 		if err != nil {
 			t.Fatalf("unexpected err: %v", err)
@@ -131,7 +132,6 @@ func TestK8sPod(t *testing.T) {
 			t.Fatalf("unexpected err: %v", err)
 		}
 
-		ce.Stdin().Close()
 		code, err := ce.Wait(ctx)
 		if err != nil {
 			t.Fatalf("unexpected err: %v", err)

internal/services/executor/executor.go

@@ -71,6 +71,7 @@ func (e *Executor) createFile(ctx context.Context, pod driver.Pod, command, user
 	var buf bytes.Buffer
 	execConfig := &driver.ExecConfig{
 		Cmd:         cmd,
+		AttachStdin: true,
 		Stdout:      &buf,
 		Stderr:      outf,
 		User:        user,
@@ -165,6 +166,7 @@ func (e *Executor) doRunStep(ctx context.Context, s *types.RunStep, t *types.Exe
 		Env:         environment,
 		WorkingDir:  workingDir,
 		User:        user,
+		AttachStdin: true,
 		Stdout:      outf,
 		Stderr:      outf,
 		Tty:         true,
@@ -214,6 +216,7 @@ func (e *Executor) doSaveToWorkspaceStep(ctx context.Context, s *types.SaveToWor
 		Cmd:         cmd,
 		Env:         t.Environment,
 		WorkingDir:  workingDir,
+		AttachStdin: true,
 		Stdout:      archivef,
 		Stderr:      logf,
 	}
@@ -273,6 +276,7 @@ func (e *Executor) expandDir(ctx context.Context, t *types.ExecutorTask, pod dri
 	execConfig := &driver.ExecConfig{
 		Cmd:         cmd,
 		Env:         t.Environment,
+		AttachStdin: true,
 		Stdout:      stdout,
 		Stderr:      logf,
 	}
@@ -300,6 +304,7 @@ func (e *Executor) mkdir(ctx context.Context, t *types.ExecutorTask, pod driver.
 	execConfig := &driver.ExecConfig{
 		Cmd:         cmd,
 		Env:         t.Environment,
+		AttachStdin: true,
 		Stdout:      logf,
 		Stderr:      logf,
 	}
@@ -336,6 +341,7 @@ func (e *Executor) template(ctx context.Context, t *types.ExecutorTask, pod driv
 		Cmd:         cmd,
 		Env:         t.Environment,
 		WorkingDir:  workingDir,
+		AttachStdin: true,
 		Stdout:      stdout,
 		Stderr:      logf,
 	}
@@ -382,6 +388,7 @@ func (e *Executor) unarchive(ctx context.Context, t *types.ExecutorTask, source
 		Cmd:         cmd,
 		Env:         t.Environment,
 		WorkingDir:  workingDir,
+		AttachStdin: true,
 		Stdout:      logf,
 		Stderr:      logf,
 	}
@@ -499,6 +506,7 @@ func (e *Executor) doSaveCacheStep(ctx context.Context, s *types.SaveCacheStep,
 		Cmd:         cmd,
 		Env:         t.Environment,
 		WorkingDir:  workingDir,
+		AttachStdin: true,
 		Stdout:      archivef,
 		Stderr:      logf,
 	}