Migrate docker autodiscovery to ECS (elastic#10898)

Fields injected by docker autodiscover provider were being placed
in alias fields introduced for ECS, change them to the new location
and add selectors accordingly. 

This PR includes elastic#10862 and elastic#10758

As a summary:
    * Autodiscover selectors using ECS structure are added to
      autodiscover events, old selectors are kept for backwards compatibility
    * Autodiscover generated metadata follows ECS
    * Dedotting of labels is added, enabled by default, will be backported for 6.7,
      but disabled

`docker.containers.labels` is not migrated, as it wasn't for `add_docker_metadata`
(see elastic#9412)

Fixes elastic#10757

Co-Authored-By: kaiyan-sheng <kaiyan.sheng@elastic.co>
Co-Authored-By: Nicolas Ruflin <spam@ruflin.com>

CHANGELOG.next.asciidoc

@@ -168,6 +168,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fixed data types for various hosts fields in `mongodb/replstatus` metricset {pull}10307[10307]
 - Added function to close sql database connection. {pull}10355[10355]
 - Fix issue with `elasticsearch/node_stats` metricset (x-pack) not indexing `source_node` field. {pull}10639[10639]
+- Migrate docker autodiscover to ECS. {issue}10757[10757] {pull}10862[10862]
 
 *Packetbeat*
 

filebeat/tests/system/test_autodiscover.py

@@ -50,6 +50,8 @@ def test_docker(self):
 
         # Check metadata is added
         assert output[0]['message'] == 'Busybox output 1'
-        assert output[0]['docker']['container']['image'] == 'busybox'
+        assert output[0]['container']['image']['name'] == 'busybox'
         assert output[0]['docker']['container']['labels'] == {}
-        assert 'name' in output[0]['docker']['container']
+        assert 'name' in output[0]['container']
+
+        self.assert_fields_are_documented(output[0])

heartbeat/tests/system/test_autodiscovery.py

@@ -60,7 +60,9 @@ def test_docker(self):
                     # We don't check all the docker fields because this is really the responsibility
                     # of libbeat's autodiscovery code.
                     event = output[0]
-                    if event['monitor']['id'] == 'myid' and event['docker']['container']['id'] is not None:
+                    if event['monitor']['id'] == 'myid' and event['container']['id'] is not None:
                         matched = True
 
         assert matched
+
+        self.assert_fields_are_documented(output[0])

libbeat/autodiscover/providers/docker/config.go

@@ -32,12 +32,14 @@ type Config struct {
 	Builders     []*common.Config        `config:"builders"`
 	Appenders    []*common.Config        `config:"appenders"`
 	Templates    template.MapperSettings `config:"templates"`
+	Dedot        bool                    `config:"labels.dedot"`
 }
 
 func defaultConfig() *Config {
 	return &Config{
 		Host:   "unix:///var/run/docker.sock",
 		Prefix: "co.elastic",
+		Dedot:  true,
 	}
 }
 

libbeat/autodiscover/providers/docker/docker.go

@@ -18,6 +18,8 @@
 package docker
 
 import (
+	"errors"
+
 	"github.com/gofrs/uuid"
 
 	"github.com/elastic/beats/libbeat/autodiscover"
@@ -119,41 +121,91 @@ func (d *Provider) Start() {
 	}()
 }
 
-func (d *Provider) emitContainer(event bus.Event, flag string) {
+type dockerMetadata struct {
+	// Old selectors [Deprecated]
+	Docker common.MapStr
+
+	// New ECS-based selectors
+	Container common.MapStr
+
+	// Metadata used to enrich events, like ECS-based selectors but can
+	// have modifications like dedotting
+	Metadata common.MapStr
+}
+
+func (d *Provider) generateMetaDocker(event bus.Event) (*docker.Container, *dockerMetadata) {
 	container, ok := event["container"].(*docker.Container)
 	if !ok {
-		logp.Err("Couldn't get a container from watcher event")
-		return
+		logp.Error(errors.New("Couldn't get a container from watcher event"))
+		return nil, nil
 	}
 
-	var host string
-	if len(container.IPAddresses) > 0 {
-		host = container.IPAddresses[0]
-	}
+	// Don't dedot selectors, dedot only metadata used for events enrichment
 	labelMap := common.MapStr{}
+	metaLabelMap := common.MapStr{}
 	for k, v := range container.Labels {
 		safemapstr.Put(labelMap, k, v)
+		if d.config.Dedot {
+			label := common.DeDot(k)
+			metaLabelMap.Put(label, v)
+		} else {
+			safemapstr.Put(metaLabelMap, k, v)
+		}
 	}
 
-	meta := common.MapStr{
-		"container": common.MapStr{
-			"id":     container.ID,
-			"name":   container.Name,
-			"image":  container.Image,
+	meta := &dockerMetadata{
+		Docker: common.MapStr{
+			"container": common.MapStr{
+				"id":     container.ID,
+				"name":   container.Name,
+				"image":  container.Image,
+				"labels": labelMap,
+			},
+		},
+		Container: common.MapStr{
+			"id":   container.ID,
+			"name": container.Name,
+			"image": common.MapStr{
+				"name": container.Image,
+			},
 			"labels": labelMap,
 		},
+		Metadata: common.MapStr{
+			"container": common.MapStr{
+				"id":   container.ID,
+				"name": container.Name,
+				"image": common.MapStr{
+					"name": container.Image,
+				},
+			},
+			"docker": common.MapStr{
+				"container": common.MapStr{
+					"labels": metaLabelMap,
+				},
+			},
+		},
+	}
+
+	return container, meta
+}
+
+func (d *Provider) emitContainer(event bus.Event, flag string) {
+	container, meta := d.generateMetaDocker(event)
+	var host string
+	if len(container.IPAddresses) > 0 {
+		host = container.IPAddresses[0]
 	}
+
 	// Without this check there would be overlapping configurations with and without ports.
 	if len(container.Ports) == 0 {
 		event := bus.Event{
-			"provider": d.uuid,
-			"id":       container.ID,
-			flag:       true,
-			"host":     host,
-			"docker":   meta,
-			"meta": common.MapStr{
-				"docker": meta,
-			},
+			"provider":  d.uuid,
+			"id":        container.ID,
+			flag:        true,
+			"host":      host,
+			"docker":    meta.Docker,
+			"container": meta.Container,
+			"meta":      meta.Metadata,
 		}
 
 		d.publish(event)
@@ -162,15 +214,14 @@ func (d *Provider) emitContainer(event bus.Event, flag string) {
 	// Emit container container and port information
 	for _, port := range container.Ports {
 		event := bus.Event{
-			"provider": d.uuid,
-			"id":       container.ID,
-			flag:       true,
-			"host":     host,
-			"port":     port.PrivatePort,
-			"docker":   meta,
-			"meta": common.MapStr{
-				"docker": meta,
-			},
+			"provider":  d.uuid,
+			"id":        container.ID,
+			flag:        true,
+			"host":      host,
+			"port":      port.PrivatePort,
+			"docker":    meta.Docker,
+			"container": meta.Container,
+			"meta":      meta.Metadata,
 		}
 
 		d.publish(event)

libbeat/autodiscover/providers/docker/docker_integration_test.go

@@ -92,17 +92,23 @@ func checkEvent(t *testing.T, listener bus.Listener, start bool) {
 				assert.Equal(t, getValue(e, "stop"), true)
 				assert.Nil(t, getValue(e, "start"))
 			}
-			assert.Equal(t, getValue(e, "docker.container.image"), "busybox")
-			assert.Equal(t, getValue(e, "docker.container.labels"), common.MapStr{
-				"label": common.MapStr{
-					"value": "foo",
-					"child": "bar",
+			assert.Equal(t, getValue(e, "container.image.name"), "busybox")
+			// labels.dedot=true by default
+			assert.Equal(t,
+				common.MapStr{
+					"label": common.MapStr{
+						"value": "foo",
+						"child": "bar",
+					},
 				},
-			})
-			assert.NotNil(t, getValue(e, "docker.container.id"))
-			assert.NotNil(t, getValue(e, "docker.container.name"))
+				getValue(e, "container.labels"),
+			)
+			assert.NotNil(t, getValue(e, "container.id"))
+			assert.NotNil(t, getValue(e, "container.name"))
 			assert.NotNil(t, getValue(e, "host"))
-			assert.Equal(t, getValue(e, "docker"), getValue(e, "meta.docker"))
+			assert.Equal(t, getValue(e, "docker.container.id"), getValue(e, "meta.container.id"))
+			assert.Equal(t, getValue(e, "docker.container.name"), getValue(e, "meta.container.name"))
+			assert.Equal(t, getValue(e, "docker.container.image"), getValue(e, "meta.container.image.name"))
 			return
 
 		case <-time.After(10 * time.Second):

libbeat/autodiscover/providers/docker/docker_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/bus"
+	"github.com/elastic/beats/libbeat/common/docker"
 )
 
 func TestGenerateHints(t *testing.T) {
@@ -105,3 +106,131 @@ func getNestedAnnotations(in common.MapStr) common.MapStr {
 	}
 	return out
 }
+
+func TestGenerateMetaDockerNoDedot(t *testing.T) {
+	event := bus.Event{
+		"container": &docker.Container{
+			ID:   "abc",
+			Name: "foobar",
+			Labels: map[string]string{
+				"do.not.include":          "true",
+				"co.elastic.logs/disable": "true",
+			},
+		},
+	}
+
+	cfg := defaultConfig()
+	cfg.Dedot = false
+	p := Provider{
+		config: cfg,
+	}
+	_, meta := p.generateMetaDocker(event)
+	expectedMeta := &dockerMetadata{
+		Docker: common.MapStr{
+			"container": common.MapStr{
+				"id":    "abc",
+				"name":  "foobar",
+				"image": "",
+				"labels": common.MapStr{
+					"do": common.MapStr{"not": common.MapStr{"include": "true"}},
+					"co": common.MapStr{"elastic": common.MapStr{"logs/disable": "true"}},
+				},
+			},
+		},
+		Container: common.MapStr{
+			"id":   "abc",
+			"name": "foobar",
+			"image": common.MapStr{
+				"name": "",
+			},
+			"labels": common.MapStr{
+				"do": common.MapStr{"not": common.MapStr{"include": "true"}},
+				"co": common.MapStr{"elastic": common.MapStr{"logs/disable": "true"}},
+			},
+		},
+		Metadata: common.MapStr{
+			"container": common.MapStr{
+				"id":   "abc",
+				"name": "foobar",
+				"image": common.MapStr{
+					"name": "",
+				},
+			},
+			"docker": common.MapStr{
+				"container": common.MapStr{
+					"labels": common.MapStr{
+						"do": common.MapStr{"not": common.MapStr{"include": "true"}},
+						"co": common.MapStr{"elastic": common.MapStr{"logs/disable": "true"}},
+					},
+				},
+			},
+		},
+	}
+	assert.Equal(t, expectedMeta.Docker, meta.Docker)
+	assert.Equal(t, expectedMeta.Container, meta.Container)
+	assert.Equal(t, expectedMeta.Metadata, meta.Metadata)
+}
+
+func TestGenerateMetaDockerWithDedot(t *testing.T) {
+	event := bus.Event{
+		"container": &docker.Container{
+			ID:   "abc",
+			Name: "foobar",
+			Labels: map[string]string{
+				"do.not.include":          "true",
+				"co.elastic.logs/disable": "true",
+			},
+		},
+	}
+
+	cfg := defaultConfig()
+	cfg.Dedot = true
+	p := Provider{
+		config: cfg,
+	}
+	_, meta := p.generateMetaDocker(event)
+	expectedMeta := &dockerMetadata{
+		Docker: common.MapStr{
+			"container": common.MapStr{
+				"id":    "abc",
+				"name":  "foobar",
+				"image": "",
+				"labels": common.MapStr{
+					"do": common.MapStr{"not": common.MapStr{"include": "true"}},
+					"co": common.MapStr{"elastic": common.MapStr{"logs/disable": "true"}},
+				},
+			},
+		},
+		Container: common.MapStr{
+			"id":   "abc",
+			"name": "foobar",
+			"image": common.MapStr{
+				"name": "",
+			},
+			"labels": common.MapStr{
+				"do": common.MapStr{"not": common.MapStr{"include": "true"}},
+				"co": common.MapStr{"elastic": common.MapStr{"logs/disable": "true"}},
+			},
+		},
+		Metadata: common.MapStr{
+			"container": common.MapStr{
+				"id":   "abc",
+				"name": "foobar",
+				"image": common.MapStr{
+					"name": "",
+				},
+			},
+			"docker": common.MapStr{
+				"container": common.MapStr{
+					"labels": common.MapStr{
+						"do_not_include":          "true",
+						"co_elastic_logs/disable": "true",
+					},
+				},
+			},
+		},
+	}
+	assert.Equal(t, expectedMeta.Docker, meta.Docker)
+	assert.Equal(t, expectedMeta.Container, meta.Container)
+	assert.Equal(t, expectedMeta.Metadata, meta.Metadata)
+}

metricbeat/tests/system/test_autodiscover.py

@@ -51,9 +51,10 @@ def test_docker(self):
         proc.check_kill_and_wait()
 
         # Check metadata is added
-        assert output[0]['docker']['container']['image'] == 'memcached:latest'
+        assert output[0]['container']['image']['name'] == 'memcached:latest'
         assert output[0]['docker']['container']['labels'] == {}
-        assert 'name' in output[0]['docker']['container']
+        assert 'name' in output[0]['container']
+        self.assert_fields_are_documented(output[0])
 
     @unittest.skipIf(not INTEGRATION_TESTS or
                      os.getenv("TESTING_ENVIRONMENT") == "2x",
@@ -93,8 +94,9 @@ def test_docker_labels(self):
         proc.check_kill_and_wait()
 
         # Check metadata is added
-        assert output[0]['docker']['container']['image'] == 'memcached:latest'
-        assert 'name' in output[0]['docker']['container']
+        assert output[0]['container']['image']['name'] == 'memcached:latest'
+        assert 'name' in output[0]['container']
+        self.assert_fields_are_documented(output[0])
 
     @unittest.skipIf(not INTEGRATION_TESTS or
                      os.getenv("TESTING_ENVIRONMENT") == "2x",
@@ -143,3 +145,4 @@ def test_config_appender(self):
 
         # Check field is added
         assert output[0]['fields']['foo'] == 'bar'
+        self.assert_fields_are_documented(output[0])