text-only output: log stdout/stderr of steps (#597)

This is part of sourcegraph/sourcegraph#23496 and adds the logging of STDOUT/STDERR of steps to the -text-only output. Please see the comment I left for more info on how the interface could be improved/extended.

Also: turns out I don't need the process.Logger that was extracted in sourcegraph/sourcegraph#24545 and instead what's needed is the IntervalLogger here that flushes the captured output with a delay (so we don't create a new huge JSON message for every line)

Author: mrnugget

go.mod

@@ -5,6 +5,8 @@ go 1.17
 require (
 	github.com/Masterminds/semver v1.5.0
 	github.com/cockroachdb/errors v1.8.6
+	github.com/cockroachdb/redact v1.1.3 // indirect
+	github.com/derision-test/glock v0.0.0-20210316032053-f5b74334bb29
 	github.com/dustin/go-humanize v1.0.0
 	github.com/gobwas/glob v0.2.3
 	github.com/google/go-cmp v0.5.6
@@ -19,7 +21,9 @@ require (
 	github.com/sourcegraph/batch-change-utils v0.0.0-20210708162152-c9f35b905d94
 	github.com/sourcegraph/go-diff v0.6.1
 	github.com/sourcegraph/jsonx v0.0.0-20200629203448-1a936bd500cf
-	github.com/sourcegraph/sourcegraph/lib v0.0.0-20210902215418-71593bf836f9
+	github.com/sourcegraph/sourcegraph/lib v0.0.0-20210903080558-3f16affc1edc
+	github.com/ssor/bom v0.0.0-20170718123548-6386211fdfcf // indirect
+	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
 	golang.org/x/net v0.0.0-20210614182718-04defd469f4e
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 	jaytaylor.com/html2text v0.0.0-20200412013138-3577fbdbcff7
@@ -28,7 +32,6 @@ require (
 require (
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/cockroachdb/logtags v0.0.0-20190617123548-eb05cc24525f // indirect
-	github.com/cockroachdb/redact v1.1.3 // indirect
 	github.com/cockroachdb/sentry-go v0.6.1-cockroachdb.2 // indirect
 	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
@@ -46,8 +49,6 @@ require (
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/rivo/uniseg v0.1.0 // indirect
 	github.com/rogpeppe/go-internal v1.8.0 // indirect
-	github.com/ssor/bom v0.0.0-20170718123548-6386211fdfcf // indirect
-	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
 	github.com/xeipuuv/gojsonschema v1.2.0 // indirect
 	golang.org/x/sys v0.0.0-20210616094352-59db8d763f22 // indirect

go.sum

@@ -43,6 +43,8 @@ github.com/dave/jennifer v1.4.1/go.mod h1:7jEdnm+qBcxl8PC0zyp7vxcpSRnzXSt9r39tpT
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/derision-test/glock v0.0.0-20210316032053-f5b74334bb29 h1:O07j7ewg0eFHRx+7bxCA/7z86HJexl7HLY9kyVMmPDo=
+github.com/derision-test/glock v0.0.0-20210316032053-f5b74334bb29/go.mod h1:jKtLdBMrF+XQatqvg46wiWdDfDSSDjdhO4dOM2FX9H4=
 github.com/derision-test/go-mockgen v1.1.2/go.mod h1:9H3VGTWYnL1VJoHHCuPKDpPFmNQ1uVyNlpX6P63l5Sk=
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
@@ -257,8 +259,8 @@ github.com/sourcegraph/go-diff v0.6.1 h1:hmA1LzxW0n1c3Q4YbrFgg4P99GSnebYa3x8gr0H
 github.com/sourcegraph/go-diff v0.6.1/go.mod h1:iBszgVvyxdc8SFZ7gm69go2KDdt3ag071iBaWPF6cjs=
 github.com/sourcegraph/jsonx v0.0.0-20200629203448-1a936bd500cf h1:oAdWFqhStsWiiMP/vkkHiMXqFXzl1XfUNOdxKJbd6bI=
 github.com/sourcegraph/jsonx v0.0.0-20200629203448-1a936bd500cf/go.mod h1:ppFaPm6kpcHnZGqQTFhUIAQRIEhdQDWP1PCv4/ON354=
-github.com/sourcegraph/sourcegraph/lib v0.0.0-20210902215418-71593bf836f9 h1:afOQUKamJAosvaIgAJKFrJ7BbQ06BxJLlNU2Z+wfzOc=
-github.com/sourcegraph/sourcegraph/lib v0.0.0-20210902215418-71593bf836f9/go.mod h1:ZK8mHKWdapYI8iIQwsO7QUxf44RjiNGKkOZa9aYIjTc=
+github.com/sourcegraph/sourcegraph/lib v0.0.0-20210903080558-3f16affc1edc h1:iMV2Nc/fCn3PrG9vgFgXN2LdY6mDfcGuzQe9ObCT/U4=
+github.com/sourcegraph/sourcegraph/lib v0.0.0-20210903080558-3f16affc1edc/go.mod h1:ZK8mHKWdapYI8iIQwsO7QUxf44RjiNGKkOZa9aYIjTc=
 github.com/sourcegraph/yaml v1.0.1-0.20200714132230-56936252f152 h1:z/MpntplPaW6QW95pzcAR/72Z5TWDyDnSo0EOcyij9o=
 github.com/sourcegraph/yaml v1.0.1-0.20200714132230-56936252f152/go.mod h1:GIjDIg/heH5DOkXY3YJ/wNhfHsQHoXGjl8G8amsYQ1I=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
@@ -274,6 +276,7 @@ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXf
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
@@ -283,8 +286,9 @@ github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyC
 github.com/valyala/fasthttp v1.6.0/go.mod h1:FstJa9V+Pj9vQ7OJie2qMHdwemEDaDiSdBnvPM1Su9w=
 github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
-github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f h1:J9EGpcZtP0E/raorCMxlFGSTBrsSlaDGf3jU/qvAE2c=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
+github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo=
+github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=

internal/batches/executor/coordinator.go

@@ -2,14 +2,14 @@ package executor
 
 import (
 	"context"
+	"io"
 	"reflect"
 	"strconv"
 	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/hashicorp/go-multierror"
 	batcheslib "github.com/sourcegraph/sourcegraph/lib/batches"
-
 	"github.com/sourcegraph/src-cli/internal/api"
 	"github.com/sourcegraph/src-cli/internal/batches"
 	"github.com/sourcegraph/src-cli/internal/batches/docker"
@@ -217,6 +217,9 @@ type TaskExecutionUI interface {
 
 	// TODO: This should be split up into methods that are more specific.
 	TaskCurrentlyExecuting(*Task, string)
+
+	StepStdoutWriter(context.Context, *Task, int) io.WriteCloser
+	StepStderrWriter(context.Context, *Task, int) io.WriteCloser
 }
 
 // Execute executes the given Tasks and the importChangeset statements in the

internal/batches/executor/coordinator_test.go

@@ -3,6 +3,7 @@ package executor
 import (
 	"context"
 	"fmt"
+	"io"
 	"strings"
 	"sync"
 	"testing"
@@ -543,7 +544,19 @@ func (d *dummyTaskExecutionUI) TaskChangesetSpecsBuilt(t *Task, specs []*batches
 	d.specs[t] = specs
 }
 
+type discardCloser struct {
+	io.Writer
+}
+
+func (discardCloser) Close() error { return nil }
+
 func (d *dummyTaskExecutionUI) TaskCurrentlyExecuting(*Task, string) {}
+func (d *dummyTaskExecutionUI) StepStdoutWriter(ctx context.Context, task *Task, step int) io.WriteCloser {
+	return discardCloser{io.Discard}
+}
+func (d *dummyTaskExecutionUI) StepStderrWriter(ctx context.Context, task *Task, step int) io.WriteCloser {
+	return discardCloser{io.Discard}
+}
 
 var _ taskExecutor = &dummyExecutor{}
 

internal/batches/executor/executor.go

@@ -177,6 +177,8 @@ func (x *executor) do(ctx context.Context, task *Task, ui TaskExecutionUI) (err
 		reportProgress: func(currentlyExecuting string) {
 			ui.TaskCurrentlyExecuting(task, currentlyExecuting)
 		},
+		newUiStdoutWriter: ui.StepStdoutWriter,
+		newUiStderrWriter: ui.StepStderrWriter,
 	}
 
 	result, stepResults, err := runSteps(runCtx, opts)
@@ -191,7 +193,6 @@ func (x *executor) do(ctx context.Context, task *Task, ui TaskExecutionUI) (err
 
 	return nil
 }
-
 func (x *executor) addResult(task *Task, result executionResult, stepResults []stepExecutionResult) {
 	x.resultsMu.Lock()
 	defer x.resultsMu.Unlock()

internal/batches/executor/run_steps.go

@@ -22,6 +22,8 @@ import (
 	"github.com/sourcegraph/src-cli/internal/batches/util"
 	"github.com/sourcegraph/src-cli/internal/batches/workspace"
 
+	"github.com/sourcegraph/sourcegraph/lib/process"
+
 	yamlv3 "gopkg.in/yaml.v3"
 )
 
@@ -65,6 +67,9 @@ type executionOpts struct {
 
 	logger         log.TaskLogger
 	reportProgress func(string)
+
+	newUiStdoutWriter func(context.Context, *Task, int) io.WriteCloser
+	newUiStderrWriter func(context.Context, *Task, int) io.WriteCloser
 }
 
 func runSteps(ctx context.Context, opts *executionOpts) (result executionResult, stepResults []stepExecutionResult, err error) {
@@ -310,14 +315,29 @@ func executeSingleStep(
 
 	var stdoutBuffer, stderrBuffer bytes.Buffer
 
-	cmd.Stdout = io.MultiWriter(&stdoutBuffer, opts.logger.PrefixWriter("stdout"))
-	cmd.Stderr = io.MultiWriter(&stderrBuffer, opts.logger.PrefixWriter("stderr"))
+	writerCtx, writerCancel := context.WithCancel(ctx)
+	defer writerCancel()
+	uiStdoutWriter := opts.newUiStdoutWriter(writerCtx, opts.task, i)
+	uiStderrWriter := opts.newUiStderrWriter(writerCtx, opts.task, i)
+	defer func() {
+		uiStdoutWriter.Close()
+		uiStderrWriter.Close()
+	}()
+
+	stdout := io.MultiWriter(&stdoutBuffer, uiStdoutWriter, opts.logger.PrefixWriter("stdout"))
+	stderr := io.MultiWriter(&stderrBuffer, uiStderrWriter, opts.logger.PrefixWriter("stderr"))
+
+	wg, err := process.PipeOutput(cmd, stdout, stderr)
+	if err != nil {
+		return bytes.Buffer{}, bytes.Buffer{}, errors.Wrap(err, "piping process output")
+	}
 
 	opts.logger.Logf("[Step %d] run: %q, container: %q", i+1, step.Run, step.Container)
 	opts.logger.Logf("[Step %d] full command: %q", i+1, strings.Join(cmd.Args, " "))
 
 	t0 := time.Now()
 	err = cmd.Run()
+	wg.Wait()
 	elapsed := time.Since(t0).Round(time.Millisecond)
 	if err != nil {
 		opts.logger.Logf("[Step %d] took %s; error running Docker container: %+v", i+1, elapsed, err)

internal/batches/ui/interval_writer.go

@@ -0,0 +1,105 @@
+package ui
+
+import (
+	"bytes"
+	"context"
+	"time"
+
+	"github.com/derision-test/glock"
+)
+
+// IntervalWriter is a io.Writer that flushes to the given sink on the given
+// interval.
+type IntervalWriter struct {
+	sink func(string)
+
+	ticker glock.Ticker
+
+	// buf is used to keep partial lines buffered before flushing them (either
+	// on the next newline or after tickDuration)
+	buf        *bytes.Buffer
+	writes     chan []byte
+	writesDone chan struct{}
+
+	closed chan struct{}
+	done   chan struct{}
+}
+
+func newIntervalWriter(ctx context.Context, ticker glock.Ticker, sink func(string)) *IntervalWriter {
+	l := &IntervalWriter{
+		sink:   sink,
+		ticker: ticker,
+
+		writes:     make(chan []byte),
+		writesDone: make(chan struct{}),
+
+		buf: &bytes.Buffer{},
+
+		closed: make(chan struct{}, 1),
+		done:   make(chan struct{}, 1),
+	}
+
+	go l.writeLines(ctx)
+
+	return l
+}
+
+// NewLogger returns a new Logger instance and spawns a goroutine in the
+// background that regularily flushed the logged output to the given sink.
+//
+// If the passed in ctx is canceled the goroutine will exit.
+func NewIntervalWriter(ctx context.Context, interval time.Duration, sink func(string)) *IntervalWriter {
+	return newIntervalWriter(ctx, glock.NewRealTicker(interval), sink)
+}
+
+func (l *IntervalWriter) flush() {
+	if l.buf.Len() == 0 {
+		return
+	}
+	l.sink(l.buf.String())
+	l.buf.Reset()
+}
+
+// Close flushes the
+func (l *IntervalWriter) Close() error {
+	l.closed <- struct{}{}
+	<-l.done
+	return nil
+}
+
+// Write handler of IntervalWriter.
+func (l *IntervalWriter) Write(p []byte) (int, error) {
+	l.writes <- p
+	<-l.writesDone
+	return len(p), nil
+}
+
+func (l *IntervalWriter) writeLines(ctx context.Context) {
+	defer func() {
+		l.flush()
+		l.ticker.Stop()
+		l.done <- struct{}{}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-l.closed:
+			return
+
+		case w, ok := <-l.writes:
+			if !ok {
+				return
+			}
+
+			if _, err := l.buf.Write(w); err != nil {
+				break
+			}
+			l.writesDone <- struct{}{}
+		case <-l.ticker.Chan():
+			l.flush()
+		}
+	}
+}

internal/batches/ui/interval_writer_test.go

@@ -0,0 +1,63 @@
+package ui
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/derision-test/glock"
+)
+
+func TestIntervalWriter(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	ch := make(chan string, 500)
+
+	sink := func(data string) {
+		ch <- data
+	}
+
+	ticker := glock.NewMockTicker(1 * time.Second)
+	writer := newIntervalWriter(ctx, ticker, sink)
+
+	writer.Write([]byte("1"))
+	select {
+	case <-ch:
+		t.Fatalf("ch has data")
+	default:
+	}
+
+	ticker.BlockingAdvance(1 * time.Second)
+
+	select {
+	case d := <-ch:
+		if d != "1" {
+			t.Fatalf("wrong data in sink")
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatalf("ch has NO data")
+	}
+
+	writer.Write([]byte("2"))
+	writer.Write([]byte("3"))
+	writer.Write([]byte("4"))
+	writer.Write([]byte("5"))
+
+	select {
+	case <-ch:
+		t.Fatalf("ch has data")
+	default:
+	}
+
+	cancel()
+	writer.Close()
+
+	select {
+	case d := <-ch:
+		if d != "2345" {
+			t.Fatalf("wrong data in sink")
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatalf("ch has NO data")
+	}
+}