Enhanced Heartbeat Telemetry

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))
 }