Add tests

input/otlp/hostmetrics/cpu.go

@@ -1,12 +1,14 @@
 package hostmetrics
 
 import (
+	"errors"
+
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 )
 
-func remapCPUMetrics(src, out pmetric.MetricSlice, dataset string) {
-	var timestamp pcommon.Timestamp
+func remapCPUMetrics(src, out pmetric.MetricSlice, dataset string) error {
+	var timestamp pcommon.Timestamp // Use the timestamp of the last read datapoint
 	var numCores int64
 	var totalPercent, idlePercent, systemPercent, userPercent, stealPercent,
 		iowaitPercent, nicePercent, irqPercent, softirqPercent float64
@@ -15,13 +17,16 @@ func remapCPUMetrics(src, out pmetric.MetricSlice, dataset string) {
 	for i := 0; i < src.Len(); i++ {
 		metric := src.At(i)
 		if metric.Name() == "system.cpu.logical.count" {
-			dp := metric.Sum().DataPoints().At(0)
+			// Take the last datapoint for num cores as this is a cumulative metric.
+			dps := metric.Sum().DataPoints()
+			dp := dps.At(dps.Len() - 1)
 			numCores = dp.IntValue()
 			timestamp = dp.Timestamp()
 		} else if metric.Name() == "system.cpu.utilization" {
-			dataPoints := metric.Gauge().DataPoints()
-			for j := 0; j < dataPoints.Len(); j++ {
-				dp := dataPoints.At(j)
+			dps := metric.Gauge().DataPoints()
+			for j := 0; j < dps.Len(); j++ {
+				dp := dps.At(j)
+				timestamp = dp.Timestamp()
 				value := dp.DoubleValue()
 
 				if state, ok := dp.Attributes().Get("state"); ok {
@@ -55,23 +60,8 @@ func remapCPUMetrics(src, out pmetric.MetricSlice, dataset string) {
 		}
 	}
 
-	totalNorm := totalPercent / float64(numCores)
-	idleNorm := idlePercent / float64(numCores)
-	systemNorm := systemPercent / float64(numCores)
-	userNorm := userPercent / float64(numCores)
-	stealNorm := stealPercent / float64(numCores)
-	iowaitNorm := iowaitPercent / float64(numCores)
-	niceNorm := nicePercent / float64(numCores)
-	irqNorm := irqPercent / float64(numCores)
-	softirqNorm := softirqPercent / float64(numCores)
-
+	// Add all metrics that are independent of cpu logical count
 	addMetrics(out, dataset,
-		metric{
-			dataType:  pmetric.MetricTypeSum,
-			name:      "system.cpu.cores",
-			timestamp: timestamp,
-			intValue:  &numCores,
-		},
 		metric{
 			dataType:    pmetric.MetricTypeGauge,
 			name:        "system.cpu.total.pct",
@@ -126,6 +116,31 @@ func remapCPUMetrics(src, out pmetric.MetricSlice, dataset string) {
 			timestamp:   timestamp,
 			doubleValue: &softirqPercent,
 		},
+	)
+
+	// TODO (lahsivjar): Remove the dependency on the system.cpu.logical.count metric. The num of cores
+	// can be derived from the system.cpu.utilization metric using the cpu dimension.
+	if numCores == 0 {
+		return errors.New("system.cpu.logical.count metric is missing in the hostmetrics")
+	}
+
+	totalNorm := totalPercent / float64(numCores)
+	idleNorm := idlePercent / float64(numCores)
+	systemNorm := systemPercent / float64(numCores)
+	userNorm := userPercent / float64(numCores)
+	stealNorm := stealPercent / float64(numCores)
+	iowaitNorm := iowaitPercent / float64(numCores)
+	niceNorm := nicePercent / float64(numCores)
+	irqNorm := irqPercent / float64(numCores)
+	softirqNorm := softirqPercent / float64(numCores)
+
+	addMetrics(out, dataset,
+		metric{
+			dataType:  pmetric.MetricTypeSum,
+			name:      "system.cpu.cores",
+			timestamp: timestamp,
+			intValue:  &numCores,
+		},
 		metric{
 			dataType:    pmetric.MetricTypeGauge,
 			name:        "system.cpu.total.norm.pct",
@@ -181,4 +196,6 @@ func remapCPUMetrics(src, out pmetric.MetricSlice, dataset string) {
 			doubleValue: &softirqNorm,
 		},
 	)
+
+	return nil
 }

input/otlp/hostmetrics/hostmetrics.go

@@ -8,6 +8,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const scopePrefix = "otelcol/hostmetricsreceiver"
+
 var scraperToElasticDataset = map[string]string{
 	"cpu":        "system.cpu",
 	"disk":       "system.diskio",
@@ -43,21 +45,30 @@ func (r *Remapper) Remap(src pmetric.ScopeMetrics, out pmetric.MetricSlice) {
 		r.logger.Warn("no dataset defined for scraper", zap.String("scraper", scraper))
 		return
 	}
-	metrics := src.Metrics()
+
+	var err error
 	switch scraper {
 	case "cpu":
-		remapCPUMetrics(metrics, out, dataset)
+		err = remapCPUMetrics(src.Metrics(), out, dataset)
 	case "memory":
-		remapMemoryMetrics(metrics, out, dataset)
+		err = remapMemoryMetrics(src.Metrics(), out, dataset)
 	case "load":
-		remapLoadMetrics(metrics, out, dataset)
+		err = remapLoadMetrics(src.Metrics(), out, dataset)
 	case "process":
-		remapProcessMetrics(metrics, out, dataset)
+		err = remapProcessMetrics(src.Metrics(), out, dataset)
 	default:
 		r.logger.Warn("no remapper found for scope", zap.String("scope", scope.Name()))
 	}
+
+	if err != nil {
+		r.logger.Warn(
+			"failed to remap OTel hostmetrics",
+			zap.String("scope", scope.Name()),
+			zap.Error(err),
+		)
+	}
 }
 
 func (r *Remapper) Valid(sm pmetric.ScopeMetrics) bool {
-	return strings.HasPrefix(sm.Scope().Name(), "otelcol/hostmetricsreceiver")
+	return strings.HasPrefix(sm.Scope().Name(), scopePrefix)
 }