From d26462f5dfaf1ad231a4be6c34671e8e1dc4b2c0 Mon Sep 17 00:00:00 2001 From: Gnanesh Date: Tue, 24 Mar 2026 00:07:36 -0400 Subject: [PATCH 1/3] Update CountMinSketch processor sketch serialization --- .../config-otlp-pathway1.yaml | 29 ++ .../config-otlp-pathway2-batch.yaml | 35 ++ .../config-otlp-pathway2-window.yaml | 38 +++ .../cmd/countminsketchcol/builder-config.yaml | 1 + .../processor/countminsketchprocessor/go.mod | 6 +- .../countminsketchprocessor/processor.go | 315 ++++-------------- .../countminsketchprocessor/processor_test.go | 4 +- 7 files changed, 179 insertions(+), 249 deletions(-) create mode 100644 opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml create mode 100644 opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-batch.yaml create mode 100644 opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-window.yaml diff --git a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml new file mode 100644 index 00000000..06d16176 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml @@ -0,0 +1,29 @@ +# 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: + logs: + level: debug + resource: + service.name: countminsketchcol-pathway1-local diff --git a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-batch.yaml b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-batch.yaml new file mode 100644 index 00000000..a09adf31 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/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: + countmin: + mode: batch + metric_name: "cms_sketch" + rows: 5 + columns: 2000 + transmit_sketch: true + group_by: [] + drop_original: true + +exporters: + otlphttp: + endpoint: http://localhost:4318 + encoding: proto + compression: gzip +service: + pipelines: + metrics: + receivers: [otlp] + processors: [countmin] + exporters: [otlphttp] + telemetry: + logs: + level: debug + resource: + service.name: countminsketchcol-pathway2-batch-local diff --git a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-window.yaml b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-window.yaml new file mode 100644 index 00000000..09987c44 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-window.yaml @@ -0,0 +1,38 @@ +# 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: + countmin: + mode: window + metric_name: "cms_sketch" + rows: 5 + columns: 2000 + transmit_sketch: true + group_by: [] + drop_original: true + window_interval: 60s + +exporters: + otlphttp: + endpoint: http://localhost:4318 + encoding: proto + compression: gzip +service: + pipelines: + metrics: + receivers: [otlp] + processors: [countmin] + exporters: [otlphttp] + telemetry: + metrics: + level: basic + logs: + level: debug + resource: + service.name: countminsketchcol-otlp-pathway2-local diff --git a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/builder-config.yaml b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/builder-config.yaml index 05a4bee2..1620a1ef 100644 --- a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/builder-config.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/builder-config.yaml @@ -280,3 +280,4 @@ replaces: # - github.com/open-telemetry/opentelemetry-collector-contrib/processor/countminsketchprocessor => /mnt/78D8516BD8512920/GARUDA_ACE/FROOT-LAB/asap-internal/DataCollector/opentelemetry-collector-contrib/processor/countminsketchprocessor - github.com/open-telemetry/opentelemetry-collector-contrib/processor/countminsketchprocessor => ../../../processor/countminsketchprocessor - go.opentelemetry.io/collector/pdata => ../../../../opentelemetry-collector/pdata + # - github.com/ProjectASAP/sketchlib-go => ../../../../../sketchlib-go diff --git a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/go.mod b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/go.mod index 92b625fe..cd6f0450 100644 --- a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/go.mod +++ b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/go.mod @@ -3,7 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/count go 1.24.0 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.47.0 go.opentelemetry.io/collector/component/componenttest v0.141.0 @@ -46,13 +46,13 @@ require ( go.yaml.in/yaml/v2 v2.4.3 // indirect golang.org/x/sys v0.38.0 // indirect golang.org/x/text v0.30.0 // indirect - google.golang.org/protobuf v1.36.11 // indirect + google.golang.org/protobuf v1.36.10 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) 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 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../../pkg/pdatatest diff --git a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go index f31bbf83..392c1439 100644 --- a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go @@ -13,19 +13,19 @@ import ( "github.com/ProjectASAP/sketchlib-go/common" cms "github.com/ProjectASAP/sketchlib-go/sketches/CountMinSketch" - "go.opentelemetry.io/collector/component" "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 used in the hot // encodeAttributesAsKey path (called on every data point). var builderPool = sync.Pool{New: func() any { return new(strings.Builder) }} + // // ───────────────────────────────────────────────────────────── // Window-level state @@ -34,7 +34,6 @@ var builderPool = sync.Pool{New: func() any { return new(strings.Builder) }} type windowSketch struct { cms *cms.CountMinSketch - attrs pcommon.Map // output attributes for this sketch group mu sync.Mutex sampleCount uint64 } @@ -50,23 +49,10 @@ type windowedCountMinSketchProcessor struct { logger *zap.Logger nextConsumer consumer.Metrics - monitor *selfmonitor.Monitor activeWindowSketches map[string]*windowSketch mu sync.RWMutex - // snapshots holds one CMS clone per partition key, taken at the end of - // each window flush. Used to compute sparse delta payloads when - // cfg.DeltaTransmission=true. - snapshots map[string]*cms.CountMinSketch - snapshotsMu sync.Mutex - - // inboundSnapshots tracks the last reconstructed full CMS per aggregation key - // received from upstream. Used to apply sparse deltas from SDK-originated - // CountMinSketchEncodingDelta payloads. - inboundMu sync.Mutex - inboundSnapshots map[string]*cms.CountMinSketch - stopCh chan struct{} doneCh chan struct{} windowStarted atomic.Bool @@ -86,8 +72,6 @@ func newProcessor( logger: logger, nextConsumer: next, activeWindowSketches: make(map[string]*windowSketch), - snapshots: make(map[string]*cms.CountMinSketch), - inboundSnapshots: make(map[string]*cms.CountMinSketch), stopCh: make(chan struct{}), doneCh: make(chan struct{}), } @@ -108,7 +92,7 @@ func (p *windowedCountMinSketchProcessor) Start( p.logger.Info( "Starting Count-Min Sketch processor", zap.String("mode", string(p.cfg.Mode)), - zap.Duration("window_duration", p.cfg.WindowDuration), + zap.Duration("window_interval", p.cfg.WindowInterval), ) // Batch mode does not require a background ticker; we flush per-batch. @@ -117,11 +101,11 @@ func (p *windowedCountMinSketchProcessor) Start( } // Window mode: start background window loop if a positive window is configured. - if p.cfg.WindowDuration <= 0 { + if p.cfg.WindowInterval <= 0 { return nil } - ticker := time.NewTicker(p.cfg.WindowDuration) + ticker := time.NewTicker(p.cfg.WindowInterval) p.windowStarted.Store(true) go func() { @@ -150,8 +134,6 @@ func (p *windowedCountMinSketchProcessor) Start( func (p *windowedCountMinSketchProcessor) Shutdown( ctx context.Context, ) error { - defer p.shutdownMonitor() - // Only wait if the window goroutine was actually started; avoids blocking // forever when Start was never called. if p.cfg.Mode != ModeWindow || !p.windowStarted.Load() { @@ -183,17 +165,36 @@ func (p *windowedCountMinSketchProcessor) ConsumeMetrics( ctx context.Context, md pmetric.Metrics, ) (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.MetricTypeSum: + dpCount += m.Sum().DataPoints().Len() + case pmetric.MetricTypeCountMinSketch: + dpCount += m.CountMinSketch().DataPoints().Len() + default: + // ignore + } + } + } + } + if p.logger != nil { + p.logger.Debug("CountMinSketch processor received data", zap.Int("resource_metrics", rmCount), zap.Int("data_points", dpCount)) + } switch p.cfg.Mode { case ModeBatch: - out := p.consumeBatch(md) - p.recordOutput(ctx, out) - return out, nil + return p.consumeBatch(md), nil case ModeWindow: p.accumulateIntoWindow(md) if !p.cfg.DropOriginal { - p.recordOutput(ctx, md) return md, nil } return pmetric.NewMetrics(), nil @@ -230,10 +231,12 @@ func (p *windowedCountMinSketchProcessor) consumeBatch(md pmetric.Metrics) pmetr if p.cfg.DropOriginal { return pmetric.NewMetrics() } + p.logBatchOutput("batch passthrough (no sketches)", md) return md } if p.cfg.DropOriginal { + p.logBatchOutput("batch output", sketches) return sketches } @@ -241,6 +244,7 @@ func (p *windowedCountMinSketchProcessor) consumeBatch(md pmetric.Metrics) pmetr out := pmetric.NewMetrics() md.ResourceMetrics().MoveAndAppendTo(out.ResourceMetrics()) sketches.ResourceMetrics().MoveAndAppendTo(out.ResourceMetrics()) + p.logBatchOutput("batch output (expansion)", out) return out } @@ -251,107 +255,35 @@ func (p *windowedCountMinSketchProcessor) ingestMetric( case pmetric.MetricTypeGauge: dps := metric.Gauge().DataPoints() for i := 0; i < dps.Len(); i++ { - dp := dps.At(i) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - p.updateWindowSketch(metric.Name(), dp) + p.updateWindowSketch(metric.Name(), dps.At(i)) } case pmetric.MetricTypeSum: dps := metric.Sum().DataPoints() for i := 0; i < dps.Len(); i++ { - dp := dps.At(i) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - p.updateWindowSketch(metric.Name(), dp) + p.updateWindowSketch(metric.Name(), dps.At(i)) } case pmetric.MetricTypeCountMinSketch: - // Pre-aggregated path: deserialize (or reconstruct from delta) and merge - // each incoming sketch into the per-aggregation-key window sketch. + // Pre-aggregated path: deserialize and merge each incoming sketch into + // the corresponding per-aggregation-key window sketch. dps := metric.CountMinSketch().DataPoints() for i := 0; i < dps.Len(); i++ { dp := dps.At(i) - if !p.matchesMatchers(dp.Attributes()) { - continue - } if len(dp.Sketch()) == 0 { continue } - aggregationKey := p.seriesKey(metric.Name(), dp.Attributes()) - incoming, err := p.inboundDecodeCMS(aggregationKey, dp) + incoming, err := deserializeCMS(dp.Sketch()) if err != nil { - p.logger.Error("countminsketchprocessor: failed to decode inbound CountMinSketch", zap.Error(err)) + p.logger.Error("countminsketchprocessor: failed to deserialize CountMinSketch dp", zap.Error(err)) continue } - if incoming == nil { - continue // delta with no snapshot yet - } - p.mergeWindowSketch(aggregationKey, dp.Attributes(), incoming) + aggregationKey := buildAggregationKey(metric.Name(), dp.Attributes()) + p.mergeWindowSketch(aggregationKey, incoming) } } } -// matchesMatchers returns true if attrs satisfies all configured LabelMatchers. -func (p *windowedCountMinSketchProcessor) 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 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 *windowedCountMinSketchProcessor) seriesKey(metricName string, attrs pcommon.Map) string { - return metricName + "::" + p.encodeKey(attrs) -} - -// encodeKey builds a stable string from the labels that form the grouping key. -// When AggregateBy is set, only those keys are used; otherwise all attributes. -func (p *windowedCountMinSketchProcessor) encodeKey(attrs pcommon.Map) string { - if len(p.cfg.AggregateBy) > 0 { - sb := builderPool.Get().(*strings.Builder) - sb.Reset() - for _, k := range p.cfg.AggregateBy { // already sorted by Validate - v, ok := attrs.Get(k) - if !ok { - continue - } - sb.WriteString(k) - sb.WriteString("=") - sb.WriteString(v.AsString()) - sb.WriteString(";") - } - s := sb.String() - builderPool.Put(sb) - return s - } - return encodeAttributesAsKey(attrs) -} - -// seriesAttrs returns the attribute map to store on a new sketch group. -// When AggregateBy is configured, only those labels are included in the output. -// Otherwise a full copy of attrs is returned. -func (p *windowedCountMinSketchProcessor) seriesAttrs(attrs pcommon.Map) pcommon.Map { - out := pcommon.NewMap() - if len(p.cfg.AggregateBy) > 0 { - for _, k := range p.cfg.AggregateBy { - if v, ok := attrs.Get(k); ok { - out.PutStr(k, v.AsString()) - } - } - return out - } - attrs.CopyTo(out) - return out -} - // mergeWindowSketch merges an incoming pre-aggregated CMS into the per-key window store. -func (p *windowedCountMinSketchProcessor) mergeWindowSketch(aggregationKey string, attrs pcommon.Map, incoming *cms.CountMinSketch) { +func (p *windowedCountMinSketchProcessor) mergeWindowSketch(aggregationKey string, incoming *cms.CountMinSketch) { p.mu.RLock() ws, exists := p.activeWindowSketches[aggregationKey] p.mu.RUnlock() @@ -373,7 +305,6 @@ func (p *windowedCountMinSketchProcessor) mergeWindowSketch(aggregationKey strin } ws.cms = newCMS } - ws.attrs = p.seriesAttrs(attrs) ws.sampleCount = 0 p.activeWindowSketches[aggregationKey] = ws } @@ -391,7 +322,7 @@ func (p *windowedCountMinSketchProcessor) updateWindowSketch( metricName string, dp pmetric.NumberDataPoint, ) { - aggregationKey := p.seriesKey(metricName, dp.Attributes()) + aggregationKey := buildAggregationKey(metricName, dp.Attributes()) p.mu.RLock() ws, exists := p.activeWindowSketches[aggregationKey] @@ -414,7 +345,6 @@ func (p *windowedCountMinSketchProcessor) updateWindowSketch( } ws.cms = newCMS } - ws.attrs = p.seriesAttrs(dp.Attributes()) ws.sampleCount = 0 p.activeWindowSketches[aggregationKey] = ws } @@ -472,47 +402,11 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M rows := ws.cms.Rows cols := ws.cms.Cols sampleCount := ws.sampleCount - outputAttrs := ws.attrs - - var payload []byte - var encoding string - var err error - - if p.cfg.TransmitSketch && p.cfg.DeltaTransmission { - // Delta path: compute sparse diff against the last snapshot. - p.snapshotsMu.Lock() - snap, hasSnap := p.snapshots[aggregationKey] - p.snapshotsMu.Unlock() - - if hasSnap { - deltaMsg, deltaErr := cms.ComputeDelta(snap, ws.cms, p.cfg.DeltaThreshold) - if deltaErr == nil { - payload, err = cms.SerializeDelta(deltaMsg) - } else { - err = deltaErr - } - encoding = "proto_delta" - } else { - // First window for this partition — send full sketch. - payload, err = serializeCMS(ws.cms) - encoding = "proto_full" - } - - // Update snapshot to current state (clone before pool return). - newSnap := cloneCMS(ws.cms) - p.snapshotsMu.Lock() - p.snapshots[aggregationKey] = newSnap - p.snapshotsMu.Unlock() - } else if p.cfg.TransmitSketch { - payload, err = serializeCMS(ws.cms) - encoding = "proto_full" - } - + payload, err := serializeCMS(ws.cms) ws.mu.Unlock() p.windowSketchPool.Put(ws) - - if p.cfg.TransmitSketch && err != nil { - p.logger.Error("Failed to serialize/delta CMS", zap.Error(err)) + if err != nil { + p.logger.Error("Failed to serialize CMS", zap.Error(err)) continue } @@ -524,13 +418,12 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M dp := gauge.DataPoints().AppendEmpty() dp.SetTimestamp(now) - outputAttrs.CopyTo(dp.Attributes()) + dp.Attributes().PutStr("aggregation_key", aggregationKey) dp.Attributes().PutInt("rows", int64(rows)) dp.Attributes().PutInt("cols", int64(cols)) dp.Attributes().PutInt("sample_count", int64(sampleCount)) if p.cfg.TransmitSketch { - dp.Attributes().PutEmptyBytes("sketch_payload").FromRaw(payload) - dp.Attributes().PutStr("encoding", encoding) + dp.Attributes().PutEmptyBytes("cms.sketch_payload").FromRaw(payload) } else { dp.SetDoubleValue(float64(sampleCount)) } @@ -544,115 +437,49 @@ func (p *windowedCountMinSketchProcessor) emitWindowAndReset() { if md.ResourceMetrics().Len() == 0 { return } - - p.recordOutput(context.Background(), md) + if p.logger != nil { + p.logBatchOutput("window flush", md) + } if err := p.nextConsumer.ConsumeMetrics(context.Background(), md); err != nil { p.logger.Error("Failed to emit windowed CMS", zap.Error(err)) } } -func (p *windowedCountMinSketchProcessor) enableSelfMonitoring(settings component.TelemetrySettings, processorID string) { - monitor, err := selfmonitor.New(settings, processorID, "countmin", p.activeSeriesCount) - if err != nil { - if p.logger != nil { - p.logger.Warn("countminsketchprocessor: failed to initialize self-monitoring", zap.Error(err)) - } - return - } - p.monitor = monitor -} - -func (p *windowedCountMinSketchProcessor) shutdownMonitor() { - if p.monitor != nil { - p.monitor.Shutdown() - } -} - -func (p *windowedCountMinSketchProcessor) recordInput(ctx context.Context, md pmetric.Metrics) { - if p.monitor != nil { - p.monitor.RecordInput(ctx, md) - } -} - -func (p *windowedCountMinSketchProcessor) recordOutput(ctx context.Context, md pmetric.Metrics) { - if p.monitor != nil { - p.monitor.RecordOutput(ctx, md) - } -} - -func (p *windowedCountMinSketchProcessor) activeSeriesCount() int64 { - p.mu.RLock() - defer p.mu.RUnlock() - return int64(len(p.activeWindowSketches)) -} - // // ───────────────────────────────────────────────────────────── // Helpers // ───────────────────────────────────────────────────────────── // -// inboundDecodeCMS decodes an incoming CountMinSketch data point, handling both -// full (Gob-encoded) and sparse-delta payloads. For delta payloads it applies -// the delta onto the last stored snapshot to reconstruct the current full state. -// Returns (nil, nil) when a delta arrives before any full snapshot. -func (p *windowedCountMinSketchProcessor) inboundDecodeCMS(aggregationKey string, dp pmetric.CountMinSketchDataPoint) (*cms.CountMinSketch, error) { - payload := dp.Sketch() - - switch dp.Encoding() { - case pmetric.CountMinSketchEncodingDelta: - p.inboundMu.Lock() - snap, hasSnap := p.inboundSnapshots[aggregationKey] - p.inboundMu.Unlock() - if !hasSnap || snap == nil { - return nil, nil - } - reconstructed := cloneCMS(snap) - if reconstructed == nil { - return nil, nil - } - deltaMsg, err := cms.DeserializeDelta(payload) - if err != nil { - return nil, err - } - cms.ApplyDelta(reconstructed, deltaMsg) - p.inboundMu.Lock() - p.inboundSnapshots[aggregationKey] = cloneCMS(reconstructed) - p.inboundMu.Unlock() - return reconstructed, nil - - default: // CountMinSketchEncodingProto or unspecified - decoded, err := deserializeCMS(payload) - if err != nil { - return nil, err +func (p *windowedCountMinSketchProcessor) logBatchOutput(kind string, md pmetric.Metrics) { + 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) + if m.Type() == pmetric.MetricTypeGauge { + dpCount += m.Gauge().DataPoints().Len() + } + } } - p.inboundMu.Lock() - p.inboundSnapshots[aggregationKey] = cloneCMS(decoded) - p.inboundMu.Unlock() - return decoded, nil + } + if p.logger != nil { + p.logger.Debug("CountMinSketch processor sending "+kind, zap.Int("resource_metrics", rmCount), zap.Int("data_points", dpCount)) } } func serializeCMS(s *cms.CountMinSketch) ([]byte, error) { - return s.SerializeProtoBytes() + env, err := s.SerializePortable() + if err != nil { + return nil, err + } + return proto.Marshal(env) } func deserializeCMS(data []byte) (*cms.CountMinSketch, error) { - return cms.DeserializeCountMinSketchFromProtoBytes(data) -} - -// cloneCMS returns a deep copy of s suitable for use as a delta snapshot. -// It serializes and deserializes to ensure full independence from the original. -func cloneCMS(s *cms.CountMinSketch) *cms.CountMinSketch { - data, err := s.SerializeProtoBytes() - if err != nil { - return nil - } - clone, err := cms.DeserializeCountMinSketchFromProtoBytes(data) - if err != nil { - return nil - } - return clone + return cms.DeserializeCountMinSketchFromBytes(data) } func buildAggregationKey( diff --git a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor_test.go b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor_test.go index c868650a..cf31e918 100644 --- a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor_test.go +++ b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor_test.go @@ -180,7 +180,7 @@ func TestProcessor_TumblingWindow_Correctness(t *testing.T) { // ========================================== // 3. Verify Binary Payload (Gob Decode) // ========================================== - payloadVal, ok := dps2[0].Attributes().Get("sketch_payload") + payloadVal, ok := dps2[0].Attributes().Get("cms.sketch_payload") require.True(t, ok, "Sketch payload must exist in attributes") rawBytes := payloadVal.Bytes().AsRaw() @@ -301,7 +301,7 @@ func TestBatchModeQueryMetricsWhenTransmitSketchDisabled(t *testing.T) { dps := getAllDataPoints(out) require.Len(t, dps, 1) assert.Equal(t, 3.0, dps[0].DoubleValue()) - _, hasPayload := dps[0].Attributes().Get("sketch_payload") + _, hasPayload := dps[0].Attributes().Get("cms.sketch_payload") assert.False(t, hasPayload) } From 277c03b6770aa59498e9b3eb49adfc2cfff9f67c Mon Sep 17 00:00:00 2001 From: Gnanesh Date: Tue, 24 Mar 2026 00:30:10 -0400 Subject: [PATCH 2/3] Restore CountMinSketch non-serialization logic --- .../countminsketchprocessor/processor.go | 306 ++++++++++++++---- 1 file changed, 242 insertions(+), 64 deletions(-) diff --git a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go index 392c1439..a6797e55 100644 --- a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go @@ -13,10 +13,12 @@ import ( "github.com/ProjectASAP/sketchlib-go/common" cms "github.com/ProjectASAP/sketchlib-go/sketches/CountMinSketch" + "go.opentelemetry.io/collector/component" "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" ) @@ -25,7 +27,6 @@ import ( // encodeAttributesAsKey path (called on every data point). var builderPool = sync.Pool{New: func() any { return new(strings.Builder) }} - // // ───────────────────────────────────────────────────────────── // Window-level state @@ -34,6 +35,7 @@ var builderPool = sync.Pool{New: func() any { return new(strings.Builder) }} type windowSketch struct { cms *cms.CountMinSketch + attrs pcommon.Map // output attributes for this sketch group mu sync.Mutex sampleCount uint64 } @@ -49,10 +51,23 @@ type windowedCountMinSketchProcessor struct { logger *zap.Logger nextConsumer consumer.Metrics + monitor *selfmonitor.Monitor activeWindowSketches map[string]*windowSketch mu sync.RWMutex + // snapshots holds one CMS clone per partition key, taken at the end of + // each window flush. Used to compute sparse delta payloads when + // cfg.DeltaTransmission=true. + snapshots map[string]*cms.CountMinSketch + snapshotsMu sync.Mutex + + // inboundSnapshots tracks the last reconstructed full CMS per aggregation key + // received from upstream. Used to apply sparse deltas from SDK-originated + // CountMinSketchEncodingDelta payloads. + inboundMu sync.Mutex + inboundSnapshots map[string]*cms.CountMinSketch + stopCh chan struct{} doneCh chan struct{} windowStarted atomic.Bool @@ -72,6 +87,8 @@ func newProcessor( logger: logger, nextConsumer: next, activeWindowSketches: make(map[string]*windowSketch), + snapshots: make(map[string]*cms.CountMinSketch), + inboundSnapshots: make(map[string]*cms.CountMinSketch), stopCh: make(chan struct{}), doneCh: make(chan struct{}), } @@ -92,7 +109,7 @@ func (p *windowedCountMinSketchProcessor) Start( p.logger.Info( "Starting Count-Min Sketch processor", zap.String("mode", string(p.cfg.Mode)), - zap.Duration("window_interval", p.cfg.WindowInterval), + zap.Duration("window_duration", p.cfg.WindowDuration), ) // Batch mode does not require a background ticker; we flush per-batch. @@ -101,11 +118,11 @@ func (p *windowedCountMinSketchProcessor) Start( } // Window mode: start background window loop if a positive window is configured. - if p.cfg.WindowInterval <= 0 { + if p.cfg.WindowDuration <= 0 { return nil } - ticker := time.NewTicker(p.cfg.WindowInterval) + ticker := time.NewTicker(p.cfg.WindowDuration) p.windowStarted.Store(true) go func() { @@ -134,6 +151,8 @@ func (p *windowedCountMinSketchProcessor) Start( func (p *windowedCountMinSketchProcessor) Shutdown( ctx context.Context, ) error { + defer p.shutdownMonitor() + // Only wait if the window goroutine was actually started; avoids blocking // forever when Start was never called. if p.cfg.Mode != ModeWindow || !p.windowStarted.Load() { @@ -165,36 +184,17 @@ func (p *windowedCountMinSketchProcessor) ConsumeMetrics( ctx context.Context, md pmetric.Metrics, ) (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.MetricTypeSum: - dpCount += m.Sum().DataPoints().Len() - case pmetric.MetricTypeCountMinSketch: - dpCount += m.CountMinSketch().DataPoints().Len() - default: - // ignore - } - } - } - } - if p.logger != nil { - p.logger.Debug("CountMinSketch processor received data", zap.Int("resource_metrics", rmCount), zap.Int("data_points", dpCount)) - } + p.recordInput(ctx, md) switch p.cfg.Mode { case ModeBatch: - return p.consumeBatch(md), nil + out := p.consumeBatch(md) + p.recordOutput(ctx, out) + return out, nil case ModeWindow: p.accumulateIntoWindow(md) if !p.cfg.DropOriginal { + p.recordOutput(ctx, md) return md, nil } return pmetric.NewMetrics(), nil @@ -231,12 +231,10 @@ func (p *windowedCountMinSketchProcessor) consumeBatch(md pmetric.Metrics) pmetr if p.cfg.DropOriginal { return pmetric.NewMetrics() } - p.logBatchOutput("batch passthrough (no sketches)", md) return md } if p.cfg.DropOriginal { - p.logBatchOutput("batch output", sketches) return sketches } @@ -244,7 +242,6 @@ func (p *windowedCountMinSketchProcessor) consumeBatch(md pmetric.Metrics) pmetr out := pmetric.NewMetrics() md.ResourceMetrics().MoveAndAppendTo(out.ResourceMetrics()) sketches.ResourceMetrics().MoveAndAppendTo(out.ResourceMetrics()) - p.logBatchOutput("batch output (expansion)", out) return out } @@ -255,35 +252,107 @@ func (p *windowedCountMinSketchProcessor) ingestMetric( case pmetric.MetricTypeGauge: dps := metric.Gauge().DataPoints() for i := 0; i < dps.Len(); i++ { - p.updateWindowSketch(metric.Name(), dps.At(i)) + dp := dps.At(i) + if !p.matchesMatchers(dp.Attributes()) { + continue + } + p.updateWindowSketch(metric.Name(), dp) } case pmetric.MetricTypeSum: dps := metric.Sum().DataPoints() for i := 0; i < dps.Len(); i++ { - p.updateWindowSketch(metric.Name(), dps.At(i)) + dp := dps.At(i) + if !p.matchesMatchers(dp.Attributes()) { + continue + } + p.updateWindowSketch(metric.Name(), dp) } case pmetric.MetricTypeCountMinSketch: - // Pre-aggregated path: deserialize and merge each incoming sketch into - // the corresponding per-aggregation-key window sketch. + // Pre-aggregated path: deserialize (or reconstruct from delta) and merge + // each incoming sketch into the per-aggregation-key window sketch. dps := metric.CountMinSketch().DataPoints() for i := 0; i < dps.Len(); i++ { dp := dps.At(i) + if !p.matchesMatchers(dp.Attributes()) { + continue + } if len(dp.Sketch()) == 0 { continue } - incoming, err := deserializeCMS(dp.Sketch()) + aggregationKey := p.seriesKey(metric.Name(), dp.Attributes()) + incoming, err := p.inboundDecodeCMS(aggregationKey, dp) if err != nil { - p.logger.Error("countminsketchprocessor: failed to deserialize CountMinSketch dp", zap.Error(err)) + p.logger.Error("countminsketchprocessor: failed to decode inbound CountMinSketch", zap.Error(err)) + continue + } + if incoming == nil { + continue // delta with no snapshot yet + } + p.mergeWindowSketch(aggregationKey, dp.Attributes(), incoming) + } + } +} + +// matchesMatchers returns true if attrs satisfies all configured LabelMatchers. +func (p *windowedCountMinSketchProcessor) 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 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 *windowedCountMinSketchProcessor) seriesKey(metricName string, attrs pcommon.Map) string { + return metricName + "::" + p.encodeKey(attrs) +} + +// encodeKey builds a stable string from the labels that form the grouping key. +// When AggregateBy is set, only those keys are used; otherwise all attributes. +func (p *windowedCountMinSketchProcessor) encodeKey(attrs pcommon.Map) string { + if len(p.cfg.AggregateBy) > 0 { + sb := builderPool.Get().(*strings.Builder) + sb.Reset() + for _, k := range p.cfg.AggregateBy { // already sorted by Validate + v, ok := attrs.Get(k) + if !ok { continue } - aggregationKey := buildAggregationKey(metric.Name(), dp.Attributes()) - p.mergeWindowSketch(aggregationKey, incoming) + sb.WriteString(k) + sb.WriteString("=") + sb.WriteString(v.AsString()) + sb.WriteString(";") } + s := sb.String() + builderPool.Put(sb) + return s } + return encodeAttributesAsKey(attrs) +} + +// seriesAttrs returns the attribute map to store on a new sketch group. +// When AggregateBy is configured, only those labels are included in the output. +// Otherwise a full copy of attrs is returned. +func (p *windowedCountMinSketchProcessor) seriesAttrs(attrs pcommon.Map) pcommon.Map { + out := pcommon.NewMap() + if len(p.cfg.AggregateBy) > 0 { + for _, k := range p.cfg.AggregateBy { + if v, ok := attrs.Get(k); ok { + out.PutStr(k, v.AsString()) + } + } + return out + } + attrs.CopyTo(out) + return out } // mergeWindowSketch merges an incoming pre-aggregated CMS into the per-key window store. -func (p *windowedCountMinSketchProcessor) mergeWindowSketch(aggregationKey string, incoming *cms.CountMinSketch) { +func (p *windowedCountMinSketchProcessor) mergeWindowSketch(aggregationKey string, attrs pcommon.Map, incoming *cms.CountMinSketch) { p.mu.RLock() ws, exists := p.activeWindowSketches[aggregationKey] p.mu.RUnlock() @@ -305,6 +374,7 @@ func (p *windowedCountMinSketchProcessor) mergeWindowSketch(aggregationKey strin } ws.cms = newCMS } + ws.attrs = p.seriesAttrs(attrs) ws.sampleCount = 0 p.activeWindowSketches[aggregationKey] = ws } @@ -322,7 +392,7 @@ func (p *windowedCountMinSketchProcessor) updateWindowSketch( metricName string, dp pmetric.NumberDataPoint, ) { - aggregationKey := buildAggregationKey(metricName, dp.Attributes()) + aggregationKey := p.seriesKey(metricName, dp.Attributes()) p.mu.RLock() ws, exists := p.activeWindowSketches[aggregationKey] @@ -345,6 +415,7 @@ func (p *windowedCountMinSketchProcessor) updateWindowSketch( } ws.cms = newCMS } + ws.attrs = p.seriesAttrs(dp.Attributes()) ws.sampleCount = 0 p.activeWindowSketches[aggregationKey] = ws } @@ -402,11 +473,47 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M rows := ws.cms.Rows cols := ws.cms.Cols sampleCount := ws.sampleCount - payload, err := serializeCMS(ws.cms) + outputAttrs := ws.attrs + + var payload []byte + var encoding string + var err error + + if p.cfg.TransmitSketch && p.cfg.DeltaTransmission { + // Delta path: compute sparse diff against the last snapshot. + p.snapshotsMu.Lock() + snap, hasSnap := p.snapshots[aggregationKey] + p.snapshotsMu.Unlock() + + if hasSnap { + deltaMsg, deltaErr := cms.ComputeDelta(snap, ws.cms, p.cfg.DeltaThreshold) + if deltaErr == nil { + payload, err = cms.SerializeDelta(deltaMsg) + } else { + err = deltaErr + } + encoding = "proto_delta" + } else { + // First window for this partition — send full sketch. + payload, err = serializeCMS(ws.cms) + encoding = "proto_full" + } + + // Update snapshot to current state (clone before pool return). + newSnap := cloneCMS(ws.cms) + p.snapshotsMu.Lock() + p.snapshots[aggregationKey] = newSnap + p.snapshotsMu.Unlock() + } else if p.cfg.TransmitSketch { + payload, err = serializeCMS(ws.cms) + encoding = "proto_full" + } + ws.mu.Unlock() p.windowSketchPool.Put(ws) - if err != nil { - p.logger.Error("Failed to serialize CMS", zap.Error(err)) + + if p.cfg.TransmitSketch && err != nil { + p.logger.Error("Failed to serialize/delta CMS", zap.Error(err)) continue } @@ -418,12 +525,13 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M dp := gauge.DataPoints().AppendEmpty() dp.SetTimestamp(now) - dp.Attributes().PutStr("aggregation_key", aggregationKey) + outputAttrs.CopyTo(dp.Attributes()) dp.Attributes().PutInt("rows", int64(rows)) dp.Attributes().PutInt("cols", int64(cols)) dp.Attributes().PutInt("sample_count", int64(sampleCount)) if p.cfg.TransmitSketch { - dp.Attributes().PutEmptyBytes("cms.sketch_payload").FromRaw(payload) + dp.Attributes().PutEmptyBytes("sketch_payload").FromRaw(payload) + dp.Attributes().PutStr("encoding", encoding) } else { dp.SetDoubleValue(float64(sampleCount)) } @@ -437,36 +545,92 @@ func (p *windowedCountMinSketchProcessor) emitWindowAndReset() { if md.ResourceMetrics().Len() == 0 { return } - if p.logger != nil { - p.logBatchOutput("window flush", md) - } + + p.recordOutput(context.Background(), md) if err := p.nextConsumer.ConsumeMetrics(context.Background(), md); err != nil { p.logger.Error("Failed to emit windowed CMS", zap.Error(err)) } } +func (p *windowedCountMinSketchProcessor) enableSelfMonitoring(settings component.TelemetrySettings, processorID string) { + monitor, err := selfmonitor.New(settings, processorID, "countmin", p.activeSeriesCount) + if err != nil { + if p.logger != nil { + p.logger.Warn("countminsketchprocessor: failed to initialize self-monitoring", zap.Error(err)) + } + return + } + p.monitor = monitor +} + +func (p *windowedCountMinSketchProcessor) shutdownMonitor() { + if p.monitor != nil { + p.monitor.Shutdown() + } +} + +func (p *windowedCountMinSketchProcessor) recordInput(ctx context.Context, md pmetric.Metrics) { + if p.monitor != nil { + p.monitor.RecordInput(ctx, md) + } +} + +func (p *windowedCountMinSketchProcessor) recordOutput(ctx context.Context, md pmetric.Metrics) { + if p.monitor != nil { + p.monitor.RecordOutput(ctx, md) + } +} + +func (p *windowedCountMinSketchProcessor) activeSeriesCount() int64 { + p.mu.RLock() + defer p.mu.RUnlock() + return int64(len(p.activeWindowSketches)) +} + // // ───────────────────────────────────────────────────────────── // Helpers // ───────────────────────────────────────────────────────────── // -func (p *windowedCountMinSketchProcessor) logBatchOutput(kind string, md pmetric.Metrics) { - 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) - if m.Type() == pmetric.MetricTypeGauge { - dpCount += m.Gauge().DataPoints().Len() - } - } +// inboundDecodeCMS decodes an incoming CountMinSketch data point, handling both +// full (Gob-encoded) and sparse-delta payloads. For delta payloads it applies +// the delta onto the last stored snapshot to reconstruct the current full state. +// Returns (nil, nil) when a delta arrives before any full snapshot. +func (p *windowedCountMinSketchProcessor) inboundDecodeCMS(aggregationKey string, dp pmetric.CountMinSketchDataPoint) (*cms.CountMinSketch, error) { + payload := dp.Sketch() + + switch dp.Encoding() { + case pmetric.CountMinSketchEncodingDelta: + p.inboundMu.Lock() + snap, hasSnap := p.inboundSnapshots[aggregationKey] + p.inboundMu.Unlock() + if !hasSnap || snap == nil { + return nil, nil } - } - if p.logger != nil { - p.logger.Debug("CountMinSketch processor sending "+kind, zap.Int("resource_metrics", rmCount), zap.Int("data_points", dpCount)) + reconstructed := cloneCMS(snap) + if reconstructed == nil { + return nil, nil + } + deltaMsg, err := cms.DeserializeDelta(payload) + if err != nil { + return nil, err + } + cms.ApplyDelta(reconstructed, deltaMsg) + p.inboundMu.Lock() + p.inboundSnapshots[aggregationKey] = cloneCMS(reconstructed) + p.inboundMu.Unlock() + return reconstructed, nil + + default: // CountMinSketchEncodingProto or unspecified + decoded, err := deserializeCMS(payload) + if err != nil { + return nil, err + } + p.inboundMu.Lock() + p.inboundSnapshots[aggregationKey] = cloneCMS(decoded) + p.inboundMu.Unlock() + return decoded, nil } } @@ -482,6 +646,20 @@ func deserializeCMS(data []byte) (*cms.CountMinSketch, error) { return cms.DeserializeCountMinSketchFromBytes(data) } +// cloneCMS returns a deep copy of s suitable for use as a delta snapshot. +// It serializes and deserializes to ensure full independence from the original. +func cloneCMS(s *cms.CountMinSketch) *cms.CountMinSketch { + data, err := s.SerializeProtoBytes() + if err != nil { + return nil + } + clone, err := cms.DeserializeCountMinSketchFromBytes(data) + if err != nil { + return nil + } + return clone +} + func buildAggregationKey( metricName string, attrs pcommon.Map, From 8a3839af145d2e11e2937f76083f0c5a1a399f34 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/countminsketchcol/asap_query_config/{config-otlp-pathway1.yaml => config-otlp-raw-metrics-flow.yaml} (81%) rename opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/{config-otlp-pathway2-batch.yaml => config-otlp-sketch-payload-flow-batch.yaml} (81%) rename opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/{config-otlp-pathway2-window.yaml => config-otlp-sketch-payload-flow-window.yaml} (83%) diff --git a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-raw-metrics-flow.yaml similarity index 81% rename from opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml rename to opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-raw-metrics-flow.yaml index 06d16176..aac37317 100644 --- a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/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: @@ -26,4 +26,4 @@ service: logs: level: debug resource: - service.name: countminsketchcol-pathway1-local + service.name: countminsketchcol-raw-metrics-flow-local diff --git a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-batch.yaml b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-sketch-payload-flow-batch.yaml similarity index 81% rename from opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-batch.yaml rename to opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-sketch-payload-flow-batch.yaml index a09adf31..71c059e9 100644 --- a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-batch.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/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: countminsketchcol-pathway2-batch-local + service.name: countminsketchcol-sketch-payload-flow-batch-local diff --git a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-window.yaml b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-sketch-payload-flow-window.yaml similarity index 83% rename from opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-window.yaml rename to opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-sketch-payload-flow-window.yaml index 09987c44..2dfe6be2 100644 --- a/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-window.yaml +++ b/opentelemetry-collector-contrib-patch/cmd/countminsketchcol/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: @@ -35,4 +35,4 @@ service: logs: level: debug resource: - service.name: countminsketchcol-otlp-pathway2-local + service.name: countminsketchcol-otlp-sketch-payload-flow-local