[Metricbeat] Migrate Ceph osd_tree to use ReporterV2 interface

metricbeat/module/ceph/osd_tree/_meta/data.json

@@ -1,9 +1,5 @@
 {
     "@timestamp": "2017-10-12T08:05:34.853Z",
-    "agent": {
-        "hostname": "host.example.com",
-        "name": "host.example.com"
-    },
     "ceph": {
         "osd_tree": {
             "children": [

metricbeat/module/ceph/osd_tree/osd_tree.go

@@ -18,7 +18,6 @@
 package osd_tree
 
 import (
-	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
@@ -61,16 +60,27 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}, nil
 }
 
-func (m *MetricSet) Fetch() ([]common.MapStr, error) {
+// Fetch methods implements the data gathering and data conversion to the right
+// format. It publishes the event which is then forwarded to the output. In case
+// of an error set the Error field of mb.Event or simply call report.Error().
+func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 	content, err := m.HTTP.FetchContent()
 	if err != nil {
-		return nil, err
+		m.Logger().Error(err)
+		reporter.Error(err)
+		return
 	}
 
 	events, err := eventsMapping(content)
 	if err != nil {
-		return nil, err
+		m.Logger().Error(err)
+		reporter.Error(err)
+		return
+	}
+
+	for _, event := range events {
+		reporter.Event(mb.Event{MetricSetFields: event})
 	}
 
-	return events, nil
+	return
 }

metricbeat/module/ceph/osd_tree/osd_tree_integration_test.go

@@ -26,9 +26,9 @@ import (
 )
 
 func TestData(t *testing.T) {
-	f := mbtest.NewEventsFetcher(t, getConfig())
-	err := mbtest.WriteEvents(f, t)
-	if err != nil {
+	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+
+	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }

metricbeat/module/ceph/osd_tree/osd_tree_test.go

@@ -49,14 +49,17 @@ func TestFetchEventContents(t *testing.T) {
 		"hosts":      []string{server.URL},
 	}
 
-	f := mbtest.NewEventsFetcher(t, config)
-	events, err := f.Fetch()
-	event := events[0]
+	f := mbtest.NewReportingMetricSetV2(t, config)
+	events, errs := mbtest.ReportingFetchV2(f)
+	if len(errs) > 0 {
+		t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
+	}
+	assert.NotEmpty(t, events)
+	nodeInfo := events[0].MetricSetFields
 
-	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event.StringToPrint())
+	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), nodeInfo.StringToPrint())
 
 	//check root bucket info
-	nodeInfo := events[0]
 	assert.EqualValues(t, "default", nodeInfo["name"])
 	assert.EqualValues(t, "root", nodeInfo["type"])
 	assert.EqualValues(t, []string{"-3"}, nodeInfo["children"])
@@ -65,7 +68,7 @@ func TestFetchEventContents(t *testing.T) {
 	assert.EqualValues(t, "", nodeInfo["father"])
 
 	//check host bucket info
-	nodeInfo = events[1]
+	nodeInfo = events[1].MetricSetFields
 	assert.EqualValues(t, "ceph-mon1", nodeInfo["name"])
 	assert.EqualValues(t, "host", nodeInfo["type"])
 	assert.EqualValues(t, []string{"1", "0"}, nodeInfo["children"])
@@ -74,7 +77,7 @@ func TestFetchEventContents(t *testing.T) {
 	assert.EqualValues(t, "default", nodeInfo["father"])
 
 	//check osd bucket info
-	nodeInfo = events[2]
+	nodeInfo = events[2].MetricSetFields
 	assert.EqualValues(t, "up", nodeInfo["status"])
 	assert.EqualValues(t, "osd.0", nodeInfo["name"])
 	assert.EqualValues(t, "osd", nodeInfo["type"])
@@ -88,7 +91,7 @@ func TestFetchEventContents(t *testing.T) {
 	assert.EqualValues(t, "ceph-mon1", nodeInfo["father"])
 	assert.EqualValues(t, 2, nodeInfo["depth"])
 
-	nodeInfo = events[3]
+	nodeInfo = events[3].MetricSetFields
 	assert.EqualValues(t, "up", nodeInfo["status"])
 	assert.EqualValues(t, "osd.1", nodeInfo["name"])
 	assert.EqualValues(t, "osd", nodeInfo["type"])