You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/08/10 03:05:34 UTC
(skywalking-banyandb) branch main updated: Optimize trace merge decision path (#1272)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 7f685a1d8 Optimize trace merge decision path (#1272)
7f685a1d8 is described below
commit 7f685a1d89de5f1f7f94a3e616a29a8724ceac0a
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Aug 10 11:05:29 2026 +0800
Optimize trace merge decision path (#1272)
---
banyand/trace/fragment_guard.go | 6 +-
banyand/trace/fragment_guard_contract_test.go | 39 ++++++
banyand/trace/merge_benchmark_observer_test.go | 1 +
banyand/trace/merger.go | 14 +-
banyand/trace/pipeline_chain.go | 154 +++++++++++++++------
banyand/trace/pipeline_chain_test.go | 66 +++++++++
.../trace-pipeline-merge-optimization-plan.md | 30 +++-
pkg/pipeline/sdk/chain.go | 31 ++++-
pkg/pipeline/sdk/chain_test.go | 33 +++++
9 files changed, 320 insertions(+), 54 deletions(-)
diff --git a/banyand/trace/fragment_guard.go b/banyand/trace/fragment_guard.go
index ec8e0ab04..a65dc2c4e 100644
--- a/banyand/trace/fragment_guard.go
+++ b/banyand/trace/fragment_guard.go
@@ -195,9 +195,13 @@ func newTraceFragmentGuard(config traceFragmentGuardConfig, catalog traceFragmen
}
func assembleTraceFragmentGuardTrace(traceID string, staged []stagedTrace) traceFragmentGuardTrace {
+ return assembleTraceFragmentGuardTraceInto(traceID, staged, make([]traceFragmentGuardBlock, 0, len(staged)))
+}
+
+func assembleTraceFragmentGuardTraceInto(traceID string, staged []stagedTrace, blocks []traceFragmentGuardBlock) traceFragmentGuardTrace {
assembled := traceFragmentGuardTrace{
TraceID: traceID,
- Blocks: make([]traceFragmentGuardBlock, 0, len(staged)),
+ Blocks: blocks[:0],
}
matched := false
complete := traceID != ""
diff --git a/banyand/trace/fragment_guard_contract_test.go b/banyand/trace/fragment_guard_contract_test.go
index ed3ad6cac..c4cba1b2f 100644
--- a/banyand/trace/fragment_guard_contract_test.go
+++ b/banyand/trace/fragment_guard_contract_test.go
@@ -1065,6 +1065,45 @@ func TestAssembleTraceFragmentGuardTraceIncludesEveryStagedBlock(t *testing.T) {
}, got)
}
+func TestAssembleTraceFragmentGuardTraceReusesCallerStorage(t *testing.T) {
+ staged := []stagedTrace{{
+ traceID: "trace-a",
+ slowBlock: &blockPointer{bm: blockMetadata{
+ traceID: "trace-a", timestamps: timestampsMetadata{min: 100, max: 110, known: true},
+ }},
+ }}
+ storage := make([]traceFragmentGuardBlock, 0, len(staged))
+
+ got := assembleTraceFragmentGuardTraceInto("trace-a", staged, storage)
+
+ require.Len(t, got.Blocks, 1)
+ require.Equal(t, &storage[:cap(storage)][0], &got.Blocks[0])
+ allocations := testing.AllocsPerRun(100, func() {
+ got = assembleTraceFragmentGuardTraceInto("trace-a", staged, got.Blocks[:0])
+ })
+ require.Zero(t, allocations)
+}
+
+func TestAssembleTraceFragmentGuardTracePreallocatesStandaloneStorage(t *testing.T) {
+ const blockCount = 8
+ staged := make([]stagedTrace, blockCount)
+ for stagedIdx := range staged {
+ staged[stagedIdx] = stagedTrace{
+ traceID: "trace-a",
+ slowBlock: &blockPointer{bm: blockMetadata{
+ traceID: "trace-a", timestamps: timestampsMetadata{min: 100, max: 110, known: true},
+ }},
+ }
+ }
+
+ var got traceFragmentGuardTrace
+ allocations := testing.AllocsPerRun(100, func() {
+ got = assembleTraceFragmentGuardTrace("trace-a", staged)
+ })
+ require.Len(t, got.Blocks, blockCount)
+ require.LessOrEqual(t, allocations, 1.0, "the standalone helper should allocate its block vector once")
+}
+
func TestMergeTwoBlocksPreservesTraceTimestampBoundsForGuard(t *testing.T) {
left := &blockPointer{
block: block{
diff --git a/banyand/trace/merge_benchmark_observer_test.go b/banyand/trace/merge_benchmark_observer_test.go
index 8b72d50b6..633093e48 100644
--- a/banyand/trace/merge_benchmark_observer_test.go
+++ b/banyand/trace/merge_benchmark_observer_test.go
@@ -39,6 +39,7 @@ func TestMergeBenchmarkCountsActualSamplerInvocations(t *testing.T) {
first := &durationEnvelopeSampler{}
second := &durationEnvelopeSampler{}
chain := newMergeChain("group", "schema", []sdk.Sampler{first, second}, 1)
+ t.Cleanup(chain.close)
batch := &sdk.TraceBatch{Traces: []sdk.TraceBlock{{TraceID: "a"}, {TraceID: "b"}}}
observation := &mergeEvaluationObservation{}
_, executeErr := chain.executeObserved(batch, time.Second, observation)
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 19b55bec8..80729280c 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -704,6 +704,9 @@ func (tst *tsTable) mergePartsThenSendIntroductionObserved(creator snapshotCreat
filter, initialReason = tst.buildHotMergeFilterDecisionAt(parts, logicalNow)
}
operation.setInitialReason(initialReason)
+ if filter != nil && filter.chain != nil {
+ defer filter.chain.close()
+ }
if filter != nil && operation != nil {
filter.observation = operation.evaluation
operation.event.EstimatedStagingBytes = filter.estimatedStagingBytes
@@ -1287,6 +1290,7 @@ type stagedEvaluationVectors struct {
traceBlocks []sdk.TraceBlock
traceIDs []string
guardRanges []stagedTraceRange
+ guardBlocks []traceFragmentGuardBlock
decisionMask []bool
}
@@ -1296,10 +1300,12 @@ func (vectors *stagedEvaluationVectors) reset() {
}
clear(vectors.traceIDs)
clear(vectors.guardRanges)
+ clear(vectors.guardBlocks)
clear(vectors.decisionMask)
vectors.traceBlocks = vectors.traceBlocks[:0]
vectors.traceIDs = vectors.traceIDs[:0]
vectors.guardRanges = vectors.guardRanges[:0]
+ vectors.guardBlocks = vectors.guardBlocks[:0]
vectors.decisionMask = vectors.decisionMask[:0]
}
@@ -1417,7 +1423,8 @@ func releaseStagedEvaluationVectors(vectors *stagedEvaluationVectors, reusable b
}
vectors.reset()
if cap(vectors.traceBlocks) > maxPooledEvaluationTraces || cap(vectors.traceIDs) > maxPooledEvaluationTraces ||
- cap(vectors.guardRanges) > maxPooledEvaluationTraces || cap(vectors.decisionMask) > maxPooledEvaluationTraces {
+ cap(vectors.guardRanges) > maxPooledEvaluationTraces || cap(vectors.guardBlocks) > maxPooledStagedBlocks ||
+ cap(vectors.decisionMask) > maxPooledEvaluationTraces {
stagedEvaluationPool.Discard(vectors)
return false
}
@@ -1844,7 +1851,10 @@ func resolveStagedDrops(filter *mergeFilter, staged []stagedTrace, assembledBatc
continue
}
guardRange := assembledBatch.vectors.guardRanges[traceIdx]
- guardTrace := assembleTraceFragmentGuardTrace(traceID, staged[guardRange.start:guardRange.end])
+ guardTrace := assembleTraceFragmentGuardTraceInto(
+ traceID, staged[guardRange.start:guardRange.end], assembledBatch.vectors.guardBlocks,
+ )
+ assembledBatch.vectors.guardBlocks = guardTrace.Blocks[:0]
decision := filter.guard.guard.Resolve(filter.guardContext(), guardTrace, traceFragmentSamplerActionDrop)
if filter.owner != nil {
filter.owner.incPipelineGuardBloomProbes(decision.BloomProbes)
diff --git a/banyand/trace/pipeline_chain.go b/banyand/trace/pipeline_chain.go
index e5ceef47e..90d02d7a4 100644
--- a/banyand/trace/pipeline_chain.go
+++ b/banyand/trace/pipeline_chain.go
@@ -35,6 +35,8 @@ var chainLog = logger.GetLogger("trace").Named("pipeline-chain")
// consecutive-timeout circuit breaker disables the chain after circuitBreakN
// timeouts.
type mergeChain struct {
+ worker *mergeDecisionWorker
+ timer *time.Timer
samplers []sdk.Sampler
group string
schema string
@@ -42,9 +44,23 @@ type mergeChain struct {
circuitOpen bool
circuitBreakN int
consecutiveTOs int
+ executionMu sync.Mutex
mu sync.Mutex
}
+type mergeDecisionRequest struct {
+ batch *sdk.TraceBatch
+ observation *mergeEvaluationObservation
+ decisionMask []bool
+}
+
+type mergeDecisionWorker struct {
+ requests chan mergeDecisionRequest
+ results chan sdk.Verdict
+ stopCh chan struct{}
+ stopOnce sync.Once
+}
+
// newMergeChain builds a chain from the ordered samplers and computes the union
// projection: Tags is the union of all plugins' projected tag names; SpanIDs and
// Spans are true if any plugin requests them.
@@ -53,10 +69,12 @@ type mergeChain struct {
func newMergeChain(group, schema string, samplers []sdk.Sampler, circuitBreakN int) *mergeChain {
var union sdk.Projection
seen := make(map[string]struct{})
+ activeSamplers := make([]sdk.Sampler, 0, len(samplers))
for _, sampler := range samplers {
if sampler == nil {
continue
}
+ activeSamplers = append(activeSamplers, sampler)
proj := sampler.Project()
for _, name := range proj.Tags {
if _, ok := seen[name]; ok {
@@ -74,7 +92,7 @@ func newMergeChain(group, schema string, samplers []sdk.Sampler, circuitBreakN i
}
return &mergeChain{
projection: union,
- samplers: samplers,
+ samplers: activeSamplers,
group: group,
schema: schema,
circuitBreakN: circuitBreakN,
@@ -99,33 +117,29 @@ func (mc *mergeChain) executeObserved(batch *sdk.TraceBatch, timeout time.Durati
func (mc *mergeChain) executeObservedInto(batch *sdk.TraceBatch, timeout time.Duration,
observation *mergeEvaluationObservation, decisionMask []bool,
) (sdk.Verdict, bool, error) {
- retainAll := func() sdk.Verdict {
- keep := make([]bool, len(batch.Traces))
- for i := range keep {
- keep[i] = true
- }
- return sdk.Verdict{Keep: keep}
- }
+ mc.executionMu.Lock()
+ defer mc.executionMu.Unlock()
mc.mu.Lock()
if mc.circuitOpen {
mc.mu.Unlock()
- return retainAll(), true, nil
+ return retainAllVerdict(len(batch.Traces), decisionMask), true, nil
}
mc.mu.Unlock()
- resultCh := make(chan sdk.Verdict, 1)
- go func() {
- resultCh <- mc.runChain(batch, observation, decisionMask)
- }()
+ worker := mc.acquireWorker()
+ worker.requests <- mergeDecisionRequest{batch: batch, observation: observation, decisionMask: decisionMask}
+ timeoutCh := mc.resetTimer(timeout)
select {
- case verdict := <-resultCh:
+ case verdict := <-worker.results:
+ mc.stopTimer()
mc.mu.Lock()
mc.consecutiveTOs = 0
mc.mu.Unlock()
return verdict, true, nil
- case <-time.After(timeout):
+ case <-timeoutCh:
+ mc.abandonWorker(worker)
mc.mu.Lock()
mc.consecutiveTOs++
opened := false
@@ -135,9 +149,87 @@ func (mc *mergeChain) executeObservedInto(batch *sdk.TraceBatch, timeout time.Du
}
mc.mu.Unlock()
if opened {
- return retainAll(), false, fmt.Errorf("circuit_open")
+ return retainAllVerdict(len(batch.Traces), nil), false, fmt.Errorf("circuit_open")
+ }
+ return retainAllVerdict(len(batch.Traces), nil), false, fmt.Errorf("timeout")
+ }
+}
+
+func retainAllVerdict(traceCount int, mask []bool) sdk.Verdict {
+ if cap(mask) < traceCount {
+ mask = make([]bool, traceCount)
+ } else {
+ mask = mask[:traceCount]
+ }
+ for traceIdx := range mask {
+ mask[traceIdx] = true
+ }
+ return sdk.Verdict{Keep: mask}
+}
+
+func (mc *mergeChain) acquireWorker() *mergeDecisionWorker {
+ if mc.worker != nil {
+ return mc.worker
+ }
+ worker := &mergeDecisionWorker{
+ requests: make(chan mergeDecisionRequest),
+ results: make(chan sdk.Verdict),
+ stopCh: make(chan struct{}),
+ }
+ mc.worker = worker
+ go worker.run(mc)
+ return worker
+}
+
+func (mdw *mergeDecisionWorker) run(chain *mergeChain) {
+ for {
+ select {
+ case request := <-mdw.requests:
+ verdict := chain.runChain(request.batch, request.observation, request.decisionMask)
+ select {
+ case mdw.results <- verdict:
+ case <-mdw.stopCh:
+ return
+ }
+ case <-mdw.stopCh:
+ return
}
- return retainAll(), false, fmt.Errorf("timeout")
+ }
+}
+
+func (mc *mergeChain) abandonWorker(worker *mergeDecisionWorker) {
+ worker.stopOnce.Do(func() { close(worker.stopCh) })
+ if mc.worker == worker {
+ mc.worker = nil
+ }
+}
+
+func (mc *mergeChain) resetTimer(timeout time.Duration) <-chan time.Time {
+ if mc.timer == nil {
+ mc.timer = time.NewTimer(timeout)
+ return mc.timer.C
+ }
+ mc.stopTimer()
+ mc.timer.Reset(timeout)
+ return mc.timer.C
+}
+
+func (mc *mergeChain) stopTimer() {
+ if mc.timer == nil || mc.timer.Stop() {
+ return
+ }
+ select {
+ case <-mc.timer.C:
+ default:
+ }
+}
+
+func (mc *mergeChain) close() {
+ mc.executionMu.Lock()
+ defer mc.executionMu.Unlock()
+ mc.stopTimer()
+ if mc.worker != nil {
+ mc.abandonWorker(mc.worker)
}
}
@@ -155,31 +247,13 @@ func (mc *mergeChain) runChain(batch *sdk.TraceBatch, observation *mergeEvaluati
}
chainLog.Warn().Err(info.Err).Str("group", mc.group).Str("schema", mc.schema).Msg("sampler link failed; bypassing (retain)")
}
- if observation == nil {
- return sdk.EvaluateChainInto(mc.samplers, batch, decisionMask, onBypass)
- }
- observedSamplers := make([]sdk.Sampler, len(mc.samplers))
- evaluatedOnce := &sync.Once{}
- for samplerIdx, sampler := range mc.samplers {
- if sampler != nil {
- observedSamplers[samplerIdx] = observedMergeSampler{Sampler: sampler, observation: observation, evaluatedOnce: evaluatedOnce}
+ if observation != nil {
+ observation.pluginCalls.Add(uint64(len(mc.samplers)))
+ if len(mc.samplers) > 0 {
+ observation.evaluated.Add(uint64(len(batch.Traces)))
}
}
- return sdk.EvaluateChainInto(observedSamplers, batch, decisionMask, onBypass)
-}
-
-type observedMergeSampler struct {
- sdk.Sampler
- observation *mergeEvaluationObservation
- evaluatedOnce *sync.Once
-}
-
-func (oms observedMergeSampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) {
- oms.observation.pluginCalls.Add(1)
- oms.evaluatedOnce.Do(func() {
- oms.observation.evaluated.Add(uint64(len(batch.Traces)))
- })
- return oms.Sampler.Decide(batch)
+ return sdk.EvaluateChainInto(mc.samplers, batch, decisionMask, onBypass)
}
// assembleTraceBlock builds a COPY-backed sdk.TraceBlock from a loaded merge
diff --git a/banyand/trace/pipeline_chain_test.go b/banyand/trace/pipeline_chain_test.go
index dd3783e1c..bc8d6d5e3 100644
--- a/banyand/trace/pipeline_chain_test.go
+++ b/banyand/trace/pipeline_chain_test.go
@@ -137,6 +137,9 @@ func newTestChain(dropIDs map[string]struct{}) *mergeChain {
// block trace ids in order plus the dropped set.
func mergeWithFilter(t *testing.T, parts []*traces, filter *mergeFilter) ([]string, map[string]struct{}) {
t.Helper()
+ if filter != nil && filter.chain != nil {
+ defer filter.chain.close()
+ }
tmpPath, defFn := test.Space(require.New(t))
defer defFn()
fileSystem := fs.NewLocalFileSystem()
@@ -731,6 +734,69 @@ type sleepSampler struct {
d time.Duration
}
+type reusableVerdictSampler struct {
+ keep []bool
+}
+
+func (rvs *reusableVerdictSampler) Kind() sdk.Kind { return sdk.KindSampler }
+func (rvs *reusableVerdictSampler) Project() sdk.Projection { return sdk.Projection{} }
+func (rvs *reusableVerdictSampler) Close() error { return nil }
+func (rvs *reusableVerdictSampler) Decide(*sdk.TraceBatch) (sdk.Verdict, error) {
+ return sdk.Verdict{Keep: rvs.keep}, nil
+}
+
+func BenchmarkMergeChainRunObserved(b *testing.B) {
+ const traceCount = 512
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, traceCount)}
+ chain := newMergeChain("g", "s", []sdk.Sampler{&reusableVerdictSampler{keep: make([]bool, traceCount)}}, 0)
+ observation := &mergeEvaluationObservation{}
+ decisionMask := make([]bool, traceCount)
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
+ chain.runChain(batch, observation, decisionMask)
+ }
+}
+
+func BenchmarkMergeChainExecuteObserved(b *testing.B) {
+ const traceCount = 512
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, traceCount)}
+ chain := newMergeChain("g", "s", []sdk.Sampler{&reusableVerdictSampler{keep: make([]bool, traceCount)}}, 0)
+ b.Cleanup(chain.close)
+ observation := &mergeEvaluationObservation{}
+ decisionMask := make([]bool, traceCount)
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
+ _, reusable, executeErr := chain.executeObservedInto(batch, time.Minute, observation, decisionMask)
+ if executeErr != nil || !reusable {
+ b.Fatalf("unexpected decision result: reusable=%t err=%v", reusable, executeErr)
+ }
+ }
+}
+
+func TestMergeChainObservedDecisionPathReusesExecutionStorage(t *testing.T) {
+ const traceCount = 16
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, traceCount)}
+ chain := newMergeChain("g", "s", []sdk.Sampler{&reusableVerdictSampler{keep: make([]bool, traceCount)}}, 0)
+ t.Cleanup(chain.close)
+ observation := &mergeEvaluationObservation{}
+ decisionMask := make([]bool, traceCount)
+ _, reusable, executeErr := chain.executeObservedInto(batch, time.Minute, observation, decisionMask)
+ require.NoError(t, executeErr)
+ require.True(t, reusable)
+
+ var measuredErr error
+ allocations := testing.AllocsPerRun(100, func() {
+ _, reusable, measuredErr = chain.executeObservedInto(batch, time.Minute, observation, decisionMask)
+ if !reusable {
+ measuredErr = fmt.Errorf("decision storage was not reusable")
+ }
+ })
+ require.NoError(t, measuredErr)
+ require.LessOrEqual(t, allocations, 1.0, "healthy decisions should reuse the worker, timer, mask, and observation path")
+}
+
func (s *sleepSampler) Kind() sdk.Kind { return sdk.KindSampler }
func (s *sleepSampler) Project() sdk.Projection { return sdk.Projection{} }
func (s *sleepSampler) Close() error { return nil }
diff --git a/docs/design/trace-pipeline-merge-optimization-plan.md b/docs/design/trace-pipeline-merge-optimization-plan.md
index f7009680e..ffe1e9e51 100644
--- a/docs/design/trace-pipeline-merge-optimization-plan.md
+++ b/docs/design/trace-pipeline-merge-optimization-plan.md
@@ -256,20 +256,38 @@ must not weaken publication-time revalidation.
**Exit gate.**
-- [ ] Sampler-wrapper slice built once per batch (or per worker) rather than per batch.
-- [ ] Chain conjunction mask reused within bounded lifetime.
-- [ ] Single retain mask for single-sampler chains.
-- [ ] Reusable decision worker (or empirical evidence that the current goroutine/channel/timer pattern is cheaper)
+- [x] Sampler-wrapper slice built once per batch (or per worker) rather than per batch.
+- [x] Chain conjunction mask reused within bounded lifetime.
+- [x] Single retain mask for single-sampler chains.
+- [x] Reusable decision worker (or empirical evidence that the current goroutine/channel/timer pattern is cheaper)
selected.
-- [ ] Fragment-guard ranges stored in compact integers; bounded guard-range vectors pooled; drop-specific probe state
+- [x] Fragment-guard ranges stored in compact integers; bounded guard-range vectors pooled; drop-specific probe state
allocated lazily.
-- [ ] AlwaysKeep verification: with no fragment-guard confirmations, no throughput regression.
+- [x] AlwaysKeep verification: with no fragment-guard confirmations, no throughput regression.
**Dependencies.** Phase 3.
**Boundary rationale.** Lower priority than raw staging because the opening controlled round only makes two plugin
calls. Following Phase 3 keeps the structural changes from masking any decision-path gains.
+**Implementation and measurement note.** The chain now filters and stores its active samplers once, records observation
+metrics without constructing per-batch wrappers, and lets a single-sampler chain copy its verdict directly into the
+pooled decision mask. Healthy decisions reuse one worker and timer for the merge lifetime. A timeout abandons that worker
+and its caller-owned storage rather than risking reuse while plugin code may still read it, preserving fail-open ownership
+and circuit-breaker behavior. Fragment-guard trace ranges remain compact pooled integer pairs, while their block scratch
+is populated lazily only for plugin-proposed drops and is retained only within the existing bounded staging-vector pool.
+
+The focused 512-trace benchmarks reduced `BenchmarkMergeChainRunObserved` from approximately 613 ns/op, 48 B/op, and
+2 allocs/op to 539-617 ns/op, 0 B/op, and 0 allocs/op. `BenchmarkMergeChainExecuteObserved` fell from approximately
+2.24-2.36 us/op, 496 B/op, and 8 allocs/op to 1.80-1.87 us/op, 0 B/op, and 0 allocs/op.
+
+Five fresh two-CPU, 4 GiB Docker runs used the frozen 15-part mature selection and AlwaysKeep plugin. Every run made five
+plugin calls, evaluated and retained 33,353 traces, wrote 147,126 rows, and preserved the core and two secondary-index
+ledgers. Against the preceding adaptive-batch acceptance medians, wall time changed by +2.97%, CPU time by +2.34%,
+allocated bytes by -4.10%, allocation count by +0.009%, peak heap by -1.93%, and peak RSS by +3.61%. Wall-time and CPU
+coefficients of variation remained below 1.6%. The full-merge result therefore shows no regression beyond the 5%
+tolerance; only the focused measurements are credited as a decision-path improvement.
+
---
### Phase 6 — Drop-Path Optimization
diff --git a/pkg/pipeline/sdk/chain.go b/pkg/pipeline/sdk/chain.go
index c6659edda..4d02c649d 100644
--- a/pkg/pipeline/sdk/chain.go
+++ b/pkg/pipeline/sdk/chain.go
@@ -86,6 +86,19 @@ func EvaluateChainInto(samplers []Sampler, batch *TraceBatch, mask []bool, onByp
} else {
mask = mask[:len(batch.Traces)]
}
+ if len(samplers) == 1 {
+ if samplers[0] != nil {
+ verdict, valid := evaluateChainLink(0, samplers[0], batch, onBypass)
+ if valid {
+ copy(mask, verdict.Keep)
+ return Verdict{Keep: mask}
+ }
+ }
+ for traceIdx := range mask {
+ mask[traceIdx] = true
+ }
+ return Verdict{Keep: mask}
+ }
for i := range mask {
mask[i] = true
}
@@ -102,6 +115,16 @@ func EvaluateChainInto(samplers []Sampler, batch *TraceBatch, mask []bool, onByp
// into mask. On panic, error, or length mismatch the link is bypassed (mask
// unchanged) and onBypass, if non-nil, is invoked with the classified reason.
func applyChainLink(idx int, sampler Sampler, batch *TraceBatch, mask []bool, onBypass func(idx int, info BypassInfo)) {
+ verdict, valid := evaluateChainLink(idx, sampler, batch, onBypass)
+ if !valid {
+ return
+ }
+ for i := range mask {
+ mask[i] = mask[i] && verdict.Keep[i]
+ }
+}
+
+func evaluateChainLink(idx int, sampler Sampler, batch *TraceBatch, onBypass func(idx int, info BypassInfo)) (Verdict, bool) {
var (
verdict Verdict
decErr error
@@ -124,15 +147,13 @@ func applyChainLink(idx int, sampler Sampler, batch *TraceBatch, mask []bool, on
}
onBypass(idx, BypassInfo{Reason: reason, Err: decErr})
}
- return
+ return Verdict{}, false
}
if len(verdict.Keep) != len(batch.Traces) {
if onBypass != nil {
onBypass(idx, BypassInfo{Reason: BypassReasonLengthMismatch, Got: len(verdict.Keep), Want: len(batch.Traces)})
}
- return
- }
- for i := range mask {
- mask[i] = mask[i] && verdict.Keep[i]
+ return Verdict{}, false
}
+ return verdict, true
}
diff --git a/pkg/pipeline/sdk/chain_test.go b/pkg/pipeline/sdk/chain_test.go
index 284e307c6..cb9f96dcb 100644
--- a/pkg/pipeline/sdk/chain_test.go
+++ b/pkg/pipeline/sdk/chain_test.go
@@ -32,6 +32,7 @@ import (
// wrong-length verdict to exercise the three bypass reasons.
type chainFakeSampler struct {
dropIDs map[string]struct{}
+ calls int
panicNow bool
errNow bool
wrongSize bool
@@ -42,6 +43,7 @@ func (f *chainFakeSampler) Project() sdk.Projection { return sdk.Projection{} }
func (f *chainFakeSampler) Close() error { return nil }
func (f *chainFakeSampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) {
+ f.calls++
if f.panicNow {
panic("boom")
}
@@ -84,6 +86,37 @@ func TestEvaluateChainIntoReusesCallerMask(t *testing.T) {
require.Equal(t, &mask[0], &verdict.Keep[0])
}
+func TestEvaluateChainIntoSingleSamplerBypassCallsOnce(t *testing.T) {
+ batch := batchOf("a", "b")
+ sampler := &chainFakeSampler{errNow: true}
+ verdict := sdk.EvaluateChainInto([]sdk.Sampler{sampler}, batch, make([]bool, len(batch.Traces)), nil)
+
+ require.Equal(t, []bool{true, true}, verdict.Keep)
+ require.Equal(t, 1, sampler.calls)
+}
+
+type reusableChainBenchmarkSampler struct {
+ keep []bool
+}
+
+func (rcbs *reusableChainBenchmarkSampler) Kind() sdk.Kind { return sdk.KindSampler }
+func (rcbs *reusableChainBenchmarkSampler) Project() sdk.Projection { return sdk.Projection{} }
+func (rcbs *reusableChainBenchmarkSampler) Close() error { return nil }
+func (rcbs *reusableChainBenchmarkSampler) Decide(*sdk.TraceBatch) (sdk.Verdict, error) {
+ return sdk.Verdict{Keep: rcbs.keep}, nil
+}
+
+func BenchmarkEvaluateChainIntoSingleSampler(b *testing.B) {
+ const traceCount = 512
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, traceCount)}
+ sampler := &reusableChainBenchmarkSampler{keep: make([]bool, traceCount)}
+ mask := make([]bool, traceCount)
+ b.ReportAllocs()
+ for range b.N {
+ sdk.EvaluateChainInto([]sdk.Sampler{sampler}, batch, mask, nil)
+ }
+}
+
func TestEvaluateChain_NilSamplerSkipped(t *testing.T) {
batch := batchOf("a", "b")
verdict := sdk.EvaluateChain([]sdk.Sampler{nil, &chainFakeSampler{}}, batch, nil)