[Feature] Add Support for Historical Data Points

docs/configuration.md

@@ -126,6 +126,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> ... ]
@@ -267,6 +272,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> ... ]
@@ -330,6 +340,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

@@ -17,7 +17,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 *config.Metric) []*model.Datapoint
@@ -48,9 +48,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.sem <- struct{}{}
-	res := c.client.GetMetricData(ctx, logger, getMetricData, namespace, length, delay, configuredRoundingPeriod)
+	res := c.client.GetMetricData(ctx, logger, getMetricData, namespace, length, delay, configuredRoundingPeriod, addHistoricalMetrics)
 	<-c.sem
 	return res
 }

pkg/clients/cloudwatch/v1/client.go

@@ -89,7 +89,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() {
@@ -113,18 +113,21 @@ 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/v2/client.go

@@ -90,7 +90,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
 
@@ -117,18 +117,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/v2/cache_test.go

@@ -461,7 +461,7 @@ func (t testClient) ListMetrics(_ context.Context, _ string, _ *config.Metric, _
 	return nil, nil
 }
 
-func (t testClient) GetMetricData(_ context.Context, _ logging.Logger, _ []*model.CloudwatchData, _ string, _ int64, _ int64, _ *int64) []*cloudwatch_client.MetricDataResult {
+func (t testClient) GetMetricData(_ context.Context, _ logging.Logger, _ []*model.CloudwatchData, _ string, _ int64, _ int64, _ *int64, _ bool) []*cloudwatch_client.MetricDataResult {
 	return nil
 }
 

pkg/config/config.go

@@ -33,6 +33,7 @@ type JobLevelMetricFields struct {
 	Delay                  int64    `yaml:"delay"`
 	NilToZero              *bool    `yaml:"nilToZero"`
 	AddCloudwatchTimestamp *bool    `yaml:"addCloudwatchTimestamp"`
+	AddHistoricalMetrics   *bool    `yaml:"addHistoricalMetrics"`
 }
 
 type Job struct {

pkg/job/custom.go

@@ -40,6 +40,11 @@ func runCustomNamespaceJob(
 
 	wg.Add(partition)
 
+	var addHistoricalMetrics bool
+	if job.AddHistoricalMetrics != nil {
+		addHistoricalMetrics = *job.AddHistoricalMetrics
+	}
+
 	for i := 0; i < metricDataLength; i += maxMetricCount {
 		go func(i int) {
 			defer wg.Done()
@@ -49,7 +54,7 @@ func runCustomNamespaceJob(
 				end = metricDataLength
 			}
 			input := getMetricDatas[i:end]
-			data := clientCloudwatch.GetMetricData(ctx, logger, input, job.Namespace, length, job.Delay, job.RoundingPeriod)
+			data := clientCloudwatch.GetMetricData(ctx, logger, input, job.Namespace, length, job.Delay, job.RoundingPeriod, addHistoricalMetrics)
 
 			if data != nil {
 				output := make([]*model.CloudwatchData, 0)

pkg/job/discovery.go

@@ -68,6 +68,11 @@ func runDiscoveryJob(
 	getMetricDataOutput := make([][]*cloudwatch.MetricDataResult, partition)
 	count := 0
 
+	var addHistoricalMetrics bool
+	if job.AddHistoricalMetrics != nil {
+		addHistoricalMetrics = *job.AddHistoricalMetrics
+	}
+
 	for i := 0; i < metricDataLength; i += maxMetricCount {
 		go func(i, n int) {
 			defer wg.Done()
@@ -76,7 +81,7 @@ func runDiscoveryJob(
 				end = metricDataLength
 			}
 			input := getMetricDatas[i:end]
-			data := clientCloudwatch.GetMetricData(ctx, logger, input, svc.Namespace, length, job.Delay, job.RoundingPeriod)
+			data := clientCloudwatch.GetMetricData(ctx, logger, input, svc.Namespace, length, job.Delay, job.RoundingPeriod, addHistoricalMetrics)
 			if data != nil {
 				getMetricDataOutput[n] = data
 			} else {
@@ -98,15 +103,34 @@ func runDiscoveryJob(
 		if data == nil {
 			continue
 		}
+		previousIdx := -1
+		previousID := ""
 		for _, metricDataResult := range data {
 			idx := findGetMetricDataByID(getMetricDatas, *metricDataResult.ID)
+			// TODO: This logic needs to be guarded by a feature flag! Also, remember to add compatibility in the client v2
 			if idx == -1 {
-				logger.Warn("GetMetricData returned unknown metric ID", "metric_id", *metricDataResult.ID)
+				if addHistoricalMetrics {
+					// Use the previousIdx to make a copy
+					if previousIdx != -1 && previousID == *metricDataResult.ID {
+						// Create a new CloudwatchData object
+						newData := *getMetricDatas[previousIdx]
+						newData.GetMetricDataPoint = metricDataResult.Datapoint
+						newData.GetMetricDataTimestamps = metricDataResult.Timestamp
+
+						getMetricDatas = append(getMetricDatas, &newData)
+					} else {
+						logger.Warn("GetMetricData returned unknown metric ID", "metric_id", *metricDataResult.ID)
+					}
+				} else {
+					logger.Warn("GetMetricData returned unknown metric ID", "metric_id", *metricDataResult.ID)
+				}
 				continue
 			}
 			getMetricDatas[idx].GetMetricDataPoint = metricDataResult.Datapoint
 			getMetricDatas[idx].GetMetricDataTimestamps = metricDataResult.Timestamp
 			getMetricDatas[idx].MetricID = nil // mark as processed
+			previousIdx = idx
+			previousID = *metricDataResult.ID
 		}
 	}
 

pkg/job/discovery_test.go

@@ -90,6 +90,7 @@ func Test_getFilteredMetricDatas(t *testing.T) {
 				{
 					AccountID:              aws.String("123123123123"),
 					AddCloudwatchTimestamp: aws.Bool(false),
+					AddHistoricalMetrics:   aws.Bool(false),
 					Dimensions: []*model.Dimension{
 						{
 							Name:  "FileSystemId",
@@ -177,6 +178,7 @@ func Test_getFilteredMetricDatas(t *testing.T) {
 				{
 					AccountID:              aws.String("123123123123"),
 					AddCloudwatchTimestamp: aws.Bool(false),
+					AddHistoricalMetrics:   aws.Bool(false),
 					Dimensions: []*model.Dimension{
 						{
 							Name:  "InstanceId",
@@ -260,6 +262,7 @@ func Test_getFilteredMetricDatas(t *testing.T) {
 				{
 					AccountID:              aws.String("123123123123"),
 					AddCloudwatchTimestamp: aws.Bool(false),
+					AddHistoricalMetrics:   aws.Bool(false),
 					Dimensions: []*model.Dimension{
 						{
 							Name:  "Cluster Name",
@@ -385,6 +388,7 @@ func Test_getFilteredMetricDatas(t *testing.T) {
 				{
 					AccountID:              aws.String("123123123123"),
 					AddCloudwatchTimestamp: aws.Bool(false),
+					AddHistoricalMetrics:   aws.Bool(false),
 					Dimensions: []*model.Dimension{
 						{
 							Name:  "LoadBalancer",
@@ -435,6 +439,9 @@ func Test_getFilteredMetricDatas(t *testing.T) {
 				if *got.AddCloudwatchTimestamp != *tt.wantGetMetricsData[i].AddCloudwatchTimestamp {
 					t.Errorf("getFilteredMetricDatas().AddCloudwatchTimestamp = %v, want %v", *got.AddCloudwatchTimestamp, *tt.wantGetMetricsData[i].AddCloudwatchTimestamp)
 				}
+				if *got.AddHistoricalMetrics != *tt.wantGetMetricsData[i].AddHistoricalMetrics {
+					t.Errorf("getFilteredMetricDatas().AddHistoricalMetrics = %v, want %v", *got.AddHistoricalMetrics, *tt.wantGetMetricsData[i].AddHistoricalMetrics)
+				}
 				if *got.NilToZero != *tt.wantGetMetricsData[i].NilToZero {
 					t.Errorf("getFilteredMetricDatas().NilToZero = %v, want %v", *got.NilToZero, *tt.wantGetMetricsData[i].NilToZero)
 				}

pkg/model/model.go

@@ -70,6 +70,7 @@ type CloudwatchData struct {
 	GetMetricDataTimestamps *time.Time
 	NilToZero               *bool
 	AddCloudwatchTimestamp  *bool
+	AddHistoricalMetrics    *bool
 	CustomTags              []Tag
 	Tags                    []Tag
 	Dimensions              []*Dimension

pkg/promutil/migrate.go

@@ -233,7 +233,9 @@ func EnsureLabelConsistencyAndRemoveDuplicates(metrics []*PrometheusMetric, obse
 			}
 		}
 
-		metricKey := fmt.Sprintf("%s-%d", *metric.Name, prom_model.LabelsToSignature(metric.Labels))
+		// Include the timestamp to avoid genuine duplicates!? At this point we have all the metrics to be exposed under the `/metrics` endpoint so
+		// we aren't able to tell if some of the metrics are present because the `addHistoricalMetrics` is set to `true`!?
+		metricKey := fmt.Sprintf("%s-%d-%d", *metric.Name, prom_model.LabelsToSignature(metric.Labels), metric.Timestamp.Unix())
 		if _, exists := metricKeys[metricKey]; !exists {
 			metricKeys[metricKey] = struct{}{}
 			output = append(output, metric)