[Metricbeat] Migrate HTTP json Metricset to use ReporterV2 interface

metricbeat/module/http/json/_meta/data.json

@@ -1,18 +1,20 @@
 {
-    "@timestamp": "2017-10-12T08:05:34.853Z",
-    "beat": {
-        "hostname": "host.example.com",
-        "name": "host.example.com"
+    "@timestamp": "2019-03-01T08:05:34.853Z",
+    "event": {
+        "dataset": "http.test",
+        "duration": 115000,
+        "module": "http"
     },
     "http": {
-        "json": {
+        "test": {
             "hello": "world"
         }
     },
     "metricset": {
-        "host": "127.0.0.1:8080",
-        "module": "http",
-        "name": "json",
-        "rtt": 115
+        "name": "json"
+    },
+    "service": {
+        "address": "127.0.0.1:55555",
+        "type": "http"
     }
 }

metricbeat/module/http/json/_meta/testdata/config.yml

@@ -0,0 +1,6 @@
+type: http
+url: "/"
+module:
+  namespace: test
+omit_documented_fields_check:
+  - "http.test.*"

metricbeat/module/http/json/_meta/testdata/docs.json

@@ -0,0 +1,3 @@
+{
+    "hello": "world"
+}

metricbeat/module/http/json/_meta/testdata/docs.json-expected.json

@@ -0,0 +1,21 @@
+[
+    {
+        "event": {
+            "dataset": "http.test",
+            "duration": 115000,
+            "module": "http"
+        },
+        "http": {
+            "test": {
+                "hello": "world"
+            }
+        },
+        "metricset": {
+            "name": "json"
+        },
+        "service": {
+            "address": "127.0.0.1:55555",
+            "type": "http"
+        }
+    }
+]

metricbeat/module/http/json/data.go

@@ -0,0 +1,78 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package json
+
+import (
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/metricbeat/mb"
+)
+
+func (m *MetricSet) processBody(response *http.Response, jsonBody interface{}) mb.Event {
+	var event common.MapStr
+
+	if m.deDotEnabled {
+		event = common.DeDotJSON(jsonBody).(common.MapStr)
+	} else {
+		event = jsonBody.(common.MapStr)
+	}
+
+	if m.requestEnabled {
+		event[mb.ModuleDataKey] = common.MapStr{
+			"request": common.MapStr{
+				"headers": m.getHeaders(response.Request.Header),
+				"method":  response.Request.Method,
+				"body": common.MapStr{
+					"content": m.body,
+				},
+			},
+		}
+	}
+
+	if m.responseEnabled {
+		phrase := strings.TrimPrefix(response.Status, strconv.Itoa(response.StatusCode)+" ")
+		event[mb.ModuleDataKey] = common.MapStr{
+			"response": common.MapStr{
+				"code":    response.StatusCode,
+				"phrase":  phrase,
+				"headers": m.getHeaders(response.Header),
+			},
+		}
+	}
+
+	return mb.Event{
+		MetricSetFields: event,
+		Namespace:       "http." + m.namespace,
+	}
+}
+
+func (m *MetricSet) getHeaders(header http.Header) map[string]string {
+	headers := make(map[string]string)
+	for k, v := range header {
+		value := ""
+		for _, h := range v {
+			value += h + " ,"
+		}
+		value = strings.TrimRight(value, " ,")
+		headers[k] = value
+	}
+	return headers
+}

metricbeat/module/http/json/json.go

@@ -20,9 +20,8 @@ package json
 import (
 	"encoding/json"
 	"io/ioutil"
-	"net/http"
-	"strconv"
-	"strings"
+
+	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/metricbeat/helper"
@@ -116,95 +115,52 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}, nil
 }
 
-func (m *MetricSet) processBody(response *http.Response, jsonBody interface{}) common.MapStr {
-	var event common.MapStr
-
-	if m.deDotEnabled {
-		event = common.DeDotJSON(jsonBody).(common.MapStr)
-	} else {
-		event = jsonBody.(common.MapStr)
-	}
-
-	if m.requestEnabled {
-		event[mb.ModuleDataKey] = common.MapStr{
-			"request": common.MapStr{
-				"headers": m.getHeaders(response.Request.Header),
-				"method":  response.Request.Method,
-				"body": common.MapStr{
-					"content": m.body,
-				},
-			},
-		}
-	}
-
-	if m.responseEnabled {
-		phrase := strings.TrimPrefix(response.Status, strconv.Itoa(response.StatusCode)+" ")
-		event[mb.ModuleDataKey] = common.MapStr{
-			"response": common.MapStr{
-				"code":    response.StatusCode,
-				"phrase":  phrase,
-				"headers": m.getHeaders(response.Header),
-			},
-		}
-	}
-
-	// Set dynamic namespace
-	event["_namespace"] = m.namespace
-
-	return event
-}
-
-// 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) error {
 	response, err := m.http.FetchResponse()
 	if err != nil {
-		return nil, err
+		return err
 	}
-	defer response.Body.Close()
-
-	var jsonBody common.MapStr
-	var jsonBodyArr []common.MapStr
-	var events []common.MapStr
+	defer func() {
+		if err := response.Body.Close(); err != nil {
+			m.Logger().Debug("error closing http body")
+		}
+	}()
 
 	body, err := ioutil.ReadAll(response.Body)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	if m.jsonIsArray {
-		err = json.Unmarshal(body, &jsonBodyArr)
-		if err != nil {
-			return nil, err
+		var jsonBodyArr []common.MapStr
+		if err = json.Unmarshal(body, &jsonBodyArr); err != nil {
+			return err
 		}
 
 		for _, obj := range jsonBodyArr {
 			event := m.processBody(response, obj)
-			events = append(events, event)
+
+			if reported := reporter.Event(event); !reported {
+				m.Logger().Debug(errors.Errorf("error reporting event: %#v", event))
+				return nil
+			}
 		}
 	} else {
-		err = json.Unmarshal(body, &jsonBody)
-		if err != nil {
-			return nil, err
+		var jsonBody common.MapStr
+		if err = json.Unmarshal(body, &jsonBody); err != nil {
+			return err
 		}
 
 		event := m.processBody(response, jsonBody)
-		events = append(events, event)
-	}
 
-	return events, nil
-}
-
-func (m *MetricSet) getHeaders(header http.Header) map[string]string {
-	headers := make(map[string]string)
-	for k, v := range header {
-		value := ""
-		for _, h := range v {
-			value += h + " ,"
+		if reported := reporter.Event(event); !reported {
+			m.Logger().Debug(errors.Errorf("error reporting event: %#v", event))
+			return nil
 		}
-		value = strings.TrimRight(value, " ,")
-		headers[k] = value
 	}
-	return headers
+
+	return nil
 }

metricbeat/module/http/json/json_integration_test.go

@@ -32,35 +32,35 @@ import (
 func TestFetchObject(t *testing.T) {
 	compose.EnsureUp(t, "http")
 
-	f := mbtest.NewEventsFetcher(t, getConfig("object"))
-	event, err := f.Fetch()
-	if !assert.NoError(t, err) {
-		t.FailNow()
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig("object"))
+	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)
 
-	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
+	t.Logf("%s/%s event: %#v", f.Module().Name(), f.Name(), events)
 }
 
 func TestFetchArray(t *testing.T) {
 	compose.EnsureUp(t, "http")
 
-	f := mbtest.NewEventsFetcher(t, getConfig("array"))
-	event, err := f.Fetch()
-	if !assert.NoError(t, err) {
-		t.FailNow()
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig("array"))
+	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)
 
-	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
+	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), events[0])
 }
 func TestData(t *testing.T) {
 	compose.EnsureUp(t, "http")
 
-	f := mbtest.NewEventsFetcher(t, getConfig("object"))
-	err := mbtest.WriteEvents(f, t)
-	if err != nil {
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig("object"))
+	if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
-
 }
 
 func getConfig(jsonType string) map[string]interface{} {

metricbeat/tests/system/test_http.py

@@ -35,8 +35,7 @@ def test_json(self):
 
         assert evt["http"]["test"]["hello"] == "world"
 
-        # Delete dynamic namespace part for fields comparison
-        del evt["http"]["test"]
+        del evt["http"]["test"]["hello"]
 
         self.assertItemsEqual(self.de_dot(HTTP_FIELDS), evt.keys(), evt)