fix(processors): forward inputs through windowed sketch processors - #211
Merged
Merged
Conversation
So a single pipeline like `[ddsketch, hll, batch]` actually works.
## Bug
ddsketch / kll / hll processors in `mode: window` returned `nil`
from `ConsumeMetrics` — they accumulated inputs into their
windowed state and dropped the input batch on the floor without
forwarding it to the next consumer:
```go
case ModeWindow:
p.accumulateIntoWindow(md)
return nil // <-- inputs go nowhere
```
In a chained pipeline `[ddsketch, hll, batch]`, that means:
1. otlp delivers raw `http_requests_total` → ddsketch.ConsumeMetrics.
ddsketch accumulates, returns nil. **HLL never sees the raw input.**
2. ddsketch's window timer fires → it emits typed-DDSketch metrics
to its nextConsumer (HLL.ConsumeMetrics).
3. HLL.accumulateIntoWindow has only `case MetricTypeGauge` and
`case MetricTypeHLLSketch` in its switch — typed DDSketch
metrics fall through and HLL **silently drops them**.
4. batch sees nothing.
Net: with multiple windowed sketch processors in one pipeline,
nothing reaches the exporter.
countminsketchprocessor / countsketchprocessor already do the
right thing — their ModeWindow paths return `md` (forward
inputs) by default, with an opt-in `drop_original` flag. The
other three were the outliers.
## Fix
Match the CMS/CountSketch pattern: forward `md` unchanged after
accumulating. Same one-liner in all three processors:
```go
case ModeWindow:
p.accumulateIntoWindow(md)
return p.nextConsumer.ConsumeMetrics(ctx, md)
```
Now the chain works as expected:
- `ddsketch.ConsumeMetrics(raw)`: accumulate, forward raw to HLL.
- `HLL.ConsumeMetrics(raw)`: accumulate, forward raw to batch.
- `batch.ConsumeMetrics(raw)`: batch & ship.
- ddsketch tick fires → typed-DDSketch → HLL (no match, forward) → batch.
- HLL tick fires → typed-HLL → batch.
So both DDSketch and HLL sketches reach the exporter in addition
to the original raw inputs. Users who don't want the raw inputs
shipped can add a `filter` processor at the end of the pipeline.
## Verification
End-to-end with `b3-delta` agent
(`processors: [ddsketch, HLL, batch]`, `delta_transmission: true`,
`window_duration: 60s`):
```
W1 (16:21:09):
Ddsketch received: dps=1000
Hllsketch received: dps=1000 ← HLL now reaches backend
ingest: 1000 routed, 0 decode-failed, 1000 unconfigured
W2 (16:22:09):
Ddsketch received: dps=1000 (delta)
ingest: 1000 routed, 0 decode-failed, 1000 unconfigured
```
Pre-fix: HLL never reached backend; only ddsketch's outputs were
emitted (and only after #210's DDSketch delta wire fix).
## Supersedes #209
PR #209 worked around this bug at the config level by splitting
`[ddsketch, HLL, batch]` into two parallel pipelines
(`metrics/ddsketch` and `metrics/hll`). That goes against the
goal of one pipeline that supports any controller-chosen
combination of sketches. With this fix, the original chained
config works directly. #209 can be closed.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
4 tasks
zzylol
added a commit
that referenced
this pull request
May 1, 2026
…able e2e (#212) Capture this session's merged work at the top, retire stale follow-ups, add new ones discovered along the way. New top section: "Single-pipeline multi-sketch + delta + queryable warm tier (2026-05-01)" — itemises #210 (DDSketch delta wire fixes — three connected bugs), #211 (windowed processors pass-through so chained `[ddsketch, HLL, batch]` works), and the companion backend PRs (#70 OTLP gRPC max msg size, #71 store overlap filter + closest-pane + window annotation). Live e2e capture pasted in as evidence. Retired follow-up #1 ("Warm-tier sketch ingest is dropped at the gateway") — superseded by the new top section. Replaced the lingering "agent-side ddsketch→batch wiring quirk" note that was the chain bug now fixed by #211. Replaced the abbreviated follow-up list (#2 cold reader, #3 reducer self-check) with the actual residual after this session, in priority order: 1. Run the P7 sweep + accuracy reducer over all five sketches. 2. Backend in-memory sketch_snapshots cache loss on restart. 3. Inference config breadth (only [1m] patterns today). 4. Cold reader torn-line tolerance (carried over). 5. Reducer self-check vs engine (carried over). 6. Inline GOPRIVATE / GOTOOLCHAIN env vars in build_sketchcollector.sh. 7. Decide on OTel submodule bumps still dirty in working tree. Updated `_Last updated_` to 2026-05-01. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This was referenced May 1, 2026
zzylol
added a commit
that referenced
this pull request
May 2, 2026
…post-#211 sink assertions (#220) Why: - DataDog/sketches-go is no longer a runtime dep (production switched to sketchlib-go in #210). Tests that still imported DataDog/sketches-go to build sketch fixtures were leftover. - After #211, window-mode ConsumeMetrics forwards raw inputs through nextConsumer so chained windowed sketch processors can see the original payload. Tests written before #211 expected sink.AllMetrics() to be empty after ConsumeMetrics; they now observe both the input pass-through AND the synthesized output emitted by flushWindow. Update assertions to count both and scan-by-name where ordering is non-deterministic. - KLL/CountSketch/CountMinSketch processors imported go.opentelemetry.io/collector/processor/selfmonitor (the local fork's selfmonitor package) but the corresponding go.mod was missing the `replace` directive that points go.opentelemetry.io/collector/processor at the local fork. Add the replace; drop the now-unused gob deserializer in two test sites where the test was reading proto-encoded emit bytes. Per-processor: - ddsketchprocessor: fix 5 window-mode tests + concurrent test; treat sink as [forwarded inputs..., synthesized flush output]. - kllprocessor: add `replace go.opentelemetry.io/collector/processor` to go.mod, fix 5 window-mode test assertions, swap DeserializeKLLSketchFromBytes (gob) → DeserializeKLLSketchFromProtoBytes (proto) in TestBatchModeTransmitSketch to match emit format. - hllprocessor: fix 2 window-mode test assertions (selfmonitor was already resolvable). - countsketchprocessor: add the `replace` directive only — tests already accounted for pass-through. - countminsketchprocessor: add the `replace` directive, swap DeserializeCountMinSketchFromBytes (gob) → DeserializeCountMinSketchFromProtoBytes (proto) to match SerializeProtoBytesFO emit path. All 5 packages pass `go test ./... -count=1` and `go build ./...`. Full sketchcollector build via build_sketchcollector.sh succeeds. gofmt is clean for all files modified by this change; pre-existing gofmt debt in unrelated files left untouched. Submodule pointer drift on opentelemetry-collector and opentelemetry-go was left unstaged (pre-existing noise). Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Stacked on the merged #210. Makes a single pipeline like `[ddsketch, HLL, batch]` actually work — without this, chained windowed sketch processors silently drop everything.
Bug
`ddsketch / kll / hll` processors in `mode: window` returned `nil` from `ConsumeMetrics` — they accumulated inputs into their windowed state and dropped the input batch on the floor without forwarding it to the next consumer:
```go
case ModeWindow:
p.accumulateIntoWindow(md)
return nil // <-- inputs go nowhere
```
In a chained pipeline:
`countminsketchprocessor` / `countsketchprocessor` already do the right thing — their ModeWindow paths return `md` (forward inputs) by default, with an opt-in `drop_original` flag. The other three were the outliers.
Fix
Match the CMS/CountSketch pattern in all three processors:
```go
case ModeWindow:
p.accumulateIntoWindow(md)
return p.nextConsumer.ConsumeMetrics(ctx, md)
```
Now the chain works:
Both sketches reach the exporter alongside raw inputs. Users who don't want raw inputs shipped can add a `filter` processor at the end (or use `drop_original` on CMS/CountSketch).
Verification
End-to-end with the original chained `b3-delta` config (`processors: [ddsketch, HLL, batch]`, `delta_transmission: true`, `window_duration: 60s`):
```
W1 (16:21:09):
Ddsketch received: dps=1000
Hllsketch received: dps=1000 ← HLL now reaches backend
ingest: 1000 routed, 0 decode-failed, 1000 unconfigured
W2 (16:22:09):
Ddsketch received: dps=1000 (delta — applied via cached snapshot)
ingest: 1000 routed, 0 decode-failed, 1000 unconfigured
```
Supersedes #209
PR #209 worked around this bug at the config level by splitting `[ddsketch, HLL, batch]` into two parallel pipelines. That goes against the user's stated goal of one pipeline that supports any controller-chosen combination of sketches. With this fix, the original chained config works directly. #209 should be closed.
Test plan
🤖 Generated with Claude Code