feat: stream output for custom workflows

server/controllers/events/events_controller_e2e_test.go

@@ -973,8 +973,9 @@ func setupE2E(t *testing.T, repoDir string) (events_controllers.VCSEventsControl
 			TerraformExecutor: terraformClient,
 		},
 		RunStepRunner: &runtime.RunStepRunner{
-			TerraformExecutor: terraformClient,
-			DefaultTFVersion:  defaultTFVersion,
+			TerraformExecutor:       terraformClient,
+			DefaultTFVersion:        defaultTFVersion,
+			ProjectCmdOutputHandler: projectCmdOutputHandler,
 		},
 		WorkingDir:       workingDir,
 		Webhooks:         &mockWebhookSender{},

server/core/runtime/env_step_runner_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/runatlantis/atlantis/server/core/terraform/mocks"
 	"github.com/runatlantis/atlantis/server/events/command"
 	"github.com/runatlantis/atlantis/server/events/models"
+	jobmocks "github.com/runatlantis/atlantis/server/jobs/mocks"
 	"github.com/runatlantis/atlantis/server/logging"
 
 	. "github.com/petergtz/pegomock"
@@ -40,9 +41,11 @@ func TestEnvStepRunner_Run(t *testing.T) {
 	tfClient := mocks.NewMockClient()
 	tfVersion, err := version.NewVersion("0.12.0")
 	Ok(t, err)
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 	runStepRunner := runtime.RunStepRunner{
-		TerraformExecutor: tfClient,
-		DefaultTFVersion:  tfVersion,
+		TerraformExecutor:       tfClient,
+		DefaultTFVersion:        tfVersion,
+		ProjectCmdOutputHandler: projectCmdOutputHandler,
 	}
 	envRunner := runtime.EnvStepRunner{
 		RunStepRunner: &runStepRunner,

server/core/runtime/multienv_step_runner_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/runatlantis/atlantis/server/core/terraform/mocks"
 	"github.com/runatlantis/atlantis/server/events/command"
 	"github.com/runatlantis/atlantis/server/events/models"
+	jobmocks "github.com/runatlantis/atlantis/server/jobs/mocks"
 	"github.com/runatlantis/atlantis/server/logging"
 	. "github.com/runatlantis/atlantis/testing"
 )
@@ -31,9 +32,11 @@ func TestMultiEnvStepRunner_Run(t *testing.T) {
 	tfClient := mocks.NewMockClient()
 	tfVersion, err := version.NewVersion("0.12.0")
 	Ok(t, err)
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 	runStepRunner := runtime.RunStepRunner{
-		TerraformExecutor: tfClient,
-		DefaultTFVersion:  tfVersion,
+		TerraformExecutor:       tfClient,
+		DefaultTFVersion:        tfVersion,
+		ProjectCmdOutputHandler: projectCmdOutputHandler,
 	}
 	multiEnvStepRunner := runtime.MultiEnvStepRunner{
 		RunStepRunner: &runStepRunner,

server/core/runtime/run_step_runner.go

@@ -1,22 +1,29 @@
 package runtime
 
 import (
+	"bufio"
+	"bytes"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/hashicorp/go-version"
 	"github.com/runatlantis/atlantis/server/events/command"
+	"github.com/runatlantis/atlantis/server/events/terraform/ansi"
+	"github.com/runatlantis/atlantis/server/jobs"
 )
 
 // RunStepRunner runs custom commands.
 type RunStepRunner struct {
 	TerraformExecutor TerraformExec
 	DefaultTFVersion  *version.Version
 	// TerraformBinDir is the directory where Atlantis downloads Terraform binaries.
-	TerraformBinDir string
+	TerraformBinDir         string
+	ProjectCmdOutputHandler jobs.ProjectCommandOutputHandler
 }
 
 func (r *RunStepRunner) Run(ctx command.ProjectContext, command string, path string, envs map[string]string) (string, error) {
@@ -66,13 +73,45 @@ func (r *RunStepRunner) Run(ctx command.ProjectContext, command string, path str
 		finalEnvVars = append(finalEnvVars, fmt.Sprintf("%s=%s", key, val))
 	}
 	cmd.Env = finalEnvVars
-	out, err := cmd.CombinedOutput()
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		err = fmt.Errorf("%s: unable to create stdout buffer", err)
+		ctx.Log.Debug("error: %s", err)
+		return "", err
+	}
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		err = fmt.Errorf("%s: unable to create stderr buffer", err)
+		ctx.Log.Debug("error: %s", err)
+		return "", err
+	}
+	cmd.Start()
+
+	var output bytes.Buffer
+	var mutex sync.Mutex
+
+	go r.streamOutput(ctx, stdout, &output, &mutex)
+	go r.streamOutput(ctx, stderr, &output, &mutex)
+
+	err = cmd.Wait()
 
 	if err != nil {
-		err = fmt.Errorf("%s: running %q in %q: \n%s", err, command, path, out)
+		err = fmt.Errorf("%s: running %q in %q: \n%s", err, command, path, output.String())
 		ctx.Log.Debug("error: %s", err)
 		return "", err
 	}
 	ctx.Log.Info("successfully ran %q in %q", command, path)
-	return string(out), nil
+	return ansi.Strip(output.String()), nil
+}
+
+func (r RunStepRunner) streamOutput(ctx command.ProjectContext, reader io.Reader, buffer *bytes.Buffer, mutex *sync.Mutex) {
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		line := scanner.Text()
+		r.ProjectCmdOutputHandler.Send(ctx, line, false)
+		mutex.Lock()
+		buffer.WriteString(line)
+		buffer.WriteString("\n")
+		mutex.Unlock()
+	}
 }

server/core/runtime/run_step_runner_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/runatlantis/atlantis/server/events/command"
 	"github.com/runatlantis/atlantis/server/events/mocks/matchers"
 	"github.com/runatlantis/atlantis/server/events/models"
+	jobmocks "github.com/runatlantis/atlantis/server/jobs/mocks"
 	"github.com/runatlantis/atlantis/server/logging"
 	. "github.com/runatlantis/atlantis/testing"
 )
@@ -38,11 +39,11 @@ func TestRunStepRunner_Run(t *testing.T) {
 		},
 		{
 			Command: `printf \'your main.tf file does not provide default region.\\ncheck\'`,
-			ExpOut:  `'your`,
+			ExpOut:  "'your\n",
 		},
 		{
 			Command: `printf 'your main.tf file does not provide default region.\ncheck'`,
-			ExpOut:  "your main.tf file does not provide default region.\ncheck",
+			ExpOut:  "your main.tf file does not provide default region.\ncheck\n",
 		},
 		{
 			Command: "echo 'a",
@@ -104,11 +105,13 @@ func TestRunStepRunner_Run(t *testing.T) {
 			ThenReturn(nil)
 
 		logger := logging.NewNoopLogger(t)
+		projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 		r := runtime.RunStepRunner{
-			TerraformExecutor: terraform,
-			DefaultTFVersion:  defaultVersion,
-			TerraformBinDir:   "/bin/dir",
+			TerraformExecutor:       terraform,
+			DefaultTFVersion:        defaultVersion,
+			TerraformBinDir:         "/bin/dir",
+			ProjectCmdOutputHandler: projectCmdOutputHandler,
 		}
 		t.Run(c.Command, func(t *testing.T) {
 			tmpDir, cleanup := TempDir(t)

server/events/project_command_runner_test.go

@@ -30,6 +30,7 @@ import (
 	eventmocks "github.com/runatlantis/atlantis/server/events/mocks"
 	"github.com/runatlantis/atlantis/server/events/mocks/matchers"
 	"github.com/runatlantis/atlantis/server/events/models"
+	jobmocks "github.com/runatlantis/atlantis/server/jobs/mocks"
 	"github.com/runatlantis/atlantis/server/logging"
 	. "github.com/runatlantis/atlantis/testing"
 )
@@ -537,9 +538,11 @@ func TestDefaultProjectCommandRunner_RunEnvSteps(t *testing.T) {
 	tfClient := tmocks.NewMockClient()
 	tfVersion, err := version.NewVersion("0.12.0")
 	Ok(t, err)
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 	run := runtime.RunStepRunner{
-		TerraformExecutor: tfClient,
-		DefaultTFVersion:  tfVersion,
+		TerraformExecutor:       tfClient,
+		DefaultTFVersion:        tfVersion,
+		ProjectCmdOutputHandler: projectCmdOutputHandler,
 	}
 	env := runtime.EnvStepRunner{
 		RunStepRunner: &run,

server/server.go

@@ -466,9 +466,10 @@ func NewServer(userConfig UserConfig, config Config) (*Server, error) {
 	defaultTfVersion := terraformClient.DefaultVersion()
 	pendingPlanFinder := &events.DefaultPendingPlanFinder{}
 	runStepRunner := &runtime.RunStepRunner{
-		TerraformExecutor: terraformClient,
-		DefaultTFVersion:  defaultTfVersion,
-		TerraformBinDir:   terraformClient.TerraformBinDir(),
+		TerraformExecutor:       terraformClient,
+		DefaultTFVersion:        defaultTfVersion,
+		TerraformBinDir:         terraformClient.TerraformBinDir(),
+		ProjectCmdOutputHandler: projectCmdOutputHandler,
 	}
 	drainer := &events.Drainer{}
 	statusController := &controllers.StatusController{