Bitrise logging optimisation v2 (#295)

* Separate closing functions

* Move raw xcbuild logs after filter to ensure waiting for them

* Flush sinks after closing filter

* Swap order of closing the filter and the outputs

* Wait for filter done when closing sinks

* Use buffer pointer for xcoderaw

* Wait for filtering and formatter to close before using the raw output

* Add comments explaining the sync closes without defer

* Separate filter closing and flushing from closing the rest of the piping

* Wrap logging and error checking on filter close

* Fix copy paste in xcpretty logs

Author: Badlazzor

logio/pipe_wiring.go

@@ -3,89 +3,87 @@ package logio
 import (
 	"bytes"
 	"errors"
-	"fmt"
 	"io"
 	"os"
 	"regexp"
+	"sync"
 )
 
 // PipeWiring is a helper struct to define the setup and binding of tools and
 // xcbuild with a filter and stdout. It is purely boilerplate reduction and it is the
 // users responsibility to choose between this and manual hooking of the in/outputs.
 // It also provides a convenient Close() method that only closes things that can/should be closed.
 type PipeWiring struct {
-	XcbuildRawout bytes.Buffer
+	XcbuildRawout *bytes.Buffer
 	XcbuildStdout io.Writer
 	XcbuildStderr io.Writer
 	ToolStdin     io.ReadCloser
 	ToolStdout    io.WriteCloser
 	ToolStderr    io.WriteCloser
 
-	closer func() error
+	toolPipeW      *io.PipeWriter
+	bufferedStdout *Sink
+	toolInSink     *Sink
+	filter         *PrefixFilter
+
+	closeFilterOnce sync.Once
+}
+
+// CloseFilter closes the filter and waits for it to finish
+func (p *PipeWiring) CloseFilter() error {
+	err := error(nil)
+	p.closeFilterOnce.Do(func() {
+		err = p.filter.Close()
+		<-p.filter.Done()
+
+	})
+	return err
 }
 
-// Close closes the PipeWiring instances that needs to be closing as part of this instance.
-//
-// In reality it can only close the filter and the tool input as everything else is
-// managed by a command or the os.
-func (i *PipeWiring) Close() error {
-	return i.closer()
+// Close ...
+func (p *PipeWiring) Close() error {
+	filterErr := p.CloseFilter()
+	toolSinkErr := p.toolInSink.Close()
+	pipeWErr := p.toolPipeW.Close()
+	bufferedStdoutErr := p.bufferedStdout.Close()
+
+	return errors.Join(filterErr, toolSinkErr, pipeWErr, bufferedStdoutErr)
 }
 
 // SetupPipeWiring creates a new PipeWiring instance that contains the usual
 // input/outputs that an xcodebuild command and a logging tool needs when we are also
 // using a logging filter.
 func SetupPipeWiring(filter *regexp.Regexp) *PipeWiring {
 	// Create a buffer to store raw xcbuild output
-	var rawXcbuild bytes.Buffer
+	rawXcbuild := bytes.NewBuffer(nil)
 	// Pipe filtered logs to tool
 	toolPipeR, toolPipeW := io.Pipe()
 
 	// Add a buffer before stdout
 	bufferedStdout := NewSink(os.Stdout)
 	// Add a buffer before tool input
-	xcbuildLogs := NewSink(toolPipeW)
+	toolInSink := NewSink(toolPipeW)
+	xcbuildLogs := io.MultiWriter(rawXcbuild, toolInSink)
 	// Create a filter for [Bitrise ...] prefixes
 	bitrisePrefixFilter := NewPrefixFilter(
 		filter,
 		bufferedStdout,
 		xcbuildLogs,
 	)
 
-	// Send raw xcbuild out to raw out and filter
-	rawInputDuplication := io.MultiWriter(&rawXcbuild, bitrisePrefixFilter)
-
 	return &PipeWiring{
 		XcbuildRawout: rawXcbuild,
-		XcbuildStdout: rawInputDuplication,
-		XcbuildStderr: rawInputDuplication,
+		XcbuildStdout: bitrisePrefixFilter,
+		XcbuildStderr: bitrisePrefixFilter,
 		ToolStdin:     toolPipeR,
 		ToolStdout:    os.Stdout,
 		ToolStderr:    os.Stderr,
-		closer: func() error {
-			// XcbuildRawout - no need to close
-			// XcbuildStdout - Multiwriter, meaning we need to close the subwriters
-			// XcbuildStderr - Multiwriter, meaning we need to close the subwriters
-			// ToolStdout - We are not closing stdout
-			// ToolSterr - We are not closing stderr
-
-			var errStr string
-
-			if err := bitrisePrefixFilter.Close(); err != nil {
-				errStr += fmt.Sprintf("failed to close log filter, error: %s", err.Error())
-			}
-			if err := toolPipeW.Close(); err != nil {
-				if len(errStr) > 0 {
-					errStr += ", "
-				}
-				errStr += fmt.Sprintf("failed to close xcodebuild-xcpretty pipe, error: %s", err.Error())
-			}
 
-			if len(errStr) > 0 {
-				return errors.New(errStr)
-			}
+		toolPipeW:      toolPipeW,
+		bufferedStdout: bufferedStdout,
+		toolInSink:     toolInSink,
+		filter:         bitrisePrefixFilter,
 
-			return nil
-		},
+		closeFilterOnce: sync.Once{},
 	}
 }

logio/pipe_wiring_test.go

@@ -0,0 +1,30 @@
+package logio_test
+
+import (
+	"io"
+	"regexp"
+	"testing"
+
+	"github.com/bitrise-io/go-xcode/v2/logio"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPipeWiring(t *testing.T) {
+	sut := logio.SetupPipeWiring(regexp.MustCompile(`^\[Bitrise.*\].*`))
+
+	out := NewChanWriterCloser()
+	go func() {
+		_, _ = io.Copy(out, sut.ToolStdin)
+		_ = out.Close()
+	}()
+
+	_, _ = sut.XcbuildStdout.Write([]byte(msg1))
+	_, _ = sut.XcbuildStdout.Write([]byte(msg2))
+	_, _ = sut.XcbuildStdout.Write([]byte(msg3))
+	_, _ = sut.XcbuildStdout.Write([]byte(msg4))
+
+	_ = sut.Close()
+
+	assert.Equal(t, msg1+msg4, sut.XcbuildRawout.String())
+	assert.Equal(t, msg1+msg4, out.Messages())
+}

logio/prefix_filter.go

@@ -20,7 +20,7 @@ type PrefixFilter struct {
 	pipeW       *io.PipeWriter
 
 	Matching *Sink
-	Filtered *Sink
+	Filtered io.Writer
 
 	// closing
 	closeOnce sync.Once
@@ -48,7 +48,7 @@ func (p *PrefixFilter) ScannerError() <-chan error { return p.scannerError }
 // NewPrefixFilter returns a new PrefixFilter. Writes are based on line prefix.
 //
 // Note: Callers are responsible for closing intercepted and target writers that implement io.Closer
-func NewPrefixFilter(prefixRegexp *regexp.Regexp, matching, filtered *Sink) *PrefixFilter {
+func NewPrefixFilter(prefixRegexp *regexp.Regexp, matching *Sink, filtered io.Writer) *PrefixFilter {
 	// This is the backing field of the bufio.ReadWriter
 	pipeR, pipeW := io.Pipe()
 	messageLost := make(chan error, 1)

logio/prefix_filter_test.go

@@ -12,7 +12,7 @@ import (
 
 const (
 	msg1 = "Log message without prefix\n"
-	msg2 = "[Bitrise Analytics] Log message with prefixs\n"
+	msg2 = "[Bitrise Analytics] Log message with prefix\n"
 	msg3 = "[Bitrise Build Cache] Log message with prefix\n"
 	msg4 = "Stuff [Bitrise Build Cache] Log message without prefix\n"
 )

xcodecommand/xcbeautify.go

@@ -82,6 +82,11 @@ func (c *XcbeautifyRunner) Run(workDir string, xcodebuildArgs []string, xcbeauti
 		}
 	}
 
+	// Closing the filter to ensure all output is flushed and processed
+	if err := loggingIO.CloseFilter(); err != nil {
+		c.logger.Warnf("logging IO failure, error: %s", err)
+	}
+
 	return Output{
 		RawOut:   loggingIO.XcbuildRawout.Bytes(),
 		ExitCode: exitCode,

xcodecommand/xcpretty.go

@@ -88,6 +88,11 @@ func (c *XcprettyCommandRunner) Run(workDir string, xcodebuildArgs []string, xcp
 		}
 	}
 
+	// Closing the filter to ensure all output is flushed and processed
+	if err := loggingIO.CloseFilter(); err != nil {
+		c.logger.Warnf("logging IO failure, error: %s", err)
+	}
+
 	return Output{
 		RawOut:   loggingIO.XcbuildRawout.Bytes(),
 		ExitCode: exitCode,

xcpretty/xcpretty.go

@@ -65,7 +65,7 @@ func (c CommandModel) Run() (string, error) {
 		Stderr: loggingIO.ToolStderr,
 	})
 
-	// Always close xcpretty outputs
+	// Wait for the filtering to finish
 	defer func() {
 		if err := loggingIO.Close(); err != nil {
 			fmt.Printf("logging IO failure, error: %s", err)
@@ -76,21 +76,32 @@ func (c CommandModel) Run() (string, error) {
 		}
 	}()
 
+	closeAndFlushFilter := func() {
+		// Closing the filter to ensure all output is flushed and processed
+		if err := loggingIO.CloseFilter(); err != nil {
+			fmt.Printf("logging IO failure, error: %s", err)
+		}
+	}
+
 	// Run
 	if err := xcodebuildCmd.Start(); err != nil {
+		closeAndFlushFilter()
 		out := loggingIO.XcbuildRawout.String()
 		return out, err
 	}
 	if err := prettyCmd.Start(); err != nil {
+		closeAndFlushFilter()
 		out := loggingIO.XcbuildRawout.String()
 		return out, err
 	}
 
 	if err := xcodebuildCmd.Wait(); err != nil {
+		closeAndFlushFilter()
 		out := loggingIO.XcbuildRawout.String()
 		return out, err
 	}
 
+	closeAndFlushFilter()
 	return loggingIO.XcbuildRawout.String(), nil
 }