Fix goroutine leak on initialization failures of log input (elastic#1…

…2125)

Outlets are created during log input initialization, and if it
fails they were never freed. Handle this case.

CHANGELOG.next.asciidoc

@@ -92,6 +92,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add missing Kubernetes metadata fields to Filebeat CoreDNS module, and fix a documentation error. {pull}11591[11591]
 - Reduce memory usage if long lines are truncated to fit `max_bytes` limit. The line buffer is copied into a smaller buffer now. This allows the runtime to release unused memory earlier. {pull}11524[11524]
 - Fix memory leak in Filebeat pipeline acker. {pull}12063[12063]
+- Fix goroutine leak caused on initialization failures of log input. {pull}12125[12125]
 
 *Heartbeat*
 

filebeat/input/log/input.go

@@ -76,6 +76,12 @@ func NewInput(
  outlet channel.Connector,
  context input.Context,
 ) (input.Input, error) {
+ cleanupNeeded := true
+ cleanupIfNeeded := func(f func() error) {
+ if cleanupNeeded {
+ f()
+ }
+ }
 
  // Note: underlying output.
  // The input and harvester do have different requirements
@@ -87,11 +93,13 @@ func NewInput(
  if err != nil {
  return nil, err
  }
+ defer cleanupIfNeeded(out.Close)
 
  // stateOut will only be unblocked if the beat is shut down.
  // otherwise it can block on a full publisher pipeline, so state updates
  // can be forwarded correctly to the registrar.
  stateOut := channel.CloseOnSignal(channel.SubOutlet(out), context.BeatDone)
+ defer cleanupIfNeeded(stateOut.Close)
 
  meta := context.Meta
  if len(meta) == 0 {
@@ -137,6 +145,7 @@ func NewInput(
 
  logp.Info("Configured paths: %v", p.config.Paths)
 
+ cleanupNeeded = false
  return p, nil
 }