hubble: Fix data races in pkg/hubble.TestRingReader_NextFollow_WithEmptyRing

pkg/hubble/container/ring_reader.go

@@ -8,13 +8,15 @@ import (
 	"sync"
 
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
+	"github.com/cilium/cilium/pkg/lock"
 )
 
 // RingReader is a reader for a Ring container.
 type RingReader struct {
 	ring          *Ring
 	idx           uint64
 	ctx           context.Context
+	mutex         lock.Mutex // protects writes to followChan
 	followChan    chan *v1.Event
 	followChanLen int
 	wg            sync.WaitGroup
@@ -75,16 +77,21 @@ func (r *RingReader) NextFollow(ctx context.Context) *v1.Event {
 	// if the context changed between invocations, we also have to restart
 	// readFrom, as the old readFrom instance will be using the old context.
 	if r.ctx != ctx {
+		r.mutex.Lock()
 		if r.followChan == nil {
 			r.followChan = make(chan *v1.Event, r.followChanLen)
 		}
+		r.mutex.Unlock()
+
 		r.wg.Add(1)
 		go func(ctx context.Context) {
 			r.ring.readFrom(ctx, r.idx, r.followChan)
+			r.mutex.Lock()
 			if ctx.Err() != nil && r.followChan != nil { // context is done
 				close(r.followChan)
 				r.followChan = nil
 			}
+			r.mutex.Unlock()
 			r.wg.Done()
 		}(ctx)
 		r.ctx = ctx
@@ -95,8 +102,12 @@ func (r *RingReader) NextFollow(ctx context.Context) *v1.Event {
 		}
 	}()
 
+	r.mutex.Lock()
+	followChan := r.followChan
+	r.mutex.Unlock()
+
 	select {
-	case e, ok := <-r.followChan:
+	case e, ok := <-followChan:
 		if !ok {
 			// the channel is closed so the context is done
 			return nil
@@ -110,9 +121,10 @@ func (r *RingReader) NextFollow(ctx context.Context) *v1.Event {
 	}
 }
 
-// Close waits for any method to return and closes the RingReader. It is not
+// Close waits for any spawned go routines to finish. It is not
 // required to call Close on a RingReader but it may be useful for specific
-// situations such as testing.
+// situations such as testing. Must not be called concurrently with NextFollow,
+// as otherwise NextFollow spawns new go routines that are not waited on.
 func (r *RingReader) Close() error {
 	r.wg.Wait()
 	return nil

pkg/hubble/container/ring_reader_test.go

@@ -289,11 +289,13 @@ func TestRingReader_NextFollow_WithEmptyRing(t *testing.T) {
 	reader := NewRingReader(ring, ring.LastWriteParallel())
 	ctx, cancel := context.WithCancel(context.Background())
 	c := make(chan *v1.Event)
+	done := make(chan struct{})
 	go func() {
 		select {
 		case <-ctx.Done():
 		case c <- reader.NextFollow(ctx):
 		}
+		close(done)
 	}()
 	select {
 	case <-c:
@@ -302,5 +304,6 @@ func TestRingReader_NextFollow_WithEmptyRing(t *testing.T) {
 		// the call blocked, we're good
 	}
 	cancel()
+	<-done
 	assert.Nil(t, reader.Close())
 }