Adds duration sampler distinct from latency in supplying two bounds

.gitignore

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

processor/tailsamplingprocessor/README.md

@@ -77,7 +77,7 @@ processors:
           {
             name: test-policy-2,
             type: latency,
-            latency: {threshold_ms: 5000}
+            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

@@ -126,8 +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"`
+	// Upper bound in milliseconds.
+	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

@@ -142,7 +142,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