[heartbeat] States and Improved Errors

CHANGELOG.next.asciidoc

@@ -136,6 +136,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 
 
 *Heartbeat*
+- Add new states field for internal use by new synthetics app. {pull}30632[30632]
 
 
 *Metricbeat*

heartbeat/_meta/fields.common.yml

@@ -65,7 +65,6 @@
           type: keyword
           description: >
             Indicator if monitor could validate the service to be available.
-
         - name: check_group
           type: keyword
           description: >
@@ -89,11 +88,77 @@
             - name: id
               type: keyword
               description: Project ID
-          fields:
             - name: name
               type: keyword
               description: Project name
 
+- key: state
+  title: "Monitor state"
+  description: state related fields
+  fields:
+    - name: state
+      type: group
+      description: "Present in the last event emitted during a check. If a monitor checks multiple endpoints, as is the case with `mode: all`."
+      fields:
+        - name: id
+          type: keyword
+          description: >
+            ID of this state
+        - name: started_at
+          type: date
+          description: >
+            First time state with this ID was seen
+        - name: duration_ms
+          type: date
+          description: >
+            Length of time this state has existed in millis
+        - name: status
+          type: keyword
+          description: >
+            The current status, "up", "down", or "flapping"
+            any state can change into flapping.
+        - name: checks
+          type: integer
+          description: total checks run
+        - name: up
+          type: integer
+          description: total up checks run
+        - name: down
+          type: integer
+          description: total down checks run
+        - name: flap_history
+          enabled: false
+        - name: ends
+          type: object
+          description: the state that was ended by this state
+          fields:
+            - name: id
+              type: integer
+              description: >
+                  ID of this state
+            - name: started_at
+              type: date
+              description: >
+                  First time state with this ID was seen
+            - name: duration_ms
+              type: date
+              description: >
+                Length of time this state has existed in millis
+            - name: status
+              type: keyword
+              description: >
+                The current status, "up", "down", or "flapping"
+                any state can change into flapping.
+            - name: checks
+              type: integer
+              description: total checks run
+            - name: up
+              type: integer
+              description: total up checks run
+            - name: down
+              type: integer
+              description: total down checks run
+
 - key: summary
   title: "Monitor summary"
   description:

heartbeat/beater/heartbeat.go

@@ -20,26 +20,27 @@ package beater
 import (
 	"errors"
 	"fmt"
+
 	"syscall"
 	"time"
 
-	"github.com/elastic/beats/v7/libbeat/publisher/pipeline"
-
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/go-elasticsearch/v8"
 
 	"github.com/elastic/beats/v7/heartbeat/config"
 	"github.com/elastic/beats/v7/heartbeat/hbregistry"
 	"github.com/elastic/beats/v7/heartbeat/monitors"
 	"github.com/elastic/beats/v7/heartbeat/monitors/plugin"
+	"github.com/elastic/beats/v7/heartbeat/monitors/wrappers/monitorstate"
 	"github.com/elastic/beats/v7/heartbeat/scheduler"
+	_ "github.com/elastic/beats/v7/heartbeat/security"
 	"github.com/elastic/beats/v7/libbeat/autodiscover"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
 	"github.com/elastic/beats/v7/libbeat/common/reload"
 	"github.com/elastic/beats/v7/libbeat/management"
-
-	_ "github.com/elastic/beats/v7/heartbeat/security"
+	"github.com/elastic/beats/v7/libbeat/publisher/pipeline"
 )
 
 // Heartbeat represents the root datastructure of this beat.
@@ -49,12 +50,29 @@ type Heartbeat struct {
 	config          config.Config
 	scheduler       *scheduler.Scheduler
 	monitorReloader *cfgfile.Reloader
-	dynamicFactory  *monitors.RunnerFactory
+	monitorFactory  *monitors.RunnerFactory
 	autodiscover    *autodiscover.Autodiscover
 }
 
+type EsConfig struct {
+	Hosts    []string `config:"hosts"`
+	Username string   `config:"username"`
+	Password string   `config:"password"`
+}
+
 // New creates a new heartbeat.
 func New(b *beat.Beat, rawConfig *conf.C) (beat.Beater, error) {
+	esc, err := getESClient(b.Config.Output.Config())
+	if err != nil {
+		return nil, err
+	}
+	var stateLoader monitorstate.StateLoader
+	if esc != nil {
+		stateLoader = monitorstate.MakeESLoader(esc, "synthetics-*,heartbeat-*")
+	} else {
+		stateLoader = monitorstate.NilStateLoader
+	}
+
 	parsedConfig := config.DefaultConfig
 	if err := rawConfig.Unpack(&parsedConfig); err != nil {
 		return nil, fmt.Errorf("error reading config file: %w", err)
@@ -89,8 +107,9 @@ func New(b *beat.Beat, rawConfig *conf.C) (beat.Beater, error) {
 		done:      make(chan struct{}),
 		config:    parsedConfig,
 		scheduler: sched,
-		// dynamicFactory is the factory used for dynamic configs, e.g. autodiscover / reload
-		dynamicFactory: monitors.NewFactory(b.Info, sched.Add, plugin.GlobalPluginsReg, pipelineClientFactory),
+		// monitorFactory is the factory used for creating all monitor instances,
+		// wiring them up to everything needed to actually execute.
+		monitorFactory: monitors.NewFactory(b.Info, sched.Add, stateLoader, plugin.GlobalPluginsReg, pipelineClientFactory),
 	}
 	return bt, nil
 }
@@ -158,7 +177,7 @@ func (bt *Heartbeat) Run(b *beat.Beat) error {
 func (bt *Heartbeat) RunStaticMonitors(b *beat.Beat) (stop func(), err error) {
 	runners := make([]cfgfile.Runner, 0, len(bt.config.Monitors))
 	for _, cfg := range bt.config.Monitors {
-		created, err := bt.dynamicFactory.Create(b.Publisher, cfg)
+		created, err := bt.monitorFactory.Create(b.Publisher, cfg)
 		if err != nil {
 			if errors.Is(err, monitors.ErrMonitorDisabled) {
 				logp.L().Info("skipping disabled monitor: %s", err)
@@ -182,21 +201,21 @@ func (bt *Heartbeat) RunStaticMonitors(b *beat.Beat) (stop func(), err error) {
 
 // RunCentralMgmtMonitors loads any central management configured configs.
 func (bt *Heartbeat) RunCentralMgmtMonitors(b *beat.Beat) {
-	mons := cfgfile.NewRunnerList(management.DebugK, bt.dynamicFactory, b.Publisher)
+	mons := cfgfile.NewRunnerList(management.DebugK, bt.monitorFactory, b.Publisher)
 	reload.Register.MustRegisterList(b.Info.Beat+".monitors", mons)
-	inputs := cfgfile.NewRunnerList(management.DebugK, bt.dynamicFactory, b.Publisher)
+	inputs := cfgfile.NewRunnerList(management.DebugK, bt.monitorFactory, b.Publisher)
 	reload.Register.MustRegisterList("inputs", inputs)
 }
 
 // RunReloadableMonitors runs the `heartbeat.config.monitors` portion of the yaml config if present.
 func (bt *Heartbeat) RunReloadableMonitors() (err error) {
 	// Check monitor configs
-	if err := bt.monitorReloader.Check(bt.dynamicFactory); err != nil {
+	if err := bt.monitorReloader.Check(bt.monitorFactory); err != nil {
 		logp.Error(fmt.Errorf("error loading reloadable monitors: %w", err))
 	}
 
 	// Execute the monitor
-	go bt.monitorReloader.Run(bt.dynamicFactory)
+	go bt.monitorReloader.Run(bt.monitorFactory)
 
 	return nil
 }
@@ -206,7 +225,7 @@ func (bt *Heartbeat) makeAutodiscover(b *beat.Beat) (*autodiscover.Autodiscover,
 	ad, err := autodiscover.NewAutodiscover(
 		"heartbeat",
 		b.Publisher,
-		bt.dynamicFactory,
+		bt.monitorFactory,
 		autodiscover.QueryConfig(),
 		bt.config.Autodiscover,
 		b.Keystore,
@@ -221,3 +240,24 @@ func (bt *Heartbeat) makeAutodiscover(b *beat.Beat) (*autodiscover.Autodiscover,
 func (bt *Heartbeat) Stop() {
 	close(bt.done)
 }
+
+// getESClient returns an ES client if one is configured. Will return nil, nil, if none is configured.
+func getESClient(outputConfig *conf.C) (esc *elasticsearch.Client, err error) {
+	esConfig := EsConfig{}
+	err = outputConfig.Unpack(&esConfig)
+	if err != nil {
+		logp.L().Info("output is not elasticsearch, error / state tracking will not be enabled: %w", err)
+		return nil, nil
+	}
+	esc, err = elasticsearch.NewClient(elasticsearch.Config{
+		Addresses: esConfig.Hosts,
+		Username:  esConfig.Username,
+		Password:  esConfig.Password,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("could not initialize elasticsearch client: %w", err)
+	}
+	logp.L().Infof("successfully connected to elasticsearch for error / state tracking: %v", esConfig.Hosts)
+
+	return esc, nil
+}

heartbeat/docker-compose.yml

@@ -1,39 +1,19 @@
 version: '2.3'
 services:
-  beat:
-    build: ${PWD}/.
-    depends_on:
-    - proxy_dep
-    environment:
-    - REDIS_HOST=redis
-    - REDIS_PORT=6379
-    - ES_HOST=elasticsearch
-    - ES_USER=heartbeat_user
-    - ES_PASS=testing
-    - ES_PORT=9200
-    working_dir: /go/src/github.com/elastic/beats/heartbeat
-    volumes:
-    - ${PWD}/..:/go/src/github.com/elastic/beats/
-    # We launch docker containers to test docker autodiscover:
-    - /var/run/docker.sock:/var/run/docker.sock
-    command: make
-
   # This is a proxy used to block beats until all services are healthy.
   # See: https://github.com/docker/compose/issues/4369
   proxy_dep:
     image: busybox
     depends_on:
       elasticsearch: { condition: service_healthy }
-      redis:         { condition: service_healthy }
 
   elasticsearch:
     extends:
-      file: ${ES_BEATS}/testing/environments/${TESTING_ENVIRONMENT}.yml
+      file: ${ES_BEATS}/testing/environments/${STACK_ENVIRONMENT}.yml
       service: elasticsearch
     healthcheck:
       test: ["CMD-SHELL", "curl -u admin:testing -s http://localhost:9200/_cat/health?h=status | grep -q green"]
       retries: 300
       interval: 1s
-
-  redis:
-    build: ${PWD}/tests/docker_support/redis
+    ports:
+      - 9200:9200