Adds upper bound to latency tail sampling as per review feedback

.gitignore

@@ -38,3 +38,4 @@ integration-coverage.html
 
 go.work*
 
+/result

processor/tailsamplingprocessor/README.md

@@ -29,7 +29,6 @@ The following configuration options are required:
 Multiple policies exist today and it is straight forward to add more. These include:
 - `always_sample`: Sample all traces
 - `latency`: Sample based on the duration of the trace. The duration is determined by looking at the earliest start time and latest end time, without taking into consideration what happened in between.
-- `duration`: Sample based on the *bounded* duration of the trace. Otherwise identical to latency
 - `numeric_attribute`: Sample based on number attributes (resource and record)
 - `probabilistic`: Sample a percentage of traces. Read [a comparison with the Probabilistic Sampling Processor](#probabilistic-sampling-processor-compared-to-the-tail-sampling-processor-with-the-probabilistic-policy).
 - `status_code`: Sample based upon the status code (`OK`, `ERROR` or `UNSET`)
@@ -77,12 +76,7 @@ processors:
           {
             name: test-policy-2,
             type: latency,
-            latency: {threshold_ms: 5000}
-          },
-          {
-            name: test-policy-2a,
-            type: duration,
-            latency: {lower_bound_ms: 5000, upper_bound_ms: 10000}
+            latency: {threshold_ms: 5000, upper_threshold_ms: 10000}
           },
           {
             name: test-policy-3,

processor/tailsamplingprocessor/and_helper_test.go

@@ -30,7 +30,7 @@ func TestAndHelper(t *testing.T) {
 		require.NoError(t, err)
 
 		expected := sampling.NewAnd(zap.NewNop(), []sampling.PolicyEvaluator{
-			sampling.NewLatency(componenttest.NewNopTelemetrySettings(), 100),
+			sampling.NewLatency(componenttest.NewNopTelemetrySettings(), 100, 0),
 		})
 		assert.Equal(t, expected, actual)
 	})

processor/tailsamplingprocessor/composite_helper_test.go

@@ -50,11 +50,11 @@ func TestCompositeHelper(t *testing.T) {
 
 		expected := sampling.NewComposite(zap.NewNop(), 1000, []sampling.SubPolicyEvalParams{
 			{
-				Evaluator:         sampling.NewLatency(componenttest.NewNopTelemetrySettings(), 100),
+				Evaluator:         sampling.NewLatency(componenttest.NewNopTelemetrySettings(), 100, 0),
 				MaxSpansPerSecond: 250,
 			},
 			{
-				Evaluator:         sampling.NewLatency(componenttest.NewNopTelemetrySettings(), 200),
+				Evaluator:         sampling.NewLatency(componenttest.NewNopTelemetrySettings(), 200, 0),
 				MaxSpansPerSecond: 500,
 			},
 		}, sampling.MonotonicClock{})

processor/tailsamplingprocessor/config.go

@@ -17,8 +17,6 @@ const (
 	AlwaysSample PolicyType = "always_sample"
 	// Latency sample traces that are longer than a given threshold.
 	Latency PolicyType = "latency"
-	// Duration sample traces that are longer than a given threshold.
-	Duration PolicyType = "duration"
 	// NumericAttribute sample traces that have a given numeric attribute in a specified
 	// range, e.g.: attribute "http.status_code" >= 399 and <= 999.
 	NumericAttribute PolicyType = "numeric_attribute"
@@ -56,8 +54,6 @@ type sharedPolicyCfg struct {
 	Type PolicyType `mapstructure:"type"`
 	// Configs for latency filter sampling policy evaluator.
 	LatencyCfg LatencyCfg `mapstructure:"latency"`
-	// Configs for duration filter sampling policy evaluator.
-	DurationCfg DurationCfg `mapstructure:"duration"`
 	// Configs for numeric attribute filter sampling policy evaluator.
 	NumericAttributeCfg NumericAttributeCfg `mapstructure:"numeric_attribute"`
 	// Configs for probabilistic sampling policy evaluator.
@@ -130,17 +126,10 @@ type PolicyCfg struct {
 // LatencyCfg holds the configurable settings to create a latency filter sampling policy
 // evaluator
 type LatencyCfg struct {
-	// ThresholdMs in milliseconds.
+	// Lower bound in milliseconds. Retaining original name for compatibility
 	ThresholdMs int64 `mapstructure:"threshold_ms"`
-}
-
-// DurationCfg holds the configurable settings to create a duration filter sampling policy
-// evaluator, which is essentially a latency sampler with two bounds
-type DurationCfg struct {
-	// Lower bound in milliseconds.
-	LowerDurationMs int64 `mapstructure:"lower_bound_ms"`
 	// Upper bound in milliseconds.
-	UpperDurationMs int64 `mapstructure:"upper_bound_ms"`
+	UpperThresholdmsMs int64 `mapstructure:"upper_threshold_ms"`
 }
 
 // NumericAttributeCfg holds the configurable settings to create a numeric attribute filter

processor/tailsamplingprocessor/internal/sampling/latency.go

@@ -13,17 +13,19 @@ import (
 )
 
 type latency struct {
-	logger      *zap.Logger
+	logger           *zap.Logger
 	thresholdMs int64
+	upperThresholdMs int64
 }
 
 var _ PolicyEvaluator = (*latency)(nil)
 
 // NewLatency creates a policy evaluator sampling traces with a duration higher than a configured threshold
-func NewLatency(settings component.TelemetrySettings, thresholdMs int64) PolicyEvaluator {
+func NewLatency(settings component.TelemetrySettings, thresholdMs int64, upperThresholdMs int64) PolicyEvaluator {
 	return &latency{
-		logger:      settings.Logger,
-		thresholdMs: thresholdMs,
+		logger:           settings.Logger,
+		thresholdMs:      thresholdMs,
+		upperThresholdMs: upperThresholdMs,
 	}
 }
 
@@ -47,6 +49,9 @@ func (l *latency) Evaluate(_ context.Context, _ pcommon.TraceID, traceData *Trac
 		}
 
 		duration := maxTime.AsTime().Sub(minTime.AsTime())
-		return duration.Milliseconds() >= l.thresholdMs
+		if l.upperThresholdMs == 0 {
+			return duration.Milliseconds() >= l.thresholdMs
+		}
+		return (l.thresholdMs < duration.Milliseconds() && duration.Milliseconds() <= l.upperThresholdMs)
 	}), nil
 }

processor/tailsamplingprocessor/internal/sampling/latency_test.go

@@ -15,7 +15,7 @@ import (
 )
 
 func TestEvaluate_Latency(t *testing.T) {
-	filter := NewLatency(componenttest.NewNopTelemetrySettings(), 5000)
+	filter := NewLatency(componenttest.NewNopTelemetrySettings(), 5000, 0)
 
 	traceID := pcommon.TraceID([16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16})
 	now := time.Now()
@@ -71,6 +71,93 @@ func TestEvaluate_Latency(t *testing.T) {
 	}
 }
 
+func TestEvaluate_Bounded_Latency(t *testing.T) {
+	filter := NewLatency(componenttest.NewNopTelemetrySettings(), 5000, 10000)
+
+	traceID := pcommon.TraceID([16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16})
+	now := time.Now()
+
+	cases := []struct {
+		Desc     string
+		Spans    []spanWithTimeAndDuration
+		Decision Decision
+	}{
+		{
+			"trace duration shorter than lower bound",
+			[]spanWithTimeAndDuration{
+				{
+					StartTime: now,
+					Duration:  4500 * time.Millisecond,
+				},
+			},
+			NotSampled,
+		},
+		{
+			"trace duration is equal to lower bound",
+			[]spanWithTimeAndDuration{
+				{
+					StartTime: now,
+					Duration:  5000 * time.Millisecond,
+				},
+			},
+			NotSampled,
+		},
+		{
+			"trace duration is within lower and upper bounds",
+			[]spanWithTimeAndDuration{
+				{
+					StartTime: now,
+					Duration:  5001 * time.Millisecond,
+				},
+			},
+			Sampled,
+		},
+		{
+			"trace duration is above upper bound",
+			[]spanWithTimeAndDuration{
+				{
+					StartTime: now,
+					Duration:  10001 * time.Millisecond,
+				},
+			},
+			NotSampled,
+		},
+		{
+			"trace duration equals upper bound",
+			[]spanWithTimeAndDuration{
+				{
+					StartTime: now,
+					Duration:  10000 * time.Millisecond,
+				},
+			},
+			Sampled,
+		},
+		{
+			"total trace duration is longer than threshold but every single span is shorter",
+			[]spanWithTimeAndDuration{
+				{
+					StartTime: now,
+					Duration:  3000 * time.Millisecond,
+				},
+				{
+					StartTime: now.Add(2500 * time.Millisecond),
+					Duration:  3000 * time.Millisecond,
+				},
+			},
+			Sampled,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.Desc, func(t *testing.T) {
+			decision, err := filter.Evaluate(context.Background(), traceID, newTraceWithSpans(c.Spans))
+
+			assert.NoError(t, err)
+			assert.Equal(t, decision, c.Decision)
+		})
+	}
+}
+
 type spanWithTimeAndDuration struct {
 	StartTime time.Time
 	Duration  time.Duration

processor/tailsamplingprocessor/processor.go

@@ -133,7 +133,7 @@ func getSharedPolicyEvaluator(settings component.TelemetrySettings, cfg *sharedP
 		return sampling.NewAlwaysSample(settings), nil
 	case Latency:
 		lfCfg := cfg.LatencyCfg
-		return sampling.NewLatency(settings, lfCfg.ThresholdMs), nil
+		return sampling.NewLatency(settings, lfCfg.ThresholdMs, lfCfg.UpperThresholdmsMs), nil
 	case NumericAttribute:
 		nafCfg := cfg.NumericAttributeCfg
 		return sampling.NewNumericAttributeFilter(settings, nafCfg.Key, nafCfg.MinValue, nafCfg.MaxValue, nafCfg.InvertMatch), nil