split kubernetes logs into multiple lines

runtime/line.go

@@ -46,19 +46,34 @@ func (w *lineWriter) Write(p []byte) (n int, err error) {
 		out = w.rep.Replace(out)
 	}
 
-	line := &Line{
-		Number:    w.num,
-		Message:   out,
-		Timestamp: int64(time.Since(w.now).Seconds()),
-	}
+	parts := []string{out}
 
-	if w.state.hook.GotLine != nil {
-		w.state.hook.GotLine(w.state, line)
+	// kubernetes buffers the output and may combine
+	// multiple lines into a single block of output.
+	// Split into multiple lines.
+	//
+	// note that docker output always inclines a line
+	// feed marker. This needs to be accounted for when
+	// splitting the output into multiple lines.
+	if strings.Contains(strings.TrimSuffix(out, "\n"), "\n") {
+		parts = strings.SplitAfter(out, "\n")
 	}
-	w.size = w.size + len(p)
-	w.num++
 
-	w.lines = append(w.lines, line)
+	for _, part := range parts {
+		line := &Line{
+			Number:    w.num,
+			Message:   part,
+			Timestamp: int64(time.Since(w.now).Seconds()),
+		}
+
+		if w.state.hook.GotLine != nil {
+			w.state.hook.GotLine(w.state, line)
+		}
+		w.size = w.size + len(part)
+		w.num++
+
+		w.lines = append(w.lines, line)
+	}
 
 	// if the write exceeds the maximum output we should
 	// write a single line to the end of the logs that

runtime/line_test.go

@@ -35,6 +35,32 @@ func TestLineWriter(t *testing.T) {
 	}
 }
 
+func TestLineWriterMulti(t *testing.T) {
+	var lines []*Line
+	hook := &Hook{}
+	state := &State{}
+
+	hook.GotLine = func(_ *State, l *Line) error {
+		lines = append(lines, l)
+		return nil
+	}
+	state.hook = hook
+	state.Step = &engine.Step{}
+	state.config = &engine.Spec{}
+
+	newWriter(state).Write([]byte("foo\nbar\nbaz"))
+
+	if len(lines) != 3 {
+		t.Error("Expect LineFunc invoked")
+	}
+	if got, want := lines[1].Message, "bar\n"; got != want {
+		t.Errorf("Got line %q, want %q", got, want)
+	}
+	if got, want := lines[1].Number, 1; got != want {
+		t.Errorf("Got line %d, want %d", got, want)
+	}
+}
+
 func TestLineReplacer(t *testing.T) {
 	secrets := []*engine.Secret{
 		{Metadata: engine.Metadata{Name: "foo"}, Data: "bar"},