diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index 7298b388919..72f3214644d 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -44,6 +44,7 @@ https://github.com/elastic/beats/compare/v6.7.2...6.8[Check the HEAD diff] - Fix goroutine leak happening when harvesters are dynamically stopped. {pull}11263[11263] - Fix initialization of the TCP input logger. {pull}11605[11605] +- Fix goroutine leak caused on initialization failures of log input. {pull}12125[12125] *Heartbeat* diff --git a/filebeat/input/log/input.go b/filebeat/input/log/input.go index cc8d575f8a8..6d5cf35ee69 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 }