[Heartbeat] Fix summarizer

heartbeat/monitors/logger/logger.go

@@ -44,6 +44,7 @@
 	Duration  int64  `json:"-"`
 	Steps     *int   `json:"steps,omitempty"`
 	Status    string `json:"status"`
+	Attempts  uint16 `json:"attempts`
 }
 
 func (m *MonitorRunInfo) MarshalJSON() ([]byte, error) {
@@ -96,6 +97,11 @@
 		errors = append(errors, err)
 	}
 
+	attempts, err := event.GetValue("summary.attempt")
+	if err != nil {
+		errors = append(errors, err)
+	}
+
 	if len(errors) > 0 {
 		return nil, fmt.Errorf("logErrors: %+v", errors)
 	}
@@ -105,6 +111,7 @@
 		Type:      monType.(string),
 		Duration:  durationUs.(int64),
 		Status:    status.(string),
+		Attempts:  attempts.(uint16),
 	}
 
 	sc, _ := event.Meta.GetValue(META_STEP_COUNT)

heartbeat/monitors/wrappers/summarizer/plugerr.go

@@ -19,6 +19,7 @@ package summarizer
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/elastic/beats/v7/heartbeat/ecserr"
 	"github.com/elastic/beats/v7/heartbeat/eventext"
@@ -31,32 +32,53 @@ import (
 // for browser monitors, preferentially using the journey/end event's
 // error field for errors.
 type BrowserErrPlugin struct {
-	summaryErrVal interface{}
+	summaryErrVal  interface{}
+	summaryErr     error
+	stepCount      int
+	journeyEndRcvd bool
 }
 
 func (esp *BrowserErrPlugin) EachEvent(event *beat.Event, eventErr error) EachEventActions {
+	// track these to determine if the journey
+	// needs an error injected due to incompleteness
+	st := synthType(event)
+	switch st {
+	case "step/end":
+		esp.stepCount++
+	case "journey/end":
+		esp.journeyEndRcvd = true
+	}
+
+	// Nothing else to do if there's no error
 	if eventErr == nil {
 		return 0
 	}
 
+	// Merge the error value into the event's "error" field
 	errVal := errToFieldVal(eventErr)
 	mergeErrVal(event, errVal)
 
-	isJourneyEnd := false
-	if synthType(event) == "journey/end" {
-		isJourneyEnd = true
-	}
-	if esp.summaryErrVal == nil || isJourneyEnd {
+	// If there is no error value OR this is the journey end event
+	// record this as the definitive error
+	if esp.summaryErrVal == nil || st == "journey/end" {
+		esp.summaryErr = eventErr
 		esp.summaryErrVal = errVal
 	}
 
 	return DropErrEvent
 }
 
 func (esp *BrowserErrPlugin) OnSummary(event *beat.Event) OnSummaryActions {
+	// If no journey end was received, make that the summary error
+	if !esp.journeyEndRcvd {
+		esp.summaryErr = fmt.Errorf("journey did not finish executing, %d steps ran: %w", esp.stepCount, esp.summaryErr)
+		esp.summaryErrVal = errToFieldVal(esp.summaryErr)
+	}
+
 	if esp.summaryErrVal != nil {
 		mergeErrVal(event, esp.summaryErrVal)
 	}
+
 	return 0
 }
 
@@ -104,16 +126,3 @@ func errToFieldVal(eventErr error) (errVal interface{}) {
 func mergeErrVal(event *beat.Event, errVal interface{}) {
 	eventext.MergeEventFields(event, mapstr.M{"error": errVal})
 }
-
-func synthType(event *beat.Event) string {
-	synthType, err := event.GetValue("synthetics.type")
-	if err != nil {
-		return ""
-	}
-
-	str, ok := synthType.(string)
-	if !ok {
-		return ""
-	}
-	return str
-}

heartbeat/monitors/wrappers/summarizer/util.go

@@ -0,0 +1,33 @@
+// 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 summarizer
+
+import "github.com/elastic/beats/v7/libbeat/beat"
+
+func synthType(event *beat.Event) string {
+	synthType, err := event.GetValue("synthetics.type")
+	if err != nil {
+		return ""
+	}
+
+	str, ok := synthType.(string)
+	if !ok {
+		return ""
+	}
+	return str
+}

x-pack/heartbeat/scenarios/stateloader_test.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/elastic/beats/v7/heartbeat/monitors/wrappers/monitorstate"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/x-pack/heartbeat/scenarios/framework"
 )
@@ -35,7 +34,7 @@ func TestStateContinuity(t *testing.T) {
 
 		lastSS := framework.LastState(mtr.Events())
 
-		assert.Equal(t, monitorstate.StatusUp, lastSS.State.Status, "monitor was unexpectedly down, synthetics console output: %s, errors", sout, errors)
+		assert.Equal(t, mtr.Meta.Status, lastSS.State.Status, "monitor had unexpected state %v, synthetics console output: %s, errors", lastSS.State.Status, sout, errors)
 
 		allSS := framework.AllStates(mtr.Events())
 		assert.Len(t, allSS, numRuns)