pomerium/pkg/telemetry/trace/debug.go

package trace

import (
	"context"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"runtime"
	"slices"
	"strings"
	"sync"
	"time"

	"go.opentelemetry.io/otel/attribute"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	oteltrace "go.opentelemetry.io/otel/trace"
	coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
	"google.golang.org/grpc/metadata"
	"google.golang.org/protobuf/encoding/protojson"
)

type DebugFlags uint32

const (
	// If set, adds the "caller" attribute to each trace with the source location
	// where the trace was started.
	TrackSpanCallers = (1 << iota)

	// If set, keeps track of all span references and will attempt to wait for
	// all traces to complete when shutting down a trace context.
	// Use with caution, this will cause increasing memory usage over time.
	TrackSpanReferences = (1 << iota)

	// If set, keeps track of all observed spans, including span context and
	// all attributes.
	// Use with caution, this will cause significantly increasing memory usage
	// over time.
	TrackAllSpans = (1 << iota) | TrackSpanCallers

	// If set, will log all trace IDs and their span counts on close.
	//
	// Enables [TrackAllSpans]
	LogTraceIDs = (1 << iota) | TrackAllSpans

	// If set, will log all spans observed by the exporter on close. These spans
	// may belong to incomplete traces.
	//
	// Enables [TrackAllSpans]
	LogAllSpans = (1 << iota) | TrackAllSpans

	// If set, will log all exported spans when a warning is issued on close
	// (requires warning flags to also be set)
	//
	// Enables [TrackAllSpans]
	LogAllSpansOnWarn = (1 << iota) | TrackAllSpans

	// If set, will log all trace ID mappings when a warning is issued on close.
	// (requires warning flags to also be set)
	LogTraceIDsOnWarn = (1 << iota)

	// If set, will print a warning to stderr on close if there are any incomplete
	// traces (traces with no observed root spans)
	WarnOnIncompleteTraces = (1 << iota)

	// If set, will print a warning to stderr on close if there are any incomplete
	// spans (spans started, but not ended)
	WarnOnIncompleteSpans = (1 << iota)

	// If set, will print a warning to stderr on close if there are any spans
	// which reference unknown parent spans.
	//
	// Enables [TrackSpanReferences]
	WarnOnUnresolvedReferences = (1 << iota) | TrackSpanReferences

	// If set, configures Envoy to flush every span individually, disabling its
	// internal buffer.
	EnvoyFlushEverySpan = (1 << iota)
)

func (df DebugFlags) Check(flags DebugFlags) bool {
	return (df & flags) == flags
}

var (
	ErrIncompleteSpans    = errors.New("exporter shut down with incomplete spans")
	ErrMissingParentSpans = errors.New("exporter shut down with missing parent spans")
)

// WaitForSpans will block up to the given max duration and wait for all
// in-flight spans from tracers created with the given context to end. This
// function can be called more than once, and is safe to call from multiple
// goroutines in parallel.
//
// This requires the [TrackSpanReferences] debug flag to have been set with
// [Options.NewContext]. Otherwise, this function is a no-op and will return
// immediately.
//
// If this function blocks for more than 10 seconds, it will print a warning
// to stderr containing a list of span IDs it is waiting for, and the IDs of
// their parents (if known). Additionally, if the [TrackAllSpans] debug flag
// is set, details about parent spans will be displayed, including call site
// and trace ID.
func WaitForSpans(ctx context.Context, maxDuration time.Duration) error {
	if sys := systemContextFromContext(ctx); sys != nil && sys.observer != nil {
		done := make(chan struct{})
		go func() {
			defer close(done)
			sys.observer.wait(10 * time.Second)
		}()
		select {
		case <-done:
			return nil
		case <-time.After(maxDuration):
			return ErrMissingParentSpans
		}
	}
	return nil
}

func DebugFlagsFromContext(ctx context.Context) DebugFlags {
	if sys := systemContextFromContext(ctx); sys != nil {
		return sys.options.DebugFlags
	}
	return 0
}

type stackTraceProcessor struct{}

// ForceFlush implements trace.SpanProcessor.
func (s *stackTraceProcessor) ForceFlush(context.Context) error {
	return nil
}

// OnEnd implements trace.SpanProcessor.
func (*stackTraceProcessor) OnEnd(sdktrace.ReadOnlySpan) {
}

// OnStart implements trace.SpanProcessor.
func (*stackTraceProcessor) OnStart(_ context.Context, s sdktrace.ReadWriteSpan) {
	_, file, line, _ := runtime.Caller(2)
	s.SetAttributes(attribute.String("caller", fmt.Sprintf("%s:%d", file, line)))
}

// Shutdown implements trace.SpanProcessor.
func (s *stackTraceProcessor) Shutdown(context.Context) error {
	return nil
}

var debugMessageWriter io.Writer

func startMsg(title string) *strings.Builder {
	msg := &strings.Builder{}
	msg.WriteString("\n==================================================\n")
	msg.WriteString(title)
	return msg
}

func endMsg(msg *strings.Builder) {
	msg.WriteString("==================================================\n")
	w := debugMessageWriter
	if w == nil {
		w = os.Stderr
	}
	fmt.Fprint(w, msg.String())
}

type DebugEvent struct {
	Timestamp time.Time                             `json:"timestamp"`
	Request   *coltracepb.ExportTraceServiceRequest `json:"request"`
}

func (e DebugEvent) MarshalJSON() ([]byte, error) {
	type debugEvent struct {
		Timestamp time.Time       `json:"timestamp"`
		Request   json.RawMessage `json:"request"`
	}
	reqData, _ := protojson.Marshal(e.Request)
	return json.Marshal(debugEvent{
		Timestamp: e.Timestamp,
		Request:   reqData,
	})
}

func (e *DebugEvent) UnmarshalJSON(b []byte) error {
	type debugEvent struct {
		Timestamp time.Time       `json:"timestamp"`
		Request   json.RawMessage `json:"request"`
	}
	var ev debugEvent
	if err := json.Unmarshal(b, &ev); err != nil {
		return err
	}
	e.Timestamp = ev.Timestamp
	var msg coltracepb.ExportTraceServiceRequest
	if err := protojson.Unmarshal(ev.Request, &msg); err != nil {
		return err
	}
	e.Request = &msg
	return nil
}

const shardCount = 64

type (
	shardedSet   [shardCount]map[oteltrace.SpanID]struct{}
	shardedLocks [shardCount]sync.Mutex
)

func (s *shardedSet) Range(f func(key oteltrace.SpanID)) {
	for i := range shardCount {
		for k := range s[i] {
			f(k)
		}
	}
}

func (s *shardedLocks) LockAll() {
	for i := range shardCount {
		s[i].Lock()
	}
}

func (s *shardedLocks) UnlockAll() {
	for i := range shardCount {
		s[i].Unlock()
	}
}

type spanTracker struct {
	inflightSpansMu shardedLocks
	inflightSpans   shardedSet
	allSpans        sync.Map
	debugFlags      DebugFlags
	observer        *spanObserver
	shutdownOnce    sync.Once
}

func newSpanTracker(observer *spanObserver, debugFlags DebugFlags) *spanTracker {
	st := &spanTracker{
		observer:   observer,
		debugFlags: debugFlags,
	}
	for i := range len(st.inflightSpans) {
		st.inflightSpans[i] = make(map[oteltrace.SpanID]struct{})
	}
	return st
}

type spanInfo struct {
	Name        string
	SpanContext oteltrace.SpanContext
	Parent      oteltrace.SpanContext
	caller      string
	startTime   time.Time
}

// ForceFlush implements trace.SpanProcessor.
func (t *spanTracker) ForceFlush(context.Context) error {
	return nil
}

// OnEnd implements trace.SpanProcessor.
func (t *spanTracker) OnEnd(s sdktrace.ReadOnlySpan) {
	id := s.SpanContext().SpanID()
	bucket := binary.BigEndian.Uint64(id[:]) % shardCount
	t.inflightSpansMu[bucket].Lock()
	defer t.inflightSpansMu[bucket].Unlock()
	delete(t.inflightSpans[bucket], id)
}

// OnStart implements trace.SpanProcessor.
func (t *spanTracker) OnStart(_ context.Context, s sdktrace.ReadWriteSpan) {
	id := s.SpanContext().SpanID()
	bucket := binary.BigEndian.Uint64(id[:]) % shardCount
	t.inflightSpansMu[bucket].Lock()
	defer t.inflightSpansMu[bucket].Unlock()
	t.inflightSpans[bucket][id] = struct{}{}

	if t.debugFlags.Check(TrackSpanReferences) {
		if s.Parent().IsValid() {
			t.observer.ObserveReference(s.Parent().SpanID(), id)
		}
		t.observer.Observe(id)
	}

	if t.debugFlags.Check(TrackAllSpans) {
		var caller string
		for _, attr := range s.Attributes() {
			if attr.Key == "caller" {
				caller = attr.Value.AsString()
				break
			}
		}
		t.allSpans.Store(id, &spanInfo{
			Name:        s.Name(),
			SpanContext: s.SpanContext(),
			Parent:      s.Parent(),
			caller:      caller,
			startTime:   s.StartTime(),
		})
	}
}

// Shutdown implements trace.SpanProcessor.
func (t *spanTracker) Shutdown(_ context.Context) error {
	if t.debugFlags == 0 {
		return nil
	}
	didWarn := false
	t.shutdownOnce.Do(func() {
		if t.debugFlags.Check(WarnOnUnresolvedReferences) {
			var unknownParentIDs []string
			for id, via := range t.observer.referencedIDs {
				if via.IsValid() {
					if t.debugFlags.Check(TrackAllSpans) {
						if viaSpan, ok := t.allSpans.Load(via); ok {
							unknownParentIDs = append(unknownParentIDs, fmt.Sprintf("%s via %s (%s)", id, via, viaSpan.(*spanInfo).Name))
						} else {
							unknownParentIDs = append(unknownParentIDs, fmt.Sprintf("%s via %s", id, via))
						}
					}
				}
			}
			if len(unknownParentIDs) > 0 {
				didWarn = true
				msg := startMsg("WARNING: parent spans referenced but never seen:\n")
				for _, str := range unknownParentIDs {
					msg.WriteString(str)
					msg.WriteString("\n")
				}
				endMsg(msg)
			}
		}
		if t.debugFlags.Check(WarnOnIncompleteSpans) {
			if t.debugFlags.Check(TrackAllSpans) {
				incompleteSpans := []*spanInfo{}
				t.inflightSpansMu.LockAll()
				t.inflightSpans.Range(func(key oteltrace.SpanID) {
					if info, ok := t.allSpans.Load(key); ok {
						incompleteSpans = append(incompleteSpans, info.(*spanInfo))
					}
				})
				t.inflightSpansMu.UnlockAll()
				if len(incompleteSpans) > 0 {
					didWarn = true
					msg := startMsg("WARNING: spans not ended:\n")
					longestName := 0
					for _, span := range incompleteSpans {
						longestName = max(longestName, len(span.Name)+2)
					}
					for _, span := range incompleteSpans {
						var startedAt string
						if span.caller != "" {
							startedAt = " | started at: " + span.caller
						}
						fmt.Fprintf(msg, "%-*s (trace: %s | span: %s | parent: %s%s)\n", longestName, "'"+span.Name+"'",
							span.SpanContext.TraceID(), span.SpanContext.SpanID(), span.Parent.SpanID(), startedAt)
					}
					endMsg(msg)
				}
			} else {
				incompleteSpans := []oteltrace.SpanID{}
				t.inflightSpansMu.LockAll()
				t.inflightSpans.Range(func(key oteltrace.SpanID) {
					incompleteSpans = append(incompleteSpans, key)
				})
				t.inflightSpansMu.UnlockAll()
				if len(incompleteSpans) > 0 {
					didWarn = true
					msg := startMsg("WARNING: spans not ended:\n")
					for _, span := range incompleteSpans {
						fmt.Fprintf(msg, "%s\n", span)
					}
					msg.WriteString("Note: set TrackAllSpans flag for more info\n")
					endMsg(msg)
				}
			}
		}

		if t.debugFlags.Check(LogAllSpans) || (t.debugFlags.Check(LogAllSpansOnWarn) && didWarn) {
			allSpans := []*spanInfo{}
			t.allSpans.Range(func(_, value any) bool {
				allSpans = append(allSpans, value.(*spanInfo))
				return true
			})
			slices.SortFunc(allSpans, func(a, b *spanInfo) int {
				return a.startTime.Compare(b.startTime)
			})
			msg := startMsg("All observed spans:\n")
			longestName := 0
			for _, span := range allSpans {
				longestName = max(longestName, len(span.Name)+2)
			}
			for _, span := range allSpans {
				var startedAt string
				if span.caller != "" {
					startedAt = " | started at: " + span.caller
				}
				fmt.Fprintf(msg, "%-*s (trace: %s | span: %s | parent: %s%s)\n", longestName, "'"+span.Name+"'",
					span.SpanContext.TraceID(), span.SpanContext.SpanID(), span.Parent.SpanID(), startedAt)
			}
			endMsg(msg)
		}

		if t.debugFlags.Check(LogTraceIDs) || (didWarn && t.debugFlags.Check(LogTraceIDsOnWarn)) {
			msg := startMsg("Known trace ids:\n")
			traceIDs := map[oteltrace.TraceID]int{}
			t.allSpans.Range(func(_, value any) bool {
				v := value.(*spanInfo)
				traceIDs[v.SpanContext.TraceID()]++
				return true
			})
			for id, n := range traceIDs {
				fmt.Fprintf(msg, "%s (%d spans)\n", id.String(), n)
			}
			endMsg(msg)
		}
	})
	if didWarn {
		return ErrIncompleteSpans
	}
	return nil
}

func newSpanObserver() *spanObserver {
	return &spanObserver{
		referencedIDs: map[oteltrace.SpanID]oteltrace.SpanID{},
		cond:          sync.NewCond(&sync.Mutex{}),
	}
}

type spanObserver struct {
	cond          *sync.Cond
	referencedIDs map[oteltrace.SpanID]oteltrace.SpanID
	unobservedIDs int
}

func (obs *spanObserver) ObserveReference(id oteltrace.SpanID, via oteltrace.SpanID) {
	obs.cond.L.Lock()
	defer obs.cond.L.Unlock()
	if _, referenced := obs.referencedIDs[id]; !referenced {
		obs.referencedIDs[id] = via // referenced, but not observed
		// It is possible for new unobserved references to come in while waiting,
		// but incrementing the counter wouldn't satisfy the condition so we don't
		// need to signal the waiters
		obs.unobservedIDs++
	}
}

func (obs *spanObserver) Observe(id oteltrace.SpanID) {
	obs.cond.L.Lock()
	defer obs.cond.L.Unlock()
	if observed, referenced := obs.referencedIDs[id]; !referenced || observed.IsValid() { // NB: subtle condition
		obs.referencedIDs[id] = zeroSpanID
		if referenced {
			obs.unobservedIDs--
			obs.cond.Broadcast()
		}
	}
}

func (obs *spanObserver) wait(warnAfter time.Duration) {
	done := make(chan struct{})
	defer close(done)
	go func() {
		select {
		case <-done:
			return
		case <-time.After(warnAfter):
			obs.debugWarnWaiting()
		}
	}()

	obs.cond.L.Lock()
	for obs.unobservedIDs > 0 {
		obs.cond.Wait()
	}
	obs.cond.L.Unlock()
}

func (obs *spanObserver) debugWarnWaiting() {
	obs.cond.L.Lock()
	msg := startMsg(fmt.Sprintf("Waiting on %d unobserved spans:\n", obs.unobservedIDs))
	for id, via := range obs.referencedIDs {
		if via.IsValid() {
			fmt.Fprintf(msg, "%s via %s\n", id, via)
		}
	}
	endMsg(msg)
	obs.cond.L.Unlock()
}

func (srv *ExporterServer) observeExport(ctx context.Context, req *coltracepb.ExportTraceServiceRequest) {
	isLocal := len(metadata.ValueFromIncomingContext(ctx, localExporterMetadataKey)) != 0
	if isLocal {
		return
	}
	for _, res := range req.ResourceSpans {
		for _, scope := range res.ScopeSpans {
			for _, span := range scope.Spans {
				id, ok := ToSpanID(span.SpanId)
				if !ok {
					continue
				}
				srv.observer.Observe(id)
				for _, attr := range span.Attributes {
					if attr.Key != "pomerium.external-parent-span" {
						continue
					}
					if bytes, err := hex.DecodeString(attr.Value.GetStringValue()); err == nil {
						if id, ok := ToSpanID(bytes); ok {
							srv.observer.Observe(id)
						}
					}
					break
				}
			}
		}
	}
}