refactor(build logs): improve build logging

improvements:
* let the caller handle streamed log lines within its own goroutine.
* stream build log to long-term storage (instead of loading it into
memory first).

Closes #7422

Signed-off-by: Max Goltzsche <mgoltzsche@cloudbees.com>

go.sum

@@ -791,6 +791,7 @@ github.com/jenkins-x/go-scm v1.5.141 h1:sd0A41HYU6i9Kfl5/jH3ldgq4VkSiwuW5+KwvRxT
 github.com/jenkins-x/go-scm v1.5.141/go.mod h1:PCT338UhP/pQ0IeEeMEf/hoLTYKcH7qjGEKd7jPkeYg=
 github.com/jenkins-x/golang-jenkins v0.0.0-20180919102630-65b83ad42314 h1:kyBMx/ucSV92S+umX/V6DDaPNynlFFOM9MGJWApltoU=
 github.com/jenkins-x/golang-jenkins v0.0.0-20180919102630-65b83ad42314/go.mod h1:C6j5HgwlHGjRU27W4XCs6jXksqYFo8OdBu+p44jqQeM=
+github.com/jenkins-x/jx v1.3.1119 h1:yArkuyBqpklDSWaYT8zg0nhdVe6uVQeZRGP2jc2tjI4=
 github.com/jenkins-x/jx-api v0.0.11 h1:MGLiiSj9VmlOfpSjzDRxq14LAwWgyCRvBHOgRZaCm94=
 github.com/jenkins-x/jx-api v0.0.11/go.mod h1:+UBXFtmDMcmQg/1aDi/HhMV9Ww0m9+p0L/ZiGAh3fgY=
 github.com/jenkins-x/jx-api v0.0.12 h1:kyq2azaV3Gh0dsKZv51tbWffhibYGY5I2Igd7ISIEG8=

pkg/cmd/controller/controller_build.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -1079,27 +1080,16 @@ func (o *ControllerBuildOptions) generateBuildLogURL(podInterface typedcorev1.Po
 		return "", errors.Wrap(err, "there was a problem obtaining one of the clients")
 	}
 
-	var logWriter logs.LogWriter
-	w := LongTermStorageLogWriter{
-		data:       []byte{},
-		kubeClient: kubeClient,
-		logMasker:  logMasker,
-	}
-	logWriter = &w
-
 	tektonLogger := logs.TektonLogger{
 		JXClient:     jx,
 		KubeClient:   kubeClient,
 		TektonClient: tektonClient,
 		Namespace:    ns,
-		LogWriter:    logWriter,
 	}
 
 	log.Logger().Debugf("Capturing running build logs for %s", activity.Name)
-	err = tektonLogger.GetRunningBuildLogs(activity, buildName, false)
-	if err != nil {
-		return "", errors.Wrapf(err, "there was a problem getting logs for build %s", buildName)
-	}
+	reader := streamMaskedRunningBuildLogs(&tektonLogger, activity, buildName, logMasker)
+	defer reader.Close()
 
 	if initGitCredentials {
 		gc := &credentials.StepGitCredentialsOptions{}
@@ -1114,13 +1104,36 @@ func (o *ControllerBuildOptions) generateBuildLogURL(podInterface typedcorev1.Po
 	}
 
 	log.Logger().Infof("storing logs for activity %s into storage at %s", activity.Name, fileName)
-	answer, err := coll.CollectData(w.data, fileName)
+	logURL, err := coll.CollectData(reader, fileName)
 	if err != nil {
 		log.Logger().Errorf("failed to store logs for activity %s into storage at %s: %s", activity.Name, fileName, err.Error())
-		return answer, err
+		return "", err
 	}
 	log.Logger().Infof("stored logs for activity %s into storage at %s", activity.Name, fileName)
-	return answer, nil
+
+	return logURL, nil
+}
+
+func streamMaskedRunningBuildLogs(tl *logs.TektonLogger, activity *v1.PipelineActivity, buildName string, logMasker *kube.LogMasker) io.ReadCloser {
+	ch := tl.GetRunningBuildLogs(activity, buildName, false)
+	reader, writer := io.Pipe()
+	go func() {
+		var err error
+		for l := range ch {
+			if err == nil {
+				line := l.Line
+				if logMasker != nil && l.ShouldMask {
+					line = logMasker.MaskLog(line)
+				}
+				_, err = writer.Write([]byte(line + "\n"))
+			}
+		}
+		if err == nil {
+			err = errors.Wrapf(tl.Err(), "getting logs for build %s", buildName)
+		}
+		writer.CloseWithError(err)
+	}()
+	return reader
 }
 
 // ensurePipelineActivityHasLabels older versions of controller build did not add labels properly

pkg/cmd/get/get_build_logs.go

@@ -156,13 +156,10 @@ func (o *GetBuildLogsOptions) getProwBuildLog(kubeClient kubernetes.Interface, t
 
 	if o.TektonLogger == nil {
 		o.TektonLogger = &logs.TektonLogger{
-			KubeClient:   kubeClient,
-			TektonClient: tektonClient,
-			JXClient:     jxClient,
-			Namespace:    ns,
-			LogWriter: &CLILogWriter{
-				CommonOptions: o.CommonOptions,
-			},
+			KubeClient:     kubeClient,
+			TektonClient:   tektonClient,
+			JXClient:       jxClient,
+			Namespace:      ns,
 			FailIfPodFails: o.FailIfPodFails,
 		}
 	}
@@ -232,38 +229,16 @@ func (o *GetBuildLogsOptions) getTektonLogs(kubeClient kubernetes.Interface, tek
 		if err != nil {
 			return false, err
 		}
-		return false, o.TektonLogger.StreamPipelinePersistentLogs(pa.Spec.BuildLogsURL, jxClient, ns, authSvc)
+		for line := range o.TektonLogger.StreamPipelinePersistentLogs(pa.Spec.BuildLogsURL, authSvc) {
+			fmt.Println(line.Line)
+		}
+		return false, o.TektonLogger.Err()
 	}
 
 	log.Logger().Infof("Build logs for %s", util.ColorInfo(name))
 	name = strings.TrimSuffix(name, " ")
-	return false, o.TektonLogger.GetRunningBuildLogs(pa, name, false)
-}
-
-// StreamLog implementation of LogWriter.StreamLog for CLILogWriter, this implementation will tail logs for the provided pod /container through the defined logger
-func (o *CLILogWriter) StreamLog(lch <-chan logs.LogLine, ech <-chan error) error {
-	for {
-		select {
-		case l, ok := <-lch:
-			if !ok {
-				return nil
-			}
-			fmt.Println(l.Line)
-		case err := <-ech:
-			return err
-		}
+	for line := range o.TektonLogger.GetRunningBuildLogs(pa, name, false) {
+		fmt.Println(line.Line)
 	}
-}
-
-// WriteLog implementation of LogWriter.WriteLog for CLILogWriter, this implementation will write the provided log line through the defined logger
-func (o *CLILogWriter) WriteLog(logLine logs.LogLine, lch chan<- logs.LogLine) error {
-	lch <- logLine
-	return nil
-}
-
-// BytesLimit defines the limit of bytes to be used to fetch the logs from the kube API
-// defaulted to 0 for this implementation
-func (o *CLILogWriter) BytesLimit() int {
-	//We are not limiting bytes with this writer
-	return 0
+	return false, o.TektonLogger.Err()
 }

pkg/collector/bucket_collector.go

@@ -2,6 +2,7 @@ package collector
 
 import (
 	"bytes"
+	"io"
 	"io/ioutil"
 	"path/filepath"
 	"time"
@@ -67,8 +68,8 @@ func (c *BucketCollector) CollectFiles(patterns []string, outputPath string, bas
 }
 
 // CollectData collects the data storing it at the given output path and returning the URL to access it
-func (c *BucketCollector) CollectData(data []byte, outputName string) (string, error) {
-	url, err := c.provider.UploadFileToBucket(bytes.NewReader(data), outputName, c.bucketURL)
+func (c *BucketCollector) CollectData(data io.Reader, outputName string) (string, error) {
+	url, err := c.provider.UploadFileToBucket(data, outputName, c.bucketURL)
 	if err != nil {
 		return "", err
 	}

pkg/collector/bucket_collector_test.go

@@ -30,7 +30,7 @@ func TestBucketCollector_CollectData(t *testing.T) {
 		provider:  mp,
 	}
 
-	finalURL, err := collector.CollectData(contents, outputName)
+	finalURL, err := collector.CollectData(bytes.NewReader(contents), outputName)
 	assert.NoError(t, err)
 	assert.Equal(t, fmt.Sprintf("%s/%s", collector.bucketURL, outputName), finalURL)
 }

pkg/collector/git_collector.go

@@ -2,6 +2,7 @@ package collector
 
 import (
 	"fmt"
+	"io"
 	"io/ioutil"
 	neturl "net/url"
 	"os"
@@ -115,7 +116,7 @@ func (c *GitCollector) CollectFiles(patterns []string, outputPath string, basedi
 
 // CollectData collects the data storing it at the given output path and returning the URL
 // to access it
-func (c *GitCollector) CollectData(data []byte, outputPath string) (string, error) {
+func (c *GitCollector) CollectData(data io.Reader, outputPath string) (string, error) {
 	u := ""
 	gitClient := c.gitter
 	storageGitInfo := c.gitInfo
@@ -135,9 +136,9 @@ func (c *GitCollector) CollectData(data []byte, outputPath string) (string, erro
 	if err != nil {
 		return u, errors.Wrapf(err, "failed to create directory file %s", toDir)
 	}
-	err = ioutil.WriteFile(toFile, data, util.DefaultWritePermissions)
+	err = writeFile(toFile, data)
 	if err != nil {
-		return u, errors.Wrapf(err, "failed to write file %s", toFile)
+		return u, errors.Wrap(err, "write temp log file")
 	}
 
 	u = c.generateURL(storageOrg, storageRepoName, outputPath)
@@ -161,6 +162,20 @@ func (c *GitCollector) CollectData(data []byte, outputPath string) (string, erro
 	return u, err
 }
 
+func writeFile(filePath string, data io.Reader) (err error) {
+	dest, err := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY, util.DefaultWritePermissions)
+	if err != nil {
+		return
+	}
+	defer func() {
+		if e := dest.Close(); e != nil && err == nil {
+			err = e
+		}
+	}()
+	_, err = io.Copy(dest, data)
+	return
+}
+
 func (c *GitCollector) generateURL(storageOrg string, storageRepoName string, rPath string) (url string) {
 	if !c.gitInfo.IsGitHub() && c.gitKind == gits.KindGitHub {
 		url = fmt.Sprintf("https://%s/raw/%s/%s/%s/%s", c.gitInfo.Host, storageOrg, storageRepoName, c.gitBranch, rPath)

pkg/collector/interface.go

@@ -1,5 +1,9 @@
 package collector
 
+import (
+	"io"
+)
+
 // Collector an interface to collect data for storage in git or cloud storage etc
 type Collector interface {
 
@@ -9,5 +13,5 @@ type Collector interface {
 
 	// CollectData collects the data storing it at the given output path and returning the URL
 	// to access it
-	CollectData(data []byte, outputPath string) (string, error)
+	CollectData(data io.Reader, outputPath string) (string, error)
 }