fix: switch to bufio.Reader for KubeTask output parsing

pkg/function/kube_task_test.go

@@ -16,7 +16,9 @@ package function
 
 import (
 	"context"
+	"fmt"
 	"os"
+	"strings"
 	"time"
 
 	. "gopkg.in/check.v1"
@@ -61,6 +63,23 @@ func (s *KubeTaskSuite) TearDownSuite(c *C) {
 	}
 }
 
+func bigOutputPhase(namespace string) crv1alpha1.BlueprintPhase {
+	longstring := strings.Repeat("a", 100000)
+	return crv1alpha1.BlueprintPhase{
+		Name: "testOutput",
+		Func: KubeTaskFuncName,
+		Args: map[string]interface{}{
+			KubeTaskNamespaceArg: namespace,
+			KubeTaskImageArg:     consts.LatestKanisterToolsImage,
+			KubeTaskCommandArg: []string{
+				"sh",
+				"-c",
+				fmt.Sprintf("kando output longstring %s", longstring),
+			},
+		},
+	}
+}
+
 func outputPhase(namespace string) crv1alpha1.BlueprintPhase {
 	return crv1alpha1.BlueprintPhase{
 		Name: "testOutput",
@@ -161,3 +180,45 @@ func (s *KubeTaskSuite) TestKubeTask(c *C) {
 		}
 	}
 }
+
+func (s *KubeTaskSuite) TestKubeTaskWithBigOutput(c *C) {
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
+	defer cancel()
+	tp := param.TemplateParams{
+		StatefulSet: &param.StatefulSetParams{
+			Namespace: s.namespace,
+		},
+		PodOverride: crv1alpha1.JSONMap{
+			"containers": []map[string]interface{}{
+				{
+					"name":            "container",
+					"imagePullPolicy": "Always",
+				},
+			},
+		},
+	}
+	expectedOut := strings.Repeat("a", 100000)
+	action := "test"
+	for _, tc := range []struct {
+		bp   *crv1alpha1.Blueprint
+		outs []map[string]interface{}
+	}{
+		{
+			bp: newTaskBlueprint(bigOutputPhase(s.namespace)),
+			outs: []map[string]interface{}{
+				{
+					"longstring": expectedOut,
+				},
+			},
+		},
+	} {
+		phases, err := kanister.GetPhases(*tc.bp, action, kanister.DefaultVersion, tp)
+		c.Assert(err, IsNil)
+		c.Assert(phases, HasLen, len(tc.outs))
+		for i, p := range phases {
+			out, err := p.Exec(ctx, *tc.bp, action, tp)
+			c.Assert(err, IsNil, Commentf("Phase %s failed", p.Name()))
+			c.Assert(out, DeepEquals, tc.outs[i])
+		}
+	}
+}

pkg/output/stream.go

@@ -36,18 +36,28 @@ func splitLines(ctx context.Context, r io.ReadCloser, f func(context.Context, st
 	}()
 
 	// Scan log lines when ready.
-	s := bufio.NewScanner(r)
-	for s.Scan() {
-		l := s.Text()
-		l = strings.TrimSpace(l)
-		if l == "" {
-			continue
+	// Don't use bufio.Scanner because it breaks if lines are too long
+	reader := bufio.NewReader(r)
+	// ReadString returns error AND a line for last line
+	line, err := reader.ReadString('\n')
+	for {
+		line = strings.TrimSpace(line)
+
+		if line != "" {
+			if err := f(ctx, line); err != nil {
+				return err
+			}
 		}
-		if err := f(ctx, l); err != nil {
-			return err
+		if err != nil {
+			break
 		}
+		line, err = reader.ReadString('\n')
 	}
-	return errors.Wrap(s.Err(), "Split lines failed")
+	if err != io.EOF {
+		return errors.Wrap(err, "Split lines failed")
+	}
+
+	return nil
 }
 
 func LogAndParse(ctx context.Context, r io.ReadCloser) (map[string]interface{}, error) {