Update mysql/status to ReporterV2 (elastic#11295)

metricbeat/module/mysql/status/status.go

@@ -26,18 +26,12 @@ package status
 import (
 	"database/sql"
 
-	"github.com/elastic/beats/libbeat/common"
-	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/module/mysql"
 
 	"github.com/pkg/errors"
 )
 
-var (
-	debugf = logp.MakeDebug("mysql-status")
-)
-
 func init() {
 	mb.Registry.MustAddMetricSet("mysql", "status", New,
 		mb.WithHostParser(mysql.ParseDSN),
@@ -57,26 +51,31 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 }
 
 // Fetch fetches status messages from a mysql host.
-func (m *MetricSet) Fetch() (common.MapStr, error) {
+func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 	if m.db == nil {
 		var err error
 		m.db, err = mysql.NewDB(m.HostData().URI)
 		if err != nil {
-			return nil, errors.Wrap(err, "mysql-status fetch failed")
+			return errors.Wrap(err, "mysql-status fetch failed")
 		}
 	}
 
 	status, err := m.loadStatus(m.db)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	event := eventMapping(status)
 
 	if m.Module().Config().Raw {
 		event["raw"] = rawEventMapping(status)
 	}
-	return event, nil
+
+	reporter.Event(mb.Event{
+		MetricSetFields: event,
+	})
+
+	return nil
 }
 
 // loadStatus loads all status entries from the given database into an array.

metricbeat/module/mysql/status/status_integration_test.go

@@ -33,12 +33,13 @@ import (
 func TestFetch(t *testing.T) {
 	compose.EnsureUp(t, "mysql")
 
-	f := mbtest.NewEventFetcher(t, getConfig(false))
-	event, err := f.Fetch()
-	if !assert.NoError(t, err) {
-		t.FailNow()
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig(false))
+	events, errs := mbtest.ReportingFetchV2Error(f)
+	if len(errs) > 0 {
+		t.Fatalf("Expected 0 errors, had %d. %v\n", len(errs), errs)
 	}
-
+	assert.NotEmpty(t, events)
+	event := events[0].MetricSetFields
 	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
 
 	// Check event fields
@@ -57,12 +58,13 @@ func TestFetch(t *testing.T) {
 func TestFetchRaw(t *testing.T) {
 	compose.EnsureUp(t, "mysql")
 
-	f := mbtest.NewEventFetcher(t, getConfig(true))
-	event, err := f.Fetch()
-	if !assert.NoError(t, err) {
-		t.FailNow()
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig(true))
+	events, errs := mbtest.ReportingFetchV2Error(f)
+	if len(errs) > 0 {
+		t.Fatalf("Expected 0 errors, had %d. %v\n", len(errs), errs)
 	}
-
+	assert.NotEmpty(t, events)
+	event := events[0].MetricSetFields
 	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
 
 	// Check event fields
@@ -81,9 +83,9 @@ func TestFetchRaw(t *testing.T) {
 }
 
 func TestData(t *testing.T) {
-	f := mbtest.NewEventFetcher(t, getConfig(false))
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig(false))
 
-	err := mbtest.WriteEvent(f, t)
+	err := mbtest.WriteEventsReporterV2Error(f, t, "")
 	if err != nil {
 		t.Fatal("write", err)
 	}