Add kubernetes apiserver metricset

CHANGELOG.asciidoc

@@ -238,6 +238,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Add Elasticsearch index_summary metricset. {pull}6918[6918]
 - Add config option `management_path_prefix` for RabbitMQ module to configure management plugin path prefix {issue}6875[6875] {pull}7074[7074]
 - Add shard metricset to Elasticsearch module. {pull}7006[7006]
+- Add apiserver metricset to Kubernetes module. {pull}7059[7059]
 
 *Packetbeat*
 

libbeat/template/processor.go

@@ -219,12 +219,12 @@ func (p *Processor) object(f *common.Field) common.MapStr {
 			dynProperties["index"] = "analyzed"
 		}
 		addDynamicTemplate(f, dynProperties, matchType("string"))
-	case "long":
-		dynProperties["type"] = f.ObjectType
-		addDynamicTemplate(f, dynProperties, matchType("long"))
 	case "keyword":
 		dynProperties["type"] = f.ObjectType
 		addDynamicTemplate(f, dynProperties, matchType("string"))
+	case "long", "double":
+		dynProperties["type"] = f.ObjectType
+		addDynamicTemplate(f, dynProperties, matchType(f.ObjectType))
 	}
 
 	properties := getDefaultProperties(f)

libbeat/template/processor_test.go

@@ -173,6 +173,12 @@ func TestProcessor(t *testing.T) {
 				"type": "long", "doc_values": false,
 			},
 		},
+		{
+			output: p.other(&common.Field{Type: "double", DocValues: &falseVar}),
+			expected: common.MapStr{
+				"type": "double", "doc_values": false,
+			},
+		},
 		{
 			output: p.other(&common.Field{Type: "text", DocValues: &trueVar}),
 			expected: common.MapStr{

metricbeat/docs/fields.asciidoc

@@ -6564,6 +6564,103 @@ Information and statistics of pods managed by kubernetes.
 
 
 
+[float]
+== apiserver fields
+
+Kubernetes API server metrics
+
+
+
+*`kubernetes.apiserver.request.client`*::
++
+--
+type: keyword
+
+Client doing the requests
+
+
+--
+
+*`kubernetes.apiserver.request.resource`*::
++
+--
+type: keyword
+
+Requested resource
+
+
+--
+
+*`kubernetes.apiserver.request.subresource`*::
++
+--
+type: keyword
+
+Requested subresource
+
+
+--
+
+*`kubernetes.apiserver.request.scope`*::
++
+--
+type: keyword
+
+Request scope (cluster, namespace, resource)
+
+
+--
+
+*`kubernetes.apiserver.request.verb`*::
++
+--
+type: keyword
+
+Request HTTP verb
+
+
+--
+
+*`kubernetes.apiserver.request.count`*::
++
+--
+type: long
+
+Total number of requests
+
+
+--
+
+*`kubernetes.apiserver.request.latency.sum`*::
++
+--
+type: long
+
+Requests latency, sum of latencies in microseconds
+
+
+--
+
+*`kubernetes.apiserver.request.latency.count`*::
++
+--
+type: long
+
+Request latency, number of requests
+
+
+--
+
+*`kubernetes.apiserver.request.latency.bucket`*::
++
+--
+type: object
+
+Request latency histagram buckets
+
+
+--
+
 [float]
 == container fields
 

metricbeat/docs/modules/kubernetes.asciidoc

@@ -57,6 +57,13 @@ metricbeat.modules:
   enabled: true
   metricsets:
     - event
+
+# Kubernetes API server
+- module: kubernetes
+  enabled: true
+  metricsets:
+    - apiserver
+  hosts: ["https://${KUBERNETES_SERVICE_HOST}:${KUBERNETES_SERVICE_PORT}"]
 ----
 
 This module supports TLS connection when using `ssl` config field, as described in <<configuration-ssl>>.

metricbeat/helper/prometheus/metric.go

@@ -1,8 +1,12 @@
 package prometheus
 
 import (
+	"math"
+	"strconv"
 	"strings"
 
+	"github.com/elastic/beats/libbeat/common"
+
 	dto "github.com/prometheus/client_model/go"
 )
 
@@ -74,6 +78,49 @@ func (m *commonMetric) GetValue(metric *dto.Metric) interface{} {
 		return gauge.GetValue()
 	}
 
+	summary := metric.GetSummary()
+	if summary != nil {
+		value := common.MapStr{}
+		value["sum"] = summary.GetSampleSum()
+		value["count"] = summary.GetSampleCount()
+
+		quantiles := summary.GetQuantile()
+		percentileMap := common.MapStr{}
+		for _, quantile := range quantiles {
+			if !math.IsNaN(quantile.GetValue()) {
+				key := strconv.FormatFloat((100 * quantile.GetQuantile()), 'f', -1, 64)
+				percentileMap[key] = quantile.GetValue()
+			}
+
+		}
+
+		if len(percentileMap) != 0 {
+			value["percentile"] = percentileMap
+		}
+
+		return value
+	}
+
+	histogram := metric.GetHistogram()
+	if histogram != nil {
+		value := common.MapStr{}
+		value["sum"] = histogram.GetSampleSum()
+		value["count"] = histogram.GetSampleCount()
+
+		buckets := histogram.GetBucket()
+		bucketMap := common.MapStr{}
+		for _, bucket := range buckets {
+			key := strconv.FormatFloat(bucket.GetUpperBound(), 'f', -1, 64)
+			bucketMap[key] = bucket.GetCumulativeCount()
+		}
+
+		if len(bucketMap) != 0 {
+			value["bucket"] = bucketMap
+		}
+
+		return value
+	}
+
 	// Other types are not supported here
 	return nil
 }

metricbeat/helper/prometheus/module.go

@@ -0,0 +1,44 @@
+package prometheus
+
+import (
+	"github.com/elastic/beats/metricbeat/mb"
+	"github.com/elastic/beats/metricbeat/mb/parse"
+)
+
+const (
+	defaultScheme = "http"
+	defaultPath   = "/metrics"
+)
+
+var (
+	// HostParser validates Prometheus URLs
+	HostParser = parse.URLHostParserBuilder{
+		DefaultScheme: defaultScheme,
+		DefaultPath:   defaultPath,
+	}.Build()
+)
+
+// MetricSetBuilder returns a builder function for a new Prometheus metricset using the given mapping
+func MetricSetBuilder(mapping *MetricsMapping) func(base mb.BaseMetricSet) (mb.MetricSet, error) {
+	return func(base mb.BaseMetricSet) (mb.MetricSet, error) {
+		prometheus, err := NewPrometheusClient(base)
+		if err != nil {
+			return nil, err
+		}
+		return &prometheusMetricSet{
+			BaseMetricSet: base,
+			prometheus:    prometheus,
+			mapping:       mapping,
+		}, nil
+	}
+}
+
+type prometheusMetricSet struct {
+	mb.BaseMetricSet
+	prometheus Prometheus
+	mapping    *MetricsMapping
+}
+
+func (m *prometheusMetricSet) Fetch(r mb.ReporterV2) {
+	m.prometheus.ReportProcessedMetrics(m.mapping, r)
+}

metricbeat/helper/prometheus/prometheus.go

@@ -19,6 +19,8 @@ type Prometheus interface {
 	GetFamilies() ([]*dto.MetricFamily, error)
 
 	GetProcessedMetrics(mapping *MetricsMapping) ([]common.MapStr, error)
+
+	ReportProcessedMetrics(mapping *MetricsMapping, r mb.ReporterV2)
 }
 
 type prometheus struct {
@@ -138,10 +140,22 @@ func (p *prometheus) GetProcessedMetrics(mapping *MetricsMapping) ([]common.MapS
 		for k, v := range mapping.ExtraFields {
 			event[k] = v
 		}
-
 		events = append(events, event)
+
 	}
 	return events, nil
+
+}
+
+func (p *prometheus) ReportProcessedMetrics(mapping *MetricsMapping, r mb.ReporterV2) {
+	events, err := p.GetProcessedMetrics(mapping)
+	if err != nil {
+		r.Error(err)
+		return
+	}
+	for _, event := range events {
+		r.Event(mb.Event{MetricSetFields: event})
+	}
 }
 
 func getEvent(m map[string]common.MapStr, labels common.MapStr) common.MapStr {

metricbeat/helper/prometheus/prometheus_test.go

@@ -11,13 +11,31 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/libbeat/common"
+	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 )
 
 const promMetrics = `
 # TYPE first_metric gauge
 first_metric{label1="value1",label2="value2",label3="value3"} 1
 # TYPE second_metric gauge
 second_metric{label1="value1",label3="othervalue"} 0
+# TYPE summary_metric summary
+summary_metric{quantile="0.5"} 29735
+summary_metric{quantile="0.9"} 47103
+summary_metric{quantile="0.99"} 50681
+summary_metric_sum 234892394
+summary_metric_count 44000
+# TYPE histogram_metric histogram
+histogram_metric_bucket{le="1000"} 1
+histogram_metric_bucket{le="10000"} 1
+histogram_metric_bucket{le="100000"} 1
+histogram_metric_bucket{le="1e+06"} 1
+histogram_metric_bucket{le="1e+08"} 1
+histogram_metric_bucket{le="1e+09"} 1
+histogram_metric_bucket{le="+Inf"} 1
+histogram_metric_sum 117
+histogram_metric_count 1
+
 `
 
 type mockFetcher struct{}
@@ -175,15 +193,65 @@ func TestPrometheus(t *testing.T) {
 				},
 			},
 		},
+		{
+			msg: "Summary metric",
+			mapping: &MetricsMapping{
+				Metrics: map[string]MetricMap{
+					"summary_metric": Metric("summary.metric"),
+				},
+			},
+			expected: []common.MapStr{
+				common.MapStr{
+					"summary.metric": common.MapStr{
+						"sum":   234892394.0,
+						"count": uint64(44000),
+						"percentile": common.MapStr{
+							"50": 29735.0,
+							"90": 47103.0,
+							"99": 50681.0,
+						},
+					},
+				},
+			},
+		},
+		{
+			msg: "Histogram metric",
+			mapping: &MetricsMapping{
+				Metrics: map[string]MetricMap{
+					"histogram_metric": Metric("histogram.metric"),
+				},
+			},
+			expected: []common.MapStr{
+				common.MapStr{
+					"histogram.metric": common.MapStr{
+						"count": uint64(1),
+						"bucket": common.MapStr{
+							"1000000000": uint64(1),
+							"+Inf":       uint64(1),
+							"1000":       uint64(1),
+							"10000":      uint64(1),
+							"100000":     uint64(1),
+							"1000000":    uint64(1),
+							"100000000":  uint64(1),
+						},
+						"sum": 117.0,
+					},
+				},
+			},
+		},
 	}
 
 	for _, test := range tests {
-		res, err := p.GetProcessedMetrics(test.mapping)
-		assert.Nil(t, err, test.msg)
+		reporter := &mbtest.CapturingReporterV2{}
+		p.ReportProcessedMetrics(test.mapping, reporter)
+		assert.Nil(t, reporter.GetErrors(), test.msg)
 		// Sort slice to avoid randomness
+		res := reporter.GetEvents()
 		sort.Slice(res, func(i, j int) bool {
-			return res[i].String() < res[j].String()
+			return res[i].MetricSetFields.String() < res[j].MetricSetFields.String()
 		})
-		assert.Equal(t, test.expected, res, test.msg)
+		for j, ev := range res {
+			assert.Equal(t, test.expected[j], ev.MetricSetFields, test.msg)
+		}
 	}
 }