[probabilisticsamplerprocessor] Support consistent intermediate span sampling (OTEP 226)

pkg/sampling/common.go

@@ -0,0 +1,100 @@
+package sampling
+
+import (
+	"errors"
+	"strings"
+
+	"go.uber.org/multierr"
+)
+
+type KV struct {
+	Key   string
+	Value string
+}
+
+var (
+	ErrTraceStateSize  = errors.New("invalid tracestate size")
+	ErrTraceStateCount = errors.New("invalid tracestate item count")
+)
+
+// keyValueScanner defines distinct scanner behaviors for lists of
+// key-values.
+type keyValueScanner struct {
+	// maxItems is 32 or -1
+	maxItems int
+	// trim is set if OWS (optional whitespace) should be removed
+	trim bool
+	// separator is , or ;
+	separator byte
+	// equality is = or :
+	equality byte
+}
+
+type commonTraceState struct {
+	kvs []KV
+}
+
+func (cts commonTraceState) HasExtraValues() bool {
+	return len(cts.kvs) != 0
+}
+
+func (cts commonTraceState) ExtraValues() []KV {
+	return cts.kvs
+}
+
+// trimOws removes optional whitespace on both ends of a string.
+func trimOws(input string) string {
+	// Hard-codes the value of owsCharset
+	for len(input) > 0 && input[0] == ' ' || input[0] == '\t' {
+		input = input[1:]
+	}
+	for len(input) > 0 && input[len(input)-1] == ' ' || input[len(input)-1] == '\t' {
+		input = input[:len(input)-1]
+	}
+	return input
+}
+
+func (s keyValueScanner) scanKeyValues(input string, f func(key, value string) error) error {
+	var rval error
+	items := 0
+	for input != "" {
+		items++
+		if s.maxItems > 0 && items >= s.maxItems {
+			// W3C specifies max 32 entries, tested here
+			// instead of via the regexp.
+			return ErrTraceStateCount
+		}
+
+		sep := strings.IndexByte(input, s.separator)
+
+		var member string
+		if sep < 0 {
+			member = input
+			input = ""
+		} else {
+			member = input[:sep]
+			input = input[sep+1:]
+		}
+
+		if s.trim {
+			// Trim only required for W3C; OTel does not
+			// specify whitespace for its value encoding.
+			member = trimOws(member)
+		}
+
+		if member == "" {
+			// W3C allows empty list members.
+			continue
+		}
+
+		eq := strings.IndexByte(member, s.equality)
+		if eq < 0 {
+			// A regexp should have rejected this input.
+			continue
+		}
+		if err := f(member[:eq], member[eq+1:]); err != nil {
+			rval = multierr.Append(rval, err)
+		}
+	}
+	return rval
+}

pkg/sampling/go.mod

@@ -0,0 +1,23 @@
+module github.com/open-telemetry/opentelemetry-collector-contrib/pkg/sampling
+
+go 1.20
+
+require (
+	github.com/stretchr/testify v1.8.2
+	go.opentelemetry.io/collector/pdata v1.0.0-rcv0011
+	go.uber.org/multierr v1.11.0
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	golang.org/x/net v0.9.0 // indirect
+	golang.org/x/sys v0.7.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
+	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
+	google.golang.org/grpc v1.54.0 // indirect
+	google.golang.org/protobuf v1.30.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

pkg/sampling/oteltracestate.go

@@ -0,0 +1,172 @@
+package sampling
+
+import (
+	"fmt"
+	"io"
+	"regexp"
+	"strconv"
+)
+
+type OTelTraceState struct {
+	commonTraceState
+
+	// sampling r, s, and t-values
+	ru uint64  // r value parsed, as unsigned
+	r  string  // 14 ASCII hex digits
+	sf float64 // s value parsed, as a probability
+	s  string  // original float syntax preserved
+	tf float64 // t value parsed, as a probability
+	t  string  // original float syntax preserved
+}
+
+const (
+	// hardMaxOTelLength is the maximum encoded size of an OTel
+	// tracestate value.
+	hardMaxOTelLength = 256
+
+	// chr        = ucalpha / lcalpha / DIGIT / "." / "_" / "-"
+	// ucalpha    = %x41-5A ; A-Z
+	// lcalpha    = %x61-7A ; a-z
+	// key        = lcalpha *(lcalpha / DIGIT )
+	// value      = *(chr)
+	// list-member = key ":" value
+	// list        = list-member *( ";" list-member )
+	otelKeyRegexp             = lcAlphaRegexp + lcDigitRegexp + `*`
+	otelValueRegexp           = `[a-zA-Z0-9._\-]*`
+	otelMemberRegexp          = `(?:` + otelKeyRegexp + `:` + otelValueRegexp + `)`
+	otelSemicolonMemberRegexp = `(?:` + `;` + otelMemberRegexp + `)`
+	otelTracestateRegexp      = `^` + otelMemberRegexp + otelSemicolonMemberRegexp + `*$`
+)
+
+var (
+	otelTracestateRe = regexp.MustCompile(otelTracestateRegexp)
+
+	ErrRandomValueRange = fmt.Errorf("r-value out of range")
+
+	otelSyntax = keyValueScanner{
+		maxItems:  -1,
+		trim:      false,
+		separator: ';',
+		equality:  ':',
+	}
+)
+
+func NewOTelTraceState(input string) (otts OTelTraceState, _ error) {
+	if len(input) > hardMaxOTelLength {
+		return otts, ErrTraceStateSize
+	}
+
+	if !otelTracestateRe.MatchString(input) {
+		return OTelTraceState{}, strconv.ErrSyntax
+	}
+
+	err := otelSyntax.scanKeyValues(input, func(key, value string) error {
+		var err error
+		switch key {
+		case "r":
+			var unsigned uint64
+			unsigned, err = strconv.ParseUint(value, 16, 64)
+			if err == nil {
+				if unsigned >= 0x1p56 {
+					err = ErrRandomValueRange
+				} else {
+					otts.r = value
+					otts.ru = unsigned
+				}
+			}
+		case "s":
+			var prob float64
+			prob, _, err = EncodedToProbabilityAndAdjustedCount(value)
+			if err == nil {
+				otts.s = value
+				otts.sf = prob
+			}
+		case "t":
+			var prob float64
+			prob, _, err = EncodedToProbabilityAndAdjustedCount(value)
+			if err == nil {
+				otts.t = value
+				otts.tf = prob
+			}
+		default:
+			otts.kvs = append(otts.kvs, KV{
+				Key:   key,
+				Value: value,
+			})
+		}
+		return err
+	})
+
+	return otts, err
+}
+
+func (otts OTelTraceState) HasRValue() bool {
+	return otts.r != ""
+}
+
+func (otts OTelTraceState) RValue() string {
+	return otts.r
+}
+
+func (otts OTelTraceState) RValueUnsigned() uint64 {
+	return otts.ru
+}
+
+func (otts OTelTraceState) HasSValue() bool {
+	return otts.s != ""
+}
+
+func (otts OTelTraceState) SValue() string {
+	return otts.s
+}
+
+func (otts OTelTraceState) SValueProbability() float64 {
+	return otts.sf
+}
+
+func (otts OTelTraceState) HasTValue() bool {
+	return otts.t != ""
+}
+
+func (otts OTelTraceState) TValue() string {
+	return otts.t
+}
+
+func (otts OTelTraceState) TValueProbability() float64 {
+	return otts.tf
+}
+
+func (otts OTelTraceState) HasAnyValue() bool {
+	return otts.HasRValue() || otts.HasSValue() || otts.HasTValue() || otts.HasExtraValues()
+}
+
+func (otts OTelTraceState) Serialize(w io.StringWriter) {
+	cnt := 0
+	sep := func() {
+		if cnt != 0 {
+			w.WriteString(";")
+		}
+		cnt++
+	}
+	if otts.HasRValue() {
+		sep()
+		w.WriteString("r:")
+		w.WriteString(otts.RValue())
+	}
+	if otts.HasSValue() {
+		sep()
+		w.WriteString("s:")
+		w.WriteString(otts.SValue())
+	}
+	if otts.HasTValue() {
+		sep()
+		w.WriteString("t:")
+		w.WriteString(otts.TValue())
+	}
+	for _, kv := range otts.ExtraValues() {
+		sep()
+		w.WriteString(kv.Key)
+		w.WriteString(":")
+		w.WriteString(kv.Value)
+	}
+}