diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 029824c74ae..e0827e0f256 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -81,10 +81,11 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Refactor Prometheus metric mappings {pull}9948[9948]
 - Removed Prometheus stats metricset in favor of just using Prometheus collector {pull}9948[9948]
 - Adjust Redis.info metricset fields to ECS. {pull}10319[10319]
-- Change type of field docker.container.ip_addresses to `ip` instead of `keyword. {pull}10364[10364]
+- Change type of field docker.container.ip_addresses to `ip` instead of `keyword`. {pull}10364[10364]
 - Rename http.request.body field to http.request.body.content. {pull}10315[10315]
 - Adjust php_fpm.process metricset fields to ECS. {pull}10366[10366]
 - Adjust mongodb.status metricset to to ECS. {pull}10368[10368]
+- Refactor munin module to collect an event per plugin and to have more strict field mappings. `namespace` option has been removed, and will be replaced by `service.name`. {pull}10322[10322]
 - Change the following fields from type text to keyword: {pull}10318[10318]
   - ceph.osd_df.name
   - ceph.osd_tree.name
diff --git a/metricbeat/docs/fields.asciidoc b/metricbeat/docs/fields.asciidoc
index 2172b0ab549..b8a006620c6 100644
--- a/metricbeat/docs/fields.asciidoc
+++ b/metricbeat/docs/fields.asciidoc
@@ -17339,11 +17339,25 @@ Munin node metrics exporter
 
 
 
-[float]
-== munin fields
+*`munin.metrics.*`*::
++
+--
+type: object
+
+Metrics exposed by a plugin of a munin node agent.
+
+
+--
 
-munin contains metrics exposed by a munin node agent
+*`munin.plugin.name`*::
++
+--
+type: keyword
 
+Name of the plugin collecting these metrics.
+
+
+--
 
 
 [[exported-fields-mysql]]
diff --git a/metricbeat/docs/modules/munin.asciidoc b/metricbeat/docs/modules/munin.asciidoc
index a30361167bb..c0b5bb8460a 100644
--- a/metricbeat/docs/modules/munin.asciidoc
+++ b/metricbeat/docs/modules/munin.asciidoc
@@ -33,7 +33,15 @@ metricbeat.modules:
   enabled: true
   period: 10s
   hosts: ["localhost:4949"]
-  node.namespace: node
+
+  # List of plugins to collect metrics from, by default it collects from
+  # all the available ones.
+  #munin.plugins: []
+
+  # If set to true, it sanitizes fields names in concordance with munin
+  # implementation (all characters that are not alphanumeric, or underscore
+  # are replaced by underscores).
+  #munin.sanitize: false
 ----
 
 [float]
diff --git a/metricbeat/metricbeat.reference.yml b/metricbeat/metricbeat.reference.yml
index 94314930f7f..4850233d174 100644
--- a/metricbeat/metricbeat.reference.yml
+++ b/metricbeat/metricbeat.reference.yml
@@ -507,7 +507,15 @@ metricbeat.modules:
   enabled: true
   period: 10s
   hosts: ["localhost:4949"]
-  node.namespace: node
+
+  # List of plugins to collect metrics from, by default it collects from
+  # all the available ones.
+  #munin.plugins: []
+
+  # If set to true, it sanitizes fields names in concordance with munin
+  # implementation (all characters that are not alphanumeric, or underscore
+  # are replaced by underscores).
+  #munin.sanitize: false
 
 #-------------------------------- MySQL Module -------------------------------
 - module: mysql
diff --git a/metricbeat/module/munin/_meta/config.reference.yml b/metricbeat/module/munin/_meta/config.reference.yml
index cdef11f3670..3fe0cb3271a 100644
--- a/metricbeat/module/munin/_meta/config.reference.yml
+++ b/metricbeat/module/munin/_meta/config.reference.yml
@@ -3,4 +3,12 @@
   enabled: true
   period: 10s
   hosts: ["localhost:4949"]
-  node.namespace: node
+
+  # List of plugins to collect metrics from, by default it collects from
+  # all the available ones.
+  #munin.plugins: []
+
+  # If set to true, it sanitizes fields names in concordance with munin
+  # implementation (all characters that are not alphanumeric, or underscore
+  # are replaced by underscores).
+  #munin.sanitize: false
diff --git a/metricbeat/module/munin/_meta/config.yml b/metricbeat/module/munin/_meta/config.yml
index c8267c385dd..e14277e55d2 100644
--- a/metricbeat/module/munin/_meta/config.yml
+++ b/metricbeat/module/munin/_meta/config.yml
@@ -3,4 +3,3 @@
   #  - node
   period: 10s
   hosts: ["localhost:4949"]
-  node.namespace: node
diff --git a/metricbeat/module/munin/_meta/fields.yml b/metricbeat/module/munin/_meta/fields.yml
index 83e41a59fa9..d36f9fd50bd 100644
--- a/metricbeat/module/munin/_meta/fields.yml
+++ b/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: >
+        Metrics exposed by a plugin of a munin node agent.
+    - name: munin.plugin.name
+      type: keyword
+      description: >
+        Name of the plugin collecting these metrics.
     - name: munin
       type: group
-      description: >
-          munin contains metrics exposed by a munin node agent
       fields:
diff --git a/metricbeat/module/munin/fields.go b/metricbeat/module/munin/fields.go
index ef7d4e1121e..bb04054e8b3 100644
--- a/metricbeat/module/munin/fields.go
+++ b/metricbeat/module/munin/fields.go
@@ -32,5 +32,5 @@ func init() {
 // AssetMunin returns asset data.
 // This is the base64 encoded gzipped contents of ../metricbeat/module/munin.
 func AssetMunin() string {
-	return "eJxsjk2qhDAQhPc5ReHeC2TxbvAOEU2NNKOdkLQw3n4wBmaEqV3/FN834snDY9tV1AEmttJj+D/nwQGRdS6STZJ6/DkAaDdoisRGKzJX8JVTMRYHFK4MlR4TLTjgIVxj9a05QsPGD+2MHZkeS0l77psfyCuthjmpBdF6g1dGTAdC/2lyYaFar39bXCZ3z3cAAAD//xiKUG4="
+	return "eJx8kU1qwzAUhPc+xZBNIBAfQIveoL1CkK2Jq0Z/SM+0vn2xpZaYmi7fzBvm09MVDy4Kfg42dIBYcVQ4va7zqQMMy5htEhuDwksHAJuHEA3hKdmOBfxKMQtzB2Q66kKFgaI74G7pTFFb8oqgPVtb38L9ZfMAWRIV4vDBUZpUh1t1TJwHx7/OzeuUbJja2vlybjsH7Bv/E3ShwbBAI7l5sgHxDl3x6gP1xCD9AXzd71dph//g8hmz+Z/gTXuuVfLOn+IxOsdRbJhWtfye9qB71zflOKemPJ+6pvaf8R0AAP//7IiXDQ=="
 }
diff --git a/metricbeat/module/munin/munin.go b/metricbeat/module/munin/munin.go
index d81a5ab0281..7c2fd40c0d8 100644
--- a/metricbeat/module/munin/munin.go
+++ b/metricbeat/module/munin/munin.go
@@ -19,23 +19,29 @@ package munin
 
 import (
 	"bufio"
-	"fmt"
 	"io"
 	"net"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
 
-	"github.com/joeshaw/multierror"
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
 )
 
 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
@@ -66,7 +72,7 @@ func (n *Node) Close() error {
 	return n.conn.Close()
 }
 
-// List of items exposed by the node
+// List of plugins exposed by the node
 func (n *Node) List() ([]string, error) {
 	_, err := io.WriteString(n.writer, "list\n")
 	if err != nil {
@@ -79,54 +85,64 @@ func (n *Node) List() ([]string, error) {
 }
 
 // Fetch metrics from munin node
-func (n *Node) Fetch(items ...string) (common.MapStr, error) {
-	var errs multierror.Errors
+func (n *Node) Fetch(plugin string, sanitize bool) (common.MapStr, error) {
+	_, err := io.WriteString(n.writer, "fetch "+plugin+"\n")
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to fetch metrics for plugin '%s'", plugin)
+	}
+
 	event := common.MapStr{}
+	scanner := bufio.NewScanner(n.reader)
+	scanner.Split(bufio.ScanWords)
+	for scanner.Scan() {
+		name := strings.TrimSpace(scanner.Text())
 
-	for _, item := range items {
-		_, err := io.WriteString(n.writer, "fetch "+item+"\n")
-		if err != nil {
-			errs = append(errs, err)
-			continue
+		// Munin delimits metrics with a dot
+		if name == "." {
+			break
 		}
 
-		scanner := bufio.NewScanner(n.reader)
-		scanner.Split(bufio.ScanWords)
-		for scanner.Scan() {
-			name := strings.TrimSpace(scanner.Text())
-
-			// Munin delimits metrics with a dot
-			if name == "." {
-				break
-			}
-
-			name = strings.TrimSuffix(name, ".value")
-
-			if !scanner.Scan() {
-				if scanner.Err() == nil {
-					errs = append(errs, errors.New("unexpected EOF when expecting value"))
-				}
-				break
+		name = strings.TrimSuffix(name, ".value")
+		if !scanner.Scan() {
+			if scanner.Err() == nil {
+				return nil, errors.New("unexpected EOF when expecting value")
 			}
-			value := scanner.Text()
+		}
+		value := scanner.Text()
 
-			key := fmt.Sprintf("%s.%s", item, name)
+		if strings.Contains(name, ".") {
+			logp.Debug("munin", "ignoring field name with dot '%s'", name)
+			continue
+		}
 
-			if value == unknownValue {
-				errs = append(errs, errors.Errorf("unknown value for %s", key))
-				continue
-			}
-			if f, err := strconv.ParseFloat(value, 64); err == nil {
-				event.Put(key, f)
-				continue
-			}
-			event.Put(key, value)
+		if value == unknownValue {
+			logp.Debug("munin", "unknown value for '%s'", name)
+			continue
 		}
 
-		if scanner.Err() != nil {
-			errs = append(errs, scanner.Err())
+		if sanitize && !nameRegexp.MatchString(name) {
+			logp.Debug("munin", "sanitizing name with invalid characters '%s'", name)
+			name = sanitizeName(name)
+		}
+		if f, err := strconv.ParseFloat(value, 64); err == nil {
+			event[name] = f
+			continue
 		}
 	}
 
-	return event, errs.Err()
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+
+	return event, nil
+}
+
+var (
+	invalidCharactersRegexp = regexp.MustCompile("(^[^a-zA-Z_]|[^a-zA-Z_0-9])")
+)
+
+// Mimic munin master implementation
+// https://github.com/munin-monitoring/munin/blob/20abb861/lib/Munin/Master/Node.pm#L385
+func sanitizeName(name string) string {
+	return invalidCharactersRegexp.ReplaceAllString(name, "_")
 }
diff --git a/metricbeat/module/munin/munin_test.go b/metricbeat/module/munin/munin_test.go
index 8312858652a..1da93165ddb 100644
--- a/metricbeat/module/munin/munin_test.go
+++ b/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,100 @@ 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)
+	responseWithWrongFields = `user.value 4679836
+nice.value 59278
+system.value 1979168
+idle.value 59957502
+user.1000.value 23456
+user.0.value 38284
+.
+`
+)
 
-	event, err := n.Fetch("test")
+func TestFetch(t *testing.T) {
+	cases := []struct {
+		title    string
+		response string
+		expected common.MapStr
+	}{
+		{
+			"normal case",
+			responseCPU,
+			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),
+			},
+		},
+		{
+			"unknown values",
+			responseUnknown,
+			common.MapStr{
+				"other": float64(42),
+			},
+		},
+		{
+			"wrong field names",
+			responseWithWrongFields,
+			common.MapStr{
+				"user":   float64(4679836),
+				"nice":   float64(59278),
+				"system": float64(1979168),
+				"idle":   float64(59957502),
+			},
+		},
+	}
 
-	assert.NotNil(t, err)
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			n := dummyNode(c.response)
+			event, err := n.Fetch("cpu", true)
+			assert.Equal(t, c.expected, event)
+			assert.NoError(t, err)
+		})
+	}
+}
 
-	expected := common.MapStr{
-		"test": common.MapStr{
-			"other": float64(42),
+func TestSanitizeName(t *testing.T) {
+	cases := []struct {
+		name     string
+		expected string
+	}{
+		{
+			"if_eth0",
+			"if_eth0",
+		},
+		{
+			"/dev/sda1",
+			"_dev_sda1",
+		},
+		{
+			"eth0:100",
+			"eth0_100",
+		},
+		{
+			"user@host",
+			"user_host",
 		},
+		{
+			"404",
+			"_04",
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.expected, sanitizeName(c.name))
+		})
 	}
-	assert.Equal(t, expected, event)
 }
diff --git a/metricbeat/module/munin/node/_meta/docs.asciidoc b/metricbeat/module/munin/node/_meta/docs.asciidoc
index 403f5fa93b9..3a5ead627e1 100644
--- a/metricbeat/module/munin/node/_meta/docs.asciidoc
+++ b/metricbeat/module/munin/node/_meta/docs.asciidoc
@@ -11,31 +11,40 @@ and sends them as events to Elastic.
 - module: munin
   metricsets: ["node"]
   hosts: ["localhost:4949"]
-  node.namespace: node
+  munin.plugins: ["cpu", "swap"]
 ---
 
-All metrics exposed by a single munin node will be sent in a single event,
-grouped by munin items, e.g:
+Metrics exposed by a single munin node will be sent in an event per plugin.
+
+For example with the previous configuration two events are sent like the
+following ones.
 
 [source,json]
 ---
 "munin": {
-  "node": {
-    "swap": {
-      "swap_in": 198609,
-      "swap_out": 612629
-    },
-    "cpu": {
-      "softirq": 680,
-      "guest": 0,
-      "user": 158212,
-      "iowait": 71095,
-      "irq": 1,
-      "system": 35906,
-      "idle": 1185709,
-      "steal": 0,
-      "nice": 1633
-    }
+  "plugin": {
+    "name": "swap"
+  },
+  "metrics": {
+    "swap_in": 198609,
+    "swap_out": 612629
+  }
+}
+
+"munin": {
+  "plugin": {
+    "name": "cpu"
+  }
+  "metrics": {
+    "softirq": 680,
+    "guest": 0,
+    "user": 158212,
+    "iowait": 71095,
+    "irq": 1,
+    "system": 35906,
+    "idle": 1185709,
+    "steal": 0,
+    "nice": 1633
   }
 }
 ---
diff --git a/metricbeat/module/munin/node/config.go b/metricbeat/module/munin/node/config.go
new file mode 100644
index 00000000000..9c649de9041
--- /dev/null
+++ b/metricbeat/module/munin/node/config.go
@@ -0,0 +1,28 @@
+// 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
+
+// Config is the configuration for munin
+type Config struct {
+	Plugins  []string `config:"munin.plugins"`
+	Sanitize bool     `config:"munin.sanitize"`
+}
+
+var defaultConfig = Config{
+	Sanitize: false,
+}
diff --git a/metricbeat/module/munin/node/node.go b/metricbeat/module/munin/node/node.go
index 9fe283db0f0..604d7553fb0 100644
--- a/metricbeat/module/munin/node/node.go
+++ b/metricbeat/module/munin/node/node.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/module/munin"
 )
@@ -42,8 +43,10 @@ func init() {
 // interface methods except for Fetch.
 type MetricSet struct {
 	mb.BaseMetricSet
-	namespace string
-	timeout   time.Duration
+	serviceType string
+	plugins     []string
+	sanitize    bool
+	timeout     time.Duration
 }
 
 // New creates a new instance of the MetricSet. New is responsible for unpacking
@@ -51,44 +54,61 @@ type MetricSet struct {
 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,
+		plugins:       config.Plugins,
+		sanitize:      config.Sanitize,
 		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
+	plugins := m.plugins
+	if len(plugins) == 0 {
+		plugins, err = node.List()
+		if err != nil {
+			r.Error(err)
+			return
+		}
 	}
 
-	event, err := node.Fetch(items...)
-	if err != nil {
-		return nil, err
-	}
+	for _, plugin := range plugins {
+		metrics, err := node.Fetch(plugin, m.sanitize)
+		if err != nil {
+			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 {
+			continue
+		}
+		event := mb.Event{
+			Service: plugin,
+			RootFields: common.MapStr{
+				"munin": common.MapStr{
+					"plugin": common.MapStr{
+						"name": plugin,
+					},
+					"metrics": metrics,
+				},
+			},
+		}
+		if !r.Event(event) {
+			logp.Debug("munin", "Failed to report event, interrupting Fetch")
+			return
+		}
 	}
-
-	return event, nil
-
 }
diff --git a/metricbeat/modules.d/munin.yml.disabled b/metricbeat/modules.d/munin.yml.disabled
index a86e6df767f..d42b1d9919e 100644
--- a/metricbeat/modules.d/munin.yml.disabled
+++ b/metricbeat/modules.d/munin.yml.disabled
@@ -6,4 +6,3 @@
   #  - node
   period: 10s
   hosts: ["localhost:4949"]
-  node.namespace: node
diff --git a/metricbeat/tests/system/test_munin.py b/metricbeat/tests/system/test_munin.py
index e1618202fa4..63cc80a94ff 100644
--- a/metricbeat/tests/system/test_munin.py
+++ b/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,
+                "munin.plugins": ["cpu"],
             },
         }])
         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"] == "cpu"
+        assert evt["munin"]["plugin"]["name"] == "cpu"
+        assert evt["munin"]["metrics"]["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')]
diff --git a/x-pack/metricbeat/metricbeat.reference.yml b/x-pack/metricbeat/metricbeat.reference.yml
index 9b9675b01fc..61534846e2f 100644
--- a/x-pack/metricbeat/metricbeat.reference.yml
+++ b/x-pack/metricbeat/metricbeat.reference.yml
@@ -525,7 +525,15 @@ metricbeat.modules:
   enabled: true
   period: 10s
   hosts: ["localhost:4949"]
-  node.namespace: node
+
+  # List of plugins to collect metrics from, by default it collects from
+  # all the available ones.
+  #munin.plugins: []
+
+  # If set to true, it sanitizes fields names in concordance with munin
+  # implementation (all characters that are not alphanumeric, or underscore
+  # are replaced by underscores).
+  #munin.sanitize: false
 
 #-------------------------------- MySQL Module --------------------------------
 - module: mysql