Report kibana_settings to X-Pack Monitoring (#7664)

Resolves #7621.

Depends on elastic/kibana#21100.

X-Pack Monitoring of Kibana requires two types of documents in the `.monitoring-kibana-*` indices: `kibana_stats` and `kibana_settings`. We made Metricbeat's `kibana/stats` metricset index `kibana_stats` documents into `.monitoring-kibana-*` in #7525. This PR makes the same metricset index `kibana_settings` documents into `.monitoring-kibana-*`.

metricbeat/module/kibana/kibana.go

@@ -29,8 +29,13 @@ import (
 	"github.com/elastic/beats/metricbeat/mb"
 )
 
-// StatsAPIAvailableVersion is the version of Kibana since when the stats API is available
-const StatsAPIAvailableVersion = "6.4.0"
+const (
+	// StatsAPIAvailableVersion is the version of Kibana since when the stats API is available
+	StatsAPIAvailableVersion = "6.4.0"
+
+	// SettingsAPIAvailableVersion is the version of Kibana since when the settings API is available
+	SettingsAPIAvailableVersion = "6.5.0"
+)
 
 // ReportErrorForMissingField reports and returns an error message for the given
 // field being missing in API response received from Kibana
@@ -67,21 +72,30 @@ func GetVersion(http *helper.HTTP, currentPath string) (string, error) {
 	return versionStr, nil
 }
 
-// IsStatsAPIAvailable returns whether the stats API is available in the given version of Kibana
-func IsStatsAPIAvailable(kibanaVersion string) (bool, error) {
-	currentVersion, err := common.NewVersion(kibanaVersion)
+func isKibanaAPIAvailable(currentKibanaVersion, apiAvailableInKibanaVersion string) (bool, error) {
+	currentVersion, err := common.NewVersion(currentKibanaVersion)
 	if err != nil {
 		return false, err
 	}
 
-	wantVersion, err := common.NewVersion(StatsAPIAvailableVersion)
+	wantVersion, err := common.NewVersion(apiAvailableInKibanaVersion)
 	if err != nil {
 		return false, err
 	}
 
 	return !currentVersion.LessThan(wantVersion), nil
 }
 
+// IsStatsAPIAvailable returns whether the stats API is available in the given version of Kibana
+func IsStatsAPIAvailable(currentKibanaVersion string) (bool, error) {
+	return isKibanaAPIAvailable(currentKibanaVersion, StatsAPIAvailableVersion)
+}
+
+// IsSettingsAPIAvailable returns whether the settings API is available in the given version of Kibana
+func IsSettingsAPIAvailable(currentKibanaVersion string) (bool, error) {
+	return isKibanaAPIAvailable(currentKibanaVersion, SettingsAPIAvailableVersion)
+}
+
 func fetchPath(http *helper.HTTP, currentPath, newPath string) ([]byte, error) {
 	currentURI := http.GetURI()
 	defer http.SetURI(currentURI) // Reset after this request

metricbeat/module/kibana/stats/_meta/test/settings.700.json

@@ -0,0 +1,18 @@
+{
+    "cluster_uuid":"u5ii0pnQRka_P0gimfmthg",
+    "settings":{
+      "xpack":{
+        "default_admin_email":"jane@doe.com"
+      },
+      "kibana":{
+        "uuid":"5b2de169-2785-441b-ae8c-186a1936b17d",
+        "name":"Janes-MBP-2",
+        "index":".kibana",
+        "host":"localhost",
+        "transport_address":"localhost:5601",
+        "version":"7.0.0-alpha1",
+        "snapshot":false,
+        "status":"green"
+      }
+    }
+  }

metricbeat/module/kibana/stats/data_xpack.go

@@ -29,7 +29,7 @@ import (
 )
 
 var (
-	schemaXPackMonitoring = s.Schema{
+	schemaXPackMonitoringStats = s.Schema{
 		"concurrent_connections": c.Int("concurrent_connections"),
 		"os": c.Dict("os", s.Schema{
 			"load": c.Dict("load", s.Schema{
@@ -105,61 +105,90 @@ var (
 	}
 )
 
-func eventMappingXPack(r mb.ReporterV2, intervalMs int64, content []byte) error {
-	var data map[string]interface{}
-	err := json.Unmarshal(content, &data)
-	if err != nil {
-		r.Error(err)
-		return err
+type dataParser func(mb.ReporterV2, common.MapStr, time.Time) (string, string, common.MapStr, error)
+
+func statsDataParser(r mb.ReporterV2, data common.MapStr, now time.Time) (string, string, common.MapStr, error) {
+	clusterUUID, ok := data["clusterUuid"].(string)
+	if !ok {
+		return "", "", nil, elastic.ReportErrorForMissingField("clusterUuid", elastic.Kibana, r)
 	}
 
-	kibanaStatsFields, err := schemaXPackMonitoring.Apply(data)
+	kibanaStatsFields, err := schemaXPackMonitoringStats.Apply(data)
 	if err != nil {
-		r.Error(err)
-		return err
+		return "", "", nil, err
 	}
 
 	process, ok := data["process"].(map[string]interface{})
 	if !ok {
-		return elastic.ReportErrorForMissingField("process", elastic.Kibana, r)
+		return "", "", nil, elastic.ReportErrorForMissingField("process", elastic.Kibana, r)
 	}
 	memory, ok := process["memory"].(map[string]interface{})
 	if !ok {
-		return elastic.ReportErrorForMissingField("process.memory", elastic.Kibana, r)
+		return "", "", nil, elastic.ReportErrorForMissingField("process.memory", elastic.Kibana, r)
 	}
-
 	rss, ok := memory["resident_set_size_bytes"].(float64)
 	if !ok {
-		return elastic.ReportErrorForMissingField("process.memory.resident_set_size_bytes", elastic.Kibana, r)
+		return "", "", nil, elastic.ReportErrorForMissingField("process.memory.resident_set_size_bytes", elastic.Kibana, r)
 	}
 	kibanaStatsFields.Put("process.memory.resident_set_size_in_bytes", int64(rss))
 
-	timestamp := time.Now()
-	kibanaStatsFields.Put("timestamp", timestamp)
+	kibanaStatsFields.Put("timestamp", now)
 
 	// Make usage field passthrough as-is
 	usage, ok := data["usage"].(map[string]interface{})
 	if !ok {
-		return elastic.ReportErrorForMissingField("usage", elastic.Kibana, r)
+		return "", "", nil, elastic.ReportErrorForMissingField("usage", elastic.Kibana, r)
 	}
 	kibanaStatsFields.Put("usage", usage)
 
-	clusterUUID, ok := data["clusterUuid"].(string)
+	return "kibana_stats", clusterUUID, kibanaStatsFields, nil
+}
+
+func settingsDataParser(r mb.ReporterV2, data common.MapStr, now time.Time) (string, string, common.MapStr, error) {
+	clusterUUID, ok := data["cluster_uuid"].(string)
+	if !ok {
+		return "", "", nil, elastic.ReportErrorForMissingField("cluster_uuid", elastic.Kibana, r)
+	}
+
+	kibanaSettingsFields, ok := data["settings"]
 	if !ok {
-		return elastic.ReportErrorForMissingField("clusterUuid", elastic.Kibana, r)
+		return "", "", nil, elastic.ReportErrorForMissingField("settings", elastic.Kibana, r)
+	}
+
+	return "kibana_settings", clusterUUID, kibanaSettingsFields.(map[string]interface{}), nil
+}
+
+func eventMappingXPack(r mb.ReporterV2, intervalMs int64, now time.Time, content []byte, dataParserFunc dataParser) error {
+	var data map[string]interface{}
+	err := json.Unmarshal(content, &data)
+	if err != nil {
+		return err
+	}
+
+	t, clusterUUID, fields, err := dataParserFunc(r, data, now)
+	if err != nil {
+		return err
 	}
 
 	var event mb.Event
 	event.RootFields = common.MapStr{
 		"cluster_uuid": clusterUUID,
-		"timestamp":    timestamp,
+		"timestamp":    now,
 		"interval_ms":  intervalMs,
-		"type":         "kibana_stats",
-		"kibana_stats": kibanaStatsFields,
+		"type":         t,
+		t:              fields,
 	}
 
 	event.Index = elastic.MakeXPackMonitoringIndexName(elastic.Kibana)
-	r.Event(event)
 
+	r.Event(event)
 	return nil
 }
+
+func eventMappingStatsXPack(r mb.ReporterV2, intervalMs int64, now time.Time, content []byte) error {
+	return eventMappingXPack(r, intervalMs, now, content, statsDataParser)
+}
+
+func eventMappingSettingsXPack(r mb.ReporterV2, intervalMs int64, now time.Time, content []byte) error {
+	return eventMappingXPack(r, intervalMs, now, content, settingsDataParser)
+}

metricbeat/module/kibana/stats/data_xpack_test.go

@@ -23,13 +23,14 @@ import (
 	"io/ioutil"
 	"path/filepath"
 	"testing"
+	"time"
 
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
-func TestEventMappingXPack(t *testing.T) {
+func TestEventMappingStatsXPack(t *testing.T) {
 
 	files, err := filepath.Glob("./_meta/test/stats-legacy.*.json")
 	assert.NoError(t, err)
@@ -39,7 +40,28 @@ func TestEventMappingXPack(t *testing.T) {
 		assert.NoError(t, err)
 
 		reporter := &mbtest.CapturingReporterV2{}
-		err = eventMappingXPack(reporter, 10000, input)
+		now := time.Now()
+
+		err = eventMappingStatsXPack(reporter, 10000, now, input)
+		assert.NoError(t, err, f)
+		assert.True(t, len(reporter.GetEvents()) >= 1, f)
+		assert.Equal(t, 0, len(reporter.GetErrors()), f)
+	}
+}
+
+func TestEventMappingSettingsXPack(t *testing.T) {
+
+	files, err := filepath.Glob("./_meta/test/settings.*.json")
+	assert.NoError(t, err)
+
+	for _, f := range files {
+		input, err := ioutil.ReadFile(f)
+		assert.NoError(t, err)
+
+		reporter := &mbtest.CapturingReporterV2{}
+		now := time.Now()
+
+		err = eventMappingSettingsXPack(reporter, 10000, now, input)
 		assert.NoError(t, err, f)
 		assert.True(t, len(reporter.GetEvents()) >= 1, f)
 		assert.Equal(t, 0, len(reporter.GetErrors()), f)

metricbeat/module/kibana/stats/stats.go

@@ -19,6 +19,8 @@ package stats
 
 import (
 	"fmt"
+	"strings"
+	"time"
 
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
 	"github.com/elastic/beats/metricbeat/helper"
@@ -36,7 +38,8 @@ func init() {
 }
 
 const (
-	statsPath = "api/stats"
+	statsPath    = "api/stats"
+	settingsPath = "api/settings"
 )
 
 var (
@@ -50,7 +53,8 @@ var (
 // MetricSet type defines all fields of the MetricSet
 type MetricSet struct {
 	mb.BaseMetricSet
-	http         *helper.HTTP
+	statsHTTP    *helper.HTTP
+	settingsHTTP *helper.HTTP
 	xPackEnabled bool
 }
 
@@ -63,22 +67,22 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, err
 	}
 
-	http, err := helper.NewHTTP(base)
+	statsHTTP, err := helper.NewHTTP(base)
 	if err != nil {
 		return nil, err
 	}
 
-	kibanaVersion, err := kibana.GetVersion(http, statsPath)
+	kibanaVersion, err := kibana.GetVersion(statsHTTP, statsPath)
 	if err != nil {
 		return nil, err
 	}
 
-	isAPIAvailable, err := kibana.IsStatsAPIAvailable(kibanaVersion)
+	isStatsAPIAvailable, err := kibana.IsStatsAPIAvailable(kibanaVersion)
 	if err != nil {
 		return nil, err
 	}
 
-	if !isAPIAvailable {
+	if !isStatsAPIAvailable {
 		const errorMsg = "The kibana stats metricset is only supported with Kibana >= %v. You are currently running Kibana %v"
 		return nil, fmt.Errorf(errorMsg, kibana.StatsAPIAvailableVersion, kibanaVersion)
 	}
@@ -87,12 +91,38 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		cfgwarn.Experimental("The experimental xpack.enabled flag in kibana/stats metricset is enabled.")
 
 		// Use legacy API response so we can passthru usage as-is
-		http.SetURI(http.GetURI() + "&legacy=true")
+		statsHTTP.SetURI(statsHTTP.GetURI() + "&legacy=true")
+	}
+
+	var settingsHTTP *helper.HTTP
+	if config.XPackEnabled {
+		cfgwarn.Experimental("The experimental xpack.enabled flag in kibana/stats metricset is enabled.")
+
+		isSettingsAPIAvailable, err := kibana.IsSettingsAPIAvailable(kibanaVersion)
+		if err != nil {
+			return nil, err
+		}
+
+		if !isSettingsAPIAvailable {
+			const errorMsg = "The kibana stats metricset with X-Pack enabled is only supported with Kibana >= %v. You are currently running Kibana %v"
+			return nil, fmt.Errorf(errorMsg, kibana.SettingsAPIAvailableVersion, kibanaVersion)
+		}
+
+		settingsHTTP, err = helper.NewHTTP(base)
+		if err != nil {
+			return nil, err
+		}
+
+		// HACK! We need to do this because there might be a basepath involved, so we
+		// only search/replace the actual API paths
+		settingsURI := strings.Replace(statsHTTP.GetURI(), statsPath, settingsPath, 1)
+		settingsHTTP.SetURI(settingsURI)
 	}
 
 	return &MetricSet{
 		base,
-		http,
+		statsHTTP,
+		settingsHTTP,
 		config.XPackEnabled,
 	}, nil
 }
@@ -101,17 +131,39 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // It returns the event which is then forward to the output. In case of an error, a
 // descriptive error must be returned.
 func (m *MetricSet) Fetch(r mb.ReporterV2) {
-	content, err := m.http.FetchContent()
+	now := time.Now()
+
+	m.fetchStats(r, now)
+	if m.xPackEnabled {
+		m.fetchSettings(r, now)
+	}
+}
+
+func (m *MetricSet) fetchStats(r mb.ReporterV2, now time.Time) {
+	content, err := m.statsHTTP.FetchContent()
 	if err != nil {
 		r.Error(err)
 		return
 	}
 
 	if m.xPackEnabled {
-		intervalMs := m.Module().Config().Period.Nanoseconds() / 1000 / 1000
-		eventMappingXPack(r, intervalMs, content)
+		intervalMs := m.calculateIntervalMs()
+		eventMappingStatsXPack(r, intervalMs, now, content)
 	} else {
 		eventMapping(r, content)
 	}
+}
+
+func (m *MetricSet) fetchSettings(r mb.ReporterV2, now time.Time) {
+	content, err := m.settingsHTTP.FetchContent()
+	if err != nil {
+		return
+	}
+
+	intervalMs := m.calculateIntervalMs()
+	eventMappingSettingsXPack(r, intervalMs, now, content)
+}
 
+func (m *MetricSet) calculateIntervalMs() int64 {
+	return m.Module().Config().Period.Nanoseconds() / 1000 / 1000
 }