[metricbeat] migrate vsphere/host to ReporterV2 (#11403)

* migrate vsphere/host to ReporterV2

metricbeat/module/vsphere/host/data.go

@@ -24,8 +24,8 @@ import (
 )
 
 func eventMapping(hs mo.HostSystem) common.MapStr {
- totalCpu := int64(hs.Summary.Hardware.CpuMhz) * int64(hs.Summary.Hardware.NumCpuCores)
- freeCpu := int64(totalCpu) - int64(hs.Summary.QuickStats.OverallCpuUsage)
+ totalCPU := int64(hs.Summary.Hardware.CpuMhz) * int64(hs.Summary.Hardware.NumCpuCores)
+ freeCPU := int64(totalCPU) - int64(hs.Summary.QuickStats.OverallCpuUsage)
  usedMemory := int64(hs.Summary.QuickStats.OverallMemoryUsage) * 1024 * 1024
  freeMemory := int64(hs.Summary.Hardware.MemorySize) - usedMemory
 
@@ -36,10 +36,10 @@ func eventMapping(hs mo.HostSystem) common.MapStr {
  "mhz": hs.Summary.QuickStats.OverallCpuUsage,
  },
  "total": common.MapStr{
- "mhz": totalCpu,
+ "mhz": totalCPU,
  },
  "free": common.MapStr{
- "mhz": freeCpu,
+ "mhz": freeCPU,
  },
  },
  "memory": common.MapStr{

metricbeat/module/vsphere/host/host.go

@@ -27,7 +27,6 @@ import (
 
  "github.com/elastic/beats/libbeat/common"
  "github.com/elastic/beats/libbeat/common/cfgwarn"
- "github.com/elastic/beats/libbeat/logp"
  "github.com/elastic/beats/metricbeat/mb"
 
  "github.com/vmware/govmomi"
@@ -38,20 +37,20 @@ import (
  "github.com/vmware/govmomi/vim25/types"
 )
 
-var logger = logp.NewLogger("vsphere")
-
 func init() {
  mb.Registry.MustAddMetricSet("vsphere", "host", New,
  mb.DefaultMetricSet(),
  )
 }
 
+// MetricSet type defines all fields of the MetricSet
 type MetricSet struct {
  mb.BaseMetricSet
  HostURL *url.URL
  Insecure bool
 }
 
+// New create a new instance of the MetricSet
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
  cfgwarn.Beta("The vsphere host metricset is beta")
 
@@ -79,16 +78,17 @@ 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) error {
 
  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()
 
- var events []common.MapStr
-
  client, err := govmomi.NewClient(ctx, m.HostURL, m.Insecure)
  if err != nil {
- return nil, err
+ return errors.Wrap(err, "error in NewClient")
  }
 
  defer client.Logout(ctx)
@@ -100,7 +100,7 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 
  v, err := mgr.CreateContainerView(ctx, c.ServiceContent.RootFolder, []string{"HostSystem"}, true)
  if err != nil {
- return nil, err
+ return errors.Wrap(err, "error in CreateContainerView")
  }
 
  defer v.Destroy(ctx)
@@ -109,7 +109,7 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
  var hst []mo.HostSystem
  err = v.Retrieve(ctx, []string{"HostSystem"}, []string{"summary"}, &hst)
  if err != nil {
- return nil, err
+ return errors.Wrap(err, "error in Retrieve")
  }
 
  for _, hs := range hst {
@@ -127,24 +127,25 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
  event.Put("memory.free.bytes", int64(hs.Summary.Hardware.MemorySize)-(int64(hs.Summary.QuickStats.OverallMemoryUsage)*1024*1024))
  event.Put("memory.total.bytes", hs.Summary.Hardware.MemorySize)
  } else {
- logger.Debug("'Hardware' or 'Summary' data not found. This is either a parsing error from vsphere library, an error trying to reach host/guest or incomplete information returned from host/guest")
+ m.Logger().Debug("'Hardware' or 'Summary' data not found. This is either a parsing error from vsphere library, an error trying to reach host/guest or incomplete information returned from host/guest")
  }
 
  if hs.Summary.Host != nil {
  networkNames, err := getNetworkNames(ctx, c, hs.Summary.Host.Reference())
  if err != nil {
- logger.Debugf("error trying to get network names: %s", err.Error())
+ m.Logger().Debugf("error trying to get network names: %s", err.Error())
  } else {
  if len(networkNames) > 0 {
  event["network_names"] = networkNames
  }
  }
  }
-
- events = append(events, event)
+ reporter.Event(mb.Event{
+ MetricSetFields: event,
+ })
  }
 
- return events, nil
+ return nil
 }
 
 func getNetworkNames(ctx context.Context, c *vim25.Client, ref types.ManagedObjectReference) ([]string, error) {

metricbeat/module/vsphere/host/host_test.go

@@ -29,22 +29,22 @@ import (
 
 func TestFetchEventContents(t *testing.T) {
  model := simulator.ESX()
- err := model.Create()
  if err := model.Create(); err != nil {
  t.Fatal(err)
  }
 
  ts := model.Service.NewServer()
  defer ts.Close()
 
- f := mbtest.NewEventsFetcher(t, getConfig(ts))
-
- events, err := f.Fetch()
- if err != nil {
- t.Fatal("fetch error", err)
+ f := mbtest.NewReportingMetricSetV2Error(t, getConfig(ts))
+ events, errs := mbtest.ReportingFetchV2Error(f)
+ if len(errs) > 0 {
+ t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
  }
 
- event := events[0]
+ assert.NotEmpty(t, events)
+
+ event := events[0].MetricSetFields
 
  t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event.StringToPrint())
 
@@ -82,9 +82,9 @@ func TestData(t *testing.T) {
  ts := model.Service.NewServer()
  defer ts.Close()
 
- f := mbtest.NewEventsFetcher(t, getConfig(ts))
+ f := mbtest.NewReportingMetricSetV2Error(t, getConfig(ts))
 
- if err := mbtest.WriteEvents(f, t); err != nil {
+ if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
  t.Fatal("write", err)
  }
 }