Convert to MetricsReporterV2

metricbeat/helper/prometheus/module.go

@@ -1,7 +1,6 @@
 package prometheus
 
 import (
-	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
 )
@@ -40,6 +39,6 @@ type prometheusMetricSet struct {
 	mapping    *MetricsMapping
 }
 
-func (m *prometheusMetricSet) Fetch() ([]common.MapStr, error) {
-	return m.prometheus.GetProcessedMetrics(m.mapping)
+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,6 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/libbeat/common"
+	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 )
 
 const promMetrics = `
@@ -241,12 +242,16 @@ func TestPrometheus(t *testing.T) {
 	}
 
 	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)
+		}
 	}
 }

metricbeat/helper/prometheus/ptest/ptest.go

@@ -9,7 +9,7 @@ import (
 	"os"
 	"testing"
 
-	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/metricbeat/mb"
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 
 	"github.com/stretchr/testify/assert"
@@ -54,13 +54,13 @@ func TestMetricSet(t *testing.T, module, metricset string, cases TestCases) {
 			"hosts":      []string{server.URL},
 		}
 
-		f := mbtest.NewEventsFetcher(t, config)
-
-		events, err := f.Fetch()
-		assert.NoError(t, err)
+		f := mbtest.NewReportingMetricSetV2(t, config)
+		reporter := &mbtest.CapturingReporterV2{}
+		f.Fetch(reporter)
+		assert.Nil(t, reporter.GetErrors(), "Errors while fetching metrics")
 
 		if *expectedFlag {
-			eventsJSON, _ := json.MarshalIndent(events, "", "\t")
+			eventsJSON, _ := json.MarshalIndent(reporter.GetEvents(), "", "\t")
 			err = ioutil.WriteFile(test.ExpectedFile, eventsJSON, 0644)
 			assert.NoError(t, err)
 		}
@@ -71,17 +71,19 @@ func TestMetricSet(t *testing.T, module, metricset string, cases TestCases) {
 			t.Fatal(err)
 		}
 
-		var expectedEvents []common.MapStr
+		var expectedEvents []mb.Event
 		err = json.Unmarshal(expected, &expectedEvents)
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		for _, event := range events {
+		for _, event := range reporter.GetEvents() {
 			// ensure the event is in expected list
 			found := -1
 			for i, expectedEvent := range expectedEvents {
-				if event.String() == expectedEvent.String() {
+				if event.RootFields.String() == expectedEvent.RootFields.String() &&
+					event.ModuleFields.String() == expectedEvent.ModuleFields.String() &&
+					event.MetricSetFields.String() == expectedEvent.MetricSetFields.String() {
 					found = i
 					break
 				}