[Metricbeat] Migrate Apache status Metricset to use ReporterV2 interface

metricbeat/module/apache/status/_meta/data.json

@@ -1,65 +1,66 @@
 {
- "@timestamp": "2017-10-12T08:05:34.853Z",
+ "@timestamp": "2019-03-01T08:05:34.853Z",
  "apache": {
  "status": {
- "bytes_per_request": 78.7692,
- "bytes_per_sec": 4.94686,
+ "bytes_per_request": 112.913,
+ "bytes_per_sec": 103.832,
  "connections": {
  "async": {
  "closing": 0,
- "keep_alive": 1,
+ "keep_alive": 0,
  "writing": 0
  },
- "total": 1
+ "total": 0
  },
  "cpu": {
  "children_system": 0,
  "children_user": 0,
- "load": 3.99758,
- "system": 16.39,
+ "load": 0.125874,
+ "system": 0.2,
  "user": 0.16
  },
- "hostname": "apache",
  "load": {
- "1": 170.11,
- "15": 77.62,
- "5": 167.51
+ "1": 2,
+ "15": 1.91,
+ "5": 1.85
  },
- "requests_per_sec": 0.0628019,
+ "requests_per_sec": 0.91958,
  "scoreboard": {
  "closing_connection": 0,
  "dns_lookup": 0,
  "gracefully_finishing": 0,
  "idle_cleanup": 0,
  "keepalive": 0,
  "logging": 0,
- "open_slot": 300,
+ "open_slot": 325,
  "reading_request": 0,
  "sending_reply": 1,
  "starting_up": 0,
  "total": 400,
- "waiting_for_connection": 99
+ "waiting_for_connection": 74
  },
- "total_accesses": 26,
- "total_kbytes": 2,
+ "total_accesses": 263,
+ "total_kbytes": 29,
  "uptime": {
- "server_uptime": 414,
- "uptime": 414
+ "server_uptime": 286,
+ "uptime": 286
  },
  "workers": {
  "busy": 1,
- "idle": 99
+ "idle": 74
  }
  }
  },
- "beat": {
- "hostname": "host.example.com",
- "name": "host.example.com"
+ "event": {
+ "dataset": "apache.status",
+ "duration": 115000,
+ "module": "apache"
  },
  "metricset": {
- "host": "apache",
- "module": "apache",
- "name": "status",
- "rtt": 115
+ "name": "status"
+ },
+ "service": {
+ "address": "127.0.0.1:55555",
+ "type": "apache"
  }
 }

metricbeat/module/apache/status/_meta/testdata/config.yml

@@ -0,0 +1,5 @@
+type: http
+url: "/server-status?auto="
+suffix: plain
+remove_fields_from_comparison:
+ - "apache.status.hostname"

metricbeat/module/apache/status/_meta/testdata/docs.plain

@@ -0,0 +1,31 @@
+172.26.0.2
+ServerVersion: Apache/2.4.20 (Unix)
+ServerMPM: event
+Server Built: Jun 7 2016 17:55:26
+CurrentTime: Thursday, 04-Apr-2019 09:32:31 UTC
+RestartTime: Thursday, 04-Apr-2019 09:27:44 UTC
+ParentServerConfigGeneration: 1
+ParentServerMPMGeneration: 0
+ServerUptimeSeconds: 286
+ServerUptime: 4 minutes 46 seconds
+Load1: 2.00
+Load5: 1.85
+Load15: 1.91
+Total Accesses: 263
+Total kBytes: 29
+CPUUser: .16
+CPUSystem: .2
+CPUChildrenUser: 0
+CPUChildrenSystem: 0
+CPULoad: .125874
+Uptime: 286
+ReqPerSec: .91958
+BytesPerSec: 103.832
+BytesPerReq: 112.913
+BusyWorkers: 1
+IdleWorkers: 74
+ConnsTotal: 0
+ConnsAsyncWriting: 0
+ConnsAsyncKeepAlive: 0
+ConnsAsyncClosing: 0
+Scoreboard: ______________________________________W____________________________________.....................................................................................................................................................................................................................................................................................................................................

metricbeat/module/apache/status/_meta/testdata/docs.plain-expected.json

@@ -0,0 +1,68 @@
+[
+ {
+ "apache": {
+ "status": {
+ "bytes_per_request": 112.913,
+ "bytes_per_sec": 103.832,
+ "connections": {
+ "async": {
+ "closing": 0,
+ "keep_alive": 0,
+ "writing": 0
+ },
+ "total": 0
+ },
+ "cpu": {
+ "children_system": 0,
+ "children_user": 0,
+ "load": 0.125874,
+ "system": 0.2,
+ "user": 0.16
+ },
+ "hostname": "127.0.0.1:43985",
+ "load": {
+ "1": 2,
+ "15": 1.91,
+ "5": 1.85
+ },
+ "requests_per_sec": 0.91958,
+ "scoreboard": {
+ "closing_connection": 0,
+ "dns_lookup": 0,
+ "gracefully_finishing": 0,
+ "idle_cleanup": 0,
+ "keepalive": 0,
+ "logging": 0,
+ "open_slot": 325,
+ "reading_request": 0,
+ "sending_reply": 1,
+ "starting_up": 0,
+ "total": 400,
+ "waiting_for_connection": 74
+ },
+ "total_accesses": 263,
+ "total_kbytes": 29,
+ "uptime": {
+ "server_uptime": 286,
+ "uptime": 286
+ },
+ "workers": {
+ "busy": 1,
+ "idle": 74
+ }
+ }
+ },
+ "event": {
+ "dataset": "apache.status",
+ "duration": 115000,
+ "module": "apache"
+ },
+ "metricset": {
+ "name": "status"
+ },
+ "service": {
+ "address": "127.0.0.1:55555",
+ "type": "apache"
+ }
+ }
+]

metricbeat/module/apache/status/status.go

@@ -19,7 +19,8 @@
 package status
 
 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"
@@ -76,13 +77,20 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
  }, nil
 }
 
-// Fetch makes an HTTP request to fetch status metrics from the mod_status endpoint.
-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 {
  scanner, err := m.http.FetchScanner()
  if err != nil {
- return nil, err
+ return errors.Wrap(err, "error fetching data")
  }
 
  data, _ := eventMapping(scanner, m.Host())
- return data, nil
+
+ if reported := reporter.Event(mb.Event{MetricSetFields: data}); !reported {
+ m.Logger().Error("error reporting event")
+ }
+
+ return nil
 }

metricbeat/module/apache/status/status_integration_test.go

@@ -32,27 +32,25 @@ import (
 func TestFetch(t *testing.T) {
  compose.EnsureUp(t, "apache")
 
- f := mbtest.NewEventFetcher(t, getConfig())
- event, err := f.Fetch()
- if !assert.NoError(t, err) {
- t.FailNow()
+ f := mbtest.NewReportingMetricSetV2Error(t, getConfig())
+ events, errs := mbtest.ReportingFetchV2Error(f)
+ if len(errs) > 0 {
+ t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
  }
+ assert.NotEmpty(t, events)
+ event := events[0]
 
  t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
 
  // Check number of fields.
- if len(event) < 11 {
+ if len(event.MetricSetFields) < 11 {
  t.Fatal("Too few top-level elements in the event")
  }
 }
 
 func TestData(t *testing.T) {
- compose.EnsureUp(t, "apache")
-
- f := mbtest.NewEventFetcher(t, getConfig())
-
- err := mbtest.WriteEvent(f, t)
- if err != nil {
+ f := mbtest.NewReportingMetricSetV2Error(t, getConfig())
+ if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
  t.Fatal("write", err)
  }
 }

metricbeat/module/apache/status/status_test.go

@@ -26,6 +26,7 @@ import (
  "net/http/httptest"
  "os"
  "path/filepath"
+ "strings"
  "sync"
  "testing"
  "time"
@@ -85,11 +86,13 @@ func TestFetchEventContents(t *testing.T) {
  "hosts": []string{server.URL},
  }
 
- f := mbtest.NewEventFetcher(t, config)
- event, err := f.Fetch()
- if !assert.NoError(t, err) {
- t.FailNow()
+ 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)
  }
+ assert.NotEmpty(t, events)
+ event := events[0].MetricSetFields
 
  t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event.StringToPrint())
 
@@ -162,13 +165,23 @@ func TestFetchTimeout(t *testing.T) {
  "timeout": "50ms",
  }
 
- f := mbtest.NewEventFetcher(t, config)
+ f := mbtest.NewReportingMetricSetV2Error(t, config)
 
  start := time.Now()
- _, err := f.Fetch()
+ events, errs := mbtest.ReportingFetchV2Error(f)
+ if len(errs) == 0 {
+ t.Fatalf("Expected an error, had %d. %v\n", len(errs), errs)
+ }
+ assert.Empty(t, events)
  elapsed := time.Since(start)
- if assert.Error(t, err) {
- assert.Contains(t, err.Error(), "request canceled (Client.Timeout exceeded")
+ var found bool
+ for _, err := range errs {
+ if strings.Contains(err.Error(), "request canceled (Client.Timeout exceeded") {
+ found = true
+ }
+ }
+ if !found {
+ assert.Failf(t, "", "expected an error containing 'request canceled (Client.Timeout exceeded'. Got %v", errs)
  }
 
  // Elapsed should be ~50ms, sometimes it can be up to 1s
@@ -201,12 +214,12 @@ func TestMultipleFetches(t *testing.T) {
  "hosts": []string{server.URL},
  }
 
- f := mbtest.NewEventFetcher(t, config)
+ f := mbtest.NewReportingMetricSetV2Error(t, config)
 
  for i := 0; i < 20; i++ {
- _, err := f.Fetch()
- if !assert.NoError(t, err) {
- t.FailNow()
+ _, errs := mbtest.ReportingFetchV2Error(f)
+ if len(errs) > 0 {
+ t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
  }
  }