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

[Heartbeat] Fix Continuation Dispatch / mode: all pings #12458

Merged
merged 3 commits into from
Jun 7, 2019
Merged

[Heartbeat] Fix Continuation Dispatch / mode: all pings #12458

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a5a6c20
[Heartbeat] Fix Continuation Dispatch / mode: all pings
andrewvc Jun 6, 2019
694c6d3
Add changelog entry
andrewvc Jun 6, 2019
1882c2c
Actually reuse hasContinuations
andrewvc Jun 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
1 change: 1 addition & 0 deletions CHANGELOG.next.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d

- Fix NPE on some monitor configuration errors. {pull}11910[11910]
- Fix NPEs / resource leaks when executing config checks. {pull}11165[11165]
- Fix duplicated IPs on `mode: all` monitors. {pull}12458[12458]

*Journalbeat*

Expand Down
21 changes: 13 additions & 8 deletions heartbeat/monitors/task.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,12 +138,12 @@ func runPublishJob(job jobs.Job, client beat.Client) []scheduler.TaskFunc {
Fields: common.MapStr{},
}

next, err := job(event)
conts, err := job(event)
if err != nil {
logp.Err("Job %v failed with: ", err)
}

hasContinuations := len(next) > 0
hasContinuations := len(conts) > 0

if event.Fields != nil && !eventext.IsEventCancelled(event) {
// If continuations are present we defensively publish a clone of the event
Expand All @@ -162,15 +162,20 @@ func runPublishJob(job jobs.Job, client beat.Client) []scheduler.TaskFunc {
}
}

if len(next) == 0 {
if !hasContinuations {
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 added this here since it makes the code more readable. I know this is a bug fix, but I think some small cleanup of this function goes to that since hard to maintain code carries bug risk.

return nil
}

continuations := make([]scheduler.TaskFunc, len(next))
for i, n := range next {
continuations[i] = func() []scheduler.TaskFunc {
return runPublishJob(n, client)
contTasks := make([]scheduler.TaskFunc, len(conts))
for i, cont := range conts {
// Move the continuation into the local block scope
// This is important since execution is deferred
// Without this only the last continuation will be executed len(conts) times
localCont := cont

contTasks[i] = func() []scheduler.TaskFunc {
return runPublishJob(localCont, client)
}
}
return continuations
return contTasks
}
24 changes: 21 additions & 3 deletions heartbeat/monitors/task_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,13 @@ import (
)

func Test_runPublishJob(t *testing.T) {
simpleJob := func(event *beat.Event) (j []jobs.Job, e error) {
eventext.MergeEventFields(event, common.MapStr{"foo": "bar"})
return nil, nil
defineJob := func(fields common.MapStr) func(event *beat.Event) (j []jobs.Job, e error) {
return func(event *beat.Event) (j []jobs.Job, e error) {
eventext.MergeEventFields(event, fields)
return nil, nil
}
}
simpleJob := defineJob(common.MapStr{"foo": "bar"})

testCases := []struct {
name string
Expand All @@ -58,6 +61,21 @@ func Test_runPublishJob(t *testing.T) {
mapval.MustCompile(mapval.Map{"foo": "bar"}),
},
},
{
"multiple conts",
func(event *beat.Event) (j []jobs.Job, e error) {
simpleJob(event)
return []jobs.Job{
defineJob(common.MapStr{"baz": "bot"}),
defineJob(common.MapStr{"blah": "blargh"}),
}, nil
},
[]mapval.Validator{
mapval.MustCompile(mapval.Map{"foo": "bar"}),
mapval.MustCompile(mapval.Map{"baz": "bot"}),
mapval.MustCompile(mapval.Map{"blah": "blargh"}),
},
},
{
"cancelled cont",
func(event *beat.Event) (j []jobs.Job, e error) {
Expand Down