LimitedSampler

_README.md

Limited Sampler

flow.jpg

main.go

package main

import (
	"go.opentelemetry.io/otel"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

func main() {
	if sampler, err = NewLimitedSampler(0.01, 0.27, 5); err != nil {
		return errors.Wrap(err, "Failed to create limited sampler")
	}

	otel.SetTracerProvider(sdktrace.NewTracerProvider(
		sdktrace.WithSampler(sdktrace.ParentBased(sampler, sdktrace.WithRemoteParentSampled(sampler))),
	))
}

sampler.go

// Source: https://github.com/googleapis/google-cloud-go/blob/v0.37.2/trace/sampling.go#L102
package main

import (
	"encoding/binary"
	"fmt"
	"sync"
	"time"

	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/trace"
	"golang.org/x/time/rate"
)

type sampler struct {
	traceIDUpperBound uint64
	skipped           float64
	*rate.Limiter
	sync.Mutex
}

var _ sdktrace.Sampler = &sampler{}

// NewLimitedSampler returns a sampling policy that randomly samples a given
// fraction of requests.  It also enforces a limit on the number of traces per
// second.  It tries to trace every request with a trace header, but will not
// exceed the maxqps and burst value to do it.
// By setting a burst you can temporarily exceed the maxqps.
// We should try to aim for around 1000 traces per hour to optimally fill
// the Cloud Trace UI. Any more traces will be invisible until zooming in.
// So we can set the maxqps to 0.277 (997 traces per hour) and a higher burst.
// The burst is particularly important when receiving sampled incoming requests
// for distributed tracing.
func NewLimitedSampler(fraction, maxqps float64, burst int) (*sampler, error) {
	if !(fraction >= 0) {
		return nil, fmt.Errorf("invalid fraction %f", fraction)
	}
	if !(maxqps >= 0) {
		return nil, fmt.Errorf("invalid maxqps %f", maxqps)
	}
	maxTokens := burst
	if maxTokens == 0 {
		// Set a limit on the number of accumulated "tokens", to limit bursts of
		// traced requests.  Use one more than a second's worth of tokens, or 100,
		// whichever is smaller.
		// See https://godoc.org/golang.org/x/time/rate#NewLimiter.
		maxTokens = 100
		if maxqps < 99.0 {
			maxTokens = 1 + int(maxqps)
		}
	}
	s := sampler{
		traceIDUpperBound: uint64(fraction * (1 << 63)),
		Limiter:           rate.NewLimiter(rate.Limit(maxqps), maxTokens),
	}
	return &s, nil
}

// ShouldSample makes the sampling decision
func (s *sampler) ShouldSample(p sdktrace.SamplingParameters) sdktrace.SamplingResult {
	// If parent is traced, trace child span
	psc := trace.SpanContextFromContext(p.ParentContext)
	if psc.IsSampled() {
		return sdktrace.SamplingResult{
			Decision:   sdktrace.RecordAndSample,
			Tracestate: psc.TraceState(),
		}
	}

	s.Lock()
	x := binary.BigEndian.Uint64(p.TraceID[0:8]) >> 1
	d := s.sample(p, time.Now(), x)
	s.Unlock()
	return d
}

// Description returns information describing the Sampler.
func (s *sampler) Description() string {
	return ""
}

// sample contains the a deterministic, time-independent logic of Sample.
func (s *sampler) sample(p sdktrace.SamplingParameters, now time.Time, x uint64) (d sdktrace.SamplingResult) {
	psc := trace.SpanContextFromContext(p.ParentContext)
	d.Tracestate = psc.TraceState()

	if x < uint64(s.traceIDUpperBound) {
		d.Decision = sdktrace.RecordAndSample
	} else {
		d.Decision = sdktrace.Drop
	}

	// Always trace if remote is sampled
	if psc.IsRemote() && psc.IsSampled() {
		d.Decision = sdktrace.RecordAndSample
	}

	if d.Decision == sdktrace.Drop {
		// We have no reason to trace this request.
		return
	}

	// We test separately that the rate limit is not tiny before calling AllowN,
	// because of overflow problems in x/time/rate.
	if s.Limit() < 1e-9 || !s.AllowN(now, 1) {
		// Rejected by the rate limit.
		if d.Decision == sdktrace.RecordAndSample {
			s.skipped++
		}
		d.Decision = sdktrace.Drop
		return
	}
	if d.Decision == sdktrace.RecordAndSample {
		s.skipped = 0.0
	}
	return
}