From 526d6c60fa347e1a9d879806cfbea2ba90faedf4 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Fri, 1 May 2026 21:13:28 -0400 Subject: [PATCH] refactor: align with sketchlib-go API renames MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update call sites to follow sketchlib-go's renamed methods (which now mirror the asap_sketchlib unified API). No semantic / wire-format changes; pure identifier rename: sketch.Add(v) → sketch.Update(v) hll.Insert(v) / .InsertValue → hll.UpdateValue(v) ddsketch.GetValueAtQuantile(q) → ddsketch.Quantile(q) hll.EstimateCardinality() → hll.Estimate() Files touched: - opentelemetry-collector-contrib-patch/processor/{ddsketch,hll,kll}processor/processor.go - opentelemetry-go-patch/sdk/metric/internal/aggregate/{hll,kll}sketch.go - opentelemetry-app/cmd/deltaaccbench/main.go - otel_collector_benchmark/{cardinality_crossover,matched_accuracy}/main.go Depends on ProjectASAP/sketchlib-go's `refactor/api-align-with-asap-sketchlib` branch (commit 7059d73). Co-Authored-By: Claude Opus 4.7 (1M context) --- opentelemetry-app/cmd/deltaaccbench/main.go | 18 +++++++++--------- .../processor/ddsketchprocessor/processor.go | 10 +++++----- .../processor/hllprocessor/processor.go | 16 ++++++++-------- .../processor/kllprocessor/processor.go | 4 ++-- .../sdk/metric/internal/aggregate/hllsketch.go | 4 ++-- .../sdk/metric/internal/aggregate/kllsketch.go | 2 +- .../cardinality_crossover/main.go | 4 ++-- .../matched_accuracy/main.go | 6 +++--- 8 files changed, 32 insertions(+), 32 deletions(-) diff --git a/opentelemetry-app/cmd/deltaaccbench/main.go b/opentelemetry-app/cmd/deltaaccbench/main.go index 684aa84b..37362a19 100644 --- a/opentelemetry-app/cmd/deltaaccbench/main.go +++ b/opentelemetry-app/cmd/deltaaccbench/main.go @@ -635,12 +635,12 @@ func benchHLL(mode string, deltaOn bool) sketchResult { if mode == "batch" { cur = hll.NewHyperLogLog() for _, v := range vals { - cur.Insert(v) + cur.UpdateValue(v) seenValues[v] = struct{}{} } } else { for _, v := range vals { - winSketch.Insert(v) + winSketch.UpdateValue(v) seenValues[v] = struct{}{} } cur = cloneHLL(winSketch) @@ -811,11 +811,11 @@ func benchDD(mode string, deltaOn bool) sketchResult { if mode == "batch" { cur = dd.New(alpha) for _, v := range vals { - cur.Add(v) + cur.Update(v) } } else { for _, v := range vals { - winSketch.Add(v) + winSketch.Update(v) } cur = winSketch.Clone() } @@ -890,7 +890,7 @@ func benchDD(mode string, deltaOn bool) sketchResult { var sumRel, maxRel float64 for _, q := range quantiles { trueQ := trueQuantile(sortedVals, q) - estQ, ok := reconSketch.GetValueAtQuantile(q) + estQ, ok := reconSketch.Quantile(q) if !ok || trueQ == 0 { continue } @@ -940,8 +940,8 @@ func ddSketchesCompatible(a, b *dd.DDSketch, quantiles []float64) bool { return false } for _, q := range quantiles { - av, aok := a.GetValueAtQuantile(q) - bv, bok := b.GetValueAtQuantile(q) + av, aok := a.Quantile(q) + bv, bok := b.Quantile(q) if aok != bok { return false } @@ -999,11 +999,11 @@ func benchKLL(mode string) sketchResult { if mode == "batch" { cur = kll.InitKLL(k) for _, v := range vals { - cur.Insert(v) + cur.Update(v) } } else { for _, v := range vals { - winSketch.Insert(v) + winSketch.Update(v) } // KLL doesn't have Clone; serialize/deserialize for snapshot. curBytes, err := winSketch.SerializeProtoBytes() diff --git a/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go index 013d92ca..9de71c67 100644 --- a/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go @@ -340,7 +340,7 @@ func (p *ddsketchProcessor) buildQuantileMetric(src pmetric.Metric, series map[s continue } for _, q := range p.cfg.Quantiles { - val, ok := s.sketch.GetValueAtQuantile(q) + val, ok := s.sketch.Quantile(q) if !ok { // sketchlib-go returns (0, false) for empty sketch // or an out-of-range quantile; skip silently to @@ -425,9 +425,9 @@ func (p *ddsketchProcessor) consumeGaugeDataPoints(dps pmetric.NumberDataPointSl } switch dp.ValueType() { case pmetric.NumberDataPointValueTypeDouble: - sk.Add(dp.DoubleValue()) + sk.Update(dp.DoubleValue()) case pmetric.NumberDataPointValueTypeInt: - sk.Add(float64(dp.IntValue())) + sk.Update(float64(dp.IntValue())) default: if p.logger != nil { p.logger.Error("unsupported gauge data point type", zap.Any("type", dp.ValueType())) @@ -784,9 +784,9 @@ func (p *ddsketchProcessor) accumulateGaugeMetric(sw *scopeWindow, metric pmetri switch dp.ValueType() { case pmetric.NumberDataPointValueTypeDouble: - sk.Add(dp.DoubleValue()) + sk.Update(dp.DoubleValue()) case pmetric.NumberDataPointValueTypeInt: - sk.Add(float64(dp.IntValue())) + sk.Update(float64(dp.IntValue())) default: if p.logger != nil { p.logger.Error("unsupported gauge data point type in window mode", zap.Any("type", dp.ValueType())) diff --git a/opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go index 2dca58ec..8e68b693 100644 --- a/opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go @@ -263,7 +263,7 @@ func (p *hllProcessor) processBatch(md pmetric.Metrics) error { continue } bs := getOrCreate(metric.Name(), metric.Unit(), dp.Attributes()) - bs.sketch.InsertValue(dp.DoubleValue()) + bs.sketch.UpdateValue(dp.DoubleValue()) bs.count++ } case pmetric.MetricTypeHLLSketch: @@ -320,7 +320,7 @@ func (p *hllProcessor) processBatch(md pmetric.Metrics) error { bs.attrs.CopyTo(dp.Attributes()) dp.SetTimestamp(now) dp.SetCount(bs.count) - dp.SetCardinality(uint64(bs.sketch.EstimateCardinality())) + dp.SetCardinality(uint64(bs.sketch.Estimate())) dp.SetSketch(payload) dp.SetEncoding(encodingTag) dp.SetPrecision(uint32(hll.HLLPrecision)) @@ -335,7 +335,7 @@ func (p *hllProcessor) processBatch(md pmetric.Metrics) error { dp := m.Gauge().DataPoints().AppendEmpty() bs.attrs.CopyTo(dp.Attributes()) dp.SetTimestamp(now) - dp.SetDoubleValue(float64(bs.sketch.EstimateCardinality())) + dp.SetDoubleValue(float64(bs.sketch.Estimate())) } } return nil @@ -500,7 +500,7 @@ func (p *hllProcessor) accumulateGaugeMetric(sw *scopeWindow, metric pmetric.Met } mw.series[attrKey] = series } - series.sketch.InsertValue(dp.DoubleValue()) + series.sketch.UpdateValue(dp.DoubleValue()) } } @@ -594,7 +594,7 @@ func (p *hllProcessor) flushWindow(ctx context.Context) error { series.attrs.CopyTo(dp.Attributes()) dp.SetTimestamp(now) dp.SetCount(0) - dp.SetCardinality(uint64(series.sketch.EstimateCardinality())) + dp.SetCardinality(uint64(series.sketch.Estimate())) dp.SetSketch(payload) dp.SetEncoding(pmetric.HLLSketchEncodingDelta) dp.SetPrecision(uint32(hll.HLLPrecision)) @@ -610,7 +610,7 @@ func (p *hllProcessor) flushWindow(ctx context.Context) error { series.attrs.CopyTo(dp.Attributes()) dp.SetTimestamp(now) dp.SetCount(0) - dp.SetCardinality(uint64(series.sketch.EstimateCardinality())) + dp.SetCardinality(uint64(series.sketch.Estimate())) dp.SetSketch(payload) dp.SetEncoding(pmetric.HLLSketchEncodingProto) dp.SetPrecision(uint32(hll.HLLPrecision)) @@ -631,7 +631,7 @@ func (p *hllProcessor) flushWindow(ctx context.Context) error { series.attrs.CopyTo(dp.Attributes()) dp.SetTimestamp(now) dp.SetCount(0) - dp.SetCardinality(uint64(series.sketch.EstimateCardinality())) + dp.SetCardinality(uint64(series.sketch.Estimate())) dp.SetSketch(payload) dp.SetEncoding(encodingTag) dp.SetPrecision(uint32(hll.HLLPrecision)) @@ -657,7 +657,7 @@ func (p *hllProcessor) flushWindow(ctx context.Context) error { dps = append(dps, struct { attrs pcommon.Map val float64 - }{series.attrs, float64(series.sketch.EstimateCardinality())}) + }{series.attrs, float64(series.sketch.Estimate())}) series.attrs = pcommon.Map{} p.seriesPool.Put(series) } diff --git a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go index deb1b60c..08b48482 100644 --- a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go @@ -193,7 +193,7 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { } bs := getOrCreate(metric.Name(), metric.Unit(), dp.Attributes()) if bs.sketch != nil { - bs.sketch.Insert(val) + bs.sketch.Update(val) } } case pmetric.MetricTypeKLLSketch: @@ -431,7 +431,7 @@ func (p *kllProcessor) accumulateGaugeMetric(sw *scopeWindow, metric pmetric.Met val = dp.DoubleValue() } if series.sketch != nil { - series.sketch.Insert(val) + series.sketch.Update(val) } } } diff --git a/opentelemetry-go-patch/sdk/metric/internal/aggregate/hllsketch.go b/opentelemetry-go-patch/sdk/metric/internal/aggregate/hllsketch.go index 4da71e68..a5a6c638 100644 --- a/opentelemetry-go-patch/sdk/metric/internal/aggregate/hllsketch.go +++ b/opentelemetry-go-patch/sdk/metric/internal/aggregate/hllsketch.go @@ -91,7 +91,7 @@ func (d *hllSketchValues[N]) measure( } } - series.sketch.InsertValue(float64(value)) + series.sketch.UpdateValue(float64(value)) series.count++ series.measuredSince = true } @@ -281,7 +281,7 @@ func (d *hllSketch[N]) exportDataPoint( encoding metricdata.HLLSketchEncoding, dest *metricdata.HLLSketchDataPoint, ) bool { - cardinality := uint64(series.sketch.EstimateCardinality()) + cardinality := uint64(series.sketch.Estimate()) dp := dest if series.seriesID != 0 { diff --git a/opentelemetry-go-patch/sdk/metric/internal/aggregate/kllsketch.go b/opentelemetry-go-patch/sdk/metric/internal/aggregate/kllsketch.go index 20ccec03..4218637d 100644 --- a/opentelemetry-go-patch/sdk/metric/internal/aggregate/kllsketch.go +++ b/opentelemetry-go-patch/sdk/metric/internal/aggregate/kllsketch.go @@ -102,7 +102,7 @@ func (d *kllSketchValues[N]) measure( } } - series.sketch.Insert(float64(value)) + series.sketch.Update(float64(value)) series.count++ series.measuredSince = true fv := float64(value) diff --git a/otel_collector_benchmark/cardinality_crossover/main.go b/otel_collector_benchmark/cardinality_crossover/main.go index 709a3f87..d3e5b070 100644 --- a/otel_collector_benchmark/cardinality_crossover/main.go +++ b/otel_collector_benchmark/cardinality_crossover/main.go @@ -276,7 +276,7 @@ func evalCountSketch(n int, tuples []tuple, gt map[uint64]int64, topKKeys []uint t0 := time.Now() for _, in := range inputs { - cs.Insert(in) + cs.Update(in) } insertSec := time.Since(t0).Seconds() @@ -322,7 +322,7 @@ func evalCountMinSketch(n int, tuples []tuple, gt map[uint64]int64, topKKeys []u t0 := time.Now() for _, in := range inputs { - cms.Insert(in) + cms.Update(in) } insertSec := time.Since(t0).Seconds() diff --git a/otel_collector_benchmark/matched_accuracy/main.go b/otel_collector_benchmark/matched_accuracy/main.go index 5771d1bc..338cc8cb 100644 --- a/otel_collector_benchmark/matched_accuracy/main.go +++ b/otel_collector_benchmark/matched_accuracy/main.go @@ -140,7 +140,7 @@ func runDDSketch(stream []float64) (float64, int64, int, func(float64) float64) d := ddsketch.NewDDSketch(0.01) t0 := time.Now() for _, v := range stream { - d.Add(v) + d.Update(v) } insertNs := float64(time.Since(t0).Nanoseconds()) / float64(len(stream)) @@ -154,7 +154,7 @@ func runDDSketch(stream []float64) (float64, int64, int, func(float64) float64) } q := func(p float64) float64 { - v, _ := d.GetValueAtQuantile(p) + v, _ := d.Quantile(p) return v } return insertNs, heap, len(wire), q @@ -172,7 +172,7 @@ func runKLL(stream []float64) (float64, int64, int, func(float64) float64) { k := kll.InitKLL(200) t0 := time.Now() for _, v := range stream { - k.Insert(v) + k.Update(v) } insertNs := float64(time.Since(t0).Nanoseconds()) / float64(len(stream))