[Metricbeat] Migrate Kubernetes container Metricset to use ReporterV2…

… interface (#10855)

metricbeat/module/kubernetes/container/container.go

@@ -18,7 +18,9 @@
 package container
 
 import (
-	"github.com/elastic/beats/libbeat/common"
+	"github.com/pkg/errors"
+
+	"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"
@@ -35,6 +37,8 @@ var (
 		DefaultScheme: defaultScheme,
 		DefaultPath:   defaultPath,
 	}.Build()
+
+	logger = logp.NewLogger("kubernetes.container")
 )
 
 // init registers the MetricSet with the central registry.
@@ -71,25 +75,34 @@ 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
+		err = errors.Wrap(err, "error doing HTTP request to fetch 'container' Metricset data")
+		logger.Error(err)
+		reporter.Error(err)
+		return
 	}
 
 	events, err := eventMapping(body, util.PerfMetrics)
 	if err != nil {
-		return nil, err
+		logger.Error(err)
+		reporter.Error(err)
+		return
 	}
 
 	m.enricher.Enrich(events)
 
-	return events, nil
+	for _, e := range events {
+		reporter.Event(mb.Event{MetricSetFields: e})
+	}
+
+	return
 }
 
 // Close stops this metricset