feat(telemetry): propagate caller trace context to aggregate load spans - #90
Conversation
PR SummaryMedium Risk Overview
Reviewed by Cursor Bugbot for commit aeb38e0. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
Warning Rate limit exceeded
Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 11 minutes and 55 seconds. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (9)
WalkthroughThe pull request threads metadata through aggregate startup and dispatcher execution, enabling W3C trace context (traceparent/tracestate) propagation from dispatch to aggregate load/populate spans, allowing those spans to be linked as children of the dispatch trace rather than appearing as orphaned roots. Changes
Sequence DiagramsequenceDiagram
participant Dispatcher
participant Supervisor
participant AggregateProcess as Aggregate Process
participant StateBuilder as StateBuilder
participant OTelHandler as OTel Handler
participant OTelContext as OTel Context
Dispatcher->>Supervisor: open_aggregate(app, module, uuid, metadata)
Note over Dispatcher: metadata contains traceparent/tracestate
Supervisor->>AggregateProcess: start_link with {state, metadata}
AggregateProcess->>AggregateProcess: init({state, metadata})
AggregateProcess->>AggregateProcess: schedule continue with metadata
AggregateProcess->>StateBuilder: populate(state, metadata)
StateBuilder->>StateBuilder: rebuild_from_events(events, metadata)
StateBuilder->>OTelHandler: telemetry event [:commanded, :aggregate, :load, :start]<br/>(includes metadata)
OTelHandler->>OTelHandler: extract_propagated_ctx(metadata)
alt Traceparent Present
OTelHandler->>OTelContext: attach(extracted_ctx)
Note over OTelContext: Span becomes child of dispatch trace
else Traceparent Missing
Note over OTelHandler: Context undefined, span is orphaned root
end
OTelHandler->>OTelHandler: create_span (with propagated parent)
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
lib/commanded/aggregates/aggregate_state_builder.ex (1)
117-188:⚠️ Potential issue | 🟡 MinorPropagate metadata consistently to load stop and populate telemetry.
metadatais included only in the load:startevent. The load:stopand populate:start/:stopevents still call helpers with the default%{}, so telemetry handlers on those events cannot see the propagated trace metadata.🔧 Proposed propagation fix
- Telemetry.stop( + Telemetry.stop( load_prefix, load_start, - load_stop_metadata(state, snapshot_used, snapshot_source_version), + load_stop_metadata(state, snapshot_used, snapshot_source_version, metadata), %{count: 0} ) @@ - {state, count} = rebuild_from_event_stream(event_stream, state) + {state, count} = rebuild_from_event_stream(event_stream, state, metadata) @@ - Telemetry.stop( + Telemetry.stop( load_prefix, load_start, - load_stop_metadata(state, snapshot_used, snapshot_source_version), + load_stop_metadata(state, snapshot_used, snapshot_source_version, metadata), %{count: count} ) @@ - defp rebuild_from_event_stream(event_stream, %Aggregate{} = state) do + defp rebuild_from_event_stream(event_stream, %Aggregate{} = state, metadata \\ %{}) do telemetry_prefix = [:commanded, :aggregate, :populate] - start_time = Telemetry.start(telemetry_prefix, telemetry_metadata(state)) + start_time = Telemetry.start(telemetry_prefix, telemetry_metadata(state, metadata)) @@ - Telemetry.stop(telemetry_prefix, start_time, telemetry_metadata(state), %{count: count}) + Telemetry.stop(telemetry_prefix, start_time, telemetry_metadata(state, metadata), %{count: count}) @@ - defp load_stop_metadata(aggregate, snapshot_used, snapshot_source_version) do - telemetry_metadata(aggregate) + defp load_stop_metadata(aggregate, snapshot_used, snapshot_source_version, metadata \\ %{}) do + telemetry_metadata(aggregate, metadata) |> Map.merge(%{ snapshot_used: snapshot_used, snapshot_source_version: snapshot_source_versionAlso applies to: 195-210
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@lib/commanded/aggregates/aggregate_state_builder.ex` around lines 117 - 188, The load :stop and populate :start/:stop telemetry calls don't receive the incoming trace metadata; update functions to propagate the metadata argument through: when calling Telemetry.start and Telemetry.stop in the loader, pass telemetry_metadata(state, metadata) instead of telemetry_metadata(state); change rebuild_from_event_stream to accept a metadata parameter (rebuild_from_event_stream(event_stream, state, metadata)) and use Telemetry.start(telemetry_prefix, telemetry_metadata(state, metadata)) and Telemetry.stop(..., telemetry_metadata(state, metadata), ...); update its call site(s) where rebuild_from_event_stream is invoked to forward the metadata; and change load_stop_metadata/3 to call telemetry_metadata(aggregate, metadata) (and update its callers) so all load/stop and populate events include the original metadata.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@lib/commanded/opentelemetry/aggregate_populate.ex`:
- Around line 33-36: The code currently calls :otel_ctx.attach(ctx) when
Helpers.extract_propagated_ctx(meta[:metadata]) returns {_links, ctx} but never
clears or restores context; change the match to call
:otel_ctx.attach(:undefined) in the {_links, :undefined} branch to explicitly
clear context and, when attaching ctx, capture the return value (prev_ctx =
:otel_ctx.attach(ctx)) and after the load span ends call
:otel_ctx.attach(prev_ctx) to restore the previous context so aggregate
processes do not retain the caller’s OTel context; reference
Helpers.extract_propagated_ctx, :otel_ctx.attach and the load span surrounding
this logic when making the change.
---
Outside diff comments:
In `@lib/commanded/aggregates/aggregate_state_builder.ex`:
- Around line 117-188: The load :stop and populate :start/:stop telemetry calls
don't receive the incoming trace metadata; update functions to propagate the
metadata argument through: when calling Telemetry.start and Telemetry.stop in
the loader, pass telemetry_metadata(state, metadata) instead of
telemetry_metadata(state); change rebuild_from_event_stream to accept a metadata
parameter (rebuild_from_event_stream(event_stream, state, metadata)) and use
Telemetry.start(telemetry_prefix, telemetry_metadata(state, metadata)) and
Telemetry.stop(..., telemetry_metadata(state, metadata), ...); update its call
site(s) where rebuild_from_event_stream is invoked to forward the metadata; and
change load_stop_metadata/3 to call telemetry_metadata(aggregate, metadata) (and
update its callers) so all load/stop and populate events include the original
metadata.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5afddf85-df7e-41dd-a3bf-7d9d1d6f3390
📒 Files selected for processing (8)
guides/explanations/fork-differences.mdlib/commanded/aggregates/aggregate.exlib/commanded/aggregates/aggregate_state_builder.exlib/commanded/aggregates/supervisor.exlib/commanded/commands/dispatcher.exlib/commanded/opentelemetry/aggregate_populate.extest/opentelemetry/aggregate_populate_test.exstest/support/factory.ex
f734fad to
659f999
Compare
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
b540b85 to
aeb38e0
Compare
1 participant
Summary
loadandpopulatespans were orphaned root spans because they fire duringhandle_continuein a GenServer process that has no OTel context from the callertraceparent/tracestate) throughSupervisor.open_aggregate→Aggregate.start_link→init→handle_continue→AggregateStateBuilder.populate→ telemetry eventsAggregatePopulateOTel handler now extracts and attaches the propagated trace context before creating theloadspan, making it a child of the dispatch trace