System test cleanup for metricbeat. (#1398)

- Fix misspelled redis-info.persistence key name.
- Added an 'integration' attribute to metricbeat system tests that require extra services like Redis.
- Documented metricset-host in fields.yml (many more to document).
- Change instantaneous_input_kbps and instantaneous_output_kbps to floats from ints. There was an
error in the log that indicated the problem. The test has been updated to check for ERR and WARN.

libbeat/scripts/Makefile

@@ -143,7 +143,7 @@ system-tests: buildbeat.test prepare-tests python-env
 # Runs system tests without coverage reports and in parallel
 .PHONY: fast-system-tests
 fast-system-tests: buildbeat.test python-env
-	. ${PYTHON_ENV}/bin/activate; nosetests -w tests/system --processes=$(PROCESSES) --process-timeout=$(TIMEOUT)
+	. ${PYTHON_ENV}/bin/activate; nosetests -w tests/system --processes=$(PROCESSES) --process-timeout=$(TIMEOUT) -a '!integration'
 
 # Run benchmark tests
 .PHONY: benchmark-tests

libbeat/tests/system/beat/beat.py

@@ -264,6 +264,18 @@ def wait_until(self, cond, max_timeout=10, poll_interval=0.1, name="cond"):
                                 "Waited {} seconds.".format(max_timeout))
             time.sleep(poll_interval)
 
+    def get_log(self, logfile=None):
+        """
+        Returns the log as a string.
+        """
+        if logfile is None:
+            logfile = self.beat_name + ".log"
+
+        with open(os.path.join(self.working_dir, logfile), 'r') as f:
+            data=f.read()
+
+        return data
+
     def log_contains(self, msg, logfile=None):
         """
         Returns true if the give logfile contains the given message.

metricbeat/beater/metricbeat.go

@@ -27,29 +27,28 @@ for each MetricSet to prevent type conflicts. Also all values are stored under t
 package beater
 
 import (
+	"fmt"
+
 	"github.com/elastic/beats/libbeat/beat"
-	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/include"
 )
 
 type Metricbeat struct {
-	done     chan struct{}
-	MbConfig *Config
+	done   chan struct{}
+	config *Config
 }
 
-// New creates a new Metricbeat instance
+// New creates and returns a new Metricbeat instance.
 func New() *Metricbeat {
 	return &Metricbeat{}
 }
 
 func (mb *Metricbeat) Config(b *beat.Beat) error {
-
-	mb.MbConfig = &Config{}
-	err := b.RawConfig.Unpack(mb.MbConfig)
+	mb.config = &Config{}
+	err := b.RawConfig.Unpack(mb.config)
 	if err != nil {
-		logp.Err("Error reading configuration file: %v", err)
-		return err
+		return fmt.Errorf("error reading configuration file. %v", err)
 	}
 
 	// List all registered modules and metricsets
@@ -64,9 +63,8 @@ func (mb *Metricbeat) Setup(b *beat.Beat) error {
 }
 
 func (mb *Metricbeat) Run(b *beat.Beat) error {
-
 	// Checks all defined metricsets and starts a module for each entry with the defined metricsets
-	for _, moduleConfig := range mb.MbConfig.Metricbeat.Modules {
+	for _, moduleConfig := range mb.config.Metricbeat.Modules {
 
 		module, err := helper.Registry.GetModule(moduleConfig)
 		if err != nil {

metricbeat/docs/fields.asciidoc

@@ -44,6 +44,11 @@ required: True
 The timestamp when the log line was read. The precision is in milliseconds. The timezone is UTC.
 
 
+==== metricset-host
+
+Hostname of the machine from which the metricset was collected. This field may not be present when the data was collected locally.
+
+
 ==== rtt
 
 type: long

metricbeat/etc/fields.yml

@@ -38,6 +38,11 @@ common:
         The timestamp when the log line was read. The precision is in
         milliseconds. The timezone is UTC.
 
+    - name: metricset-host
+      description: >
+        Hostname of the machine from which the metricset was collected. This
+        field may not be present when the data was collected locally.
+
     - name: rtt
       type: long
       required: true

metricbeat/etc/fields_base.yml

@@ -38,6 +38,11 @@ common:
         The timestamp when the log line was read. The precision is in
         milliseconds. The timezone is UTC.
 
+    - name: metricset-host
+      description: >
+        Hostname of the machine from which the metricset was collected. This
+        field may not be present when the data was collected locally.
+
     - name: rtt
       type: long
       required: true

metricbeat/module/redis/info/data.go

@@ -35,7 +35,7 @@ func eventMapping(info map[string]string) common.MapStr {
 			"used_memory_lua":  toInt(info["used_memory_lua"]),
 			"mem_allocator":    info["mem_allocator"], // Could be moved to server as it rarely changes
 		},
-		"presistence": common.MapStr{
+		"persistence": common.MapStr{
 			"loading":                      toBool(info["loading"]),
 			"rdb_changes_since_last_save":  toInt(info["rdb_changes_since_last_save"]),
 			"rdb_bgsave_in_progress":       toBool(info["rdb_bgsave_in_progress"]),
@@ -85,8 +85,8 @@ func eventMapping(info map[string]string) common.MapStr {
 			"instantaneous_ops_per_sec":  toInt(info["instantaneous_ops_per_sec"]),
 			"total_net_input_bytes":      toInt(info["total_net_input_bytes"]),
 			"total_net_output_bytes":     toInt(info["total_net_output_bytes"]),
-			"instantaneous_input_kbps":   toInt(info["instantaneous_input_kbps"]),
-			"instantaneous_output_kbps":  toInt(info["instantaneous_output_kbps"]),
+			"instantaneous_input_kbps":   toFloat(info["instantaneous_input_kbps"]),
+			"instantaneous_output_kbps":  toFloat(info["instantaneous_output_kbps"]),
 			"rejected_connections":       toInt(info["rejected_connections"]),
 			"sync_full":                  toInt(info["sync_full"]),
 			"sync_partial_ok":            toInt(info["sync_partial_ok"]),

metricbeat/module/redis/info/info.go

@@ -12,6 +12,10 @@ import (
 	"github.com/elastic/beats/metricbeat/helper"
 )
 
+var (
+	debugf = logp.MakeDebug("redis")
+)
+
 func init() {
 	helper.Registry.AddMetricSeter("redis", "info", New)
 }
@@ -75,14 +79,13 @@ func createPool(host, password, network string, maxConn int, timeout time.Durati
 }
 
 func (m *MetricSeter) Fetch(ms *helper.MetricSet, host string) (events common.MapStr, err error) {
-
 	// Fetch default INFO
 	info, err := m.fetchRedisStats(host, "default")
-
 	if err != nil {
 		return nil, err
 	}
 
+	debugf("Redis INFO from %s: %+v", host, info)
 	return eventMapping(info), nil
 }
 

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

@@ -3,7 +3,7 @@ metricbeat:
     {% if redis %}
     - module: redis
       hosts:
-        - "{{ redis_host | default("127.0.0.1", true) }}:6379"
+        - "{{ redis_host|default("127.0.0.1", true) }}:6379"
       metricsets:
         - info
       period: 2s
@@ -12,7 +12,7 @@ metricbeat:
     {% endif %}
     {% if mysql %}
     - module: mysql:
-      # This expectd a full mysql dsn
+      # This expects a full MySQL Data Source Name (DSN).
       # Example: [username[:password]@][protocol[(address)]]/
       hosts:
         - "@tcp(127.0.0.1:3306)/"
@@ -22,44 +22,11 @@ metricbeat:
       enabled: true
     {% endif %}
 
-
-############################# Output ############################################
-
-# Configure what outputs to use when sending the data collected by metricbeat.
-# You can enable one or multiple outputs by setting enabled option to true.
 output:
-
-  # Elasticsearch as output
-  # Options:
-  # host, port: where Elasticsearch is listening on
-  # save_topology: specify if the topology is saved in Elasticsearch
-  #elasticsearch:
-  #  enabled: false
-  #  host: localhost
-  #  port: 9200
-  #  save_topology: true
-
-  # Redis as output
-  # Options:
-  # host, port: where Redis is listening on
-  # save_topology: specify if the topology is saved in Redis
-  #redis:
-  #  enabled: false
-  #  host: localhost
-  #  port: 6379
-  #  save_topology: true
-
-  # File as output
-  # Options
-  # path: where to save the files
-  # filename: name of the files
-  # rotate_every_kb: maximum size of the files in path
-  # number of files: maximum number of files in path
   file:
     enabled: true
     path: {{ output_file_path|default(beat.working_dir + "/output") }}
     filename: "{{ output_file_filename|default("metricbeat") }}"
     rotate_every_kb: 1000
-    #number_of_files: 7
 
 # vim: set ft=jinja:

metricbeat/tests/system/metricbeat.py

@@ -0,0 +1,26 @@
+import sys
+
+sys.path.append('../../../libbeat/tests/system')
+from beat.beat import TestCase
+
+COMMON_FIELDS = ["@timestamp", "beat", "metricset", "metricset-host",
+                 "module", "rtt", "type"]
+
+
+class BaseTest(TestCase):
+    @classmethod
+    def setUpClass(self):
+        self.beat_name = "metricbeat"
+        self.build_path = "../../build/system-tests/"
+        self.beat_path = "../../metricbeat.test"
+
+    def assert_fields_are_documented(self, evt):
+        """
+        Assert that all keys present in evt are documented in fields.yml.
+        """
+        expected_fields, _ = self.load_fields()
+        flat = self.flatten_object(evt, [])
+
+        for key in flat.keys():
+            if key not in expected_fields:
+                raise Exception("Key '{}' found in event is not documented!".format(key))

metricbeat/tests/system/test_base.py

@@ -1,15 +1,24 @@
-from metricbeat.metricbeat import BaseTest
+import re
+from metricbeat import BaseTest
 
-class Test(BaseTest):
 
-    def test_base(self):
+class Test(BaseTest):
+    def test_start_stop(self):
         """
-        Basic test with exiting Mockbeat normally
+        Metricbeat starts and stops without error.
         """
-        self.render_config_template(
-        )
-
+        self.render_config_template()
         proc = self.start_beat()
-        self.wait_until( lambda: self.log_contains("Setup Beat"))
-        exit_code = proc.kill_and_wait()
-        assert exit_code == 0
+        self.wait_until(lambda: self.log_contains("Setup Beat"))
+        proc.check_kill_and_wait()
+
+        # Ensure no errors or warnings exist in the log.
+        log = self.get_log()
+        self.assertNotRegexpMatches(log, "ERR|WARN")
+
+        # Ensure all Beater stages are used.
+        self.assertRegexpMatches(log, re.compile(".*".join([
+            "Setup Beat: metricbeat",
+            "metricbeat start running",
+            "metricbeat cleanup"
+        ]), re.DOTALL))