fix(podlogs): follow mode exit bug and test improvements (cloudnative-pg#9599)

The test "should catch extra logs if given the follow option" was flaky
in CI because it tested implementation details (counting loop iterations
with tight timing) rather than actual behavior.

Redesigned the test to verify what Follow=true actually does: it keeps
streaming until the context is cancelled. The test now uses
Eventually/Consistently patterns that handle timing variations
gracefully, making it robust across different environments.

The improved test exposed a bug in the production code: the streaming
function would exit when all current log streams completed, even when
Follow=true was set. This caused premature exit instead of continuing to
poll for new or restarted pods. Fixed by changing the exit condition to
only apply when Follow=false.

Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>

Author: mnencia

pkg/podlogs/cluster_writer.go

@@ -220,7 +220,7 @@ func (csr *ClusterWriter) SingleStream(ctx context.Context, writer io.Writer) er
 				}
 			}
 		}
-		if streamSet.isZero() {
+		if !csr.Options.Follow && streamSet.isZero() {
 			return nil
 		}
 

pkg/podlogs/cluster_writer_test.go

@@ -22,7 +22,6 @@ package podlogs
 import (
 	"bytes"
 	"context"
-	"strings"
 	"sync"
 	"time"
 
@@ -153,40 +152,50 @@ var _ = Describe("Cluster logging tests", func() {
 		Expect(logBuffer.String()).To(BeEquivalentTo("fake logs\nfake logs\n"))
 	})
 
-	It("should catch extra logs if given the follow option", func(ctx context.Context) {
+	It("should continue streaming when follow option is enabled", func(ctx context.Context) {
 		client := fake.NewClientset(pod)
-
-		wg := sync.WaitGroup{}
+		var wg sync.WaitGroup
 		wg.Add(1)
 		var logBuffer syncBuffer
+		errChan := make(chan error, 1)
+		streamCtx, cancel := context.WithCancel(ctx)
 
-		// let's set a short follow-wait, and keep the cluster streaming for two
-		// cycles
-		followWaiting := 150 * time.Millisecond
-		ctx2, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
 		go func() {
-			// we always invoke done no matter what happens
 			defer wg.Done()
 			defer GinkgoRecover()
 			streamClusterLogs := ClusterWriter{
 				Cluster: cluster,
 				Options: &corev1.PodLogOptions{
 					Follow: true,
 				},
-				FollowWaiting: followWaiting,
+				FollowWaiting: 50 * time.Millisecond, // Short interval for test speed
 				Client:        client,
 			}
-			err := streamClusterLogs.SingleStream(ctx2, &logBuffer)
-			// we cannot reliably now if we will close the function before the context
-			// deadline, so we accept both nil and context.DeadlineExceeded
-			Expect(err).To(Or(BeNil(), Equal(context.DeadlineExceeded)))
+			err := streamClusterLogs.SingleStream(streamCtx, &logBuffer)
+			errChan <- err
 		}()
 
-		time.Sleep(350 * time.Millisecond)
+		Eventually(func() bool {
+			return len(logBuffer.String()) > 0
+		}, 2*time.Second, 10*time.Millisecond).Should(BeTrue(),
+			"streaming should capture logs")
+
+		Consistently(func() bool {
+			select {
+			case <-errChan:
+				return false
+			default:
+				return true
+			}
+		}, 200*time.Millisecond, 50*time.Millisecond).Should(BeTrue(),
+			"streaming should continue until cancelled")
+
 		cancel()
 		wg.Wait()
 
-		fakeLogCount := strings.Count(logBuffer.String(), "fake logs\n")
-		Expect(fakeLogCount).To(BeNumerically(">=", 2))
+		var streamErr error
+		Eventually(errChan, time.Second).Should(Receive(&streamErr))
+		Expect(streamErr).To(Equal(context.Canceled))
+		Expect(logBuffer.String()).To(ContainSubstring("fake logs"))
 	})
 })