Stop waiting for signals on closed outleters (#11263) (#11489)

Outleters start a goroutine to handle the finalization of filebeat. If the
outleter is closed by other means the goroutine will be kept running
even if it has nothing to do, leaking goroutines.

Stop this goroutine if the outleter is closed.

(cherry picked from commit 57c9891)

CHANGELOG.next.asciidoc

@@ -37,6 +37,7 @@ https://github.com/elastic/beats/compare/v6.6.2...6.6[Check the HEAD diff]
 *Filebeat*
 
 - Fix a bug with the convert_timezone option using the incorrect timezone field. {issue}11055[11055] {pull}11164[11164]
+- Fix goroutine leak happening when harvesters are dynamically stopped. {pull}11263[11263]
 
 *Heartbeat*
 

filebeat/channel/interface.go

@@ -32,5 +32,6 @@ type Connector func(*common.Config, *common.MapStrPointer) (Outleter, error)
 // Outleter is the outlet for an input
 type Outleter interface {
  Close() error
+ Done() <-chan struct{}
  OnEvent(data *util.Data) bool
 }

filebeat/channel/outlet.go

@@ -27,25 +27,32 @@ type outlet struct {
  wg eventCounter
  client beat.Client
  isOpen atomic.Bool
+ done chan struct{}
 }
 
 func newOutlet(client beat.Client, wg eventCounter) *outlet {
  o := &outlet{
  wg: wg,
  client: client,
  isOpen: atomic.MakeBool(true),
+ done: make(chan struct{}),
  }
  return o
 }
 
 func (o *outlet) Close() error {
  isOpen := o.isOpen.Swap(false)
  if isOpen {
+ close(o.done)
  return o.client.Close()
  }
  return nil
 }
 
+func (o *outlet) Done() <-chan struct{} {
+ return o.done
+}
+
 func (o *outlet) OnEvent(d *util.Data) bool {
  if !o.isOpen.Load() {
  return false

filebeat/channel/util.go

@@ -71,6 +71,10 @@ func (o *subOutlet) Close() error {
  return nil
 }
 
+func (o *subOutlet) Done() <-chan struct{} {
+ return o.done
+}
+
 func (o *subOutlet) OnEvent(d *util.Data) bool {
 
  o.mutex.Lock()
@@ -114,8 +118,12 @@ func (o *subOutlet) OnEvent(d *util.Data) bool {
 func CloseOnSignal(outlet Outleter, sig <-chan struct{}) Outleter {
  if sig != nil {
  go func() {
- <-sig
- outlet.Close()
+ select {
+ case <-outlet.Done():
+ return
+ case <-sig:
+ outlet.Close()
+ }
  }()
  }
  return outlet

filebeat/input/log/input_other_test.go

@@ -169,3 +169,4 @@ type TestOutlet struct{}
 
 func (o TestOutlet) OnEvent(event *util.Data) bool { return true }
 func (o TestOutlet) Close() error { return nil }
+func (o TestOutlet) Done() <-chan struct{} { return nil }