Support shipping directly to monitoring cluster

CHANGELOG.next.asciidoc

@@ -89,6 +89,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - New processor: `copy_fields`. {pull}11303[11303]
 - Add `error.message` to events when `fail_on_error` is set in `rename` and `copy_fields` processors. {pull}11303[11303]
 - New processor: `truncate_fields`. {pull}11297[11297]
+- Allow a beat to ship monitoring data directly to an Elasticsearch monitoring clsuter. {pull}9260[9260]
 
 *Auditbeat*
 

libbeat/cmd/instance/beat.go

@@ -39,6 +39,10 @@ import (
 	errw "github.com/pkg/errors"
 	"go.uber.org/zap"
 
+	sysinfo "github.com/elastic/go-sysinfo"
+	"github.com/elastic/go-sysinfo/types"
+	ucfg "github.com/elastic/go-ucfg"
+
 	"github.com/elastic/beats/libbeat/api"
 	"github.com/elastic/beats/libbeat/asset"
 	"github.com/elastic/beats/libbeat/beat"
@@ -66,9 +70,6 @@ import (
 	"github.com/elastic/beats/libbeat/publisher/processing"
 	svc "github.com/elastic/beats/libbeat/service"
 	"github.com/elastic/beats/libbeat/version"
-	sysinfo "github.com/elastic/go-sysinfo"
-	"github.com/elastic/go-sysinfo/types"
-	ucfg "github.com/elastic/go-ucfg"
 )
 
 // Beat provides the runnable and configurable instance of a beat.
@@ -102,8 +103,10 @@ type beatConfig struct {
 	Keystore      *common.Config `config:"keystore"`
 
 	// output/publishing related configurations
-	Pipeline   pipeline.Config `config:",inline"`
-	Monitoring *common.Config  `config:"xpack.monitoring"`
+	Pipeline pipeline.Config `config:",inline"`
+
+	// monitoring settings
+	MonitoringBeatConfig monitoring.BeatConfig `config:",inline"`
 
 	// central management settings
 	Management *common.Config `config:"management"`
@@ -284,6 +287,11 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 		return nil, err
 	}
 
+	err = b.registerClusterUUIDFetching()
+	if err != nil {
+		return nil, err
+	}
+
 	reg := monitoring.Default.GetRegistry("libbeat")
 	if reg == nil {
 		reg = monitoring.Default.NewRegistry("libbeat")
@@ -361,11 +369,17 @@ func (b *Beat) launch(settings Settings, bt beat.Creator) error {
 		return err
 	}
 
-	if b.Config.Monitoring.Enabled() {
+	monitoringCfg, reporterSettings, err := monitoring.SelectConfig(b.Config.MonitoringBeatConfig)
+	if err != nil {
+		return err
+	}
+
+	if monitoringCfg.Enabled() {
 		settings := report.Settings{
 			DefaultUsername: settings.Monitoring.DefaultUsername,
+			Format:          reporterSettings.Format,
 		}
-		reporter, err := report.New(b.Info, settings, b.Config.Monitoring, b.Config.Output)
+		reporter, err := report.New(b.Info, settings, monitoringCfg, b.Config.Output)
 		if err != nil {
 			return err
 		}
@@ -759,8 +773,7 @@ func (b *Beat) registerESIndexManagement() error {
 	return nil
 }
 
-// Build and return a callback to load index template into ES
-func (b *Beat) indexSetupCallback() func(esClient *elasticsearch.Client) error {
+func (b *Beat) indexSetupCallback() elasticsearch.ConnectCallback {
 	return func(esClient *elasticsearch.Client) error {
 		m := b.index.Manager(esClient, idxmgmt.BeatsAssets(b.Fields))
 		return m.Setup(true, true)
@@ -790,6 +803,47 @@ func (b *Beat) createOutput(stats outputs.Observer, cfg common.ConfigNamespace)
 	return outputs.Load(b.index, b.Info, stats, cfg.Name(), cfg.Config())
 }
 
+func (b *Beat) registerClusterUUIDFetching() error {
+	if b.Config.Output.Name() == "elasticsearch" {
+		callback, err := b.clusterUUIDFetchingCallback()
+		if err != nil {
+			return err
+		}
+		elasticsearch.RegisterConnectCallback(callback)
+	}
+	return nil
+}
+
+// Build and return a callback to fetch the Elasticsearch cluster_uuid for monitoring
+func (b *Beat) clusterUUIDFetchingCallback() (elasticsearch.ConnectCallback, error) {
+	stateRegistry := monitoring.GetNamespace("state").GetRegistry()
+	elasticsearchRegistry := stateRegistry.NewRegistry("outputs.elasticsearch")
+	clusterUUIDRegVar := monitoring.NewString(elasticsearchRegistry, "cluster_uuid")
+
+	callback := func(esClient *elasticsearch.Client) error {
+		var response struct {
+			ClusterUUID string `json:"cluster_uuid"`
+		}
+
+		status, body, err := esClient.Request("GET", "/", "", nil, nil)
+		if err != nil {
+			return errw.Wrap(err, "error querying /")
+		}
+		if status > 299 {
+			return fmt.Errorf("Error querying /. Status: %d. Response body: %s", status, body)
+		}
+		err = json.Unmarshal(body, &response)
+		if err != nil {
+			return fmt.Errorf("Error unmarshaling json when querying /. Body: %s", body)
+		}
+
+		clusterUUIDRegVar.Set(response.ClusterUUID)
+		return nil
+	}
+
+	return callback, nil
+}
+
 // handleError handles the given error by logging it and then returning the
 // error. If the err is nil or is a GracefulExit error then the method will
 // return nil without logging anything.

libbeat/monitoring/monitoring.go

@@ -17,7 +17,19 @@
 
 package monitoring
 
-import "errors"
+import (
+	"errors"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	"github.com/elastic/beats/libbeat/monitoring/report"
+)
+
+// BeatConfig represents the part of the $BEAT.yml to do with monitoring settings
+type BeatConfig struct {
+	XPackMonitoring *common.Config `config:"xpack.monitoring"`
+	Monitoring      *common.Config `config:"monitoring"`
+}
 
 type Mode uint8
 
@@ -30,6 +42,11 @@ const (
 	Full
 )
 
+var (
+	errMonitoringBothConfigEnabled = errors.New("both xpack.monitoring.* and monitoring.* cannot be set. Prefer to set monitoring.* and set monitoring.elasticsearch.hosts to monitoring cluster hosts")
+	warnMonitoringDeprecatedConfig = "xpack.monitoring.* settings are deprecated. Use monitoring.* instead, but set monitoring.elasticsearch.hosts to monitoring cluster hosts"
+)
+
 // Default is the global default metrics registry provided by the monitoring package.
 var Default = NewRegistry()
 
@@ -67,3 +84,21 @@ func Remove(name string) {
 func Clear() error {
 	return Default.Clear()
 }
+
+// SelectConfig selects the appropriate monitoring configuration based on the user's settings in $BEAT.yml. Users may either
+// use xpack.monitoring.* settings OR monitoring.* settings but not both.
+func SelectConfig(beatCfg BeatConfig) (*common.Config, *report.Settings, error) {
+	switch {
+	case beatCfg.Monitoring.Enabled() && beatCfg.XPackMonitoring.Enabled():
+		return nil, nil, errMonitoringBothConfigEnabled
+	case beatCfg.XPackMonitoring.Enabled():
+		cfgwarn.Deprecate("7.0", warnMonitoringDeprecatedConfig)
+		monitoringCfg := beatCfg.XPackMonitoring
+		return monitoringCfg, &report.Settings{Format: report.FormatXPackMonitoringBulk}, nil
+	case beatCfg.Monitoring.Enabled():
+		monitoringCfg := beatCfg.Monitoring
+		return monitoringCfg, &report.Settings{Format: report.FormatBulk}, nil
+	default:
+		return nil, nil, nil
+	}
+}

libbeat/monitoring/report/elasticsearch/client.go

@@ -20,6 +20,7 @@ package elasticsearch
 import (
 	"encoding/json"
 	"fmt"
+	"time"
 
 	"github.com/pkg/errors"
 
@@ -34,17 +35,20 @@ import (
 type publishClient struct {
 	es     *esout.Client
 	params map[string]string
+	format report.Format
 }
 
 func newPublishClient(
 	es *esout.Client,
 	params map[string]string,
-) *publishClient {
+	format report.Format,
+) (*publishClient, error) {
 	p := &publishClient{
 		es:     es,
 		params: params,
+		format: format,
 	}
-	return p
+	return p, nil
 }
 
 func (c *publishClient) Connect() error {
@@ -84,6 +88,7 @@ func (c *publishClient) Connect() error {
 	}
 
 	debugf("XPack monitoring is enabled")
+
 	return nil
 }
 
@@ -97,13 +102,18 @@ func (c *publishClient) Publish(batch publisher.Batch) error {
 	var reason error
 	for _, event := range events {
 
-		// Extract time
+		// Extract type
 		t, err := event.Content.Meta.GetValue("type")
 		if err != nil {
 			logp.Err("Type not available in monitoring reported. Please report this error: %s", err)
 			continue
 		}
 
+		typ, ok := t.(string)
+		if !ok {
+			logp.Err("monitoring type is not a string")
+		}
+
 		var params = map[string]string{}
 		// Copy params
 		for k, v := range c.params {
@@ -120,23 +130,13 @@ func (c *publishClient) Publish(batch publisher.Batch) error {
 			}
 		}
 
-		meta := common.MapStr{
-			"_index":   "",
-			"_routing": nil,
-			"_type":    t,
-		}
-		bulk := [2]interface{}{
-			common.MapStr{"index": meta},
-			report.Event{
-				Timestamp: event.Content.Timestamp,
-				Fields:    event.Content.Fields,
-			},
+		switch c.format {
+		case report.FormatXPackMonitoringBulk:
+			err = c.publishXPackBulk(params, event, typ)
+		case report.FormatBulk:
+			err = c.publishBulk(event, typ)
 		}
 
-		// Currently one request per event is sent. Reason is that each event can contain different
-		// interval params and X-Pack requires to send the interval param.
-		_, err = c.es.SendMonitoringBulk(params, bulk[:])
-
 		if err != nil {
 			failed = append(failed, event)
 			reason = err
@@ -158,3 +158,75 @@ func (c *publishClient) Test(d testing.Driver) {
 func (c *publishClient) String() string {
 	return "publish(" + c.es.String() + ")"
 }
+
+func (c *publishClient) publishXPackBulk(params map[string]string, event publisher.Event, typ string) error {
+	meta := common.MapStr{
+		"_index":   "",
+		"_routing": nil,
+		"_type":    typ,
+	}
+	bulk := [2]interface{}{
+		common.MapStr{"index": meta},
+		report.Event{
+			Timestamp: event.Content.Timestamp,
+			Fields:    event.Content.Fields,
+		},
+	}
+
+	// Currently one request per event is sent. Reason is that each event can contain different
+	// interval params and X-Pack requires to send the interval param.
+	_, err := c.es.SendMonitoringBulk(params, bulk[:])
+	return err
+}
+
+func (c *publishClient) publishBulk(event publisher.Event, typ string) error {
+	meta := common.MapStr{
+		"_index":   getMonitoringIndexName(),
+		"_routing": nil,
+	}
+
+	if c.es.GetVersion().Major < 7 {
+		meta["_type"] = "doc"
+	}
+
+	action := common.MapStr{
+		"index": meta,
+	}
+
+	event.Content.Fields.Put("timestamp", event.Content.Timestamp)
+
+	fields := common.MapStr{
+		"type": typ,
+		typ:    event.Content.Fields,
+	}
+
+	interval, err := event.Content.Meta.GetValue("interval_ms")
+	if err != nil {
+		return errors.Wrap(err, "could not determine interval_ms field")
+	}
+	fields.Put("interval_ms", interval)
+
+	clusterUUID, err := event.Content.Meta.GetValue("cluster_uuid")
+	if err != nil && err != common.ErrKeyNotFound {
+		return errors.Wrap(err, "could not determine cluster_uuid field")
+	}
+	fields.Put("cluster_uuid", clusterUUID)
+
+	document := report.Event{
+		Timestamp: event.Content.Timestamp,
+		Fields:    fields,
+	}
+	bulk := [2]interface{}{action, document}
+
+	// Currently one request per event is sent. Reason is that each event can contain different
+	// interval params and X-Pack requires to send the interval param.
+	// FIXME: index name (first param below)
+	_, err = c.es.BulkWith(getMonitoringIndexName(), "", nil, nil, bulk[:])
+	return err
+}
+
+func getMonitoringIndexName() string {
+	version := 7
+	date := time.Now().Format("2006.01.02")
+	return fmt.Sprintf(".monitoring-beats-%v-%s", version, date)
+}

libbeat/monitoring/report/elasticsearch/config.go

@@ -20,6 +20,8 @@ package elasticsearch
 import (
 	"time"
 
+	"github.com/elastic/beats/libbeat/monitoring/report"
+
 	"github.com/elastic/beats/libbeat/common/transport/tlscommon"
 )
 
@@ -43,6 +45,7 @@ type config struct {
 	BufferSize       int               `config:"buffer_size"`
 	Tags             []string          `config:"tags"`
 	Backoff          backoff           `config:"backoff"`
+	Format           report.Format     `config:"_format"`
 }
 
 type backoff struct {