[heartbeat] States and Improved Errors

heartbeat/monitors/factory.go

@@ -200,8 +200,8 @@ func newCommonPublishConfigs(info beat.Info, beatLocation *config.LocationWithID
 	// Generally speaking direct HB users would use the beat location, and the synthetics service may as well (TBD)
 	// while Fleet configured monitors will always use a per location monitor
 	var loc *config.LocationWithID
-	if sf.Location != nil {
-		loc = sf.Location
+	if sf.RunFrom != nil {
+		loc = sf.RunFrom
 	} else {
 		loc = beatLocation
 	}

heartbeat/monitors/mocks.go

@@ -90,10 +90,11 @@ func makeMockFactory(pluginsReg *plugin.PluginsReg) (factory *RunnerFactory, sch
 }
 
 type mockClient struct {
-	publishLog []*beat.Event
-	pipeline   beat.Pipeline
-	closed     bool
-	mtx        sync.Mutex
+	publishLog   []*beat.Event
+	pipeline     beat.Pipeline
+	closed       bool
+	mtx          sync.Mutex
+	clientConfig beat.ClientConfig
 }
 
 func (c *mockClient) IsClosed() bool {
@@ -104,6 +105,10 @@ func (c *mockClient) IsClosed() bool {
 }
 
 func (c *mockClient) Publish(e beat.Event) {
+	if c.clientConfig.Processing.Processor != nil {
+		outE, _ := c.clientConfig.Processing.Processor.Run(&e)
+		e = *outE
+	}
 	c.PublishAll([]beat.Event{e})
 }
 
@@ -148,11 +153,11 @@ func (pc *MockPipeline) Connect() (beat.Client, error) {
 	return pc.ConnectWith(beat.ClientConfig{})
 }
 
-func (pc *MockPipeline) ConnectWith(beat.ClientConfig) (beat.Client, error) {
+func (pc *MockPipeline) ConnectWith(cc beat.ClientConfig) (beat.Client, error) {
 	pc.mtx.Lock()
 	defer pc.mtx.Unlock()
 
-	c := &mockClient{pipeline: pc}
+	c := &mockClient{pipeline: pc, clientConfig: cc}
 
 	pc.Clients = append(pc.Clients, c)
 

heartbeat/monitors/stdfields/stdfields.go

@@ -50,7 +50,7 @@ type StdMonitorFields struct {
 		ZipUrl *config.C `config:"zip_url"`
 		Local  *config.C `config:"local"`
 	} `config:"source"`
-	Location *hbconfig.LocationWithID `config:"location"`
+	RunFrom *hbconfig.LocationWithID `config:"run_from"`
 }
 
 func ConfigToStdMonitorFields(conf *config.C) (StdMonitorFields, error) {

heartbeat/monitors/wrappers/monitorstate/esloader.go

@@ -56,9 +56,9 @@ func MakeESLoader(esc *elasticsearch.Client, indexPattern string, beatLocation *
 			},
 		}
 
-		if sf.Location != nil {
+		if sf.RunFrom != nil {
 			queryMustClauses = append(queryMustClauses, mapstr.M{
-				"match": mapstr.M{"observer.name": sf.Location.ID},
+				"match": mapstr.M{"observer.name": sf.RunFrom.ID},
 			})
 		}
 

heartbeat/monitors/wrappers/monitorstate/monitorstate.go

@@ -44,7 +44,7 @@ func newMonitorState(sf stdfields.StdMonitorFields, status StateStatus, ctr int,
 		// ID is unique and sortable by time for easier aggregations
 		// Note that we add an incrementing counter to help with the fact that
 		// millisecond res isn't quite enough for uniqueness (esp. in tests)
-		ID:              fmt.Sprintf("%s-%s-%x-%x", sf.ID, sf.Location, now.UnixMilli(), ctr),
+		ID:              fmt.Sprintf("%s-%s-%x-%x", sf.ID, sf.RunFrom, now.UnixMilli(), ctr),
 		StartedAt:       now,
 		DurationMs:      0,
 		Status:          status,

x-pack/heartbeat/scenarios/basics_test.go

@@ -78,3 +78,18 @@ func TestLightweightSummaries(t *testing.T) {
 		}
 	})
 }
+
+func TestRunFromOverride(t *testing.T) {
+	scenarioDB.RunAllWithATwist(t, TwistAddRunFrom, func(t *testing.T, mtr *framework.MonitorTestRun, err error) {
+		for _, e := range mtr.Events() {
+			testslike.Test(t, lookslike.MustCompile(map[string]interface{}{
+				"observer": map[string]interface{}{
+					"name": TestLocationDefault.ID,
+					"geo": map[string]interface{}{
+						"name": TestLocationDefault.Geo.Name,
+					},
+				},
+			}), e.Fields)
+		}
+	})
+}

x-pack/heartbeat/scenarios/framework/fakeloader.go

@@ -27,8 +27,8 @@ func newLoaderDB() *loaderDB {
 }
 
 func loaderDbKey(sf stdfields.StdMonitorFields) string {
-	if sf.Location != nil {
-		return fmt.Sprintf("%s-%s", sf.ID, sf.Location.ID)
+	if sf.RunFrom != nil {
+		return fmt.Sprintf("%s-%s", sf.ID, sf.RunFrom.ID)
 	}
 	return sf.ID
 }

x-pack/heartbeat/scenarios/framework/framework.go

@@ -37,7 +37,7 @@ type Scenario struct {
 	Type         string
 	Runner       ScenarioRun
 	Tags         []string
-	Location     *hbconfig.LocationWithID
+	RunFrom      *hbconfig.LocationWithID
 	NumberOfRuns int
 }
 
@@ -63,9 +63,9 @@ func MultiTwist(twists ...Twist) Twist {
 
 func (s Scenario) clone() Scenario {
 	copy := s
-	if s.Location != nil {
-		locationCopy := *s.Location
-		copy.Location = &locationCopy
+	if s.RunFrom != nil {
+		locationCopy := *s.RunFrom
+		copy.RunFrom = &locationCopy
 	}
 	return copy
 }
@@ -102,7 +102,7 @@ func (s Scenario) Run(t *testing.T, twist Twist, callback func(t *testing.T, mtr
 		var conf mapstr.M
 		for i := 0; i < numberRuns; i++ {
 			var mtr *MonitorTestRun
-			mtr, err = runMonitorOnce(t, cfgMap, runS.Location, loaderDB.StateLoader())
+			mtr, err = runMonitorOnce(t, cfgMap, runS.RunFrom, loaderDB.StateLoader())
 
 			mtr.wait()
 			events = append(events, mtr.Events()...)

x-pack/heartbeat/scenarios/framework/framework_test.go

@@ -25,7 +25,7 @@ var testScenario Scenario = Scenario{
 			"schedule": "@every 10s",
 		}, nil, nil
 	},
-	Location: &config.LocationWithID{
+	RunFrom: &config.LocationWithID{
 		ID: "TestID",
 		Geo: util.GeoConfig{
 			Name: "TestName",
@@ -39,9 +39,9 @@ func TestClone(t *testing.T) {
 	require.NotEqual(t, testScenario.Name, clone.Name)
 	clone.Tags = []string{"CloneTag"}
 	require.NotEqual(t, testScenario.Tags, clone.Tags)
-	clone.Location.ID = "CloneID"
-	require.NotEqual(t, testScenario.Location.ID, clone.Location.ID)
-	clone.Location.Geo.Name = "CloneGeoName"
-	require.NotEqual(t, testScenario.Location.Geo.Name, clone.Location.Geo.Name)
+	clone.RunFrom.ID = "CloneID"
+	require.NotEqual(t, testScenario.RunFrom.ID, clone.RunFrom.ID)
+	clone.RunFrom.Geo.Name = "CloneGeoName"
+	require.NotEqual(t, testScenario.RunFrom.Geo.Name, clone.RunFrom.Geo.Name)
 
 }

x-pack/heartbeat/scenarios/stateloader_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/elastic/beats/v7/x-pack/heartbeat/scenarios/framework"
 )
 
-var esIntegTwists = framework.MultiTwist(TwistAddLocation, TwistMultiRun(3))
+var esIntegTwists = framework.MultiTwist(TwistAddRunFrom, TwistMultiRun(3))
 
 func TestStateContinuity(t *testing.T) {
 	scenarioDB.RunAllWithATwist(t, esIntegTwists, func(t *testing.T, mtr *framework.MonitorTestRun, err error) {

x-pack/heartbeat/scenarios/twists.go

@@ -22,8 +22,8 @@ var TestLocationMpls = &config.LocationWithID{
 	},
 }
 
-var TwistAddLocation = framework.MakeTwist("add location", func(s framework.Scenario) framework.Scenario {
-	s.Location = TestLocationDefault
+var TwistAddRunFrom = framework.MakeTwist("add run_from", func(s framework.Scenario) framework.Scenario {
+	s.RunFrom = TestLocationDefault
 	return s
 })