From 53e3f1ace0c4198e9debd59dfa6687e96ea993ea Mon Sep 17 00:00:00 2001 From: Gnanesh Date: Tue, 24 Mar 2026 00:07:17 -0400 Subject: [PATCH 1/3] Update KLL processor sketch serialization --- .../config-otlp-pathway1.yaml | 31 +++ .../config-otlp-pathway2-batch.yaml | 35 +++ .../config-otlp-pathway2-window.yaml | 36 +++ .../cmd/kll/build-config.yaml | 8 +- .../processor/kllprocessor/go.mod | 6 +- .../processor/kllprocessor/go.sum | 12 - .../processor/kllprocessor/processor.go | 231 ++++++++---------- .../processor/kllprocessor/processor_test.go | 3 + 8 files changed, 219 insertions(+), 143 deletions(-) create mode 100644 opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway1.yaml create mode 100644 opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-batch.yaml create mode 100644 opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-window.yaml diff --git a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway1.yaml b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway1.yaml new file mode 100644 index 00000000..fc97a86e --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway1.yaml @@ -0,0 +1,31 @@ +# OTLP Pathway 1 - Local testing +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:53217 + http: + endpoint: 0.0.0.0:53218 + +processors: + batch: + send_batch_size: 5000 + +exporters: + otlphttp: + endpoint: http://localhost:4318 + encoding: proto + compression: gzip +service: + pipelines: + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlphttp] + telemetry: + metrics: + level: basic + logs: + level: info + resource: + service.name: kllcol-otlp-pathway1-local diff --git a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-batch.yaml b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-batch.yaml new file mode 100644 index 00000000..2675f507 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-batch.yaml @@ -0,0 +1,35 @@ +# OTLP Pathway 2 - Batch mode - Local testing +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:53217 + http: + endpoint: 0.0.0.0:53218 + +processors: + KLL: + mode: batch + k: 256 + transmit_sketch: true + quantiles: [0.5, 0.9, 0.99] + drop_original: true + +exporters: + otlphttp: + endpoint: http://localhost:4318 + encoding: proto + compression: gzip +service: + pipelines: + metrics: + receivers: [otlp] + processors: [KLL] + exporters: [otlphttp] + telemetry: + metrics: + level: basic + logs: + level: debug + resource: + service.name: kllcol-otlp-pathway2-local diff --git a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-window.yaml b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-window.yaml new file mode 100644 index 00000000..76bf4d02 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-window.yaml @@ -0,0 +1,36 @@ +# OTLP Pathway 2 - Window mode - Local testing +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:53217 + http: + endpoint: 0.0.0.0:53218 + +processors: + KLL: + mode: window + window_duration: 10s + k: 256 + transmit_sketch: true + quantiles: [0.5, 0.9, 0.99] + drop_original: true + +exporters: + otlphttp: + endpoint: http://localhost:4318 + encoding: proto + compression: gzip +service: + pipelines: + metrics: + receivers: [otlp] + processors: [KLL] + exporters: [otlphttp] + telemetry: + metrics: + level: basic + logs: + level: debug + resource: + service.name: kllcol-otlp-pathway2-window-local diff --git a/opentelemetry-collector-contrib-patch/cmd/kll/build-config.yaml b/opentelemetry-collector-contrib-patch/cmd/kll/build-config.yaml index d00cbaf0..ae9f5968 100644 --- a/opentelemetry-collector-contrib-patch/cmd/kll/build-config.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/kll/build-config.yaml @@ -4,17 +4,19 @@ dist: output_path: . exporters: - - gomod: go.opentelemetry.io/collector/exporter/nopexporter v0.141.0 + - gomod: go.opentelemetry.io/collector/exporter/nopexporter v0.142.0 + - gomod: go.opentelemetry.io/collector/exporter/otlphttpexporter v0.142.0 - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/fileexporter v0.142.0 - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.142.0 processors: - - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.141.0 + - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.142.0 - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/kllprocessor v0.0.0 receivers: - - gomod: go.opentelemetry.io/collector/receiver/otlpreceiver v0.141.0 + - gomod: go.opentelemetry.io/collector/receiver/otlpreceiver v0.142.0 replaces: - github.com/open-telemetry/opentelemetry-collector-contrib/processor/kllprocessor => ./processor/kllprocessor - go.opentelemetry.io/collector/pdata => ../opentelemetry-collector/pdata + # - github.com/ProjectASAP/sketchlib-go => ../../sketchlib-go diff --git a/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.mod b/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.mod index 52d844e0..c8b492c6 100644 --- a/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.mod +++ b/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.mod @@ -3,12 +3,13 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/kllpr go 1.25.4 require ( - github.com/ProjectASAP/sketchlib-go v0.0.0-20260321024028-d20a9f9151b5 + github.com/ProjectASAP/sketchlib-go v0.0.0-20260310013347-2c5db0c75da8 github.com/stretchr/testify v1.11.1 go.opentelemetry.io/collector/component v1.48.0 go.opentelemetry.io/collector/component/componenttest v0.142.0 go.opentelemetry.io/collector/consumer v1.48.0 go.opentelemetry.io/collector/consumer/consumertest v0.142.0 + google.golang.org/protobuf v1.36.11 ) require ( @@ -32,7 +33,6 @@ require ( go.yaml.in/yaml/v2 v2.4.3 // indirect golang.org/x/sys v0.39.0 // indirect golang.org/x/text v0.30.0 // indirect - google.golang.org/protobuf v1.36.11 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) @@ -54,4 +54,4 @@ require ( replace go.opentelemetry.io/collector/pdata => ../../../opentelemetry-collector/pdata -replace go.opentelemetry.io/collector/processor => ../../../opentelemetry-collector/processor +// replace github.com/ProjectASAP/sketchlib-go => ../../../../sketchlib-go diff --git a/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.sum b/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.sum index c42b4b6e..e6369c6b 100644 --- a/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.sum +++ b/opentelemetry-collector-contrib-patch/processor/kllprocessor/go.sum @@ -1,13 +1,3 @@ -github.com/ProjectASAP/sketchlib-go v0.0.0-20260316040945-49890c71f035 h1:aLrGKic2Q7KJJmfTzqB3fSvjjW2KfRvY5BDcRvYI2zE= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260316040945-49890c71f035/go.mod h1:Nzyu+hI1KUBmgeyhSlmUnTp5LvI/EwD8VuoZyNQoEZs= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260320220729-3ba826ceb054 h1:HYCqQ0IEu/hDk5UOi4AHF5yuNb/DX3jD8x2PtO4bApQ= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260320220729-3ba826ceb054/go.mod h1:VmV0RYT6+rXpb1gVxTG3Z17sttm0nfj7RuesrL1oDTI= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260321021603-cddd774cb224 h1:mYN1NJLRp9elj4RoLqlHMcoZeQxI7jPNqnxbxxS7wBw= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260321021603-cddd774cb224/go.mod h1:VmV0RYT6+rXpb1gVxTG3Z17sttm0nfj7RuesrL1oDTI= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260321023259-ecebc36fb5aa h1:nRWz+s995NYgaRI6a7YHu2lnfGhyzn3Sn8uDFbuNEO0= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260321023259-ecebc36fb5aa/go.mod h1:VmV0RYT6+rXpb1gVxTG3Z17sttm0nfj7RuesrL1oDTI= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260321024028-d20a9f9151b5 h1:e1If3BxB2QRDt/57ho0PoxxIwouEvuaojy8m9u21IC0= -github.com/ProjectASAP/sketchlib-go v0.0.0-20260321024028-d20a9f9151b5/go.mod h1:VmV0RYT6+rXpb1gVxTG3Z17sttm0nfj7RuesrL1oDTI= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -117,8 +107,6 @@ golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk= golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= -google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= -google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go index a4fbc6ba..c255ab82 100644 --- a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go @@ -13,19 +13,19 @@ import ( "go.opentelemetry.io/collector/consumer" "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/pmetric" - "go.opentelemetry.io/collector/processor/selfmonitor" "go.uber.org/zap" + "google.golang.org/protobuf/proto" ) // builderPool recycles strings.Builder instances to avoid per-call heap // allocations in the hot attributesKey path. var builderPool = sync.Pool{New: func() any { return new(strings.Builder) }} + type kllProcessor struct { cfg *Config logger *zap.Logger nextConsumer consumer.Metrics - monitor *selfmonitor.Monitor mu sync.Mutex windowStore map[string]*resourceWindow @@ -106,8 +106,6 @@ func (p *kllProcessor) Start(ctx context.Context, _ component.Host) error { } func (p *kllProcessor) Shutdown(ctx context.Context) error { - defer p.shutdownMonitor() - if p.cfg.Mode != ModeWindow || !p.windowStarted.Load() { return nil } @@ -121,15 +119,44 @@ func (p *kllProcessor) Shutdown(ctx context.Context) error { } func (p *kllProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error { - p.recordInput(ctx, md) - + rmCount := md.ResourceMetrics().Len() + dpCount := 0 + for i := 0; i < rmCount; i++ { + for j := 0; j < md.ResourceMetrics().At(i).ScopeMetrics().Len(); j++ { + metrics := md.ResourceMetrics().At(i).ScopeMetrics().At(j).Metrics() + for k := 0; k < metrics.Len(); k++ { + m := metrics.At(k) + switch m.Type() { + case pmetric.MetricTypeGauge: + dpCount += m.Gauge().DataPoints().Len() + case pmetric.MetricTypeKLLSketch: + dpCount += m.KLLSketch().DataPoints().Len() + default: + // ignore + } + } + } + } + if p.logger != nil { + p.logger.Debug("KLL processor received data", zap.Int("resource_metrics", rmCount), zap.Int("data_points", dpCount)) + } switch p.cfg.Mode { case ModeBatch: if err := p.processBatch(md); err != nil { return err } - p.recordOutput(ctx, md) - return p.nextConsumer.ConsumeMetrics(ctx, md) + if p.logger != nil { + p.logger.Debug("KLL processor sending batch output") + } + out := md + if p.cfg.DropOriginal { + var ok bool + out, ok = p.batchOutputOnly(md) + if !ok { + return nil + } + } + return p.nextConsumer.ConsumeMetrics(ctx, out) case ModeWindow: p.accumulateIntoWindow(md) return nil @@ -153,10 +180,12 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { batched := make(map[string]*batchSeries) // key = metricName + "::" + attributesKey(attrs) getOrCreate := func(name, unit string, attrs pcommon.Map) *batchSeries { - key := name + "::" + p.seriesKey(attrs) + key := name + "::" + attributesKey(attrs) bs := batched[key] if bs == nil { - bs = &batchSeries{name: name, unit: unit, attrs: p.seriesAttrs(attrs), sketch: newKLLSketch(p.cfg.K)} + attrCopy := pcommon.NewMap() + attrs.CopyTo(attrCopy) + bs = &batchSeries{name: name, unit: unit, attrs: attrCopy, sketch: newKLLSketch(p.cfg.K)} batched[key] = bs } return bs @@ -174,9 +203,6 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { dps := metric.Gauge().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) - if !p.matchesMatchers(dp.Attributes()) { - continue - } var val float64 if p.cfg.ReadAsInt { val = float64(dp.IntValue()) @@ -192,12 +218,9 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { dps := metric.KLLSketch().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) - if !p.matchesMatchers(dp.Attributes()) { - continue - } bs := getOrCreate(metric.Name(), metric.Unit(), dp.Attributes()) if bs.sketch != nil && len(dp.Sketch()) > 0 { - incoming, err := kll.DeserializeKLLSketchFromProtoBytes(dp.Sketch()) + incoming, err := kll.DeserializeKLLSketchFromBytes(dp.Sketch()) if err == nil { _ = bs.sketch.Merge(incoming) } else if p.logger != nil { @@ -214,7 +237,9 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { return nil } - scope := md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty() + // Append a new ResourceMetrics for our output (processBatch mutates md in place). + outputRm := md.ResourceMetrics().AppendEmpty() + scope := outputRm.ScopeMetrics().AppendEmpty() scope.Scope().SetName("otelcol/kllprocessor") now := pcommon.NewTimestampFromTime(time.Now()) @@ -246,6 +271,41 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { return nil } +// batchOutputOnly builds a new pmetric.Metrics containing only the KLL output +// (quantiles or sketches) from processBatch, with resource copied from the first +// input ResourceMetrics. Returns (metrics, true) if there is output to forward, +// or (empty metrics, false) if not. +func (p *kllProcessor) batchOutputOnly(md pmetric.Metrics) (pmetric.Metrics, bool) { + rms := md.ResourceMetrics() + if rms.Len() < 2 { + // Need at least 1 input rm + 1 output rm (appended by processBatch) + return pmetric.NewMetrics(), false + } + firstRm := rms.At(0) + outputRm := rms.At(rms.Len() - 1) + + // Check if output has any data points + dpCount := 0 + for j := 0; j < outputRm.ScopeMetrics().Len(); j++ { + for k := 0; k < outputRm.ScopeMetrics().At(j).Metrics().Len(); k++ { + m := outputRm.ScopeMetrics().At(j).Metrics().At(k) + if m.Type() == pmetric.MetricTypeGauge { + dpCount += m.Gauge().DataPoints().Len() + } + } + } + if dpCount == 0 { + return pmetric.NewMetrics(), false + } + + // Build result with only output. Copy output rm and set resource from first input. + out := pmetric.NewMetrics() + dstRm := out.ResourceMetrics().AppendEmpty() + outputRm.CopyTo(dstRm) + firstRm.Resource().CopyTo(dstRm.Resource()) + return out, true +} + func attributesKey(attrs pcommon.Map) string { keys := make([]string, 0, attrs.Len()) attrs.Range(func(k string, _ pcommon.Value) bool { @@ -270,58 +330,6 @@ func attributesKey(attrs pcommon.Map) string { return s } -// matchesMatchers returns true if attrs satisfies all configured LabelMatchers. -func (p *kllProcessor) matchesMatchers(attrs pcommon.Map) bool { - for _, m := range p.cfg.LabelMatchers { - v, ok := attrs.Get(m.Key) - if !ok || v.AsString() != m.Value { - return false - } - } - return true -} - -// seriesKey returns the map key used to locate a series in the window/batch store. -// When AggregateBy is configured, only those label values form the key (cross-series -// aggregation). Otherwise the full attribute set is used (per-series, default). -func (p *kllProcessor) seriesKey(attrs pcommon.Map) string { - if len(p.cfg.AggregateBy) == 0 { - return attributesKey(attrs) - } - b := builderPool.Get().(*strings.Builder) - b.Reset() - for _, k := range p.cfg.AggregateBy { // already sorted by Validate - v, ok := attrs.Get(k) - if !ok { - continue - } - b.WriteString(k) - b.WriteByte('=') - b.WriteString(v.AsString()) - b.WriteByte(';') - } - key := b.String() - builderPool.Put(b) - return key -} - -// seriesAttrs returns the attribute map to store on a new series entry. -// When AggregateBy is configured, only those labels are included in the output. -// Otherwise a full copy of attrs is returned. -func (p *kllProcessor) seriesAttrs(attrs pcommon.Map) pcommon.Map { - out := pcommon.NewMap() - if len(p.cfg.AggregateBy) == 0 { - attrs.CopyTo(out) - return out - } - for _, k := range p.cfg.AggregateBy { - if v, ok := attrs.Get(k); ok { - out.PutStr(k, v.AsString()) - } - } - return out -} - func (p *kllProcessor) accumulateIntoWindow(md pmetric.Metrics) { rms := md.ResourceMetrics() if rms.Len() == 0 { @@ -390,14 +398,13 @@ func (p *kllProcessor) accumulateGaugeMetric(sw *scopeWindow, metric pmetric.Met dps := metric.Gauge().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - attrKey := p.seriesKey(dp.Attributes()) + attrKey := attributesKey(dp.Attributes()) series := mw.series[attrKey] if series == nil { + attrCopy := pcommon.NewMap() + dp.Attributes().CopyTo(attrCopy) series = p.seriesPool.Get().(*kllSeries) - series.attrs = p.seriesAttrs(dp.Attributes()) + series.attrs = attrCopy if series.sketch != nil { series.sketch.Reset() } else { @@ -424,14 +431,13 @@ func (p *kllProcessor) accumulateKLLSketchMetric(sw *scopeWindow, metric pmetric dps := metric.KLLSketch().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - attrKey := p.seriesKey(dp.Attributes()) + attrKey := attributesKey(dp.Attributes()) series := mw.series[attrKey] if series == nil { + attrCopy := pcommon.NewMap() + dp.Attributes().CopyTo(attrCopy) series = p.seriesPool.Get().(*kllSeries) - series.attrs = p.seriesAttrs(dp.Attributes()) + series.attrs = attrCopy if series.sketch != nil { series.sketch.Reset() } else { @@ -440,7 +446,7 @@ func (p *kllProcessor) accumulateKLLSketchMetric(sw *scopeWindow, metric pmetric mw.series[attrKey] = series } if series.sketch != nil && len(dp.Sketch()) > 0 { - incoming, err := kll.DeserializeKLLSketchFromProtoBytes(dp.Sketch()) + incoming, err := kll.DeserializeKLLSketchFromBytes(dp.Sketch()) if err == nil { _ = series.sketch.Merge(incoming) } else if p.logger != nil { @@ -549,52 +555,23 @@ func (p *kllProcessor) flushWindow(ctx context.Context) error { if out.ResourceMetrics().Len() == 0 { return nil } - p.recordOutput(ctx, out) - return p.nextConsumer.ConsumeMetrics(ctx, out) -} - -func (p *kllProcessor) enableSelfMonitoring(settings component.TelemetrySettings, processorID string) { - monitor, err := selfmonitor.New(settings, processorID, "KLL", p.activeSeriesCount) - if err != nil { - if p.logger != nil { - p.logger.Warn("kllprocessor: failed to initialize self-monitoring", zap.Error(err)) - } - return - } - p.monitor = monitor -} - -func (p *kllProcessor) shutdownMonitor() { - if p.monitor != nil { - p.monitor.Shutdown() - } -} - -func (p *kllProcessor) recordInput(ctx context.Context, md pmetric.Metrics) { - if p.monitor != nil { - p.monitor.RecordInput(ctx, md) - } -} - -func (p *kllProcessor) recordOutput(ctx context.Context, md pmetric.Metrics) { - if p.monitor != nil { - p.monitor.RecordOutput(ctx, md) - } -} - -func (p *kllProcessor) activeSeriesCount() int64 { - p.mu.Lock() - defer p.mu.Unlock() - - var total int64 - for _, rw := range p.windowStore { - for _, sw := range rw.scopes { - for _, mw := range sw.metrics { - total += int64(len(mw.series)) + outRmCount := out.ResourceMetrics().Len() + outDpCount := 0 + for i := 0; i < outRmCount; i++ { + for j := 0; j < out.ResourceMetrics().At(i).ScopeMetrics().Len(); j++ { + metrics := out.ResourceMetrics().At(i).ScopeMetrics().At(j).Metrics() + for k := 0; k < metrics.Len(); k++ { + m := metrics.At(k) + if m.Type() == pmetric.MetricTypeGauge { + outDpCount += m.Gauge().DataPoints().Len() + } } } } - return total + if p.logger != nil { + p.logger.Debug("KLL processor sending window flush", zap.Int("resource_metrics", outRmCount), zap.Int("data_points", outDpCount)) + } + return p.nextConsumer.ConsumeMetrics(ctx, out) } func findOrCreateGaugeMetric(metrics pmetric.MetricSlice, name, unit string) pmetric.Metric { @@ -629,7 +606,11 @@ func serializeKLLSketch(sketch *kll.KLLSketch) ([]byte, error) { if sketch == nil { return nil, nil } - return sketch.SerializeProtoBytes() + env, err := sketch.SerializePortable() + if err != nil { + return nil, err + } + return proto.Marshal(env) } func (p *kllProcessor) sketchMetricName(base string) string { diff --git a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor_test.go b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor_test.go index 0d39c10f..98cfcefb 100644 --- a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor_test.go +++ b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor_test.go @@ -68,6 +68,7 @@ func TestBatchModeTransmitSketch(t *testing.T) { cfg.Mode = ModeBatch cfg.TransmitSketch = true cfg.Quantiles = nil + cfg.DropOriginal = true require.NoError(t, cfg.Validate()) sink := new(consumertest.MetricsSink) @@ -290,10 +291,12 @@ func TestEmptyInput(t *testing.T) { } // TestEmptyResourceMetrics verifies ResourceMetrics with zero ScopeMetrics is handled in batch mode. +// Uses DropOriginal=false to test passthrough when there is no data to aggregate. func TestEmptyResourceMetrics(t *testing.T) { cfg := createDefaultConfig().(*Config) cfg.Mode = ModeBatch cfg.Quantiles = []float64{0.5} + cfg.DropOriginal = false require.NoError(t, cfg.Validate()) sink := new(consumertest.MetricsSink) From c1cb24820c4639b82929cfb2cd8d2a3b3988873a Mon Sep 17 00:00:00 2001 From: Gnanesh Date: Tue, 24 Mar 2026 00:29:39 -0400 Subject: [PATCH 2/3] Restore KLL non-serialization logic --- .../processor/kllprocessor/processor.go | 220 ++++++++++-------- 1 file changed, 122 insertions(+), 98 deletions(-) diff --git a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go index c255ab82..4baa3712 100644 --- a/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/kllprocessor/processor.go @@ -13,6 +13,7 @@ import ( "go.opentelemetry.io/collector/consumer" "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/pmetric" + "go.opentelemetry.io/collector/processor/selfmonitor" "go.uber.org/zap" "google.golang.org/protobuf/proto" ) @@ -21,11 +22,11 @@ import ( // allocations in the hot attributesKey path. var builderPool = sync.Pool{New: func() any { return new(strings.Builder) }} - type kllProcessor struct { cfg *Config logger *zap.Logger nextConsumer consumer.Metrics + monitor *selfmonitor.Monitor mu sync.Mutex windowStore map[string]*resourceWindow @@ -106,6 +107,8 @@ func (p *kllProcessor) Start(ctx context.Context, _ component.Host) error { } func (p *kllProcessor) Shutdown(ctx context.Context) error { + defer p.shutdownMonitor() + if p.cfg.Mode != ModeWindow || !p.windowStarted.Load() { return nil } @@ -119,44 +122,15 @@ func (p *kllProcessor) Shutdown(ctx context.Context) error { } func (p *kllProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error { - rmCount := md.ResourceMetrics().Len() - dpCount := 0 - for i := 0; i < rmCount; i++ { - for j := 0; j < md.ResourceMetrics().At(i).ScopeMetrics().Len(); j++ { - metrics := md.ResourceMetrics().At(i).ScopeMetrics().At(j).Metrics() - for k := 0; k < metrics.Len(); k++ { - m := metrics.At(k) - switch m.Type() { - case pmetric.MetricTypeGauge: - dpCount += m.Gauge().DataPoints().Len() - case pmetric.MetricTypeKLLSketch: - dpCount += m.KLLSketch().DataPoints().Len() - default: - // ignore - } - } - } - } - if p.logger != nil { - p.logger.Debug("KLL processor received data", zap.Int("resource_metrics", rmCount), zap.Int("data_points", dpCount)) - } + p.recordInput(ctx, md) + switch p.cfg.Mode { case ModeBatch: if err := p.processBatch(md); err != nil { return err } - if p.logger != nil { - p.logger.Debug("KLL processor sending batch output") - } - out := md - if p.cfg.DropOriginal { - var ok bool - out, ok = p.batchOutputOnly(md) - if !ok { - return nil - } - } - return p.nextConsumer.ConsumeMetrics(ctx, out) + p.recordOutput(ctx, md) + return p.nextConsumer.ConsumeMetrics(ctx, md) case ModeWindow: p.accumulateIntoWindow(md) return nil @@ -180,12 +154,10 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { batched := make(map[string]*batchSeries) // key = metricName + "::" + attributesKey(attrs) getOrCreate := func(name, unit string, attrs pcommon.Map) *batchSeries { - key := name + "::" + attributesKey(attrs) + key := name + "::" + p.seriesKey(attrs) bs := batched[key] if bs == nil { - attrCopy := pcommon.NewMap() - attrs.CopyTo(attrCopy) - bs = &batchSeries{name: name, unit: unit, attrs: attrCopy, sketch: newKLLSketch(p.cfg.K)} + bs = &batchSeries{name: name, unit: unit, attrs: p.seriesAttrs(attrs), sketch: newKLLSketch(p.cfg.K)} batched[key] = bs } return bs @@ -203,6 +175,9 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { dps := metric.Gauge().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) + if !p.matchesMatchers(dp.Attributes()) { + continue + } var val float64 if p.cfg.ReadAsInt { val = float64(dp.IntValue()) @@ -218,6 +193,9 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { dps := metric.KLLSketch().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) + if !p.matchesMatchers(dp.Attributes()) { + continue + } bs := getOrCreate(metric.Name(), metric.Unit(), dp.Attributes()) if bs.sketch != nil && len(dp.Sketch()) > 0 { incoming, err := kll.DeserializeKLLSketchFromBytes(dp.Sketch()) @@ -237,9 +215,7 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { return nil } - // Append a new ResourceMetrics for our output (processBatch mutates md in place). - outputRm := md.ResourceMetrics().AppendEmpty() - scope := outputRm.ScopeMetrics().AppendEmpty() + scope := md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty() scope.Scope().SetName("otelcol/kllprocessor") now := pcommon.NewTimestampFromTime(time.Now()) @@ -271,41 +247,6 @@ func (p *kllProcessor) processBatch(md pmetric.Metrics) error { return nil } -// batchOutputOnly builds a new pmetric.Metrics containing only the KLL output -// (quantiles or sketches) from processBatch, with resource copied from the first -// input ResourceMetrics. Returns (metrics, true) if there is output to forward, -// or (empty metrics, false) if not. -func (p *kllProcessor) batchOutputOnly(md pmetric.Metrics) (pmetric.Metrics, bool) { - rms := md.ResourceMetrics() - if rms.Len() < 2 { - // Need at least 1 input rm + 1 output rm (appended by processBatch) - return pmetric.NewMetrics(), false - } - firstRm := rms.At(0) - outputRm := rms.At(rms.Len() - 1) - - // Check if output has any data points - dpCount := 0 - for j := 0; j < outputRm.ScopeMetrics().Len(); j++ { - for k := 0; k < outputRm.ScopeMetrics().At(j).Metrics().Len(); k++ { - m := outputRm.ScopeMetrics().At(j).Metrics().At(k) - if m.Type() == pmetric.MetricTypeGauge { - dpCount += m.Gauge().DataPoints().Len() - } - } - } - if dpCount == 0 { - return pmetric.NewMetrics(), false - } - - // Build result with only output. Copy output rm and set resource from first input. - out := pmetric.NewMetrics() - dstRm := out.ResourceMetrics().AppendEmpty() - outputRm.CopyTo(dstRm) - firstRm.Resource().CopyTo(dstRm.Resource()) - return out, true -} - func attributesKey(attrs pcommon.Map) string { keys := make([]string, 0, attrs.Len()) attrs.Range(func(k string, _ pcommon.Value) bool { @@ -330,6 +271,58 @@ func attributesKey(attrs pcommon.Map) string { return s } +// matchesMatchers returns true if attrs satisfies all configured LabelMatchers. +func (p *kllProcessor) matchesMatchers(attrs pcommon.Map) bool { + for _, m := range p.cfg.LabelMatchers { + v, ok := attrs.Get(m.Key) + if !ok || v.AsString() != m.Value { + return false + } + } + return true +} + +// seriesKey returns the map key used to locate a series in the window/batch store. +// When AggregateBy is configured, only those label values form the key (cross-series +// aggregation). Otherwise the full attribute set is used (per-series, default). +func (p *kllProcessor) seriesKey(attrs pcommon.Map) string { + if len(p.cfg.AggregateBy) == 0 { + return attributesKey(attrs) + } + b := builderPool.Get().(*strings.Builder) + b.Reset() + for _, k := range p.cfg.AggregateBy { // already sorted by Validate + v, ok := attrs.Get(k) + if !ok { + continue + } + b.WriteString(k) + b.WriteByte('=') + b.WriteString(v.AsString()) + b.WriteByte(';') + } + key := b.String() + builderPool.Put(b) + return key +} + +// seriesAttrs returns the attribute map to store on a new series entry. +// When AggregateBy is configured, only those labels are included in the output. +// Otherwise a full copy of attrs is returned. +func (p *kllProcessor) seriesAttrs(attrs pcommon.Map) pcommon.Map { + out := pcommon.NewMap() + if len(p.cfg.AggregateBy) == 0 { + attrs.CopyTo(out) + return out + } + for _, k := range p.cfg.AggregateBy { + if v, ok := attrs.Get(k); ok { + out.PutStr(k, v.AsString()) + } + } + return out +} + func (p *kllProcessor) accumulateIntoWindow(md pmetric.Metrics) { rms := md.ResourceMetrics() if rms.Len() == 0 { @@ -398,13 +391,14 @@ func (p *kllProcessor) accumulateGaugeMetric(sw *scopeWindow, metric pmetric.Met dps := metric.Gauge().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) - attrKey := attributesKey(dp.Attributes()) + if !p.matchesMatchers(dp.Attributes()) { + continue + } + attrKey := p.seriesKey(dp.Attributes()) series := mw.series[attrKey] if series == nil { - attrCopy := pcommon.NewMap() - dp.Attributes().CopyTo(attrCopy) series = p.seriesPool.Get().(*kllSeries) - series.attrs = attrCopy + series.attrs = p.seriesAttrs(dp.Attributes()) if series.sketch != nil { series.sketch.Reset() } else { @@ -431,13 +425,14 @@ func (p *kllProcessor) accumulateKLLSketchMetric(sw *scopeWindow, metric pmetric dps := metric.KLLSketch().DataPoints() for l := 0; l < dps.Len(); l++ { dp := dps.At(l) - attrKey := attributesKey(dp.Attributes()) + if !p.matchesMatchers(dp.Attributes()) { + continue + } + attrKey := p.seriesKey(dp.Attributes()) series := mw.series[attrKey] if series == nil { - attrCopy := pcommon.NewMap() - dp.Attributes().CopyTo(attrCopy) series = p.seriesPool.Get().(*kllSeries) - series.attrs = attrCopy + series.attrs = p.seriesAttrs(dp.Attributes()) if series.sketch != nil { series.sketch.Reset() } else { @@ -555,23 +550,52 @@ func (p *kllProcessor) flushWindow(ctx context.Context) error { if out.ResourceMetrics().Len() == 0 { return nil } - outRmCount := out.ResourceMetrics().Len() - outDpCount := 0 - for i := 0; i < outRmCount; i++ { - for j := 0; j < out.ResourceMetrics().At(i).ScopeMetrics().Len(); j++ { - metrics := out.ResourceMetrics().At(i).ScopeMetrics().At(j).Metrics() - for k := 0; k < metrics.Len(); k++ { - m := metrics.At(k) - if m.Type() == pmetric.MetricTypeGauge { - outDpCount += m.Gauge().DataPoints().Len() - } - } + p.recordOutput(ctx, out) + return p.nextConsumer.ConsumeMetrics(ctx, out) +} + +func (p *kllProcessor) enableSelfMonitoring(settings component.TelemetrySettings, processorID string) { + monitor, err := selfmonitor.New(settings, processorID, "KLL", p.activeSeriesCount) + if err != nil { + if p.logger != nil { + p.logger.Warn("kllprocessor: failed to initialize self-monitoring", zap.Error(err)) } + return } - if p.logger != nil { - p.logger.Debug("KLL processor sending window flush", zap.Int("resource_metrics", outRmCount), zap.Int("data_points", outDpCount)) + p.monitor = monitor +} + +func (p *kllProcessor) shutdownMonitor() { + if p.monitor != nil { + p.monitor.Shutdown() } - return p.nextConsumer.ConsumeMetrics(ctx, out) +} + +func (p *kllProcessor) recordInput(ctx context.Context, md pmetric.Metrics) { + if p.monitor != nil { + p.monitor.RecordInput(ctx, md) + } +} + +func (p *kllProcessor) recordOutput(ctx context.Context, md pmetric.Metrics) { + if p.monitor != nil { + p.monitor.RecordOutput(ctx, md) + } +} + +func (p *kllProcessor) activeSeriesCount() int64 { + p.mu.Lock() + defer p.mu.Unlock() + + var total int64 + for _, rw := range p.windowStore { + for _, sw := range rw.scopes { + for _, mw := range sw.metrics { + total += int64(len(mw.series)) + } + } + } + return total } func findOrCreateGaugeMetric(metrics pmetric.MetricSlice, name, unit string) pmetric.Metric { From 6ef24b0e72a5f8cdb8f313ee86f0c91dfb3f35ac Mon Sep 17 00:00:00 2001 From: Gnanesh Date: Tue, 24 Mar 2026 20:00:32 -0400 Subject: [PATCH 3/3] rename pathway references --- ...g-otlp-pathway1.yaml => config-otlp-raw-metrics-flow.yaml} | 4 ++-- ...-batch.yaml => config-otlp-sketch-payload-flow-batch.yaml} | 4 ++-- ...indow.yaml => config-otlp-sketch-payload-flow-window.yaml} | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) rename opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/{config-otlp-pathway1.yaml => config-otlp-raw-metrics-flow.yaml} (83%) rename opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/{config-otlp-pathway2-batch.yaml => config-otlp-sketch-payload-flow-batch.yaml} (82%) rename opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/{config-otlp-pathway2-window.yaml => config-otlp-sketch-payload-flow-window.yaml} (82%) diff --git a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway1.yaml b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-raw-metrics-flow.yaml similarity index 83% rename from opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway1.yaml rename to opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-raw-metrics-flow.yaml index fc97a86e..35510dc2 100644 --- a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway1.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-raw-metrics-flow.yaml @@ -1,4 +1,4 @@ -# OTLP Pathway 1 - Local testing +# OTLP Raw Metrics Flow - Local testing receivers: otlp: protocols: @@ -28,4 +28,4 @@ service: logs: level: info resource: - service.name: kllcol-otlp-pathway1-local + service.name: kllcol-otlp-raw-metrics-flow-local diff --git a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-batch.yaml b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-sketch-payload-flow-batch.yaml similarity index 82% rename from opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-batch.yaml rename to opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-sketch-payload-flow-batch.yaml index 2675f507..154ef95a 100644 --- a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-batch.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-sketch-payload-flow-batch.yaml @@ -1,4 +1,4 @@ -# OTLP Pathway 2 - Batch mode - Local testing +# OTLP Sketch Payload Flow - Batch mode - Local testing receivers: otlp: protocols: @@ -32,4 +32,4 @@ service: logs: level: debug resource: - service.name: kllcol-otlp-pathway2-local + service.name: kllcol-otlp-sketch-payload-flow-local diff --git a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-window.yaml b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-sketch-payload-flow-window.yaml similarity index 82% rename from opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-window.yaml rename to opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-sketch-payload-flow-window.yaml index 76bf4d02..22857dd5 100644 --- a/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-pathway2-window.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/kll/asap_query_config/config-otlp-sketch-payload-flow-window.yaml @@ -1,4 +1,4 @@ -# OTLP Pathway 2 - Window mode - Local testing +# OTLP Sketch Payload Flow - Window mode - Local testing receivers: otlp: protocols: @@ -33,4 +33,4 @@ service: logs: level: debug resource: - service.name: kllcol-otlp-pathway2-window-local + service.name: kllcol-otlp-sketch-payload-flow-window-local