Atomic commit

metricbeat/docs/modules/ceph/monitor_health.asciidoc

@@ -13,9 +13,3 @@ include::../../../module/ceph/monitor_health/_meta/docs.asciidoc[]
 For a description of each field in the metricset, see the
 <<exported-fields-ceph,exported fields>> section.
 
-Here is an example document generated by this metricset:
-
-[source,json]
-----
-include::../../../module/ceph/monitor_health/_meta/data.json[]
-----

metricbeat/module/ceph/monitor_health/_meta/testdata/config.yml

@@ -0,0 +1,3 @@
+type: http
+url: "/api/v0.1/health"
+suffix: plain

metricbeat/module/ceph/monitor_health/_meta/testdata/input.plain

@@ -0,0 +1 @@
+{"status": "OK", "output": {"detail": [], "timechecks": {"round_status": "finished", "epoch": 3, "round": 0}, "health": {"health_services": [{"mons": [{"last_updated": "2019-03-13 11:21:24.667025", "name": "26c372192772", "avail_percent": 82, "kb_total": 936145620, "kb_avail": 773013080, "health": "HEALTH_OK", "kb_used": 115509168, "store_stats": {"bytes_total": 4195407, "bytes_log": 4128768, "last_updated": "0.000000", "bytes_misc": 65552, "bytes_sst": 1087}}]}]}, "overall_status": "HEALTH_OK", "summary": []}}

metricbeat/module/ceph/monitor_health/_meta/testdata/input.plain-expected.json

@@ -0,0 +1,48 @@
+[
+ {
+ "ceph": {
+ "monitor_health": {
+ "available": {
+ "kb": 773013080,
+ "pct": 82
+ },
+ "health": "HEALTH_OK",
+ "last_updated": "2019-03-13T11:21:24.667025Z",
+ "name": "26c372192772",
+ "store_stats": {
+ "last_updated": "0.000000",
+ "log": {
+ "bytes": 4128768
+ },
+ "misc": {
+ "bytes": 65552
+ },
+ "sst": {
+ "bytes": 1087
+ },
+ "total": {
+ "bytes": 4195407
+ }
+ },
+ "total": {
+ "kb": 936145620
+ },
+ "used": {
+ "kb": 115509168
+ }
+ }
+ },
+ "event": {
+ "dataset": "ceph.monitor_health",
+ "duration": 115000,
+ "module": "ceph"
+ },
+ "metricset": {
+ "name": "monitor_health"
+ },
+ "service": {
+ "address": "127.0.0.1:55555",
+ "type": "ceph"
+ }
+ }
+]

metricbeat/module/ceph/monitor_health/data.go

@@ -21,8 +21,9 @@ import (
  "encoding/json"
  "time"
 
+ "github.com/pkg/errors"
+
  "github.com/elastic/beats/libbeat/common"
- "github.com/elastic/beats/libbeat/logp"
 )
 
 type Tick struct {
@@ -85,11 +86,11 @@ type HealthRequest struct {
  Output Output `json:"output"`
 }
 
-func eventsMapping(content []byte) []common.MapStr {
+func eventsMapping(content []byte) ([]common.MapStr, error) {
  var d HealthRequest
  err := json.Unmarshal(content, &d)
  if err != nil {
- logp.Err("Error: %+v", err)
+ return nil, errors.Wrapf(err, "could not get HealthRequest data")
  }
 
  events := []common.MapStr{}
@@ -131,5 +132,5 @@ func eventsMapping(content []byte) []common.MapStr {
  }
  }
 
- return events
+ return events, nil
 }

metricbeat/module/ceph/monitor_health/monitor_health.go

@@ -18,7 +18,6 @@
 package monitor_health
 
 import (
- "github.com/elastic/beats/libbeat/common"
  "github.com/elastic/beats/metricbeat/helper"
  "github.com/elastic/beats/metricbeat/mb"
  "github.com/elastic/beats/metricbeat/mb/parse"
@@ -61,11 +60,23 @@ 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 {
  content, err := m.HTTP.FetchContent()
  if err != nil {
- return nil, err
+ return err
+ }
+
+ events, err := eventsMapping(content)
+ if err != nil {
+ return err
+ }
+
+ for _, event := range events {
+ reporter.Event(mb.Event{MetricSetFields: event})
  }
 
- return eventsMapping(content), nil
+ return nil
 }

metricbeat/module/ceph/monitor_health/monitor_health_test.go

@@ -32,6 +32,7 @@ import (
 
 func TestFetchEventContents(t *testing.T) {
  absPath, err := filepath.Abs("../_meta/testdata/")
+ assert.NoError(t, err)
 
  response, err := ioutil.ReadFile(absPath + "/sample_response.json")
  server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -47,12 +48,13 @@ func TestFetchEventContents(t *testing.T) {
  "hosts": []string{server.URL},
  }
 
- f := mbtest.NewEventsFetcher(t, config)
- events, err := f.Fetch()
- if err != nil {
- t.Fatal(err)
+ 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)
  }
- event := events[0]
+ assert.NotEmpty(t, events)
+ event := events[0].MetricSetFields
 
  t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event.StringToPrint())