Added option "total/perdevice" to Docker input (#1525)

Like cpu plugin, add two option "total" and "perdevice" to send network
and diskio metrics either per device and/or the sum of all devices.

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### Features
 
 - [#1413](https://github.com/influxdata/telegraf/issues/1413): Separate container_version from container_image tag.
+- [#1525](https://github.com/influxdata/telegraf/pull/1525): Support setting per-device and total metrics for Docker network and blockio.
 
 ### Bugfixes
 

etc/telegraf.conf

@@ -666,6 +666,11 @@
 #   container_names = []
 #   ## Timeout for docker list, info, and stats commands
 #   timeout = "5s"
+#   ## Whether to report for each container per-device blkio (8:0, 8:1...) and
+#   ## network (eth0, eth1, ...) stats or not
+#   perdevice = true
+#   ## Whether to report for each container total blkio and network stats or not
+#   total = false
 
 
 # # Read statistics from one or many dovecot servers

plugins/inputs/docker/docker.go

@@ -25,6 +25,8 @@ type Docker struct {
 	Endpoint       string
 	ContainerNames []string
 	Timeout        internal.Duration
+	PerDevice      bool `toml:"perdevice"`
+	Total          bool `toml:"total"`
 
 	client DockerClient
 }
@@ -58,6 +60,13 @@ var sampleConfig = `
   container_names = []
   ## Timeout for docker list, info, and stats commands
   timeout = "5s"
+
+  ## Whether to report for each container per-device blkio (8:0, 8:1...) and
+  ## network (eth0, eth1, ...) stats or not
+  perdevice = true
+  ## Whether to report for each container total blkio and network stats or not
+  total = false
+
 `
 
 // Description returns input description
@@ -246,7 +255,7 @@ func (d *Docker) gatherContainer(
 		tags[k] = label
 	}
 
-	gatherContainerStats(v, acc, tags, container.ID)
+	gatherContainerStats(v, acc, tags, container.ID, d.PerDevice, d.Total)
 
 	return nil
 }
@@ -256,6 +265,8 @@ func gatherContainerStats(
 	acc telegraf.Accumulator,
 	tags map[string]string,
 	id string,
+	perDevice bool,
+	total bool,
 ) {
 	now := stat.Read
 
@@ -323,6 +334,7 @@ func gatherContainerStats(
 		acc.AddFields("docker_container_cpu", fields, percputags, now)
 	}
 
+	totalNetworkStatMap := make(map[string]interface{})
 	for network, netstats := range stat.Networks {
 		netfields := map[string]interface{}{
 			"rx_dropped":   netstats.RxDropped,
@@ -336,12 +348,35 @@ func gatherContainerStats(
 			"container_id": id,
 		}
 		// Create a new network tag dictionary for the "network" tag
+		if perDevice {
 			nettags := copyTags(tags)
 			nettags["network"] = network
 			acc.AddFields("docker_container_net", netfields, nettags, now)
 		}
+		if total {
+			for field, value := range netfields {
+				if field == "container_id" {
+					continue
+				}
+				_, ok := totalNetworkStatMap[field]
+				if ok {
+					totalNetworkStatMap[field] = totalNetworkStatMap[field].(uint64) + value.(uint64)
+				} else {
+					totalNetworkStatMap[field] = value
+				}
+			}
+		}
+	}
 
-	gatherBlockIOMetrics(stat, acc, tags, now, id)
+	// totalNetworkStatMap could be empty if container is running with --net=host.
+	if total && len(totalNetworkStatMap) != 0 {
+		nettags := copyTags(tags)
+		nettags["network"] = "total"
+		totalNetworkStatMap["container_id"] = id
+		acc.AddFields("docker_container_net", totalNetworkStatMap, nettags, now)
+	}
+
+	gatherBlockIOMetrics(stat, acc, tags, now, id, perDevice, total)
 }
 
 func calculateMemPercent(stat *types.StatsJSON) float64 {
@@ -370,6 +405,8 @@ func gatherBlockIOMetrics(
 	tags map[string]string,
 	now time.Time,
 	id string,
+	perDevice bool,
+	total bool,
 ) {
 	blkioStats := stat.BlkioStats
 	// Make a map of devices to their block io stats
@@ -431,12 +468,34 @@ func gatherBlockIOMetrics(
 		deviceStatMap[device]["sectors_recursive"] = metric.Value
 	}
 
+	totalStatMap := make(map[string]interface{})
 	for device, fields := range deviceStatMap {
+		fields["container_id"] = id
+		if perDevice {
 			iotags := copyTags(tags)
 			iotags["device"] = device
-		fields["container_id"] = id
 			acc.AddFields("docker_container_blkio", fields, iotags, now)
 		}
+		if total {
+			for field, value := range fields {
+				if field == "container_id" {
+					continue
+				}
+				_, ok := totalStatMap[field]
+				if ok {
+					totalStatMap[field] = totalStatMap[field].(uint64) + value.(uint64)
+				} else {
+					totalStatMap[field] = value
+				}
+			}
+		}
+	}
+	if total {
+		totalStatMap["container_id"] = id
+		iotags := copyTags(tags)
+		iotags["device"] = "total"
+		acc.AddFields("docker_container_blkio", totalStatMap, iotags, now)
+	}
 }
 
 func copyTags(in map[string]string) map[string]string {
@@ -480,6 +539,7 @@ func parseSize(sizeStr string) (int64, error) {
 func init() {
 	inputs.Add("docker", func() telegraf.Input {
 		return &Docker{
+			PerDevice: true,
 			Timeout:   internal.Duration{Duration: time.Second * 5},
 		}
 	})

plugins/inputs/docker/docker_test.go

@@ -24,7 +24,7 @@ func TestDockerGatherContainerStats(t *testing.T) {
 		"container_name":  "redis",
 		"container_image": "redis/image",
 	}
-	gatherContainerStats(stats, &acc, tags, "123456789")
+	gatherContainerStats(stats, &acc, tags, "123456789", true, true)
 
 	// test docker_container_net measurement
 	netfields := map[string]interface{}{
@@ -42,6 +42,21 @@ func TestDockerGatherContainerStats(t *testing.T) {
 	nettags["network"] = "eth0"
 	acc.AssertContainsTaggedFields(t, "docker_container_net", netfields, nettags)
 
+	netfields = map[string]interface{}{
+		"rx_dropped":   uint64(6),
+		"rx_bytes":     uint64(8),
+		"rx_errors":    uint64(10),
+		"tx_packets":   uint64(12),
+		"tx_dropped":   uint64(6),
+		"rx_packets":   uint64(8),
+		"tx_errors":    uint64(10),
+		"tx_bytes":     uint64(12),
+		"container_id": "123456789",
+	}
+	nettags = copyTags(tags)
+	nettags["network"] = "total"
+	acc.AssertContainsTaggedFields(t, "docker_container_net", netfields, nettags)
+
 	// test docker_blkio measurement
 	blkiotags := copyTags(tags)
 	blkiotags["device"] = "6:0"
@@ -52,6 +67,15 @@ func TestDockerGatherContainerStats(t *testing.T) {
 	}
 	acc.AssertContainsTaggedFields(t, "docker_container_blkio", blkiofields, blkiotags)
 
+	blkiotags = copyTags(tags)
+	blkiotags["device"] = "total"
+	blkiofields = map[string]interface{}{
+		"io_service_bytes_recursive_read": uint64(100),
+		"io_serviced_recursive_write":     uint64(302),
+		"container_id":                    "123456789",
+	}
+	acc.AssertContainsTaggedFields(t, "docker_container_blkio", blkiofields, blkiotags)
+
 	// test docker_container_mem measurement
 	memfields := map[string]interface{}{
 		"max_usage":                 uint64(1001),
@@ -186,6 +210,17 @@ func testStats() *types.StatsJSON {
 		TxBytes:   4,
 	}
 
+	stats.Networks["eth1"] = types.NetworkStats{
+		RxDropped: 5,
+		RxBytes:   6,
+		RxErrors:  7,
+		TxPackets: 8,
+		TxDropped: 5,
+		RxPackets: 6,
+		TxErrors:  7,
+		TxBytes:   8,
+	}
+
 	sbr := types.BlkioStatEntry{
 		Major: 6,
 		Minor: 0,
@@ -198,11 +233,19 @@ func testStats() *types.StatsJSON {
 		Op:    "write",
 		Value: 101,
 	}
+	sr2 := types.BlkioStatEntry{
+		Major: 6,
+		Minor: 1,
+		Op:    "write",
+		Value: 201,
+	}
 
 	stats.BlkioStats.IoServiceBytesRecursive = append(
 		stats.BlkioStats.IoServiceBytesRecursive, sbr)
 	stats.BlkioStats.IoServicedRecursive = append(
 		stats.BlkioStats.IoServicedRecursive, sr)
+	stats.BlkioStats.IoServicedRecursive = append(
+		stats.BlkioStats.IoServicedRecursive, sr2)
 
 	return stats
 }