Network Scraper Translation to ES Metrics (open-telemetry#3)

* Network scrapper

* Add network scrapper  to hostmetrics

* Add system.network.packets

* Add more metrics to network scrapper

* Move attribute addition to the network.go file

* Add code for Network Scraper

* remove disk for now

* comment a particular part

* Have just a single metric

* Uncomment the code

* update process.go

* Improve the code of Network Scraper

* Change in network scraper code

* Update the metric.attributes

processor/elasticprocessor/internal/hostmetrics/hostmetrics.go

@@ -51,6 +51,8 @@ func AddElasticSystemMetrics(scopeMetrics pmetric.ScopeMetrics, resource pcommon
 		return addProcessMetrics(scopeMetrics.Metrics(), resource, dataset)
 	case "processes":
 		return addProcessSummaryMetrics(scopeMetrics.Metrics(), resource, dataset)
+	case "network":
+		return addNetworkMetrics(scopeMetrics.Metrics(), resource, dataset)
 	default:
 		return fmt.Errorf("no matching transform function found for scope '%s'", scope.Name())
 	}

processor/elasticprocessor/internal/hostmetrics/metric.go

@@ -19,6 +19,7 @@ type metric struct {
 	startTimestamp pcommon.Timestamp
 	intValue       *int64
 	doubleValue    *float64
+	attributes     *pcommon.Map
 }
 
 func addMetrics(ms pmetric.MetricSlice, resource pcommon.Resource, dataset string, metrics ...metric) {
@@ -47,7 +48,11 @@ func addMetrics(ms pmetric.MetricSlice, resource pcommon.Resource, dataset strin
 			dp.SetStartTimestamp(metric.startTimestamp)
 		}
 
+		if metric.attributes != nil {
+			metric.attributes.CopyTo(dp.Attributes())
+		}
 		if dataset == "system.process" {
+			// Add resource attribute as an attribute to each datapoint
 			addProcessAttributes(resource, dp)
 		}
 

processor/elasticprocessor/internal/hostmetrics/network.go

@@ -0,0 +1,65 @@
+package hostmetrics
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+func addNetworkMetrics(metrics pmetric.MetricSlice, resource pcommon.Resource, dataset string) error {
+	for i := 0; i < metrics.Len(); i++ {
+		metric := metrics.At(i)
+		dataPoints := metric.Sum().DataPoints()
+		for j := 0; j < dataPoints.Len(); j++ {
+			dp := dataPoints.At(j)
+
+			var device string
+			if d, ok := dp.Attributes().Get("device"); ok {
+				device = d.Str()
+			} else {
+				continue
+			}
+
+			if direction, ok := dp.Attributes().Get("direction"); ok {
+				name := metric.Name()
+				timestamp := dp.Timestamp()
+				value := dp.IntValue()
+
+				switch direction.Str() {
+				case "receive":
+					addDeviceMetric(metrics, resource, dataset, name, device, "in", timestamp, value)
+				case "transmit":
+					addDeviceMetric(metrics, resource, dataset, name, device, "out", timestamp, value)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func addDeviceMetric(metrics pmetric.MetricSlice, resource pcommon.Resource,
+	dataset, name, device, direction string, timestamp pcommon.Timestamp, value int64) {
+
+	metricsToAdd := map[string]string{
+		"system.network.io":      "system.network.%s.bytes",
+		"system.network.packets": "system.network.%s.packets",
+		"system.network.dropped": "system.network.%s.dropped",
+		"system.network.errors":  "system.network.%s.errors",
+	}
+
+	if metricNetworkES, ok := metricsToAdd[name]; ok {
+		attributes := pcommon.NewMap()
+		attributes.PutStr("system.network.name", device)
+
+		addMetrics(metrics, resource, dataset,
+			metric{
+				dataType:   Sum,
+				name:       fmt.Sprintf(metricNetworkES, direction),
+				timestamp:  timestamp,
+				intValue:   &value,
+				attributes: &attributes,
+			})
+	}
+}