Skip to content

Commit

Permalink
Cherry-pick #7820 to 6.5: Fix goroutine leak in SubOutlet (#9592)
Browse files Browse the repository at this point in the history
Cherry-pick of PR #7820 to 6.5 branch. Original message: 

This fixes a goroutine leaking every time that a harvester is closed.
  • Loading branch information
ph authored Dec 17, 2018
1 parent 3891278 commit bd8922f
Show file tree
Hide file tree
Showing 2 changed files with 25 additions and 16 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ https://github.com/elastic/beats/compare/v6.5.3...6.5[Check the HEAD diff]
*Filebeat*

- Fix saved objects in filebeat haproxy dashboard. {pull}9417[9417]
- Fixed a memory leak when harvesters are closed. {pull}7820[7820]

*Heartbeat*

Expand Down
40 changes: 24 additions & 16 deletions filebeat/channel/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,17 +18,19 @@
package channel

import (
"sync"

"github.com/elastic/beats/filebeat/util"
"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/common/atomic"
)

type subOutlet struct {
isOpen atomic.Bool
done chan struct{}
ch chan *util.Data
res chan bool
done chan struct{}
ch chan *util.Data
res chan bool
mutex sync.Mutex
closeOnce sync.Once
}

// ConnectTo creates a new Connector, combining a beat.Pipeline with an outlet Factory.
Expand All @@ -42,10 +44,9 @@ func ConnectTo(pipeline beat.Pipeline, factory Factory) Connector {
// underlying outlet.
func SubOutlet(out Outleter) Outleter {
s := &subOutlet{
isOpen: atomic.MakeBool(true),
done: make(chan struct{}),
ch: make(chan *util.Data),
res: make(chan bool, 1),
done: make(chan struct{}),
ch: make(chan *util.Data),
res: make(chan bool, 1),
}

go func() {
Expand All @@ -58,21 +59,30 @@ func SubOutlet(out Outleter) Outleter {
}

func (o *subOutlet) Close() error {
isOpen := o.isOpen.Swap(false)
if isOpen {
o.closeOnce.Do(func() {
// Signal OnEvent() to terminate
close(o.done)
}
// This mutex prevents the event channel to be closed if OnEvent is
// still running.
o.mutex.Lock()
defer o.mutex.Unlock()
close(o.ch)
})
return nil
}

func (o *subOutlet) OnEvent(d *util.Data) bool {
if !o.isOpen.Load() {

o.mutex.Lock()
defer o.mutex.Unlock()
select {
case <-o.done:
return false
default:
}

select {
case <-o.done:
close(o.ch)
return false

case o.ch <- d:
Expand All @@ -92,8 +102,6 @@ func (o *subOutlet) OnEvent(d *util.Data) bool {
// Once all messages are in the publisher pipeline, in correct order,
// it depends on registrar/publisher pipeline if state is finally updated
// in the registrar.

close(o.ch)
return true

case ret := <-o.res:
Expand Down

0 comments on commit bd8922f

Please sign in to comment.