[Metricbeat] Migrate Kubernetes node Metricset to use ReporterV2 inte…

…rface

metricbeat/module/kubernetes/node/node.go

@@ -20,6 +20,7 @@ package node
 import (
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/kubernetes"
+	"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 +37,8 @@ var (
 		DefaultScheme: defaultScheme,
 		DefaultPath:   defaultPath,
 	}.Build()
+
+	logger = logp.NewLogger("kubernetes.node")
 )
 
 // init registers the MetricSet with the central registry.
@@ -73,25 +76,31 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}, nil
 }
 
-// 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() (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) {
 	m.enricher.Start()
 
 	body, err := m.http.FetchContent()
 	if err != nil {
-		return nil, err
+		logger.Error(err)
+		reporter.Error(err)
+		return
 	}
 
 	event, err := eventMapping(body)
 	if err != nil {
-		return nil, err
+		logger.Error(err)
+		reporter.Error(err)
+		return
 	}
 
 	m.enricher.Enrich([]common.MapStr{event})
 
-	return event, nil
+	reporter.Event(mb.Event{MetricSetFields: event})
+
+	return
 }
 
 // Close stops this metricset