diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index 9146e8aca1e..4ae8ab7195e 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -39,6 +39,7 @@ https://github.com/elastic/beats/compare/v7.0.0...7.1[Check the HEAD diff] *Filebeat* +- Fix goroutine leak caused on initialization failures of log input. {pull}12125[12125] - Fix memory leak in Filebeat pipeline acker. {pull}12063[12063] *Heartbeat* diff --git a/filebeat/input/log/input.go b/filebeat/input/log/input.go index 875c3a6b104..52ab225a4e6 100644 --- a/filebeat/input/log/input.go +++ b/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 }