feat: reimplement history data export (#1642)

Signed-off-by: Lukas Wöhrl <[email protected]>
Co-authored-by: Kyle Eckhart <[email protected]>

Author: woehrl01

docs/configuration.md

@@ -137,6 +137,14 @@ statistics:
 # Export the metric with the original CloudWatch timestamp (General Setting for all metrics in this job)
 [ addCloudwatchTimestamp: <boolean> ]
 
+# Enables the inclusion of past metric data points from the CloudWatch response if available.
+# This is useful when a metric is configured with a 60-second period and a 300-second duration, ensuring that all
+# five data points are exposed at the metrics endpoint instead of only the latest one.
+# Note: This option requires `addCloudwatchTimestamp` to be enabled.
+# The metric destination must support out of order timestamps, see https://prometheus.io/docs/prometheus/latest/configuration/configuration/#tsdb
+# (General Setting for all metrics in this job)
+[ exportAllDataPoints: <boolean> ]
+
 # List of metric definitions
 metrics:
   [ - <metric_config> ... ]
@@ -276,6 +284,14 @@ statistics:
 # Export the metric with the original CloudWatch timestamp (General Setting for all metrics in this job)
 [ addCloudwatchTimestamp: <boolean> ]
 
+# Enables the inclusion of past metric data points from the CloudWatch response if available.
+# This is useful when a metric is configured with a 60-second period and a 300-second duration, ensuring that all
+# five data points are exposed at the metrics endpoint instead of only the latest one.
+# Note: This option requires `addCloudwatchTimestamp` to be enabled.
+# The metric destination must support out of order timestamps, see https://prometheus.io/docs/prometheus/latest/configuration/configuration/#tsdb
+# (General Setting for all metrics in this job)
+[ exportAllDataPoints: <boolean> ]
+
 # List of metric definitions
 metrics:
   [ - <metric_config> ... ]
@@ -333,12 +349,20 @@ statistics:
 
 # Export the metric with the original CloudWatch timestamp (Overrides job level setting)
 [ addCloudwatchTimestamp: <boolean> ]
+
+# Enables the inclusion of past metric data points from the CloudWatch response if available.
+# This is useful when a metric is configured with a 60-second period and a 300-second duration, ensuring that all
+# five data points are exposed at the metrics endpoint instead of only the latest one.
+# Note: This option requires `addCloudwatchTimestamp` to be enabled.
+# The metric destination must support out of order timestamps, see https://prometheus.io/docs/prometheus/latest/configuration/configuration/#tsdb
+# (General Setting for all metrics in this job)
+[ exportAllDataPoints: <boolean> ]
 ```
 
 Notes:
 - Available statistics: `Maximum`, `Minimum`, `Sum`, `SampleCount`, `Average`, `pXX` (e.g. `p90`).
 
-- Watch out using `addCloudwatchTimestamp` for sparse metrics, e.g from S3, since Prometheus won't scrape metrics containing timestamps older than 2-3 hours.
+- Watch out using `addCloudwatchTimestamp` for sparse metrics, e.g from S3, since Prometheus won't scrape metrics containing timestamps older than 2-3 hours. Also the same applies when enabling `exportAllDataPoints` in any metric.
 
 ### `exported_tags_config`
 
@@ -390,4 +414,4 @@ This is an example of the `dimensions_config` block:
 dimensions:
   - name: AutoScalingGroupName
     value: MyGroup
-```
+```

examples/historic-data.yml

@@ -0,0 +1,29 @@
+apiVersion: v1alpha1
+discovery:
+  jobs:
+    - type: AWS/SQS
+      regions:
+        - us-east-1
+      period: 60
+      length: 300
+      addCloudwatchTimestamp: true
+      exportAllDataPoints: true
+      metrics:
+        - name: NumberOfMessagesSent
+          statistics: [Sum]
+        - name: NumberOfMessagesReceived
+          statistics: [Sum]
+        - name: NumberOfMessagesDeleted
+          statistics: [Sum]
+        - name: ApproximateAgeOfOldestMessage
+          statistics: [Average]
+        - name: NumberOfEmptyReceives
+          statistics: [Sum]
+        - name: SentMessageSize
+          statistics: [Average]
+        - name: ApproximateNumberOfMessagesNotVisible
+          statistics: [Sum]
+        - name: ApproximateNumberOfMessagesDelayed
+          statistics: [Sum]
+        - name: ApproximateNumberOfMessagesVisible
+          statistics: [Sum]

pkg/clients/cloudwatch/client.go

@@ -37,7 +37,7 @@ type Client interface {
 	GetMetricData(ctx context.Context, getMetricData []*model.CloudwatchData, namespace string, startTime time.Time, endTime time.Time) []MetricDataResult
 
 	// GetMetricStatistics returns the output of the GetMetricStatistics CloudWatch API.
-	GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.Datapoint
+	GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.MetricStatisticsResult
 }
 
 // ConcurrencyLimiter limits the concurrency when calling AWS CloudWatch APIs. The functions implemented
@@ -55,9 +55,12 @@ type ConcurrencyLimiter interface {
 }
 
 type MetricDataResult struct {
-	ID string
-	// A nil datapoint is a marker for no datapoint being found
-	Datapoint *float64
+	ID         string
+	DataPoints []DataPoint
+}
+
+type DataPoint struct {
+	Value     *float64
 	Timestamp time.Time
 }
 
@@ -73,7 +76,7 @@ func NewLimitedConcurrencyClient(client Client, limiter ConcurrencyLimiter) Clie
 	}
 }
 
-func (c limitedConcurrencyClient) GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.Datapoint {
+func (c limitedConcurrencyClient) GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.MetricStatisticsResult {
 	c.limiter.Acquire(getMetricStatisticsCall)
 	res := c.client.GetMetricStatistics(ctx, logger, dimensions, namespace, metric)
 	c.limiter.Release(getMetricStatisticsCall)

pkg/clients/cloudwatch/v1/client.go

@@ -95,6 +95,7 @@ func toModelDimensions(dimensions []*cloudwatch.Dimension) []model.Dimension {
 
 func (c client) GetMetricData(ctx context.Context, getMetricData []*model.CloudwatchData, namespace string, startTime time.Time, endTime time.Time) []cloudwatch_client.MetricDataResult {
 	metricDataQueries := make([]*cloudwatch.MetricDataQuery, 0, len(getMetricData))
+	exportAllDataPoints := false
 	for _, data := range getMetricData {
 		metricStat := &cloudwatch.MetricStat{
 			Metric: &cloudwatch.Metric{
@@ -110,6 +111,7 @@ func (c client) GetMetricData(ctx context.Context, getMetricData []*model.Cloudw
 			MetricStat: metricStat,
 			ReturnData: aws.Bool(true),
 		})
+		exportAllDataPoints = exportAllDataPoints || data.MetricMigrationParams.ExportAllDataPoints
 	}
 	input := &cloudwatch.GetMetricDataInput{
 		EndTime:           &endTime,
@@ -137,23 +139,31 @@ func (c client) GetMetricData(ctx context.Context, getMetricData []*model.Cloudw
 		c.logger.Error("GetMetricData error", "err", err)
 		return nil
 	}
-	return toMetricDataResult(resp)
+	return toMetricDataResult(resp, exportAllDataPoints)
 }
 
-func toMetricDataResult(resp cloudwatch.GetMetricDataOutput) []cloudwatch_client.MetricDataResult {
+func toMetricDataResult(resp cloudwatch.GetMetricDataOutput, exportAllDataPoints bool) []cloudwatch_client.MetricDataResult {
 	output := make([]cloudwatch_client.MetricDataResult, 0, len(resp.MetricDataResults))
 	for _, metricDataResult := range resp.MetricDataResults {
-		mappedResult := cloudwatch_client.MetricDataResult{ID: *metricDataResult.Id}
-		if len(metricDataResult.Values) > 0 {
-			mappedResult.Datapoint = metricDataResult.Values[0]
-			mappedResult.Timestamp = *metricDataResult.Timestamps[0]
+		mappedResult := cloudwatch_client.MetricDataResult{
+			ID:         *metricDataResult.Id,
+			DataPoints: make([]cloudwatch_client.DataPoint, 0, len(metricDataResult.Timestamps))}
+		for i := 0; i < len(metricDataResult.Timestamps); i++ {
+			mappedResult.DataPoints = append(mappedResult.DataPoints, cloudwatch_client.DataPoint{
+				Value:     metricDataResult.Values[i],
+				Timestamp: *metricDataResult.Timestamps[i],
+			})
+
+			if !exportAllDataPoints {
+				break
+			}
 		}
 		output = append(output, mappedResult)
 	}
 	return output
 }
 
-func (c client) GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.Datapoint {
+func (c client) GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.MetricStatisticsResult {
 	filter := createGetMetricStatisticsInput(dimensions, &namespace, metric, logger)
 
 	c.logger.Debug("GetMetricStatistics", "input", filter)
@@ -171,14 +181,14 @@ func (c client) GetMetricStatistics(ctx context.Context, logger *slog.Logger, di
 		return nil
 	}
 
-	return toModelDatapoints(resp.Datapoints)
+	return toModelDataPoints(resp.Datapoints)
 }
 
-func toModelDatapoints(cwDatapoints []*cloudwatch.Datapoint) []*model.Datapoint {
-	modelDataPoints := make([]*model.Datapoint, 0, len(cwDatapoints))
+func toModelDataPoints(cwDataPoints []*cloudwatch.Datapoint) []*model.MetricStatisticsResult {
+	modelDataPoints := make([]*model.MetricStatisticsResult, 0, len(cwDataPoints))
 
-	for _, cwDatapoint := range cwDatapoints {
-		modelDataPoints = append(modelDataPoints, &model.Datapoint{
+	for _, cwDatapoint := range cwDataPoints {
+		modelDataPoints = append(modelDataPoints, &model.MetricStatisticsResult{
 			Average:            cwDatapoint.Average,
 			ExtendedStatistics: cwDatapoint.ExtendedStatistics,
 			Maximum:            cwDatapoint.Maximum,

pkg/clients/cloudwatch/v1/client_test.go

@@ -47,11 +47,13 @@ func Test_toMetricDataResult(t *testing.T) {
 		name                      string
 		getMetricDataOutput       cloudwatch.GetMetricDataOutput
 		expectedMetricDataResults []cloudwatch_client.MetricDataResult
+		exportAllDataPoints       bool
 	}
 
 	testCases := []testCase{
 		{
-			name: "all metrics present",
+			name:                "all metrics present",
+			exportAllDataPoints: false,
 			getMetricDataOutput: cloudwatch.GetMetricDataOutput{
 				MetricDataResults: []*cloudwatch.MetricDataResult{
 					{
@@ -67,12 +69,21 @@ func Test_toMetricDataResult(t *testing.T) {
 				},
 			},
 			expectedMetricDataResults: []cloudwatch_client.MetricDataResult{
-				{ID: "metric-1", Datapoint: aws.Float64(1.0), Timestamp: ts.Add(10 * time.Minute)},
-				{ID: "metric-2", Datapoint: aws.Float64(2.0), Timestamp: ts},
+				{
+					ID: "metric-1", DataPoints: []cloudwatch_client.DataPoint{
+						{Value: aws.Float64(1.0), Timestamp: ts.Add(10 * time.Minute)},
+					},
+				},
+				{
+					ID: "metric-2", DataPoints: []cloudwatch_client.DataPoint{
+						{Value: aws.Float64(2.0), Timestamp: ts},
+					},
+				},
 			},
 		},
 		{
-			name: "metric with no values",
+			name:                "metric with no values",
+			exportAllDataPoints: false,
 			getMetricDataOutput: cloudwatch.GetMetricDataOutput{
 				MetricDataResults: []*cloudwatch.MetricDataResult{
 					{
@@ -88,15 +99,54 @@ func Test_toMetricDataResult(t *testing.T) {
 				},
 			},
 			expectedMetricDataResults: []cloudwatch_client.MetricDataResult{
-				{ID: "metric-1", Datapoint: aws.Float64(1.0), Timestamp: ts.Add(10 * time.Minute)},
-				{ID: "metric-2", Datapoint: nil, Timestamp: time.Time{}},
+				{
+					ID: "metric-1", DataPoints: []cloudwatch_client.DataPoint{
+						{Value: aws.Float64(1.0), Timestamp: ts.Add(10 * time.Minute)},
+					},
+				},
+				{
+					ID:         "metric-2",
+					DataPoints: []cloudwatch_client.DataPoint{},
+				},
+			},
+		},
+		{
+			name:                "export all data points",
+			exportAllDataPoints: true,
+			getMetricDataOutput: cloudwatch.GetMetricDataOutput{
+				MetricDataResults: []*cloudwatch.MetricDataResult{
+					{
+						Id:         aws.String("metric-1"),
+						Values:     []*float64{aws.Float64(1.0), aws.Float64(2.0), aws.Float64(3.0)},
+						Timestamps: []*time.Time{aws.Time(ts.Add(10 * time.Minute)), aws.Time(ts.Add(5 * time.Minute)), aws.Time(ts)},
+					},
+					{
+						Id:         aws.String("metric-2"),
+						Values:     []*float64{aws.Float64(2.0)},
+						Timestamps: []*time.Time{aws.Time(ts)},
+					},
+				},
+			},
+			expectedMetricDataResults: []cloudwatch_client.MetricDataResult{
+				{
+					ID: "metric-1", DataPoints: []cloudwatch_client.DataPoint{
+						{Value: aws.Float64(1.0), Timestamp: ts.Add(10 * time.Minute)},
+						{Value: aws.Float64(2.0), Timestamp: ts.Add(5 * time.Minute)},
+						{Value: aws.Float64(3.0), Timestamp: ts},
+					},
+				},
+				{
+					ID: "metric-2", DataPoints: []cloudwatch_client.DataPoint{
+						{Value: aws.Float64(2.0), Timestamp: ts},
+					},
+				},
 			},
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			metricDataResults := toMetricDataResult(tc.getMetricDataOutput)
+			metricDataResults := toMetricDataResult(tc.getMetricDataOutput, tc.exportAllDataPoints)
 			require.Equal(t, tc.expectedMetricDataResults, metricDataResults)
 		})
 	}

pkg/clients/cloudwatch/v2/client.go

@@ -98,6 +98,7 @@ func toModelDimensions(dimensions []types.Dimension) []model.Dimension {
 
 func (c client) GetMetricData(ctx context.Context, getMetricData []*model.CloudwatchData, namespace string, startTime time.Time, endTime time.Time) []cloudwatch_client.MetricDataResult {
 	metricDataQueries := make([]types.MetricDataQuery, 0, len(getMetricData))
+	exportAllDataPoints := false
 	for _, data := range getMetricData {
 		metricStat := &types.MetricStat{
 			Metric: &types.Metric{
@@ -113,6 +114,7 @@ func (c client) GetMetricData(ctx context.Context, getMetricData []*model.Cloudw
 			MetricStat: metricStat,
 			ReturnData: aws.Bool(true),
 		})
+		exportAllDataPoints = exportAllDataPoints || data.MetricMigrationParams.ExportAllDataPoints
 	}
 
 	input := &cloudwatch.GetMetricDataInput{
@@ -143,23 +145,32 @@ func (c client) GetMetricData(ctx context.Context, getMetricData []*model.Cloudw
 
 	c.logger.Debug("GetMetricData", "output", resp)
 
-	return toMetricDataResult(resp)
+	return toMetricDataResult(resp, exportAllDataPoints)
 }
 
-func toMetricDataResult(resp cloudwatch.GetMetricDataOutput) []cloudwatch_client.MetricDataResult {
+func toMetricDataResult(resp cloudwatch.GetMetricDataOutput, exportAllDataPoints bool) []cloudwatch_client.MetricDataResult {
 	output := make([]cloudwatch_client.MetricDataResult, 0, len(resp.MetricDataResults))
 	for _, metricDataResult := range resp.MetricDataResults {
-		mappedResult := cloudwatch_client.MetricDataResult{ID: *metricDataResult.Id}
-		if len(metricDataResult.Values) > 0 {
-			mappedResult.Datapoint = &metricDataResult.Values[0]
-			mappedResult.Timestamp = metricDataResult.Timestamps[0]
+		mappedResult := cloudwatch_client.MetricDataResult{
+			ID:         *metricDataResult.Id,
+			DataPoints: make([]cloudwatch_client.DataPoint, 0, len(metricDataResult.Timestamps)),
+		}
+		for i := 0; i < len(metricDataResult.Timestamps); i++ {
+			mappedResult.DataPoints = append(mappedResult.DataPoints, cloudwatch_client.DataPoint{
+				Value:     &metricDataResult.Values[i],
+				Timestamp: metricDataResult.Timestamps[i],
+			})
+
+			if !exportAllDataPoints {
+				break
+			}
 		}
 		output = append(output, mappedResult)
 	}
 	return output
 }
 
-func (c client) GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.Datapoint {
+func (c client) GetMetricStatistics(ctx context.Context, logger *slog.Logger, dimensions []model.Dimension, namespace string, metric *model.MetricConfig) []*model.MetricStatisticsResult {
 	filter := createGetMetricStatisticsInput(logger, dimensions, &namespace, metric)
 	c.logger.Debug("GetMetricStatistics", "input", filter)
 
@@ -181,18 +192,18 @@ func (c client) GetMetricStatistics(ctx context.Context, logger *slog.Logger, di
 		ptrs = append(ptrs, &datapoint)
 	}
 
-	return toModelDatapoints(ptrs)
+	return toModelDataPoints(ptrs)
 }
 
-func toModelDatapoints(cwDatapoints []*types.Datapoint) []*model.Datapoint {
-	modelDataPoints := make([]*model.Datapoint, 0, len(cwDatapoints))
+func toModelDataPoints(cwDataPoints []*types.Datapoint) []*model.MetricStatisticsResult {
+	modelDataPoints := make([]*model.MetricStatisticsResult, 0, len(cwDataPoints))
 
-	for _, cwDatapoint := range cwDatapoints {
+	for _, cwDatapoint := range cwDataPoints {
 		extendedStats := make(map[string]*float64, len(cwDatapoint.ExtendedStatistics))
 		for name, value := range cwDatapoint.ExtendedStatistics {
 			extendedStats[name] = &value
 		}
-		modelDataPoints = append(modelDataPoints, &model.Datapoint{
+		modelDataPoints = append(modelDataPoints, &model.MetricStatisticsResult{
 			Average:            cwDatapoint.Average,
 			ExtendedStatistics: extendedStats,
 			Maximum:            cwDatapoint.Maximum,