Infer types in Prometheus remote_write

CHANGELOG.next.asciidoc

@@ -596,6 +596,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Added `performance` and `query` metricsets to `mysql` module. {pull}18955[18955]
 - The `elasticsearch-xpack/index` metricset now reports hidden indices as such. {issue}18639[18639] {pull}18706[18706]
 - Adds support for app insights metrics in the azure module. {issue}18570[18570] {pull}18940[18940]
+- Infer types in Prometheus remote_write. {pull}19944[19944]
 - 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]

metricbeat/helper/prometheus/prometheus.go

@@ -22,6 +22,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"regexp"
 
 	"github.com/pkg/errors"
 	dto "github.com/prometheus/client_model/go"
@@ -284,3 +285,29 @@ func getLabels(metric *dto.Metric) common.MapStr {
 	}
 	return labels
 }
+
+func CompilePatternList(patterns *[]string) ([]*regexp.Regexp, error) {
+	var compiledPatterns []*regexp.Regexp
+	compiledPatterns = []*regexp.Regexp{}
+	if patterns != nil {
+		for _, pattern := range *patterns {
+			r, err := regexp.Compile(pattern)
+			if err != nil {
+				return nil, errors.Wrapf(err, "compiling pattern '%s'", pattern)
+			}
+			compiledPatterns = append(compiledPatterns, r)
+		}
+		return compiledPatterns, nil
+	}
+	return []*regexp.Regexp{}, nil
+}
+
+func MatchMetricFamily(family string, matchMetrics []*regexp.Regexp) bool {
+	for _, checkMetric := range matchMetrics {
+		matched := checkMetric.MatchString(family)
+		if matched {
+			return true
+		}
+	}
+	return false
+}

metricbeat/module/prometheus/collector/collector.go

@@ -111,11 +111,11 @@ func MetricSetBuilder(namespace string, genFactory PromEventsGeneratorFactory) f
 		}
 		// store host here to use it as a pointer when building `up` metric
 		ms.host = ms.Host()
-		ms.excludeMetrics, err = compilePatternList(config.MetricsFilters.ExcludeMetrics)
+		ms.excludeMetrics, err = p.CompilePatternList(config.MetricsFilters.ExcludeMetrics)
 		if err != nil {
 			return nil, errors.Wrapf(err, "unable to compile exclude patterns")
 		}
-		ms.includeMetrics, err = compilePatternList(config.MetricsFilters.IncludeMetrics)
+		ms.includeMetrics, err = p.CompilePatternList(config.MetricsFilters.IncludeMetrics)
 		if err != nil {
 			return nil, errors.Wrapf(err, "unable to compile include patterns")
 		}
@@ -237,39 +237,13 @@ func (m *MetricSet) skipFamilyName(family string) bool {
 
 	// if include_metrics are defined, check if this metric should be included
 	if len(m.includeMetrics) > 0 {
-		if !matchMetricFamily(family, m.includeMetrics) {
+		if !p.MatchMetricFamily(family, m.includeMetrics) {
 			return true
 		}
 	}
 	// now exclude the metric if it matches any of the given patterns
 	if len(m.excludeMetrics) > 0 {
-		if matchMetricFamily(family, m.excludeMetrics) {
-			return true
-		}
-	}
-	return false
-}
-
-func compilePatternList(patterns *[]string) ([]*regexp.Regexp, error) {
-	var compiledPatterns []*regexp.Regexp
-	compiledPatterns = []*regexp.Regexp{}
-	if patterns != nil {
-		for _, pattern := range *patterns {
-			r, err := regexp.Compile(pattern)
-			if err != nil {
-				return nil, errors.Wrapf(err, "compiling pattern '%s'", pattern)
-			}
-			compiledPatterns = append(compiledPatterns, r)
-		}
-		return compiledPatterns, nil
-	}
-	return []*regexp.Regexp{}, nil
-}
-
-func matchMetricFamily(family string, matchMetrics []*regexp.Regexp) bool {
-	for _, checkMetric := range matchMetrics {
-		matched := checkMetric.MatchString(family)
-		if matched {
+		if p.MatchMetricFamily(family, m.excludeMetrics) {
 			return true
 		}
 	}

metricbeat/module/prometheus/collector/collector_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	p "github.com/elastic/beats/v7/metricbeat/helper/prometheus"
 	mbtest "github.com/elastic/beats/v7/metricbeat/mb/testing"
 
 	_ "github.com/elastic/beats/v7/metricbeat/module/prometheus"
@@ -330,8 +331,8 @@ func TestSkipMetricFamily(t *testing.T) {
 	}
 
 	// test with no filters
-	ms.includeMetrics, _ = compilePatternList(&[]string{})
-	ms.excludeMetrics, _ = compilePatternList(&[]string{})
+	ms.includeMetrics, _ = p.CompilePatternList(&[]string{})
+	ms.excludeMetrics, _ = p.CompilePatternList(&[]string{})
 	metricsToKeep := 0
 	for _, testFamily := range testFamilies {
 		if !ms.skipFamily(testFamily) {
@@ -341,8 +342,8 @@ func TestSkipMetricFamily(t *testing.T) {
 	assert.Equal(t, metricsToKeep, len(testFamilies))
 
 	// test with only one include filter
-	ms.includeMetrics, _ = compilePatternList(&[]string{"http_request_duration_microseconds_a_*"})
-	ms.excludeMetrics, _ = compilePatternList(&[]string{})
+	ms.includeMetrics, _ = p.CompilePatternList(&[]string{"http_request_duration_microseconds_a_*"})
+	ms.excludeMetrics, _ = p.CompilePatternList(&[]string{})
 	metricsToKeep = 0
 	for _, testFamily := range testFamilies {
 		if !ms.skipFamily(testFamily) {
@@ -352,8 +353,8 @@ func TestSkipMetricFamily(t *testing.T) {
 	assert.Equal(t, metricsToKeep, 2)
 
 	// test with only one exclude filter
-	ms.includeMetrics, _ = compilePatternList(&[]string{""})
-	ms.excludeMetrics, _ = compilePatternList(&[]string{"http_request_duration_microseconds_a_*"})
+	ms.includeMetrics, _ = p.CompilePatternList(&[]string{""})
+	ms.excludeMetrics, _ = p.CompilePatternList(&[]string{"http_request_duration_microseconds_a_*"})
 	metricsToKeep = 0
 	for _, testFamily := range testFamilies {
 		if !ms.skipFamily(testFamily) {
@@ -363,8 +364,8 @@ func TestSkipMetricFamily(t *testing.T) {
 	assert.Equal(t, len(testFamilies)-2, metricsToKeep)
 
 	// test with ine include and one exclude
-	ms.includeMetrics, _ = compilePatternList(&[]string{"http_request_duration_microseconds_a_*"})
-	ms.excludeMetrics, _ = compilePatternList(&[]string{"http_request_duration_microseconds_a_b_*"})
+	ms.includeMetrics, _ = p.CompilePatternList(&[]string{"http_request_duration_microseconds_a_*"})
+	ms.excludeMetrics, _ = p.CompilePatternList(&[]string{"http_request_duration_microseconds_a_b_*"})
 	metricsToKeep = 0
 	for _, testFamily := range testFamilies {
 		if !ms.skipFamily(testFamily) {

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

@@ -61,3 +61,104 @@ remote_write:
         # Disable validation of the server certificate.
         #insecure_skip_verify: true
 ------------------------------------------------------------------------------
+
+
+[float]
+[role="xpack"]
+=== Histograms and types
+
+beta[]
+
+[source,yaml]
+-------------------------------------------------------------------------------------
+metricbeat.modules:
+- module: prometheus
+  metricsets: ["remote_write"]
+  host: "localhost"
+  port: "9201"
+-------------------------------------------------------------------------------------
+
+`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]
+==== Types' patterns
+
+Unline `collector` metricset, `remote_write` receives metrics in raw form from the prometheus server.
+In this, the module has to internally use a heuristic in order to identify efficiently the type of each raw metric.
+For these purpose some name patterns are used in order to identify the type of each metric.
+The default patterns are the following:
+
+. `_total` suffix: the metric is of Counter type
+. `_sum` suffix: the metric is of Counter type
+. `_count` suffix: the metric is of Counter type
+. `_bucket` suffix and `le` in labels: the metric is of Histogram type
+
+Everything else is handled as a Gauge.
+
+Users have the flexibility to add their own patterns using the following configuration:
+
+[source,yaml]
+-------------------------------------------------------------------------------------
+metricbeat.modules:
+- module: prometheus
+  metricsets: ["remote_write"]
+  host: "localhost"
+  port: "9201"
+  types_patterns:
+    counter_patterns: ["_my_counter_suffix"]
+    histogram_patterns: ["_my_histogram_suffix"]
+-------------------------------------------------------------------------------------
+
+The configuration above will consider metrics with names that match `_my_counter_suffix` as Counters
+and those that match `_my_histogram_suffix` (and have `le` in their labels) as Histograms.
+
+
+To match only specific metrics, anchor the start and the end of the regexp of each metric:
+
+- the caret ^ matches the beginning of a text or line,
+- the dollar sign $ matches the end of a text.
+
+[source,yaml]
+-------------------------------------------------------------------------------------
+metricbeat.modules:
+- module: prometheus
+  metricsets: ["remote_write"]
+  host: "localhost"
+  port: "9201"
+  types_patterns:
+    histogram_patterns: ["^my_histogram_metric$"]
+-------------------------------------------------------------------------------------

metricbeat/module/prometheus/remote_write/data.go

@@ -26,7 +26,17 @@ import (
 	"github.com/elastic/beats/v7/metricbeat/mb"
 )
 
-func samplesToEvents(metrics model.Samples) map[string]mb.Event {
+// DefaultRemoteWriteEventsGeneratorFactory returns the default prometheus events generator
+func DefaultRemoteWriteEventsGeneratorFactory(ms mb.BaseMetricSet) (RemoteWriteEventsGenerator, error) {
+	return &remoteWriteEventGenerator{}, nil
+}
+
+type remoteWriteEventGenerator struct{}
+
+func (p *remoteWriteEventGenerator) Start() {}
+func (p *remoteWriteEventGenerator) Stop()  {}
+
+func (p *remoteWriteEventGenerator) GenerateEvents(metrics model.Samples) map[string]mb.Event {
 	eventList := map[string]mb.Event{}
 
 	for _, metric := range metrics {