[Metricbeat] Migrate NGINX stubstatus to use ReporterV2 interface

metricbeat/module/nginx/stubstatus/_meta/data.json

@@ -1,27 +1,33 @@
 {
     "@timestamp": "2017-10-12T08:05:34.853Z",
-    "beat": {
+    "agent": {
         "hostname": "host.example.com",
         "name": "host.example.com"
     },
+    "event": {
+        "dataset": "nginx.stubstatus",
+        "duration": 115000,
+        "module": "nginx"
+    },
     "metricset": {
-        "host": "nginx",
-        "module": "nginx",
-        "name": "stubstatus",
-        "rtt": 115
+        "name": "stubstatus"
     },
     "nginx": {
         "stubstatus": {
-            "accepts": 83,
+            "accepts": 6254,
             "active": 2,
-            "current": 83,
+            "current": 1,
             "dropped": 0,
-            "handled": 83,
-            "hostname": "nginx",
+            "handled": 6254,
+            "hostname": "127.0.0.1",
             "reading": 0,
-            "requests": 83,
+            "requests": 6259,
             "waiting": 1,
             "writing": 1
         }
+    },
+    "service": {
+        "address": "127.0.0.1",
+        "type": "nginx"
     }
 }

metricbeat/module/nginx/stubstatus/stubstatus.go

@@ -19,7 +19,7 @@
 package stubstatus
 
 import (
-	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
@@ -42,6 +42,8 @@ var (
 	}.Build()
 )
 
+var logger = logp.NewLogger("nginx.stubstatus")
+
 func init() {
 	mb.Registry.MustAddMetricSet("nginx", "stubstatus", New,
 		mb.WithHostParser(hostParser),
@@ -68,12 +70,18 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}, nil
 }
 
-// Fetch makes an HTTP request to fetch status metrics from the stubstatus endpoint.
-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) {
 	scanner, err := m.http.FetchScanner()
 	if err != nil {
-		return nil, err
+		logger.Error(err)
+		reporter.Error(err)
+		return
 	}
+	event, _ := eventMapping(scanner, m)
+	reporter.Event(mb.Event{MetricSetFields: event})
 
-	return eventMapping(scanner, m)
+	return
 }

metricbeat/module/nginx/stubstatus/stubstatus_integration_test.go

@@ -32,23 +32,31 @@ import (
 func TestFetch(t *testing.T) {
 	compose.EnsureUp(t, "nginx")
 
-	f := mbtest.NewEventFetcher(t, getConfig())
-	event, err := f.Fetch()
-	if !assert.NoError(t, err) {
-		t.FailNow()
+	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+	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)
 
-	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
+	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), events[0])
 
 	// Check number of fields.
+	event := events[0].MetricSetFields
 	assert.Equal(t, 10, len(event))
 }
 
 func TestData(t *testing.T) {
-	f := mbtest.NewEventFetcher(t, getConfig())
+	compose.EnsureUp(t, "nginx")
+
+	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+	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)
 
-	err := mbtest.WriteEvent(f, t)
-	if err != nil {
+	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }