Fix race condition for harvester Start / Stop in registry

filebeat/harvester/harvester.go

@@ -53,7 +53,7 @@ type Harvester struct {
 	encoding        encoding.Encoding
 	done            chan struct{}
 	stopOnce        sync.Once
-	stopWg          *sync.WaitGroup
+	StopWg          *sync.WaitGroup
 	outlet          Outlet
 	ID              uuid.UUID
 	processors      *processors.Processors
@@ -71,7 +71,7 @@ func NewHarvester(
 		state:  state,
 		states: states,
 		done:   make(chan struct{}),
-		stopWg: &sync.WaitGroup{},
+		StopWg: &sync.WaitGroup{},
 		outlet: outlet,
 		ID:     uuid.NewV4(),
 	}

filebeat/harvester/log.go

@@ -54,7 +54,6 @@ func (h *Harvester) Harvest(r reader.Reader) {
 	harvesterStarted.Add(1)
 	harvesterRunning.Add(1)
 
-	h.stopWg.Add(1)
 	defer func() {
 		// Channel to stop internal harvester routines
 		h.stop()
@@ -64,7 +63,7 @@ func (h *Harvester) Harvest(r reader.Reader) {
 		harvesterRunning.Add(-1)
 
 		// Marks harvester stopping completed
-		h.stopWg.Done()
+		h.StopWg.Done()
 	}()
 
 	// Closes reader after timeout or when done channel is closed
@@ -164,7 +163,7 @@ func (h *Harvester) stop() {
 // Stop stops harvester and waits for completion
 func (h *Harvester) Stop() {
 	h.stop()
-	h.stopWg.Wait()
+	h.StopWg.Wait()
 }
 
 // sendEvent sends event to the spooler channel

filebeat/prospector/registry.go

@@ -12,11 +12,13 @@ type harvesterRegistry struct {
 	sync.Mutex
 	harvesters map[uuid.UUID]*harvester.Harvester
 	wg         sync.WaitGroup
+	done       chan struct{}
 }
 
 func newHarvesterRegistry() *harvesterRegistry {
 	return &harvesterRegistry{
 		harvesters: map[uuid.UUID]*harvester.Harvester{},
+		done:       make(chan struct{}),
 	}
 }
 
@@ -33,16 +35,21 @@ func (hr *harvesterRegistry) remove(h *harvester.Harvester) {
 }
 
 func (hr *harvesterRegistry) Stop() {
+
 	hr.Lock()
+	defer func() {
+		hr.Unlock()
+		hr.waitForCompletion()
+	}()
+	close(hr.done)
 	for _, hv := range hr.harvesters {
 		hr.wg.Add(1)
 		go func(h *harvester.Harvester) {
 			hr.wg.Done()
 			h.Stop()
 		}(hv)
 	}
-	hr.Unlock()
-	hr.waitForCompletion()
+
 }
 
 func (hr *harvesterRegistry) waitForCompletion() {
@@ -51,7 +58,23 @@ func (hr *harvesterRegistry) waitForCompletion() {
 
 func (hr *harvesterRegistry) start(h *harvester.Harvester, r reader.Reader) {
 
+	// Make sure stop is not called during starting a harvester
+	hr.Lock()
+
+	// Make sure no new harvesters are started after stop was called
+	select {
+	case <-hr.done:
+		return
+	default:
+	}
+
 	hr.wg.Add(1)
+	h.StopWg.Add(1)
+
+	hr.Unlock()
+
+	// TODO: It could happen that stop is called here
+
 	hr.add(h)
 
 	// Update state before staring harvester