[netpath] Fix panic in npCollector cause by forwader.SendEventPlatformEventBlocking() call

todo

Author: AlexandreYang

comp/networkpath/npcollector/npcollectorimpl/npcollector.go

@@ -12,6 +12,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"sync"
 	"time"
 
 	model "github.com/DataDog/agent-payload/v5/process"
@@ -62,6 +63,7 @@ type npCollectorImpl struct {
 	workers       int
 	stopChan      chan struct{}
 	flushLoopDone chan struct{}
+	workersDone   chan struct{}
 	runDone       chan struct{}
 	flushInterval time.Duration
 
@@ -122,6 +124,7 @@ func newNpCollectorImpl(epForwarder eventplatform.Forwarder, collectorConfigs *c
 		stopChan:      make(chan struct{}),
 		runDone:       make(chan struct{}),
 		flushLoopDone: make(chan struct{}),
+		workersDone:   make(chan struct{}),
 
 		runTraceroute: runTraceroute,
 		statsdClient:  statsd,
@@ -274,7 +277,7 @@ func (s *npCollectorImpl) start() error {
 
 	go s.listenPathtests()
 	go s.flushLoop()
-	s.startWorkers()
+	go s.startWorkers()
 
 	return nil
 }
@@ -286,6 +289,7 @@ func (s *npCollectorImpl) stop() {
 	}
 	close(s.stopChan)
 	<-s.flushLoopDone
+	<-s.workersDone
 	<-s.runDone
 	s.running = false
 }
@@ -496,13 +500,24 @@ func (s *npCollectorImpl) getReverseDNSResult(ipAddr string, results map[string]
 
 func (s *npCollectorImpl) startWorkers() {
 	s.logger.Debugf("Starting workers (%d)", s.workers)
+
+	// TODO: TEST ME
+	// TODO: TEST ME
+	// TODO: TEST ME
+	var wg sync.WaitGroup
 	for w := 0; w < s.workers; w++ {
+		wg.Add(1)
 		s.logger.Debugf("Starting worker #%d", w)
-		go s.startWorker(w)
+		go func() {
+			defer wg.Done()
+			s.runWorker(w)
+		}()
 	}
+	wg.Wait()
+	s.workersDone <- struct{}{}
 }
 
-func (s *npCollectorImpl) startWorker(workerID int) {
+func (s *npCollectorImpl) runWorker(workerID int) {
 	for {
 		select {
 		case <-s.stopChan:

comp/networkpath/npcollector/npcollectorimpl/npcollector_test.go

@@ -548,7 +548,7 @@ func Test_npCollectorImpl_stopWorker(t *testing.T) {
 
 	stopped := make(chan bool, 1)
 	go func() {
-		npCollector.startWorker(42)
+		npCollector.runWorker(42)
 		stopped <- true
 	}()
 	close(npCollector.stopChan)