ZING-32005: Fix data races. (#27)

* fix data race

* fix data race on writer

* add sleep to manager_test.go

* fix Makefile

* replace Addwg()

* Use different strategy to fix dataraces

---------

Co-authored-by: Danylo Kondratiev <knightpp@proton.me>

Makefile

@@ -54,19 +54,18 @@ test: COVERAGE_PROFILE := coverprofile.out
 test: COVERAGE_HTML    := $(COVERAGE_DIR)/index.html
 test: COVERAGE_XML     := $(COVERAGE_DIR)/coverage.xml
 test: fmt lint
-	@echo "Please ENABLE RACE DETECTOR"
 	@mkdir -p $(COVERAGE_DIR)
 	@$(GINKGO) \
 		run \
 		-r \
 		--tags integration \
 		--cover \
 		--coverprofile $(COVERAGE_PROFILE) \
-		--covermode=count \
+		--covermode=atomic \
+		--race \
 		--junit-report=junit.xml
 	$Q $(GO) tool cover -html=$(COVERAGE_PROFILE) -o $(COVERAGE_HTML)
 	$Q $(GOCOV) convert $(COVERAGE_PROFILE) | $(GOCOVXML) > $(COVERAGE_XML)
-	@echo "Please ENABLE RACE DETECTOR"
 
 .PHONY: test-containerized
 test-containerized:

health/collector.go

@@ -54,7 +54,9 @@ func StopCollectorSingleton() {
 	collectorLock.Lock()
 	defer collectorLock.Unlock()
 	if collector != nil {
-		close(collector.done)
+		collector.doneOnce.Do(func() {
+			close(collector.done)
+		})
 	}
 }
 
@@ -79,6 +81,7 @@ type healthCollector struct {
 	cycleDuration time.Duration
 	metricsIn     chan<- *TargetMeasurement
 	done          chan struct{}
+	doneOnce      sync.Once
 }
 
 func (hc *healthCollector) HeartBeat(targetID string) (context.CancelFunc, error) {

health/manager.go

@@ -16,6 +16,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	logging "github.com/zenoss/zenoss-go-sdk/health/log"
@@ -42,19 +43,41 @@ func FrameworkStart(ctx context.Context, cfg *Config, m Manager, writer w.Health
 	c := NewCollector(cfg.CollectionCycle, measurementsCh)
 	SetCollectorSingleton(c)
 
+	ctx, cancel := context.WithCancel(ctx)
+	var doneWg sync.WaitGroup
+
+	doneWg.Add(1)
 	go func() {
+		defer doneWg.Done()
+		defer cancel()
 		<-ctx.Done()
+
 		StopCollectorSingleton()
 	}()
 
-	go m.Start(ctx, measurementsCh, healthCh, targetCh)
+	doneWg.Add(1)
+	go func() {
+		defer doneWg.Done()
+		defer cancel()
 
-	go writer.Start(ctx, healthCh, targetCh)
+		m.Start(ctx, measurementsCh, healthCh, targetCh)
+	}()
+
+	doneWg.Add(1)
+	go func() {
+		defer doneWg.Done()
+		defer cancel()
+
+		writer.Start(ctx, healthCh, targetCh)
+	}()
 
 	return func() {
+		cancel()
+
 		StopCollectorSingleton()
 		m.Shutdown()
 		writer.Shutdown()
+		doneWg.Wait()
 	}
 }
 
@@ -104,7 +127,7 @@ type healthManager struct {
 	wg *sync.WaitGroup
 
 	mu      sync.Mutex
-	started bool
+	started atomic.Bool
 }
 
 func (hm *healthManager) Start(
@@ -131,10 +154,10 @@ func (hm *healthManager) Start(
 		hm.healthForwarder(ctx, healthIn)
 	}()
 
-	hm.started = true
+	hm.started.Store(true)
 	go func() {
 		hm.wg.Wait()
-		hm.started = false
+		hm.started.Store(false)
 		hm.stopWait <- struct{}{}
 	}()
 
@@ -150,7 +173,7 @@ func (hm *healthManager) Shutdown() {
 }
 
 func (hm *healthManager) IsStarted() bool {
-	return hm.started
+	return hm.started.Load()
 }
 
 func (hm *healthManager) AddTargets(targets []*target.Target) {
@@ -159,7 +182,7 @@ func (hm *healthManager) AddTargets(targets []*target.Target) {
 
 	for _, newTarget := range targets {
 		hm.registry.setRawHealthForTarget(newRawHealth(newTarget))
-		if hm.started {
+		if hm.IsStarted() {
 			hm.targetIn <- newTarget
 		}
 	}

health/manager_test.go

@@ -307,6 +307,7 @@ var _ = Describe("Health Manager", func() {
 					CounterChange: counterValue,
 				}
 				mesuresCh <- counterMeasure
+				time.Sleep(200 * time.Microsecond)
 
 				go func() {
 					testTarget := <-targetCh

health/writer/writer.go

@@ -7,7 +7,6 @@ package writer
 
 import (
 	"context"
-	"sync"
 
 	"github.com/zenoss/zenoss-go-sdk/health/log"
 	"github.com/zenoss/zenoss-go-sdk/health/target"
@@ -29,7 +28,6 @@ func New(dests []Destination) HealthWriter {
 	return &writer{
 		destinations: dests,
 		stopSig:      stopSig,
-		wg:           &sync.WaitGroup{},
 	}
 }
 
@@ -38,13 +36,9 @@ type writer struct {
 	// we can also create and add some data transformers if
 	// health data should be changed somehow before send
 	stopSig chan struct{}
-	wg      *sync.WaitGroup
 }
 
 func (w *writer) Start(ctx context.Context, healthIn <-chan *target.Health, targetIn <-chan *target.Target) {
-	w.wg.Add(1)
-	defer w.wg.Done()
-
 	for {
 		select {
 		case healthData, more := <-healthIn:
@@ -75,5 +69,4 @@ func (w *writer) Start(ctx context.Context, healthIn <-chan *target.Health, targ
 
 func (w *writer) Shutdown() {
 	close(w.stopSig)
-	w.wg.Wait()
 }

health/writer/writer_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -83,14 +84,21 @@ var _ = Describe("Writer", func() {
 			)
 			h := target.NewHealth(targetID, utils.DefaultTargetType)
 
+			var wg sync.WaitGroup
 			hCh := make(chan *target.Health)
 			tCh := make(chan *target.Target)
-			go healthWriter.Start(ctx, hCh, tCh)
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				healthWriter.Start(ctx, hCh, tCh)
+			}()
 			tCh <- hTarget
 			hCh <- h
 			time.Sleep(1 * time.Second)
 			close(hCh)
 			close(tCh)
+			wg.Wait()
 			out := buf.String()
 
 			Ω(out).Should(ContainSubstring(fmt.Sprintf("TargetID: %s, Status=Healthy", targetID)))