fix: [CI-14200]: Trimming new line suffix from Log Line to avoid extr…

…a new line space (#314)

* fix: [CI-14200]: Trimming new line suffix from Log Line to avoid extra new line space

* fix: [CI-14200]: debugging

api/api.go

@@ -162,10 +162,11 @@ type (
 	}
 
 	LogConfig struct {
-		AccountID      string `json:"account_id,omitempty"`
-		IndirectUpload bool   `json:"indirect_upload,omitempty"` // Whether to directly upload via signed link or using log service
-		URL            string `json:"url,omitempty"`
-		Token          string `json:"token,omitempty"`
+		AccountID         string `json:"account_id,omitempty"`
+		IndirectUpload    bool   `json:"indirect_upload,omitempty"` // Whether to directly upload via signed link or using log service
+		URL               string `json:"url,omitempty"`
+		Token             string `json:"token,omitempty"`
+		TrimNewLineSuffix bool   `json:"trim_new_line_suffix,omitempty"`
 	}
 
 	TIConfig struct {

livelog/livelog.go

@@ -52,27 +52,28 @@ type Writer struct {
 	history       []*logstream.Line
 	prev          []byte
 
-	closed bool
-	close  chan struct{}
-	ready  chan struct{}
-
-	lastFlushTime time.Time
+	closed            bool
+	close             chan struct{}
+	ready             chan struct{}
+	trimNewLineSuffix bool
+	lastFlushTime     time.Time
 }
 
 // New returns a new writer
-func New(client logstream.Client, key, name string, nudges []logstream.Nudge, printToStdout bool) *Writer {
+func New(client logstream.Client, key, name string, nudges []logstream.Nudge, printToStdout bool, trimNewLineSuffix bool) *Writer {
 	b := &Writer{
-		client:        client,
-		key:           key,
-		name:          name,
-		now:           time.Now(),
-		printToStdout: printToStdout,
-		limit:         defaultLimit,
-		interval:      defaultInterval,
-		nudges:        nudges,
-		close:         make(chan struct{}),
-		ready:         make(chan struct{}, 1),
-		lastFlushTime: time.Now(),
+		client:            client,
+		key:               key,
+		name:              name,
+		now:               time.Now(),
+		printToStdout:     printToStdout,
+		limit:             defaultLimit,
+		interval:          defaultInterval,
+		nudges:            nudges,
+		close:             make(chan struct{}),
+		ready:             make(chan struct{}, 1),
+		lastFlushTime:     time.Now(),
+		trimNewLineSuffix: trimNewLineSuffix,
 	}
 	go b.Start()
 	return b
@@ -114,6 +115,11 @@ func (b *Writer) Write(p []byte) (n int, err error) {
 		if part == "" {
 			continue
 		}
+
+		if b.trimNewLineSuffix {
+			part = strings.TrimSuffix(part, "\n")
+		}
+
 		line := &logstream.Line{
 			Level:       defaultLevel,
 			Message:     truncate(part, maxLineLimit),

livelog/livelog_test.go

@@ -16,7 +16,7 @@ import (
 
 func TestLineWriterSingle(t *testing.T) {
 	client := new(mockClient)
-	w := New(client, "1", "1", nil, false)
+	w := New(client, "1", "1", nil, false, false)
 	w.SetInterval(time.Duration(0))
 	w.num = 4
 	_, _ = w.Write([]byte("foo\nbar\n"))
@@ -40,6 +40,32 @@ func TestLineWriterSingle(t *testing.T) {
 	}
 }
 
+func TestLineWriterSingleWithTrimNewLineSuffixEnabled(t *testing.T) {
+	client := new(mockClient)
+	w := New(client, "1", "1", nil, false, true)
+	w.SetInterval(time.Duration(0))
+	w.num = 4
+	_, _ = w.Write([]byte("foo\nbar\n"))
+
+	a := w.pending
+	b := []*logstream.Line{
+		{Number: 4, Message: "foo"},
+		{Number: 5, Message: "bar"},
+	}
+	if err := compare(a, b); err != nil {
+		t.Fail()
+		fmt.Print(a)
+		t.Log(err)
+	}
+
+	w.Close()
+	a = client.uploaded
+	if err := compare(a, b); err != nil {
+		t.Fail()
+		t.Log(err)
+	}
+}
+
 func compare(a, b []*logstream.Line) error {
 	if len(a) != len(b) {
 		return fmt.Errorf("expected size: %d, actual: %d", len(a), len(b))

pipeline/runtime/common.go

@@ -191,7 +191,7 @@ func GetReplacer(
 	cfg api.LogConfig, logKey, name string, secrets []string,
 ) logstream.Writer {
 	client := getLogServiceClient(cfg)
-	wc := livelog.New(client, logKey, name, []logstream.Nudge{}, false)
+	wc := livelog.New(client, logKey, name, []logstream.Nudge{}, false, cfg.TrimNewLineSuffix)
 	return logstream.NewReplacer(wc, secrets)
 }
 

pipeline/runtime/step_executor.go

@@ -389,7 +389,8 @@ func getLogStreamWriter(r *api.StartStepRequest) logstream.Writer {
 
 	// Create a log stream for step logs
 	client := pipelineState.GetLogStreamClient()
-	wc := livelog.New(client, r.LogKey, r.Name, getNudges(), false)
+
+	wc := livelog.New(client, r.LogKey, r.Name, getNudges(), false, pipelineState.GetLogConfig().TrimNewLineSuffix)
 	wr := logstream.NewReplacer(wc, secrets)
 	go wr.Open() //nolint:errcheck
 	return wr

pipeline/state.go

@@ -81,6 +81,13 @@ func (s *State) GetTIConfig() *tiCfg.Cfg {
 	return &s.tiConfig
 }
 
+func (s *State) GetLogConfig() *api.LogConfig {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return &s.logConfig
+}
+
 func GetState() *State {
 	once.Do(func() {
 		state = &State{