From c72e96ad6c684dcd22950a6561249ea2f8754c91 Mon Sep 17 00:00:00 2001 From: Garry Cairns Date: Fri, 15 Sep 2023 15:00:57 +0100 Subject: [PATCH] Adds upper bound to latency tail sampling as per review feedback --- .gitignore | 1 + processor/tailsamplingprocessor/README.md | 8 +- .../tailsamplingprocessor/and_helper_test.go | 2 +- .../composite_helper_test.go | 4 +- processor/tailsamplingprocessor/config.go | 15 +--- .../internal/sampling/latency.go | 15 ++-- .../internal/sampling/latency_test.go | 89 ++++++++++++++++++- processor/tailsamplingprocessor/processor.go | 2 +- 8 files changed, 106 insertions(+), 30 deletions(-) diff --git a/.gitignore b/.gitignore index fd937a38f1b4..52616038f614 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,4 @@ integration-coverage.html go.work* +/result diff --git a/processor/tailsamplingprocessor/README.md b/processor/tailsamplingprocessor/README.md index 9b394256cecc..c035a49ed6b8 100644 --- a/processor/tailsamplingprocessor/README.md +++ b/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, diff --git a/processor/tailsamplingprocessor/and_helper_test.go b/processor/tailsamplingprocessor/and_helper_test.go index f1cbe8a1e587..e46237ed95ed 100644 --- a/processor/tailsamplingprocessor/and_helper_test.go +++ b/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) }) diff --git a/processor/tailsamplingprocessor/composite_helper_test.go b/processor/tailsamplingprocessor/composite_helper_test.go index bc7d3cc2053e..578f3be289b9 100644 --- a/processor/tailsamplingprocessor/composite_helper_test.go +++ b/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{}) diff --git a/processor/tailsamplingprocessor/config.go b/processor/tailsamplingprocessor/config.go index e37c9da57e90..df76dfaf55b1 100644 --- a/processor/tailsamplingprocessor/config.go +++ b/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 diff --git a/processor/tailsamplingprocessor/internal/sampling/latency.go b/processor/tailsamplingprocessor/internal/sampling/latency.go index 760b4e9f6768..20e3710b2f76 100644 --- a/processor/tailsamplingprocessor/internal/sampling/latency.go +++ b/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 } diff --git a/processor/tailsamplingprocessor/internal/sampling/latency_test.go b/processor/tailsamplingprocessor/internal/sampling/latency_test.go index 839feddcbf9d..b5a2537edf65 100644 --- a/processor/tailsamplingprocessor/internal/sampling/latency_test.go +++ b/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 diff --git a/processor/tailsamplingprocessor/processor.go b/processor/tailsamplingprocessor/processor.go index bfcbaac0f8d5..6259bb72825f 100644 --- a/processor/tailsamplingprocessor/processor.go +++ b/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