[Metricbeat] Migrate ceph to reporter V2 with error handling (#11979)

* migrate ceph to reporter V2 with error handling
elastic · May 1, 2019 · 0a2ae60 · 0a2ae60
1 parent 8bd2510
commit 0a2ae60
Show file tree

Hide file tree

Showing 20 changed files with 125 additions and 92 deletions.
diff --git a/metricbeat/module/ceph/cluster_health/cluster_health.go b/metricbeat/module/ceph/cluster_health/cluster_health.go
@@ -18,6 +18,8 @@
 package cluster_health
 
 import (
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
@@ -42,11 +44,13 @@ func init() {
 	)
 }
 
+// MetricSet type defines all fields of the MetricSet
 type MetricSet struct {
 	mb.BaseMetricSet
 	*helper.HTTP
 }
 
+// New creates a new instance of the cluster_health MetricSet.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	http, err := helper.NewHTTP(base)
 	if err != nil {
@@ -63,22 +67,18 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, 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) {
+func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 	content, err := m.HTTP.FetchContent()
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return errors.Wrap(err, "error in fetch")
 	}
 
 	events, err := eventMapping(content)
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return errors.Wrap(err, "error in mapping")
 	}
 
 	reporter.Event(mb.Event{MetricSetFields: events})
 
-	return
+	return nil
 }
diff --git a/metricbeat/module/ceph/cluster_health/cluster_health_integration_test.go b/metricbeat/module/ceph/cluster_health/cluster_health_integration_test.go
@@ -26,9 +26,9 @@ import (
 )
 
 func TestData(t *testing.T) {
-	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig())
 
-	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
+	if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }

diff --git a/metricbeat/module/ceph/cluster_health/cluster_health_test.go b/metricbeat/module/ceph/cluster_health/cluster_health_test.go
@@ -50,8 +50,8 @@ func TestFetchEventContents(t *testing.T) {
 		"hosts":      []string{server.URL},
 	}
 
-	f := mbtest.NewReportingMetricSetV2(t, config)
-	events, errs := mbtest.ReportingFetchV2(f)
+	f := mbtest.NewReportingMetricSetV2Error(t, config)
+	events, errs := mbtest.ReportingFetchV2Error(f)
 	if len(errs) > 0 {
 		t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
 	}

diff --git a/metricbeat/module/ceph/cluster_health/data.go b/metricbeat/module/ceph/cluster_health/data.go
@@ -25,17 +25,20 @@ import (
 	"github.com/elastic/beats/libbeat/common"
 )
 
+// Timecheck contains part of the response from a HealthRequest
 type Timecheck struct {
 	RoundStatus string `json:"round_status"`
 	Epoch       int64  `json:"epoch"`
 	Round       int64  `json:"round"`
 }
 
+// Output is the body of the status response
 type Output struct {
 	OverallStatus string    `json:"overall_status"`
 	Timechecks    Timecheck `json:"timechecks"`
 }
 
+// HealthRequest represents the response to a cluster health request
 type HealthRequest struct {
 	Status string `json:"status"`
 	Output Output `json:"output"`

diff --git a/metricbeat/module/ceph/cluster_status/cluster_status.go b/metricbeat/module/ceph/cluster_status/cluster_status.go
@@ -18,6 +18,8 @@
 package cluster_status
 
 import (
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
@@ -42,11 +44,13 @@ func init() {
 	)
 }
 
+// MetricSet type defines all fields of the MetricSet
 type MetricSet struct {
 	mb.BaseMetricSet
 	*helper.HTTP
 }
 
+// New creates a new instance of the cluster_status MetricSet.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	http, err := helper.NewHTTP(base)
 	if err != nil {
@@ -63,24 +67,23 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, 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) {
+func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 	content, err := m.HTTP.FetchContent()
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return errors.Wrap(err, "error in fetch")
 	}
 
 	events, err := eventsMapping(content)
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return errors.Wrap(err, "error in mapping")
 	}
 
 	for _, event := range events {
-		reporter.Event(mb.Event{MetricSetFields: event})
+		reported := reporter.Event(mb.Event{MetricSetFields: event})
+		if !reported {
+			return nil
+		}
 	}
 
-	return
+	return nil
 }
diff --git a/metricbeat/module/ceph/cluster_status/cluster_status_integration_test.go b/metricbeat/module/ceph/cluster_status/cluster_status_integration_test.go
@@ -26,9 +26,9 @@ import (
 )
 
 func TestData(t *testing.T) {
-	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig())
 
-	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
+	if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }

diff --git a/metricbeat/module/ceph/cluster_status/cluster_status_test.go b/metricbeat/module/ceph/cluster_status/cluster_status_test.go
@@ -50,8 +50,8 @@ func TestFetchEventContents(t *testing.T) {
 		"hosts":      []string{server.URL},
 	}
 
-	f := mbtest.NewReportingMetricSetV2(t, config)
-	events, errs := mbtest.ReportingFetchV2(f)
+	f := mbtest.NewReportingMetricSetV2Error(t, config)
+	events, errs := mbtest.ReportingFetchV2Error(f)
 	if len(errs) > 0 {
 		t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
 	}
@@ -100,23 +100,23 @@ func TestFetchEventContents(t *testing.T) {
 	assert.EqualValues(t, 2872860672, pgInfo["used_bytes"])
 
 	//check pg_state info
-	pg_stateInfo := events[1].MetricSetFields["pg_state"].(common.MapStr)
-	assert.EqualValues(t, "active+undersized+degraded", pg_stateInfo["state_name"])
-	assert.EqualValues(t, 109, pg_stateInfo["count"])
-	assert.EqualValues(t, 813, pg_stateInfo["version"])
-
-	pg_stateInfo = events[2].MetricSetFields["pg_state"].(common.MapStr)
-	assert.EqualValues(t, "undersized+degraded+peered", pg_stateInfo["state_name"])
-	assert.EqualValues(t, 101, pg_stateInfo["count"])
-	assert.EqualValues(t, 813, pg_stateInfo["version"])
-
-	pg_stateInfo = events[3].MetricSetFields["pg_state"].(common.MapStr)
-	assert.EqualValues(t, "active+remapped", pg_stateInfo["state_name"])
-	assert.EqualValues(t, 55, pg_stateInfo["count"])
-	assert.EqualValues(t, 813, pg_stateInfo["version"])
-
-	pg_stateInfo = events[4].MetricSetFields["pg_state"].(common.MapStr)
-	assert.EqualValues(t, "active+undersized+degraded+remapped", pg_stateInfo["state_name"])
-	assert.EqualValues(t, 55, pg_stateInfo["count"])
-	assert.EqualValues(t, 813, pg_stateInfo["version"])
+	pgStateInfo := events[1].MetricSetFields["pg_state"].(common.MapStr)
+	assert.EqualValues(t, "active+undersized+degraded", pgStateInfo["state_name"])
+	assert.EqualValues(t, 109, pgStateInfo["count"])
+	assert.EqualValues(t, 813, pgStateInfo["version"])
+
+	pgStateInfo = events[2].MetricSetFields["pg_state"].(common.MapStr)
+	assert.EqualValues(t, "undersized+degraded+peered", pgStateInfo["state_name"])
+	assert.EqualValues(t, 101, pgStateInfo["count"])
+	assert.EqualValues(t, 813, pgStateInfo["version"])
+
+	pgStateInfo = events[3].MetricSetFields["pg_state"].(common.MapStr)
+	assert.EqualValues(t, "active+remapped", pgStateInfo["state_name"])
+	assert.EqualValues(t, 55, pgStateInfo["count"])
+	assert.EqualValues(t, 813, pgStateInfo["version"])
+
+	pgStateInfo = events[4].MetricSetFields["pg_state"].(common.MapStr)
+	assert.EqualValues(t, "active+undersized+degraded+remapped", pgStateInfo["state_name"])
+	assert.EqualValues(t, 55, pgStateInfo["count"])
+	assert.EqualValues(t, 813, pgStateInfo["version"])
 }
diff --git a/metricbeat/module/ceph/cluster_status/data.go b/metricbeat/module/ceph/cluster_status/data.go
@@ -25,11 +25,13 @@ import (
 	"github.com/elastic/beats/libbeat/common"
 )
 
+// PgState represents placement group state
 type PgState struct {
 	Count     int64  `json:"count"`
 	StateName string `json:"state_name"`
 }
 
+// Pgmap represents data from a placement group
 type Pgmap struct {
 	AvailByte int64 `json:"bytes_avail"`
 	TotalByte int64 `json:"bytes_total"`
@@ -54,6 +56,7 @@ type Pgmap struct {
 	PgStates []PgState `json:"pgs_by_state"`
 }
 
+// Osdmap represents data from an OSD
 type Osdmap struct {
 	Epoch      int64 `json:"epoch"`
 	Full       bool  `json:"full"`
@@ -64,15 +67,18 @@ type Osdmap struct {
 	RemapedPgs int64 `json:"num_remapped_pgs"`
 }
 
+// Osdmap_ is a placeholder for the json parser
 type Osdmap_ struct {
 	Osdmap Osdmap `json:"osdmap"`
 }
 
+// Output is the response body
 type Output struct {
 	Pgmap  Pgmap   `json:"pgmap"`
 	Osdmap Osdmap_ `json:"osdmap"`
 }
 
+// HealthRequest represents the response to a health request
 type HealthRequest struct {
 	Status string `json:"status"`
 	Output Output `json:"output"`
@@ -122,26 +128,26 @@ func eventsMapping(content []byte) ([]common.MapStr, error) {
 	pg["used_bytes"] = pgmap.UsedByte
 	pg["data_bytes"] = pgmap.DataByte
 
-	state_event := common.MapStr{}
-	state_event["osd"] = osdState
-	state_event["traffic"] = traffic
-	state_event["misplace"] = misplace
-	state_event["degraded"] = degraded
-	state_event["pg"] = pg
-	state_event["version"] = pgmap.Version
+	stateEvent := common.MapStr{}
+	stateEvent["osd"] = osdState
+	stateEvent["traffic"] = traffic
+	stateEvent["misplace"] = misplace
+	stateEvent["degraded"] = degraded
+	stateEvent["pg"] = pg
+	stateEvent["version"] = pgmap.Version
 
 	events := []common.MapStr{}
-	events = append(events, state_event)
+	events = append(events, stateEvent)
 
 	//pg state info
 	for _, state := range pgmap.PgStates {
-		state_evn := common.MapStr{
+		stateEvn := common.MapStr{
 			"count":      state.Count,
 			"state_name": state.StateName,
 			"version":    pgmap.Version,
 		}
 		evt := common.MapStr{
-			"pg_state": state_evn,
+			"pg_state": stateEvn,
 		}
 		events = append(events, evt)
 	}

diff --git a/metricbeat/module/ceph/osd_df/data.go b/metricbeat/module/ceph/osd_df/data.go
@@ -25,6 +25,7 @@ import (
 	"github.com/elastic/beats/libbeat/common"
 )
 
+// Node represents a node object
 type Node struct {
 	ID          int64  `json:"id"`
 	Name        string `json:"name"`
@@ -35,10 +36,12 @@ type Node struct {
 	DeviceClass string `json:"device_class"`
 }
 
+// Output contains a node list from the df response
 type Output struct {
 	Nodes []Node `json:"nodes"`
 }
 
+// OsdDfRequest contains the df response
 type OsdDfRequest struct {
 	Status string `json:"status"`
 	Output Output `json:"output"`

diff --git a/metricbeat/module/ceph/osd_df/osd_df.go b/metricbeat/module/ceph/osd_df/osd_df.go
@@ -18,6 +18,8 @@
 package osd_df
 
 import (
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
@@ -42,11 +44,13 @@ func init() {
 	)
 }
 
+// MetricSet type defines all fields of the MetricSet
 type MetricSet struct {
 	mb.BaseMetricSet
 	*helper.HTTP
 }
 
+// New creates a new instance of the osd_df MetricSet.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	http, err := helper.NewHTTP(base)
 	if err != nil {
@@ -63,24 +67,23 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, 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) {
+func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 	content, err := m.HTTP.FetchContent()
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return errors.Wrap(err, "error in fetch")
 	}
 
 	events, err := eventsMapping(content)
 	if err != nil {
-		m.Logger().Error(err)
-		reporter.Error(err)
-		return
+		return errors.Wrap(err, "error in mapping")
 	}
 
 	for _, event := range events {
-		reporter.Event(mb.Event{MetricSetFields: event})
+		reported := reporter.Event(mb.Event{MetricSetFields: event})
+		if !reported {
+			return nil
+		}
 	}
 
-	return
+	return nil
 }
diff --git a/metricbeat/module/ceph/osd_df/osd_df_integration_test.go b/metricbeat/module/ceph/osd_df/osd_df_integration_test.go
@@ -26,9 +26,9 @@ import (
 )
 
 func TestData(t *testing.T) {
-	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig())
 
-	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
+	if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }

diff --git a/metricbeat/module/ceph/osd_df/osd_df_test.go b/metricbeat/module/ceph/osd_df/osd_df_test.go
@@ -49,8 +49,8 @@ func TestFetchEventContents(t *testing.T) {
 		"hosts":      []string{server.URL},
 	}
 
-	f := mbtest.NewReportingMetricSetV2(t, config)
-	events, errs := mbtest.ReportingFetchV2(f)
+	f := mbtest.NewReportingMetricSetV2Error(t, config)
+	events, errs := mbtest.ReportingFetchV2Error(f)
 	if len(errs) > 0 {
 		t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
 	}