From dbf39c1640c2aa530ef4d608b52d0a6e83a4dd0a Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Fri, 12 Apr 2019 06:40:34 -0700 Subject: [PATCH] Refactoring: use new Fetch interface that automatically reports and logs errors (#11767) --- .../logstash/logstash_integration_test.go | 8 +++---- metricbeat/module/logstash/node/data.go | 24 +++++-------------- metricbeat/module/logstash/node/node.go | 12 +++------- metricbeat/module/logstash/node_stats/data.go | 20 ++++------------ .../module/logstash/node_stats/node_stats.go | 12 +++------- 5 files changed, 21 insertions(+), 55 deletions(-) diff --git a/metricbeat/module/logstash/logstash_integration_test.go b/metricbeat/module/logstash/logstash_integration_test.go index 6987b44535c..9f3ac7ab123 100644 --- a/metricbeat/module/logstash/logstash_integration_test.go +++ b/metricbeat/module/logstash/logstash_integration_test.go @@ -40,8 +40,8 @@ func TestFetch(t *testing.T) { compose.EnsureUp(t, "logstash") for _, metricSet := range metricSets { - f := mbtest.NewReportingMetricSetV2(t, logstash.GetConfig(metricSet)) - events, errs := mbtest.ReportingFetchV2(f) + f := mbtest.NewReportingMetricSetV2Error(t, logstash.GetConfig(metricSet)) + events, errs := mbtest.ReportingFetchV2Error(f) assert.Empty(t, errs) if !assert.NotEmpty(t, events) { @@ -58,8 +58,8 @@ func TestData(t *testing.T) { for _, metricSet := range metricSets { config := logstash.GetConfig(metricSet) - f := mbtest.NewReportingMetricSetV2(t, config) - err := mbtest.WriteEventsReporterV2(f, t, metricSet) + f := mbtest.NewReportingMetricSetV2Error(t, config) + err := mbtest.WriteEventsReporterV2Error(f, t, metricSet) if err != nil { t.Fatal("write", err) } diff --git a/metricbeat/module/logstash/node/data.go b/metricbeat/module/logstash/node/data.go index 25cac855012..525ea1b014e 100644 --- a/metricbeat/module/logstash/node/data.go +++ b/metricbeat/module/logstash/node/data.go @@ -50,24 +50,18 @@ func eventMapping(r mb.ReporterV2, content []byte) error { var data map[string]interface{} err := json.Unmarshal(content, &data) if err != nil { - event.Error = errors.Wrap(err, "failure parsing Logstash Node API response") - r.Event(event) - return event.Error + return errors.Wrap(err, "failure parsing Logstash Node API response") } fields, err := schema.Apply(data) if err != nil { - event.Error = errors.Wrap(err, "failure applying node schema") - r.Event(event) - return event.Error + return errors.Wrap(err, "failure applying node schema") } // Set service ID serviceID, err := fields.GetValue("id") if err != nil { - event.Error = elastic.MakeErrorForMissingField("id", elastic.Logstash) - r.Event(event) - return event.Error + return elastic.MakeErrorForMissingField("id", elastic.Logstash) } event.RootFields.Put("service.id", serviceID) fields.Delete("id") @@ -75,9 +69,7 @@ func eventMapping(r mb.ReporterV2, content []byte) error { // Set service hostname host, err := fields.GetValue("host") if err != nil { - event.Error = elastic.MakeErrorForMissingField("host", elastic.Logstash) - r.Event(event) - return event.Error + return elastic.MakeErrorForMissingField("host", elastic.Logstash) } event.RootFields.Put("service.hostname", host) fields.Delete("host") @@ -85,9 +77,7 @@ func eventMapping(r mb.ReporterV2, content []byte) error { // Set service version version, err := fields.GetValue("version") if err != nil { - event.Error = elastic.MakeErrorForMissingField("version", elastic.Logstash) - r.Event(event) - return event.Error + return elastic.MakeErrorForMissingField("version", elastic.Logstash) } event.RootFields.Put("service.version", version) fields.Delete("version") @@ -95,9 +85,7 @@ func eventMapping(r mb.ReporterV2, content []byte) error { // Set PID pid, err := fields.GetValue("jvm.pid") if err != nil { - event.Error = elastic.MakeErrorForMissingField("jvm.pid", elastic.Logstash) - r.Event(event) - return event.Error + return elastic.MakeErrorForMissingField("jvm.pid", elastic.Logstash) } event.RootFields.Put("process.pid", pid) fields.Delete("jvm.pid") diff --git a/metricbeat/module/logstash/node/node.go b/metricbeat/module/logstash/node/node.go index db0cef279d5..1e8048a0f41 100644 --- a/metricbeat/module/logstash/node/node.go +++ b/metricbeat/module/logstash/node/node.go @@ -19,7 +19,6 @@ package node import ( "github.com/elastic/beats/metricbeat/helper" - "github.com/elastic/beats/metricbeat/helper/elastic" "github.com/elastic/beats/metricbeat/mb" "github.com/elastic/beats/metricbeat/mb/parse" "github.com/elastic/beats/metricbeat/module/logstash" @@ -69,16 +68,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) { // Fetch methods implements the data gathering and data conversion to the right format // It returns the event which is then forward to the output. In case of an error, a // descriptive error must be returned. -func (m *MetricSet) Fetch(r mb.ReporterV2) { +func (m *MetricSet) Fetch(r mb.ReporterV2) error { content, err := m.http.FetchContent() if err != nil { - elastic.ReportAndLogError(err, r, m.Logger()) - return + return err } - err = eventMapping(r, content) - if err != nil { - m.Logger().Error(err) - return - } + return eventMapping(r, content) } diff --git a/metricbeat/module/logstash/node_stats/data.go b/metricbeat/module/logstash/node_stats/data.go index c8896df8482..052a5b4ad31 100644 --- a/metricbeat/module/logstash/node_stats/data.go +++ b/metricbeat/module/logstash/node_stats/data.go @@ -51,24 +51,18 @@ func eventMapping(r mb.ReporterV2, content []byte) error { var data map[string]interface{} err := json.Unmarshal(content, &data) if err != nil { - event.Error = errors.Wrap(err, "failure parsing Logstash Node Stats API response") - r.Event(event) - return event.Error + return errors.Wrap(err, "failure parsing Logstash Node Stats API response") } fields, err := schema.Apply(data) if err != nil { - event.Error = errors.Wrap(err, "failure applying node stats schema") - r.Event(event) - return event.Error + return errors.Wrap(err, "failure applying node stats schema") } // Set service ID serviceID, err := fields.GetValue("id") if err != nil { - event.Error = elastic.MakeErrorForMissingField("id", elastic.Logstash) - r.Event(event) - return event.Error + return elastic.MakeErrorForMissingField("id", elastic.Logstash) } event.RootFields.Put("service.id", serviceID) fields.Delete("id") @@ -76,9 +70,7 @@ func eventMapping(r mb.ReporterV2, content []byte) error { // Set service hostname host, err := fields.GetValue("host") if err != nil { - event.Error = elastic.MakeErrorForMissingField("host", elastic.Logstash) - r.Event(event) - return event.Error + return elastic.MakeErrorForMissingField("host", elastic.Logstash) } event.RootFields.Put("service.hostname", host) fields.Delete("host") @@ -86,9 +78,7 @@ func eventMapping(r mb.ReporterV2, content []byte) error { // Set service version version, err := fields.GetValue("version") if err != nil { - event.Error = elastic.MakeErrorForMissingField("version", elastic.Logstash) - r.Event(event) - return event.Error + return elastic.MakeErrorForMissingField("version", elastic.Logstash) } event.RootFields.Put("service.version", version) fields.Delete("version") diff --git a/metricbeat/module/logstash/node_stats/node_stats.go b/metricbeat/module/logstash/node_stats/node_stats.go index 373ca22add2..5ac114f9a9e 100644 --- a/metricbeat/module/logstash/node_stats/node_stats.go +++ b/metricbeat/module/logstash/node_stats/node_stats.go @@ -19,7 +19,6 @@ package node_stats import ( "github.com/elastic/beats/metricbeat/helper" - "github.com/elastic/beats/metricbeat/helper/elastic" "github.com/elastic/beats/metricbeat/mb" "github.com/elastic/beats/metricbeat/mb/parse" @@ -70,16 +69,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) { // Fetch methods implements the data gathering and data conversion to the right format // It returns the event which is then forward to the output. In case of an error, a // descriptive error must be returned. -func (m *MetricSet) Fetch(r mb.ReporterV2) { +func (m *MetricSet) Fetch(r mb.ReporterV2) error { content, err := m.http.FetchContent() if err != nil { - elastic.ReportAndLogError(err, r, m.Logger()) - return + return err } - err = eventMapping(r, content) - if err != nil { - m.Logger().Error(err) - return - } + return eventMapping(r, content) }