From f0e003f16264dcdb29056591de26e54ef1be7c7c Mon Sep 17 00:00:00 2001 From: nir-sagi Date: Wed, 8 Jul 2026 12:45:33 +0000 Subject: [PATCH 1/2] Add monitoring.metrics-return-all-points flag Report every data point in the requested interval (each stamped with its own end time) instead of only the most recent one. Defaults to false, so existing behavior is unchanged. In all-points mode a series' points share labels and differ only by timestamp, so the deduplicator now mixes the point end time into its signature (gated on the new flag) to avoid collapsing them into one. Co-Authored-By: Claude Opus 4.8 (1M context) --- collectors/deduplicator.go | 15 ++- collectors/deduplicator_benchmark_test.go | 3 +- collectors/deduplicator_test.go | 18 ++++ collectors/monitoring_collector.go | 112 ++++++++++++++-------- stackdriver_exporter.go | 5 + 5 files changed, 108 insertions(+), 45 deletions(-) diff --git a/collectors/deduplicator.go b/collectors/deduplicator.go index 12d94097..393a3ddf 100644 --- a/collectors/deduplicator.go +++ b/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)) diff --git a/collectors/deduplicator_benchmark_test.go b/collectors/deduplicator_benchmark_test.go index 5f92d975..0436e5ea 100644 --- a/collectors/deduplicator_benchmark_test.go +++ b/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{}) } } diff --git a/collectors/deduplicator_test.go b/collectors/deduplicator_test.go index 2d1aea7f..fe3b4a0c 100644 --- a/collectors/deduplicator_test.go +++ b/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") diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index c6dde002..46fbebba 100644 --- a/collectors/monitoring_collector.go +++ b/collectors/monitoring_collector.go @@ -61,6 +61,7 @@ type MonitoringCollector struct { descriptorCache DescriptorCache enableSystemLabels bool userLabelsOverride bool + returnAllPoints bool deduplicator *MetricDeduplicator // Metrics for tracking dropped data @@ -96,6 +97,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 +244,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 +562,83 @@ 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. + pointsToReport := []*monitoring.Point{newestTSPoint} + if c.returnAllPoints { + pointsToReport = timeSeries.Points } - switch timeSeries.ValueType { - case "BOOL": - metricValue = 0 - if *newestTSPoint.Value.BoolValue { - metricValue = 1 + for _, tsPoint := range pointsToReport { + if tsPoint == nil { + continue } - 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) + + pointEndTime := newestEndTime + if c.returnAllPoints { + parsed, err := time.Parse(time.RFC3339Nano, tsPoint.Interval.EndTime) + if err != nil { + return fmt.Errorf("Error parsing TimeSeries Point interval end time `%s`: %s", tsPoint.Interval.EndTime, err) + } + pointEndTime = parsed + } + + // Check for duplicate metrics using deduplicator + if c.deduplicator.CheckAndMark(timeSeries.Metric.Type, labelKeys, labelValues, pointEndTime) { + continue // Duplicate detected and logged by deduplicator + } + + 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 diff --git a/stackdriver_exporter.go b/stackdriver_exporter.go index 90c33b86..e8c9bd4a 100644 --- a/stackdriver_exporter.go +++ b/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 From 55d0d83d117730c36a1dee2ea50f4d6bbc2c4ad6 Mon Sep 17 00:00:00 2001 From: nir-sagi Date: Wed, 8 Jul 2026 13:22:56 +0000 Subject: [PATCH 2/2] Sort points ascending in return-all-points mode DELTA metrics under AggregateDeltas only accumulate strictly-newer samples in the counter store. GCP returns points newest-first, so feeding them in that order made the store keep only the newest and drop the rest, silently undercounting. Sort points oldest-first so the full window accumulates in chronological order; the store's ReportTime check still prevents double-counting across overlapping scrape windows. Order is irrelevant for gauges/distributions, so this is harmless there. Co-Authored-By: Claude Opus 4.8 (1M context) --- collectors/monitoring_collector.go | 40 +++++++++++++++++++----------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index 46fbebba..f620e68a 100644 --- a/collectors/monitoring_collector.go +++ b/collectors/monitoring_collector.go @@ -18,6 +18,7 @@ import ( "fmt" "log/slog" "math" + "sort" "strings" "sync" "time" @@ -563,25 +564,34 @@ func (c *MonitoringCollector) reportTimeSeriesMetrics( } // 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. - pointsToReport := []*monitoring.Point{newestTSPoint} - if c.returnAllPoints { - pointsToReport = timeSeries.Points + // 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 } - - for _, tsPoint := range pointsToReport { - if tsPoint == nil { - continue - } - - pointEndTime := newestEndTime - if c.returnAllPoints { - parsed, err := time.Parse(time.RFC3339Nano, tsPoint.Interval.EndTime) + 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", tsPoint.Interval.EndTime, err) + return fmt.Errorf("Error parsing TimeSeries Point interval end time `%s`: %s", p.Interval.EndTime, err) } - pointEndTime = parsed + 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}} + } + + 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) {