From 4ab7dce2cd52c07f0943d17dfed94d11e9b9d420 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20P=C3=A9rez-Aradros=20Herce?= Date: Fri, 10 May 2019 13:10:39 +0200 Subject: [PATCH] Cherry-pick #12063 to 7.0: Fix memory leak in Filebeat pipeline acker (#12140) * Fix memory leak in Filebeat pipeline acker (#12063) * Fix memory leak in Filebeat pipeline acker Before this change acker goroutine was kept forever as processed events count was not correctly updated. Filebeat sends an empty event to update file states, this event is not published, but treated as dropped, without updating counters. This change makes sures that `a.events` count gets updated for dropped events also, so the acker gets closed after all ACKs happen. (cherry picked from commit 965310573289806993cec957e5c9391bb3612e04) * Update CHANGELOG.next.asciidoc --- CHANGELOG.next.asciidoc | 2 ++ libbeat/publisher/pipeline/acker.go | 9 ++++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index ac7dc75b7fd..7509bb0917b 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -36,6 +36,8 @@ https://github.com/elastic/beats/compare/v7.0.0...7.0[Check the HEAD diff] *Filebeat* +- Fix memory leak in Filebeat pipeline acker. {pull}12063[12063] + *Heartbeat* *Journalbeat* diff --git a/libbeat/publisher/pipeline/acker.go b/libbeat/publisher/pipeline/acker.go index 1d4155677ac..5689b86d445 100644 --- a/libbeat/publisher/pipeline/acker.go +++ b/libbeat/publisher/pipeline/acker.go @@ -139,6 +139,12 @@ func (a *gapCountACK) ackLoop() { case <-a.done: closing = true a.done = nil + if a.events.Load() == 0 { + // stop worker, if all events accounted for have been ACKed. + // If new events are added after this acker won't handle them, which may + // result in duplicates + return + } case <-a.pipeline.ackDone: return @@ -146,12 +152,13 @@ func (a *gapCountACK) ackLoop() { case n := <-acks: empty := a.handleACK(n) if empty && closing && a.events.Load() == 0 { - // stop worker, iff all events accounted for have been ACKed + // stop worker, if and only if all events accounted for have been ACKed return } case <-drop: // TODO: accumulate multiple drop events + flush count with timer + a.events.Sub(1) a.fn(1, 0) } }