Refactor serializer sketches support

comp/dogstatsd/http/impl/sketch_iterator.go

@@ -13,25 +13,40 @@ import (
 	pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/dogstatsdhttp"
 )
 
-// sketchData implements metrics.SketchData from reader-provided sketch columns and summary.
+// sketchData holds one sketch point's reader-provided columns and summary.
 type sketchData struct {
+	ts                 int64
 	k                  []int32
 	n                  []uint32
 	cnt                int64
 	min, max, sum, avg float64
 }
 
-func (s *sketchData) Cols() ([]int32, []uint32) {
-	return s.k, s.n
+type dogstatsdSketchSeries struct {
+	metrics.DistributionMetadata
+	Points []sketchData
 }
 
-func (s *sketchData) BasicStats() (int64, float64, float64, float64, float64) {
-	return s.cnt, s.min, s.max, s.sum, s.avg
+// GetName returns the metric name.
+func (s *dogstatsdSketchSeries) GetName() string {
+	return s.Name
+}
+
+// WriteTo emits the buffered points to the writer's DDSketch flavor. May be
+// called multiple times on the same value; iteration always starts over.
+func (s *dogstatsdSketchSeries) WriteTo(w metrics.DistributionWriter) error {
+	return w.WriteDDSketch(s.DistributionMetadata, len(s.Points), s)
+}
+
+// GetDDSketchPoint returns the buffered sketch point at index i.
+func (s *dogstatsdSketchSeries) GetDDSketchPoint(i int) (ts, cnt int64, min, max, sum, avg float64, k []int32, n []uint32) {
+	p := s.Points[i]
+	return p.ts, p.cnt, p.min, p.max, p.sum, p.avg, p.k, p.n
 }
 
 type sketchIterator struct {
 	iteratorCommon
-	buffer metrics.SketchSeries
+	buffer dogstatsdSketchSeries
 }
 
 func newSketchIterator(payload *pb.Payload, origin origin, hostname string) (*sketchIterator, error) {
@@ -92,25 +107,24 @@ func (it *sketchIterator) MoveNext() bool {
 		if cnt > 0 {
 			avg = sum / float64(cnt)
 		}
-		b.Points = append(b.Points, metrics.SketchPoint{
-			Ts: it.reader.Timestamp(),
-			Sketch: &sketchData{
+		b.Points = append(b.Points,
+			sketchData{
+				ts:  it.reader.Timestamp(),
 				k:   k,
 				n:   n,
 				cnt: int64(cnt),
 				min: min,
 				max: max,
 				sum: sum,
 				avg: avg,
-			},
-		})
+			})
 	}
 
 	return true
 }
 
 // Current returns the internal sketch series buffer, populated by MoveNext.
-func (it *sketchIterator) Current() *metrics.SketchSeries {
+func (it *sketchIterator) Current() metrics.Distribution {
 	return &it.buffer
 }
 

comp/dogstatsd/http/impl/sketch_iterator_test.go

@@ -48,53 +48,47 @@ func TestSketchIterator(t *testing.T) {
 	require.NotNil(t, it)
 
 	require.True(t, it.MoveNext())
-	s := it.Current()
+	s := it.Current().(*dogstatsdSketchSeries)
 	require.Equal(t, "foo", s.Name)
 	require.Equal(t, "default", s.Host)
 	require.Equal(t, metrics.MetricSourceDogstatsd, s.Source)
 	require.Len(t, s.Points, 2)
 
 	pt := s.Points[0]
-	require.Equal(t, int64(1000), pt.Ts)
-	k, n := pt.Sketch.Cols()
-	require.Equal(t, []int32{0, 2}, k)
-	require.Equal(t, []uint32{1, 1}, n)
-	cnt, min, max, sum, avg := pt.Sketch.BasicStats()
-	require.Equal(t, int64(2), cnt)
-	require.Equal(t, 1.0, min)
-	require.Equal(t, 3.0, max)
-	require.Equal(t, 4.0, sum)
-	require.Equal(t, 2.0, avg)
+	require.Equal(t, int64(1000), pt.ts)
+	require.Equal(t, []int32{0, 2}, pt.k)
+	require.Equal(t, []uint32{1, 1}, pt.n)
+	require.Equal(t, int64(2), pt.cnt)
+	require.Equal(t, 1.0, pt.min)
+	require.Equal(t, 3.0, pt.max)
+	require.Equal(t, 4.0, pt.sum)
+	require.Equal(t, 2.0, pt.avg)
 
 	pt = s.Points[1]
-	require.Equal(t, int64(2000), pt.Ts)
-	k, n = pt.Sketch.Cols()
-	require.Equal(t, []int32{1}, k)
-	require.Equal(t, []uint32{3}, n)
-	cnt, min, max, sum, avg = pt.Sketch.BasicStats()
-	require.Equal(t, int64(3), cnt)
-	require.Equal(t, 2.0, min)
-	require.Equal(t, 4.0, max)
-	require.Equal(t, 6.0, sum)
-	require.Equal(t, 2.0, avg)
+	require.Equal(t, int64(2000), pt.ts)
+	require.Equal(t, []int32{1}, pt.k)
+	require.Equal(t, []uint32{3}, pt.n)
+	require.Equal(t, int64(3), pt.cnt)
+	require.Equal(t, 2.0, pt.min)
+	require.Equal(t, 4.0, pt.max)
+	require.Equal(t, 6.0, pt.sum)
+	require.Equal(t, 2.0, pt.avg)
 
 	require.True(t, it.MoveNext())
-	s = it.Current()
+	s = it.Current().(*dogstatsdSketchSeries)
 	require.Equal(t, "bar", s.Name)
 	require.Equal(t, "default", s.Host)
 	require.Len(t, s.Points, 1)
 
 	pt = s.Points[0]
-	require.Equal(t, int64(3000), pt.Ts)
-	k, n = pt.Sketch.Cols()
-	require.Empty(t, k)
-	require.Empty(t, n)
-	cnt, min, max, sum, avg = pt.Sketch.BasicStats()
-	require.Equal(t, int64(5), cnt)
-	require.Equal(t, 0.0, min)
-	require.Equal(t, 0.0, max)
-	require.Equal(t, 0.0, sum)
-	require.Equal(t, 0.0, avg) // sum=0/cnt=5=0
+	require.Equal(t, int64(3000), pt.ts)
+	require.Empty(t, pt.k)
+	require.Empty(t, pt.n)
+	require.Equal(t, int64(5), pt.cnt)
+	require.Equal(t, 0.0, pt.min)
+	require.Equal(t, 0.0, pt.max)
+	require.Equal(t, 0.0, pt.sum)
+	require.Equal(t, 0.0, pt.avg) // sum=0/cnt=5=0
 
 	require.False(t, it.MoveNext())
 	require.NoError(t, it.err)
@@ -136,12 +130,12 @@ func TestSketchIteratorTagMerging(t *testing.T) {
 	require.NoError(t, err)
 
 	require.True(t, it.MoveNext())
-	s := it.Current()
+	s := it.Current().(*dogstatsdSketchSeries)
 	require.Equal(t, "foo", s.Name)
 	require.Equal(t, tagset.NewCompositeTags([]string{"low", "orch", "high"}, []string{"ook"}), s.Tags)
 
 	require.True(t, it.MoveNext())
-	s = it.Current()
+	s = it.Current().(*dogstatsdSketchSeries)
 	require.Equal(t, "bar", s.Name)
 	require.Equal(t, tagset.NewCompositeTags([]string{"low"}, []string{"ook"}), s.Tags)
 
@@ -177,7 +171,7 @@ func TestSketchIteratorHostOverride(t *testing.T) {
 	it, err := newSketchIterator(payload, origin, "default")
 	require.NoError(t, err)
 	require.True(t, it.MoveNext())
-	s := it.Current()
+	s := it.Current().(*dogstatsdSketchSeries)
 	require.Equal(t, "foo", s.Name)
 	require.Equal(t, "", s.Host)
 	require.False(t, it.MoveNext())

comp/otelcol/otlp/components/exporter/serializerexporter/consumer.go

@@ -142,15 +142,17 @@ func (c *serializerConsumer) ConsumeSketch(_ context.Context, dimensions *otlpme
 		msrc = metrics.MetricSourceOpenTelemetryCollectorUnknown
 	}
 	c.sketches = append(c.sketches, &metrics.SketchSeries{
-		Name:     dimensions.Name(),
-		Tags:     tagset.CompositeTagsFromSlice(enrichTags(c.extraTags, dimensions)),
-		Host:     dimensions.Host(),
-		Interval: interval,
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name:     dimensions.Name(),
+			Tags:     tagset.CompositeTagsFromSlice(enrichTags(c.extraTags, dimensions)),
+			Host:     dimensions.Host(),
+			Interval: interval,
+			Source:   msrc,
+		},
 		Points: []metrics.SketchPoint{{
 			Ts:     int64(ts / 1e9),
 			Sketch: qsketch,
 		}},
-		Source: msrc,
 	})
 }
 

comp/otelcol/otlp/components/exporter/serializerexporter/exporter_test.go

@@ -47,7 +47,7 @@ func (r *metricRecorder) SendSketch(s metrics.SketchesSource) error {
 		if c == nil {
 			continue
 		}
-		r.sketchSeriesList = append(r.sketchSeriesList, c)
+		r.sketchSeriesList = append(r.sketchSeriesList, c.(*metrics.SketchSeries))
 	}
 	return nil
 }

pkg/aggregator/check_sampler.go

@@ -97,10 +97,12 @@ func (cs *CheckSampler) newSketchSeries(ck ckey.ContextKey, points []metrics.Ske
 		return nil
 	}
 	ss := &metrics.SketchSeries{
-		Name: ctx.Name,
-		Tags: ctx.Tags(),
-		Host: ctx.Host,
-		// Interval: TODO: investigate
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: ctx.Name,
+			Tags: ctx.Tags(),
+			Host: ctx.Host,
+			// Interval: TODO: investigate
+		},
 		Points: points,
 	}
 

pkg/aggregator/check_sampler_test.go

@@ -257,8 +257,10 @@ func testCheckHistogramBucketSampling(t *testing.T, store *tags.Store) {
 
 	// ~3% error seen in this test case for sums (sum error is additive so it's always the worst)
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
-		Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+			Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12345.0, Sketch: expSketch},
 		},
@@ -289,8 +291,10 @@ func testCheckHistogramBucketSampling(t *testing.T, store *tags.Store) {
 	assert.Equal(t, 1, len(flushed))
 	// ~3% error seen in this test case for sums (sum error is additive so it's always the worst)
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
-		Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+			Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12400.0, Sketch: expSketch},
 		},
@@ -352,8 +356,10 @@ func testCheckHistogramBucketDontFlushFirstValue(t *testing.T, store *tags.Store
 	assert.Equal(t, 1, len(flushed))
 	// ~3% error seen in this test case for sums (sum error is additive so it's always the worst)
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
-		Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+			Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12400.0, Sketch: expSketch},
 		},
@@ -421,14 +427,18 @@ func testCheckHistogramBucketReset(t *testing.T, store *tags.Store) {
 
 	require.Len(t, flushed, 2)
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12410, Sketch: sketchOf(10, 20, 3)},
 		},
 	}, flushed[0], 0.01)
 
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12420, Sketch: sketchOf(10, 20, 2)},
 		},
@@ -492,8 +502,10 @@ func testCheckHistogramBucketMultipleBucketsSampling(t *testing.T, store *tags.S
 	expSketch.InsertInterpolate(10.0, 20.0, 4)
 	expSketch.InsertInterpolate(30.0, 40.0, 6)
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
-		Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+			Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12345.0, Sketch: expSketch.Finish()},
 		},
@@ -533,8 +545,10 @@ func testCheckHistogramBucketMultipleBucketsSampling(t *testing.T, store *tags.S
 	expSketch.InsertInterpolate(10.0, 20.0, 3)
 	expSketch.InsertInterpolate(30.0, 40.0, 5)
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
-		Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+			Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12400.0, Sketch: expSketch.Finish()},
 		},
@@ -582,8 +596,10 @@ func testCheckHistogramBucketInfinityBucket(t *testing.T, store *tags.Store) {
 
 	// ~3% error seen in this test case for sums (sum error is additive so it's always the worst)
 	metrics.AssertSketchSeriesApproxEqual(t, &metrics.SketchSeries{
-		Name: "my.histogram",
-		Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.histogram",
+			Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12345.0, Sketch: expSketch},
 		},
@@ -619,8 +635,10 @@ func testCheckDistribution(t *testing.T, store *tags.Store) {
 	expSketch.Insert(quantile.Default(), 1)
 
 	metrics.AssertSketchSeriesEqual(t, &metrics.SketchSeries{
-		Name: "my.metric.name",
-		Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		DistributionMetadata: metrics.DistributionMetadata{
+			Name: "my.metric.name",
+			Tags: tagset.CompositeTagsFromSlice([]string{"foo", "bar"}),
+		},
 		Points: []metrics.SketchPoint{
 			{Ts: 12345.0, Sketch: expSketch},
 		},
@@ -771,7 +789,7 @@ func testFilteredSketches(t *testing.T, store *tags.Store) {
 	require.Equal(t, 3, len(sketches))
 	sketchNames := make(map[string]bool)
 	for _, sketch := range sketches {
-		sketchNames[sketch.Name] = true
+		sketchNames[sketch.GetName()] = true
 	}
 
 	assert.True(t, sketchNames["custom.distribution.one"])

pkg/aggregator/demultiplexer.go

@@ -126,7 +126,7 @@ func createIterableMetrics(
 			if hostTags != nil {
 				sketch.Tags = tagset.CombineCompositeTagsAndSlice(sketch.Tags, hostTagProvider.GetHostTags())
 			}
-			tagsetTlm.updateHugeSketchesTelemetry(sketch)
+			tagsetTlm.updateHugeSketchesTelemetry(&sketch.DistributionMetadata)
 		}, flushAndSerializeInParallel.BufferSize, flushAndSerializeInParallel.ChannelSize)
 	}
 	return series, sketches