Repository navigation
Conversation
Add a tracing integration for github.com/riverqueue/river. NewMiddleware returns a River middleware that creates a river.insert_many producer span per insert batch and a river.work consumer span per job attempt. The insert span's context, including its sampling decision, is stored in each job's metadata under "_datadog", so work spans continue the trace that inserted the job, retries included. WithDistributedTracing(false) turns this off. Snoozes are not errors, cancels are, and panics are recorded and re-raised with their original value. Orchestrion adds the middleware to every river.Client by prepending to river.NewClient, on a copy of the caller's config. Fixes DataDog#3103
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: fb666373d3
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| tags := map[string]any{ | ||
| ext.ResourceName: job.Kind, | ||
| ext.MessagingDestinationName: job.Queue, | ||
| ext.MessagingMessageID: job.ID, |
There was a problem hiding this comment.
On every consumer span, this passes River's int64 job ID directly to the tracer, which stores numeric tag values in the span metrics map as float64; messaging.message.id is an identity/string attribute, so BIGINT IDs above 2^53 are rounded and even smaller IDs are emitted in the wrong payload field, preventing reliable correlation with the River job row. Format the ID as a base-10 string before adding the tag.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Good catch, fixed in 2cdbd78: the job ID is now formatted as a base-10 string, and the test asserts the string value.
River job IDs are int64. Numeric tags are stored as float64 metrics, which rounds IDs above 2^53 and puts an identifier in the metrics map.
![datadog-datadog-prod-us1-2[bot]](https://avatars.githubusercontent.com/u/365230?s=60&v=4){kind=small}
There was a problem hiding this comment.
Orchestrion adds duplicate tracing when River middleware is registered through Config.Plugins. This produces duplicate spans and bypasses an explicitly disabled propagation setting.
🤖 Bits Code Review · Commit fb66637
| rivertrace: github.com/DataDog/dd-trace-go/contrib/riverqueue/river/v2 | ||
| template: |- | ||
| {{- $cfg := .Function.Argument 1 -}} | ||
| if {{ $cfg }} != nil { |
There was a problem hiding this comment.
When an application registers NewMiddleware through Config.Plugins, Orchestrion injects another instance because it checks only the middleware fields. River executes both instances, producing duplicate insert and worker spans. The added default instance also propagates trace context despite WithDistributedTracing(false) on the application's instance. Check Plugins before injecting into either middleware configuration path.
| if {{ $cfg }} != nil { | |
| __dd_tracedPlugin := false | |
| if {{ $cfg }} != nil { | |
| for _, __dd_plugin := range {{ $cfg }}.Plugins { | |
| if _, ok := __dd_plugin.(*rivertrace.Middleware); ok { | |
| __dd_tracedPlugin = true | |
| break | |
| } | |
| } | |
| } | |
| if {{ $cfg }} != nil && !__dd_tracedPlugin { |
React: 👍 useful · 👎 wrong · 😕 unclear · 👀 right, but not acting on it
What each reaction means
- 👍 Useful: correct, and you're fixing it or will.
- 👎 Wrong: the finding is incorrect, for example the bug can't happen or the code was misread.
- 😕 Unclear: you can't tell what it means or why it matters.
- 👀 Right, not acting: correct, but intended, out of scope for this PR, a known risk, or too minor.
🤖 Bits Code Review · Open Bits AI session
There was a problem hiding this comment.
Right, fixed in 076a5f4: the woven code now also checks Config.Plugins before adding the middleware. Verified with an Orchestrion build where the app registers NewMiddleware as a plugin: one insert span and one work span per job.
…lugin River runs a middleware registered through Config.Plugins like any other. Skip clients that already have the tracing middleware there, as the woven code already did for the deprecated WorkerMiddleware field.
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| // spliceTraceContext returns the JSON object {"_datadog":traceContext,rest}, | ||
| // where rest holds the members of another object, without its braces. | ||
| func spliceTraceContext(rest, traceContext []byte) []byte { | ||
| out := make([]byte, 0, len(metadataKey)+len(traceContext)+len(rest)+6) |
| @@ -0,0 +1,103 @@ | |||
| module github.com/DataDog/dd-trace-go/contrib/riverqueue/river/v2 | |||
❌ ErrorsYour PR has failed checks. Please review the issues below and take necessary action before merging. 🚦 10 Pipeline jobs failed
Useful? React with 👍 / 👎 This comment will be updated automatically if new data arrives.🔗 Commit SHA: 076a5f4 | Docs | View more details | Give us feedback! |
What does this PR do?
Adds
contrib/riverqueue/river, a tracing integration for River (github.com/riverqueue/river), the Postgres job queue, with Orchestrion support.rivertrace.NewMiddleware()is a River middleware (rivertype.JobInsertMiddleware+rivertype.WorkerMiddleware) that creates:river.insert_many(producer) around every insert call (Insert,InsertTx,InsertMany, ...). One span per batch; resource is the job kind, ormultiplefor mixed batches. Tags:messaging.system=river,messaging.destination.name(queue, when the batch shares one),messaging.batch.message_count,river.insert.unique_skipped_as_duplicate_count.river.work(consumer) around every job attempt. Resource is the job kind. Tags:messaging.destination.name,messaging.message.id(job ID),river.job.kind,river.job.attempt,river.job.max_attempts,river.job.priority.Propagation. The insert span's context is stored in each job's metadata under
_datadog(same key as the SQS/SNS integrations), and the work span is its child. This is the part the OTel route (rivercontrib/otelriver) doesn't give Datadog users, as noted in #3103: the sampling decision of the request that enqueued the job now carries over to every attempt of that job, including retries that run hours later. If the worker's context already holds a span (for example one that was active whenclient.Startwas called), the propagated context still wins.WithDistributedTracing(false)turns propagation off.Errors. Snoozes (
river.JobSnooze) are not errors and getriver.job.snoozed. Cancels (river.JobCancel) are errors and getriver.job.cancelled. Panics are recorded and re-raised with the original value; the end-to-end test checks that River still records the worker's own stack frame inAttemptError.Trace.WithErrorCheckcan ignore errors. Tracing never fails an insert: metadata that isn't a JSON object is left for River to reject as it does today.Options.
WithService(defaultDD_SERVICE, resolved lazily like the connect integration, since Orchestrion can build clients beforetracer.Start),WithCustomTag,WithErrorCheck,WithDistributedTracing.Orchestrion.
river.NewClientis generic and takes its middleware inside*river.Config, so the aspect prepends toNewClient's body: it copies the config and adds the middleware first (outermost) withrivertrace.AddMiddleware. The caller'sConfigand slices are never modified (theTestCaseExistingMiddlewaresuite checks spare capacity too), a client that already has the middleware (inMiddleware,Pluginsor the deprecated fields) isn't traced twice, and configs using the deprecatedJobInsertMiddleware/WorkerMiddlewarefields get it there instead, since River rejects mixing them withMiddleware. The contrib package only importsrivertype, so injecting it into packageriverhas no import cycle.Performance. Static tags are built once per middleware (
NewStartSpanConfig), per-call tags go through oneWithTags, ordered ascontrib/INTEGRATIONS.mdrequires. The trace context is encoded once per batch. Empty metadata (River's default) and objects without a trace context get the entry spliced in without decoding the metadata.BenchmarkWithTraceContext, median of 6 runs on Apple Silicon arm64, compared to decoding and re-encoding the object:ns/op went from about 2.8 us to 1.2 us and 7.2 us to 2.6 us, but that machine was noisy, so I'd treat the alloc numbers as the reliable ones.
BenchmarkInsertManyandBenchmarkWorkcover the full middleware.Motivation
Fixes #3103. Revives #3245 (closed in favour of OTel), following the v2 contrib layout and the review feedback there: no
.v0suffix, propagation can be turned off, no swallowed errors in tests, low-cardinality resource names.Testing
go test -race -shuffle=on ./...incontrib/riverqueue/river, 98.4% coverage. Table-driven unit tests for both middleware methods, plus end-to-end tests that run a real River client on River's SQLite driver (riversqlite+modernc.org/sqlite), so they need no database server: trace continuation through the database round trip, sampling priority (manual.keepon the inserting request survives to the work span), batch inserts, user metadata kept, and panics.internal/orchestrion/_integration/riverqueue_river: threeTestCases (no middleware, existing application middleware, deprecatedWorkerMiddlewarefield), also on SQLite. Ran locally withgo run github.com/DataDog/orchestrion go test ./riverqueue_river/; all pass. After the review fixes I re-checked the woven code with a smaller Orchestrion build (no middleware, deprecated fields, and manual middleware or plugin plus the woven one, each traced once).go test ./scripts/ciselect/,go test ./instrumentation/ ./ddtrace/tracer/,golangci-lintwith the repo config on the new packages, the copyright check andconfiginverter check. Generated files were regenerated with their own generators (gencontribdoc,gencontribs, theorchestrion/alland_integrationgenerators,configinverter generate), and the three touched modules were tidied; I did not run the fullmake generate/make fix-modulesover every module.riverpgxv5). River's behaviour is driver independent here and metadata goes throughjsonbthe same way, but I couldn't run Docker on this machine, so there is no Postgres test in this PR. Happy to add one with the testcontainers helper if you'd like.Notes for reviewers
go.modrequires River v0.51.0 (current). River's owngo.modrequirespgx/v5v5.11.0, soorchestrion/alland_integrationmove from pgx v5.9.2 to v5.11.0.AddMiddlewareis exported for the Orchestrion aspect. I kept it generic over River's three middleware slice types rather than adding three helpers.EnvVarPrefixisRIVER, soTestInstrumentation_AnalyticsRateneedsDD_TRACE_RIVER_ANALYTICS_ENABLEDinsupported_configurations.json(boolean, defaultfalse, implementation A). Could a maintainer register it in the configuration registry? The registry validation job will fail until then.Reviewer's Checklist
make lintlocally.make testlocally. (Ran the affected packages, see Testing.)make generatelocally. (Per generator, see Testing.)make fix-moduleslocally.