fix: wait for work goroutine to exit in Close()

Address review feedback: Close() now waits for the work goroutine
to fully exit before returning. Added doneChan that is closed when
the work loop exits, and Close() blocks on receiving from it.

Also moved work goroutine startup to newPodEventLogger to ensure
it's only started once (not per-namespace).

Author: kacpersaw

logger.go

@@ -86,8 +86,12 @@ func newPodEventLogger(ctx context.Context, opts podEventLoggerOptions) (*podEve
 			},
 			maxRetries: opts.maxRetries,
 		},
+		doneChan: make(chan struct{}),
 	}
 
+	// Start the work goroutine once
+	go reporter.lq.work(reporter.ctx, reporter.doneChan)
+
 	// If no namespaces are provided, we listen for events in all namespaces.
 	if len(opts.namespaces) == 0 {
 		if err := reporter.initNamespace(""); err != nil {
@@ -122,6 +126,8 @@ type podEventLogger struct {
 
 	// closeOnce ensures Close() is idempotent
 	closeOnce sync.Once
+	// doneChan is closed when the work goroutine exits
+	doneChan chan struct{}
 }
 
 // resolveEnvValue resolves the value of an environment variable, supporting both
@@ -164,8 +170,6 @@ func (p *podEventLogger) initNamespace(namespace string) error {
 	// This is to prevent us from sending duplicate events.
 	startTime := time.Now()
 
-	go p.lq.work(p.ctx)
-
 	podFactory := informers.NewSharedInformerFactoryWithOptions(p.client, 0, informers.WithNamespace(namespace), informers.WithTweakListOptions(func(lo *v1.ListOptions) {
 		lo.FieldSelector = p.fieldSelector
 		lo.LabelSelector = p.labelSelector
@@ -421,6 +425,8 @@ func (p *podEventLogger) Close() error {
 		close(p.stopChan)
 		close(p.errChan)
 	})
+	// Wait for the work goroutine to exit
+	<-p.doneChan
 	return nil
 }
 
@@ -509,8 +515,9 @@ type logQueuer struct {
 	maxRetries int
 }
 
-func (l *logQueuer) work(ctx context.Context) {
+func (l *logQueuer) work(ctx context.Context, done chan struct{}) {
 	defer l.cleanup()
+	defer close(done)
 
 	for ctx.Err() == nil {
 		select {

logger_test.go

@@ -675,7 +675,7 @@ func Test_logQueuer(t *testing.T) {
 
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
-		go lq.work(ctx)
+		go lq.work(ctx, make(chan struct{}))
 
 		ch <- agentLog{
 			op:           opLog,
@@ -742,7 +742,7 @@ func Test_logQueuer(t *testing.T) {
 
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
-		go lq.work(ctx)
+		go lq.work(ctx, make(chan struct{}))
 
 		token := "retry-token"
 		ch <- agentLog{
@@ -905,7 +905,7 @@ func Test_logQueuer(t *testing.T) {
 
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		go lq.work(ctx)
+		go lq.work(ctx, make(chan struct{}))
 
 		token := "max-retry-token"
 		ch <- agentLog{
@@ -1111,7 +1111,7 @@ func Test_logCache(t *testing.T) {
 
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		go lq.work(ctx)
+		go lq.work(ctx, make(chan struct{}))
 
 		token := "test-token"