Enhanced Heartbeat Telemetry (#8621) (#8719)

Add heartbeat telemetry.

We now track:

1. The total number of active monitors
2. The total number of active monitors per protocol
3. The total number of checked endpoints per protocol
4. Counter versions of the above stats for the stats endpoint

This also removes an accidentally comitted test case that tests nothing that broke with these refactors.

(cherry picked from commit 596a0dd)

heartbeat/monitors/active/http/http.go

@@ -39,15 +39,15 @@ var debugf = logp.MakeDebug("http")
 func create(
 	name string,
 	cfg *common.Config,
-) ([]monitors.Job, error) {
+) (jobs []monitors.Job, endpoints int, err error) {
 	config := defaultConfig
 	if err := cfg.Unpack(&config); err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	tls, err := outputs.LoadTLSConfig(config.TLS)
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	var body []byte
@@ -58,44 +58,44 @@ func create(
 		compression := config.Check.Request.Compression
 		enc, err = getContentEncoder(compression.Type, compression.Level)
 		if err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 
 		buf := bytes.NewBuffer(nil)
 		err = enc.Encode(buf, bytes.NewBufferString(config.Check.Request.SendBody))
 		if err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 
 		body = buf.Bytes()
 	}
 
 	validator := makeValidateResponse(&config.Check.Response)
 
-	jobs := make([]monitors.Job, len(config.URLs))
+	jobs = make([]monitors.Job, len(config.URLs))
 
 	if config.ProxyURL != "" {
 		transport, err := newRoundTripper(&config, tls)
 		if err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 
 		for i, url := range config.URLs {
 			jobs[i], err = newHTTPMonitorHostJob(url, &config, transport, enc, body, validator)
 			if err != nil {
-				return nil, err
+				return nil, 0, err
 			}
 		}
 	} else {
 		for i, url := range config.URLs {
 			jobs[i], err = newHTTPMonitorIPsJob(&config, url, tls, enc, body, validator)
 			if err != nil {
-				return nil, err
+				return nil, 0, err
 			}
 		}
 	}
 
-	return jobs, nil
+	return jobs, len(config.URLs), nil
 }
 
 func newRoundTripper(config *Config, tls *transport.TLSConfig) (*http.Transport, error) {

heartbeat/monitors/active/http/http_test.go

@@ -54,14 +54,16 @@ func testTLSRequest(t *testing.T, testURL string, certPath string) beat.Event {
 	config, err := common.NewConfigFrom(configSrc)
 	require.NoError(t, err)
 
-	jobs, err := create("tls", config)
+	jobs, endpoints, err := create("tls", config)
 	require.NoError(t, err)
 
 	job := jobs[0]
 
 	event, _, err := job.Run()
 	require.NoError(t, err)
 
+	require.Equal(t, 1, endpoints)
+
 	return event
 }
 

heartbeat/monitors/active/icmp/icmp.go

@@ -37,10 +37,10 @@ var debugf = logp.MakeDebug("icmp")
 func create(
 	name string,
 	cfg *common.Config,
-) ([]monitors.Job, error) {
+) (jobs []monitors.Job, endpoints int, err error) {
 	config := DefaultConfig
 	if err := cfg.Unpack(&config); err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	// TODO: check icmp is support by OS + check we've
@@ -49,7 +49,6 @@ func create(
 	// TODO: replace icmp package base reader/sender using raw sockets with
 	//       OS specific solution
 
-	var jobs []monitors.Job
 	addJob := func(t monitors.Job, err error) error {
 		if err != nil {
 			return err
@@ -61,7 +60,7 @@ func create(
 	ipVersion := config.Mode.Network()
 	if len(config.Hosts) > 0 && ipVersion == "" {
 		err := fmt.Errorf("pinging hosts requires ipv4 or ipv6 mode enabled")
-		return nil, err
+		return nil, 0, err
 	}
 
 	var loopErr error
@@ -71,11 +70,11 @@ func create(
 	})
 	if loopErr != nil {
 		debugf("Failed to initialize ICMP loop %v", loopErr)
-		return nil, loopErr
+		return nil, 0, loopErr
 	}
 
 	if err := loop.checkNetworkMode(ipVersion); err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	network := config.Mode.Network()
@@ -90,11 +89,11 @@ func create(
 		settings := monitors.MakeHostJobSettings(jobName, host, config.Mode)
 		err := addJob(monitors.MakeByHostJob(settings, pingFactory))
 		if err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 	}
 
-	return jobs, nil
+	return jobs, len(config.Hosts), nil
 }
 
 func createPingIPFactory(config *Config) func(*net.IPAddr) (common.MapStr, error) {

heartbeat/monitors/active/tcp/tcp.go

@@ -47,33 +47,32 @@ type connURL struct {
 func create(
 	name string,
 	cfg *common.Config,
-) ([]monitors.Job, error) {
+) (jobs []monitors.Job, endpoints int, err error) {
 	config := DefaultConfig
 	if err := cfg.Unpack(&config); err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	tls, err := outputs.LoadTLSConfig(config.TLS)
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	defaultScheme := "tcp"
 	if tls != nil {
 		defaultScheme = "ssl"
 	}
 
-	endpoints, err := collectHosts(&config, defaultScheme)
+	schemeHosts, err := collectHosts(&config, defaultScheme)
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	typ := config.Name
 	timeout := config.Timeout
 	validator := makeValidateConn(&config)
 
-	var jobs []monitors.Job
-	for scheme, eps := range endpoints {
+	for scheme, eps := range schemeHosts {
 		schemeTLS := tls
 		if scheme == "tcp" || scheme == "plain" {
 			schemeTLS = nil
@@ -85,20 +84,25 @@ func create(
 			TLS:     schemeTLS,
 		})
 		if err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 
 		epJobs, err := dialchain.MakeDialerJobs(db, typ, scheme, eps, config.Mode,
 			func(dialer transport.Dialer, addr string) (common.MapStr, error) {
 				return pingHost(dialer, addr, timeout, validator)
 			})
 		if err != nil {
-			return nil, err
+			return nil, 0, err
 		}
 
 		jobs = append(jobs, epJobs...)
 	}
-	return jobs, nil
+
+	numHosts := 0
+	for _, hosts := range schemeHosts {
+		numHosts += len(hosts)
+	}
+	return jobs, numHosts, nil
 }
 
 func collectHosts(config *Config, defaultScheme string) (map[string][]dialchain.Endpoint, error) {

heartbeat/monitors/active/tcp/tcp_test.go

@@ -45,14 +45,16 @@ func testTCPCheck(t *testing.T, host string, port uint16) *beat.Event {
 	})
 	require.NoError(t, err)
 
-	jobs, err := create("tcp", config)
+	jobs, endpoints, err := create("tcp", config)
 	require.NoError(t, err)
 
 	job := jobs[0]
 
 	event, _, err := job.Run()
 	require.NoError(t, err)
 
+	require.Equal(t, 1, endpoints)
+
 	return &event
 }
 
@@ -65,14 +67,16 @@ func testTLSTCPCheck(t *testing.T, host string, port uint16, certFileName string
 	})
 	require.NoError(t, err)
 
-	jobs, err := create("tcp", config)
+	jobs, endpoints, err := create("tcp", config)
 	require.NoError(t, err)
 
 	job := jobs[0]
 
 	event, _, err := job.Run()
 	require.NoError(t, err)
 
+	require.Equal(t, 1, endpoints)
+
 	return &event
 }
 

heartbeat/monitors/mocks_test.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/monitoring"
 )
 
 type MockBeatClient struct {
@@ -99,11 +100,13 @@ func createMockJob(name string, cfg *common.Config) ([]Job, error) {
 }
 
 func mockPluginBuilder() pluginBuilder {
-	return pluginBuilder{"test", ActiveMonitor, func(s string, config *common.Config) ([]Job, error) {
+	reg := monitoring.NewRegistry()
+
+	return pluginBuilder{"test", ActiveMonitor, func(s string, config *common.Config) ([]Job, int, error) {
 		c := common.Config{}
 		j, err := createMockJob("test", &c)
-		return j, err
-	}}
+		return j, 1, err
+	}, newPluginCountersRecorder("test", reg)}
 }
 
 func mockPluginsReg() *pluginsReg {

heartbeat/monitors/monitor.go

@@ -42,6 +42,8 @@ type Monitor struct {
 	scheduler  *scheduler.Scheduler
 	jobTasks   []*task
 	enabled    bool
+	// endpoints is a count of endpoints this monitor measures.
+	endpoints int
 	// internalsMtx is used to synchronize access to critical
 	// internal datastructures
 	internalsMtx sync.Mutex
@@ -51,6 +53,10 @@ type Monitor struct {
 	watch          watcher.Watch
 
 	pipelineConnector beat.PipelineConnector
+
+	// stats is the countersRecorder used to record lifecycle events
+	// for global metrics + telemetry
+	stats registryRecorder
 }
 
 // String prints a description of the monitor in a threadsafe way. It is important that this use threadsafe
@@ -95,9 +101,11 @@ func newMonitor(
 		watchPollTasks:    []*task{},
 		internalsMtx:      sync.Mutex{},
 		config:            config,
+		stats:             monitorPlugin.stats,
 	}
 
-	jobs, err := monitorPlugin.create(config)
+	jobs, endpoints, err := monitorPlugin.create(config)
+	m.endpoints = endpoints
 	if err != nil {
 		return nil, fmt.Errorf("job err %v", err)
 	}
@@ -181,7 +189,8 @@ func (m *Monitor) makeWatchTasks(monitorPlugin pluginBuilder) error {
 					return
 				}
 
-				watchJobs, err := monitorPlugin.create(merged)
+				watchJobs, endpoints, err := monitorPlugin.create(merged)
+				m.endpoints = endpoints
 				if err != nil {
 					logp.Err("Could not create job from watch file: %v", err)
 				}
@@ -227,6 +236,8 @@ func (m *Monitor) Start() {
 	for _, t := range m.watchPollTasks {
 		t.Start()
 	}
+
+	m.stats.startMonitor(int64(m.endpoints))
 }
 
 // Stop stops the Monitor's execution in its configured scheduler.
@@ -242,4 +253,6 @@ func (m *Monitor) Stop() {
 	for _, t := range m.watchPollTasks {
 		t.Stop()
 	}
+
+	m.stats.stopMonitor(int64(m.endpoints))
 }