Add mapping for munin and options to override service type and name

CHANGELOG.next.asciidoc

@@ -72,6 +72,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Metricbeat*
 
+- Replace `namespace` option in munin module with `service.type` and `service.name` to override the ECS fields with the same names. {pull}10322[10322]
+- Add checks on valid field names and more strict mapping to munin module. {pull}10322[10322]
 - Refactor Prometheus metric mappings {pull}9948[9948]
 - Removed Prometheus stats metricset in favor of just using Prometheus collector {pull}9948[9948]
 

metricbeat/docs/fields.asciidoc

@@ -16986,11 +16986,22 @@ Munin node metrics exporter
 
 
 
-[float]
-== munin fields
+*`munin.metrics.*.*`*::
++
+--
+type: object
+
+munin.metrics contains metrics exposed by a munin node agent
+
 
-munin contains metrics exposed by a munin node agent
+--
+
+*`munin.metrics.*`*::
++
+--
+type: object
 
+--
 
 
 [[exported-fields-mysql]]

metricbeat/module/munin/_meta/fields.yml

@@ -4,8 +4,16 @@
     Munin node metrics exporter
   release: beta
   fields:
+    - name: munin.metrics.*.*
+      type: object
+      object_type: double
+      object_type_mapping_type: '*'
+      description: >
+          munin.metrics contains metrics exposed by a munin node agent
+    - name: munin.metrics.*
+      type: object
+      object_type: object
+      object_type_mapping_type: '*'
     - name: munin
       type: group
-      description: >
-          munin contains metrics exposed by a munin node agent
       fields:

metricbeat/module/munin/munin.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -36,6 +37,12 @@ const (
 	unknownValue = "U"
 )
 
+var (
+	// Field names must match with this expression
+	// http://guide.munin-monitoring.org/en/latest/reference/plugin.html#notes-on-fieldnames
+	nameRegexp = regexp.MustCompile("^[a-zA-Z_][a-zA-Z0-9_]*$")
+)
+
 // Node connection
 type Node struct {
 	conn net.Conn
@@ -110,6 +117,11 @@ func (n *Node) Fetch(items ...string) (common.MapStr, error) {
 			}
 			value := scanner.Text()
 
+			if !nameRegexp.MatchString(name) {
+				errs = append(errs, fmt.Errorf("field name '%s' doesn't match expected format '%s'", name, nameRegexp.String()))
+				continue
+			}
+
 			key := fmt.Sprintf("%s.%s", item, name)
 
 			if value == unknownValue {

metricbeat/module/munin/munin_test.go

@@ -45,8 +45,8 @@ func TestList(t *testing.T) {
 	assert.ElementsMatch(t, expected, list)
 }
 
-func TestFetch(t *testing.T) {
-	response := `user.value 4679836
+const (
+	responseCPU = `user.value 4679836
 nice.value 59278
 system.value 1979168
 idle.value 59957502
@@ -57,43 +57,80 @@ steal.value 0
 guest.value 0
 .
 `
-	n := dummyNode(response)
-
-	event, err := n.Fetch("cpu", "swap")
-
-	assert.Nil(t, err)
-
-	expected := common.MapStr{
-		"cpu": common.MapStr{
-			"user":    float64(4679836),
-			"nice":    float64(59278),
-			"system":  float64(1979168),
-			"idle":    float64(59957502),
-			"iowait":  float64(705373),
-			"irq":     float64(76),
-			"softirq": float64(36404),
-			"steal":   float64(0),
-			"guest":   float64(0),
-		},
-	}
-	assert.Equal(t, expected, event)
-}
-
-func TestFetchUnknown(t *testing.T) {
-	response := `some.value U
+	responseUnknown = `some.value U
 other.value 42
 .
 `
-	n := dummyNode(response)
-
-	event, err := n.Fetch("test")
-
-	assert.NotNil(t, err)
+	responseWithWrongFields = `user.value 4679836
+nice.value 59278
+system.value 1979168
+idle.value 59957502
+user.1000.value 23456
+user.0.value 38284
+.
+`
+)
 
-	expected := common.MapStr{
-		"test": common.MapStr{
-			"other": float64(42),
+func TestFetch(t *testing.T) {
+	cases := []struct {
+		title    string
+		response string
+		expected common.MapStr
+		errors   bool
+	}{
+		{
+			"normal case",
+			responseCPU,
+			common.MapStr{
+				"cpu": common.MapStr{
+					"user":    float64(4679836),
+					"nice":    float64(59278),
+					"system":  float64(1979168),
+					"idle":    float64(59957502),
+					"iowait":  float64(705373),
+					"irq":     float64(76),
+					"softirq": float64(36404),
+					"steal":   float64(0),
+					"guest":   float64(0),
+				},
+			},
+			false,
+		},
+		{
+			"unknown values",
+			responseUnknown,
+			common.MapStr{
+				"cpu": common.MapStr{
+					"other": float64(42),
+				},
+			},
+			true,
+		},
+		{
+			"wrong field names",
+			responseWithWrongFields,
+			common.MapStr{
+				"cpu": common.MapStr{
+					"user":   float64(4679836),
+					"nice":   float64(59278),
+					"system": float64(1979168),
+					"idle":   float64(59957502),
+				},
+			},
+			true,
 		},
 	}
-	assert.Equal(t, expected, event)
+
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			n := dummyNode(c.response)
+			event, err := n.Fetch("cpu", "swap")
+			assert.Equal(t, c.expected, event)
+			if c.errors {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
 }

metricbeat/module/munin/node/config.go

@@ -0,0 +1,27 @@
+// 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 node
+
+type Config struct {
+	ServiceName string `config:"service.name"`
+	ServiceType string `config:"service.type"`
+}
+
+var defaultConfig = Config{
+	ServiceType: "munin",
+}

metricbeat/module/munin/node/node.go

@@ -42,53 +42,63 @@ func init() {
 // interface methods except for Fetch.
 type MetricSet struct {
 	mb.BaseMetricSet
-	namespace string
-	timeout   time.Duration
+	serviceName string
+	serviceType string
+	timeout     time.Duration
 }
 
 // New creates a new instance of the MetricSet. New is responsible for unpacking
 // any MetricSet specific configuration options if there are any.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	cfgwarn.Beta("The munin node metricset is beta.")
 
-	config := struct {
-		Namespace string `config:"node.namespace" validate:"required"`
-	}{}
+	config := defaultConfig
 	if err := base.Module().UnpackConfig(&config); err != nil {
 		return nil, err
 	}
 
 	return &MetricSet{
 		BaseMetricSet: base,
-		namespace:     config.Namespace,
+		serviceName:   config.ServiceName,
+		serviceType:   config.ServiceType,
 		timeout:       base.Module().Config().Timeout,
 	}, nil
 }
 
 // Fetch method implements the data gathering
-func (m *MetricSet) Fetch() (common.MapStr, error) {
+func (m *MetricSet) Fetch(r mb.ReporterV2) {
 	node, err := munin.Connect(m.Host(), m.timeout)
 	if err != nil {
-		return nil, err
+		r.Error(err)
+		return
 	}
 	defer node.Close()
 
 	items, err := node.List()
 	if err != nil {
-		return nil, err
+		r.Error(err)
+		return
 	}
 
-	event, err := node.Fetch(items...)
+	metrics, err := node.Fetch(items...)
 	if err != nil {
-		return nil, err
+		r.Error(err)
 	}
 
-	// Set dynamic namespace.
-	_, err = event.Put(mb.NamespaceKey, m.namespace)
-	if err != nil {
-		return nil, err
+	// Even if there was some error, keep sending succesfully collected metrics if any
+	if len(metrics) == 0 {
+		return
 	}
-
-	return event, nil
-
+	event := mb.Event{
+		Service: m.serviceType,
+		RootFields: common.MapStr{
+			"munin": common.MapStr{
+				"metrics": metrics,
+			},
+		},
+	}
+	if m.serviceName != "" {
+		event.RootFields.Put("service.name", m.serviceName)
+	}
+	r.Event(event)
 }

metricbeat/tests/system/test_munin.py

@@ -18,7 +18,7 @@ def test_munin_node(self):
             "hosts": self.get_hosts(),
             "period": "1s",
             "extras": {
-                "node.namespace": namespace,
+                "service.name": namespace,
             },
         }])
         proc = self.start_beat()
@@ -31,8 +31,12 @@ def test_munin_node(self):
         evt = output[0]
         print(evt)
 
-        assert evt["munin"][namespace]["cpu"]["user"] > 0
+        assert evt["service"]["type"] == "munin"
+        assert evt["service"]["name"] == namespace
+        assert evt["munin"]["metrics"]["cpu"]["user"] > 0
+
+        self.assert_fields_are_documented(evt)
 
     def get_hosts(self):
-        return [os.getenv('MUNIN_HOST', 'localhost') + ':' +
+        return [self.compose_hosts()[0] + ':' +
                 os.getenv('MUNIN_PORT', '4949')]