Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix memory leak in Filebeat pipeline acker #12063

Merged
merged 5 commits into from
May 7, 2019
Merged

Fix memory leak in Filebeat pipeline acker #12063

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0c942e7
Fix memory leak in Filebeat pipeline acker
May 5, 2019
290c4a5
fix `empty` var scope
May 6, 2019
f830863
clarify acker closing condition
May 6, 2019
0a06af9
Add CHANGELOG
May 6, 2019
899d1b6
Update libbeat/publisher/pipeline/acker.go
jsoriano May 6, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions libbeat/publisher/pipeline/acker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,25 +133,31 @@ func (a *gapCountACK) ackLoop() {

acks, drop := a.acks, a.drop
closing := false
empty := false
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the empty variable is only used in the case n := <-acks block. As the state is changed asynchronously the 'empty' variable is MUST only be used after a call the handleACK and if closing is true. Let's reduce the scope where empty is available to the block where it is actually initialised and used.


for {
select {
case <-a.done:
closing = true
a.done = nil
if a.events.Load() == 0 {
// stop worker, if all events accounted for have been ACKed
return
}
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This one is tricky and potentially racy. The addEvent function is called asynchronously, meaning that events could still be added although the acker is already closed. It might still be ok if the client rejects events on close and guarantees that no more events will be added before the acker is closed.

Checking the client code, I found the addEvent can indeed be called after the done channel is closed and loop has been returned.

Might be ok, though. The ACKer also ignores pending ACKs if shutdown is forced.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've added a comment to explain the situation here


case <-a.pipeline.ackDone:
return

case n := <-acks:
empty := a.handleACK(n)
empty = a.handleACK(n)
if empty && closing && a.events.Load() == 0 {
// stop worker, iff all events accounted for have been ACKed
// stop worker, if all events accounted for have been ACKed
exekias marked this conversation as resolved.
Show resolved Hide resolved
return
}

case <-drop:
// TODO: accumulate multiple drop events + flush count with timer
a.events.Sub(1)
a.fn(1, 0)
}
}
Expand Down