[processor/tailsampling] Simplify tsp locking (open-telemetry#43671)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
This PR changes the tailsampling processor to do all work (append traces
to a batch, decide on which traces from a batch are sampled, forward to
the next processor) in a single goroutine. At Grafana Labs we have found
that there is significant contention between the various mutexes such
that little performance is gained by allowing traces to be updated while
a policy evaluation is occurring. In addition, the locking adds a lot of
complexity which makes future improvements more challenging, and
introduces subtle concurrency issues such as seen in the tracking issue
linked below.

We (mostly @Logiraptor) also experimented with using multiple workers,
each with their own goroutine to process traces but we found little
improvement as yet again various contention issues showed up. See
Logiraptor#1 (or
the first commit in this PR) for more details.

If additional performance is necessary for slow decision tick latencies
I would suggest adding parallelism just to the call to `makeDecision`,
which would allow additional CPUs to be used without adding contention
on `idToTrace` or other locations.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#41656 
Working towards
open-telemetry#43876

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Tests were updated to make sure the tick behavior remains correct and
consistent. In addition, the syncBatcher needed to be updated to
properly implement `Stop`.

---------

Co-authored-by: Patrick Oyarzun <[email protected]>

Author: csmarchbanks

.chloggen/simplify-tsp-locking.yaml

@@ -0,0 +1,30 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: processor/tail_sampling
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Simplify the locking used by the tail sampling
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [41656, 43671]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  There are two small breaking changes as part of this work:
+  1. Pending traces are now passed through the decision logic during shutdown by default. If this is not desired it can be turned off using `drop_pending_traces_on_shutdown`.
+  2. The mutex in `samplingpolicy.TraceData` has been removed and `samplingpolicy.SpanCount` is now an `int64` instead of `*atomic.Int64`. Custom extensions using these fields will need to be updated.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

processor/tailsamplingprocessor/README.md

@@ -61,6 +61,8 @@ The following configuration options can also be modified:
     persisting the "drop" decisions for traces that may have already been released from memory.
     By default, the size is 0 and the cache is inactive.
 - `sample_on_first_match`: Make decision as soon as a policy matches
+- `drop_pending_traces_on_shutdown`: Drop pending traces on shutdown instead of making a decision with the partial data
+  already ingested.
 
 
 Each policy will result in a decision, and the processor will evaluate them to make a final decision:

processor/tailsamplingprocessor/config.go

@@ -291,4 +291,7 @@ type Config struct {
 	Options []Option `mapstructure:"-"`
 	// Make decision as soon as a policy matches
 	SampleOnFirstMatch bool `mapstructure:"sample_on_first_match"`
+	// DropPendingTracesOnShutdown will drop all traces that are part of batches that have not yet reached the decision
+	// wait when the processor is shutdown.
+	DropPendingTracesOnShutdown bool `mapstructure:"drop_pending_traces_on_shutdown"`
 }

processor/tailsamplingprocessor/go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/golang-lru/v2 v2.0.7
-	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.140.1
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.140.1 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.140.1
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.140.1
 	github.com/stretchr/testify v1.11.1

processor/tailsamplingprocessor/internal/sampling/boolean_tag_filter.go

@@ -36,8 +36,6 @@ func NewBooleanAttributeFilter(settings component.TelemetrySettings, key string,
 
 // Evaluate looks at the trace data and returns a corresponding SamplingDecision.
 func (baf *booleanAttributeFilter) Evaluate(_ context.Context, _ pcommon.TraceID, trace *samplingpolicy.TraceData) (samplingpolicy.Decision, error) {
-	trace.Lock()
-	defer trace.Unlock()
 	batches := trace.ReceivedBatches
 
 	if baf.invertMatch {

processor/tailsamplingprocessor/internal/sampling/composite.go

@@ -111,7 +111,7 @@ func (c *Composite) Evaluate(ctx context.Context, traceID pcommon.TraceID, trace
 			// The subpolicy made a decision to Sample. Now we need to make our decision.
 
 			// Calculate resulting SPS counter if we decide to sample this trace
-			spansInSecondIfSampled := sub.sampledSPS + trace.SpanCount.Load()
+			spansInSecondIfSampled := sub.sampledSPS + trace.SpanCount
 
 			// Check if the rate will be within the allocated bandwidth.
 			if spansInSecondIfSampled <= sub.allocatedSPS && spansInSecondIfSampled <= c.maxTotalSPS {

processor/tailsamplingprocessor/internal/sampling/composite_test.go

@@ -3,7 +3,6 @@
 package sampling
 
 import (
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -28,9 +27,7 @@ func (f FakeTimeProvider) getCurSecond() int64 {
 var traceID = pcommon.TraceID([16]byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x52, 0x96, 0x9A, 0x89, 0x55, 0x57, 0x1A, 0x3F})
 
 func createTrace() *samplingpolicy.TraceData {
-	spanCount := &atomic.Int64{}
-	spanCount.Store(1)
-	trace := &samplingpolicy.TraceData{SpanCount: spanCount, ReceivedBatches: ptrace.NewTraces()}
+	trace := &samplingpolicy.TraceData{SpanCount: 1, ReceivedBatches: ptrace.NewTraces()}
 	return trace
 }
 
@@ -49,11 +46,9 @@ func newTraceWithKV(traceID pcommon.TraceID, key string, val int64) *samplingpol
 	))
 	span.Attributes().PutInt(key, val)
 
-	spanCount := &atomic.Int64{}
-	spanCount.Store(1)
 	return &samplingpolicy.TraceData{
 		ReceivedBatches: traces,
-		SpanCount:       spanCount,
+		SpanCount:       1,
 	}
 }
 

processor/tailsamplingprocessor/internal/sampling/latency.go

@@ -35,8 +35,6 @@ func NewLatency(settings component.TelemetrySettings, thresholdMs, upperThreshol
 func (l *latency) Evaluate(_ context.Context, _ pcommon.TraceID, traceData *samplingpolicy.TraceData) (samplingpolicy.Decision, error) {
 	l.logger.Debug("Evaluating spans in latency filter")
 
-	traceData.Lock()
-	defer traceData.Unlock()
 	batches := traceData.ReceivedBatches
 
 	var minTime pcommon.Timestamp

processor/tailsamplingprocessor/internal/sampling/numeric_tag_filter.go

@@ -44,8 +44,6 @@ func NewNumericAttributeFilter(settings component.TelemetrySettings, key string,
 
 // Evaluate looks at the trace data and returns a corresponding SamplingDecision.
 func (naf *numericAttributeFilter) Evaluate(_ context.Context, _ pcommon.TraceID, trace *samplingpolicy.TraceData) (samplingpolicy.Decision, error) {
-	trace.Lock()
-	defer trace.Unlock()
 	batches := trace.ReceivedBatches
 
 	// Get the effective min/max values

processor/tailsamplingprocessor/internal/sampling/ottl.go

@@ -64,8 +64,6 @@ func (ocf *ottlConditionFilter) Evaluate(ctx context.Context, traceID pcommon.Tr
 		return samplingpolicy.NotSampled, nil
 	}
 
-	trace.Lock()
-	defer trace.Unlock()
 	batches := trace.ReceivedBatches
 
 	for i := 0; i < batches.ResourceSpans().Len(); i++ {