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]: catch all error handler for browser jobs #30013

Merged
merged 4 commits into from
Feb 1, 2022
Merged

[Heartbeat]: catch all error handler for browser jobs #30013

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b6c95f7
[Heartbeat]: catch all error handler for browser jobs
vigneshshanmugam Jan 26, 2022
d145c0d
fix wrapper tests
vigneshshanmugam Jan 26, 2022
0d6f438
add tests and consume all events
vigneshshanmugam Jan 29, 2022
c0b4cc0
address review and improve test
vigneshshanmugam Feb 1, 2022
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
24 changes: 12 additions & 12 deletions heartbeat/monitors/wrappers/wrappers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,23 +97,23 @@ func addMonitorMetaFields(event *beat.Event, started time.Time, sf stdfields.Std
id = fmt.Sprintf("%s-%x", sf.ID, urlHash)
}

// Allow jobs to override the ID, useful for browser suites
// which do this logic on their own
if v, _ := event.GetValue("monitor.id"); v != nil {
id = fmt.Sprintf("%s-%s", sf.ID, v.(string))
}
if v, _ := event.GetValue("monitor.name"); v != nil {
name = fmt.Sprintf("%s - %s", sf.Name, v.(string))
}

fieldsToMerge := common.MapStr{
"monitor": common.MapStr{
"id": id,
"name": name,
"type": sf.Type,
"timespan": timespan(started, sf.Schedule, sf.Timeout),
},
}
// Browser monitor fields are enriched separately
// in the synthexec package - x-pack/heartbeat/monitors/browser/synthexec/enrich.go
if sf.Type != "browser" {
vigneshshanmugam marked this conversation as resolved.
Show resolved Hide resolved
fieldsToMerge = common.MapStr{
"monitor": common.MapStr{
"id": id,
"name": name,
"type": sf.Type,
"timespan": timespan(started, sf.Schedule, sf.Timeout),
},
}
}

// Add service.name for APM interop
if sf.Service.Name != "" {
Expand Down
57 changes: 33 additions & 24 deletions heartbeat/monitors/wrappers/wrappers_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,6 @@ var testMonFields = stdfields.StdMonitorFields{
}

var testBrowserMonFields = stdfields.StdMonitorFields{
ID: "myid",
Name: "myname",
Type: "browser",
Schedule: schedule.MustParse("@every 1s"),
Timeout: 1,
Expand Down Expand Up @@ -396,23 +394,37 @@ func TestTimespan(t *testing.T) {
}
}

type BrowserMonitor struct {
id string
name string
checkGroup string
}

var inlineMonitorValues = BrowserMonitor{
id: "inline",
name: "inline",
checkGroup: "inline-check-group",
}

func makeInlineBrowserJob(t *testing.T, u string) jobs.Job {
parsed, err := url.Parse(u)
require.NoError(t, err)
return func(event *beat.Event) (i []jobs.Job, e error) {
eventext.MergeEventFields(event, common.MapStr{
"url": URLFields(parsed),
"monitor": common.MapStr{
"check_group": "inline-check-group",
"type": "browser",
"id": inlineMonitorValues.id,
"name": inlineMonitorValues.name,
"check_group": inlineMonitorValues.checkGroup,
},
})
return nil, nil
}
}

// Inline browser jobs function very similarly to lightweight jobs
// in that they don't override the ID.
// They do not, however, get a summary field added, nor duration.
// Browser inline jobs monitor information should not be altered
// by the wrappers as they are handled separately in synth enricher
func TestInlineBrowserJob(t *testing.T) {
fields := testBrowserMonFields
testCommonWrap(t, testDef{
Expand All @@ -425,10 +437,10 @@ func TestInlineBrowserJob(t *testing.T) {
urlValidator(t, "http://foo.com"),
lookslike.MustCompile(map[string]interface{}{
"monitor": map[string]interface{}{
"id": testMonFields.ID,
"name": testMonFields.Name,
"type": fields.Type,
"check_group": "inline-check-group",
"type": "browser",
"id": inlineMonitorValues.id,
"name": inlineMonitorValues.name,
"check_group": inlineMonitorValues.checkGroup,
},
}),
hbtestllext.MonitorTimespanValidator,
Expand All @@ -439,13 +451,9 @@ func TestInlineBrowserJob(t *testing.T) {
})
}

var suiteBrowserJobValues = struct {
id string
name string
checkGroup string
}{
id: "journey_1",
name: "Journey 1",
var suiteMonitorValues = BrowserMonitor{
id: "suite-journey_1",
name: "suite-Journey 1",
checkGroup: "journey-1-check-group",
}

Expand All @@ -456,9 +464,10 @@ func makeSuiteBrowserJob(t *testing.T, u string, summary bool, suiteErr error) j
eventext.MergeEventFields(event, common.MapStr{
"url": URLFields(parsed),
"monitor": common.MapStr{
"id": suiteBrowserJobValues.id,
"name": suiteBrowserJobValues.name,
"check_group": suiteBrowserJobValues.checkGroup,
"type": "browser",
"id": suiteMonitorValues.id,
"name": suiteMonitorValues.name,
"check_group": suiteMonitorValues.checkGroup,
},
})
if summary {
Expand All @@ -482,10 +491,10 @@ func TestSuiteBrowserJob(t *testing.T) {
urlU, _ := url.Parse(urlStr)
expectedMonFields := lookslike.MustCompile(map[string]interface{}{
"monitor": map[string]interface{}{
"id": fmt.Sprintf("%s-%s", testMonFields.ID, suiteBrowserJobValues.id),
"name": fmt.Sprintf("%s - %s", testMonFields.Name, suiteBrowserJobValues.name),
"type": fields.Type,
"check_group": suiteBrowserJobValues.checkGroup,
"type": "browser",
"id": suiteMonitorValues.id,
"name": suiteMonitorValues.name,
"check_group": suiteMonitorValues.checkGroup,
"timespan": common.MapStr{
"gte": hbtestllext.IsTime,
"lt": hbtestllext.IsTime,
Expand Down
14 changes: 12 additions & 2 deletions x-pack/heartbeat/monitors/browser/suite.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,16 @@ func (s *Suite) FilterJourneys() synthexec.FilterJourneyConfig {
return s.suiteCfg.FilterJourneys
}

func (s *Suite) Fields() synthexec.StandardSuiteFields {
_, inline := s.InlineSource()
vigneshshanmugam marked this conversation as resolved.
Show resolved Hide resolved
return synthexec.StandardSuiteFields{
Name: s.suiteCfg.Name,
Id: s.suiteCfg.Id,
Inline: inline,
Type: "browser",
}
}

func (s *Suite) Close() error {
if s.suiteCfg.Source.ActiveMemo != nil {
s.suiteCfg.Source.ActiveMemo.Close()
Expand Down Expand Up @@ -102,14 +112,14 @@ func (s *Suite) extraArgs() []string {
func (s *Suite) jobs() []jobs.Job {
var j jobs.Job
if src, ok := s.InlineSource(); ok {
j = synthexec.InlineJourneyJob(context.TODO(), src, s.Params(), s.extraArgs()...)
j = synthexec.InlineJourneyJob(context.TODO(), src, s.Params(), s.Fields(), s.extraArgs()...)
} else {
j = func(event *beat.Event) ([]jobs.Job, error) {
err := s.Fetch()
if err != nil {
return nil, fmt.Errorf("could not fetch for suite job: %w", err)
}
sj, err := synthexec.SuiteJob(context.TODO(), s.Workdir(), s.Params(), s.FilterJourneys(), s.extraArgs()...)
sj, err := synthexec.SuiteJob(context.TODO(), s.Workdir(), s.Params(), s.FilterJourneys(), s.Fields(), s.extraArgs()...)
if err != nil {
return nil, err
}
Expand Down
38 changes: 29 additions & 9 deletions x-pack/heartbeat/monitors/browser/synthexec/enrich.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,18 +18,18 @@ import (
"github.com/elastic/beats/v7/libbeat/common"
)

type enricher func(event *beat.Event, se *SynthEvent) error
type enricher func(event *beat.Event, se *SynthEvent, fields StandardSuiteFields) error

type streamEnricher struct {
je *journeyEnricher
}

func (e *streamEnricher) enrich(event *beat.Event, se *SynthEvent) error {
func (e *streamEnricher) enrich(event *beat.Event, se *SynthEvent, fields StandardSuiteFields) error {
if e.je == nil || (se != nil && se.Type == "journey/start") {
e.je = newJourneyEnricher()
}

return e.je.enrich(event, se)
return e.je.enrich(event, se, fields)
}

// journeyEnricher holds state across received SynthEvents retaining fields
Expand Down Expand Up @@ -62,7 +62,7 @@ func makeUuid() string {
return u.String()
}

func (je *journeyEnricher) enrich(event *beat.Event, se *SynthEvent) error {
func (je *journeyEnricher) enrich(event *beat.Event, se *SynthEvent, fields StandardSuiteFields) error {
if se == nil {
return nil
}
Expand All @@ -84,20 +84,40 @@ func (je *journeyEnricher) enrich(event *beat.Event, se *SynthEvent) error {
}

eventext.MergeEventFields(event, common.MapStr{
"event": common.MapStr{
"type": se.Type,
},
"monitor": common.MapStr{
"check_group": je.checkGroup,
},
})
// Inline jobs have no journey

// Id and name differs for inline and suite monitors
// - We use the monitor id and name for inline journeys
// - Monitor id/name is concatenated with the journey id/name for
// suite journeys
id := fields.Id
name := fields.Name
if je.journey != nil {
id = fmt.Sprintf("%s-%s", id, je.journey.Id)
name = fmt.Sprintf("%s - %s", name, je.journey.Name)
}
eventext.MergeEventFields(event, common.MapStr{
"monitor": common.MapStr{
"id": id,
"name": name,
},
})

// Write suite level fields for suite jobs
if !fields.Inline {
eventext.MergeEventFields(event, common.MapStr{
"monitor": common.MapStr{
"id": je.journey.Id,
"name": je.journey.Name,
"suite": common.MapStr{
"id": id,
"name": name,
},
})
}

return je.enrichSynthEvent(event, se)
}

Expand Down
6 changes: 4 additions & 2 deletions x-pack/heartbeat/monitors/browser/synthexec/enrich_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,8 +84,9 @@ func TestJourneyEnricher(t *testing.T) {
// on the nil data.
for idx, se := range synthEvents {
e := &beat.Event{}
stdFields := StandardSuiteFields{Inline: false}
t.Run(fmt.Sprintf("event %d", idx), func(t *testing.T) {
enrichErr := je.enrich(e, se)
enrichErr := je.enrich(e, se, stdFields)

if se != nil && se.Type != "journey/end" {
// Test that the created event includes the mapped
Expand Down Expand Up @@ -258,8 +259,9 @@ func TestNoSummaryOnAfterHook(t *testing.T) {

for idx, se := range synthEvents {
e := &beat.Event{}
stdFields := StandardSuiteFields{Inline: false}
t.Run(fmt.Sprintf("event %d", idx), func(t *testing.T) {
enrichErr := je.enrich(e, se)
enrichErr := je.enrich(e, se, stdFields)

if se != nil && se.Type == "cmd/status" {
t.Run("no summary in cmd/status", func(t *testing.T) {
Expand Down
16 changes: 8 additions & 8 deletions x-pack/heartbeat/monitors/browser/synthexec/execmultiplexer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,20 +28,20 @@ func (e ExecMultiplexer) writeSynthEvent(se *SynthEvent) {
e.currentJourney.Store(true)
e.eventCounter.Store(-1)
}
hasCurrentJourney := e.currentJourney.Load()
se.index = e.eventCounter.Inc()

if se.Type == "journey/end" || se.Type == "cmd/status" {
e.currentJourney.Store(false)
}

se.index = e.eventCounter.Inc()
if hasCurrentJourney {
e.synthEvents <- se
}
e.synthEvents <- se
vigneshshanmugam marked this conversation as resolved.
Show resolved Hide resolved
Copy link
Contributor

Choose a reason for hiding this comment

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

There's a weird thing to consider, which is errors between journeys.

We could still need to check for journey/end to account for errors that might happen between journeys. I think we can just set e.currentJourney.store(false) again, no? Then it just gets reported as an error on the suite.

Our long term goal is for heartbeat to really only execute one journey per monitor (with discovery mode) so I'm OK with not handling this for the moment and just tacking those errors on the previous journey. Also, I don't know what an error between journeys would even be.

Copy link
Member Author

Choose a reason for hiding this comment

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

I am not sure if there would be error between journeys, But with this change we would still get stderr messages on the suite level, wouldnt that be enough to figure out any errors?

}

// SynthEvents returns a read only channel for synth events
func (e ExecMultiplexer) SynthEvents() <-chan *SynthEvent {
return e.synthEvents
func (e ExecMultiplexer) SynthEvents(inline bool) <-chan *SynthEvent {
if inline || e.currentJourney.Load() {
return e.synthEvents
}
return make(chan *SynthEvent)
vigneshshanmugam marked this conversation as resolved.
Show resolved Hide resolved
}

// Done returns a channel that is closed when all output has been received
Expand Down
Loading