From 671e6090e8bab23d7406554a22183318de744f81 Mon Sep 17 00:00:00 2001 From: Jaime Soriano Pastor Date: Fri, 10 May 2019 17:49:44 +0200 Subject: [PATCH] Fix goroutine leak on initialization failures of log input (#12125) (#12142) Outlets are created during log input initialization, and if it fails they were never freed. Handle this case. (cherry picked from commit 99f8ded132f3df77d710f6262b8b39a672a50a77) --- CHANGELOG.next.asciidoc | 1 + filebeat/input/log/input.go | 9 +++++++++ 2 files changed, 10 insertions(+) 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 }