Skip to content

Commit 92ae805

Browse files
darcciodarccio
authored
Merge branch 'main' into dario.castane/ktlo/flaky-testtracer-prioritysampler
2 parents 5b78dab + 1ebb74b commit 92ae805

37 files changed

Lines changed: 6539 additions & 249 deletions

‎contrib/redis/rueidis/rueidis.go‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -76,8 +76,6 @@ func (c *client) startSpan(ctx context.Context, cmd command) (*tracer.Span, cont
7676
tracer.ServiceName(c.cfg.serviceName),
7777
tracer.ResourceName(cmd.statement),
7878
tracer.SpanType(ext.SpanTypeRedis),
79-
tracer.Tag(ext.TargetHost, c.host),
80-
tracer.Tag(ext.TargetPort, c.port),
8179
tracer.Tag(ext.Component, instrumentation.PackageRedisRueidis),
8280
tracer.Tag(ext.SpanKind, ext.SpanKindClient),
8381
tracer.Tag(ext.DBSystem, ext.DBSystemRedis),

‎contrib/redis/rueidis/rueidis_test.go‎

Lines changed: 0 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -300,54 +300,6 @@ func TestNewClient(t *testing.T) {
300300
},
301301
wantServiceName: "global-service",
302302
},
303-
{
304-
name: "Test SET command with canceled context and custom error check",
305-
opts: []Option{
306-
WithErrorCheck(func(err error) bool {
307-
return err != nil && !rueidis.IsRedisNil(err) && !errors.Is(err, context.Canceled)
308-
}),
309-
},
310-
runTest: func(t *testing.T, ctx context.Context, client rueidis.Client) {
311-
ctx, cancel := context.WithCancel(ctx)
312-
cancel()
313-
require.Error(t, client.Do(ctx, client.B().Set().Key("test_key").Value("test_value").Build()).Error())
314-
},
315-
assertSpans: func(t *testing.T, spans []*mocktracer.Span) {
316-
require.Len(t, spans, 1)
317-
318-
span := spans[0]
319-
assert.Equal(t, "SET", span.Tag(ext.ResourceName))
320-
assert.Nil(t, span.Tag(ext.RedisRawCommand))
321-
assert.Equal(t, "false", span.Tag(ext.RedisClientCacheHit))
322-
assert.Less(t, span.Tag(ext.RedisClientCacheTTL), float64(0))
323-
assert.Less(t, span.Tag(ext.RedisClientCachePXAT), float64(0))
324-
assert.Less(t, span.Tag(ext.RedisClientCachePTTL), float64(0))
325-
assert.Nil(t, span.Tag(ext.Error))
326-
},
327-
wantServiceName: "global-service",
328-
},
329-
{
330-
name: "Test redis nil not attached to span",
331-
opts: []Option{
332-
WithRawCommand(true),
333-
},
334-
runTest: func(t *testing.T, ctx context.Context, client rueidis.Client) {
335-
require.Error(t, client.Do(ctx, client.B().Get().Key("404").Build()).Error())
336-
},
337-
assertSpans: func(t *testing.T, spans []*mocktracer.Span) {
338-
require.Len(t, spans, 1)
339-
340-
span := spans[0]
341-
assert.Equal(t, "GET", span.Tag(ext.ResourceName))
342-
assert.Equal(t, "GET 404", span.Tag(ext.RedisRawCommand))
343-
assert.Equal(t, "false", span.Tag(ext.RedisClientCacheHit))
344-
assert.Less(t, span.Tag(ext.RedisClientCacheTTL), float64(0))
345-
assert.Less(t, span.Tag(ext.RedisClientCachePXAT), float64(0))
346-
assert.Less(t, span.Tag(ext.RedisClientCachePTTL), float64(0))
347-
assert.Nil(t, span.Tag(ext.Error))
348-
},
349-
wantServiceName: "global-service",
350-
},
351303
}
352304
for _, tt := range tests {
353305
t.Run(tt.name, func(t *testing.T) {

‎contrib/valkey-io/valkey-go/valkey.go‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -210,8 +210,6 @@ func (c *client) startSpan(ctx context.Context, cmd command) (*tracer.Span, cont
210210
tracer.ServiceName(c.cfg.serviceName),
211211
tracer.ResourceName(cmd.statement),
212212
tracer.SpanType(ext.SpanTypeValkey),
213-
tracer.Tag(ext.TargetHost, c.host),
214-
tracer.Tag(ext.TargetPort, c.port),
215213
tracer.Tag(ext.Component, instrumentation.PackageValkeyIoValkeyGo),
216214
tracer.Tag(ext.SpanKind, ext.SpanKindClient),
217215
tracer.Tag(ext.DBSystem, ext.DBSystemValkey),

‎contrib/valkey-io/valkey-go/valkey_test.go‎

Lines changed: 0 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -355,54 +355,6 @@ func TestNewClient(t *testing.T) {
355355
},
356356
wantServiceName: "global-service",
357357
},
358-
{
359-
name: "Test SET command with canceled context and custom error check",
360-
opts: []Option{
361-
WithErrorCheck(func(err error) bool {
362-
return err != nil && !valkey.IsValkeyNil(err) && !errors.Is(err, context.Canceled)
363-
}),
364-
},
365-
runTest: func(t *testing.T, ctx context.Context, client valkey.Client) {
366-
ctx, cancel := context.WithCancel(ctx)
367-
cancel()
368-
require.Error(t, client.Do(ctx, client.B().Set().Key("test_key").Value("test_value").Build()).Error())
369-
},
370-
assertSpans: func(t *testing.T, spans []*mocktracer.Span) {
371-
require.Len(t, spans, 1)
372-
373-
span := spans[0]
374-
assert.Equal(t, "SET", span.Tag(ext.ResourceName))
375-
assert.Nil(t, span.Tag(ext.ValkeyRawCommand))
376-
assert.Equal(t, "false", span.Tag(ext.ValkeyClientCacheHit))
377-
assert.Less(t, span.Tag(ext.ValkeyClientCacheTTL), float64(0))
378-
assert.Less(t, span.Tag(ext.ValkeyClientCachePXAT), float64(0))
379-
assert.Less(t, span.Tag(ext.ValkeyClientCachePTTL), float64(0))
380-
assert.Nil(t, span.Tag(ext.Error))
381-
},
382-
wantServiceName: "global-service",
383-
},
384-
{
385-
name: "Test valkey nil not attached to span",
386-
opts: []Option{
387-
WithRawCommand(true),
388-
},
389-
runTest: func(t *testing.T, ctx context.Context, client valkey.Client) {
390-
require.Error(t, client.Do(ctx, client.B().Get().Key("404").Build()).Error())
391-
},
392-
assertSpans: func(t *testing.T, spans []*mocktracer.Span) {
393-
require.Len(t, spans, 1)
394-
395-
span := spans[0]
396-
assert.Equal(t, "GET", span.Tag(ext.ResourceName))
397-
assert.Equal(t, "GET 404", span.Tag(ext.ValkeyRawCommand))
398-
assert.Equal(t, "false", span.Tag(ext.ValkeyClientCacheHit))
399-
assert.Less(t, span.Tag(ext.ValkeyClientCacheTTL), float64(0))
400-
assert.Less(t, span.Tag(ext.ValkeyClientCachePXAT), float64(0))
401-
assert.Less(t, span.Tag(ext.ValkeyClientCachePTTL), float64(0))
402-
assert.Nil(t, span.Tag(ext.Error))
403-
},
404-
wantServiceName: "global-service",
405-
},
406358
}
407359
for _, tt := range tests {
408360
t.Run(tt.name, func(t *testing.T) {

‎ddtrace/ext/tags.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,4 +154,10 @@ const (
154154

155155
// CloudResourceID is the cloud provider resource identifier.
156156
CloudResourceID = "cloud.resource_id"
157+
158+
// DSMTransactionID is the span tag key for a Data Streams transaction identifier.
159+
DSMTransactionID = "dsm.transaction.id"
160+
161+
// DSMTransactionCheckpoint is the span tag key for a Data Streams transaction checkpoint name.
162+
DSMTransactionCheckpoint = "dsm.transaction.checkpoint"
157163
)

‎ddtrace/tracer/api.txt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,8 @@ func StartSpanFromContext(context.Context, string, ...StartSpanOption) (*Span, c
1414
// Package Functions
1515
func SetDataStreamsCheckpoint(context.Context, ...string) (context.Context, bool)
1616
func SetDataStreamsCheckpointWithParams(context.Context, options.CheckpointParams, ...string) (context.Context, bool)
17-
func TrackDataStreamsTransaction(string)
17+
func TrackDataStreamsTransaction(context.Context, string)
18+
func TrackDataStreamsTransactionAt(context.Context, string, time.Time)
1819
func TrackKafkaCommitOffset(string, int32, int64)
1920
func TrackKafkaHighWatermarkOffset(string, string, int32, int64)
2021
func TrackKafkaProduceOffset(string, int32, int64)

‎ddtrace/tracer/data_streams.go‎

Lines changed: 29 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,10 @@ package tracer
77

88
import (
99
"context"
10+
"time"
1011

1112
"github.com/DataDog/dd-trace-go/v2/datastreams/options"
13+
"github.com/DataDog/dd-trace-go/v2/ddtrace/ext"
1214
idatastreams "github.com/DataDog/dd-trace-go/v2/internal/datastreams"
1315
)
1416

@@ -73,19 +75,40 @@ func TrackKafkaHighWatermarkOffset(cluster string, topic string, partition int32
7375
}
7476

7577
// TrackDataStreamsTransaction records a manual transaction checkpoint observation
76-
// for Data Streams Monitoring. Call this each time a transaction with the given
77-
// transactionID is observed at the named checkpoint in your pipeline.
78+
// for Data Streams Monitoring. The active span in ctx (if any) is tagged with the
79+
// transaction ID and checkpoint name for correlation in the Datadog UI.
80+
// Pass context.Background() if you do not need span tagging.
7881
//
7982
// transactionID is an application-defined identifier for the transaction (e.g. a
8083
// message ID or correlation ID). IDs longer than 255 bytes are silently truncated.
8184
//
8285
// checkpointName is a stable label for the processing stage (e.g. "ingested",
8386
// "processed", "delivered"). A maximum of 254 unique checkpoint names are supported
8487
// per processor lifetime; additional names beyond this limit are silently dropped.
85-
func TrackDataStreamsTransaction(transactionID, checkpointName string) {
86-
if t, ok := getGlobalTracer().(dataStreamsContainer); ok {
87-
if p := t.GetDataStreamsProcessor(); p != nil {
88-
p.TrackTransaction(transactionID, checkpointName)
88+
func TrackDataStreamsTransaction(ctx context.Context, transactionID, checkpointName string) {
89+
TrackDataStreamsTransactionAt(ctx, transactionID, checkpointName, time.Now())
90+
}
91+
92+
// TrackDataStreamsTransactionAt is like TrackDataStreamsTransaction but records the
93+
// observation at time t instead of the current time. Use this when the transaction
94+
// timestamp is already known (e.g. embedded in a message header).
95+
func TrackDataStreamsTransactionAt(ctx context.Context, transactionID, checkpointName string, t time.Time) {
96+
tagActiveSpan(ctx, transactionID, checkpointName)
97+
if tr, ok := getGlobalTracer().(dataStreamsContainer); ok {
98+
if p := tr.GetDataStreamsProcessor(); p != nil {
99+
p.TrackTransactionAt(transactionID, checkpointName, t)
89100
}
90101
}
91102
}
103+
104+
// tagActiveSpan sets the DSM transaction tags on the span stored in ctx, if any.
105+
// If ctx is nil or contains no span, this is a no-op.
106+
func tagActiveSpan(ctx context.Context, transactionID, checkpointName string) {
107+
if ctx == nil {
108+
return
109+
}
110+
if span, ok := SpanFromContext(ctx); ok {
111+
span.SetTag(ext.DSMTransactionID, transactionID)
112+
span.SetTag(ext.DSMTransactionCheckpoint, checkpointName)
113+
}
114+
}

‎ddtrace/tracer/data_streams_test.go‎

Lines changed: 74 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,14 @@
66
package tracer
77

88
import (
9+
"context"
910
"testing"
11+
"time"
12+
13+
"github.com/DataDog/dd-trace-go/v2/ddtrace/ext"
1014

1115
"github.com/stretchr/testify/assert"
16+
"github.com/stretchr/testify/require"
1217
)
1318

1419
// TestTrackDataStreamsTransactionPublicAPI verifies that TrackDataStreamsTransaction correctly
@@ -24,5 +29,73 @@ func TestTrackDataStreamsTransactionPublicAPI(t *testing.T) {
2429
assert.NotNil(t, tr.GetDataStreamsProcessor(), "DSM processor should be non-nil when DD_DATA_STREAMS_ENABLED=true")
2530

2631
// Should not panic and should reach the processor.
27-
TrackDataStreamsTransaction("msg-001", "ingested")
32+
TrackDataStreamsTransaction(context.Background(), "msg-001", "ingested")
33+
}
34+
35+
// TestTrackDataStreamsTransactionTagsSpan verifies that the active span in the context
36+
// is tagged with the DSM transaction ID and checkpoint name.
37+
func TestTrackDataStreamsTransactionTagsSpan(t *testing.T) {
38+
t.Setenv("DD_DATA_STREAMS_ENABLED", "true")
39+
t.Setenv("DD_INSTRUMENTATION_TELEMETRY_ENABLED", "false")
40+
Start(withNoopStats())
41+
defer Stop()
42+
43+
span, ctx := StartSpanFromContext(context.Background(), "test.op")
44+
defer span.Finish()
45+
46+
TrackDataStreamsTransaction(ctx, "tx-span-tag", "processed")
47+
48+
s, ok := SpanFromContext(ctx)
49+
require.True(t, ok)
50+
meta := s.getMetadata()
51+
assert.Equal(t, "tx-span-tag", meta[ext.DSMTransactionID])
52+
assert.Equal(t, "processed", meta[ext.DSMTransactionCheckpoint])
53+
}
54+
55+
// TestTrackDataStreamsTransactionNoSpanInContextNoops verifies that when the context
56+
// contains no span, tagging is silently skipped and the function does not panic.
57+
func TestTrackDataStreamsTransactionNoSpanInContextNoops(t *testing.T) {
58+
t.Setenv("DD_DATA_STREAMS_ENABLED", "true")
59+
t.Setenv("DD_INSTRUMENTATION_TELEMETRY_ENABLED", "false")
60+
Start(withNoopStats())
61+
defer Stop()
62+
63+
assert.NotPanics(t, func() {
64+
TrackDataStreamsTransaction(context.Background(), "tx-no-span", "ingested")
65+
})
66+
}
67+
68+
// TestTrackDataStreamsTransactionAtDelegatesToProcessor verifies that
69+
// TrackDataStreamsTransactionAt forwards the provided time to the processor.
70+
func TestTrackDataStreamsTransactionAtDelegatesToProcessor(t *testing.T) {
71+
t.Setenv("DD_DATA_STREAMS_ENABLED", "true")
72+
t.Setenv("DD_INSTRUMENTATION_TELEMETRY_ENABLED", "false")
73+
Start(withNoopStats())
74+
defer Stop()
75+
76+
fixedTime := time.Date(2024, 1, 15, 12, 0, 0, 0, time.UTC)
77+
assert.NotPanics(t, func() {
78+
TrackDataStreamsTransactionAt(context.Background(), "tx-at-001", "delivered", fixedTime)
79+
})
80+
}
81+
82+
// TestTrackDataStreamsTransactionAtTagsSpan verifies that TrackDataStreamsTransactionAt
83+
// also tags the active span in the context.
84+
func TestTrackDataStreamsTransactionAtTagsSpan(t *testing.T) {
85+
t.Setenv("DD_DATA_STREAMS_ENABLED", "true")
86+
t.Setenv("DD_INSTRUMENTATION_TELEMETRY_ENABLED", "false")
87+
Start(withNoopStats())
88+
defer Stop()
89+
90+
span, ctx := StartSpanFromContext(context.Background(), "test.op")
91+
defer span.Finish()
92+
93+
fixedTime := time.Date(2024, 1, 15, 12, 0, 0, 0, time.UTC)
94+
TrackDataStreamsTransactionAt(ctx, "tx-at-span", "delivered", fixedTime)
95+
96+
s, ok := SpanFromContext(ctx)
97+
require.True(t, ok)
98+
meta := s.getMetadata()
99+
assert.Equal(t, "tx-at-span", meta[ext.DSMTransactionID])
100+
assert.Equal(t, "delivered", meta[ext.DSMTransactionCheckpoint])
28101
}

‎internal/datastreams/processor.go‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -640,20 +640,33 @@ func (p *Processor) TrackKafkaHighWatermarkOffset(_ string, topic string, partit
640640
}
641641
}
642642

643-
// TrackTransaction records a manual transaction checkpoint observation. Use this to
644-
// track when a specific transaction ID is seen at a named checkpoint in a data pipeline.
645-
// transactionID identifies the transaction (e.g. a message ID or correlation ID).
646-
// checkpointName is a stable label for the processing stage (e.g. "ingested", "processed").
647-
func (p *Processor) TrackTransaction(transactionID, checkpointName string) {
643+
// trackTransactionAt is the shared implementation for TrackTransaction and TrackTransactionAt.
644+
func (p *Processor) trackTransactionAt(transactionID, checkpointName string, t time.Time) {
648645
dropped := p.in.push(&processorInput{
649646
typ: pointTypeTransaction,
650647
transactionEntry: transactionEntry{
651648
transactionID: transactionID,
652649
checkpointName: checkpointName,
653-
timestamp: p.time().UnixNano(),
650+
timestamp: t.UnixNano(),
654651
},
655652
})
656653
if dropped {
657654
atomic.AddInt64(&p.stats.dropped, 1)
658655
}
659656
}
657+
658+
// TrackTransaction records a manual transaction checkpoint observation. Use this to
659+
// track when a specific transaction ID is seen at a named checkpoint in a data pipeline.
660+
// transactionID identifies the transaction (e.g. a message ID or correlation ID).
661+
// checkpointName is a stable label for the processing stage (e.g. "ingested", "processed").
662+
func (p *Processor) TrackTransaction(transactionID, checkpointName string) {
663+
p.trackTransactionAt(transactionID, checkpointName, p.time())
664+
}
665+
666+
// TrackTransactionAt records a manual transaction checkpoint observation at the
667+
// provided time t instead of the current time. Use this when the observation time
668+
// is already known (e.g. a timestamp embedded in a message header). t is expected
669+
// to be in UTC.
670+
func (p *Processor) TrackTransactionAt(transactionID, checkpointName string, t time.Time) {
671+
p.trackTransactionAt(transactionID, checkpointName, t)
672+
}

‎internal/datastreams/processor_test.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -546,6 +546,41 @@ func TestTrackTransactionViaMethod(t *testing.T) {
546546
assert.NotEmpty(t, found.TransactionCheckpointIds)
547547
}
548548

549+
// TestTrackTransactionAtUsesProvidedTime verifies that TrackTransactionAt stores the
550+
// caller-supplied timestamp rather than the processor's clock.
551+
func TestTrackTransactionAtUsesProvidedTime(t *testing.T) {
552+
p := NewProcessor(nil, "env", "service", "v1", &url.URL{Scheme: "http", Host: "agent-address"}, nil)
553+
// Set the processor clock to a different time to confirm it is not used.
554+
processorTime := time.Now().Truncate(bucketDuration)
555+
p.timeSource = func() time.Time { return processorTime }
556+
557+
customTime := processorTime.Add(-5 * time.Minute)
558+
p.TrackTransactionAt("tx-custom", "ingested", customTime)
559+
560+
in := p.in.pop()
561+
require.NotNil(t, in)
562+
assert.Equal(t, pointTypeTransaction, in.typ)
563+
assert.Equal(t, "tx-custom", in.transactionEntry.transactionID)
564+
assert.Equal(t, "ingested", in.transactionEntry.checkpointName)
565+
// The stored timestamp must match the caller-supplied time, not the processor clock.
566+
assert.Equal(t, customTime.UnixNano(), in.transactionEntry.timestamp)
567+
568+
p.processInput(in)
569+
payloads := p.flush(customTime.Add(bucketDuration * 2))
570+
571+
var found *StatsBucket
572+
for _, payload := range payloads {
573+
for i := range payload.Stats {
574+
if len(payload.Stats[i].Transactions) > 0 {
575+
found = &payload.Stats[i]
576+
break
577+
}
578+
}
579+
}
580+
require.NotNil(t, found, "expected a bucket containing transactions for the custom timestamp")
581+
assert.NotEmpty(t, found.Transactions)
582+
}
583+
549584
type noOpTransport struct{}
550585

551586
// RoundTrip does nothing and returns a dummy response.

0 commit comments

Comments
 (0)