Add in-memory tail-sampling prototype (#372)

* Add in-memory tail-sampling prototype

This brings the tail-sampling prototype to the collector.

Overview of this prototype at https://omnition.io/docs/opencensus-tail-based-sampling-demo
Design at https://omnition.io/docs/opencensus-tail-based-sampling

Goals for the prototype:
* Allow early experimentation
* Collect information to guide future design decisions

Tail sampling is implemented as a SpanProcessor that holds traces in memory until the configured decision wait time is reached. Each trace is then evaluated against the configured sample policy and it is sampled or not. After the decision the trace still stays in memory for some time, so late arriving spans can be properly handled, either by discarding them, in case of not sampled traces, or forwarded to the proper exporters, if the trace was sampled.

Clean-up and more tests

Remove demo file

* PR Feedback 00

* Move tail sampling processor to its own package

Author: Paulo Janotti

cmd/occollector/app/builder/builder.go

@@ -30,12 +30,13 @@ const (
 	zipkinScribeEntry = "zipkin-scribe"
 
 	// flags
-	configCfg               = "config"
-	jaegerReceiverFlg       = "receive-jaeger"
-	ocReceiverFlg           = "receive-oc-trace"
-	zipkinReceiverFlg       = "receive-zipkin"
-	zipkinScribeReceiverFlg = "receive-zipkin-scribe"
-	debugProcessorFlg       = "debug-processor"
+	configCfg                   = "config"
+	jaegerReceiverFlg           = "receive-jaeger"
+	ocReceiverFlg               = "receive-oc-trace"
+	zipkinReceiverFlg           = "receive-zipkin"
+	zipkinScribeReceiverFlg     = "receive-zipkin-scribe"
+	debugProcessorFlg           = "debug-processor"
+	useTailSamplingAlwaysSample = "tail-sampling-always-sample"
 )
 
 // Flags adds flags related to basic building of the collector application to the given flagset.
@@ -50,6 +51,8 @@ func Flags(flags *flag.FlagSet) {
 	flags.Bool(zipkinScribeReceiverFlg, false,
 		fmt.Sprintf("Flag to run the Zipkin Scribe receiver, default settings: %+v", *NewDefaultZipkinScribeReceiverCfg()))
 	flags.Bool(debugProcessorFlg, false, "Flag to add a debug processor (combine with log level DEBUG to log incoming spans)")
+	flags.Bool(useTailSamplingAlwaysSample, false, "Flag to use a tail-based sampling processor with an always sample policy, "+
+		"unless tail sampling setting is present on configuration file.")
 }
 
 // GetConfigFile gets the config file from the config file flag.
@@ -62,6 +65,11 @@ func DebugProcessorEnabled(v *viper.Viper) bool {
 	return v.GetBool(debugProcessorFlg)
 }
 
+// DebugTailSamplingEnabled returns true if the debug processor is enabled, and false otherwise
+func DebugTailSamplingEnabled(v *viper.Viper) bool {
+	return v.GetBool(useTailSamplingAlwaysSample)
+}
+
 // JaegerReceiverCfg holds configuration for Jaeger receivers.
 type JaegerReceiverCfg struct {
 	// ThriftTChannelPort is the port that the relay receives on for jaeger thrift tchannel requests

cmd/occollector/app/builder/builder_test.go

@@ -15,7 +15,9 @@
 package builder
 
 import (
+	"encoding/json"
 	"reflect"
+	"sort"
 	"testing"
 	"time"
 
@@ -126,6 +128,88 @@ func TestMultiAndQueuedSpanProcessorConfig(t *testing.T) {
 	}
 }
 
+func TestTailSamplingPoliciesConfiguration(t *testing.T) {
+	v, err := loadViperFromFile("./testdata/sampling_config.yaml")
+	if err != nil {
+		t.Fatalf("Failed to load viper from test file: %v", err)
+	}
+
+	wCfg := NewDefaultSamplingCfg()
+	if wCfg.Mode != NoSampling {
+		t.Fatalf("Default SamplingCfg Mode should be NoSampling")
+	}
+	wCfg.Mode = TailSampling
+	wCfg.Policies = []*PolicyCfg{
+		{
+			Name:      "string-tag-filter1",
+			Type:      StringTagFilter,
+			Exporters: []string{"jaeger1"},
+			Configuration: &StringTagFilterCfg{
+				Tag:    "test",
+				Values: []string{"value 1", "value 2"},
+			},
+		},
+		{
+			Name:      "numeric-tag-filter2",
+			Type:      NumericTagFilter,
+			Exporters: []string{"jaeger2"},
+			Configuration: &NumericTagFilterCfg{
+				Tag:      "http.status_code",
+				MinValue: 400,
+				MaxValue: 999,
+			},
+		},
+		{
+			Name:      "string-tag-filter3",
+			Type:      StringTagFilter,
+			Exporters: []string{"jaeger3"},
+			Configuration: &StringTagFilterCfg{
+				Tag:    "test.different",
+				Values: []string{"key 1", "key 2"},
+			},
+		},
+		{
+			Name:      "numeric-tag-filter4",
+			Type:      NumericTagFilter,
+			Exporters: []string{"jaeger4", "jaeger5"},
+			Configuration: &NumericTagFilterCfg{
+				Tag:      "http.status_code",
+				MinValue: 400,
+				MaxValue: 999,
+			},
+		},
+	}
+
+	gCfg := NewDefaultSamplingCfg().InitFromViper(v)
+	sort.Slice(gCfg.Policies, func(i, j int) bool {
+		if len(gCfg.Policies[i].Exporters) == len(gCfg.Policies[j].Exporters) {
+			return gCfg.Policies[i].Exporters[0] < gCfg.Policies[j].Exporters[0]
+		}
+		return len(gCfg.Policies[i].Exporters) < len(gCfg.Policies[j].Exporters)
+	})
+
+	if !reflect.DeepEqual(gCfg, wCfg) {
+		gb, _ := json.MarshalIndent(gCfg, "", " ")
+		t.Fatalf("Wanted %+v but got %+v\ngot json:\n%s", *wCfg, *gCfg, string(gb))
+	}
+}
+
+func TestTailSamplingConfig(t *testing.T) {
+	v, err := loadViperFromFile("./testdata/sampling_config.yaml")
+	if err != nil {
+		t.Fatalf("Failed to load viper from test file: %v", err)
+	}
+
+	wCfg := NewDefaultTailBasedCfg()
+	wCfg.DecisionWait = 31 * time.Second
+	wCfg.NumTraces = 20001
+
+	gCfg := NewDefaultTailBasedCfg().InitFromViper(v)
+	if !reflect.DeepEqual(gCfg, wCfg) {
+		t.Fatalf("Wanted %+v but got %+v", *wCfg, *gCfg)
+	}
+}
+
 func loadViperFromFile(file string) (*viper.Viper, error) {
 	v := viper.New()
 	v.SetConfigFile(file)

cmd/occollector/app/builder/sampling_builder.go

@@ -0,0 +1,187 @@
+// Copyright 2019, OpenCensus Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package builder
+
+import (
+	"time"
+
+	"github.com/spf13/viper"
+)
+
+const (
+	modeTag     = "mode"
+	policiesTag = "policies"
+	samplingTag = "sampling"
+)
+
+// Mode indicates the sampling mode
+type Mode string
+
+const (
+	// NoSampling mode is the default and means that all data arriving at the collector
+	// is passed ahead.
+	NoSampling Mode = "no-sampling"
+	// TailSampling is the mode in which trace data is temporarily retained until an evaluation
+	// if the trace should be sampled is performed.
+	TailSampling Mode = "tail"
+)
+
+// PolicyType indicates the type of sampling policy.
+type PolicyType string
+
+const (
+	// AlwaysSample samples all traces, typically used for debugging.
+	AlwaysSample PolicyType = "always-sample"
+	// NumericTagFilter sample traces that have a given numberic tag in a specified
+	// range, e.g.: tag "http.status_code" >= 399 and <= 999.
+	NumericTagFilter PolicyType = "numeric-tag-filter"
+	// StringTagFilter sample traces that a tag, of type string, matching
+	// one of the listed values.
+	StringTagFilter PolicyType = "string-tag-filter"
+	// RateLimiting allows all traces until the specified limits are satisfied.
+	RateLimiting PolicyType = "rate-limiting"
+)
+
+// 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
+	// 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{}
+}
+
+// NumericTagFilterCfg holds the configurable settings to create a numeric tag filter
+// sampling policy evaluator.
+type NumericTagFilterCfg struct {
+	// Tag that the filter is going to be matching against.
+	Tag string `mapstructure:"tag"`
+	// MinValue is the minimum value of the tag to be considered a match.
+	MinValue int64 `mapstructure:"min-value"`
+	// MaxValue is the maximum value of the tag to be considered a match.
+	MaxValue int64 `mapstructure:"max-value"`
+}
+
+// StringTagFilterCfg holds the configurable settings to create a string tag filter
+// sampling policy evaluator.
+type StringTagFilterCfg struct {
+	// Tag that the filter is going to be matching against.
+	Tag string `mapstructure:"tag"`
+	// Values is the set of tag values that if any is equal to the actual tag valueto be considered a match.
+	Values []string `mapstructure:"values"`
+}
+
+// RateLimitingCfg holds the configurable settings to create a string tag filter
+// sampling policy evaluator.
+type RateLimitingCfg struct {
+	// SpansPerSecond limit to the number of spans per second
+	SpansPerSecond int64 `mapstructure:"spans-per-second"`
+}
+
+// SamplingCfg holds the sampling configuration.
+type SamplingCfg struct {
+	// Mode specifies the sampling mode to be used.
+	Mode Mode `mapstructure:"mode"`
+	// Policies contains the list of policies to be used by sampling.
+	Policies []*PolicyCfg `mapstructure:"policies"`
+}
+
+// NewDefaultSamplingCfg creates a SamplingCfg with the default values.
+func NewDefaultSamplingCfg() *SamplingCfg {
+	return &SamplingCfg{
+		Mode: NoSampling,
+	}
+}
+
+// InitFromViper initializes SamplingCfg with properties from viper.
+func (sCfg *SamplingCfg) InitFromViper(v *viper.Viper) *SamplingCfg {
+	sv := v.Sub(samplingTag)
+	if sv == nil {
+		return sCfg
+	}
+
+	sCfg.Mode = Mode(sv.GetString(modeTag))
+
+	pv := sv.Sub(policiesTag)
+	if pv == nil {
+		return sCfg
+	}
+
+	for policyName := range sv.GetStringMap(policiesTag) {
+		polSub := pv.Sub(policyName)
+		polCfg := &PolicyCfg{}
+		polCfg.Name = policyName
+		polCfg.Type = PolicyType(polSub.GetString("policy"))
+		polCfg.Exporters = polSub.GetStringSlice("exporters")
+
+		cfgSub := polSub.Sub("configuration")
+		if cfgSub != nil {
+			// As the number of polices grow this likely should be in a map.
+			var cfg interface{}
+			switch polCfg.Type {
+			case NumericTagFilter:
+				numTagFilterCfg := &NumericTagFilterCfg{}
+				cfg = numTagFilterCfg
+			case StringTagFilter:
+				strTagFilterCfg := &StringTagFilterCfg{}
+				cfg = strTagFilterCfg
+			case RateLimiting:
+				rateLimitingCfg := &RateLimitingCfg{}
+				cfg = rateLimitingCfg
+			}
+			cfgSub.Unmarshal(cfg)
+			polCfg.Configuration = cfg
+		}
+
+		sCfg.Policies = append(sCfg.Policies, polCfg)
+	}
+	return sCfg
+}
+
+// TailBasedCfg holds the configuration for tail-based sampling.
+type TailBasedCfg struct {
+	// DecisionWait is the desired wait time from the arrival of the first span of
+	// trace until the decision about sampling it or not is evaluated.
+	DecisionWait time.Duration `mapstructure:"decision-wait"`
+	// NumTraces is the number of traces kept on memory. Typically most of the data
+	// of a trace is released after a sampling decision is taken.
+	NumTraces uint64 `mapstructure:"num-traces"`
+}
+
+// NewDefaultTailBasedCfg creates a TailBasedCfg with the default values.
+func NewDefaultTailBasedCfg() *TailBasedCfg {
+	return &TailBasedCfg{
+		DecisionWait: 30 * time.Second,
+		NumTraces:    50000,
+	}
+}
+
+// InitFromViper initializes TailBasedCfg with properties from viper.
+func (tCfg *TailBasedCfg) InitFromViper(v *viper.Viper) *TailBasedCfg {
+	tv := v.Sub(samplingTag)
+	if tv == nil {
+		return tCfg
+	}
+	if tv == nil || tv.GetString(modeTag) != string(TailSampling) {
+		return tCfg
+	}
+
+	tv.Unmarshal(tCfg)
+	return tCfg
+}

cmd/occollector/app/builder/testdata/queued_exporters.yaml

@@ -1,4 +1,4 @@
- queued-exporters:
+queued-exporters:
   proc-tchannel:
     num-workers: 13
     queue-size: 1300