Skip to content

feat(delta): implement delta transmission over time windows and series - #65

Merged
zzylol merged 20 commits into
mainfrom
63-delta-transmission-over-time-windows
Mar 21, 2026
Merged

feat(delta): implement delta transmission over time windows and series#65
zzylol merged 20 commits into
mainfrom
63-delta-transmission-over-time-windows

Conversation

@zzylol

@zzylol zzylol commented Mar 20, 2026

Copy link
Copy Markdown
Contributor

Summary

Implements the delta transmission design from docs/delta-transmission-design.md.

sketchlib-go (PR #43, stacked on add-octosketch)

  • Phase 1: New proto messages in proto/sketchlib.proto — SketchDeltaEnvelope, CountMinDelta, CountSketchDelta, HLLDelta, DDSketchDelta and their cell/register types. Regenerated sketchlib.pb.go.
  • Phase 2: ComputeDelta/ApplyDelta in each sketch package:
    • CountMinSketch/delta.go — sparse cell diff with threshold; L1/L2 always included
    • CountSketch/delta.go — signed cell diff; TopK retransmitted in full
    • HLL/delta.go — increased registers only (max semantics, zero threshold)
    • DDSketch/delta.go — bucket count diff >= threshold; sum/min/max always transmitted

DataCollector (this PR)

  • Phase 3: Delta mode in countminsketchprocessor and countsketchprocessor
    • New config: delta_transmission: bool, delta_threshold: float64
    • Snapshot map per partition key; ComputeDelta on flush; encoding="proto_delta"/"proto_full" attribute
  • Phase 4: New receiver-side accumulation processors
    • countminsketchmergeprocessor — reconstructs CMS from proto_full/proto_delta stream
    • countsketchmergeprocessor — same for CountSketch
  • Phase 5: Delta mode in hllprocessor
    • New config: delta_transmission: bool
    • Register delta via ComputeRegisterDelta; hll.encoding attribute on data points
  • Phase 6: Delta mode in ddsketchprocessor
    • New config: delta_transmission: bool, delta_threshold: uint64
    • Bucket-level delta using sketchpb.DDSketch Store comparison
  • Phase 7: KLL guard in kllprocessor
    • delta_transmission: true → Validate() returns error (KLL is not additively mergeable)

Wire protocol

Data points carry two new attributes when delta_transmission=true:

  • encoding: "proto_delta" or "proto_full"
  • sketch_payload: bytes (proto-marshalled delta or full sketch)

First window per partition always sends proto_full. Subsequent windows send proto_delta if snapshot exists.

Test plan

  • All existing processor tests pass (verified locally)
  • CMS: ComputeDelta(snapshot, current, threshold) → ApplyDelta roundtrip produces same estimates
  • CS: same, plus TopK heap preserved after delta
  • HLL: max-semantics — applying same delta twice is idempotent
  • DDSketch: quantile estimates within relative accuracy after delta apply
  • KLL: Validate() rejects delta_transmission: true
  • Merge processors: first batch (proto_full) initializes accumulator; second (proto_delta) updates it

Related

🤖 Generated with Claude Code

@zzylol @claude
chore: update opentelemetry-collector-contrib submodule to delta-tran…
0a57a6f 
…smission commit

Points to 307870c9fb which adds delta transmission for CMS, CS, HLL,
DDSketch, and KLL processors (Phases 3-7).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol zzylol linked an issue Mar 20, 2026 that may be closed by this pull request
delta transmission over time windows #63
Closed
zzylol and others added 13 commits March 20, 2026 17:33
@zzylol
Revert "chore: update opentelemetry-collector-contrib submodule to de…
f8182cf 
…lta-transmission commit"

This reverts commit 0a57a6f.
@zzylol @claude
feat(delta): add delta transmission patches for CMS, CS, HLL, DDSketc…
d3ef3a2 
…h, KLL processors

Copies updated processor files into opentelemetry-collector-contrib-patch/
per the repo's patch-overlay workflow (backup_otel_collector_contrib_patches.sh).

Changes in patch overlay:
- processor/countminsketchprocessor: DeltaTransmission + DeltaThreshold config,
  snapshot map + ComputeDelta/ApplyDelta flush path, go.mod local sketchlib-go replace
- processor/countsketchprocessor: same delta mode + fix Insert/Estimate API for
  new sketchlib-go version
- processor/countminsketchmergeprocessor: NEW — receiver-side CMS accumulator
  (decodes proto_delta/proto_full, ApplyDelta/Merge per aggregation_key)
- processor/countsketchmergeprocessor: NEW — receiver-side CS accumulator
- processor/hllprocessor: DeltaTransmission config, register delta via
  ComputeRegisterDelta, hll.encoding attribute, API fixes for new sketchlib-go
- processor/ddsketchprocessor: DeltaTransmission + DeltaThreshold config,
  sparse bucket delta via sketchpb.DDSketch Store diffing
- processor/kllprocessor: delta_transmission guard in Validate() — KLL is not
  additively mergeable

Also adds docs/delta-transmission-design.md (the design spec for this feature).

Corresponding sketchlib-go changes: PR #43
(proto delta messages + ComputeDelta/ApplyDelta for CMS/CS/HLL/DDSketch)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
feat(delta): fix CMS/CS patches to restore all-3-modes + window_durat…
2246a76 
…ion unification

- Restore AggregateBy/LabelMatchers/LabelMatcher in CMS and CS configs
  (regressed in d3ef3a2 which used old GroupBy naming)
- Rename WindowInterval (CMS) and WindowSize (CS) to WindowDuration
  to match HLL/DDSketch/KLL naming
- Upgrade CS processor from global row/col sketch to per-partition
  windowSketch architecture (supports all 3 modes: batch, window, matrix)
- Add LabelMatchers filtering to CS ingestMetric
- CMS processor: preserve AggregateBy/seriesKey/seriesAttrs/matchesMatchers
  logic while adding delta transmission snapshot tracking
- All tests pass for both processors

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
fix(config): rename window_interval/window_size to window_duration in…
8927df6 
… cmd yamls

Update countminsketchcol and countsketchcol config.yaml files to use
unified window_duration key.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
chore: add opentelemetry-go backup/restore patch scripts
7715e48 
Submodule opentelemetry-go must stay at the upstream commit
(4575a977). All local SDK changes live in opentelemetry-go-patch/.
Add backup_opentelemetry_go_patches.sh and
restore_opentelemetry_go_patches.sh (mirroring the pattern used for
collector/contrib/proto), and wire them into the umbrella
backup_otel_patches.sh / restore_otel_patches.sh.

Workflow:
  ./restore_opentelemetry_go_patches.sh  — copy patches → submodule (enables compilation)
  ./backup_opentelemetry_go_patches.sh   — copy submodule diffs → patch folder (saves work)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
feat(sdk): add SDK-side sparse delta transmission for cumulative sket…
4bcfdf9 
…ch exports

Phase 14: SDK-to-Collector delta transmission.

Add sparse delta encoding to the cumulative() export path of the three
SDK sketch aggregators so the application sends only changed cells/registers
to the collector instead of the full sketch every export cycle.

Changes:
- metricdata/data.go: add HLLSketchEncodingDelta, CountMinSketchEncodingDelta,
  CountSketchEncodingDelta encoding constants
- aggregation.go: add DeltaTransmission bool + DeltaThreshold float64 to
  AggregationCountSketch, AggregationCountMinSketch; add DeltaTransmission
  to AggregationHLLSketch (no threshold — HLL registers are monotone)
- pipeline.go: thread new fields through to Builder method calls
- aggregate.go: update Builder.CountSketch/CountMinSketch/HLLSketch signatures;
  add Builder.Noop() for the no-transmit path
- hllsketch.go: add snapshots map + payloadFor helper; delta encoding in
  cumulative() via hll.ComputeRegisterDelta; eviction cleans snapshots
- countminsketch.go: add snapshots map + payloadFor helper; delta encoding in
  cumulative() via cms.ComputeDelta; cloneCMSketch helper
- countsketch.go: add snapshots map + payloadFor helper; delta encoding in
  cumulative() via countsketch.ComputeDelta; cloneCSSketch helper
- sdk/metric/go.mod: add replace directive to use local sketchlib-go
  (add-delta-transmission branch) with ComputeDelta/ComputeRegisterDelta

Design doc: add §14 SDK-to-Collector Delta Transmission with architecture,
config table, encoding wire values, implementation file list, snapshot
lifecycle, and interaction with collector-side delta (§1-7).

All sdk/metric tests pass.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
feat(processors): restore deleted main-branch code + add delta transm…
b0b7694 
…ission to ddsketch/hll processors

- ddsketchprocessor/config.go: restore LabelMatcher struct, AggregateBy, LabelMatchers fields from main; add sort.Strings(AggregateBy) in Validate()
- ddsketchprocessor/processor.go: restore matchesMatchers/seriesKey/seriesAttrs; add delta transmission logic in buildMergedSketchMetric; add computeDDSketchDelta/storeDelta/storeToMap helpers
- ddsketchprocessor/processor_test.go: restore TestDDAggregateByCollapsesSeries, TestDDLabelMatchersFilterGauge, TestDDAggregateByWindowModeDDSketchInput
- hllprocessor/config.go: restore LabelMatcher struct, AggregateBy, LabelMatchers fields from main; keep DeltaTransmission field added in this branch
- hllprocessor/processor.go: restore matchesMatchers/seriesKey/seriesAttrs; fix HLL API (InsertValue/EstimateCardinality); add delta transmission logic + appendHLLDeltaDataPoint/cloneHLL helpers
- hllprocessor/processor_test.go: restore TestHLLAggregateByCollapsesSeries, TestHLLLabelMatchersFilter
- kllprocessor/config.go: restore LabelMatcher struct, AggregateBy, LabelMatchers, sort.Strings; keep DeltaTransmission with validation error
- kllprocessor/processor.go: restore matchesMatchers/seriesKey/seriesAttrs
- kllprocessor/processor_test.go: restore TestAggregateByCollapsesSeries, TestLabelMatchersFilterGauge, TestAggregateByWithLabelMatchersWindowMode

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
feat(sdk): add delta transmission to DDSketch SDK aggregator
5b8044e 
Mirror the pattern already implemented for HLL, CountMin, and CountSketch:
- metricdata/data.go: add DDSketchEncodingProtoDelta constant
- aggregation.go: add DeltaTransmission bool + DeltaThreshold uint64 to AggregationDDSketch
- pipeline.go: pass new fields through to Builder.DDSketch()
- aggregate/aggregate.go: update Builder.DDSketch() signature
- aggregate/ddsketch.go: add snapshots map + snapshotsMu; implement payloadFor()
  that computes sparse delta (ddSketchDeltaPayload/ddStoreDelta/ddStoreToMap)
  in cumulative() path; exportDataPoint now accepts pre-computed payload+encoding;
  snapshot evicted alongside series to prevent memory leak
- aggregate/ddsketch_test.go: update existing test calls to new signature (deltaTransmission=false, threshold=0)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
feat: SDK→collector inbound delta reconstruction + controller sizing …
0f17dbd 
…design

## Wire format (proto + pdata)
- opentelemetry-proto: add DDSKETCH_ENCODING_PROTO_DELTA=2, HLL_SKETCH_ENCODING_DELTA=2,
  COUNT_MIN_SKETCH_ENCODING_DELTA=2, COUNT_SKETCH_ENCODING_DELTA=2 to encoding enums
- opentelemetry-collector pdata/internal: add delta enum constants to all 4 generated
  encoding enum files
- opentelemetry-collector pdata/pmetric: add DDSketchEncodingProtoDelta,
  HLLSketchEncodingDelta, CountMinSketchEncodingDelta, CountSketchEncodingDelta
  exported constants with String() support

## Collector inbound delta reconstruction
- hllprocessor: add inboundSnapshots map + inboundMergeHLL helper; applies
  hll.ApplyRegisterDelta for HLLSketchEncodingDelta payloads in processBatch
  and accumulateHLLSketchMetric; stores full snapshot on HLLSketchEncodingBinary
- ddsketchprocessor: add inboundSnapshots map; convert decodeDDSketchDataPoint to
  method with seriesKey; add applyDDSketchDelta/applyDDStore helpers; reconstructs
  full sketch from proto delta before merging; returns nil,nil when no snapshot yet
- countminsketchprocessor: add inboundSnapshots map + inboundDecodeCMS helper;
  applies cms.ApplyDelta for CountMinSketchEncodingDelta in ingestMetric

## Design doc
- §14: update scope to include DDSketch; expand §14.4 encoding table (SDK + proto);
  expand §14.5 with full implementation file table including inbound reconstruction;
  update §14.7 to describe two-layer snapshot model (inbound + outbound independent)
- §16 (new): Controller-Driven Sketch Sizing and Delta Window Configuration —
  sizing rules for CMS/CS/HLL/DDSketch, target_window_duration derivation, joint
  co-optimization loop, new AgentCollectorConfig fields, monitor feedback loop

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
feat(countsketchprocessor): add inbound delta reconstruction for pre-…
94b7e6d 
…aggregated CS

- Add inboundSnapshots map + inboundMu to countSketchProcessor struct
- inboundDecodeCS: handles CountSketchEncodingGob (full) and CountSketchEncodingDelta;
  applies countsketch.ApplyDelta to reconstruct current full sketch from sparse delta;
  returns nil,nil when delta arrives before any full snapshot
- mergeWindowCS: merges reconstructed CS into the running window sketch via Merge();
  creates window sketch (via newConfiguredCountSketch) if not yet present
- MetricTypeCountSketch case in ingestMetric: routes sketch payload through
  inboundDecodeCS + mergeWindowCS; falls back to raw sample (1.0) only when
  Sketch() is empty (backward compat with count-only data points)
- Update §14.5 design doc: countsketchprocessor now listed with full inbound
  reconstruction support alongside hll/ddsketch/cms processors

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
fix: sync docs/config with main (window_duration naming, README fields)
49bcfd7 
- countsketchprocessor README: window_size -> window_duration, add mode/
  drop_original/aggregate_by/label_matchers fields that exist on main
- countmin_bench.sh: window_interval -> window_duration in comment
- config.yaml: remove stale "# NEW:" annotation on window_duration

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
chore: replace local sketchlib-go path with GitHub pseudo-version
417308d 
Remove `replace github.com/ProjectASAP/sketchlib-go => /tmp/sketchlib-go`
from all go.mod files and update to the resolved pseudo-version
v0.0.0-20260320220729-3ba826ceb054 from the add-delta-transmission branch.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
docs+fix: sync design doc and SDK go.mod with actual implementation
0d6f653 
Code fix:
- opentelemetry-go-patch/sdk/metric/go.mod: remove local /tmp/sketchlib-go
  replace directive; update to GitHub pseudo-version v0.0.0-20260320220729-3ba826ceb054

Doc fixes (delta-transmission-design.md):
- §6 Phase 3/4: sync.RWMutex → sync.Mutex (matches actual snapshotsMu/accMu types)
- §6 Phase 3: WindowSize → WindowDuration (matches actual config field name)
- §12 table: WindowSize → WindowDuration; add AggregateBy, LabelMatchers to
  current-state column for cms/cs processors
- §13.2: window_size → window_duration in fill rate formula
- §13.3: window_size: 30s → window_duration: 30s in YAML example
- §14.3: add AggregationDDSketch DeltaTransmission/DeltaThreshold rows (DDSketch
  is implemented in the SDK aggregator)
- §14.5: sdk/metric/go.mod entry updated to reflect GitHub pseudo-version instead
  of local path replace

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol zzylol changed the title feat(delta): implement delta transmission over time windows (Phases 1-7) feat(delta): implement delta transmission over time windows Mar 21, 2026
zzylol and others added 6 commits March 20, 2026 21:22
@zzylol @claude
refactor: rename gob/binary encoding enums to proto across all sketch…
b031975 
… types

Rename HLL_SKETCH_ENCODING_BINARY→PROTO, COUNT_MIN_SKETCH_ENCODING_GOB→PROTO,
COUNT_SKETCH_ENCODING_GOB→PROTO, KLL_SKETCH_ENCODING_GOB→PROTO in metrics.proto,
pdata/internal generated enum files, and pdata/pmetric encoding files.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
refactor: use SerializeProtoBytes for full-sketch in SDK aggregators;…
e746a3b 
… bump sketchlib-go

- Replace gob encoding with SerializeProtoBytes in countminsketch aggregator
- Replace SerializeToBytes with SerializeProtoBytes in countsketch, hll, kll aggregators
- Update cloneXxx helpers to use proto round-trip
- Rename encoding constants to Proto variants in metricdata/data.go
- Bump sketchlib-go to v0.0.0-20260321021603-cddd774cb224

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
refactor: use SerializeProtoBytes/DeserializeFromProtoBytes in all pr…
ae19f01 
…ocessors; bump sketchlib-go

- countminsketchprocessor: serializeCMS/deserializeCMS/cloneCMS use proto methods
- countsketchprocessor: SerializeProtoBytes, DeserializeCountSketchFromProtoBytes
- hllprocessor: SerializeProtoBytes, DeserializeHyperLogLogFromProtoBytes
- kllprocessor: SerializeProtoBytes, DeserializeKLLSketchFromProtoBytes
- countminsketchmergeprocessor: DeserializeCountMinSketchFromProtoBytes
- countsketchmergeprocessor: DeserializeCountSketchFromProtoBytes
- Bump all processor go.mod to sketchlib-go v0.0.0-20260321021603-cddd774cb224

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol
fix: update stale HLLSketchEncodingBinary comment to HLLSketchEncodin…
a437307 
…gProto
@zzylol @claude
refactor: use split delta API (algo+codec) in all processors and SDK …
e7f448d 
…aggregators

sketchlib-go ecebc36: delta.go = pure algorithm on sketch structs (no proto);
delta_codec.go = bytes <-> proto <-> sketch struct in one pass.

Callers now explicitly:
  ComputeDelta(snap, current) -> *Delta -> SerializeDelta -> []byte  (send)
  DeserializeDelta([]byte) -> *Delta -> ApplyDelta(target)           (recv)

Bump sketchlib-go to v0.0.0-20260321023259-ecebc36fb5aa across all go.mod files.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol @claude
test: add delta transmission tests for CMS, CS, and HLL processors; b…
71505bf 
…ump sketchlib-go

Adds delta_transmission_test.go to countminsketchprocessor (5 tests),
countsketchprocessor (6 tests), and hllprocessor (5 tests). All 16 pass.

Coverage per processor:
- FirstWindowSendsFullSketch: no prior snapshot → proto_full encoding
- SubsequentWindowsSendDelta: snapshot exists → proto_delta encoding
- RoundTrip: ApplyDelta(clone(snap), delta) == independent reference sketch
- MultipleWindowsConvergence: 5 windows, receiver reconstructs correct state each time
- DisabledAlwaysSendsFullSketch: DeltaTransmission=false → always proto_full
- PartitionKeyIsolation (CS only): two services independently transition full→delta
- MaxSemanticsIdempotent (HLL only): applying delta twice == applying once

Also bumps sketchlib-go to v0.0.0-20260321024028-d20a9f9151b5 (threshold
bug fixes + delta tests) in all go.mod files.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@zzylol
zzylol merged commit f6735ca into main Mar 21, 2026
@zzylol
zzylol deleted the 63-delta-transmission-over-time-windows branch March 21, 2026 03:00
@zzylol zzylol changed the title feat(delta): implement delta transmission over time windows feat(delta): implement delta transmission over time windows and series Mar 21, 2026
SieDeta pushed a commit that referenced this pull request Apr 17, 2026
@zzylol
Merge pull request #65 from ProjectASAP/63-delta-transmission-over-ti…
6efccb0 
…me-windows

feat(delta): implement delta transmission over time windows
@zzylol zzylol mentioned this pull request May 27, 2026
Merged
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

delta transmission over time windows

1 participant

@zzylol