Support "cluster" scope in Metricbeat elasticsearch module

CHANGELOG.next.asciidoc

@@ -610,6 +610,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Added cache and connection_errors metrics to status metricset of MySQL module {issue}16955[16955] {pull}19844[19844]
 - Update MySQL dashboard with connection errors and cache metrics {pull}19913[19913] {issue}16955[16955]
 - Add cloud.instance.name into aws ec2 metricset. {pull}20077[20077]
+- Add `scope` setting for elasticsearch module, allowing it to monitor an Elasticsearch cluster behind a load-balancing proxy. {issue}18539[18539] {pull}18547[18547]
 
 *Packetbeat*
 

metricbeat/docs/modules/elasticsearch.asciidoc

@@ -13,7 +13,16 @@ The `elasticsearch` module collects metrics about {es}.
 The `elasticsearch` module works with {es} 6.7.0 and later.
 
 [float]
-=== Usage for Stack Monitoring
+=== Module-specific configuration notes
+
+Like other {beatname_uc} modules, the `elasticsearch` module accepts a `hosts` configuration setting.
+This setting can contain a list of entries. The related `scope` setting determines how each entry in
+the `hosts` list is interpreted by the module.
+
+* If `scope` is set to `node` (default), each entry in the `hosts` list indicates a distinct node in an
+  {es} cluster.
+* If `scope` is set to `cluster`, each entry in the `hosts` list indicates a single endpoint for a distinct
+  {es} cluster (for example, a load-balancing proxy fronting the cluster).
 
 The `elasticsearch` module can be used to collect metrics shown in our {stack} {monitor-features}
 UI in {kib}. To enable this usage, set `xpack.enabled: true` and remove any `metricsets`
@@ -45,12 +54,9 @@ metricbeat.modules:
   #password: "changeme"
   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
-  # Set to false to fetch all entries
   #index_recovery.active_only: true
-
-  # Set to true to send data collected by module to X-Pack
-  # Monitoring instead of metricbeat-* indices.
   #xpack.enabled: false
+  #scope: node
 ----
 
 This module supports TLS connections when using `ssl` config field, as described in <<configuration-ssl>>.

metricbeat/metricbeat.reference.yml

@@ -273,12 +273,9 @@ metricbeat.modules:
   #password: "changeme"
   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
-  # Set to false to fetch all entries
   #index_recovery.active_only: true
-
-  # Set to true to send data collected by module to X-Pack
-  # Monitoring instead of metricbeat-* indices.
   #xpack.enabled: false
+  #scope: node
 
 #------------------------------ Envoyproxy Module ------------------------------
 - module: envoyproxy

metricbeat/module/elasticsearch/_meta/config.reference.yml

@@ -13,9 +13,6 @@
   #password: "changeme"
   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
 
-  # Set to false to fetch all entries
   #index_recovery.active_only: true
-
-  # Set to true to send data collected by module to X-Pack
-  # Monitoring instead of metricbeat-* indices.
   #xpack.enabled: false
+  #scope: node

metricbeat/module/elasticsearch/_meta/docs.asciidoc

@@ -6,7 +6,16 @@ The `elasticsearch` module collects metrics about {es}.
 The `elasticsearch` module works with {es} 6.7.0 and later.
 
 [float]
-=== Usage for Stack Monitoring
+=== Module-specific configuration notes
+
+Like other {beatname_uc} modules, the `elasticsearch` module accepts a `hosts` configuration setting.
+This setting can contain a list of entries. The related `scope` setting determines how each entry in
+the `hosts` list is interpreted by the module.
+
+* If `scope` is set to `node` (default), each entry in the `hosts` list indicates a distinct node in an
+  {es} cluster.
+* If `scope` is set to `cluster`, each entry in the `hosts` list indicates a single endpoint for a distinct
+  {es} cluster (for example, a load-balancing proxy fronting the cluster).
 
 The `elasticsearch` module can be used to collect metrics shown in our {stack} {monitor-features}
 UI in {kib}. To enable this usage, set `xpack.enabled: true` and remove any `metricsets`

metricbeat/module/elasticsearch/ccr/ccr.go

@@ -56,14 +56,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch gathers stats for each follower shard from the _ccr/stats API
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.GetServiceURI())
+	shouldSkip, err := m.ShouldSkipFetch()
 	if err != nil {
-		return errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		return err
 	}
-
-	// Not master, no event sent
-	if !isMaster {
-		m.Logger().Debug("trying to fetch ccr stats from a non-master node")
+	if shouldSkip {
 		return nil
 	}
 

metricbeat/module/elasticsearch/cluster_stats/cluster_stats.go

@@ -18,8 +18,6 @@
 package cluster_stats
 
 import (
-	"github.com/pkg/errors"
-
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/module/elasticsearch"
 )
@@ -51,14 +49,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch methods implements the data gathering and data conversion to the right format
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.HostData().SanitizedURI+clusterStatsPath)
+	shouldSkip, err := m.ShouldSkipFetch()
 	if err != nil {
-		return errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		return err
 	}
-
-	// Not master, no event sent
-	if !isMaster {
-		m.Logger().Debug("trying to fetch cluster stats from a non-master node")
+	if shouldSkip {
 		return nil
 	}
 

metricbeat/module/elasticsearch/elasticsearch.go

@@ -436,6 +436,28 @@ func IsMLockAllEnabled(http *helper.HTTP, resetURI, nodeID string) (bool, error)
 	return false, fmt.Errorf("could not determine if mlockall is enabled on node ID = %v", nodeID)
 }
 
+// GetMasterNodeID returns the ID of the Elasticsearch cluster's master node
+func GetMasterNodeID(http *helper.HTTP, resetURI string) (string, error) {
+	content, err := fetchPath(http, resetURI, "_nodes/_master", "filter_path=nodes.*.name")
+	if err != nil {
+		return "", err
+	}
+
+	var response struct {
+		Nodes map[string]interface{} `json:"nodes"`
+	}
+
+	if err := json.Unmarshal(content, &response); err != nil {
+		return "", err
+	}
+
+	for nodeID, _ := range response.Nodes {
+		return nodeID, nil
+	}
+
+	return "", errors.New("could not determine master node ID")
+}
+
 // PassThruField copies the field at the given path from the given source data object into
 // the same path in the given target data object.
 func PassThruField(fieldPath string, sourceData, targetData common.MapStr) error {

metricbeat/module/elasticsearch/enrich/enrich.go

@@ -55,14 +55,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch gathers stats for each enrich coordinator node
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.GetServiceURI())
+	shouldSkip, err := m.ShouldSkipFetch()
 	if err != nil {
-		return errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		return err
 	}
-
-	// Not master, no event sent
-	if !isMaster {
-		m.Logger().Debug("trying to fetch enrich stats from a non-master node")
+	if shouldSkip {
 		return nil
 	}
 

metricbeat/module/elasticsearch/index/index.go

@@ -59,15 +59,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch gathers stats for each index from the _stats API
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
-
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.HostData().SanitizedURI+statsPath)
+	shouldSkip, err := m.ShouldSkipFetch()
 	if err != nil {
-		return errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		return err
 	}
-
-	// Not master, no event sent
-	if !isMaster {
-		m.Logger().Debug("trying to fetch index stats from a non-master node")
+	if shouldSkip {
 		return nil
 	}
 

metricbeat/module/elasticsearch/index_recovery/index_recovery.go

@@ -18,8 +18,6 @@
 package index_recovery
 
 import (
-	"github.com/pkg/errors"
-
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/module/elasticsearch"
 )
@@ -67,14 +65,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch gathers stats for each index from the _stats API
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.GetServiceURI())
+	shouldSkip, err := m.ShouldSkipFetch()
 	if err != nil {
-		return errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		return err
 	}
-
-	// Not master, no event sent
-	if !isMaster {
-		m.Logger().Debug("trying to fetch index recovery stats from a non-master node")
+	if shouldSkip {
 		return nil
 	}
 

metricbeat/module/elasticsearch/index_summary/index_summary.go

@@ -62,14 +62,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch gathers stats for each index from the _stats API
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.HostData().SanitizedURI+statsPath)
+	shouldSkip, err := m.ShouldSkipFetch()
 	if err != nil {
-		return errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		return err
 	}
-
-	// Not master, no event sent
-	if !isMaster {
-		m.Logger().Debug("trying to fetch index summary stats from a non-master node")
+	if shouldSkip {
 		return nil
 	}
 

metricbeat/module/elasticsearch/metricset.go

@@ -18,6 +18,10 @@
 package elasticsearch
 
 import (
+	"fmt"
+
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/v7/metricbeat/helper"
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/mb/parse"
@@ -36,13 +40,39 @@ var (
 	}.Build()
 )
 
+type Scope int
+
+const (
+	// Indicates that each item in the hosts list points to a distinct Elasticsearch node in a
+	// cluster.
+	ScopeNode Scope = iota
+
+	// Indicates that each item in the hosts lists points to a endpoint for a distinct Elasticsearch
+	// cluster (e.g. a load-balancing proxy) fronting the cluster.
+	ScopeCluster
+)
+
+func (h *Scope) Unpack(str string) error {
+	switch str {
+	case "node":
+		*h = ScopeNode
+	case "cluster":
+		*h = ScopeCluster
+	default:
+		return fmt.Errorf("invalid scope: %v", str)
+	}
+
+	return nil
+}
+
 // MetricSet can be used to build other metric sets that query RabbitMQ
 // management plugin
 type MetricSet struct {
 	mb.BaseMetricSet
 	servicePath string
 	*helper.HTTP
 	XPack bool
+	Scope Scope
 }
 
 // NewMetricSet creates an metric set that can be used to build other metric
@@ -54,9 +84,11 @@ func NewMetricSet(base mb.BaseMetricSet, servicePath string) (*MetricSet, error)
 	}
 
 	config := struct {
-		XPack bool `config:"xpack.enabled"`
+		XPack bool  `config:"xpack.enabled"`
+		Scope Scope `config:"scope"`
 	}{
 		XPack: false,
+		Scope: ScopeNode,
 	}
 	if err := base.Module().UnpackConfig(&config); err != nil {
 		return nil, err
@@ -67,6 +99,7 @@ func NewMetricSet(base mb.BaseMetricSet, servicePath string) (*MetricSet, error)
 		servicePath,
 		http,
 		config.XPack,
+		config.Scope,
 	}
 
 	ms.SetServiceURI(servicePath)
@@ -84,3 +117,22 @@ func (m *MetricSet) SetServiceURI(servicePath string) {
 	m.servicePath = servicePath
 	m.HTTP.SetURI(m.GetServiceURI())
 }
+
+func (m *MetricSet) ShouldSkipFetch() (bool, error) {
+	// If we're talking to a set of ES nodes directly, only collect stats from the master node so
+	// we don't collect the same stats from every node and end up duplicating them.
+	if m.Scope == ScopeNode {
+		isMaster, err := IsMaster(m.HTTP, m.GetServiceURI())
+		if err != nil {
+			return false, errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		}
+
+		// Not master, no event sent
+		if !isMaster {
+			m.Logger().Debugf("trying to fetch %v stats from a non-master node", m.Name())
+			return true, nil
+		}
+	}
+
+	return false, nil
+}

metricbeat/module/elasticsearch/ml_job/ml_job.go

@@ -18,8 +18,6 @@
 package ml_job
 
 import (
-	"github.com/pkg/errors"
-
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/module/elasticsearch"
 )
@@ -54,15 +52,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch methods implements the data gathering and data conversion to the right format
 func (m *MetricSet) Fetch(r mb.ReporterV2) error {
-
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.GetServiceURI())
+	shouldSkip, err := m.ShouldSkipFetch()
 	if err != nil {
-		return errors.Wrap(err, "error determining if connected Elasticsearch node is master")
+		return err
 	}
-
-	// Not master, no event sent
-	if !isMaster {
-		m.Logger().Debug("trying to fetch machine learning job stats from a non-master node")
+	if shouldSkip {
 		return nil
 	}
 

metricbeat/module/elasticsearch/node_stats/data_xpack.go

@@ -187,18 +187,14 @@ func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info,
 		return errors.Wrap(err, "failure parsing Elasticsearch Node Stats API response")
 	}
 
-	// Normally the nodeStruct should only contain one node. But if _local is removed
-	// from the path and Metricbeat is not installed on the same machine as the node
-	// it will provid the data for multiple nodes. This will mean the detection of the
-	// master node will not be accurate anymore as often in these cases a proxy is in front
-	// of ES and it's not know if the request will be routed to the same node as before.
+	masterNodeID, err := elasticsearch.GetMasterNodeID(m.HTTP, m.HTTP.GetURI())
+	if err != nil {
+		return err
+	}
+
 	var errs multierror.Errors
 	for nodeID, node := range nodesStruct.Nodes {
-		isMaster, err := elasticsearch.IsMaster(m.HTTP, m.HTTP.GetURI())
-		if err != nil {
-			errs = append(errs, errors.Wrap(err, "error determining if connected Elasticsearch node is master"))
-			continue
-		}
+		isMaster := nodeID == masterNodeID
 
 		event := mb.Event{}
 
@@ -207,6 +203,7 @@ func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info,
 			errs = append(errs, errors.Wrap(err, "failure to apply node schema"))
 			continue
 		}
+
 		nodeData["node_master"] = isMaster
 		nodeData["node_id"] = nodeID