Remap host.network.* based on transformed metrics

[lahsivjar]

This code remaps host.network.* metrics based on a pre-transformed metric generated by system.network.*. The pre-transformation is required because the source metric from network scrapper (system.network.io and system.network.packets) are cumulative metric with monotonically increasing values. For the remapping purposes, the OTel's system.network.{io, packets} metric is transformed to 2 metrics:

1. system.network.{in, out}.{bytes, packets} which is a monotonically increasing cumulative metric, and
2. host.network.{ingress, egress}.{bytes, packets} which is a delta cumulative metric

This means that we cannot just translate the source OTel metrics to delta. The PR relies on doing a metric transformation of the source OTel metric via a processor and then transforming the processed metric to delta. Example configuration:

processors:
  metricstransform:
    transforms:
      - include: "^system\\.network\\.(io|packets)$$"
        match_type: regexp
        action: insert
        new_name: host.network.$${1}
        operations:
          - action: aggregate_labels
            label_set: [direction]
            aggregation_type: sum
  cumulativetodelta:
    include:
      metrics:
        - "^host\\.network\\.(io|packets)$$"
      match_type: regexp

The above processors can be shipped along with the hostmetrics configuration to the customers.

Related issues

https://github.com/elastic/opentelemetry-lib/issues/14 https://github.com/elastic/opentelemetry-lib/issues/9

[shmsr]

As I am not familiar with the core logic, my review will primarily focus on code style, readability, and potential improvements to the Go-specific aspects of the codebase. The following patch suggests mostly nitpicks and suggestions for enhancing the code quality.

diff --git a/remappers/hostmetrics/network.go b/remappers/hostmetrics/network.go
index 01bf1fa..8bc4b70 100644
--- a/remappers/hostmetrics/network.go
+++ b/remappers/hostmetrics/network.go
@@ -25,7 +25,7 @@ import (
    "go.opentelemetry.io/collector/pdata/pmetric"
 )

-var metricsMapping = map[string]string{
+var metricsFormatMapping = map[string]string{
    "system.network.io":      "system.network.%s.bytes",
    "system.network.packets": "system.network.%s.packets",
    "system.network.dropped": "system.network.%s.dropped",
@@ -43,10 +43,11 @@ func remapNetworkMetrics(
        metricsetPeriod    int64
        metricsetTimestamp pcommon.Timestamp
    )
+
    for i := 0; i < src.Len(); i++ {
        m := src.At(i)

-       elasticMetric, ok := metricsMapping[m.Name()]
+       elasticMetric, ok := metricsFormatMapping[m.Name()]
        if !ok {
            continue
        }
@@ -54,27 +55,23 @@ func remapNetworkMetrics(
        // Special case handling for host network metrics produced by aggregating
        // system network metrics and converting to delta temporality. These host
        // metrics have been aggregated to drop all labels other than direction.
-       transformedHostMetrics := strings.HasPrefix(m.Name(), "host.")
+       isHostMetric := strings.HasPrefix(m.Name(), "host.")

        dataPoints := m.Sum().DataPoints()
        for j := 0; j < dataPoints.Len(); j++ {
            dp := dataPoints.At(j)

-           device, ok := dp.Attributes().Get("device")
-           if !ok && !transformedHostMetrics {
-               continue
-           }
-
-           direction, ok := dp.Attributes().Get("direction")
-           if !ok {
+           device, deviceOk := dp.Attributes().Get("device")
+           direction, directionOk := dp.Attributes().Get("direction")
+           if (!isHostMetric && !deviceOk) || !directionOk {
                continue
            }

            ts := dp.Timestamp()
            value := dp.IntValue()
            metricDataType := pmetric.MetricTypeSum
-           if transformedHostMetrics {
-               // transformed metrics are gauges as they are trasformed from cumulative to delta
+           if isHostMetric {
+               // Transformed metrics are gauges as they are transformed from cumulative to delta.
                metricDataType = pmetric.MetricTypeGauge
                startTs := dp.StartTimestamp()
                if metricsetPeriod == 0 && startTs != 0 && startTs < ts {
@@ -90,11 +87,8 @@ func remapNetworkMetrics(
                    }
                },
                metric{
-                   dataType: metricDataType,
-                   name: fmt.Sprintf(
-                       elasticMetric,
-                       normalizeDirection(direction, transformedHostMetrics),
-                   ),
+                   dataType:  metricDataType,
+                   name:      fmt.Sprintf(elasticMetric, normalizeDirection(direction, isHostMetric)),
                    timestamp: ts,
                    intValue:  &value,
                },
@@ -114,9 +108,9 @@ func remapNetworkMetrics(
    return nil
 }

-// normalize direction normalizes the OTel direction attribute to Elastic direction.
+// normalizeDirection normalizes the OTel direction attribute to Elastic direction.
 func normalizeDirection(dir pcommon.Value, hostMetrics bool) string {
-   var in, out = "in", "out"
+   in, out := "in", "out"
    if hostMetrics {
        in, out = "ingress", "egress"
    }

[ChrsMark]

Do I assume this correctly that our mapped system.network.{in, out}.{bytes, packets} will be equal to the system.network.{io, packets} of OTel?

[ishleenk17]

Do I assume this correctly that our mapped system.network.{in, out}.{bytes, packets} will be equal to the system.network.{io, packets} of OTel?

    "system.network.io":      "system.network.%s.bytes",
    "system.network.packets": "system.network.%s.packets",

This is the mapping for them, where %s can be in/out.