[processor/probabilisticsampler] Fix logs sampling priority mechanism (#38470)

#### Description

Fixes #38468.

#### Testing

Adds testing.

#### Documentation

✔️

Author: jmacd

.chloggen/logsprioritysampler.yaml

@@ -0,0 +1,6 @@
+change_type: bug_fix
+component: probabilisticsampler
+note: Logs priority sampling behavior applies only when the priority attribute is present.
+issues: [38468]
+subtext:
+change_logs: [user]

processor/probabilisticsamplerprocessor/README.md

@@ -112,10 +112,11 @@ attribute is not configurable, and is called `sampling.priority`.
 
 In logs pipelines, when the priority attribute has value 0, the
 configured probability will by modified to 0%, and the item will not
-pass the sampler.  Otherwise, the logs sampling priority attribute is
+pass the sampler. Otherwise, the logs sampling priority attribute is
 interpreted as a percentage, with values >= 100 equal to 100%
-sampling.  The logs sampling priority attribute is configured via
-`sampling_priority`.
+sampling. The logs sampling priority attribute is configured via
+`sampling_priority`. If the logs sampling priority attribute is not
+set on a log record, the default sampling percentage applies.
 
 ## Mode Selection
 

processor/probabilisticsamplerprocessor/logsprocessor.go

@@ -240,20 +240,16 @@ func (lsp *logsProcessor) priorityFunc(logRec plog.LogRecord, rnd randomnessName
 	// Note: in logs, unlike traces, the sampling priority
 	// attribute is interpreted as a request to be sampled.
 	if lsp.samplingPriority != "" {
-		priorityThreshold := lsp.logRecordToPriorityThreshold(logRec)
-
-		if priorityThreshold == sampling.NeverSampleThreshold {
-			threshold = priorityThreshold
-			rnd = newSamplingPriorityMethod(rnd.randomness()) // override policy name
-		} else if sampling.ThresholdLessThan(priorityThreshold, threshold) {
+		priorityThreshold, has := lsp.logRecordToPriorityThreshold(logRec)
+		if has {
 			threshold = priorityThreshold
 			rnd = newSamplingPriorityMethod(rnd.randomness()) // override policy name
 		}
 	}
 	return rnd, threshold
 }
 
-func (lsp *logsProcessor) logRecordToPriorityThreshold(logRec plog.LogRecord) sampling.Threshold {
+func (lsp *logsProcessor) logRecordToPriorityThreshold(logRec plog.LogRecord) (sampling.Threshold, bool) {
 	if localPriority, ok := logRec.Attributes().Get(lsp.samplingPriority); ok {
 		// Potentially raise the sampling probability to minProb
 		minProb := 0.0
@@ -266,9 +262,9 @@ func (lsp *logsProcessor) logRecordToPriorityThreshold(logRec plog.LogRecord) sa
 		if minProb != 0 {
 			if th, err := sampling.ProbabilityToThresholdWithPrecision(minProb, lsp.precision); err == nil {
 				// The record has supplied a valid alternative sampling probability
-				return th
+				return th, true
 			}
 		}
 	}
-	return sampling.NeverSampleThreshold
+	return sampling.NeverSampleThreshold, false
 }

processor/probabilisticsamplerprocessor/logsprocessor_test.go

@@ -151,13 +151,64 @@ func TestLogsSampling(t *testing.T) {
 			received: 29, // probabilistic... doesn't yield the same results as foo
 		},
 		{
-			name: "sampling_priority",
+			name: "sampling_priority_0_record",
 			cfg: &Config{
 				SamplingPercentage: 0,
 				SamplingPriority:   "priority",
+				AttributeSource:    recordAttributeSource,
+				FromAttribute:      "rando",
 			},
 			received: 25,
 		},
+		{
+			name: "sampling_priority_50_record",
+			cfg: &Config{
+				SamplingPercentage: 50,
+				SamplingPriority:   "priority",
+				AttributeSource:    recordAttributeSource,
+				FromAttribute:      "rando",
+			},
+			// Expected value is (0.5*75)+25 == 62.5
+			received: 64,
+		},
+		{
+			name: "sampling_priority_100_record",
+			cfg: &Config{
+				SamplingPercentage: 100,
+				SamplingPriority:   "priority",
+				AttributeSource:    recordAttributeSource,
+				FromAttribute:      "rando",
+			},
+			received: 100,
+		},
+		{
+			name: "sampling_priority_0_traceid",
+			cfg: &Config{
+				SamplingPercentage: 0,
+				SamplingPriority:   "priority",
+				AttributeSource:    traceIDAttributeSource,
+			},
+			received: 25,
+		},
+		{
+			name: "sampling_priority_50_traceid",
+			cfg: &Config{
+				SamplingPercentage: 50,
+				SamplingPriority:   "priority",
+				AttributeSource:    traceIDAttributeSource,
+			},
+			// Expected value is (0.5*75)+25 == 62.5
+			received: 57,
+		},
+		{
+			name: "sampling_priority_100_traceid",
+			cfg: &Config{
+				SamplingPercentage: 100,
+				SamplingPriority:   "priority",
+				AttributeSource:    traceIDAttributeSource,
+			},
+			received: 100,
+		},
 		{
 			name: "sampling_priority with sampling field",
 			cfg: &Config{
@@ -185,6 +236,7 @@ func TestLogsSampling(t *testing.T) {
 				// encodes historical behavior, we leave it as-is.
 				traceID := [16]byte{0, 0, 0, 0, 0, 0, 0, 0, ib, ib, ib, ib, ib, ib, ib, ib}
 				record.SetTraceID(traceID)
+				record.Attributes().PutInt("rando", int64(computeHash(traceID[:], 123)))
 				// set half of records with a foo (bytes) and a bar (string) attribute
 				if i%2 == 0 {
 					b := record.Attributes().PutEmptyBytes("foo")

processor/probabilisticsamplerprocessor/sampler_mode.go

@@ -236,7 +236,7 @@ type hashingSampler struct {
 	// Logs only: name of attribute to obtain randomness
 	logsRandomnessSourceAttribute string
 
-	// Logs only: name of attribute to obtain randomness
+	// Logs only: whether traceID is being used
 	logsTraceIDEnabled bool
 }