pkg/*: fix data races in PubSub and ticker test (#4155)

Fixes a data race in PubSub singleton instantiation
where the singleton object creation is not protected
from concurrent initialization.
Fixes a data race in the ticker test where a variable
is written/read to/from without protected access from
multiple goroutines.

Signed-off-by: Shashank Ram <[email protected]>

pkg/k8s/events/event_pubsub.go

@@ -1,6 +1,8 @@
 package events
 
 import (
+	"sync"
+
 	"github.com/cskr/pubsub"
 
 	"github.com/openservicemesh/osm/pkg/announcements"
@@ -14,6 +16,9 @@ const (
 var (
 	// Globally accessible instance, through singleton pattern using getPubSubInstance()
 	pubSubInstance *pubsub.PubSub
+
+	// pubSubOnce is used to ensure PubSub object creation happens just once
+	pubSubOnce sync.Once
 )
 
 // Subscribe is the Subscribe implementation for PubSub
@@ -59,8 +64,8 @@ func Unsub(unsubChan chan interface{}) {
 // Note that spawning the instance is not thread-safe. First call should happen on
 // a single-routine context to avoid races.
 func getPubSubInstance() *pubsub.PubSub {
-	if pubSubInstance == nil {
+	pubSubOnce.Do(func() {
 		pubSubInstance = pubsub.New(defaultAnnouncementChannelSize)
-	}
+	})
 	return pubSubInstance
 }

pkg/ticker/ticker_test.go

@@ -1,6 +1,7 @@
 package ticker
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -44,14 +45,17 @@ func TestTicker(t *testing.T) {
 	broadcastEvents := events.Subscribe(announcements.ScheduleProxyBroadcast)
 	defer events.Unsub(broadcastEvents)
 
+	var counterMutex sync.Mutex
 	broadcastsReceived := 0
 	stop := make(chan struct{})
 	defer close(stop)
 	go func() {
 		for {
 			select {
 			case <-broadcastEvents:
+				counterMutex.Lock()
 				broadcastsReceived++
+				counterMutex.Unlock()
 			case <-stop:
 				return
 			}
@@ -73,6 +77,8 @@ func TestTicker(t *testing.T) {
 
 	// broadcast events should increase in the next few seconds
 	assert.Eventually(func() bool {
+		counterMutex.Lock()
+		defer counterMutex.Unlock()
 		return broadcastsReceived > 0
 	}, 3*time.Second, 500*time.Millisecond)
 
@@ -83,6 +89,8 @@ func TestTicker(t *testing.T) {
 
 	// Should stop increasing
 	assert.Eventually(func() bool {
+		counterMutex.Lock()
+		defer counterMutex.Unlock()
 		firstRead := broadcastsReceived
 		time.Sleep(1 * time.Second)
 		secondRead := broadcastsReceived