feat(performance): improve Log Processing performance (#5647)

* feat(performance): buffer the logs sent from the container, to avoid sending message for each line
* feat(performance): batch Test Workflow's result updates
* fix(testworkflows): handle getting long container logs after the log rotation happens
   @see {@link https://stackoverflow.com/a/68673451}
* feat(testworkflows): optimize reading timestamp from Kubernetes logs
* feat(testworkflows): optimize buffering logs
* feat(testworkflows): use native channel instead of heavier Channels for WatchInstrumentedPod
* feat(testworkflows): increase buffer size for logs buffering

pkg/testworkflows/testworkflowcontroller/channel.go

@@ -20,6 +20,7 @@ type Channel[T any] interface {
 	Channel() <-chan ChannelMessage[T]
 	Close()
 	Done() <-chan struct{}
+	CtxErr() error
 }
 
 type channel[T any] struct {
@@ -168,3 +169,7 @@ func (c *channel[T]) Close() {
 func (c *channel[T]) Done() <-chan struct{} {
 	return c.ctx.Done()
 }
+
+func (c *channel[T]) CtxErr() error {
+	return c.ctx.Err()
+}

pkg/testworkflows/testworkflowcontroller/controller.go

@@ -212,7 +212,7 @@ func (c *controller) StopController() {
 }
 
 func (c *controller) Watch(parentCtx context.Context) <-chan ChannelMessage[Notification] {
-	w, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
+	ch, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
 		JobEvents: c.jobEvents,
 		Job:       c.job,
 	})
@@ -222,7 +222,7 @@ func (c *controller) Watch(parentCtx context.Context) <-chan ChannelMessage[Noti
 		v.Close()
 		return v.Channel()
 	}
-	return w.Channel()
+	return ch
 }
 
 // TODO: Make it actually light
@@ -281,15 +281,15 @@ func (c *controller) Logs(parentCtx context.Context, follow bool) io.Reader {
 		case <-c.podEvents.Peek(parentCtx):
 		case <-alignTimeoutCh:
 		}
-		w, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
+		ch, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
 			JobEvents: c.jobEvents,
 			Job:       c.job,
 			Follow:    common.Ptr(follow),
 		})
 		if err != nil {
 			return
 		}
-		for v := range w.Channel() {
+		for v := range ch {
 			if v.Error == nil && v.Value.Log != "" && !v.Value.Temporary {
 				if ref != v.Value.Ref && v.Value.Ref != "" {
 					ref = v.Value.Ref