Add reporting interface with error

metricbeat/module/ceph/cluster_disk/cluster_disk.go

@@ -63,23 +63,20 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, 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) {
+func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 	content, err := m.HTTP.FetchContent()
-
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return err
 	}
 
 	event, err := eventMapping(content)
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return err
 	}
 
-	reporter.Event(mb.Event{MetricSetFields: event})
+	if reported := reporter.Event(mb.Event{MetricSetFields: event}); !reported {
+		m.Logger().Debug("error reporting event")
+	}
 
-	return
+	return nil
 }

metricbeat/module/ceph/cluster_disk/cluster_disk_integration_test.go

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

metricbeat/module/ceph/cluster_disk/cluster_disk_test.go

@@ -50,8 +50,8 @@ func TestFetchEventContents(t *testing.T) {
 		"hosts":      []string{server.URL},
 	}
 
-	f := mbtest.NewReportingMetricSetV2(t, config)
-	events, errs := mbtest.ReportingFetchV2(f)
+	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)
 	}