Add docker diskio stats on Windows (#8126) (#8134)

Docker stats use a different data structure for blkio stats in Windows, the main difference is that in posix systems (Linux at least) stats are separated by device by Major/Minor (lists of BlkioStatsEntry), and in Windows they are directly summarized by reads and writes, by operations and volume (StorageStats).

This change counts the metrics of both data structures and aggregates the result.

Fixes #6815

(cherry picked from commit 63f25a4)

CHANGELOG.asciidoc

@@ -56,6 +56,7 @@ https://github.com/elastic/beats/compare/v6.4.0...6.x[Check the HEAD diff]
 
 - Fix golang.heap.gc.cpu_fraction type from long to float in Golang module. {pull}7789[7789]
 - Fixed the RPM by designating the modules.d config files as configuration data in the RPM spec. {issue}8075[8075]
+- Add docker diskio stats on Windows. {issue}6815[6815] {pull}8126[8126]
 
 *Packetbeat*
 

metricbeat/module/docker/diskio/diskio_test.go

@@ -222,7 +222,7 @@ func setTime(index int) {
 	newBlkioRaw[index].Time = oldBlkioRaw[index].Time.Add(time.Duration(2000000000))
 }
 
-func TestGetBlkioStats(t *testing.T) {
+func TestGetBlkioStatsList(t *testing.T) {
 	start := time.Now()
 	later := start.Add(10 * time.Second)
 
@@ -232,7 +232,7 @@ func TestGetBlkioStats(t *testing.T) {
 		},
 	}
 
-	dockerStats := &docker.Stat{
+	dockerStats := []docker.Stat{{
 		Container: &types.Container{
 			ID:    "cebada",
 			Names: []string{"test"},
@@ -258,9 +258,49 @@ func TestGetBlkioStats(t *testing.T) {
 				},
 			},
 		}},
+	}}
+
+	statsList := blkioService.getBlkioStatsList(dockerStats, true)
+	stats := statsList[0]
+	assert.Equal(t, float64(5), stats.reads)
+	assert.Equal(t, float64(10), stats.writes)
+	assert.Equal(t, float64(15), stats.totals)
+	assert.Equal(t,
+		BlkioRaw{Time: later, reads: 150, writes: 300, totals: 450},
+		stats.serviced)
+	assert.Equal(t,
+		BlkioRaw{Time: later, reads: 1500, writes: 3000, totals: 4500},
+		stats.servicedBytes)
+}
+
+func TestGetBlkioStatsListWindows(t *testing.T) {
+	start := time.Now()
+	later := start.Add(10 * time.Second)
+
+	blkioService := BlkioService{
+		map[string]BlkioRaw{
+			"cebada": {Time: start, reads: 100, writes: 200, totals: 300},
+		},
 	}
 
-	stats := blkioService.getBlkioStats(dockerStats, true)
+	dockerStats := []docker.Stat{{
+		Container: &types.Container{
+			ID:    "cebada",
+			Names: []string{"test"},
+		},
+		Stats: types.StatsJSON{Stats: types.Stats{
+			Read: later,
+			StorageStats: types.StorageStats{
+				ReadCountNormalized:  150,
+				WriteCountNormalized: 300,
+				ReadSizeBytes:        1500,
+				WriteSizeBytes:       3000,
+			},
+		}},
+	}}
+
+	statsList := blkioService.getBlkioStatsList(dockerStats, true)
+	stats := statsList[0]
 	assert.Equal(t, float64(5), stats.reads)
 	assert.Equal(t, float64(10), stats.writes)
 	assert.Equal(t, float64(15), stats.totals)

metricbeat/module/docker/diskio/helper.go

@@ -36,13 +36,30 @@ type BlkioStats struct {
 	servicedBytes BlkioRaw
 }
 
+// Add adds blkio stats
+func (s *BlkioStats) Add(o *BlkioStats) {
+	s.reads += o.reads
+	s.writes += o.writes
+	s.totals += o.totals
+
+	s.serviced.Add(&o.serviced)
+	s.servicedBytes.Add(&o.servicedBytes)
+}
+
 type BlkioRaw struct {
 	Time   time.Time
 	reads  uint64
 	writes uint64
 	totals uint64
 }
 
+// Add adds blkio raw stats
+func (s *BlkioRaw) Add(o *BlkioRaw) {
+	s.reads += o.reads
+	s.writes += o.writes
+	s.totals += o.totals
+}
+
 // BlkioService is a helper to collect and calculate disk I/O metrics
 type BlkioService struct {
 	lastStatsPerContainer map[string]BlkioRaw
@@ -61,34 +78,59 @@ func (io *BlkioService) getBlkioStatsList(rawStats []docker.Stat, dedot bool) []
 	statsPerContainer := make(map[string]BlkioRaw)
 	for _, myRawStats := range rawStats {
 		stats := io.getBlkioStats(&myRawStats, dedot)
-		statsPerContainer[myRawStats.Container.ID] = stats.serviced
+		storageStats := io.getStorageStats(&myRawStats, dedot)
+		stats.Add(&storageStats)
+
+		oldStats, exist := io.lastStatsPerContainer[stats.Container.ID]
+		if exist {
+			stats.reads = io.getReadPs(&oldStats, &stats.serviced)
+			stats.writes = io.getWritePs(&oldStats, &stats.serviced)
+			stats.totals = io.getTotalPs(&oldStats, &stats.serviced)
+		}
+
+		statsPerContainer[stats.Container.ID] = stats.serviced
 		formattedStats = append(formattedStats, stats)
 	}
 
 	io.lastStatsPerContainer = statsPerContainer
 	return formattedStats
 }
 
-func (io *BlkioService) getBlkioStats(myRawStat *docker.Stat, dedot bool) BlkioStats {
-	newBlkioStats := io.getNewStats(myRawStat.Stats.Read, myRawStat.Stats.BlkioStats.IoServicedRecursive)
-	bytesBlkioStats := io.getNewStats(myRawStat.Stats.Read, myRawStat.Stats.BlkioStats.IoServiceBytesRecursive)
+// getStorageStats collects diskio metrics from StorageStats structure, that
+// is populated in Windows systems only
+func (io *BlkioService) getStorageStats(myRawStats *docker.Stat, dedot bool) BlkioStats {
+	return BlkioStats{
+		Time:      myRawStats.Stats.Read,
+		Container: docker.NewContainer(myRawStats.Container, dedot),
+
+		serviced: BlkioRaw{
+			reads:  myRawStats.Stats.StorageStats.ReadCountNormalized,
+			writes: myRawStats.Stats.StorageStats.WriteCountNormalized,
+			totals: myRawStats.Stats.StorageStats.ReadCountNormalized + myRawStats.Stats.StorageStats.WriteCountNormalized,
+		},
+
+		servicedBytes: BlkioRaw{
+			reads:  myRawStats.Stats.StorageStats.ReadSizeBytes,
+			writes: myRawStats.Stats.StorageStats.WriteSizeBytes,
+			totals: myRawStats.Stats.StorageStats.ReadSizeBytes + myRawStats.Stats.StorageStats.WriteSizeBytes,
+		},
+	}
+}
 
-	myBlkioStats := BlkioStats{
+// getBlkioStats collects diskio metrics from BlkioStats structures, that
+// are not populated in Windows
+func (io *BlkioService) getBlkioStats(myRawStat *docker.Stat, dedot bool) BlkioStats {
+	return BlkioStats{
 		Time:      myRawStat.Stats.Read,
 		Container: docker.NewContainer(myRawStat.Container, dedot),
 
-		serviced:      newBlkioStats,
-		servicedBytes: bytesBlkioStats,
+		serviced: io.getNewStats(
+			myRawStat.Stats.Read,
+			myRawStat.Stats.BlkioStats.IoServicedRecursive),
+		servicedBytes: io.getNewStats(
+			myRawStat.Stats.Read,
+			myRawStat.Stats.BlkioStats.IoServiceBytesRecursive),
 	}
-
-	oldBlkioStats, exist := io.lastStatsPerContainer[myRawStat.Container.ID]
-	if exist {
-		myBlkioStats.reads = io.getReadPs(&oldBlkioStats, &newBlkioStats)
-		myBlkioStats.writes = io.getWritePs(&oldBlkioStats, &newBlkioStats)
-		myBlkioStats.totals = io.getTotalPs(&oldBlkioStats, &newBlkioStats)
-	}
-
-	return myBlkioStats
 }
 
 func (io *BlkioService) getNewStats(time time.Time, blkioEntry []types.BlkioStatEntry) BlkioRaw {