add back tkn cli style per log line prefixing

rh-pre-commit.version: 2.2.0
rh-pre-commit.check-secrets: ENABLED

pkg/watcher/reconciler/dynamic/dynamic.go

@@ -15,10 +15,12 @@
 package dynamic
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"fmt"
 	"io"
+	"strings"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -346,7 +348,8 @@ func (r *Reconciler) sendLog(ctx context.Context, o results.Object) error {
 	return nil
 }
 
-func (r *Reconciler) getPodLogs(ctx context.Context, ns, pod, container string) ([]byte, error) {
+func (r *Reconciler) getPodLogs(ctx context.Context, ns, pod, container, pipeline, task string) ([]byte, error) {
+	logger := logging.FromContext(ctx)
 	podLogOpts := corev1.PodLogOptions{
 		Container: container,
 	}
@@ -365,9 +368,43 @@ func (r *Reconciler) getPodLogs(ctx context.Context, ns, pod, container string)
 	if err != nil {
 		return nil, err
 	}
+	logger.Infow("GGM starting new logging",
+		zap.String("ns", ns),
+		zap.String("pod", pod),
+		zap.String("container", container),
+		zap.String("pipeline", pipeline),
+		zap.String("task", task),
+	)
+	rdr := bufio.NewReader(podLogs)
 	buf := new(bytes.Buffer)
-	_, err = io.Copy(buf, podLogs)
-	return buf.Bytes(), err
+	for {
+		var line []byte
+		line, _, err = rdr.ReadLine()
+		logger.Infow("GGM readline return line",
+			zap.String("line", string(line)))
+		if err != nil && err == io.EOF {
+			logger.Infow("GGM readline eof")
+			break
+		}
+		if err != nil {
+			logger.Infow("GGM readline unexpect err",
+				zap.String("err", err.Error()))
+			return nil, err
+		}
+		s := ""
+		stepName := strings.TrimPrefix(container, "step-")
+		s = fmt.Sprintf("[%s : %s] %s", task, stepName, string(line))
+		if len(pipeline) == 0 {
+			s = fmt.Sprintf("[%s] %s", container, string(line))
+		}
+		logger.Infow("GGM readLine formatted line",
+			zap.String("line", s))
+		_, err = buf.Write([]byte(s))
+		if err != nil {
+			return nil, err
+		}
+	}
+	return buf.Bytes(), nil
 }
 
 func (r *Reconciler) streamLogs(ctx context.Context, o results.Object, labelKey, logName string) error {
@@ -401,12 +438,12 @@ func (r *Reconciler) streamLogs(ctx context.Context, o results.Object, labelKey,
 		copy(containers, pod.Spec.InitContainers)
 		containers = append(containers, pod.Spec.Containers...)
 		for _, container := range containers {
-			ba, podLogsErr := r.getPodLogs(ctx, o.GetNamespace(), pod.Name, container.Name)
+			pipelineName := pod.Labels[pipeline.PipelineLabelKey]
+			taskName := pod.Labels[pipeline.TaskLabelKey]
+			ba, podLogsErr := r.getPodLogs(ctx, o.GetNamespace(), pod.Name, container.Name, pipelineName, taskName)
 			if podLogsErr != nil {
 				return podLogsErr
 			}
-			hdr := fmt.Sprintf("*** Logs for pod %s container %s ***\n", pod.Name, container.Name)
-			inMemWriteBufferStdout.Write([]byte(hdr))
 			inMemWriteBufferStdout.Write(ba)
 		}
 	}

test/e2e/e2e_gcs_test.go

@@ -115,7 +115,7 @@ func TestGCSLog(t *testing.T) {
 				t.Logf("Error getting Log for %s: %v", logName, cerr)
 				return false, nil
 			}
-			want := "hello world!"
+			want := "[hello : hello] hello world!"
 			if log == nil {
 				t.Logf("Nil return from logClient.Recv()")
 				return false, nil