Skip to content

Commit

Permalink
Fix a race condition on add_host_metadata (elastic#8653) (elastic#8677)
Browse files Browse the repository at this point in the history
add_host_metadata keeps a cache of the host data collected, this cache
is now updated atomically.

(cherry picked from commit 6d25fd9)
  • Loading branch information
jsoriano authored Oct 23, 2018
1 parent 588a90b commit df83ad9
Show file tree
Hide file tree
Showing 2 changed files with 41 additions and 23 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ https://github.com/elastic/beats/compare/v6.4.1...6.4[Check the HEAD diff]

*Affecting all Beats*

- Fix a race condition with the `add_host_metadata` and the event serialization. {pull}8223[8223] {pull}8653[8653]
- Fix race condition when publishing monitoring data. {pull}8646[8646]

*Auditbeat*
Expand Down
63 changes: 40 additions & 23 deletions libbeat/processors/add_host_metadata/add_host_metadata.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ package add_host_metadata
import (
"fmt"
"net"
"sync"
"time"

"github.com/joeshaw/multierror"
Expand All @@ -40,9 +41,12 @@ func init() {

type addHostMetadata struct {
info types.HostInfo
lastUpdate time.Time
data common.MapStr
config Config
lastUpdate struct {
time.Time
sync.Mutex
}
data common.MapStrPointer
config Config
}

const (
Expand All @@ -63,42 +67,55 @@ func newHostMetadataProcessor(cfg *common.Config) (processors.Processor, error)
p := &addHostMetadata{
info: h.Info(),
config: config,
data: common.NewMapStrPointer(nil),
}
p.loadData()
return p, nil
}

// Run enriches the given event with the host meta data
func (p *addHostMetadata) Run(event *beat.Event) (*beat.Event, error) {
p.loadData()
event.Fields.DeepUpdate(p.data.Clone())
event.Fields.DeepUpdate(p.data.Get().Clone())
return event, nil
}

func (p *addHostMetadata) loadData() {
func (p *addHostMetadata) expired() bool {
p.lastUpdate.Lock()
defer p.lastUpdate.Unlock()

// Check if cache is expired
if p.lastUpdate.Add(cacheExpiration).Before(time.Now()) {
p.data = host.MapHostInfo(p.info)
if p.lastUpdate.Add(cacheExpiration).After(time.Now()) {
return false
}
p.lastUpdate.Time = time.Now()
return true
}

if p.config.NetInfoEnabled {
// IP-address and MAC-address
var ipList, hwList, err = p.getNetInfo()
if err != nil {
logp.Info("Error when getting network information %v", err)
}
func (p *addHostMetadata) loadData() {
if !p.expired() {
return
}

if len(ipList) > 0 {
p.data.Put("host.ip", ipList)
}
if len(hwList) > 0 {
p.data.Put("host.mac", hwList)
}
data := host.MapHostInfo(p.info)
if p.config.NetInfoEnabled {
// IP-address and MAC-address
var ipList, hwList, err = p.getNetInfo()
if err != nil {
logp.Info("Error when getting network information %v", err)
}

if len(ipList) > 0 {
data.Put("host.ip", ipList)
}
if len(hwList) > 0 {
data.Put("host.mac", hwList)
}
p.lastUpdate = time.Now()
}

p.data.Set(data)
}

func (p addHostMetadata) getNetInfo() ([]string, []string, error) {
func (p *addHostMetadata) getNetInfo() ([]string, []string, error) {
var ipList []string
var hwList []string

Expand Down Expand Up @@ -143,7 +160,7 @@ func (p addHostMetadata) getNetInfo() ([]string, []string, error) {
return ipList, hwList, errs.Err()
}

func (p addHostMetadata) String() string {
func (p *addHostMetadata) String() string {
return fmt.Sprintf("%v=[netinfo.enabled=[%v]]",
processorName, p.config.NetInfoEnabled)
}

0 comments on commit df83ad9

Please sign in to comment.