Add support for disk scraper (#17)

* Add support of disk scraper * Tidying up * Add tests and sync to latest changes * Complete the tests for disk metrics * Refactored code * Move metricsToAdd as a global variable * Refactoring the code further * Updated go mod * Update variable name

Author: ishleenk17

remappers/hostmetrics/disk.go

@@ -0,0 +1,86 @@
+// 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.
+
+package hostmetrics
+
+import (
+"errors"
+"fmt"
+
+remappers "github.com/elastic/opentelemetry-lib/remappers/internal"
+"go.opentelemetry.io/collector/pdata/pcommon"
+"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+var metricsToAdd = map[string]string{
+"system.disk.io": "system.diskio.%s.bytes",
+"system.disk.operations": "system.diskio.%s.count",
+"system.disk.pending_operations": "system.diskio.io.%sops",
+"system.disk.operation_time": "system.diskio.%s.time",
+"system.disk.io_time": "system.diskio.io.%stime",
+}
+
+// remapDiskMetrics remaps disk-related metrics from the source to the output metric slice.
+func remapDiskMetrics(src, out pmetric.MetricSlice, _ pcommon.Resource, dataset string) error {
+var errs []error
+for i := 0; i < src.Len(); i++ {
+var err error
+metric := src.At(i)
+switch metric.Name() {
+case "system.disk.io", "system.disk.operations", "system.disk.pending_operations":
+err = addDiskMetric(metric, out, dataset, 1)
+case "system.disk.operation_time", "system.disk.io_time":
+err = addDiskMetric(metric, out, dataset, 1000)
+}
+if err != nil {
+errs = append(errs, err)
+}
+}
+return errors.Join(errs...)
+}
+
+func addDiskMetric(metric pmetric.Metric, out pmetric.MetricSlice, dataset string, multiplier int64) error {
+metricDiskES, ok := metricsToAdd[metric.Name()]
+if !ok {
+return fmt.Errorf("unexpected metric name: %s", metric.Name())
+}
+
+dps := metric.Sum().DataPoints()
+for i := 0; i < dps.Len(); i++ {
+dp := dps.At(i)
+if device, ok := dp.Attributes().Get("device"); ok {
+direction, _ := dp.Attributes().Get("direction")
+remappedMetric := remappers.Metric{
+DataType: pmetric.MetricTypeSum,
+Name: fmt.Sprintf(metricDiskES, direction.Str()),
+Timestamp: dp.Timestamp(),
+}
+switch dp.ValueType() {
+case pmetric.NumberDataPointValueTypeInt:
+v := dp.IntValue() * multiplier
+remappedMetric.IntValue = &v
+case pmetric.NumberDataPointValueTypeDouble:
+v := dp.DoubleValue() * float64(multiplier)
+remappedMetric.DoubleValue = &v
+}
+remappers.AddMetrics(out, dataset, func(dp pmetric.NumberDataPoint) {
+dp.Attributes().PutStr("system.diskio.name", device.Str())
+}, remappedMetric)
+}
+}
+return nil
+}

remappers/hostmetrics/hostmetrics.go

@@ -49,6 +49,7 @@ var remapFuncs = map[string]remapFunc{
 "process": remapProcessMetrics,
 "processes": remapProcessesMetrics,
 "network": remapNetworkMetrics,
+"disk": remapDiskMetrics,
 }
 
 // Remapper maps the OTel hostmetrics to Elastic system metrics. These remapped

remappers/hostmetrics/hostmetrics_test.go

@@ -41,6 +41,7 @@ var (
 ProcPath string = "/bin/run"
 ProcName string = "runner"
 Device string = "en0"
+Disk string = "nvme0n1p128"
 )
 
 func TestRemap(t *testing.T) {
@@ -66,6 +67,8 @@ func doTestRemap(t *testing.T, id string, remapOpts ...Option) {
 m["process.name"] = ProcName
 case "network":
 m["system.network.name"] = Device
+case "disk":
+m["system.diskio.name"] = Disk
 }
 return m
 }
@@ -267,6 +270,30 @@ func doTestRemap(t *testing.T, id string, remapOpts ...Option) {
 {Type: Sum, Name: "system.network.out.errors", DP: testDP{Ts: now, Int: ptr(int64(2)), Attrs: outAttr("network")}},
 },
 },
+{
+name: "disk",
+scraper: "disk",
+input: []testMetric{
+{Type: Sum, Name: "system.disk.io", DP: testDP{Ts: now, Int: ptr(int64(1888256)), Attrs: map[string]any{"device": Disk, "direction": "read"}}},
+{Type: Sum, Name: "system.disk.io", DP: testDP{Ts: now, Int: ptr(int64(512)), Attrs: map[string]any{"device": Disk, "direction": "write"}}},
+{Type: Sum, Name: "system.disk.operations", DP: testDP{Ts: now, Int: ptr(int64(15390)), Attrs: map[string]any{"device": Disk, "direction": "read"}}},
+{Type: Sum, Name: "system.disk.operations", DP: testDP{Ts: now, Int: ptr(int64(371687)), Attrs: map[string]any{"device": Disk, "direction": "write"}}},
+{Type: Sum, Name: "system.disk.operation_time", DP: testDP{Ts: now, Dbl: ptr(11.182), Attrs: map[string]any{"device": Disk, "direction": "read"}}},
+{Type: Sum, Name: "system.disk.operation_time", DP: testDP{Ts: now, Dbl: ptr(617.289), Attrs: map[string]any{"device": Disk, "direction": "write"}}},
+{Type: Sum, Name: "system.disk.io_time", DP: testDP{Ts: now, Dbl: ptr(520.3), Attrs: map[string]any{"device": Disk}}},
+{Type: Sum, Name: "system.disk.pending_operations", DP: testDP{Ts: now, Int: ptr(int64(102)), Attrs: map[string]any{"device": Disk}}},
+},
+expected: []testMetric{
+{Type: Sum, Name: "system.diskio.read.bytes", DP: testDP{Ts: now, Int: ptr(int64(1888256)), Attrs: outAttr("disk")}},
+{Type: Sum, Name: "system.diskio.write.bytes", DP: testDP{Ts: now, Int: ptr(int64(512)), Attrs: outAttr("disk")}},
+{Type: Sum, Name: "system.diskio.read.count", DP: testDP{Ts: now, Int: ptr(int64(15390)), Attrs: outAttr("disk")}},
+{Type: Sum, Name: "system.diskio.write.count", DP: testDP{Ts: now, Int: ptr(int64(371687)), Attrs: outAttr("disk")}},
+{Type: Sum, Name: "system.diskio.read.time", DP: testDP{Ts: now, Dbl: ptr(11182.0), Attrs: outAttr("disk")}},
+{Type: Sum, Name: "system.diskio.write.time", DP: testDP{Ts: now, Dbl: ptr(617289.0), Attrs: outAttr("disk")}},
+{Type: Sum, Name: "system.diskio.io.time", DP: testDP{Ts: now, Dbl: ptr(520300.0), Attrs: outAttr("disk")}},
+{Type: Sum, Name: "system.diskio.io.ops", DP: testDP{Ts: now, Int: ptr(int64(102)), Attrs: outAttr("disk")}},
+},
+},
 } {
 t.Run(fmt.Sprintf("%s/%s", tc.name, id), func(t *testing.T) {
 sm := pmetric.NewScopeMetrics()