Fix Prometheus metric types parser (#39743)

* store metric types in a map * add test case * include error in fatal log * add changelog entry (cherry picked from commit 35158d8)
elastic · Jul 2, 2024 · f41dfc7 · f41dfc7
1 parent e9455e2
commit f41dfc7
Show file tree

Hide file tree

Showing 3 changed files with 196 additions and 2 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -117,6 +117,21 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 
 *Metricbeat*
 
+- Fix `namespace` filter option on metricset `state_namespace` enricher. {pull}39934[39934]
+- Fix `namespace` filter option at Kubernetes provider level. {pull}39881[39881]
+- Fix Azure Monitor 429 error by causing metricbeat to retry the request again. {pull}38294[38294]
+- Fix fields not being parsed correctly in postgresql/database {issue}25301[25301] {pull}37720[37720]
+- rabbitmq/queue - Change the mapping type of `rabbitmq.queue.consumers.utilisation.pct` to `scaled_float` from `long` because the values fall within the range of `[0.0, 1.0]`. Previously, conversion to integer resulted in reporting either `0` or `1`.
+- Fix timeout caused by the retrival of which indices are hidden {pull}39165[39165]
+- Fix Azure Monitor support for multiple aggregation types {issue}39192[39192] {pull}39204[39204]
+- Fix handling of access errors when reading process metrics {pull}39627[39627]
+- Fix behavior of cgroups path discovery when monitoring the host system from within a container {pull}39627[39627]
+- Fix issue where beats may report incorrect metrics for its own process when running inside a container {pull}39627[39627]
+- Fix for MySQL/Performance - Query failure for MySQL versions below v8.0.1, for performance metric `quantile_95`. {pull}38710[38710]
+- Fix Prometheus helper text parser to store each metric family type. {pull}39743[39743]
+- Normalize AWS RDS CPU Utilization values before making the metadata API call. {pull}39664[39664]
+- Fix behavior of pagetypeinfo metrics {pull}39985[39985]
+- Fix query logic for temp and non-temp tablespaces in Oracle module. {issue}38051[38051] {pull}39787[39787]
 
 *Osquerybeat*
 

diff --git a/metricbeat/helper/prometheus/textparse.go b/metricbeat/helper/prometheus/textparse.go
@@ -488,7 +488,8 @@ func ParseMetricFamilies(b []byte, contentType string, ts time.Time, logger *log
 		summariesByName      = map[string]map[string]*OpenMetric{}
 		histogramsByName     = map[string]map[string]*OpenMetric{}
 		fam                  *MetricFamily
-		mt                   = textparse.MetricTypeUnknown
+		// metricTypes stores the metric type for each metric name.
+		metricTypes = make(map[string]textparse.MetricType)
 	)
 	var err error
 
@@ -530,7 +531,8 @@ func ParseMetricFamilies(b []byte, contentType string, ts time.Time, logger *log
 			} else {
 				fam.Type = t
 			}
-			mt = t
+			// Store the metric type for each base metric name.
+			metricTypes[s] = t
 			continue
 		case textparse.EntryHelp:
 			buf, t := parser.Help()
@@ -611,6 +613,23 @@ func ParseMetricFamilies(b []byte, contentType string, ts time.Time, logger *log
 		lookupMetricName := metricName
 		var exm *exemplar.Exemplar
 
+		mt, ok := metricTypes[metricName]
+		if !ok {
+			// Splitting is necessary to find the base metric name type in the metricTypes map.
+			// This allows us to group related metrics together under the same base metric name.
+			// For example, the metric family `summary_metric` can have the metrics
+			// `summary_metric_count` and `summary_metric_sum`, all having the same metric type.
+			parts := strings.Split(metricName, "_")
+			baseMetricNamekey := strings.Join(parts[:len(parts)-1], "_")
+
+			// If the metric type is not found, default to unknown
+			if metricTypeFound, ok := metricTypes[baseMetricNamekey]; ok {
+				mt = metricTypeFound
+			} else {
+				mt = textparse.MetricTypeUnknown
+			}
+		}
+
 		// Suffixes - https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md#suffixes
 		switch mt {
 		case textparse.MetricTypeCounter:

diff --git a/metricbeat/helper/prometheus/textparse_test.go b/metricbeat/helper/prometheus/textparse_test.go
@@ -144,6 +144,166 @@ process_cpu 20
 	require.ElementsMatch(t, expected, result)
 }
 
+func TestGroupWithHeaderBlockPrometheus(t *testing.T) {
+	input := `
+# HELP nginx_sts_server_bytes_total The request/response bytes
+# TYPE nginx_sts_server_bytes_total counter
+# HELP nginx_sts_server_connects_total The connects counter
+# TYPE nginx_sts_server_connects_total counter
+# HELP nginx_sts_server_session_seconds_total The session duration time
+# TYPE nginx_sts_server_session_seconds_total counter
+# HELP nginx_sts_server_session_seconds The average of session duration time in seconds
+# TYPE nginx_sts_server_session_seconds gauge
+# HELP nginx_sts_server_session_duration_seconds The histogram of session duration in seconds
+# TYPE nginx_sts_server_session_duration_seconds histogram
+nginx_sts_server_bytes_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",direction="in"} 0
+nginx_sts_server_bytes_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",direction="out"} 0
+nginx_sts_server_connects_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",code="1xx"} 0
+nginx_sts_server_connects_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",code="2xx"} 0
+nginx_sts_server_connects_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",code="3xx"} 0
+nginx_sts_server_connects_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",code="4xx"} 0
+nginx_sts_server_connects_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",code="5xx"} 171
+nginx_sts_server_connects_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP",code="total"} 171
+nginx_sts_server_session_seconds_total{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP"} 0.016
+nginx_sts_server_session_seconds{listen="TCP:8091:127.0.0.1",port="8091",protocol="TCP"} 0.000
+`
+
+	expected := []*MetricFamily{
+		{
+			Name: stringp("nginx_sts_server_bytes_total"),
+			Help: stringp("The request/response bytes"),
+			Type: "counter",
+			Metric: []*OpenMetric{
+				{
+					Name: stringp("nginx_sts_server_bytes_total"),
+					Label: []*labels.Label{
+						{Name: "direction", Value: "in"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(0)},
+				},
+				{
+					Name: stringp("nginx_sts_server_bytes_total"),
+					Label: []*labels.Label{
+						{Name: "direction", Value: "out"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(0)},
+				},
+			},
+		},
+		{
+			Name: stringp("nginx_sts_server_connects_total"),
+			Help: stringp("The connects counter"),
+			Type: "counter",
+			Metric: []*OpenMetric{
+				{
+					Name: stringp("nginx_sts_server_connects_total"),
+					Label: []*labels.Label{
+						{Name: "code", Value: "1xx"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(0)},
+				},
+				{
+					Name: stringp("nginx_sts_server_connects_total"),
+					Label: []*labels.Label{
+						{Name: "code", Value: "2xx"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(0)},
+				},
+				{
+					Name: stringp("nginx_sts_server_connects_total"),
+					Label: []*labels.Label{
+						{Name: "code", Value: "3xx"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(0)},
+				},
+				{
+					Name: stringp("nginx_sts_server_connects_total"),
+					Label: []*labels.Label{
+						{Name: "code", Value: "4xx"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(0)},
+				},
+				{
+					Name: stringp("nginx_sts_server_connects_total"),
+					Label: []*labels.Label{
+						{Name: "code", Value: "5xx"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(171)},
+				},
+				{
+					Name: stringp("nginx_sts_server_connects_total"),
+					Label: []*labels.Label{
+						{Name: "code", Value: "total"},
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(171)},
+				},
+			},
+		},
+		{
+			Name: stringp("nginx_sts_server_session_seconds_total"),
+			Help: stringp("The session duration time"),
+			Type: "counter",
+			Metric: []*OpenMetric{
+				{
+					Name: stringp("nginx_sts_server_session_seconds_total"),
+					Label: []*labels.Label{
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Counter: &Counter{Value: float64p(0.016)},
+				},
+			},
+		},
+		{
+			Name: stringp("nginx_sts_server_session_seconds"),
+			Help: stringp("The average of session duration time in seconds"),
+			Type: "gauge",
+			Metric: []*OpenMetric{
+				{
+					Name: stringp("nginx_sts_server_session_seconds"),
+					Label: []*labels.Label{
+						{Name: "listen", Value: "TCP:8091:127.0.0.1"},
+						{Name: "port", Value: "8091"},
+						{Name: "protocol", Value: "TCP"},
+					},
+					Gauge: &Gauge{Value: float64p(0.000)},
+				},
+			},
+		},
+	}
+
+	result, err := ParseMetricFamilies([]byte(input), ContentTypeTextFormat, time.Now(), nil)
+	if err != nil {
+		t.Fatalf("ParseMetricFamilies for content type %s returned an error: %v", ContentTypeTextFormat, err)
+	}
+	require.ElementsMatch(t, expected, result)
+}
+
 func TestGaugeOpenMetrics(t *testing.T) {
 	input := `
 # TYPE first_metric gauge