[Metricbeat] Migrate EnvoyProxy server to use ReporterV2 interface

metricbeat/module/envoyproxy/server/_meta/data.json

@@ -1,74 +1,72 @@
 {
-  "@timestamp": "2018-07-06T14:01:31.031Z",
-  "beat":{
-	"hostname":"beathost",
-	"name":"beathost"
-	},
-  "metricset":{
-	"host":"localhost",
-	"module":"envoyproxy",
-	"name":"server",
-	"rtt":44269
+    "@timestamp": "2017-10-12T08:05:34.853Z",
+    "agent": {
+        "hostname": "host.example.com",
+        "name": "host.example.com"
     },
-  "envoyproxy": {
-    "server": {
-      "filesystem": {
-        "write_buffered": 30,
-        "write_completed": 29,
-        "write_total_buffered": 0,
-        "flushed_by_timer": 311,
-        "reopen_failed": 0
-      },
-      "runtime": {
-        "override_dir_not_exists": 0,
-        "admin_overrides_active": 0,
-        "load_error": 0,
-        "load_success": 0,
-        "num_keys": 0,
-        "override_dir_exists": 0
-      },
-      "listener_manager": {
-        "listener_modified": 0,
-        "listener_removed": 0,
-        "total_listeners_active": 1,
-        "total_listeners_draining": 0,
-        "total_listeners_warming": 0,
-        "listener_added": 1,
-        "listener_create_failure": 0,
-        "listener_create_success": 4
-      },
-      "stats": {
-        "overflow": 0
-      },
-      "server": {
-        "version": 4151803,
-        "memory_allocated": 3170848,
-        "watchdog_miss": 0,
-        "uptime": 3146,
-        "watchdog_mega_miss": 0,
-        "hot_restart_epoch": 0,
-        "days_until_first_cert_expiring": 2147483647,
-        "live": 1,
-        "memory_heap_size": 4194304,
-        "parent_connections": 0,
-        "total_connections": 0
-      },
-      "http2": {
-	    "header_overflow":0,
-	    "headers_cb_no_stream":0,
-	    "rx_reset":0,
-	    "too_many_header_frames":0,
-	    "trailers":0,
-	    "tx_reset":0
-	  },
-      "cluster_manager": {
-        "active_clusters": 1,
-        "cluster_added": 1,
-        "cluster_modified": 0,
-        "cluster_removed": 0,
-        "warming_clusters": 0
-      }
+    "envoyproxy": {
+        "server": {
+            "cluster_manager": {
+                "active_clusters": 1,
+                "cluster_added": 1,
+                "cluster_modified": 0,
+                "cluster_removed": 0,
+                "warming_clusters": 0
+            },
+            "filesystem": {
+                "flushed_by_timer": 15,
+                "reopen_failed": 0,
+                "write_buffered": 151,
+                "write_completed": 31,
+                "write_total_buffered": 518
+            },
+            "http2": {},
+            "listener_manager": {
+                "listener_added": 1,
+                "listener_create_failure": 0,
+                "listener_create_success": 12,
+                "listener_modified": 0,
+                "listener_removed": 0,
+                "total_listeners_active": 1,
+                "total_listeners_draining": 0,
+                "total_listeners_warming": 0
+            },
+            "runtime": {
+                "admin_overrides_active": 0,
+                "load_error": 0,
+                "load_success": 0,
+                "num_keys": 0,
+                "override_dir_exists": 0,
+                "override_dir_not_exists": 0
+            },
+            "server": {
+                "days_until_first_cert_expiring": 2147483647,
+                "hot_restart_epoch": 0,
+                "live": 1,
+                "memory_allocated": 3302416,
+                "memory_heap_size": 4194304,
+                "parent_connections": 0,
+                "total_connections": 0,
+                "uptime": 150,
+                "version": 4151803,
+                "watchdog_mega_miss": 0,
+                "watchdog_miss": 0
+            },
+            "stats": {
+                "overflow": 0
+            }
+        }
+    },
+    "event": {
+        "dataset": "envoyproxy.server",
+        "duration": 115000,
+        "module": "envoyproxy"
+    },
+    "metricset": {
+        "name": "server"
+    },
+    "service": {
+        "address": "127.0.0.1:9901",
+        "type": "envoyproxy"
     }
-  },
-  "type":"metricsets"
-}	
+}

metricbeat/module/envoyproxy/server/server.go

@@ -18,7 +18,7 @@
 package server
 
 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"
@@ -36,6 +36,8 @@ var (
 	}.Build()
 )
 
+var logger = logp.NewLogger("envoyproxy.server")
+
 func init() {
 	mb.Registry.MustAddMetricSet("envoyproxy", "server", New,
 		mb.WithHostParser(hostParser),
@@ -64,14 +66,19 @@ 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) {
 	content, err := m.http.FetchContent()
 	if err != nil {
-		return nil, err
+		logger.Error(err)
+		reporter.Error(err)
+		return
 	}
-	event, err := eventMapping(content)
-	if err != nil {
-		return nil, err
-	}
-	return event, nil
+
+	event, _ := eventMapping(content)
+	reporter.Event(mb.Event{MetricSetFields: event})
+
+	return
 }

metricbeat/module/envoyproxy/server/server_integration_test.go

@@ -29,27 +29,32 @@ import (
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 )
 
-func TestData(t *testing.T) {
+func TestFetch(t *testing.T) {
 	compose.EnsureUp(t, "envoyproxy")
 
-	f := mbtest.NewEventFetcher(t, getConfig())
-	err := mbtest.WriteEvent(f, t)
-	if err != nil {
-		t.Fatal("write", err)
+	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(), events[0])
 }
 
-func TestFetch(t *testing.T) {
+func TestData(t *testing.T) {
 	compose.EnsureUp(t, "envoyproxy")
 
-	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)
 
-	assert.NotNil(t, event)
-	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
+	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
+		t.Fatal("write", err)
+	}
 }
 
 func getConfig() map[string]interface{} {

metricbeat/module/envoyproxy/server/server_test.go

@@ -22,6 +22,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -92,13 +93,14 @@ func TestFetchEventContent(t *testing.T) {
 		"hosts":      []string{server.URL},
 	}
 
-	f := mbtest.NewEventFetcher(t, config)
-	event, err := f.Fetch()
-	if !assert.NoError(t, err) {
-		t.FailNow()
+	f := mbtest.NewReportingMetricSetV2(t, config)
+	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])
 }
 
 func testValue(t *testing.T, event common.MapStr, field string, value interface{}) {
@@ -129,13 +131,23 @@ func TestFetchTimeout(t *testing.T) {
 		"timeout":    "50ms",
 	}
 
-	f := mbtest.NewEventFetcher(t, config)
+	f := mbtest.NewReportingMetricSetV2(t, config)
 
 	start := time.Now()
-	_, err = f.Fetch()
+	events, errs := mbtest.ReportingFetchV2(f)
+	if len(errs) == 0 {
+		t.Fatalf("Expected an error, had %d. %v\n", len(errs), errs)
+	}
+	assert.Empty(t, events)
 	elapsed := time.Since(start)
-	if assert.Error(t, err) {
-		assert.Contains(t, err.Error(), "request canceled (Client.Timeout exceeded")
+	var found bool
+	for _, err := range errs {
+		if strings.Contains(err.Error(), "request canceled (Client.Timeout exceeded") {
+			found = true
+		}
+	}
+	if !found {
+		assert.Failf(t, "", "expected an error containing 'request canceled (Client.Timeout exceeded'. Got %v", errs)
 	}
 
 	assert.True(t, elapsed < 5*time.Second, "elapsed time: %s", elapsed.String())