[Metricbeat] Migrate Ceph monitor_health to use ReporterV2 interface

metricbeat/module/ceph/monitor_health/_meta/testdata/config.yml

@@ -0,0 +1,3 @@
+type: http
+url: "/api/v0.1/health"
+suffix: plain

metricbeat/module/ceph/monitor_health/_meta/testdata/input.plain

@@ -0,0 +1 @@
+{"status": "OK", "output": {"detail": [], "timechecks": {"round_status": "finished", "epoch": 3, "round": 0}, "health": {"health_services": [{"mons": [{"last_updated": "2019-03-13 11:21:24.667025", "name": "26c372192772", "avail_percent": 82, "kb_total": 936145620, "kb_avail": 773013080, "health": "HEALTH_OK", "kb_used": 115509168, "store_stats": {"bytes_total": 4195407, "bytes_log": 4128768, "last_updated": "0.000000", "bytes_misc": 65552, "bytes_sst": 1087}}]}]}, "overall_status": "HEALTH_OK", "summary": []}}

metricbeat/module/ceph/monitor_health/_meta/testdata/input.plain-expected.json

@@ -0,0 +1,48 @@
+[
+    {
+        "ceph": {
+            "monitor_health": {
+                "available": {
+                    "kb": 773013080,
+                    "pct": 82
+                },
+                "health": "HEALTH_OK",
+                "last_updated": "2019-03-13T11:21:24.667025Z",
+                "name": "26c372192772",
+                "store_stats": {
+                    "last_updated": "0.000000",
+                    "log": {
+                        "bytes": 4128768
+                    },
+                    "misc": {
+                        "bytes": 65552
+                    },
+                    "sst": {
+                        "bytes": 1087
+                    },
+                    "total": {
+                        "bytes": 4195407
+                    }
+                },
+                "total": {
+                    "kb": 936145620
+                },
+                "used": {
+                    "kb": 115509168
+                }
+            }
+        },
+        "event": {
+            "dataset": "ceph.monitor_health",
+            "duration": 115000,
+            "module": "ceph"
+        },
+        "metricset": {
+            "name": "monitor_health"
+        },
+        "service": {
+            "address": "127.0.0.1:55555",
+            "type": "ceph"
+        }
+    }
+]

metricbeat/module/ceph/monitor_health/data.go

@@ -21,8 +21,9 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/libbeat/common"
-	"github.com/elastic/beats/libbeat/logp"
 )
 
 type Tick struct {
@@ -85,11 +86,11 @@ type HealthRequest struct {
 	Output Output `json:"output"`
 }
 
-func eventsMapping(content []byte) []common.MapStr {
+func eventsMapping(content []byte) ([]common.MapStr, error) {
 	var d HealthRequest
 	err := json.Unmarshal(content, &d)
 	if err != nil {
-		logp.Err("Error: %+v", err)
+		return nil, errors.Wrapf(err, "could not get HealthRequest data")
 	}
 
 	events := []common.MapStr{}
@@ -131,5 +132,5 @@ func eventsMapping(content []byte) []common.MapStr {
 		}
 	}
 
-	return events
+	return events, nil
 }

metricbeat/module/ceph/monitor_health/monitor_health.go

@@ -18,7 +18,6 @@
 package monitor_health
 
 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,11 +60,23 @@ 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) error {
 	content, err := m.HTTP.FetchContent()
 	if err != nil {
-		return nil, err
+		return err
+	}
+
+	events, err := eventsMapping(content)
+	if err != nil {
+		return err
+	}
+
+	for _, event := range events {
+		reporter.Event(mb.Event{MetricSetFields: event})
 	}
 
-	return eventsMapping(content), nil
+	return nil
 }

metricbeat/module/ceph/monitor_health/monitor_health_test.go

@@ -32,6 +32,7 @@ import (
 
 func TestFetchEventContents(t *testing.T) {
 	absPath, err := filepath.Abs("../_meta/testdata/")
+	assert.NoError(t, err)
 
 	response, err := ioutil.ReadFile(absPath + "/sample_response.json")
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -47,12 +48,13 @@ func TestFetchEventContents(t *testing.T) {
 		"hosts":      []string{server.URL},
 	}
 
-	f := mbtest.NewEventsFetcher(t, config)
-	events, err := f.Fetch()
-	if err != nil {
-		t.Fatal(err)
+	f := mbtest.NewReportingMetricSetV2Error(t, config)
+	events, errs := mbtest.ReportingFetchV2Error(f)
+	if len(errs) > 0 {
+		t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
 	}
-	event := events[0]
+	assert.NotEmpty(t, events)
+	event := events[0].MetricSetFields
 
 	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event.StringToPrint())