always add timestamp even if some metrics are disabled

ishleenk17 · Feb 13, 2024 · 98e6638 · 98e6638
1 parent 6e3ef4a
commit 98e6638
Show file tree

Hide file tree

Showing 5 changed files with 99 additions and 31 deletions.
diff --git a/processor/elasticprocessor/internal/hostmetrics/cpu.go b/processor/elasticprocessor/internal/hostmetrics/cpu.go
@@ -16,14 +16,20 @@ func addCPUMetrics(metrics pmetric.MetricSlice, dataset string) error {
 		metric := metrics.At(i)
 		if metric.Name() == "system.cpu.logical.count" {
 			dp := metric.Sum().DataPoints().At(0)
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
+
 			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)
-				value := dp.DoubleValue()
+				if timestamp == 0 {
+					timestamp = dp.Timestamp()
+				}
 
+				value := dp.DoubleValue()
 				if state, ok := dp.Attributes().Get("state"); ok {
 					switch state.Str() {
 					case "idle":

diff --git a/processor/elasticprocessor/internal/hostmetrics/load.go b/processor/elasticprocessor/internal/hostmetrics/load.go
@@ -13,13 +13,21 @@ func addLoadMetrics(metrics pmetric.MetricSlice, dataset string) error {
 		metric := metrics.At(i)
 		if metric.Name() == "system.cpu.load_average.1m" {
 			dp := metric.Gauge().DataPoints().At(0)
-			timestamp = dp.Timestamp()
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
 			l1 = dp.DoubleValue()
 		} else if metric.Name() == "system.cpu.load_average.5m" {
 			dp := metric.Gauge().DataPoints().At(0)
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
 			l5 = dp.DoubleValue()
 		} else if metric.Name() == "system.cpu.load_average.15m" {
 			dp := metric.Gauge().DataPoints().At(0)
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
 			l15 = dp.DoubleValue()
 		}
 	}

diff --git a/processor/elasticprocessor/internal/hostmetrics/memory.go b/processor/elasticprocessor/internal/hostmetrics/memory.go
@@ -17,8 +17,9 @@ func addMemoryMetrics(metrics pmetric.MetricSlice, dataset string) error {
 			dataPoints := metric.Sum().DataPoints()
 			for j := 0; j < dataPoints.Len(); j++ {
 				dp := dataPoints.At(j)
-
-				timestamp = dp.Timestamp()
+				if timestamp == 0 {
+					timestamp = dp.Timestamp()
+				}
 
 				value := dp.IntValue()
 				if state, ok := dp.Attributes().Get("state"); ok {
@@ -47,8 +48,11 @@ func addMemoryMetrics(metrics pmetric.MetricSlice, dataset string) error {
 			dataPoints := metric.Gauge().DataPoints()
 			for j := 0; j < dataPoints.Len(); j++ {
 				dp := dataPoints.At(j)
-				value := dp.DoubleValue()
+				if timestamp == 0 {
+					timestamp = dp.Timestamp()
+				}
 
+				value := dp.DoubleValue()
 				if state, ok := dp.Attributes().Get("state"); ok {
 					switch state.Str() {
 					case "free":

diff --git a/processor/elasticprocessor/internal/hostmetrics/process.go b/processor/elasticprocessor/internal/hostmetrics/process.go
@@ -6,33 +6,67 @@ import (
 )
 
 func addProcessMetrics(metrics pmetric.MetricSlice, dataset string) error {
-	var timestamp, startTimestamp pcommon.Timestamp
-	var threads, memUsage, memVirtual, fdOpen, ioReadBytes, ioWriteBytes, ioReadOperations, ioWriteOperations int64
+	var timestamp pcommon.Timestamp
+	var startTime, threads, memUsage, memVirtual, fdOpen, ioReadBytes, ioWriteBytes, ioReadOperations, ioWriteOperations int64
 	var memUtil, memUtilPct, total, cpuTimeValue, systemCpuTime, userCpuTime float64
 
 	for i := 0; i < metrics.Len(); i++ {
 		metric := metrics.At(i)
 		if metric.Name() == "process.threads" {
 			dp := metric.Sum().DataPoints().At(0)
-			timestamp = dp.Timestamp()
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
+			if startTime == 0 {
+				startTime = dp.StartTimestamp().AsTime().UnixMilli()
+			}
 			threads = dp.IntValue()
 		} else if metric.Name() == "process.memory.utilization" {
 			dp := metric.Gauge().DataPoints().At(0)
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
+			if startTime == 0 {
+				startTime = dp.StartTimestamp().AsTime().UnixMilli()
+			}
 			memUtil = dp.DoubleValue()
 		} else if metric.Name() == "process.memory.usage" {
 			dp := metric.Sum().DataPoints().At(0)
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
+			if startTime == 0 {
+				startTime = dp.StartTimestamp().AsTime().UnixMilli()
+			}
 			memUsage = dp.IntValue()
 		} else if metric.Name() == "process.memory.virtual" {
 			dp := metric.Sum().DataPoints().At(0)
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
+			if startTime == 0 {
+				startTime = dp.StartTimestamp().AsTime().UnixMilli()
+			}
 			memVirtual = dp.IntValue()
 		} else if metric.Name() == "process.open_file_descriptors" {
 			dp := metric.Sum().DataPoints().At(0)
+			if timestamp == 0 {
+				timestamp = dp.Timestamp()
+			}
+			if startTime == 0 {
+				startTime = dp.StartTimestamp().AsTime().UnixMilli()
+			}
 			fdOpen = dp.IntValue()
 		} else if metric.Name() == "process.cpu.time" {
 			dataPoints := metric.Sum().DataPoints()
 			for j := 0; j < dataPoints.Len(); j++ {
 				dp := dataPoints.At(j)
-				startTimestamp = dp.StartTimestamp()
+				if timestamp == 0 {
+					timestamp = dp.Timestamp()
+				}
+				if startTime == 0 {
+					startTime = dp.StartTimestamp().AsTime().UnixMilli()
+				}
 				value := dp.DoubleValue()
 				if state, ok := dp.Attributes().Get("state"); ok {
 					switch state.Str() {
@@ -52,6 +86,12 @@ func addProcessMetrics(metrics pmetric.MetricSlice, dataset string) error {
 			dataPoints := metric.Sum().DataPoints()
 			for j := 0; j < dataPoints.Len(); j++ {
 				dp := dataPoints.At(j)
+				if timestamp == 0 {
+					timestamp = dp.Timestamp()
+				}
+				if startTime == 0 {
+					startTime = dp.StartTimestamp().AsTime().UnixMilli()
+				}
 				value := dp.IntValue()
 				if direction, ok := dp.Attributes().Get("direction"); ok {
 					switch direction.Str() {
@@ -66,6 +106,12 @@ func addProcessMetrics(metrics pmetric.MetricSlice, dataset string) error {
 			dataPoints := metric.Sum().DataPoints()
 			for j := 0; j < dataPoints.Len(); j++ {
 				dp := dataPoints.At(j)
+				if timestamp == 0 {
+					timestamp = dp.Timestamp()
+				}
+				if startTime == 0 {
+					startTime = dp.StartTimestamp().AsTime().UnixMilli()
+				}
 				value := dp.IntValue()
 				if direction, ok := dp.Attributes().Get("direction"); ok {
 					switch direction.Str() {
@@ -85,6 +131,12 @@ func addProcessMetrics(metrics pmetric.MetricSlice, dataset string) error {
 	userCpuTime = userCpuTime * 1000
 
 	addMetrics(metrics, dataset,
+		metric{
+			dataType:  Sum,
+			name:      "process.cpu.start_time",
+			timestamp: timestamp,
+			intValue:  &startTime,
+		},
 		metric{
 			dataType:  Sum,
 			name:      "system.process.num_threads",
@@ -123,32 +175,28 @@ func addProcessMetrics(metrics pmetric.MetricSlice, dataset string) error {
 			doubleValue: &memUtilPct,
 		},
 		metric{
-			dataType:       Sum,
-			name:           "system.process.cpu.total.value",
-			timestamp:      timestamp,
-			startTimestamp: startTimestamp,
-			doubleValue:    &cpuTimeValue,
+			dataType:    Sum,
+			name:        "system.process.cpu.total.value",
+			timestamp:   timestamp,
+			doubleValue: &cpuTimeValue,
 		},
 		metric{
-			dataType:       Sum,
-			name:           "system.process.cpu.system.ticks",
-			timestamp:      timestamp,
-			startTimestamp: startTimestamp,
-			doubleValue:    &systemCpuTime,
+			dataType:    Sum,
+			name:        "system.process.cpu.system.ticks",
+			timestamp:   timestamp,
+			doubleValue: &systemCpuTime,
 		},
 		metric{
-			dataType:       Sum,
-			name:           "system.process.cpu.user.ticks",
-			timestamp:      timestamp,
-			startTimestamp: startTimestamp,
-			doubleValue:    &userCpuTime,
+			dataType:    Sum,
+			name:        "system.process.cpu.user.ticks",
+			timestamp:   timestamp,
+			doubleValue: &userCpuTime,
 		},
 		metric{
-			dataType:       Sum,
-			name:           "system.process.cpu.total.ticks",
-			timestamp:      timestamp,
-			startTimestamp: startTimestamp,
-			doubleValue:    &cpuTimeValue,
+			dataType:    Sum,
+			name:        "system.process.cpu.total.ticks",
+			timestamp:   timestamp,
+			doubleValue: &cpuTimeValue,
 		},
 		metric{
 			dataType:  Sum,

diff --git a/processor/elasticprocessor/internal/hostmetrics/processes.go b/processor/elasticprocessor/internal/hostmetrics/processes.go
@@ -17,7 +17,9 @@ func addProcessSummaryMetrics(metrics pmetric.MetricSlice, dataset string) error
 			// iterate over the datapoints corresponding to different 'status' attributes
 			for j := 0; j < dataPoints.Len(); j++ {
 				dp := dataPoints.At(j)
-				timestamp = dp.Timestamp()
+				if timestamp == 0 {
+					timestamp = dp.Timestamp()
+				}
 				value := dp.IntValue()
 				if status, ok := dp.Attributes().Get("status"); ok {
 					switch status.Str() {