diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index 573c72b1..d19012d8 100644 --- a/guides/explanations/fork-differences.md +++ b/guides/explanations/fork-differences.md @@ -187,6 +187,31 @@ end - Event handlers can extract trace context to create span links or parent-child relationships - Non-invasive - only adds metadata when a span is active +### **W3C Baggage Propagation Middleware** +[PR #108](https://github.com/straw-hat-team/commanded/pull/108) + +**Changes:** +- Added `Commanded.Middleware.BaggagePropagator` middleware for propagating OpenTelemetry baggage +- Captures the current baggage and stores it in event metadata using the W3C Baggage standard +- Consumer-side helpers updated to rebuild baggage into the extracted OpenTelemetry context, including the baggage-only case (no `traceparent`) + +**Usage:** +```elixir +defmodule MyApp.Router do + use Commanded.Commands.Router + + middleware Commanded.Middleware.TraceContextPropagator + middleware Commanded.Middleware.BaggagePropagator + + # ... your command routes +end +``` + +**Benefits:** +- Propagates user-defined name/value pairs (e.g. `tenant_id`, `user_id`) from command dispatch to event handlers +- Independent of trace context — W3C Baggage and W3C Trace Context are separate specifications, so each is opt-in via its own middleware +- Router-level declaration makes baggage propagation visible at the call site, since baggage entries persist in event metadata for the lifetime of the stream + ### **Aggregate Identity Protocol** [PR #43](https://github.com/straw-hat-team/commanded/pull/43) diff --git a/lib/commanded/middleware/baggage_propagator.ex b/lib/commanded/middleware/baggage_propagator.ex new file mode 100644 index 00000000..dc80e142 --- /dev/null +++ b/lib/commanded/middleware/baggage_propagator.ex @@ -0,0 +1,91 @@ +if Code.ensure_loaded?(:otel_propagator_text_map) do + defmodule Commanded.Middleware.BaggagePropagator do + @moduledoc """ + A middleware for propagating W3C Baggage to Event Handlers. + + This middleware captures the current baggage from the OpenTelemetry context + and stores it in event metadata following the + [W3C Baggage](https://www.w3.org/TR/baggage/) specification, allowing event + handlers to read user-defined name/value pairs that travel alongside trace + context. + + Baggage and W3C Trace Context are independent specifications. + `Commanded.Middleware.TraceContextPropagator` propagates `traceparent` and + `tracestate`; this middleware propagates `baggage`. Use both together to + propagate the full OpenTelemetry context. + + ## W3C Baggage + + The middleware stores baggage using the standard W3C header name: + + * `baggage` - A comma-separated list of `key=value` pairs, with optional + per-entry metadata (e.g. `userId=alice,serverNode=DF%2028,isProduction=false`). + + ## Persistence considerations + + Baggage entries are written into event metadata and therefore persisted in + the event store for the lifetime of the stream. Do not place PII, secrets, + or unbounded values into baggage when this middleware is enabled. Filter or + redact at the producer site (before `OpenTelemetry.Baggage.set/1`) if + necessary. + + ## Dependencies + + This middleware requires `opentelemetry_api` to be installed and the + `baggage` propagator to be present in the global text map propagator + composite (the OpenTelemetry SDK's default is `[trace_context, baggage]`). + + {:opentelemetry_api, "~> 1.0"} + + ## Usage + + ```elixir + defmodule BankRouter do + use Commanded.Commands.Router + + middleware Commanded.Middleware.TraceContextPropagator + middleware Commanded.Middleware.BaggagePropagator + + dispatch [OpenAccount, DepositMoney], + to: BankAccount, + identity: :account_number + end + ``` + """ + + @behaviour Commanded.Middleware + + alias Commanded.Middleware.Pipeline + + @doc """ + Injects the W3C `baggage` header into the command pipeline metadata. + + Called before command dispatch to capture the current baggage. If no + baggage is set on the current OpenTelemetry context, the pipeline is + returned unchanged. + """ + @impl true + def before_dispatch(%Pipeline{} = pipeline) do + case :otel_propagator_text_map.inject([]) do + [] -> + pipeline + + headers -> + maybe_assign(pipeline, "baggage", List.keyfind(headers, "baggage", 0)) + end + end + + defp maybe_assign(pipeline, _key, nil), do: pipeline + + defp maybe_assign(pipeline, key, {_, value}), + do: Pipeline.assign_metadata(pipeline, key, value) + + @doc false + @impl true + def after_dispatch(pipeline), do: pipeline + + @doc false + @impl true + def after_failure(pipeline), do: pipeline + end +end diff --git a/lib/commanded/opentelemetry/helpers.ex b/lib/commanded/opentelemetry/helpers.ex index 92c9be43..2e32e19e 100644 --- a/lib/commanded/opentelemetry/helpers.ex +++ b/lib/commanded/opentelemetry/helpers.ex @@ -21,10 +21,12 @@ defmodule Commanded.OpenTelemetry.Helpers do |> :otel_propagator_text_map.extract_to(headers) span_ctx = :otel_tracer.current_span_ctx(ctx) + baggage = :otel_baggage.get_all(ctx) - case span_ctx do - :undefined -> {[], :undefined} - span_ctx -> {[OpenTelemetry.link(span_ctx)], ctx} + case {span_ctx, map_size(baggage)} do + {:undefined, 0} -> {[], :undefined} + {:undefined, _} -> {[], ctx} + {span_ctx, _} -> {[OpenTelemetry.link(span_ctx)], ctx} end end @@ -36,6 +38,7 @@ defmodule Commanded.OpenTelemetry.Helpers do [] |> maybe_add_header(metadata, "traceparent") |> maybe_add_header(metadata, "tracestate") + |> maybe_add_header(metadata, "baggage") end defp maybe_add_header(headers, metadata, key) do diff --git a/test/middleware/baggage_propagator_test.exs b/test/middleware/baggage_propagator_test.exs new file mode 100644 index 00000000..b16fb2f6 --- /dev/null +++ b/test/middleware/baggage_propagator_test.exs @@ -0,0 +1,107 @@ +defmodule Commanded.Middleware.BaggagePropagatorTest do + use ExUnit.Case, async: false + + alias Commanded.Middleware.BaggagePropagator + alias Commanded.Middleware.Commands.Fail + alias Commanded.Middleware.Pipeline + + setup do + on_exit(fn -> :otel_baggage.clear() end) + :otel_baggage.clear() + :ok + end + + describe "before_dispatch/1" do + test "does not modify metadata when no baggage is set" do + pipeline = %Pipeline{command: %Fail{}, metadata: %{}} + + result = BaggagePropagator.before_dispatch(pipeline) + + assert result.metadata == %{} + end + + test "captures baggage when present on the current context" do + :otel_baggage.set(%{"userId" => "alice", "tenant" => "acme"}) + + pipeline = %Pipeline{command: %Fail{}, metadata: %{}} + + result = BaggagePropagator.before_dispatch(pipeline) + + assert %{"baggage" => header} = result.metadata + assert header =~ "userId=alice" + assert header =~ "tenant=acme" + end + + test "preserves existing metadata" do + :otel_baggage.set(%{"userId" => "alice"}) + + pipeline = %Pipeline{ + command: %Fail{}, + metadata: %{"correlation_id" => "abc-123"} + } + + result = BaggagePropagator.before_dispatch(pipeline) + + assert result.metadata["correlation_id"] == "abc-123" + assert result.metadata["baggage"] =~ "userId=alice" + end + + test "does not modify pipeline command" do + :otel_baggage.set(%{"userId" => "alice"}) + + command = %Fail{} + pipeline = %Pipeline{command: command, metadata: %{}} + + result = BaggagePropagator.before_dispatch(pipeline) + + assert result.command == command + end + + test "does not modify pipeline assigns" do + :otel_baggage.set(%{"userId" => "alice"}) + + pipeline = %Pipeline{ + command: %Fail{}, + metadata: %{}, + assigns: %{existing: "value"} + } + + result = BaggagePropagator.before_dispatch(pipeline) + + assert result.assigns == %{existing: "value"} + end + + test "round-trips through W3C baggage decoding" do + :otel_baggage.set(%{"userId" => "alice", "tenant" => "acme"}) + + pipeline = %Pipeline{command: %Fail{}, metadata: %{}} + + result = BaggagePropagator.before_dispatch(pipeline) + + ctx = + :otel_ctx.new() + |> :otel_propagator_text_map.extract_to([{"baggage", result.metadata["baggage"]}]) + + decoded = :otel_baggage.get_all(ctx) + + assert decoded["userId"] == {"alice", []} + assert decoded["tenant"] == {"acme", []} + end + end + + describe "after_dispatch/1" do + test "returns pipeline unchanged" do + pipeline = %Pipeline{command: %Fail{}, metadata: %{"test" => "value"}} + + assert BaggagePropagator.after_dispatch(pipeline) == pipeline + end + end + + describe "after_failure/1" do + test "returns pipeline unchanged" do + pipeline = %Pipeline{command: %Fail{}, metadata: %{"test" => "value"}} + + assert BaggagePropagator.after_failure(pipeline) == pipeline + end + end +end diff --git a/test/opentelemetry/helpers_test.exs b/test/opentelemetry/helpers_test.exs index f9a717b8..6c499e11 100644 --- a/test/opentelemetry/helpers_test.exs +++ b/test/opentelemetry/helpers_test.exs @@ -64,6 +64,32 @@ defmodule Commanded.OpenTelemetry.HelpersTest do after_ctx = :otel_ctx.get_current() assert before_ctx == after_ctx end + + test "returns ctx with baggage when only baggage is present" do + {links, ctx} = Helpers.extract_propagated_ctx(%{"baggage" => "userId=alice"}) + + assert links == [] + refute ctx == :undefined + assert :otel_baggage.get_all(ctx) == %{"userId" => {"alice", []}} + end + + test "returns ctx with both span and baggage when both are present" do + {_trace_id, _span_id, traceparent} = make_traceparent() + + metadata = %{ + "traceparent" => traceparent, + "baggage" => "userId=alice,tenant=acme" + } + + {links, ctx} = Helpers.extract_propagated_ctx(metadata) + + refute ctx == :undefined + assert [%{}] = links + + baggage = :otel_baggage.get_all(ctx) + assert baggage["userId"] == {"alice", []} + assert baggage["tenant"] == {"acme", []} + end end describe "clear_ctx/0" do