CountMinSketch: OTLP export and sketch serialization

[GnaneshGnani]

Changes

1. Code changes

   • Added OTLP sketch payload export behavior for CountMinSketch output flow.
   • Switched CountMinSketch sketch payload emission to a portable protobuf envelope in the CountMinSketch processor.
   • Updated processor:
     • processor/countminsketchprocessor/processor.go

2. Implementation pattern used in the processor

   • Build sketch state using sketchlib.
   • Convert sketch to a portable envelope using SerializePortable().
   • Encode the envelope with proto.Marshal(env).
   • Attach encoded bytes into OTLP metric attributes for Sketch Payload Flow.
   • Keep Raw Metrics Flow behavior (raw metric forwarding) unchanged unless the selected mode intentionally emits sketch-only output.

3. Go module file changes

   • Updated CountMinSketch processor-local module file:
     • processor/countminsketchprocessor/go.mod

4. OTLP exporter compatibility settings

   • encoding: proto
   • compression: gzip or none

Compression notes

• gzip is supported in both flows and is currently configured in CountMinSketch test configs for both Raw Metrics Flow and Sketch Payload Flow.
• gzip is recommended for sketch-heavy Sketch Payload Flow traffic.
• none is supported and useful for debugging payload size behavior, but it can hit request size limits more easily.

Known issue

1. Reproducible issue
   • CountMinSketch Sketch Payload Flow Batch mode with compression none can return repeated HTTP 413 responses.
   • In Sketch Payload Flow Batch mode, multiple serialized sketch payloads are sent in OTLP HTTP export batches.
   • With compression disabled, request bodies are larger and can exceed upstream body size limits (receiver, proxy, or ingress limits), triggering HTTP 413 (Payload Too Large).
   • With gzip enabled, payload size is reduced enough to stay under the effective limit in validated runs.
   • Preferred: enable compression: gzip for Sketch Payload Flow Batch.
   • Reduce per-request payload size by lowering batch send size or timeout so each request carries fewer sketch payloads.
   • Reduce sketch payload growth (for example, tune sketch parameters that influence serialized size).
   • Increase allowed HTTP body size on the receiving path (ASAPQuery receiver and any reverse proxy or ingress in front of it).
   • Optionally add exporter-side safeguards to split large batches before send.

OTLP message shape

Common wire format (both flows):

• OTLP HTTP endpoint: POST /v1/metrics
• OTLP gRPC service/method: opentelemetry.proto.collector.metrics.v1.MetricsService/Export
• Body: Protobuf ExportMetricsServiceRequest
• Optional HTTP compression: gzip (or none)

gRPC transport notes

• gRPC uses the same ExportMetricsServiceRequest message as HTTP.
• There is no HTTP path like /v1/metrics in gRPC; the RPC method is MetricsService.Export.
• In local runs in this integration, gRPC receiver is typically on :4317 and HTTP receiver on :4318.
• Message shape inside the payload is identical across HTTP and gRPC (same resource_metrics -> scope_metrics -> metrics -> data_points structure).

Shared top-level structure:

ExportMetricsServiceRequest {
    resource_metrics: [
        {
            resource: { attributes: [...] },
            scope_metrics: [
                {
                    scope: { name, version, attributes[...] },
                    metrics: [ ... ]
                }
            ]
        }
    ]
}

Raw Metrics Flow message shape

In this flow, data points carry numeric values directly (gauge or sum).

How ASAPQuery treats it:

• No sketch payload attribute is present.
• Points are parsed as regular metrics and routed as Raw Metrics Flow.

Sketch Payload Flow message shape (CountMinSketch)

In this flow, processor attaches serialized sketch bytes into OTLP attributes.
Sketch bytes are produced as:

1. SerializePortable()
2. proto.Marshal(...)
3. store resulting bytes in a sketch payload attribute

Expected sketch payload attribute key in this PR:

• cms.sketch_payload

How ASAPQuery treats it:

• If cms.sketch_payload is present, the point is classified as Sketch Payload Flow.
• Payload bytes are logged and processed as sketch data, not as a regular raw numeric point.

How to test these changes

1) Start ASAPQuery OTLP receiver

From ASAPQuery/:

cd ASAPQuery
RUST_LOG=query_engine_rust=debug ./target/release/query_engine_rust \
    --enable-otel-ingest \
    --otel-grpc-port 4317 \
    --otel-http-port 4318 \
    --kafka-topic dummy \
    --input-format json \
    --config asap-query-engine/examples/promql/inference_config.yaml \
    --streaming-config asap-query-engine/examples/promql/streaming_config.yaml \
    --streaming-engine arroyo \
    --prometheus-scrape-interval 15 \
    --output-dir /tmp/asapquery \
    --query-language promql \
    --lock-strategy global

Expected startup logs include OTLP listeners on :4317 (gRPC) and :4318 (HTTP).

2) Run CountMinSketch processor config

From DataCollector/opentelemetry-collector-contrib-patch/:

Raw Metrics Flow:

./cmd/countminsketchcol/dist/countminsketchcol --config cmd/countminsketchcol/asap_query_config/config-otlp-pathway1.yaml

Sketch Payload Flow (batch):

./cmd/countminsketchcol/dist/countminsketchcol --config cmd/countminsketchcol/asap_query_config/config-otlp-pathway2-batch.yaml

3) Run load generator

From DataCollector/otel_collector_benchmark/:

cd ProjectASAP/DataCollector/otel_collector_benchmark
go run main.go --endpoint localhost:53217 --workers 2 --hosts 5 --metrics 10 --duration 15s

Notes:

• Use localhost:53217 when the processor otlp receiver gRPC endpoint is 0.0.0.0:53217.
• Prefer gzip for Sketch Payload Flow batch mode.