Adjust logger types (#3859)

agent/logger.go

@@ -33,7 +33,9 @@ const (
 )
 
 func (r *Runner) createLogger(_logger zerolog.Logger, uploads *sync.WaitGroup, workflow *rpc.Workflow) pipeline.Logger {
-	return func(step *backend.Step, rc io.Reader) error {
+	return func(step *backend.Step, rc io.ReadCloser) error {
+		defer rc.Close()
+
 		logger := _logger.With().
 			Str("image", step.Image).
 			Str("workflow_id", workflow.ID).

cli/exec/exec.go

@@ -277,7 +277,7 @@ func convertPathForWindows(path string) string {
 }
 
 const maxLogLineLength = 1024 * 1024 // 1mb
-var defaultLogger = pipeline.Logger(func(step *backendTypes.Step, rc io.Reader) error {
+var defaultLogger = pipeline.Logger(func(step *backendTypes.Step, rc io.ReadCloser) error {
 	logWriter := NewLineWriter(step.Name, step.UUID)
 	return pipelineLog.CopyLineByLine(logWriter, rc, maxLogLineLength)
 })

pipeline/log/line_writer.go

@@ -40,7 +40,7 @@ type LineWriter struct {
 }
 
 // NewLineWriter returns a new line reader.
-func NewLineWriter(peer rpc.Peer, stepUUID string, secret ...string) io.WriteCloser {
+func NewLineWriter(peer rpc.Peer, stepUUID string, secret ...string) io.Writer {
 	lw := &LineWriter{
 		peer:      peer,
 		stepUUID:  stepUUID,
@@ -73,7 +73,3 @@ func (w *LineWriter) Write(p []byte) (n int, err error) {
 
 	return len(data), nil
 }
-
-func (w *LineWriter) Close() error {
-	return nil
-}

pipeline/log/line_writer_test.go

@@ -31,7 +31,6 @@ func TestLineWriter(t *testing.T) {
 
 	secrets := []string{"world"}
 	lw := log.NewLineWriter(peer, "e9ea76a5-44a1-4059-9c4a-6956c478b26d", secrets...)
-	defer lw.Close()
 
 	_, err := lw.Write([]byte("hello world\n"))
 	assert.NoError(t, err)

pipeline/log/utils.go

@@ -20,7 +20,7 @@ import (
 	"io"
 )
 
-func writeChunks(dst io.WriteCloser, data []byte, size int) error {
+func writeChunks(dst io.Writer, data []byte, size int) error {
 	if len(data) <= size {
 		_, err := dst.Write(data)
 		return err
@@ -41,9 +41,8 @@ func writeChunks(dst io.WriteCloser, data []byte, size int) error {
 	return nil
 }
 
-func CopyLineByLine(dst io.WriteCloser, src io.Reader, maxSize int) error {
+func CopyLineByLine(dst io.Writer, src io.Reader, maxSize int) error {
 	r := bufio.NewReader(src)
-	defer dst.Close()
 
 	for {
 		// TODO: read til newline or maxSize directly

pipeline/log/utils_test.go

@@ -144,3 +144,18 @@ func TestCopyLineByLineSizeLimit(t *testing.T) {
 
 	wg.Wait()
 }
+
+func TestStringReader(t *testing.T) {
+	r := io.NopCloser(strings.NewReader("123\n4567\n890"))
+
+	testWriter := &testWriter{
+		Mutex:  &sync.Mutex{},
+		writes: make([]string, 0),
+	}
+
+	err := log.CopyLineByLine(testWriter, r, 1024)
+	assert.NoError(t, err)
+
+	writes := testWriter.GetWrites()
+	assert.Lenf(t, writes, 3, "expected 3 writes, got: %v", writes)
+}

pipeline/logger.go

@@ -21,4 +21,4 @@ import (
 )
 
 // Logger handles the process logging.
-type Logger func(*backend.Step, io.Reader) error
+type Logger func(*backend.Step, io.ReadCloser) error