Skip to content
27 changes: 22 additions & 5 deletions processor/tailsamplingprocessor/processor.go
Original file line number Diff line number Diff line change
Expand Up @@ -350,8 +350,11 @@ func (tsp *tailSamplingSpanProcessor) samplingPolicyOnTick() {
trace.ReceivedBatches = ptrace.NewTraces()
trace.Unlock()

if decision == sampling.Sampled {
switch decision {
case sampling.Sampled:
tsp.releaseSampledTrace(context.Background(), id, allSpans)
case sampling.NotSampled:
tsp.releaseNotSampledTrace(id)
}
}

Expand Down Expand Up @@ -517,7 +520,7 @@ func (tsp *tailSamplingSpanProcessor) processTraces(resourceSpans ptrace.Resourc
appendToTraces(traceTd, resourceSpans, spans)
tsp.releaseSampledTrace(tsp.ctx, id, traceTd)
case sampling.NotSampled:
tsp.nonSampledIDCache.Put(id, true)
tsp.releaseNotSampledTrace(id)
tsp.telemetry.ProcessorTailSamplingSamplingLateSpanAge.Record(tsp.ctx, int64(time.Since(actualData.DecisionTime)/time.Second))
default:
tsp.logger.Warn("Encountered unexpected sampling decision",
Expand Down Expand Up @@ -562,16 +565,30 @@ func (tsp *tailSamplingSpanProcessor) dropTrace(traceID pcommon.TraceID, deletio
tsp.telemetry.ProcessorTailSamplingSamplingTraceRemovalAge.Record(tsp.ctx, int64(deletionTime.Sub(trace.ArrivalTime)/time.Second))
}

// releaseSampledTrace sends the trace data to the next consumer.
// It additionally adds the trace ID to the cache of sampled trace IDs.
// It does not (yet) delete the spans from the internal map.
// releaseSampledTrace sends the trace data to the next consumer. It
// additionally adds the trace ID to the cache of sampled trace IDs. If the
// trace ID is cached, it deletes the spans from the internal map.
func (tsp *tailSamplingSpanProcessor) releaseSampledTrace(ctx context.Context, id pcommon.TraceID, td ptrace.Traces) {
tsp.sampledIDCache.Put(id, true)
if err := tsp.nextConsumer.ConsumeTraces(ctx, td); err != nil {
tsp.logger.Warn(
"Error sending spans to destination",
zap.Error(err))
}
_, ok := tsp.sampledIDCache.Get(id)
if ok {
tsp.dropTrace(id, time.Now())
}
}

// releaseNotSampledTrace adds the trace ID to the cache of not sampled trace
// IDs. If the trace ID is cached, it deletes the spans from the internal map.
func (tsp *tailSamplingSpanProcessor) releaseNotSampledTrace(id pcommon.TraceID) {
tsp.nonSampledIDCache.Put(id, true)
_, ok := tsp.nonSampledIDCache.Get(id)
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This Get() would be unnecessary if the cache interface Put() returned bool, error. Saving for a potential future enhancement.

if ok {
tsp.dropTrace(id, time.Now())
}
}

func appendToTraces(dest ptrace.Traces, rss ptrace.ResourceSpans, spanAndScopes []spanAndScope) {
Expand Down
Loading