[metricbeat] Migrate vsphere/datastore to ReporterV2

metricbeat/module/vsphere/datastore/datastore.go

@@ -21,6 +21,8 @@ import (
 	"context"
 	"net/url"
 
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
 	"github.com/elastic/beats/metricbeat/mb"
@@ -36,12 +38,14 @@ func init() {
 	)
 }
 
+// MetricSet type defines all fields of the MetricSet
 type MetricSet struct {
 	mb.BaseMetricSet
 	HostURL  *url.URL
 	Insecure bool
 }
 
+// New create a new instance of the MetricSet
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	cfgwarn.Beta("The vsphere datastore metricset is beta")
 
@@ -69,16 +73,17 @@ 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 {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	var events []common.MapStr
-
 	client, err := govmomi.NewClient(ctx, m.HostURL, m.Insecure)
 	if err != nil {
-		return nil, err
+		return errors.Wrap(err, "error in NewClient")
 	}
 
 	defer client.Logout(ctx)
@@ -90,7 +95,7 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 
 	v, err := mgr.CreateContainerView(ctx, c.ServiceContent.RootFolder, []string{"Datastore"}, true)
 	if err != nil {
-		return nil, err
+		return errors.Wrap(err, "error in CreateContainerView")
 	}
 
 	defer v.Destroy(ctx)
@@ -99,7 +104,7 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 	var dst []mo.Datastore
 	err = v.Retrieve(ctx, []string{"Datastore"}, []string{"summary"}, &dst)
 	if err != nil {
-		return nil, err
+		return errors.Wrap(err, "error in Retrieve")
 	}
 
 	for _, ds := range dst {
@@ -126,8 +131,10 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 			},
 		}
 
-		events = append(events, event)
+		reporter.Event(mb.Event{
+			MetricSetFields: event,
+		})
 	}
 
-	return events, nil
+	return nil
 }

metricbeat/module/vsphere/datastore/datastore_test.go

@@ -35,13 +35,15 @@ func TestFetchEventContents(t *testing.T) {
 	ts := model.Service.NewServer()
 	defer ts.Close()
 
-	f := mbtest.NewEventsFetcher(t, getConfig(ts))
-	events, err := f.Fetch()
-	if err != nil {
-		t.Fatal("fetch error", err)
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig(ts))
+	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())
 
@@ -83,9 +85,9 @@ func TestData(t *testing.T) {
 	ts := model.Service.NewServer()
 	defer ts.Close()
 
-	f := mbtest.NewEventsFetcher(t, getConfig(ts))
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig(ts))
 
-	if err := mbtest.WriteEvents(f, t); err != nil {
+	if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }