Refactor http.server metricset

CHANGELOG.asciidoc

@@ -44,6 +44,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Fixed typo in values for `state_container` `status.phase`, from `terminate` to `terminated`. {pull}6916[6916]
 - RabbitMQ management plugin path is now configured at the module level instead of having to do it in each of the metricsets. New `management_path_prefix` option should be used now {pull}7074[7074]
 - RabbitMQ node metricset only collects metrics of the instance it connects to, `node.collect: cluster` can be used to collect all nodes as before. {issue}6556[6556] {pull}6971[6971]
+- Change http/server metricset to put events by default under http.server and prefix config options with server.. {pull}7100[7100]
 
 *Packetbeat*
 

metricbeat/helper/server/http/http.go

@@ -62,7 +62,7 @@ func (h *HttpServer) Start() error {
 
 		logp.Info("Starting http server on %s", h.server.Addr)
 		err := h.server.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			logp.Critical("Unable to start HTTP server due to error: %v", err)
 		}
 	}()

metricbeat/module/http/server/_meta/docs.asciidoc

@@ -1 +1,15 @@
 This is the server metricset of the module http.
+
+Events sent to the http endpoint will be put by default under the `http.server` prefix. To change this use the `server.paths`
+config options. In the example below every request to `/foo` will be put under `http.foo`.
+
+["source","yaml",subs="attributes"]
+------------------------------------------------------------------------------
+- module: http
+  metricsets: ["server"]
+  host: "localhost"
+  port: "8080"
+  server.paths:
+    - path: "/foo"
+      namespace: "foo"
+------------------------------------------------------------------------------

metricbeat/module/http/server/config.go

@@ -7,8 +7,8 @@ import (
 )
 
 type HttpServerConfig struct {
-	Paths       []PathConfig `config:"paths"`
-	DefaultPath PathConfig   `config:"default_path"`
+	Paths       []PathConfig `config:"server.paths"`
+	DefaultPath PathConfig   `config:"server.default_path"`
 }
 
 type PathConfig struct {
@@ -21,7 +21,7 @@ func defaultHttpServerConfig() HttpServerConfig {
 	return HttpServerConfig{
 		DefaultPath: PathConfig{
 			Path:      "/",
-			Namespace: "http",
+			Namespace: "server",
 		},
 	}
 }

metricbeat/module/http/server/server.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
 	serverhelper "github.com/elastic/beats/metricbeat/helper/server"
 	"github.com/elastic/beats/metricbeat/helper/server/http"
@@ -50,7 +51,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 }
 
 // Run method provides the Graphite server with a reporter with which events can be reported.
-func (m *MetricSet) Run(reporter mb.PushReporter) {
+func (m *MetricSet) Run(reporter mb.PushReporterV2) {
 	// Start event watcher
 	m.server.Start()
 
@@ -60,10 +61,15 @@ func (m *MetricSet) Run(reporter mb.PushReporter) {
 			m.server.Stop()
 			return
 		case msg := <-m.server.GetEvents():
-			event, err := m.processor.Process(msg)
+			fields, err := m.processor.Process(msg)
 			if err != nil {
 				reporter.Error(err)
 			} else {
+				event := mb.Event{}
+				event.ModuleFields = common.MapStr{}
+				metricSetName := fields[mb.NamespaceKey].(string)
+				delete(fields, mb.NamespaceKey)
+				event.ModuleFields.Put(metricSetName, fields)
 				reporter.Event(event)
 			}
 

metricbeat/tests/system/config/metricbeat.yml.j2

@@ -41,6 +41,10 @@ metricbeat.modules:
   metrics_path: {{ m.path }}
   {% endif -%}
 
+  {% if m.port -%}
+  port: {{ m.port }}
+  {% endif -%}
+
   {% if m.timeout -%}
   timeout: {{ m.timeout }}
   {% endif -%}

metricbeat/tests/system/test_http.py

@@ -2,6 +2,8 @@
 import metricbeat
 import unittest
 from nose.plugins.attrib import attr
+import requests
+import time
 
 HTTP_FIELDS = metricbeat.COMMON_FIELDS + ["http"]
 
@@ -18,7 +20,7 @@ def test_json(self):
         self.render_config_template(modules=[{
             "name": "http",
             "metricsets": ["json"],
-            "hosts": self.get_hosts(),
+            "hosts": [self.get_host()],
             "period": "5s",
             "namespace": "test",
         }])
@@ -40,6 +42,38 @@ def test_json(self):
 
         self.assert_fields_are_documented(evt)
 
-    def get_hosts(self):
-        return ["http://" + os.getenv('HTTP_HOST', 'localhost') + ':' +
-                os.getenv('HTTP_PORT', '8080')]
+    def test_server(self):
+        """
+        http server metricset test
+        """
+        port = 8082
+        host = "localhost"
+        self.render_config_template(modules=[{
+            "name": "http",
+            "metricsets": ["server"],
+            "port": port,
+            "host": host,
+        }])
+        proc = self.start_beat()
+        self.wait_until(lambda: self.log_contains("Starting http server on "))
+        requests.post("http://" + host + ":" + str(port),
+                      json={'hello': 'world'}, headers={'Content-Type': 'application/json'})
+        self.wait_until(lambda: self.output_lines() > 0)
+        proc.check_kill_and_wait()
+        self.assert_no_logged_warnings()
+
+        output = self.read_output_json()
+        self.assertEqual(len(output), 1)
+        evt = output[0]
+
+        assert evt["http"]["server"]["hello"] == "world"
+
+        # Delete dynamic namespace part for fields comparison
+        del evt["http"]["server"]
+
+        self.assertItemsEqual(self.de_dot(HTTP_FIELDS), evt.keys(), evt)
+
+        self.assert_fields_are_documented(evt)
+
+    def get_host(self):
+        return "http://" + os.getenv('HTTP_HOST', 'localhost') + ':' + os.getenv('HTTP_PORT', '8080')