[Metricbeat] Migrate ETCD self to use ReporterV2 interface (#10816)

elastic · Feb 22, 2019 · 86e77e2 · 86e77e2
1 parent af41b93
commit 86e77e2
Show file tree

Hide file tree

Showing 4 changed files with 52 additions and 29 deletions.
diff --git a/metricbeat/module/etcd/self/_meta/data.json b/metricbeat/module/etcd/self/_meta/data.json
@@ -1,6 +1,6 @@
 {
     "@timestamp": "2017-10-12T08:05:34.853Z",
-    "beat": {
+    "agent": {
         "hostname": "host.example.com",
         "name": "host.example.com"
     },
@@ -9,8 +9,8 @@
             "id": "8e9e05c52164694d",
             "leaderinfo": {
                 "leader": "8e9e05c52164694d",
-                "starttime": "2017-12-07T07:20:33.241031712Z",
-                "uptime": "9m4.919923089s"
+                "starttime": "2019-02-18T21:42:23.191073075Z",
+                "uptime": "23m12.747903053s"
             },
             "name": "default",
             "recv": {
@@ -27,14 +27,20 @@
                 "bandwidthrate": 0,
                 "pkgrate": 0
             },
-            "starttime": "2017-12-07T07:20:32.340598119Z",
+            "starttime": "2019-02-18T21:42:21.490011868Z",
             "state": "StateLeader"
         }
     },
+    "event": {
+        "dataset": "etcd.self",
+        "duration": 115000,
+        "module": "etcd"
+    },
     "metricset": {
-        "host": "etcd:2379",
-        "module": "etcd",
-        "name": "self",
-        "rtt": 115
+        "name": "self"
+    },
+    "service": {
+        "address": "127.0.0.1:2379",
+        "type": "etcd"
     }
 }
diff --git a/metricbeat/module/etcd/self/self.go b/metricbeat/module/etcd/self/self.go
@@ -18,7 +18,7 @@
 package self
 
 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"
@@ -43,6 +43,8 @@ func init() {
 	)
 }
 
+var logger = logp.NewLogger("etcd.self")
+
 type MetricSet struct {
 	mb.BaseMetricSet
 	http *helper.HTTP
@@ -65,10 +67,18 @@ 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
 	}
-	return eventMapping(content), nil
+
+	reporter.Event(mb.Event{
+		MetricSetFields: eventMapping(content),
+	})
 }
diff --git a/metricbeat/module/etcd/self/self_integration_test.go b/metricbeat/module/etcd/self/self_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, "etcd")
 
-	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, "etcd")
 
-	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{} {

diff --git a/metricbeat/module/etcd/self/self_test.go b/metricbeat/module/etcd/self/self_test.go
@@ -41,6 +41,7 @@ func TestEventMapping(t *testing.T) {
 
 func TestFetchEventContent(t *testing.T) {
 	absPath, err := filepath.Abs("../_meta/test/")
+	assert.NoError(t, err)
 
 	response, err := ioutil.ReadFile(absPath + "/selfstats.json")
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -55,11 +56,12 @@ func TestFetchEventContent(t *testing.T) {
 		"metricsets": []string{"self"},
 		"hosts":      []string{server.URL},
 	}
-	f := mbtest.NewEventFetcher(t, config)
-	event, err := f.Fetch()
-	if err != nil {
-		t.Fatal(err)
+	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])
 }