[Feature] Add Support for Historical Data Points

docs/configuration.md

@@ -137,6 +137,11 @@ statistics:
 # Export the metric with the original CloudWatch timestamp (General Setting for all metrics in this job)
 [ addCloudwatchTimestamp: <boolean> ]
 
+# Include any metrics in the past if they are present in the CloudWatch metric response. This is useful, for example, if a metric is setup with
+# period 60s and length 300s so all the 5 data points are exposed in the metrics endpoint and not just the last one
+# (General Setting for all metrics in this job)
+[ addHistoricalMetrics: <boolean> ]
+
 # List of metric definitions
 metrics:
   [ - <metric_config> ... ]
@@ -276,6 +281,11 @@ statistics:
 # Export the metric with the original CloudWatch timestamp (General Setting for all metrics in this job)
 [ addCloudwatchTimestamp: <boolean> ]
 
+# Include any metrics in the past if they are present in the CloudWatch metric response. This is useful, for example, if a metric is setup with
+# period 60s and length 300s so all the 5 data points are exposed in the metrics endpoint and not just the last one
+# (General Setting for all metrics in this job)
+[ addHistoricalMetrics: <boolean> ]
+
 # List of metric definitions
 metrics:
   [ - <metric_config> ... ]
@@ -339,6 +349,7 @@ 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.
+Also the same applies when enabling `addHistoricalMetrics` in any metric
 
 ### `exported_tags_config`
 

pkg/clients/cloudwatch/client.go

@@ -22,7 +22,7 @@ type Client interface {
 
 	// GetMetricData returns the output of the GetMetricData CloudWatch API.
 	// Results pagination is handled automatically.
-	GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64) []MetricDataResult
+	GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64, addHistoricalMetrics bool) []MetricDataResult
 
 	// GetMetricStatistics returns the output of the GetMetricStatistics CloudWatch API.
 	GetMetricStatistics(ctx context.Context, logger logging.Logger, dimensions []*model.Dimension, namespace string, metric *model.MetricConfig) []*model.Datapoint
@@ -67,9 +67,9 @@ func (c limitedConcurrencyClient) GetMetricStatistics(ctx context.Context, logge
 	return res
 }
 
-func (c limitedConcurrencyClient) GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64) []MetricDataResult {
+func (c limitedConcurrencyClient) GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64, addHistoricalMetrics bool) []MetricDataResult {
 	c.limiter.Acquire(getMetricDataCall)
-	res := c.client.GetMetricData(ctx, logger, getMetricData, namespace, length, delay, configuredRoundingPeriod)
+	res := c.client.GetMetricData(ctx, logger, getMetricData, namespace, length, delay, configuredRoundingPeriod, addHistoricalMetrics)
 	c.limiter.Release(getMetricDataCall)
 	return res
 }

pkg/clients/cloudwatch/v1/client.go

@@ -84,7 +84,7 @@ func toModelDimensions(dimensions []*cloudwatch.Dimension) []*model.Dimension {
 	return modelDimensions
 }
 
-func (c client) GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64) []cloudwatch_client.MetricDataResult {
+func (c client) GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64, addHistoricalMetrics bool) []cloudwatch_client.MetricDataResult {
 	var resp cloudwatch.GetMetricDataOutput
 	filter := createGetMetricDataInput(getMetricData, &namespace, length, delay, configuredRoundingPeriod, logger)
 	if c.logger.IsDebugEnabled() {
@@ -109,18 +109,22 @@ func (c client) GetMetricData(ctx context.Context, logger logging.Logger, getMet
 		c.logger.Error(err, "GetMetricData error")
 		return nil
 	}
-	return toMetricDataResult(resp)
+	return toMetricDataResult(resp, addHistoricalMetrics)
 }
 
-func toMetricDataResult(resp cloudwatch.GetMetricDataOutput) []cloudwatch_client.MetricDataResult {
+func toMetricDataResult(resp cloudwatch.GetMetricDataOutput, addHistoricalMetrics 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]
+
+		for i := 0; i < len(metricDataResult.Values); i++ {
+			mappedResult := cloudwatch_client.MetricDataResult{ID: *metricDataResult.Id}
+			mappedResult.Datapoint = metricDataResult.Values[i]
+			mappedResult.Timestamp = *metricDataResult.Timestamps[i]
+			output = append(output, mappedResult)
+			if !addHistoricalMetrics {
+				break
+			}
 		}
-		output = append(output, mappedResult)
 	}
 	return output
 }

pkg/clients/cloudwatch/v1/client_test.go

@@ -84,7 +84,7 @@ func Test_toMetricDataResult(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			metricDataResults := toMetricDataResult(tc.getMetricDataOutput)
+			metricDataResults := toMetricDataResult(tc.getMetricDataOutput, false)
 			require.Equal(t, tc.expectedMetricDataResults, metricDataResults)
 		})
 	}

pkg/clients/cloudwatch/v2/client.go

@@ -87,7 +87,7 @@ func toModelDimensions(dimensions []types.Dimension) []*model.Dimension {
 	return modelDimensions
 }
 
-func (c client) GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64) []cloudwatch_client.MetricDataResult {
+func (c client) GetMetricData(ctx context.Context, logger logging.Logger, getMetricData []*model.CloudwatchData, namespace string, length int64, delay int64, configuredRoundingPeriod *int64, addHistoricalMetrics bool) []cloudwatch_client.MetricDataResult {
 	filter := createGetMetricDataInput(logger, getMetricData, &namespace, length, delay, configuredRoundingPeriod)
 	var resp cloudwatch.GetMetricDataOutput
 
@@ -115,18 +115,21 @@ func (c client) GetMetricData(ctx context.Context, logger logging.Logger, getMet
 		c.logger.Debug("GetMetricData", "output", resp)
 	}
 
-	return toMetricDataResult(resp)
+	return toMetricDataResult(resp, addHistoricalMetrics)
 }
 
-func toMetricDataResult(resp cloudwatch.GetMetricDataOutput) []cloudwatch_client.MetricDataResult {
+func toMetricDataResult(resp cloudwatch.GetMetricDataOutput, addHistoricalMetrics 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]
+		for i := 0; i < len(metricDataResult.Values); i++ {
+			mappedResult := cloudwatch_client.MetricDataResult{ID: *metricDataResult.Id}
+			mappedResult.Datapoint = &metricDataResult.Values[i]
+			mappedResult.Timestamp = metricDataResult.Timestamps[i]
+			output = append(output, mappedResult)
+			if !addHistoricalMetrics {
+				break
+			}
 		}
-		output = append(output, mappedResult)
 	}
 	return output
 }

pkg/clients/cloudwatch/v2/client_test.go

@@ -69,7 +69,7 @@ func Test_toMetricDataResult(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			metricDataResults := toMetricDataResult(tc.getMetricDataOutput)
+			metricDataResults := toMetricDataResult(tc.getMetricDataOutput, false)
 			require.Equal(t, tc.expectedMetricDataResults, metricDataResults)
 		})
 	}