elastic · jsoriano · Feb 25, 2019 · Feb 25, 2019 · Feb 25, 2019 · Feb 25, 2019
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -92,6 +92,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Update rabbitmq.* fields to map to ECS. {pull}10563[10563]
 - Update haproxy.* fields to map to ECS. {pull}10558[10558] {pull}10568[10568]
 - Collect all EC2 meta data from all instances in all states. {pull}10628[10628]
+- Migrate docker module to ECS. {pull}10927[10927]
 
 *Packetbeat*
 

diff --git a/metricbeat/module/docker/container/_meta/data.json b/metricbeat/module/docker/container/_meta/data.json
@@ -1,37 +1,59 @@
 {
     "@timestamp": "2017-10-12T08:05:34.853Z",
-    "beat": {
+    "agent": {
         "hostname": "host.example.com",
         "name": "host.example.com"
     },
+    "container": {
+        "id": "cc78e58acfda4501105dc4de8e3ae218f2da616213e6e3af168c40103829302a",
+        "image": {
+            "name": "metricbeat_elasticsearch"
+        },
+        "name": "metricbeat_elasticsearch_1_df866b3a7b3d",
+        "runtime": "docker"
+    },
     "docker": {
         "container": {
-            "command": "go test -tags=integration github.com/elastic/beats/metricbeat/module/... -data",
-            "created": "2017-12-07T07:20:57.000Z",
-            "id": "d88e67bb6961a5bb70c1c1c48094c6030e43768eed91e827f437111888f9967e",
-            "image": "metricbeatsnapshotnoxpack700alpha1d71419298b58ed8b0a5b60a6d1e4a476ffaf80a8_beat",
+            "command": "/usr/local/bin/docker-entrypoint.sh eswrapper",
+            "created": "2019-02-25T10:18:10.000Z",
             "ip_addresses": [
-                "172.18.0.27"
+                "172.23.0.2"
             ],
             "labels": {
+                "com_docker_compose_config-hash": "e3e0a2c6e5d1afb741bc8b1ecb09cda0395886b7a3e5084a9fd110be46d70f78",
                 "com_docker_compose_container-number": "1",
-                "com_docker_compose_oneoff": "True",
-                "com_docker_compose_project": "metricbeatsnapshotnoxpack700alpha1d71419298b58ed8b0a5b60a6d1e4a476ffaf80a8",
-                "com_docker_compose_service": "beat",
-                "com_docker_compose_version": "1.16.1"
+                "com_docker_compose_oneoff": "False",
+                "com_docker_compose_project": "metricbeat",
+                "com_docker_compose_service": "elasticsearch",
+                "com_docker_compose_slug": "df866b3a7b3d50c0802350cbe58ee5b34fa32b7f6ba7fe9e48cde2c12dd0201d",
+                "com_docker_compose_version": "1.23.1",
+                "license": "Elastic License",
+                "org_label-schema_build-date": "20181006",
+                "org_label-schema_license": "GPLv2",
+                "org_label-schema_name": "elasticsearch",
+                "org_label-schema_schema-version": "1.0",
+                "org_label-schema_url": "https://www.elastic.co/products/elasticsearch",
+                "org_label-schema_vcs-url": "https://github.com/elastic/elasticsearch-docker",
+                "org_label-schema_vendor": "Elastic",
+                "org_label-schema_version": "6.5.1"
             },
-            "name": "metricbeatsnapshotnoxpack700alpha1d71419298b58ed8b0a5b60a6d1e4a476ffaf80a8_beat_run_1",
             "size": {
                 "root_fs": 0,
                 "rw": 0
             },
-            "status": "Up 6 minutes (healthy)"
+            "status": "Up 7 minutes (healthy)"
         }
     },
+    "event": {
+        "dataset": "docker.container",
+        "duration": 115000,
+        "module": "docker"
+    },
     "metricset": {
-        "host": "/var/run/docker.sock",
-        "module": "docker",
-        "name": "container",
-        "rtt": 115
+        "name": "container"
+    },
+    "service": {
+        "address": "/var/run/docker.sock",
+        "type": "docker"
     }
 }
diff --git a/metricbeat/module/docker/container/container.go b/metricbeat/module/docker/container/container.go
@@ -22,12 +22,15 @@ import (
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/client"
+	"github.com/pkg/errors"
 
-	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/module/docker"
 )
 
+var logger = logp.NewLogger("docker.container")
+
 func init() {
 	mb.Registry.MustAddMetricSet("docker", "container", New,
 		mb.WithHostParser(docker.HostParser),
@@ -62,11 +65,14 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch returns a list of all containers as events.
 // This is based on https://docs.docker.com/engine/reference/api/docker_remote_api_v1.24/#/list-containers.
-func (m *MetricSet) Fetch() ([]common.MapStr, error) {
+func (m *MetricSet) Fetch(r mb.ReporterV2) {
 	// Fetch a list of all containers.
 	containers, err := m.dockerClient.ContainerList(context.Background(), types.ContainerListOptions{})
 	if err != nil {
-		return nil, err
+		err = errors.Wrap(err, "failed to get docker containers list")
+		logger.Error(err)
+		r.Error(err)
+		return
 	}
-	return eventsMapping(containers, m.dedot), nil
+	eventsMapping(r, containers, m.dedot)
 }
diff --git a/metricbeat/module/docker/container/container_integration_test.go b/metricbeat/module/docker/container/container_integration_test.go
@@ -22,13 +22,20 @@ package container
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 )
 
 func TestData(t *testing.T) {
-	f := mbtest.NewEventsFetcher(t, getConfig())
-	err := mbtest.WriteEvents(f, t)
-	if err != nil {
+	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+	events, errs := mbtest.ReportingFetchV2(f)
+	if len(errs) > 0 {
+		t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
+	}
+	assert.NotEmpty(t, events)
+
+	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }

diff --git a/metricbeat/module/docker/container/data.go b/metricbeat/module/docker/container/data.go
@@ -23,39 +23,49 @@ import (
 	"github.com/docker/docker/api/types"
 
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/module/docker"
 )
 
-func eventsMapping(containersList []types.Container, dedot bool) []common.MapStr {
-	myEvents := []common.MapStr{}
+func eventsMapping(r mb.ReporterV2, containersList []types.Container, dedot bool) {
 	for _, container := range containersList {
-		myEvents = append(myEvents, eventMapping(&container, dedot))
+		eventMapping(r, &container, dedot)
 	}
-	return myEvents
 }
 
-func eventMapping(cont *types.Container, dedot bool) common.MapStr {
+func eventMapping(r mb.ReporterV2, cont *types.Container, dedot bool) {
 	event := common.MapStr{
-		"created":      common.Time(time.Unix(cont.Created, 0)),
-		"id":           cont.ID,
-		"name":         docker.ExtractContainerName(cont.Names),
-		"command":      cont.Command,
-		"image":        cont.Image,
-		"ip_addresses": extractIPAddresses(cont.NetworkSettings),
-		"size": common.MapStr{
-			"root_fs": cont.SizeRootFs,
-			"rw":      cont.SizeRw,
+		"container": common.MapStr{
+			"id": cont.ID,
+			"image": common.MapStr{
+				"name": cont.Image,
+			},
+			"name":    docker.ExtractContainerName(cont.Names),
+			"runtime": "docker",
+		},
+		"docker": common.MapStr{
+			"container": common.MapStr{
+				"created":      common.Time(time.Unix(cont.Created, 0)),
+				"command":      cont.Command,
+				"ip_addresses": extractIPAddresses(cont.NetworkSettings),
+				"size": common.MapStr{
+					"root_fs": cont.SizeRootFs,
+					"rw":      cont.SizeRw,
+				},
+				"status": cont.Status,
+			},
 		},
-		"status": cont.Status,
 	}
 
 	labels := docker.DeDotLabels(cont.Labels, dedot)
 
 	if len(labels) > 0 {
-		event["labels"] = labels
+		event.Put("docker.container.labels", labels)
 	}
 
-	return event
+	r.Event(mb.Event{
+		RootFields: event,
+	})
 }
 
 func extractIPAddresses(networks *types.SummaryNetworkSettings) []string {

diff --git a/metricbeat/module/docker/cpu/_meta/data.json b/metricbeat/module/docker/cpu/_meta/data.json
@@ -1,67 +1,84 @@
 {
     "@timestamp": "2017-10-12T08:05:34.853Z",
-    "beat": {
+    "agent": {
         "hostname": "host.example.com",
         "name": "host.example.com"
     },
+    "container": {
+        "id": "cc78e58acfda4501105dc4de8e3ae218f2da616213e6e3af168c40103829302a",
+        "image": {
+            "name": "metricbeat_elasticsearch"
+        },
+        "name": "metricbeat_elasticsearch_1_df866b3a7b3d",
+        "runtime": "docker"
+    },
     "docker": {
         "container": {
-            "id": "bbdcbc751e7eda7a50e773b4a5d8c2800af664f835ef9c0ad6bbb9c160c50d83",
             "labels": {
-                "build-date": "20170911",
-                "com_docker_compose_config-hash": "371e477ae73fd44b19bcbcf0d4feaf4de9adfb69137a8f16d09cff749724ca99",
+                "com_docker_compose_config-hash": "e3e0a2c6e5d1afb741bc8b1ecb09cda0395886b7a3e5084a9fd110be46d70f78",
                 "com_docker_compose_container-number": "1",
                 "com_docker_compose_oneoff": "False",
                 "com_docker_compose_project": "metricbeat",
                 "com_docker_compose_service": "elasticsearch",
-                "com_docker_compose_version": "1.21.0",
-                "license": "GPLv2",
-                "maintainer": "Elastic Docker Team \u003cdocker@elastic.co\u003e",
-                "name": "CentOS Base Image",
-                "vendor": "CentOS"
-            },
-            "name": "metricbeat_elasticsearch_1"
+                "com_docker_compose_slug": "df866b3a7b3d50c0802350cbe58ee5b34fa32b7f6ba7fe9e48cde2c12dd0201d",
+                "com_docker_compose_version": "1.23.1",
+                "license": "Elastic License",
+                "org_label-schema_build-date": "20181006",
+                "org_label-schema_license": "GPLv2",
+                "org_label-schema_name": "elasticsearch",
+                "org_label-schema_schema-version": "1.0",
+                "org_label-schema_url": "https://www.elastic.co/products/elasticsearch",
+                "org_label-schema_vcs-url": "https://github.com/elastic/elasticsearch-docker",
+                "org_label-schema_vendor": "Elastic",
+                "org_label-schema_version": "6.5.1"
+            }
         },
         "cpu": {
             "core": {
                 "0": {
-                    "pct": 0.01012583677581864,
-                    "ticks": 9528454911
+                    "pct": 0.03263313721518987,
+                    "ticks": 38346196894
                 },
                 "1": {
-                    "pct": 0.0069975889168765746,
-                    "ticks": 11916812270
+                    "pct": 0.014317838987341772,
+                    "ticks": 37143007802
                 },
                 "2": {
-                    "pct": 0.001329603022670025,
-                    "ticks": 10894346015
+                    "pct": 0.0028625296202531647,
+                    "ticks": 37194678570
                 },
                 "3": {
-                    "pct": 0.0018390015113350126,
-                    "ticks": 10847487614
+                    "pct": 0.005687502784810126,
+                    "ticks": 39335551141
                 }
             },
             "kernel": {
-                "pct": 0.010075566750629723,
-                "ticks": 1960000000
+                "pct": 0.010126582278481013,
+                "ticks": 10560000000
             },
             "system": {
                 "pct": 4,
-                "ticks": 1092479570000000
+                "ticks": 5566563680000000
             },
             "total": {
-                "pct": 0.02029203022670025
+                "pct": 0.05550100860759494
             },
             "user": {
-                "pct": 0.010075566750629723,
-                "ticks": 40960000000
+                "pct": 0.05063291139240506,
+                "ticks": 139520000000
             }
         }
     },
+    "event": {
+        "dataset": "docker.cpu",
+        "duration": 115000,
+        "module": "docker"
+    },
     "metricset": {
-        "host": "/var/run/docker.sock",
-        "module": "docker",
-        "name": "cpu",
-        "rtt": 115
+        "name": "cpu"
+    },
+    "service": {
+        "address": "/var/run/docker.sock",
+        "type": "docker"
     }
 }
diff --git a/metricbeat/module/docker/cpu/cpu.go b/metricbeat/module/docker/cpu/cpu.go
@@ -19,12 +19,15 @@ package cpu
 
 import (
 	"github.com/docker/docker/client"
+	"github.com/pkg/errors"
 
-	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/module/docker"
 )
 
+var logger = logp.NewLogger("docker.cpu")
+
 func init() {
 	mb.Registry.MustAddMetricSet("docker", "cpu", New,
 		mb.WithHostParser(docker.HostParser),
@@ -69,12 +72,15 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 }
 
 // Fetch returns a list of docker CPU stats.
-func (m *MetricSet) Fetch() ([]common.MapStr, error) {
+func (m *MetricSet) Fetch(r mb.ReporterV2) {
 	stats, err := docker.FetchStats(m.dockerClient, m.Module().Config().Timeout)
 	if err != nil {
-		return nil, err
+		err = errors.Wrap(err, "failed to get docker stats")
+		logger.Error(err)
+		r.Error(err)
+		return
 	}
 
 	formattedStats := m.cpuService.getCPUStatsList(stats, m.dedot)
-	return eventsMapping(formattedStats), nil
+	eventsMapping(r, formattedStats)
 }
diff --git a/metricbeat/module/docker/cpu/cpu_integration_test.go b/metricbeat/module/docker/cpu/cpu_integration_test.go
@@ -22,13 +22,20 @@ package cpu
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 )
 
 func TestData(t *testing.T) {
-	f := mbtest.NewEventsFetcher(t, getConfig())
-	err := mbtest.WriteEvents(f, t)
-	if err != nil {
+	f := mbtest.NewReportingMetricSetV2(t, getConfig())
+	events, errs := mbtest.ReportingFetchV2(f)
+	if len(errs) > 0 {
+		t.Fatalf("Expected 0 error, had %d. %v\n", len(errs), errs)
+	}
+	assert.NotEmpty(t, events)
+
+	if err := mbtest.WriteEventsReporterV2(f, t, ""); err != nil {
 		t.Fatal("write", err)
 	}
 }