Finish up configs for tail sampling policies

Fixes open-telemetry#146.

Note one additional change is now each tail sampling processor can only
accept one policy, because one processor can only have one next
consumer. If users need more than one policies they can define multiple
tail sampling processors in the config.

internal/collector/sampling/policy.go

@@ -26,8 +26,8 @@ import (
 // TraceData stores the sampling related trace data.
 type TraceData struct {
 	sync.Mutex
-	// Decision gives the current status of the sampling decision for each policy.
-	Decision []Decision
+	// Decision gives the current status of the sampling decision.
+	Decision Decision
 	// Arrival time the first span for the trace was received.
 	ArrivalTime time.Time
 	// Decisiontime time when sampling decision was taken.

processor/tailsampling/config.go

@@ -32,21 +32,22 @@ const (
 	// StringAttributeFilter sample traces that a attribute, of type string, matching
 	// one of the listed values.
 	StringAttributeFilter PolicyType = "string-attribute-filter"
-	// RateLimiting allows all traces until the specified limits are satisfied.
-	RateLimiting PolicyType = "rate-limiting"
+	// RateLimitingFilter allows all traces until the specified limits are satisfied.
+	RateLimitingFilter PolicyType = "rate-limiting-filter"
 )
 
 // PolicyCfg holds the common configuration to all policies.
 type PolicyCfg struct {
 	// Name given to the instance of the policy to make easy to identify it in metrics and logs.
-	Name string
+	Name string `mapstructure:"name"`
 	// Type of the policy this will be used to match the proper configuration of the policy.
-	Type PolicyType
-	// Exporters hold the name of the exporters that the policy evaluator uses to make decisions
-	// about whether or not sending the traces.
-	Exporters []string
-	// Configuration holds the settings specific to the policy.
-	Configuration interface{}
+	Type PolicyType `mapstructure:"type"`
+	// Configs for numeric attribute filter sampling policy evaluator.
+	NumericAttributeFilterCfg NumericAttributeFilterCfg `mapstructure:"numeric-attribute-filter"`
+	// Configs for string attribute filter sampling policy evaluator.
+	StringAttributeFilterCfg StringAttributeFilterCfg `mapstructure:"string-attribute-filter"`
+	// Configs for rate limiting filter sampling policy evaluator.
+	RateLimitingFilterCfg RateLimitingFilterCfg `mapstructure:"rate-limiting-filter"`
 }
 
 // NumericAttributeFilterCfg holds the configurable settings to create a numeric attribute filter
@@ -69,6 +70,13 @@ type StringAttributeFilterCfg struct {
 	Values []string `mapstructure:"values"`
 }
 
+// RateLimitingFilterCfg holds the configurable settings to create a rate limiting
+// sampling policy evaluator.
+type RateLimitingFilterCfg struct {
+	// SpansPerSecond sets the limit on the maximum nuber of spans that can be processed each second.
+	SpansPerSecond int64 `mapstructure:"spans-per-second"`
+}
+
 // Config holds the configuration for tail-based sampling.
 type Config struct {
 	configmodels.ProcessorSettings `mapstructure:",squash"`
@@ -81,4 +89,7 @@ type Config struct {
 	// ExpectedNewTracesPerSec sets the expected number of new traces sending to the tail sampling processor
 	// per second. This helps with allocating data structures with closer to actual usage size.
 	ExpectedNewTracesPerSec uint64 `mapstructure:"expected-new-traces-per-sec"`
+	// Policy sets the tail-based sampling policy which makes a sampling decision
+	// for a given trace when requested.
+	PolicyCfg PolicyCfg `mapstructure:"policy"`
 }

processor/tailsampling/config_test.go

@@ -36,19 +36,69 @@ func TestLoadConfig(t *testing.T) {
 	cfg, err := config.LoadConfigFile(
 		t, path.Join(".", "testdata", "tail_sampling_config.yaml"), receivers, processors, exporters,
 	)
-
 	require.Nil(t, err)
 	require.NotNil(t, cfg)
 
-	p0 := cfg.Processors["tail-sampling"]
-	assert.Equal(t, p0,
+	assert.Equal(t, cfg.Processors["tail-sampling"],
 		&Config{
 			ProcessorSettings: configmodels.ProcessorSettings{
 				TypeVal: "tail-sampling",
 				NameVal: "tail-sampling",
 			},
-			DecisionWait:            31 * time.Second,
-			NumTraces:               20001,
-			ExpectedNewTracesPerSec: 100,
+			DecisionWait:            10 * time.Second,
+			NumTraces:               100,
+			ExpectedNewTracesPerSec: 10,
+			PolicyCfg: PolicyCfg{
+				Name: "test-policy-1",
+				Type: AlwaysSample,
+			},
+		})
+
+	assert.Equal(t, cfg.Processors["tail-sampling/2"],
+		&Config{
+			ProcessorSettings: configmodels.ProcessorSettings{
+				TypeVal: "tail-sampling",
+				NameVal: "tail-sampling/2",
+			},
+			DecisionWait:            20 * time.Second,
+			NumTraces:               200,
+			ExpectedNewTracesPerSec: 20,
+			PolicyCfg: PolicyCfg{
+				Name:                      "test-policy-2",
+				Type:                      NumericAttributeFilter,
+				NumericAttributeFilterCfg: NumericAttributeFilterCfg{Key: "key1", MinValue: 50, MaxValue: 100},
+			},
+		})
+
+	assert.Equal(t, cfg.Processors["tail-sampling/3"],
+		&Config{
+			ProcessorSettings: configmodels.ProcessorSettings{
+				TypeVal: "tail-sampling",
+				NameVal: "tail-sampling/3",
+			},
+			DecisionWait:            30 * time.Second,
+			NumTraces:               300,
+			ExpectedNewTracesPerSec: 30,
+			PolicyCfg: PolicyCfg{
+				Name:                     "test-policy-3",
+				Type:                     StringAttributeFilter,
+				StringAttributeFilterCfg: StringAttributeFilterCfg{Key: "key2", Values: []string{"value1", "value2"}},
+			},
+		})
+
+	assert.Equal(t, cfg.Processors["tail-sampling/4"],
+		&Config{
+			ProcessorSettings: configmodels.ProcessorSettings{
+				TypeVal: "tail-sampling",
+				NameVal: "tail-sampling/4",
+			},
+			DecisionWait:            40 * time.Second,
+			NumTraces:               400,
+			ExpectedNewTracesPerSec: 40,
+			PolicyCfg: PolicyCfg{
+				Name:                  "test-policy-4",
+				Type:                  RateLimitingFilter,
+				RateLimitingFilterCfg: RateLimitingFilterCfg{SpansPerSecond: 35},
+			},
 		})
 }

processor/tailsampling/factory_test.go

@@ -33,7 +33,13 @@ func TestCreateDefaultConfig(t *testing.T) {
 func TestCreateProcessor(t *testing.T) {
 	factory := &Factory{}
 
-	cfg := factory.CreateDefaultConfig()
+	cfg := factory.CreateDefaultConfig().(*Config)
+	// Manually set required fields
+	cfg.ExpectedNewTracesPerSec = 64
+	cfg.PolicyCfg = PolicyCfg{
+		Name: "test-policy",
+		Type: AlwaysSample,
+	}
 
 	tp, err := factory.CreateTraceProcessor(zap.NewNop(), exportertest.NewNopTraceExporter(), cfg)
 	assert.NotNil(t, tp)