Fix: Make pod metrics StopRefreshLoop synchronous

The TestMetricsRefresh test in pod_metrics_test.go was flaky due to a
race condition. The `StopRefreshLoop` method would signal the metrics
refresh goroutine to stop but did not wait for its actual termination.
If the test updated the mock metrics client immediately after calling
`StopRefreshLoop`, the refresh goroutine could, in rare cases, perform
a final metrics fetch with the new data before fully exiting. This
resulted in the test asserting against unexpected metric values.

This commit resolves the issue by making `StopRefreshLoop` a synchronous
operation. A `sync.WaitGroup` is now used:
- `wg.Add(1)` is called before the refresh goroutine starts.
- `defer wg.Done()` is used within the refresh goroutine to signal
   completion.
- `wg.Wait()` is called in `StopRefreshLoop` to block until the
   goroutine has fully terminated.
- `stopOnce` is used to ensure the `done` channel is only closed once
  (for idempotency and protection against concurrent calls).

This change ensures that when `StopRefreshLoop` returns, the refresh
goroutine is guaranteed to have stopped, eliminating the race condition.

Author: LukeAVanDrie

pkg/epp/backend/metrics/pod_metrics.go

@@ -42,8 +42,10 @@ type podMetrics struct {
 	ds       Datastore
 	interval time.Duration
 
-	once sync.Once // ensure the StartRefreshLoop is only called once.
-	done chan struct{}
+	startOnce sync.Once // ensures the refresh loop goroutine is started only once
+	stopOnce  sync.Once // ensures the done channel is closed only once
+	done      chan struct{}
+	wg        sync.WaitGroup // to wait for the refresh loop to terminate
 
 	logger logr.Logger
 }
@@ -86,9 +88,11 @@ func toInternalPod(pod *corev1.Pod) *backend.Pod {
 // start starts a goroutine exactly once to periodically update metrics. The goroutine will be
 // stopped either when stop() is called, or the given ctx is cancelled.
 func (pm *podMetrics) startRefreshLoop(ctx context.Context) {
-	pm.once.Do(func() {
+	pm.startOnce.Do(func() {
+		pm.wg.Add(1)
 		go func() {
 			pm.logger.V(logutil.DEFAULT).Info("Starting refresher", "pod", pm.GetPod())
+			defer pm.wg.Done()
 			ticker := time.NewTicker(pm.interval)
 			defer ticker.Stop()
 			for {
@@ -138,5 +142,8 @@ func (pm *podMetrics) refreshMetrics() error {
 
 func (pm *podMetrics) StopRefreshLoop() {
 	pm.logger.V(logutil.DEFAULT).Info("Stopping refresher", "pod", pm.GetPod())
-	close(pm.done)
+	pm.stopOnce.Do(func() {
+		close(pm.done)
+	})
+	pm.wg.Wait()
 }

pkg/epp/backend/metrics/types.go

@@ -43,12 +43,13 @@ type PodMetricsFactory struct {
 func (f *PodMetricsFactory) NewPodMetrics(parentCtx context.Context, in *corev1.Pod, ds Datastore) PodMetrics {
 	pod := toInternalPod(in)
 	pm := &podMetrics{
-		pmc:      f.pmc,
-		ds:       ds,
-		interval: f.refreshMetricsInterval,
-		once:     sync.Once{},
-		done:     make(chan struct{}),
-		logger:   log.FromContext(parentCtx).WithValues("pod", pod.NamespacedName),
+		pmc:       f.pmc,
+		ds:        ds,
+		interval:  f.refreshMetricsInterval,
+		startOnce: sync.Once{},
+		stopOnce:  sync.Once{},
+		done:      make(chan struct{}),
+		logger:    log.FromContext(parentCtx).WithValues("pod", pod.NamespacedName),
 	}
 	pm.pod.Store(pod)
 	pm.metrics.Store(newMetrics())