From 33d30e1a5dd618815b55c0efd2b0087e31885303 Mon Sep 17 00:00:00 2001 From: Anthony Super Date: Tue, 23 Jun 2026 12:54:11 -0400 Subject: [PATCH 1/7] Enable Trace Propagation When the configuration parameter `EnableTracePropagation` is enabled, the traces generated by river workers will be *linked* to the traces that enqueued them. This is done by use of metadata parameters, following the example in River's documentation for the middleware functionality (https://riverqueue.com/docs/middleware). This uses a *link* instead of directly making a child span which is semantically more correct. Fixes #41 (I believe, at least). --- otelriver/middleware.go | 70 ++++++++++++++++++++++++++++++++- otelriver/middleware_test.go | 75 ++++++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 1 deletion(-) diff --git a/otelriver/middleware.go b/otelriver/middleware.go index 63173f3..445b6e7 100644 --- a/otelriver/middleware.go +++ b/otelriver/middleware.go @@ -3,6 +3,7 @@ package otelriver import ( "cmp" "context" + "encoding/json" "errors" "slices" "time" @@ -11,6 +12,7 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/trace" "github.com/riverqueue/river" @@ -45,6 +47,13 @@ type MiddlewareConfig struct { // metric names, with attributes differentiating them. EnableSemanticMetrics bool + // EnableTracePropagation injects W3C trace context (traceparent/tracestate) + // into job metadata on insert and extracts it on work, adding a span link + // from the work span back to the span that enqueued the job. A link is used + // rather than a parent so the work span's timeline is independent of the + // insert span (the two may be separated by minutes or hours). + EnableTracePropagation bool + // EnableWorkSpanJobKindSuffix appends the job kind a suffix to work spans // so they look like `river.work/my_job` instead of `river.work`. EnableWorkSpanJobKindSuffix bool @@ -208,6 +217,12 @@ func (m *Middleware) InsertMany(ctx context.Context, manyParams []*rivertype.Job } }() + if m.config.EnableTracePropagation { + for i := range manyParams { + manyParams[i].Metadata = injectTraceContext(ctx, manyParams[i].Metadata) + } + } + insertRes, err = doInner(ctx) panicked = false return insertRes, err @@ -219,8 +234,14 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu spanName += "/" + job.Kind } + var startOpts []trace.SpanStartOption + if m.config.EnableTracePropagation { + if sc := extractSpanContext(ctx, job.Metadata); sc.IsValid() { + startOpts = append(startOpts, trace.WithLinks(trace.Link{SpanContext: sc})) + } + } ctx, span := m.tracer.Start(ctx, spanName, - trace.WithSpanKind(trace.SpanKindConsumer)) + append(startOpts, trace.WithSpanKind(trace.SpanKindConsumer))...) defer span.End() attrs := []attribute.KeyValue{ @@ -336,6 +357,53 @@ func mustInt64Counter(meter metric.Meter, name string, options ...metric.Int64Co return metric } +// injectTraceContext injects the current span context from ctx into metadata +// JSON under the W3C "traceparent" (and optionally "tracestate") key. If +// injection fails for any reason the original metadata is returned unchanged. +func injectTraceContext(ctx context.Context, metadata []byte) []byte { + carrier := make(propagation.MapCarrier) + propagation.TraceContext{}.Inject(ctx, carrier) + if len(carrier) == 0 { + return metadata + } + if len(metadata) == 0 { + metadata = []byte("{}") + } + var meta map[string]any + if err := json.Unmarshal(metadata, &meta); err != nil { + return metadata + } + for k, v := range carrier { + meta[k] = v + } + injected, err := json.Marshal(meta) + if err != nil { + return metadata + } + return injected +} + +// extractSpanContext reads W3C trace context from metadata JSON and returns the +// remote SpanContext it encodes. Returns a zero SpanContext (IsValid() == false) +// if no traceparent is present or the metadata cannot be parsed. +func extractSpanContext(ctx context.Context, metadata []byte) trace.SpanContext { + if len(metadata) == 0 { + return trace.SpanContext{} + } + var meta map[string]any + if err := json.Unmarshal(metadata, &meta); err != nil { + return trace.SpanContext{} + } + carrier := make(propagation.MapCarrier) + for k, v := range meta { + if s, ok := v.(string); ok { + carrier[k] = s + } + } + extracted := propagation.TraceContext{}.Extract(ctx, carrier) + return trace.SpanFromContext(extracted).SpanContext() +} + // Sets success status on the given span and within the set of attributes. The // index of the status attribute is required ahead of time as a minor // optimization. diff --git a/otelriver/middleware_test.go b/otelriver/middleware_test.go index 9b35758..ac8cd1a 100644 --- a/otelriver/middleware_test.go +++ b/otelriver/middleware_test.go @@ -2,6 +2,7 @@ package otelriver import ( "context" + "encoding/json" "errors" "fmt" "testing" @@ -10,6 +11,7 @@ import ( "github.com/stretchr/testify/require" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" "go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest" @@ -304,6 +306,34 @@ func TestMiddleware(t *testing.T) { getAttribute(t, spans[0].Attributes, "kinds").AsStringSlice()) }) + t.Run("InsertManyInjectsTraceparent", func(t *testing.T) { + t.Parallel() + + middleware, bundle := setupConfig(t, &MiddlewareConfig{EnableTracePropagation: true}) + + params := []*rivertype.JobInsertParams{{Kind: "no_op"}} + doInner := func(ctx context.Context) ([]*rivertype.JobInsertResult, error) { + return []*rivertype.JobInsertResult{{Job: &rivertype.JobRow{ID: 1}}}, nil + } + + _, err := middleware.InsertMany(ctx, params, doInner) + require.NoError(t, err) + + // The insert_many span's context should have been injected into the params metadata. + require.NotNil(t, params[0].Metadata) + var meta map[string]any + require.NoError(t, json.Unmarshal(params[0].Metadata, &meta)) + traceparent, ok := meta["traceparent"].(string) + require.True(t, ok, "expected traceparent key in job metadata") + + // The traceparent must reference the insert_many span's trace and span IDs. + spans := bundle.traceExporter.GetSpans() + require.Len(t, spans, 1) + insertSpan := spans[0] + require.Contains(t, traceparent, insertSpan.SpanContext.TraceID().String()) + require.Contains(t, traceparent, insertSpan.SpanContext.SpanID().String()) + }) + t.Run("InsertManyDurationUnitMS", func(t *testing.T) { t.Parallel() @@ -750,6 +780,51 @@ func TestMiddleware(t *testing.T) { } }) + t.Run("WorkExtractsTraceparent", func(t *testing.T) { + t.Parallel() + + middleware, bundle := setupConfig(t, &MiddlewareConfig{EnableTracePropagation: true}) + + // Build a synthetic traceparent pointing to a remote parent span. + parentTraceID := "4bf92f3577b34da6a3ce929d0e0e4736" + parentSpanID := "00f067aa0ba902b7" + carrier := propagation.MapCarrier{ + "traceparent": fmt.Sprintf("00-%s-%s-01", parentTraceID, parentSpanID), + } + metadata, err := json.Marshal(carrier) + require.NoError(t, err) + + err = middleware.Work(ctx, &rivertype.JobRow{ + Kind: "no_op", + Metadata: metadata, + }, func(ctx context.Context) error { return nil }) + require.NoError(t, err) + + spans := bundle.traceExporter.GetSpans() + require.Len(t, spans, 1) + workSpan := spans[0] + + // The work span must be linked to (not a child of) the insert span. + require.False(t, workSpan.Parent.IsValid(), "work span should not be a child of the insert span") + require.Len(t, workSpan.Links, 1) + require.Equal(t, parentTraceID, workSpan.Links[0].SpanContext.TraceID().String()) + require.Equal(t, parentSpanID, workSpan.Links[0].SpanContext.SpanID().String()) + }) + + t.Run("WorkExtractsTraceparentMissingMetadata", func(t *testing.T) { + t.Parallel() + + middleware, bundle := setupConfig(t, &MiddlewareConfig{EnableTracePropagation: true}) + + // No traceparent in metadata — work span should be a root span. + err := middleware.Work(ctx, &rivertype.JobRow{Kind: "no_op"}, func(ctx context.Context) error { return nil }) + require.NoError(t, err) + + spans := bundle.traceExporter.GetSpans() + require.Len(t, spans, 1) + require.False(t, spans[0].Parent.IsValid(), "expected no parent span when metadata has no traceparent") + }) + t.Run("WorkEnableWorkSpanJobKindSuffix ", func(t *testing.T) { t.Parallel() From d8f72fb705cc90593ba9e6844733e0b5b3631aba Mon Sep 17 00:00:00 2001 From: Anthony Super Date: Mon, 29 Jun 2026 13:51:07 -0600 Subject: [PATCH 2/7] Use sjson instead of json --- otelriver/middleware.go | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/otelriver/middleware.go b/otelriver/middleware.go index 445b6e7..aa25df2 100644 --- a/otelriver/middleware.go +++ b/otelriver/middleware.go @@ -17,6 +17,7 @@ import ( "github.com/riverqueue/river" "github.com/riverqueue/river/rivertype" + "github.com/tidwall/sjson" ) const ( @@ -369,18 +370,15 @@ func injectTraceContext(ctx context.Context, metadata []byte) []byte { if len(metadata) == 0 { metadata = []byte("{}") } - var meta map[string]any - if err := json.Unmarshal(metadata, &meta); err != nil { - return metadata - } + original := metadata for k, v := range carrier { - meta[k] = v - } - injected, err := json.Marshal(meta) - if err != nil { - return metadata + var err error + metadata, err = sjson.SetBytes(metadata, k, v) + if err != nil { + return original + } } - return injected + return metadata } // extractSpanContext reads W3C trace context from metadata JSON and returns the From 0ea8d939f6be4a49cdba0449c5bf5a28d84e3181 Mon Sep 17 00:00:00 2001 From: Anthony Super Date: Mon, 29 Jun 2026 13:56:57 -0600 Subject: [PATCH 3/7] Don't allow for pointless fallback --- otelriver/middleware.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/otelriver/middleware.go b/otelriver/middleware.go index aa25df2..e730a3f 100644 --- a/otelriver/middleware.go +++ b/otelriver/middleware.go @@ -398,7 +398,11 @@ func extractSpanContext(ctx context.Context, metadata []byte) trace.SpanContext carrier[k] = s } } - extracted := propagation.TraceContext{}.Extract(ctx, carrier) + // We use context.Background here because the only purpose of this function is to return + // a span context for *linking*. If one doesn't exist, we don't want to extract anything - + // and we certainly don't want to extract the span from `ctx`, which would most often lead to us + // linking to ourselves, which is pretty obviously incorrect! + extracted := propagation.TraceContext{}.Extract(context.Background(), carrier) return trace.SpanFromContext(extracted).SpanContext() } From c6e4073523f90e66e8f2f991a4938fa981e92ea9 Mon Sep 17 00:00:00 2001 From: Anthony Super Date: Mon, 29 Jun 2026 13:59:17 -0600 Subject: [PATCH 4/7] Be less verbose --- otelriver/middleware.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/otelriver/middleware.go b/otelriver/middleware.go index e730a3f..ab66d00 100644 --- a/otelriver/middleware.go +++ b/otelriver/middleware.go @@ -50,9 +50,7 @@ type MiddlewareConfig struct { // EnableTracePropagation injects W3C trace context (traceparent/tracestate) // into job metadata on insert and extracts it on work, adding a span link - // from the work span back to the span that enqueued the job. A link is used - // rather than a parent so the work span's timeline is independent of the - // insert span (the two may be separated by minutes or hours). + // from the work span back to the span that enqueued the job. EnableTracePropagation bool // EnableWorkSpanJobKindSuffix appends the job kind a suffix to work spans @@ -238,6 +236,9 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu var startOpts []trace.SpanStartOption if m.config.EnableTracePropagation { if sc := extractSpanContext(ctx, job.Metadata); sc.IsValid() { + // We use a *link* to the span that enqueued this value, because river jobs are async by nature, so they may happen + // minutes, hours, or even days after they're enqueued, which can lead to really weird span contexts if a direct parent + // relationship is used. startOpts = append(startOpts, trace.WithLinks(trace.Link{SpanContext: sc})) } } From 648c0f8712cecb4eb03a246d9f20fb073a3167cf Mon Sep 17 00:00:00 2001 From: Anthony Super Date: Mon, 29 Jun 2026 14:01:35 -0600 Subject: [PATCH 5/7] Add changelog entry --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d8ef8f..8208aab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add `EnableTracePropagation` configuration method, which enables linking river job spans to the span that enqueued them for tracing *why* a job was enqueued. [PR #65](https://github.com/riverqueue/rivercontrib/pull/65/changes) + ## [0.10.0] - 2026-06-06 ### Added From 4ab26061f4b280a3d8155e67c5d7238c4e052fc9 Mon Sep 17 00:00:00 2001 From: Anthony Super Date: Mon, 29 Jun 2026 15:12:40 -0600 Subject: [PATCH 6/7] Update CHANGELOG.md Co-authored-by: Brandur --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8208aab..a930709 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Add `EnableTracePropagation` configuration method, which enables linking river job spans to the span that enqueued them for tracing *why* a job was enqueued. [PR #65](https://github.com/riverqueue/rivercontrib/pull/65/changes) +- Add `EnableTracePropagation` configuration method, which enables linking river job spans to the span that enqueued them for tracing *why* a job was enqueued. [PR #65](https://github.com/riverqueue/rivercontrib/pull/65) ## [0.10.0] - 2026-06-06 From 2a27c330b932fb7a0b247782d237c2d7c4904755 Mon Sep 17 00:00:00 2001 From: Anthony Super Date: Mon, 29 Jun 2026 15:19:09 -0600 Subject: [PATCH 7/7] Skip context parameter This is no longer used. Also fixes lints. --- otelriver/middleware.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/otelriver/middleware.go b/otelriver/middleware.go index ab66d00..27ea91e 100644 --- a/otelriver/middleware.go +++ b/otelriver/middleware.go @@ -8,6 +8,7 @@ import ( "slices" "time" + "github.com/tidwall/sjson" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" @@ -17,7 +18,6 @@ import ( "github.com/riverqueue/river" "github.com/riverqueue/river/rivertype" - "github.com/tidwall/sjson" ) const ( @@ -235,7 +235,8 @@ func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner fu var startOpts []trace.SpanStartOption if m.config.EnableTracePropagation { - if sc := extractSpanContext(ctx, job.Metadata); sc.IsValid() { + //nolint:contextcheck + if sc := extractSpanContext(job.Metadata); sc.IsValid() { // We use a *link* to the span that enqueued this value, because river jobs are async by nature, so they may happen // minutes, hours, or even days after they're enqueued, which can lead to really weird span contexts if a direct parent // relationship is used. @@ -385,7 +386,7 @@ func injectTraceContext(ctx context.Context, metadata []byte) []byte { // extractSpanContext reads W3C trace context from metadata JSON and returns the // remote SpanContext it encodes. Returns a zero SpanContext (IsValid() == false) // if no traceparent is present or the metadata cannot be parsed. -func extractSpanContext(ctx context.Context, metadata []byte) trace.SpanContext { +func extractSpanContext(metadata []byte) trace.SpanContext { if len(metadata) == 0 { return trace.SpanContext{} }