[processor/tailsampling] add missing sub policies to AND policy

processor/tailsamplingprocessor/and_helper.go

@@ -15,49 +15,25 @@
 package tailsamplingprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor"
 
 import (
-	"fmt"
-
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/sampling"
 )
 
-func getNewAndPolicy(logger *zap.Logger, config AndCfg) (sampling.PolicyEvaluator, error) {
+func getNewAndPolicy(logger *zap.Logger, config *AndCfg) (sampling.PolicyEvaluator, error) {
 	var subPolicyEvaluators []sampling.PolicyEvaluator
 	for i := range config.SubPolicyCfg {
-		policyCfg := config.SubPolicyCfg[i]
-		policy, _ := getAndSubPolicyEvaluator(logger, &policyCfg)
+		policyCfg := &config.SubPolicyCfg[i]
+		policy, err := getAndSubPolicyEvaluator(logger, policyCfg)
+		if err != nil {
+			return nil, err
+		}
 		subPolicyEvaluators = append(subPolicyEvaluators, policy)
 	}
 	return sampling.NewAnd(logger, subPolicyEvaluators), nil
 }
 
 // Return instance of and sub-policy
 func getAndSubPolicyEvaluator(logger *zap.Logger, cfg *AndSubPolicyCfg) (sampling.PolicyEvaluator, error) {
-	switch cfg.Type {
-	case AlwaysSample:
-		return sampling.NewAlwaysSample(logger), nil
-	case NumericAttribute:
-		nafCfg := cfg.NumericAttributeCfg
-		return sampling.NewNumericAttributeFilter(logger, nafCfg.Key, nafCfg.MinValue, nafCfg.MaxValue), nil
-	case StringAttribute:
-		safCfg := cfg.StringAttributeCfg
-		return sampling.NewStringAttributeFilter(logger, safCfg.Key, safCfg.Values, safCfg.EnabledRegexMatching, safCfg.CacheMaxSize, safCfg.InvertMatch), nil
-	case RateLimiting:
-		rlfCfg := cfg.RateLimitingCfg
-		return sampling.NewRateLimiting(logger, rlfCfg.SpansPerSecond), nil
-	case StatusCode:
-		return sampling.NewStatusCodeFilter(logger, cfg.StatusCodeCfg.StatusCodes)
-	case Probabilistic:
-		pfCfg := cfg.ProbabilisticCfg
-		return sampling.NewProbabilisticSampler(logger, pfCfg.HashSalt, pfCfg.SamplingPercentage), nil
-	case TraceState:
-		tsfCfg := cfg.TraceStateCfg
-		return sampling.NewTraceStateFilter(logger, tsfCfg.Key, tsfCfg.Values), nil
-	case SpanCount:
-		scfCfg := cfg.SpanCountCfg
-		return sampling.NewSpanCount(logger, scfCfg.MinSpans), nil
-	default:
-		return nil, fmt.Errorf("unknown sampling policy type %s", cfg.Type)
-	}
+	return getSharedPolicyEvaluator(logger, &cfg.sharedPolicyCfg)
 }

processor/tailsamplingprocessor/and_helper_test.go

@@ -17,58 +17,45 @@ package tailsamplingprocessor
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/sampling"
 )
 
 func TestAndHelper(t *testing.T) {
-	andCfg := &AndCfg{
-		SubPolicyCfg: []AndSubPolicyCfg{
-			{
-				Name: "test-and-policy-1",
-				Type: AlwaysSample,
-			},
-			{
-				Name:                "test-and-policy-2",
-				Type:                NumericAttribute,
-				NumericAttributeCfg: NumericAttributeCfg{Key: "key1", MinValue: 50, MaxValue: 100},
-			},
-			{
-				Name:               "test-and-policy-3",
-				Type:               StringAttribute,
-				StringAttributeCfg: StringAttributeCfg{Key: "key2", Values: []string{"value1", "value2"}},
-			},
-			{
-				Name:            "test-and-policy-4",
-				Type:            RateLimiting,
-				RateLimitingCfg: RateLimitingCfg{SpansPerSecond: 10},
-			},
-			{
-				Name:          "test-and-policy-5",
-				Type:          StatusCode,
-				StatusCodeCfg: StatusCodeCfg{StatusCodes: []string{"ERROR", "UNSET"}},
-			},
-			{
-				Name:             "test-and-policy-6",
-				Type:             Probabilistic,
-				ProbabilisticCfg: ProbabilisticCfg{HashSalt: "salt", SamplingPercentage: 10},
-			},
-			{
-				Name:          "test-and-policy-3",
-				Type:          TraceState,
-				TraceStateCfg: TraceStateCfg{Key: "key3", Values: []string{"value1", "value2"}},
-			},
-			{
-				Name:         "test-and-policy-1",
-				Type:         SpanCount,
-				SpanCountCfg: SpanCountCfg{MinSpans: 2},
-			},
-		},
-	}
+	t.Run("valid", func(t *testing.T) {
+		actual, err := getNewAndPolicy(zap.NewNop(), &AndCfg{
+			SubPolicyCfg: []AndSubPolicyCfg{
+				{
+					sharedPolicyCfg: sharedPolicyCfg{
+						Name:       "test-and-policy-1",
+						Type:       Latency,
+						LatencyCfg: LatencyCfg{ThresholdMs: 100},
+					},
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		expected := sampling.NewAnd(zap.NewNop(), []sampling.PolicyEvaluator{
+			sampling.NewLatency(zap.NewNop(), 100),
+		})
+		assert.Equal(t, expected, actual)
+	})
 
-	for i := range andCfg.SubPolicyCfg {
-		policy, e := getAndSubPolicyEvaluator(zap.NewNop(), &andCfg.SubPolicyCfg[i])
-		require.NotNil(t, policy)
-		require.NoError(t, e)
-	}
+	t.Run("unsupported sampling policy type", func(t *testing.T) {
+		_, err := getNewAndPolicy(zap.NewNop(), &AndCfg{
+			SubPolicyCfg: []AndSubPolicyCfg{
+				{
+					sharedPolicyCfg: sharedPolicyCfg{
+						Name: "test-and-policy-2",
+						Type: And, // nested and is not allowed
+					},
+				},
+			},
+		})
+		require.EqualError(t, err, "unknown sampling policy type and")
+	})
 }

processor/tailsamplingprocessor/composite_helper.go

@@ -15,19 +15,20 @@
 package tailsamplingprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor"
 
 import (
-	"fmt"
-
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/sampling"
 )
 
-func getNewCompositePolicy(logger *zap.Logger, config CompositeCfg) (sampling.PolicyEvaluator, error) {
+func getNewCompositePolicy(logger *zap.Logger, config *CompositeCfg) (sampling.PolicyEvaluator, error) {
 	var subPolicyEvalParams []sampling.SubPolicyEvalParams
 	rateAllocationsMap := getRateAllocationMap(config)
 	for i := range config.SubPolicyCfg {
-		policyCfg := config.SubPolicyCfg[i]
-		policy, _ := getSubPolicyEvaluator(logger, &policyCfg)
+		policyCfg := &config.SubPolicyCfg[i]
+		policy, err := getCompositeSubPolicyEvaluator(logger, policyCfg)
+		if err != nil {
+			return nil, err
+		}
 
 		evalParams := sampling.SubPolicyEvalParams{
 			Evaluator:         policy,
@@ -39,7 +40,7 @@ func getNewCompositePolicy(logger *zap.Logger, config CompositeCfg) (sampling.Po
 }
 
 // Apply rate allocations to the sub-policies
-func getRateAllocationMap(config CompositeCfg) map[string]float64 {
+func getRateAllocationMap(config *CompositeCfg) map[string]float64 {
 	rateAllocationsMap := make(map[string]float64)
 	maxTotalSPS := float64(config.MaxTotalSpansPerSecond)
 	// Default SPS determined by equally diving number of sub policies
@@ -55,37 +56,11 @@ func getRateAllocationMap(config CompositeCfg) map[string]float64 {
 }
 
 // Return instance of composite sub-policy
-func getSubPolicyEvaluator(logger *zap.Logger, cfg *SubPolicyCfg) (sampling.PolicyEvaluator, error) {
+func getCompositeSubPolicyEvaluator(logger *zap.Logger, cfg *CompositeSubPolicyCfg) (sampling.PolicyEvaluator, error) {
 	switch cfg.Type {
-	case AlwaysSample:
-		return sampling.NewAlwaysSample(logger), nil
-	case Latency:
-		lfCfg := cfg.LatencyCfg
-		return sampling.NewLatency(logger, lfCfg.ThresholdMs), nil
-	case NumericAttribute:
-		nafCfg := cfg.NumericAttributeCfg
-		return sampling.NewNumericAttributeFilter(logger, nafCfg.Key, nafCfg.MinValue, nafCfg.MaxValue), nil
-	case Probabilistic:
-		pfCfg := cfg.ProbabilisticCfg
-		return sampling.NewProbabilisticSampler(logger, pfCfg.HashSalt, pfCfg.SamplingPercentage), nil
-	case StatusCode:
-		scCfg := cfg.StatusCodeCfg
-		return sampling.NewStatusCodeFilter(logger, scCfg.StatusCodes)
-	case StringAttribute:
-		safCfg := cfg.StringAttributeCfg
-		return sampling.NewStringAttributeFilter(logger, safCfg.Key, safCfg.Values, safCfg.EnabledRegexMatching, safCfg.CacheMaxSize, safCfg.InvertMatch), nil
-	case RateLimiting:
-		rlfCfg := cfg.RateLimitingCfg
-		return sampling.NewRateLimiting(logger, rlfCfg.SpansPerSecond), nil
 	case And:
-		return getNewAndPolicy(logger, cfg.AndCfg)
-	case SpanCount:
-		scCfg := cfg.SpanCountCfg
-		return sampling.NewSpanCount(logger, scCfg.MinSpans), nil
-	case TraceState:
-		tsfCfg := cfg.TraceStateCfg
-		return sampling.NewTraceStateFilter(logger, tsfCfg.Key, tsfCfg.Values), nil
+		return getNewAndPolicy(logger, &cfg.AndCfg)
 	default:
-		return nil, fmt.Errorf("unknown sampling policy type %s", cfg.Type)
+		return getSharedPolicyEvaluator(logger, &cfg.sharedPolicyCfg)
 	}
 }

processor/tailsamplingprocessor/composite_helper_test.go

@@ -16,86 +16,72 @@ package tailsamplingprocessor
 
 import (
 	"testing"
-	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.opentelemetry.io/collector/config"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/sampling"
 )
 
 func TestCompositeHelper(t *testing.T) {
-	cfg := &Config{
-		ProcessorSettings:       config.NewProcessorSettings(config.NewComponentID(typeStr)),
-		DecisionWait:            10 * time.Second,
-		NumTraces:               100,
-		ExpectedNewTracesPerSec: 10,
-		PolicyCfgs: []PolicyCfg{
-			{
-				Name: "composite-policy-1",
-				Type: Composite,
-				CompositeCfg: CompositeCfg{
-					MaxTotalSpansPerSecond: 1000,
-					PolicyOrder:            []string{"test-composite-policy-1", "test-composite-policy-2", "test-composite-policy-3", "test-composite-policy-4", "test-composite-policy-5", "test-composite-policy-6", "test-composite-policy-7", "test-composite-policy-8"},
-					SubPolicyCfg: []SubPolicyCfg{
-						{
-							Name:                "test-composite-policy-1",
-							Type:                NumericAttribute,
-							NumericAttributeCfg: NumericAttributeCfg{Key: "key1", MinValue: 50, MaxValue: 100},
-						},
-						{
-							Name:               "test-composite-policy-2",
-							Type:               StringAttribute,
-							StringAttributeCfg: StringAttributeCfg{Key: "key2", Values: []string{"value1", "value2"}},
-						},
-						{
-							Name:            "test-composite-policy-3",
-							Type:            RateLimiting,
-							RateLimitingCfg: RateLimitingCfg{SpansPerSecond: 10},
-						},
-						{
-							Name:             "test-composite-policy-4",
-							Type:             Probabilistic,
-							ProbabilisticCfg: ProbabilisticCfg{HashSalt: "salt", SamplingPercentage: 10},
-						},
-						{
-							Name:       "test-composite-policy-5",
-							Type:       Latency,
-							LatencyCfg: LatencyCfg{ThresholdMs: 10},
-						},
-						{
-							Name:          "test-composite-policy-6",
-							Type:          StatusCode,
-							StatusCodeCfg: StatusCodeCfg{StatusCodes: []string{"200", "404"}},
-						},
-						{
-							Name: "test-composite-policy-7",
-							Type: AlwaysSample,
-						},
-						{
-							Name: "test-composite-policy-8",
-						},
+	t.Run("valid", func(t *testing.T) {
+		actual, err := getNewCompositePolicy(zap.NewNop(), &CompositeCfg{
+			MaxTotalSpansPerSecond: 1000,
+			PolicyOrder:            []string{"test-composite-policy-1"},
+			SubPolicyCfg: []CompositeSubPolicyCfg{
+				{
+					sharedPolicyCfg: sharedPolicyCfg{
+						Name:       "test-composite-policy-1",
+						Type:       Latency,
+						LatencyCfg: LatencyCfg{ThresholdMs: 100},
+					},
+				},
+				{
+					sharedPolicyCfg: sharedPolicyCfg{
+						Name:       "test-composite-policy-2",
+						Type:       Latency,
+						LatencyCfg: LatencyCfg{ThresholdMs: 200},
 					},
-					RateAllocation: []RateAllocationCfg{
-						{
-							Policy:  "test-composite-policy-1",
-							Percent: 50,
-						},
-						{
-							Policy:  "test-composite-policy-2",
-							Percent: 25,
-						},
+				},
+			},
+			RateAllocation: []RateAllocationCfg{
+				{
+					Policy:  "test-composite-policy-1",
+					Percent: 25,
+				},
+				{
+					Policy:  "test-composite-policy-2",
+					Percent: 0, // will be populated with default
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		expected := sampling.NewComposite(zap.NewNop(), 1000, []sampling.SubPolicyEvalParams{
+			{
+				Evaluator:         sampling.NewLatency(zap.NewNop(), 100),
+				MaxSpansPerSecond: 250,
+			},
+			{
+				Evaluator:         sampling.NewLatency(zap.NewNop(), 200),
+				MaxSpansPerSecond: 500,
+			},
+		}, sampling.MonotonicClock{})
+		assert.Equal(t, expected, actual)
+	})
+
+	t.Run("unsupported sampling policy type", func(t *testing.T) {
+		_, err := getNewCompositePolicy(zap.NewNop(), &CompositeCfg{
+			SubPolicyCfg: []CompositeSubPolicyCfg{
+				{
+					sharedPolicyCfg: sharedPolicyCfg{
+						Name: "test-composite-policy-1",
+						Type: Composite, // nested composite is not allowed
 					},
 				},
 			},
-		},
-	}
-	rlfCfg := cfg.PolicyCfgs[0].CompositeCfg
-	composite, e := getNewCompositePolicy(zap.NewNop(), rlfCfg)
-	require.NotNil(t, composite)
-	require.NotNil(t, cfg.ProcessorSettings)
-	require.Equal(t, 10*time.Second, cfg.DecisionWait)
-	require.Equal(t, uint64(100), cfg.NumTraces)
-	require.Equal(t, uint64(10), cfg.ExpectedNewTracesPerSec)
-	require.NoError(t, e)
-	// TBD add more assertions
+		})
+		require.EqualError(t, err, "unknown sampling policy type composite")
+	})
 }