ddtrace/tracer: create debug mode for old, open spans

ddtrace/tracer/abandoned_spans.go

@@ -0,0 +1,139 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package tracer
+
+import (
+	"container/list"
+	"fmt"
+	"strings"
+	"time"
+
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
+)
+
+var tickerInterval = time.Minute
+
+// reportAbandonedSpans periodically finds and reports potentially abandoned
+// spans that are older than the given interval. These spans are stored in a
+// bucketed linked list, sorted by their `Start` time.
+func (t *tracer) reportAbandonedSpans(interval time.Duration) {
+	tick := time.NewTicker(tickerInterval)
+	defer tick.Stop()
+	for {
+		select {
+		case <-tick.C:
+			Print(t.abandonedSpans, true, interval)
+		case s := <-t.cIn:
+			bNode := t.abandonedSpans.Front()
+			if bNode == nil {
+				b := list.New()
+				b.PushBack(s)
+				t.abandonedSpans.PushBack(b)
+				break
+			}
+			for bNode != nil {
+				bucket, bOk := bNode.Value.(*list.List)
+				if !bOk {
+					bNode = bNode.Next()
+					continue
+				}
+				sNode := bucket.Front()
+				if sNode == nil {
+					bNode = bNode.Next()
+					continue
+				}
+				sp, sOk := sNode.Value.(*span)
+				if sOk && sp != nil && s.Start-sp.Start <= interval.Nanoseconds() {
+					bucket.PushBack(s)
+					break
+				}
+				bNode = bNode.Next()
+			}
+			if bNode != nil {
+				break
+			}
+			b := list.New()
+			b.PushBack(s)
+			t.abandonedSpans.PushBack(b)
+		case s := <-t.cOut:
+			for node := t.abandonedSpans.Front(); node != nil; node = node.Next() {
+				bucket, ok := node.Value.(*list.List)
+				if !ok || bucket.Front() == nil {
+					continue
+				}
+				spNode := bucket.Front()
+				sp, ok := spNode.Value.(*span)
+				if !ok {
+					continue
+				}
+				if s.Start-sp.Start <= interval.Nanoseconds() {
+					bucket.Remove(spNode)
+					if bucket.Front() == nil {
+						t.abandonedSpans.Remove(node)
+					}
+					break
+				}
+			}
+		case <-t.stop:
+			Print(t.abandonedSpans, false, interval)
+			return
+		}
+	}
+}
+
+// Print returns a string containing potentially abandoned spans. If `filter` is true,
+// it will only return spans that are older than the provided time `interval`. If false,
+// it will return all unfinished spans.
+func Print(l *list.List, filter bool, interval time.Duration) {
+	var sb strings.Builder
+	nowTime := now()
+	spanCount := 0
+
+	for bucketNode := l.Front(); bucketNode != nil; bucketNode = bucketNode.Next() {
+		bucket, ok := bucketNode.Value.(*list.List)
+		if !ok || bucket == nil {
+			continue
+		}
+
+		// since spans are bucketed by time, finding a bucket that is newer
+		// than the allowed time interval means that all spans in this bucket
+		// and future buckets will be younger than `interval`, and thus aren't
+		// worth checking.
+		if filter {
+			spanNode := bucket.Front()
+			if spanNode == nil {
+				continue
+			}
+			sp, ok := spanNode.Value.(*span)
+			if !ok || sp == nil {
+				continue
+			}
+			if nowTime-sp.Start < interval.Nanoseconds() {
+				continue
+			}
+		}
+		for spanNode := bucket.Front(); spanNode != nil; spanNode = spanNode.Next() {
+			sp, ok := spanNode.Value.(*span)
+			if !ok || sp == nil {
+				continue
+			}
+
+			// despite quitting early, spans within the same bucket can still fall on either side
+			// of the timeout. We should still check if the span is too old or not.
+			if filter && nowTime-sp.Start < interval.Nanoseconds() {
+				break
+			}
+			sb.WriteString(fmt.Sprintf("[name: %s, span_id: %d, trace_id: %d, age: %d],", sp.Name, sp.SpanID, sp.TraceID, sp.Duration))
+			spanCount += 1
+		}
+	}
+
+	if spanCount == 0 {
+		return
+	}
+	log.Warn("%d abandoned spans:", spanCount)
+	log.Warn(sb.String())
+}

ddtrace/tracer/abandoned_spans_test.go

@@ -0,0 +1,161 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package tracer
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/telemetry"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/version"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var warnPrefix = fmt.Sprintf("Datadog Tracer %v WARN: ", version.Tag)
+
+func TestReportAbandonedSpans(t *testing.T) {
+	assert := assert.New(t)
+	tp := new(log.RecordLogger)
+
+	tickerInterval = 100 * time.Millisecond
+
+	tp.Ignore("appsec: ", telemetry.LogPrefix)
+
+	t.Run("on", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(500*time.Millisecond))
+		defer stop()
+		assert.True(tracer.config.debugAbandonedSpans)
+		assert.Equal(tracer.config.spanTimeout, 500*time.Millisecond)
+	})
+
+	t.Run("finished", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(500*time.Millisecond))
+		defer stop()
+		s := tracer.StartSpan("operation").(*span)
+		s.Finish()
+		expected := fmt.Sprintf("%s[name: %s, span_id: %d, trace_id: %d, age: %d],", warnPrefix, s.Name, s.SpanID, s.TraceID, s.Duration)
+		assert.NotContains(tp.Logs(), expected)
+	})
+
+	t.Run("open", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(500*time.Millisecond))
+		defer stop()
+		s := tracer.StartSpan("operation").(*span)
+		time.Sleep(time.Second)
+		expected := fmt.Sprintf("%s[name: %s, span_id: %d, trace_id: %d, age: %d],", warnPrefix, s.Name, s.SpanID, s.TraceID, s.Duration)
+		assert.Contains(tp.Logs(), fmt.Sprintf("%s%d abandoned spans:", warnPrefix, 1))
+		assert.Contains(tp.Logs(), expected)
+		s.Finish()
+	})
+
+	t.Run("both", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(500*time.Millisecond))
+		defer stop()
+		sf := tracer.StartSpan("op").(*span)
+		sf.Finish()
+		s := tracer.StartSpan("op2").(*span)
+		time.Sleep(time.Second)
+		notExpected := fmt.Sprintf("%s[name: %s, span_id: %d, trace_id: %d, age: %d],[name: %s, span_id: %d, trace_id: %d, age: %d],", warnPrefix, sf.Name, sf.SpanID, sf.TraceID, sf.Duration, s.Name, s.SpanID, s.TraceID, s.Duration)
+		expected := fmt.Sprintf("%s[name: %s, span_id: %d, trace_id: %d, age: %d],", warnPrefix, s.Name, s.SpanID, s.TraceID, s.Duration)
+		assert.Contains(tp.Logs(), fmt.Sprintf("%s%d abandoned spans:", warnPrefix, 1))
+		assert.NotContains(tp.Logs(), notExpected)
+		assert.Contains(tp.Logs(), expected)
+		s.Finish()
+	})
+
+	t.Run("many", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(500*time.Millisecond))
+		defer stop()
+		var sb strings.Builder
+		sb.WriteString(warnPrefix)
+		for i := 0; i < 10; i++ {
+			s := tracer.StartSpan(fmt.Sprintf("operation%d", i)).(*span)
+			if i%2 == 0 {
+				s.Finish()
+			} else {
+				e := fmt.Sprintf("[name: %s, span_id: %d, trace_id: %d, age: %d],", s.Name, s.SpanID, s.TraceID, s.Duration)
+				sb.WriteString(e)
+				time.Sleep(500 * time.Millisecond)
+			}
+		}
+		time.Sleep(time.Second)
+		assert.Contains(tp.Logs(), sb.String())
+	})
+
+	t.Run("many buckets", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(500*time.Millisecond))
+		defer stop()
+		var sb strings.Builder
+		sb.WriteString(warnPrefix)
+
+		for i := 0; i < 5; i++ {
+			s := tracer.StartSpan(fmt.Sprintf("operation%d", i))
+			s.Finish()
+			time.Sleep(150 * time.Millisecond)
+		}
+		for i := 0; i < 5; i++ {
+			s := tracer.StartSpan(fmt.Sprintf("operation2-%d", i)).(*span)
+			sb.WriteString(fmt.Sprintf("[name: %s, span_id: %d, trace_id: %d, age: %d],", s.Name, s.SpanID, s.TraceID, s.Duration))
+			time.Sleep(150 * time.Millisecond)
+		}
+		time.Sleep(time.Second)
+
+		assert.Contains(tp.Logs(), fmt.Sprintf("%s%d abandoned spans:", warnPrefix, 5))
+		assert.Contains(tp.Logs(), sb.String())
+	})
+
+	t.Run("stop", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(100*time.Millisecond))
+		var sb strings.Builder
+		sb.WriteString(warnPrefix)
+
+		for i := 0; i < 5; i++ {
+			s := tracer.StartSpan(fmt.Sprintf("operation%d", i)).(*span)
+			sb.WriteString(fmt.Sprintf("[name: %s, span_id: %d, trace_id: %d, age: %d],", s.Name, s.SpanID, s.TraceID, s.Duration))
+		}
+		stop()
+		time.Sleep(time.Second)
+		assert.Contains(tp.Logs(), fmt.Sprintf("%s%d abandoned spans:", warnPrefix, 5))
+		assert.Contains(tp.Logs(), sb.String())
+	})
+
+	t.Run("wait", func(t *testing.T) {
+		tracer, _, _, stop := startTestTracer(t, WithLogger(tp), WithDebugSpansMode(500*time.Millisecond))
+		defer stop()
+
+		s := tracer.StartSpan("operation").(*span)
+		expected := fmt.Sprintf("%s[name: %s, span_id: %d, trace_id: %d, age: %d],", warnPrefix, s.Name, s.SpanID, s.TraceID, s.Duration)
+
+		assert.NotContains(tp.Logs(), expected)
+		time.Sleep(time.Second)
+		assert.Contains(tp.Logs(), expected)
+		s.Finish()
+	})
+}
+
+func TestDebugAbandonedSpansOff(t *testing.T) {
+	assert := assert.New(t)
+	tp := new(log.RecordLogger)
+	tracer, _, _, stop := startTestTracer(t, WithLogger(tp))
+	defer stop()
+
+	tp.Reset()
+	tp.Ignore("appsec: ", telemetry.LogPrefix)
+
+	t.Run("default", func(t *testing.T) {
+		assert.False(tracer.config.debugAbandonedSpans)
+		assert.Equal(time.Duration(0), tracer.config.spanTimeout)
+		s := tracer.StartSpan("operation")
+		time.Sleep(time.Second)
+		expected := fmt.Sprintf("%s Trace %v waiting on span %v", warnPrefix, s.Context().TraceID(), s.Context().SpanID())
+		assert.NotContains(tp.Logs(), expected)
+		s.Finish()
+	})
+}

ddtrace/tracer/option.go

@@ -227,6 +227,13 @@ type config struct {
 	// peerServiceMappings holds a set of service mappings to dynamically rename peer.service values.
 	peerServiceMappings map[string]string
 
+	// debugAbandonedSpans controls if the tracer should log when old, open spans are found
+	debugAbandonedSpans bool
+
+	// spanTimeout represents how old a span can be before it should be logged as a possible
+	// misconfiguration
+	spanTimeout time.Duration
+
 	// partialFlushMinSpans is the number of finished spans in a single trace to trigger a
 	// partial flush, or 0 if partial flushing is disabled.
 	// Value from DD_TRACE_PARTIAL_FLUSH_MIN_SPANS, default 1000.
@@ -314,6 +321,10 @@ func newConfig(opts ...StartOption) *config {
 	c.profilerEndpoints = internal.BoolEnv(traceprof.EndpointEnvVar, true)
 	c.profilerHotspots = internal.BoolEnv(traceprof.CodeHotspotsEnvVar, true)
 	c.enableHostnameDetection = internal.BoolEnv("DD_CLIENT_HOSTNAME_ENABLED", true)
+	c.debugAbandonedSpans = internal.BoolEnv("DD_TRACE_DEBUG_ABANDONED_SPANS", false)
+	if c.debugAbandonedSpans {
+		c.spanTimeout = internal.DurationEnv("DD_TRACE_ABANDONED_SPAN_TIMEOUT", 10*time.Minute)
+	}
 	c.statsComputationEnabled = internal.BoolEnv("DD_TRACE_STATS_COMPUTATION_ENABLED", false)
 	c.partialFlushEnabled = internal.BoolEnv("DD_TRACE_PARTIAL_FLUSH_ENABLED", false)
 	c.partialFlushMinSpans = internal.IntEnv("DD_TRACE_PARTIAL_FLUSH_MIN_SPANS", partialFlushMinSpansDefault)
@@ -450,7 +461,6 @@ func newConfig(opts ...StartOption) *config {
 		}
 		c.dogstatsdAddr = addr
 	}
-
 	return c
 }
 
@@ -996,6 +1006,20 @@ func WithProfilerEndpoints(enabled bool) StartOption {
 	}
 }
 
+// WithDebugSpansMode enables debugging old spans that may have been
+// abandoned, which may prevent traces from being set to the Datadog
+// Agent, especially if partial flushing is off.
+// This setting can also be configured by setting DD_TRACE_DEBUG_ABANDONED_SPANS
+// to true. The timeout will default to 10 minutes, unless overwritten
+// by DD_TRACE_ABANDONED_SPAN_TIMEOUT.
+// This feature is disabled by default.
+func WithDebugSpansMode(timeout time.Duration) StartOption {
+	return func(c *config) {
+		c.debugAbandonedSpans = true
+		c.spanTimeout = timeout
+	}
+}
+
 // WithPartialFlushing enables flushing of partially finished traces.
 // This is done after "numSpans" have finished in a single local trace at
 // which point all finished spans in that trace will be flushed, freeing up