Cherry-pick elastic#7820 to 6.5: Fix goroutine leak in SubOutlet (ela…

…stic#9592)

Cherry-pick of PR elastic#7820 to 6.5 branch. Original message: 

This fixes a goroutine leaking every time that a harvester is closed.

CHANGELOG.asciidoc

@@ -40,6 +40,7 @@ https://github.com/elastic/beats/compare/v6.5.3...6.5[Check the HEAD diff]
 *Filebeat*
 
 - Fix saved objects in filebeat haproxy dashboard. {pull}9417[9417]
+- Fixed a memory leak when harvesters are closed. {pull}7820[7820]
 
 *Heartbeat*
 

filebeat/channel/util.go

@@ -18,17 +18,19 @@
 package channel
 
 import (
+	"sync"
+
 	"github.com/elastic/beats/filebeat/util"
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
-	"github.com/elastic/beats/libbeat/common/atomic"
 )
 
 type subOutlet struct {
-	isOpen atomic.Bool
-	done   chan struct{}
-	ch     chan *util.Data
-	res    chan bool
+	done      chan struct{}
+	ch        chan *util.Data
+	res       chan bool
+	mutex     sync.Mutex
+	closeOnce sync.Once
 }
 
 // ConnectTo creates a new Connector, combining a beat.Pipeline with an outlet Factory.
@@ -42,10 +44,9 @@ func ConnectTo(pipeline beat.Pipeline, factory Factory) Connector {
 // underlying outlet.
 func SubOutlet(out Outleter) Outleter {
 	s := &subOutlet{
-		isOpen: atomic.MakeBool(true),
-		done:   make(chan struct{}),
-		ch:     make(chan *util.Data),
-		res:    make(chan bool, 1),
+		done: make(chan struct{}),
+		ch:   make(chan *util.Data),
+		res:  make(chan bool, 1),
 	}
 
 	go func() {
@@ -58,21 +59,30 @@ func SubOutlet(out Outleter) Outleter {
 }
 
 func (o *subOutlet) Close() error {
-	isOpen := o.isOpen.Swap(false)
-	if isOpen {
+	o.closeOnce.Do(func() {
+		// Signal OnEvent() to terminate
 		close(o.done)
-	}
+		// This mutex prevents the event channel to be closed if OnEvent is
+		// still running.
+		o.mutex.Lock()
+		defer o.mutex.Unlock()
+		close(o.ch)
+	})
 	return nil
 }
 
 func (o *subOutlet) OnEvent(d *util.Data) bool {
-	if !o.isOpen.Load() {
+
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	select {
+	case <-o.done:
 		return false
+	default:
 	}
 
 	select {
 	case <-o.done:
-		close(o.ch)
 		return false
 
 	case o.ch <- d:
@@ -92,8 +102,6 @@ func (o *subOutlet) OnEvent(d *util.Data) bool {
 			//  Once all messages are in the publisher pipeline, in correct order,
 			//  it depends on registrar/publisher pipeline if state is finally updated
 			//  in the registrar.
-
-			close(o.ch)
 			return true
 
 		case ret := <-o.res: