[Heartbeat] Fix incorrect 'Up' status for all mode

heartbeat/eventext/eventext.go

@@ -25,8 +25,30 @@ import (
 // MergeEventFields merges the given common.MapStr into the given Event's Fields.
 func MergeEventFields(e *beat.Event, merge common.MapStr) {
 	if e.Fields != nil {
-		e.Fields.DeepUpdate(merge)
+		e.Fields.DeepUpdate(merge.Clone())
 	} else {
-		e.Fields = merge
+		e.Fields = merge.Clone()
 	}
 }
+
+// EventCancelledMetaKey is the path to the @metadata key marking an event as cancelled.
+const EventCancelledMetaKey = "__hb_evt_cancel__"
+
+// CancelEvent marks the event as cancelled. Downstream consumers of it should not emit nor output this event.
+func CancelEvent(event *beat.Event) {
+	if event != nil {
+		if event.Meta == nil {
+			event.Meta = common.MapStr{}
+		}
+		event.Meta.Put(EventCancelledMetaKey, true)
+	}
+}
+
+// IsEventCancelled checks for the marker left by CancelEvent.
+func IsEventCancelled(event *beat.Event) bool {
+	if event == nil || event.Meta == nil {
+		return false
+	}
+	v, err := event.Meta.GetValue(EventCancelledMetaKey)
+	return err == nil && v == true
+}

heartbeat/monitors/task.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/pkg/errors"
 
+	"github.com/elastic/beats/heartbeat/eventext"
 	"github.com/elastic/beats/heartbeat/monitors/jobs"
 	"github.com/elastic/beats/heartbeat/scheduler"
 	"github.com/elastic/beats/heartbeat/scheduler/schedule"
@@ -86,42 +87,7 @@ func (e ProcessorsError) Error() string {
 
 func (t *configuredJob) prepareSchedulerJob(job jobs.Job) scheduler.TaskFunc {
 	return func() []scheduler.TaskFunc {
-		event := &beat.Event{
-			Fields: common.MapStr{},
-		}
-		next, err := job(event)
-		hasContinuations := len(next) > 0
-
-		if err != nil {
-			logp.Err("Job %v failed with: ", err)
-		}
-
-		if event != nil && event.Fields != nil {
-			// If continuations are present we defensively publish a clone of the event
-			// in the chance that the event shares underlying data with the events for continuations
-			// This prevents races where the pipeline publish could accidentally alter multiple events.
-			if hasContinuations {
-				clone := beat.Event{
-					Timestamp: event.Timestamp,
-					Meta:      event.Meta.Clone(),
-					Fields:    event.Fields.Clone(),
-				}
-				t.client.Publish(clone)
-			} else {
-				// no clone needed if no continuations
-				t.client.Publish(*event)
-			}
-		}
-
-		if len(next) == 0 {
-			return nil
-		}
-
-		continuations := make([]scheduler.TaskFunc, len(next))
-		for i, n := range next {
-			continuations[i] = t.prepareSchedulerJob(n)
-		}
-		return continuations
+		return runPublishJob(job, t.client)
 	}
 }
 
@@ -166,3 +132,45 @@ func (t *configuredJob) Stop() {
 		t.client.Close()
 	}
 }
+
+func runPublishJob(job jobs.Job, client beat.Client) []scheduler.TaskFunc {
+	event := &beat.Event{
+		Fields: common.MapStr{},
+	}
+
+	next, err := job(event)
+	if err != nil {
+		logp.Err("Job %v failed with: ", err)
+	}
+
+	hasContinuations := len(next) > 0
+
+	if event.Fields != nil && !eventext.IsEventCancelled(event) {
+		// If continuations are present we defensively publish a clone of the event
+		// in the chance that the event shares underlying data with the events for continuations
+		// This prevents races where the pipeline publish could accidentally alter multiple events.
+		if hasContinuations {
+			clone := beat.Event{
+				Timestamp: event.Timestamp,
+				Meta:      event.Meta.Clone(),
+				Fields:    event.Fields.Clone(),
+			}
+			client.Publish(clone)
+		} else {
+			// no clone needed if no continuations
+			client.Publish(*event)
+		}
+	}
+
+	if len(next) == 0 {
+		return nil
+	}
+
+	continuations := make([]scheduler.TaskFunc, len(next))
+	for i, n := range next {
+		continuations[i] = func() []scheduler.TaskFunc {
+			return runPublishJob(n, client)
+		}
+	}
+	return continuations
+}

heartbeat/monitors/task_test.go

@@ -0,0 +1,96 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package monitors
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/heartbeat/eventext"
+	"github.com/elastic/beats/heartbeat/monitors/jobs"
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/mapval"
+)
+
+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
+	}
+
+	testCases := []struct {
+		name       string
+		job        jobs.Job
+		validators []mapval.Validator
+	}{
+		{
+			"simple",
+			simpleJob,
+			[]mapval.Validator{
+				mapval.MustCompile(mapval.Map{"foo": "bar"}),
+			},
+		},
+		{
+			"one cont",
+			func(event *beat.Event) (j []jobs.Job, e error) {
+				simpleJob(event)
+				return []jobs.Job{simpleJob}, nil
+			},
+			[]mapval.Validator{
+				mapval.MustCompile(mapval.Map{"foo": "bar"}),
+				mapval.MustCompile(mapval.Map{"foo": "bar"}),
+			},
+		},
+		{
+			"cancelled cont",
+			func(event *beat.Event) (j []jobs.Job, e error) {
+				eventext.CancelEvent(event)
+				return []jobs.Job{simpleJob}, nil
+			},
+			[]mapval.Validator{
+				mapval.MustCompile(mapval.Map{"foo": "bar"}),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			client := &MockBeatClient{}
+			queue := runPublishJob(tc.job, client)
+			for {
+				if len(queue) == 0 {
+					break
+				}
+				tf := queue[0]
+				queue = queue[1:]
+				conts := tf()
+				for _, cont := range conts {
+					queue = append(queue, cont)
+				}
+			}
+			client.Close()
+
+			require.Len(t, client.publishes, len(tc.validators))
+			for idx, event := range client.publishes {
+				mapval.Test(t, tc.validators[idx], event.Fields)
+			}
+		})
+	}
+}

heartbeat/monitors/util.go

@@ -23,6 +23,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/elastic/beats/heartbeat/eventext"
 	"github.com/elastic/beats/heartbeat/look"
 	"github.com/elastic/beats/heartbeat/monitors/jobs"
 	"github.com/elastic/beats/heartbeat/monitors/wrappers"
@@ -211,7 +212,11 @@ func makeByHostAllIPJob(
 			ipFields := resolveIPEvent(ip.String(), resolveRTT)
 			cont[i] = wrappers.WithFields(ipFields, pingFactory(addr))
 		}
-		return cont, nil
+		// Ideally we would test this invocation. This function however is really hard to to test given all the extra context it takes in
+		// In a future refactor we could perhaps test that this in correctly invoked.
+		eventext.CancelEvent(event)
+
+		return cont, err
 	}
 }
 

heartbeat/monitors/wrappers/monitors.go

@@ -155,13 +155,18 @@ func makeAddSummary() jobs.JobWrapper {
 			state.mtx.Lock()
 			defer state.mtx.Unlock()
 
-			// After each job
-			eventStatus, _ := event.GetValue("monitor.status")
-			if eventStatus == "up" {
-				state.up++
-			} else {
-				state.down++
+			// If the event is cancelled we don't record it as being either up or down since
+			// we discard the event anyway.
+			if !eventext.IsEventCancelled(event) {
+				// After each job
+				eventStatus, _ := event.GetValue("monitor.status")
+				if eventStatus == "up" {
+					state.up++
+				} else {
+					state.down++
+				}
 			}
+
 			// No error check needed here
 			event.PutValue("monitor.check_group", state.checkGroup)