Use Elasticsearch histogram type to store Prometheus histograms (elas…

…tic#17061)

* Allow to override already defined metricsets

This is useful when we want to have different behaviors between flavours

* Add new Prometheus skeleton for basic types usage

* Refactor Prometheus module to accommodate newer uses

* Add typed schema to Prometheus module

This should make room for using different Elasticsearch types depending
on the Prometheus metric

* Convert Prometheus histograms to ES histograms

Co-authored-by: Chris Mark <chrismarkou92@gmail.com>
Co-authored-by: Jaime Soriano Pastor <jaime.soriano@elastic.co>

CHANGELOG.next.asciidoc

@@ -262,6 +262,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add support for CouchDB v2 {issue}16352[16352] {pull}16455[16455]
 - Add dashboards for the azure container metricsets. {pull}17194[17194]
 - Replace vpc metricset into vpn, transitgateway and natgateway metricsets. {pull}16892[16892]
+- Use Elasticsearch histogram type to store Prometheus histograms {pull}17061[17061]
+- Allow to rate Prometheus counters when scraping them {pull}17061[17061]
 - Release Oracle module as GA. {issue}14279[14279] {pull}16833[16833]
 - Release vsphere module as GA. {issue}15798[15798] {pull}17119[17119]
 - Add Storage metricsets to GCP module {pull}15598[15598]

metricbeat/docs/fields.asciidoc

@@ -66,6 +66,7 @@ grouped in the following categories:
 * <<exported-fields-postgresql>>
 * <<exported-fields-process>>
 * <<exported-fields-prometheus>>
+* <<exported-fields-prometheus-xpack>>
 * <<exported-fields-rabbitmq>>
 * <<exported-fields-redis>>
 * <<exported-fields-redisenterprise>>
@@ -31639,6 +31640,53 @@ query metricset
 remote write metrics from Prometheus server
 
 
+[[exported-fields-prometheus-xpack]]
+== Prometheus typed metrics fields
+
+Stats scraped from a Prometheus endpoint.
+
+
+
+*`prometheus.*.value`*::
++
+--
+Prometheus gauge metric
+
+
+type: object
+
+--
+
+*`prometheus.*.counter`*::
++
+--
+Prometheus counter metric
+
+
+type: object
+
+--
+
+*`prometheus.*.rate`*::
++
+--
+Prometheus rated counter metric
+
+
+type: object
+
+--
+
+*`prometheus.*.histogram`*::
++
+--
+Prometheus histogram metric - release: ga
+
+
+type: object
+
+--
+
 [[exported-fields-rabbitmq]]
 == RabbitMQ fields
 

metricbeat/mb/registry.go

@@ -68,6 +68,7 @@ type MetricSetRegistration struct {
 	IsDefault  bool
 	HostParser HostParser
 	Namespace  string
+	Replace    bool
 }
 
 // MetricSetOption sets an option for a MetricSetFactory that is being
@@ -99,6 +100,15 @@ func WithNamespace(namespace string) MetricSetOption {
 	}
 }
 
+// MustReplace specifies that the MetricSetFactory must be replacing an existing
+// metricset with the same name. An error will happen if there is no metricset
+// defined with the same params.
+func MustReplace() MetricSetOption {
+	return func(r *MetricSetRegistration) {
+		r.Replace = true
+	}
+}
+
 // Register contains the factory functions for creating new Modules and new
 // MetricSets. Registers are thread safe for concurrent usage.
 type Register struct {
@@ -201,22 +211,28 @@ func (r *Register) addMetricSet(module, name string, factory MetricSetFactory, o
 	module = strings.ToLower(module)
 	name = strings.ToLower(name)
 
+	// Set the options.
+	msInfo := MetricSetRegistration{Name: name, Factory: factory}
+	for _, opt := range options {
+		opt(&msInfo)
+	}
+
 	if metricsets, ok := r.metricSets[module]; !ok {
+		if msInfo.Replace {
+			return fmt.Errorf("metricset '%s/%s' should be replacing an existing metricset, none found", module, name)
+		}
+
 		r.metricSets[module] = map[string]MetricSetRegistration{}
 	} else if _, exists := metricsets[name]; exists {
-		return fmt.Errorf("metricset '%s/%s' is already registered", module, name)
+		if !msInfo.Replace {
+			return fmt.Errorf("metricset '%s/%s' is already registered", module, name)
+		}
 	}
 
 	if factory == nil {
 		return fmt.Errorf("metricset '%s/%s' cannot be registered with a nil factory", module, name)
 	}
 
-	// Set the options.
-	msInfo := MetricSetRegistration{Name: name, Factory: factory}
-	for _, opt := range options {
-		opt(&msInfo)
-	}
-
 	r.metricSets[module][name] = msInfo
 	r.log.Infof("MetricSet registered: %s/%s", module, name)
 	return nil

metricbeat/mb/registry_test.go

@@ -192,6 +192,14 @@ func TestDefaultMetricSet(t *testing.T) {
 	assert.Contains(t, names, metricSetName)
 }
 
+func TestMustReplaceMetricSet(t *testing.T) {
+	registry := NewRegister()
+	err := registry.addMetricSet(moduleName, metricSetName, fakeMetricSetFactory, MustReplace())
+	if assert.Error(t, err) {
+		assert.Equal(t, "metricset 'mymodule/mymetricset' should be replacing an existing metricset, none found", err.Error())
+	}
+}
+
 func TestMetricSetQuery(t *testing.T) {
 	registry := NewRegister()
 	err := registry.AddMetricSet(moduleName, metricSetName, fakeMetricSetFactory)

metricbeat/mb/testing/testdata.go

@@ -336,9 +336,22 @@ func documentedFieldCheck(foundKeys common.MapStr, knownKeys map[string]interfac
 					return nil
 				}
 			}
-			// If a field is defined as object it can also be defined as `status_codes.*`
-			// So this checks if such a key with the * exists by removing the last part.
+			// If a field is defined as object it can also have a * somewhere
+			// So this checks if such a key with the * exists by testing with it
 			splits := strings.Split(foundKey, ".")
+			found := false
+			for pos := 1; pos < len(splits)-1; pos++ {
+				key := strings.Join(splits[0:pos], ".") + ".*." + strings.Join(splits[pos+1:len(splits)], ".")
+				if _, ok := knownKeys[key]; ok {
+					found = true
+					break
+				}
+			}
+			if found {
+				continue
+			}
+
+			// last case `status_codes.*`:
 			prefix := strings.Join(splits[0:len(splits)-1], ".")
 			if _, ok := knownKeys[prefix+".*"]; ok {
 				continue

metricbeat/module/prometheus/collector/_meta/docs.asciidoc

@@ -24,6 +24,58 @@ to retrieve the metrics from (`/metrics` by default) can be configured with `met
 -------------------------------------------------------------------------------------
 
 
+[float]
+[role="xpack"]
+=== Histograms and types
+
+beta[]
+
+[source,yaml]
+-------------------------------------------------------------------------------------
+metricbeat.modules:
+- module: prometheus
+  period: 10s
+  hosts: ["localhost:9090"]
+  use_types: true
+  rate_counters: false
+-------------------------------------------------------------------------------------
+
+`use_types` paramater (default: false) enables a different layout for metrics storage, leveraging Elasticsearch
+types, including https://www.elastic.co/guide/en/elasticsearch/reference/current/histogram.html[histograms].
+
+`rate_counters` paramater (default: false) enables calculating a rate out of Prometheus counters. When enabled, Metricbeat stores
+the counter increment since the last collection. This metric should make some aggregations easier and with better
+performance. This parameter can only be enabled in combination with `use_types`.
+
+When `use_types` and `rate_counters` are enabled, metrics are stored like this:
+
+[source,json]
+----
+{
+    "prometheus": {
+        "labels": {
+            "instance": "172.27.0.2:9090",
+            "job": "prometheus"
+        },
+        "prometheus_target_interval_length_seconds_count": {
+            "counter": 1,
+            "rate": 0
+        },
+        "prometheus_target_interval_length_seconds_sum": {
+            "counter": 15.000401344,
+            "rate": 0
+        }
+        "prometheus_tsdb_compaction_chunk_range_seconds_bucket": {
+            "histogram": {
+                "values": [50, 300, 1000, 4000, 16000],
+                "counts": [10, 2, 34, 7]
+            }
+        }
+    },
+}
+----
+
+
 [float]
 === Scraping all metrics from a Prometheus server