fix: k8s log stream cpu issue

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.19  AS build-env
+FROM golang:1.18  AS build-env
 
 RUN echo $GOPATH
 RUN apt update

DockerfileEA

@@ -1,4 +1,4 @@
-FROM golang:1.19  AS build-env
+FROM golang:1.18  AS build-env
 
 RUN echo $GOPATH
 RUN apt update

api/connector/Connector.go

@@ -33,6 +33,7 @@ import (
 	"net/http"
 	"regexp"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 )
@@ -104,28 +105,39 @@ func (impl PumpImpl) StartK8sStreamWithHeartBeat(w http.ResponseWriter, isReconn
 		done <- true
 	}()
 
-	// heartbeat end
-	for {
-		sc := bufio.NewScanner(stream)
-		for sc.Scan() {
-			log := sc.Text()
-			a := regexp.MustCompile(" ")
-			splitLog := a.Split(log, 2)
-			timeParsed, err := time.Parse(time.RFC3339, splitLog[0])
-			if err != nil {
-				impl.logger.Errorw("error in writing data over sse", "err", err)
-				return
-			}
-			mux.Lock()
-			err = impl.sendEvent([]byte(strconv.FormatInt(timeParsed.UnixNano(), 10)), nil, []byte(splitLog[1]), w)
-			mux.Unlock()
-			if err != nil {
-				impl.logger.Errorw("error in writing data over sse", "err", err)
+	bufReader := bufio.NewReader(stream)
+	eof := false
+	for !eof {
+		log, err := bufReader.ReadString('\n')
+		if err == io.EOF {
+			eof = true
+			// stop if we reached end of stream and the next line is empty
+			if log == "" {
 				return
 			}
-			f.Flush()
+		} else if err != nil && err != io.EOF {
+			impl.logger.Errorw("error in reading buffer string, StartK8sStreamWithHeartBeat", "err", err)
+			return
+		}
+		log = strings.TrimSpace(log) // Remove trailing line ending
+		a := regexp.MustCompile(" ")
+		splitLog := a.Split(log, 2)
+		parsedTime, err := time.Parse(time.RFC3339, splitLog[0])
+		if err != nil {
+			impl.logger.Errorw("error in writing data over sse", "err", err)
+			return
 		}
+		eventId := strconv.FormatInt(parsedTime.UnixNano(), 10)
+		mux.Lock()
+		err = impl.sendEvent([]byte(eventId), nil, []byte(splitLog[1]), w)
+		mux.Unlock()
+		if err != nil {
+			impl.logger.Errorw("error in writing data over sse", "err", err)
+			return
+		}
+		f.Flush()
 	}
+	// heartbeat end
 }
 
 func (impl PumpImpl) StartStreamWithHeartBeat(w http.ResponseWriter, isReconnect bool, recv func() (*application.LogEntry, error), err error) {
@@ -223,11 +235,9 @@ func (impl *PumpImpl) sendEvent(eventId []byte, eventName []byte, payload []byte
 		res = append(res, payload...)
 	}
 	res = append(res, '\n', '\n')
-	if i, err := w.Write(res); err != nil {
+	if _, err := w.Write(res); err != nil {
 		impl.logger.Errorf("Failed to send response chunk: %v", err)
 		return err
-	} else {
-		impl.logger.Debugw("msg written", "count", i)
 	}
 
 	return nil

util/k8s/k8sApplicationRestHandler.go

@@ -1,6 +1,7 @@
 package k8s
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -548,6 +549,17 @@ func (handler *K8sApplicationRestHandlerImpl) GetPodLogs(w http.ResponseWriter,
 		isReconnect = true
 	}
 	stream, err := handler.k8sApplicationService.GetPodLogs(r.Context(), request)
+	ctx, cancel := context.WithCancel(r.Context())
+	if cn, ok := w.(http.CloseNotifier); ok {
+		go func(done <-chan struct{}, closed <-chan bool) {
+			select {
+			case <-done:
+			case <-closed:
+				cancel()
+			}
+		}(ctx.Done(), cn.CloseNotify())
+	}
+	defer cancel()
 	defer util.Close(stream, handler.logger)
 	handler.pump.StartK8sStreamWithHeartBeat(w, isReconnect, stream, err)
 }