Refactoring prometheus module to aggregate metrics based on metric family

CHANGELOG.asciidoc

@@ -35,6 +35,7 @@ https://github.com/elastic/beats/compare/v5.1.1...master[Check the HEAD diff]
   metricset. The configuration option for the feature was renamed from
   `cgroups` to `process.cgroups.enabled`. {pull}3519[3519]
 - Change fieldnames couchbase.node.couch.*.actual_disk_size.* to couchbase.node.couch.*.disk_size.* {pull}3545[3545]
+- Fixing prometheus collector to aggregate metrics based on metric family. {pull}4075[4075]
 
 *Packetbeat*
 - Remove deprecated geoip. {pull}3766[3766]
@@ -79,6 +80,7 @@ https://github.com/elastic/beats/compare/v5.1.1...master[Check the HEAD diff]
 - Support common.Time in mapstriface.toTime() {pull}3812[3812]
 - Fixing panic on prometheus collector when label has , {pull}3947[3947]
 - Fix MongoDB dbstats fields mapping. {pull}4025[4025]
+- Fixing prometheus collector to aggregate metrics based on metric family. {pull}4075[4075]
 
 *Packetbeat*
 

NOTICE

@@ -2187,6 +2187,38 @@ The above copyright notice and this permission notice shall be included in all c
 
 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
+--------------------------------------------------------------------
+github.com/matttproud/golang_protobuf_extensions
+--------------------------------------------------------------------
+Apache License
+
+-------NOTICE-----
+Copyright 2012 Matt T. Proud (matt.proud@gmail.com)
+
+--------------------------------------------------------------------
+github.com/prometheus/client_model
+--------------------------------------------------------------------
+Apache License
+
+-------NOTICE-----
+Data model artifacts for Prometheus.
+Copyright 2012-2015 The Prometheus Authors
+
+This product includes software developed at
+SoundCloud Ltd. (http://soundcloud.com/).
+
+--------------------------------------------------------------------
+github.com/prometheus/common
+--------------------------------------------------------------------
+Apache License
+
+-------NOTICE-----
+Common libraries shared by Prometheus Go components.
+Copyright 2015 The Prometheus Authors
+
+This product includes software developed at
+SoundCloud Ltd. (http://soundcloud.com/).
+
 --------------------------------------------------------------------
 github.com/davecgh/go-spew
 --------------------------------------------------------------------

metricbeat/module/prometheus/collector/_meta/data.json

@@ -16,8 +16,10 @@
             "label": {
                 "type": "quarantine_completed"
             },
-            "prometheus_local_storage_series_ops_total": 0
+            "prometheus_local_storage_series_ops_total": {
+                "value": 0
+            }
         }
     },
     "type": "metricsets"
-}
+}

metricbeat/module/prometheus/collector/collector.go

@@ -1,11 +1,14 @@
 package collector
 
 import (
+	"fmt"
+
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
+	"github.com/elastic/beats/metricbeat/module/prometheus"
 )
 
 const (
@@ -53,36 +56,36 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 
-	scanner, err := m.http.FetchScanner()
+	resp, err := m.http.FetchResponse()
+	defer resp.Body.Close()
 	if err != nil {
 		return nil, err
 	}
+	families, err := prometheus.GetMetricFamiliesFromResponse(resp)
+
+	if err != nil {
+		return nil, fmt.Errorf("Unable to decode response from prometheus endpoint")
+	}
+
 	eventList := map[string]common.MapStr{}
 
-	// Iterate through all events to gather data
-	for scanner.Scan() {
-		line := scanner.Text()
-		// Skip comment lines
-		if line[0] == '#' {
-			continue
-		}
+	for _, family := range families {
+		promEvents := GetPromEventsFromMetricFamily(family)
 
-		promEvent := NewPromEvent(line)
-		if promEvent.value == nil {
-			continue
-		}
+		for _, promEvent := range promEvents {
+			if _, ok := eventList[promEvent.labelHash]; !ok {
+				eventList[promEvent.labelHash] = common.MapStr{}
 
-		// If MapString for this label group does not exist yet, it is created
-		if _, ok := eventList[promEvent.labelHash]; !ok {
-			eventList[promEvent.labelHash] = common.MapStr{}
+				// Add labels
+				if len(promEvent.labels) > 0 {
+					eventList[promEvent.labelHash]["label"] = promEvent.labels
+				}
 
-			// Add labels
-			if len(promEvent.labels) > 0 {
-				eventList[promEvent.labelHash]["label"] = promEvent.labels
 			}
 
+			eventList[promEvent.labelHash][promEvent.key] = promEvent.value
+
 		}
-		eventList[promEvent.labelHash][promEvent.key] = promEvent.value
 	}
 
 	// Converts hash list to slice

metricbeat/module/prometheus/collector/collector_test.go

@@ -5,71 +5,139 @@ package collector
 import (
 	"testing"
 
-	"github.com/stretchr/testify/assert"
-
 	"github.com/elastic/beats/libbeat/common"
+
+	"github.com/golang/protobuf/proto"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/assert"
 )
 
-func TestDecodeLine(t *testing.T) {
+func TestGetPromEventsFromMetricFamily(t *testing.T) {
+	labels := common.MapStr{
+		"handler": "query",
+	}
 	tests := []struct {
-		Line  string
-		Event PromEvent
+		Family *dto.MetricFamily
+		Event  PromEvent
 	}{
 		{
-			Line: `http_request_duration_microseconds{handler="query",quantile="0.99"} 17`,
+			Family: &dto.MetricFamily{
+				Name: proto.String("http_request_duration_microseconds"),
+				Help: proto.String("foo"),
+				Type: dto.MetricType_COUNTER.Enum(),
+				Metric: []*dto.Metric{
+					{
+						Label: []*dto.LabelPair{
+							{
+								Name:  proto.String("handler"),
+								Value: proto.String("query"),
+							},
+						},
+						Counter: &dto.Counter{
+							Value: proto.Float64(10),
+						},
+					},
+				},
+			},
 			Event: PromEvent{
-				key:       "http_request_duration_microseconds",
-				value:     int64(17),
-				labelHash: `handler="query",quantile="0.99"`,
-				labels: common.MapStr{
-					"handler":  "query",
-					"quantile": 0.99,
+				key: "http_request_duration_microseconds",
+				value: common.MapStr{
+					"value": int64(10),
 				},
+				labelHash: labels.String(),
+				labels:    labels,
 			},
 		},
 		{
-			Line: `http_request_duration_microseconds{handler="query",quantile="0.99"} NaN`,
+			Family: &dto.MetricFamily{
+				Name: proto.String("http_request_duration_microseconds"),
+				Help: proto.String("foo"),
+				Type: dto.MetricType_GAUGE.Enum(),
+				Metric: []*dto.Metric{
+					{
+						Gauge: &dto.Gauge{
+							Value: proto.Float64(10),
+						},
+					},
+				},
+			},
 			Event: PromEvent{
-				key:       "http_request_duration_microseconds",
-				value:     nil,
-				labelHash: `handler="query",quantile="0.99"`,
-				labels: common.MapStr{
-					"handler":  "query",
-					"quantile": 0.99,
+				key: "http_request_duration_microseconds",
+				value: common.MapStr{
+					"value": float64(10),
 				},
+				labelHash: "#",
 			},
 		},
 		{
-			Line: `http_request_duration_microseconds{handler="query",quantile="0.99"} 13.2`,
+			Family: &dto.MetricFamily{
+				Name: proto.String("http_request_duration_microseconds"),
+				Help: proto.String("foo"),
+				Type: dto.MetricType_SUMMARY.Enum(),
+				Metric: []*dto.Metric{
+					{
+						Summary: &dto.Summary{
+							SampleCount: proto.Uint64(10),
+							SampleSum:   proto.Float64(10),
+							Quantile: []*dto.Quantile{
+								{
+									Quantile: proto.Float64(0.99),
+									Value:    proto.Float64(10),
+								},
+							},
+						},
+					},
+				},
+			},
 			Event: PromEvent{
-				key:       "http_request_duration_microseconds",
-				value:     13.2,
-				labelHash: `handler="query",quantile="0.99"`,
-				labels: common.MapStr{
-					"handler":  "query",
-					"quantile": 0.99,
+				key: "http_request_duration_microseconds",
+				value: common.MapStr{
+					"count": uint64(10),
+					"sum":   float64(10),
+					"percentile": common.MapStr{
+						"99": float64(10),
+					},
 				},
+				labelHash: "#",
 			},
 		},
 		{
-			Line: `apiserver_request_count{client="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36",code="200",contentType="",resource="elasticsearchclusters",verb="LIST"} 1`,
+			Family: &dto.MetricFamily{
+				Name: proto.String("http_request_duration_microseconds"),
+				Help: proto.String("foo"),
+				Type: dto.MetricType_HISTOGRAM.Enum(),
+				Metric: []*dto.Metric{
+					{
+						Histogram: &dto.Histogram{
+							SampleCount: proto.Uint64(10),
+							SampleSum:   proto.Float64(10),
+							Bucket: []*dto.Bucket{
+								{
+									UpperBound:      proto.Float64(0.99),
+									CumulativeCount: proto.Uint64(10),
+								},
+							},
+						},
+					},
+				},
+			},
 			Event: PromEvent{
-				key:       "apiserver_request_count",
-				value:     int64(1),
-				labelHash: `client="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36",code="200",contentType="",resource="elasticsearchclusters",verb="LIST"`,
-				labels: common.MapStr{
-					"client":      "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36",
-					"code":        int64(200),
-					"contentType": "",
-					"resource":    "elasticsearchclusters",
-					"verb":        "LIST",
+				key: "http_request_duration_microseconds",
+				value: common.MapStr{
+					"count": uint64(10),
+					"sum":   float64(10),
+					"bucket": common.MapStr{
+						"0.99": uint64(10),
+					},
 				},
+				labelHash: "#",
 			},
 		},
 	}
 
 	for _, test := range tests {
-		event := NewPromEvent(test.Line)
-		assert.Equal(t, event, test.Event)
+		event := GetPromEventsFromMetricFamily(test.Family)
+		assert.Equal(t, len(event), 1)
+		assert.Equal(t, event[0], test.Event)
 	}
 }