From c23bd5771fad632936339811c19de528ce74b108 Mon Sep 17 00:00:00 2001
From: Jacek Wysocki <jacek.wysocki@gmail.com>
Date: Mon, 10 Jun 2024 14:11:04 +0200
Subject: [PATCH] fix: logs proxy for slaves followup - split subject with pod
 name (#5558)

* fix: separate pod name and log stream topic for slave pods

* fix: pass valid data into main
---
 cmd/logs-sidecar/main.go  | 10 ++--------
 pkg/logs/sidecar/proxy.go | 28 ++++++++++++++++------------
 2 files changed, 18 insertions(+), 20 deletions(-)

diff --git a/cmd/logs-sidecar/main.go b/cmd/logs-sidecar/main.go
index 2ccca29c7e6..c91e440c1e5 100644
--- a/cmd/logs-sidecar/main.go
+++ b/cmd/logs-sidecar/main.go
@@ -53,16 +53,10 @@ func main() {
 		return
 	}
 
-	podName := cfg.PodName
-	// fallback to exection id for older tests
-	if podName == "" {
-		podName = cfg.ExecutionId
-	}
-
-	log.Debugw("starting logs proxy", "podName", podName, "namespace", cfg.Namespace, "source", cfg.Source, "executionId", cfg.ExecutionId)
+	log.Debugw("starting logs proxy", "podName", cfg.PodName, "namespace", cfg.Namespace, "source", cfg.Source, "executionId", cfg.ExecutionId)
 
 	// run Sidecar Logs Proxy - it will proxy logs from pod to nats
-	proxy := sidecar.NewProxy(clientset, podsClient, logsStream, js, log, cfg.Namespace, podName, cfg.Source)
+	proxy := sidecar.NewProxy(clientset, podsClient, logsStream, js, log, cfg.Namespace, cfg.PodName, cfg.ExecutionId, cfg.Source)
 	if err := proxy.Run(ctx); err != nil {
 		log.Errorw("error proxying logs", "error", err)
 	}
diff --git a/pkg/logs/sidecar/proxy.go b/pkg/logs/sidecar/proxy.go
index 105a5ca3ddd..cbad9ab94a5 100644
--- a/pkg/logs/sidecar/proxy.go
+++ b/pkg/logs/sidecar/proxy.go
@@ -37,13 +37,14 @@ const (
 )
 
 func NewProxy(clientset kubernetes.Interface, podsClient tcorev1.PodInterface, logsStream client.Stream, js jetstream.JetStream, log *zap.SugaredLogger,
-	namespace, podName, source string) *Proxy {
+	namespace, podName, subject, source string) *Proxy {
 	return &Proxy{
 		log:        log.With("service", "logs-proxy", "namespace", namespace, "podName", podName),
 		js:         js,
 		clientset:  clientset,
 		namespace:  namespace,
 		podName:    podName,
+		subject:    subject,
 		podsClient: podsClient,
 		logsStream: logsStream,
 		source:     source,
@@ -51,15 +52,15 @@ func NewProxy(clientset kubernetes.Interface, podsClient tcorev1.PodInterface, l
 }
 
 type Proxy struct {
-	log         *zap.SugaredLogger
-	js          jetstream.JetStream
-	clientset   kubernetes.Interface
-	namespace   string
-	executionId string
-	source      string
-	podsClient  tcorev1.PodInterface
-	logsStream  client.InitializedStreamPusher
-	podName     string
+	log        *zap.SugaredLogger
+	js         jetstream.JetStream
+	clientset  kubernetes.Interface
+	namespace  string
+	subject    string
+	source     string
+	podsClient tcorev1.PodInterface
+	logsStream client.InitializedStreamPusher
+	podName    string
 }
 
 func (p *Proxy) Run(ctx context.Context) error {
@@ -106,7 +107,7 @@ func (p *Proxy) Run(ctx context.Context) error {
 }
 
 func (p *Proxy) streamLogs(ctx context.Context, logs chan *events.Log) (err error) {
-	pods, err := executor.GetJobPods(ctx, p.podsClient, p.getSubject(), 1, 10)
+	pods, err := executor.GetJobPods(ctx, p.podsClient, p.getPodName(), 1, 10)
 	if err != nil {
 		p.handleError(err, "error getting job pods")
 		return err
@@ -255,10 +256,13 @@ func (p *Proxy) handleError(err error, title string) {
 		if err != nil {
 			p.log.Errorw("error pushing error to stream", "title", title, "error", err)
 		}
-
 	}
 }
 
 func (p *Proxy) getSubject() string {
+	return p.subject
+}
+
+func (p *Proxy) getPodName() string {
 	return p.podName
 }