Add monitoring.metrics-return-all-points flag

collectors/deduplicator.go

@@ -30,6 +30,10 @@ type MetricDeduplicator struct {
 	sentSignatures map[uint64]struct{}
 	logger         *slog.Logger
 
+	// includeTimestamp mixes the point end time into the signature. Needed in return-all-points mode
+	// where multiple points of the same series share labels and differ only by timestamp.
+	includeTimestamp bool
+
 	// Prometheus metrics
 	duplicatesTotal    prometheus.Counter
 	checksTotal        prometheus.Counter
@@ -83,7 +87,7 @@ func (d *MetricDeduplicator) CheckAndMark(name string, labelKeys, labelValues []
 
 	d.checksTotal.Inc()
 
-	signature := d.hashLabels(name, labelKeys, labelValues)
+	signature := d.hashLabels(name, labelKeys, labelValues, ts)
 
 	if _, exists := d.sentSignatures[signature]; exists {
 		d.duplicatesTotal.Inc()
@@ -97,7 +101,7 @@ func (d *MetricDeduplicator) CheckAndMark(name string, labelKeys, labelValues []
 }
 
 func (d *MetricDeduplicator) RevertMark(fqName string, labelKeys, labelValues []string, ts time.Time) {
-	signature := d.hashLabels(fqName, labelKeys, labelValues)
+	signature := d.hashLabels(fqName, labelKeys, labelValues, ts)
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
@@ -106,11 +110,16 @@ func (d *MetricDeduplicator) RevertMark(fqName string, labelKeys, labelValues []
 }
 
 // hashLabels calculates a hash based on FQName and sorted labels.
-func (d *MetricDeduplicator) hashLabels(fqName string, labelKeys, labelValues []string) uint64 {
+func (d *MetricDeduplicator) hashLabels(fqName string, labelKeys, labelValues []string, ts time.Time) uint64 {
 	h := hash.New()
 	h = hash.Add(h, fqName)
 	h = hash.AddByte(h, hash.SeparatorByte)
 
+	if d.includeTimestamp {
+		h = hash.Add(h, ts.Format(time.RFC3339Nano))
+		h = hash.AddByte(h, hash.SeparatorByte)
+	}
+
 	if len(labelKeys) > 0 {
 		// Create indices [0, 1, 2, ...]
 		indices := make([]int, len(labelKeys))

collectors/deduplicator_benchmark_test.go

@@ -17,6 +17,7 @@ import (
 	"log/slog"
 	"os"
 	"testing"
+	"time"
 )
 
 func BenchmarkHashLabels(b *testing.B) {
@@ -28,6 +29,6 @@ func BenchmarkHashLabels(b *testing.B) {
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		dedup.hashLabels(fqName, keys, vals)
+		dedup.hashLabels(fqName, keys, vals, time.Time{})
 	}
 }

collectors/deduplicator_test.go

@@ -59,6 +59,24 @@ func TestMetricDeduplicator_CheckAndMark(t *testing.T) {
 	assert.False(t, isDuplicate, "Call with different metric name should not be a duplicate")
 }
 
+func TestMetricDeduplicator_IncludeTimestamp(t *testing.T) {
+	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	dedup := NewMetricDeduplicator(logger, "test_project")
+	dedup.includeTimestamp = true // return-all-points mode
+
+	fqName := "test_metric"
+	labelKeys := []string{"label1", "label2"}
+	labelValues := []string{"value1", "value2"}
+	ts := time.Now()
+
+	// First point: not a duplicate.
+	assert.False(t, dedup.CheckAndMark(fqName, labelKeys, labelValues, ts))
+	// Same labels, same timestamp: duplicate.
+	assert.True(t, dedup.CheckAndMark(fqName, labelKeys, labelValues, ts))
+	// Same labels, different timestamp: NOT a duplicate when timestamp is included.
+	assert.False(t, dedup.CheckAndMark(fqName, labelKeys, labelValues, ts.Add(time.Second)))
+}
+
 func TestMetricDeduplicator_LabelOrdering(t *testing.T) {
 	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
 	dedup := NewMetricDeduplicator(logger, "test_project")

collectors/monitoring_collector.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"log/slog"
 	"math"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -61,6 +62,7 @@ type MonitoringCollector struct {
 	descriptorCache                 DescriptorCache
 	enableSystemLabels              bool
 	userLabelsOverride              bool
+	returnAllPoints                 bool
 	deduplicator                    *MetricDeduplicator
 
 	// Metrics for tracking dropped data
@@ -96,6 +98,9 @@ type MonitoringCollectorOptions struct {
 	EnableSystemLabels bool
 	// UserLabelsOverride decides if user labels should override any conflicting labels
 	UserLabelsOverride bool
+	// ReturnAllPoints decides if every data point in the requested interval is reported (each stamped with its
+	// own end time) instead of only the most recent one.
+	ReturnAllPoints bool
 }
 
 func isGoogleMetric(name string) bool {
@@ -240,9 +245,13 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv
 		descriptorCache:                 descriptorCache,
 		enableSystemLabels:              opts.EnableSystemLabels,
 		userLabelsOverride:              opts.UserLabelsOverride,
+		returnAllPoints:                 opts.ReturnAllPoints,
 		deduplicator:                    NewMetricDeduplicator(logger, projectID),
 		droppedMetricsTotal:             droppedMetricsTotal,
 	}
+	// In return-all-points mode a series' points share labels and differ only by time, so the
+	// deduplicator must include the timestamp to avoid collapsing them into one.
+	monitoringCollector.deduplicator.includeTimestamp = opts.ReturnAllPoints
 
 	return monitoringCollector, nil
 }
@@ -554,61 +563,92 @@ func (c *MonitoringCollector) reportTimeSeriesMetrics(
 			continue
 		}
 
-		// Check for duplicate metrics using deduplicator
-		if c.deduplicator.CheckAndMark(timeSeries.Metric.Type, labelKeys, labelValues, newestEndTime) {
-			continue // Duplicate detected and logged by deduplicator
+		// By default only the newest point is reported. In return-all-points mode every point in the
+		// interval is reported, each stamped with its own end time, ordered oldest-first so DELTA
+		// aggregation (which only accumulates strictly-newer samples) sums them in chronological order.
+		type reportPoint struct {
+			point   *monitoring.Point
+			endTime time.Time
+		}
+		var pointsToReport []reportPoint
+		if c.returnAllPoints {
+			for _, p := range timeSeries.Points {
+				if p == nil {
+					continue
+				}
+				endTime, err := time.Parse(time.RFC3339Nano, p.Interval.EndTime)
+				if err != nil {
+					return fmt.Errorf("Error parsing TimeSeries Point interval end time `%s`: %s", p.Interval.EndTime, err)
+				}
+				pointsToReport = append(pointsToReport, reportPoint{p, endTime})
+			}
+			sort.Slice(pointsToReport, func(i, j int) bool {
+				return pointsToReport[i].endTime.Before(pointsToReport[j].endTime)
+			})
+		} else if newestTSPoint != nil {
+			pointsToReport = []reportPoint{{newestTSPoint, newestEndTime}}
 		}
 
-		switch timeSeries.ValueType {
-		case "BOOL":
-			metricValue = 0
-			if *newestTSPoint.Value.BoolValue {
-				metricValue = 1
+		for _, rp := range pointsToReport {
+			tsPoint := rp.point
+			pointEndTime := rp.endTime
+
+			// Check for duplicate metrics using deduplicator
+			if c.deduplicator.CheckAndMark(timeSeries.Metric.Type, labelKeys, labelValues, pointEndTime) {
+				continue // Duplicate detected and logged by deduplicator
 			}
-		case "INT64":
-			metricValue = float64(*newestTSPoint.Value.Int64Value)
-		case "DOUBLE":
-			metricValue = *newestTSPoint.Value.DoubleValue
-		case "DISTRIBUTION":
-			dist := newestTSPoint.Value.DistributionValue
-			buckets, err := c.generateHistogramBuckets(dist)
-
-			if err == nil {
-				timeSeriesMetrics.CollectNewConstHistogram(timeSeries, newestEndTime, labelKeys, dist, buckets, labelValues, timeSeries.MetricKind)
-			} else {
-				c.deduplicator.RevertMark(timeSeries.Metric.Type, labelKeys, labelValues, newestEndTime)
+
+			switch timeSeries.ValueType {
+			case "BOOL":
+				metricValue = 0
+				if *tsPoint.Value.BoolValue {
+					metricValue = 1
+				}
+			case "INT64":
+				metricValue = float64(*tsPoint.Value.Int64Value)
+			case "DOUBLE":
+				metricValue = *tsPoint.Value.DoubleValue
+			case "DISTRIBUTION":
+				dist := tsPoint.Value.DistributionValue
+				buckets, err := c.generateHistogramBuckets(dist)
+
+				if err == nil {
+					timeSeriesMetrics.CollectNewConstHistogram(timeSeries, pointEndTime, labelKeys, dist, buckets, labelValues, timeSeries.MetricKind)
+				} else {
+					c.deduplicator.RevertMark(timeSeries.Metric.Type, labelKeys, labelValues, pointEndTime)
+					c.droppedMetricsTotal.WithLabelValues(
+						"distribution_bucket_error",
+						timeSeries.Metric.Type,
+						timeSeries.Resource.Type,
+						timeSeries.MetricKind,
+						timeSeries.ValueType,
+					).Inc()
+					c.logger.Warn("dropping distribution metric due to bucket error",
+						"metric", timeSeries.Metric.Type,
+						"resource_type", timeSeries.Resource.Type,
+						"metric_kind", timeSeries.MetricKind,
+						"err", err)
+				}
+				continue
+			default:
+				c.deduplicator.RevertMark(timeSeries.Metric.Type, labelKeys, labelValues, pointEndTime)
 				c.droppedMetricsTotal.WithLabelValues(
-					"distribution_bucket_error",
+					"unknown_value_type",
 					timeSeries.Metric.Type,
 					timeSeries.Resource.Type,
 					timeSeries.MetricKind,
 					timeSeries.ValueType,
 				).Inc()
-				c.logger.Warn("dropping distribution metric due to bucket error",
+				c.logger.Warn("dropping metric with unknown value type",
 					"metric", timeSeries.Metric.Type,
 					"resource_type", timeSeries.Resource.Type,
 					"metric_kind", timeSeries.MetricKind,
-					"err", err)
+					"value_type", timeSeries.ValueType)
+				continue
 			}
-			continue
-		default:
-			c.deduplicator.RevertMark(timeSeries.Metric.Type, labelKeys, labelValues, newestEndTime)
-			c.droppedMetricsTotal.WithLabelValues(
-				"unknown_value_type",
-				timeSeries.Metric.Type,
-				timeSeries.Resource.Type,
-				timeSeries.MetricKind,
-				timeSeries.ValueType,
-			).Inc()
-			c.logger.Warn("dropping metric with unknown value type",
-				"metric", timeSeries.Metric.Type,
-				"resource_type", timeSeries.Resource.Type,
-				"metric_kind", timeSeries.MetricKind,
-				"value_type", timeSeries.ValueType)
-			continue
-		}
 
-		timeSeriesMetrics.CollectNewConstMetric(timeSeries, newestEndTime, labelKeys, metricValueType, metricValue, labelValues, timeSeries.MetricKind)
+			timeSeriesMetrics.CollectNewConstMetric(timeSeries, pointEndTime, labelKeys, metricValueType, metricValue, labelValues, timeSeries.MetricKind)
+		}
 	}
 	timeSeriesMetrics.Complete(begun)
 	return nil

stackdriver_exporter.go

@@ -113,6 +113,10 @@ var (
 		"monitoring.metrics-ingest-delay", "Offset for the Google Stackdriver Monitoring Metrics interval into the past by the ingest delay from the metric's metadata.",
 	).Default("false").Bool()
 
+	monitoringMetricsReturnAllPoints = kingpin.Flag(
+		"monitoring.metrics-return-all-points", "Return all data points in the requested interval (each stamped with its own end time) instead of only the most recent one.",
+	).Default("false").Bool()
+
 	collectorFillMissingLabels = kingpin.Flag(
 		"collector.fill-missing-labels", "Fill missing metrics labels with empty string to avoid label dimensions inconsistent failure.",
 	).Default("true").Bool()
@@ -255,6 +259,7 @@ func (h *handler) getCollector(project string, filters map[string]bool) (*collec
 		AggregateDeltas:           *monitoringMetricsAggregateDeltas,
 		DescriptorCacheTTL:        *monitoringDescriptorCacheTTL,
 		DescriptorCacheOnlyGoogle: *monitoringDescriptorCacheOnlyGoogle,
+		ReturnAllPoints:           *monitoringMetricsReturnAllPoints,
 	}, h.logger, delta.NewInMemoryCounterStore(h.logger, *monitoringMetricsDeltasTTL), delta.NewInMemoryHistogramStore(h.logger, *monitoringMetricsDeltasTTL))
 	if err != nil {
 		return nil, err