Replace WMI query to get the system/diskio metrics for Windows

CHANGELOG.next.asciidoc

@@ -98,6 +98,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add _bucket to histogram metrics in Prometheus Collector {pull}11578[11578]
 - Prevent the docker/memory metricset from processing invalid events before container start {pull}11676[11676]
 - Change `add_cloud_metadata` processor to not overwrite `cloud` field when it already exist in the event. {pull}11612[11612] {issue}11305[11305]
+- Change diskio metrics retrieval method (only for Windows) from wmi query to DeviceIOControl function using the IOCTL_DISK_PERFORMANCE control code {pull}11635[11635]
 - Call GetMetricData api per region instead of per instance. {issue}11820[11820] {pull}11882[11882]
 
 *Packetbeat*

metricbeat/module/system/diskio/diskio.go

@@ -25,7 +25,6 @@ import (
 	"github.com/elastic/beats/metricbeat/mb/parse"
 
 	"github.com/pkg/errors"
-	"github.com/shirou/gopsutil/disk"
 )
 
 func init() {
@@ -60,7 +59,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch fetches disk IO metrics from the OS.
 func (m *MetricSet) Fetch(r mb.ReporterV2) {
-	stats, err := disk.IOCounters(m.includeDevices...)
+	stats, err := IOCounters(m.includeDevices...)
 	if err != nil {
 		r.Error(errors.Wrap(err, "disk io counters"))
 		return
@@ -89,8 +88,8 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 				"time": counters.IoTime,
 			},
 		}
-
-		extraMetrics, err := m.statistics.CalIOStatistics(counters)
+		var extraMetrics DiskIOMetric
+		err := m.statistics.CalIOStatistics(&extraMetrics, counters)
 		if err == nil {
 			event["iostat"] = common.MapStr{
 				"read": common.MapStr{

metricbeat/module/system/diskio/diskstat_linux.go

@@ -27,38 +27,44 @@ import (
 )
 
 func Get_CLK_TCK() uint32 {
-	//return uint32(C.sysconf(C._SC_CLK_TCK))
-	//NOTE: _SC_CLK_TCK should be fetched from sysconf using cgo
+	// return uint32(C.sysconf(C._SC_CLK_TCK))
+	// NOTE: _SC_CLK_TCK should be fetched from sysconf using cgo
 	return uint32(100)
 }
 
+// IOCounters should map functionality to disk package for linux os.
+func IOCounters(names ...string) (map[string]disk.IOCountersStat, error) {
+	return disk.IOCounters(names...)
+}
+
+// NewDiskIOStat :init DiskIOStat object.
 func NewDiskIOStat() *DiskIOStat {
-	d := &DiskIOStat{}
-	d.lastDiskIOCounters = make(map[string]disk.IOCountersStat)
-	return d
+	return &DiskIOStat{
+		lastDiskIOCounters: map[string]disk.IOCountersStat{},
+	}
 }
 
-// create current cpu sampling
-// need call as soon as get IOCounters
+// OpenSampling creates current cpu sampling
+// need call as soon as get IOCounters.
 func (stat *DiskIOStat) OpenSampling() error {
 	return stat.curCpu.Get()
 }
 
-func (stat *DiskIOStat) CalIOStatistics(counter disk.IOCountersStat) (DiskIOMetric, error) {
+// CalIOStatistics calculates IO statistics.
+func (stat *DiskIOStat) CalIOStatistics(result *DiskIOMetric, counter disk.IOCountersStat) error {
 	var last disk.IOCountersStat
 	var ok bool
-	var result DiskIOMetric
 
 	// if last counter not found, create one and return all 0
 	if last, ok = stat.lastDiskIOCounters[counter.Name]; !ok {
 		stat.lastDiskIOCounters[counter.Name] = counter
-		return result, nil
+		return nil
 	}
 
 	// calculate the delta ms between the CloseSampling and OpenSampling
 	deltams := 1000.0 * float64(stat.curCpu.Total()-stat.lastCpu.Total()) / float64(cpu.NumCores) / float64(Get_CLK_TCK())
 	if deltams <= 0 {
-		return result, errors.New("The delta cpu time between close sampling and open sampling is less or equal to 0")
+		return errors.New("The delta cpu time between close sampling and open sampling is less or equal to 0")
 	}
 
 	rd_ios := counter.ReadCount - last.ReadCount
@@ -111,7 +117,7 @@ func (stat *DiskIOStat) CalIOStatistics(counter disk.IOCountersStat) (DiskIOMetr
 	}
 
 	stat.lastDiskIOCounters[counter.Name] = counter
-	return result, nil
+	return nil
 
 }
 

metricbeat/module/system/diskio/diskstat_linux_test.go

@@ -104,8 +104,8 @@ func TestDiskIOStat_CalIOStatistics(t *testing.T) {
 		AvgReadAwaitTime:  1.2,
 		AvgWriteAwaitTime: 1,
 	}
-
-	got, err := stat.CalIOStatistics(counter)
+	var got DiskIOMetric
+	err := stat.CalIOStatistics(&got, counter)
 	if err != nil {
 		t.Fatal(err)
 	}

metricbeat/module/system/diskio/diskstat_other.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-// +build darwin,cgo freebsd windows
+// +build darwin,cgo freebsd
 
 package diskio
 
@@ -24,21 +24,27 @@ import (
 	"github.com/shirou/gopsutil/disk"
 )
 
+// NewDiskIOStat :init DiskIOStat object.
 func NewDiskIOStat() *DiskIOStat {
-	d := &DiskIOStat{}
-	d.lastDiskIOCounters = make(map[string]disk.IOCountersStat)
-	return d
+	return &DiskIOStat{
+		lastDiskIOCounters: map[string]disk.IOCountersStat{},
+	}
 }
 
+// OpenSampling stub for linux implementation.
 func (stat *DiskIOStat) OpenSampling() error {
 	return nil
 }
 
-func (stat *DiskIOStat) CalIOStatistics(counter disk.IOCountersStat) (DiskIOMetric, error) {
-	var result DiskIOMetric
-	return result, errors.New("Not implemented out of linux")
+// CalIOStatistics stub for linux implementation.
+func (stat *DiskIOStat) CalIOStatistics(result *DiskIOMetric, counter disk.IOCountersStat) error {
+	return errors.New("not implemented out of linux")
 }
 
-func (stat *DiskIOStat) CloseSampling() {
-	return
+// CloseSampling stub for linux implementation.
+func (stat *DiskIOStat) CloseSampling() {}
+
+// IOCounters should map functionality to disk package for linux os.
+func IOCounters(names ...string) (map[string]disk.IOCountersStat, error) {
+	return disk.IOCounters(names...)
 }

metricbeat/module/system/diskio/diskstat_windows.go

@@ -0,0 +1,50 @@
+// 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.
+
+// +build windows
+
+package diskio
+
+import (
+	"github.com/pkg/errors"
+	"github.com/shirou/gopsutil/disk"
+)
+
+// NewDiskIOStat :init DiskIOStat object.
+func NewDiskIOStat() *DiskIOStat {
+	return &DiskIOStat{
+		lastDiskIOCounters: map[string]disk.IOCountersStat{},
+	}
+}
+
+// OpenSampling stub for linux implementation.
+func (stat *DiskIOStat) OpenSampling() error {
+	return nil
+}
+
+// CalIOStatistics stub for linux implementation.
+func (stat *DiskIOStat) CalIOStatistics(result *DiskIOMetric, counter disk.IOCountersStat) error {
+	return errors.New("iostat is not implement for Windows")
+}
+
+// CloseSampling stub for linux implementation.
+func (stat *DiskIOStat) CloseSampling() {}
+
+// IOCounters should map functionality to disk package for linux os.
+func IOCounters(names ...string) (map[string]disk.IOCountersStat, error) {
+	return ioCounters(names...)
+}