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..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" @@ -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 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