diff --git a/RELEASE.md b/RELEASE.md index 3286dcc89..5b0cb781a 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -63,6 +63,20 @@ dotnet nuget push -k NUGET_API_KEY -s https://api.nuget.org/v3/index.json ./pack ## `main` (`7.x`) branch +### Collect release-note prose + +`CHANGELOG.md` and `HISTORY.md` are generated by `tools/generate-changelog.sh` and only carry the mechanical PR and issue list. User-visible behaviour notes (dashboards moving, error rates changing, deprecations) live in the GitHub Release body instead. + +Before creating the release, sweep merged PRs in the milestone labeled `release-notes` and fold each PR's `## For the X.Y.Z release notes` section into a `## Notable behavioural changes` section of the release body: + +``` +gh pr list --repo rabbitmq/rabbitmq-dotnet-client \ + --state merged --search 'milestone:X.Y.Z label:release-notes' \ + --json number,title,url +``` + +### Cut the release + * Close the appropriate milestone, and make a note of the link to the milestone with closed issues visible * Use the GitHub web UI or `gh release create` command to create the new release * GitHub actions will build and publish the release to NuGet diff --git a/docs/internal/opentelemetry-tracing-review.md b/docs/internal/opentelemetry-tracing-review.md new file mode 100644 index 000000000..51d41e7fb --- /dev/null +++ b/docs/internal/opentelemetry-tracing-review.md @@ -0,0 +1,348 @@ +# OpenTelemetry Tracing: Implementation Review + +This document records a full review of the client's OpenTelemetry tracing implementation, carried out for issue #1967 before #1923 locks the public tracing API for 7.3.0. Every claim below that is marked *verified* was settled by driving the real SDK pipeline against a live broker, not by reading code. + +Read this before changing anything under `RabbitMQActivitySource`, `RabbitMQ.Client.OpenTelemetry`, or the `Activity.Current` call sites in `SessionBase` / `Connection`. + +## Status + +The findings were split into three groups, because they carry very different risk: + +| Group | Content | Status | +|---|---|---| +| A | Behavioural defects: ambient-span pollution, failures never recorded, untagged `tcp connection attempt`, the null-`Headers` extractor path, a wrong comment | **Fixed.** Sections below are marked `FIXED` individually. | +| B | Semantic-convention conformance. Each one changes emitted span names or attributes, and several break existing test assertions. | Open | +| C | Public API: per-provider tracing configuration. Must land before #1923. | Open | + +Group A was separated out precisely because none of it changes a conforming attribute value or span name, so it can ship without a downstream consumer having to re-key anything. Groups B and C build on this branch as stacked PRs. + +### One observable change did come with Group A: publish-span duration + +Moving the `using` out of the inner `try` was necessary to record failures at all, but it also put the `finally` inside the activity's scope - and the `finally` is where the publisher confirmation is awaited. The span therefore now covers the full publish-and-confirm round trip rather than just handing frames to the socket. + +**Measured**, 300 warm iterations with publisher confirmations and tracking enabled: + +| | span p50 | span p95 | span/wall p50 ratio | +|---|---|---|---| +| `main` | 37us | 117us | 0.047 | +| Group A | 372us | 728us | 0.960 | + +The span went from ~5% of the wall-clock publish call to ~96% of it - about a 10x increase in reported duration. Nothing became slower: wall-clock publish time was the same order in both runs (p50 776us on `main`, 388us on Group A), and that gap is run-to-run noise between two separate probe processes, not a speedup. + +End-to-end is the more useful semantic, and it is what a consumer would expect a `publish` span to cover when confirmations are enabled. But it is a visible change and belongs in the 7.3.0 release notes: publish-latency dashboards keyed on this span will jump by roughly an order of magnitude on upgrade. + +An earlier version of this document described this as only "the span now starts before flow control, so its duration includes any flow-control blocking." That understated it considerably - the earlier start is the minor half, the confirmation await is the dominant one. + +## The three activity sources + +| Source | Spans | Created in | +|---|---|---| +| `RabbitMQ.Client.Connection` | `connection attempt`, `tcp connection attempt` | `ConnectionFactory`, `AutorecoveringConnection`, `IEndpointResolverExtensions` | +| `RabbitMQ.Client.Publisher` | `publish` | `Channel.BasicPublish.cs` | +| `RabbitMQ.Client.Subscriber` | `fetch`, `fetch (empty)`, `deliver` | `Channel.cs` (`BasicGetAsync`), `AsyncConsumerDispatcher` | + +`ConnectionSourceName` is the only tracing member still in `PublicAPI.Unshipped.txt`. Everything else - `TracingOptions`, `ContextInjector`, `ContextExtractor`, `UseRoutingKeyAsOperationName`, and all of `RabbitMQTracingOptions` - shipped in 7.2.1. + +## Guard pattern: the shape that matters + +Each activity factory tests `HasListeners()` on **its own** source before creating anything, then tests `IsAllDataRequested` before setting tags. That part is correct and consistent. + +The subtle part is `SetNetworkTags`. It has *no* listener check of its own: + +```csharp +// projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs +internal static void SetNetworkTags(this Activity? activity, IFrameHandler frameHandler) +{ + if (activity?.IsAllDataRequested ?? false) + { +``` + +In 7.2.1 that check was `PublisherHasListeners && activity != null && activity.IsAllDataRequested`. It was moved out to the call site in `Connection.WriteAsync` deliberately: connection spans are created when the `Publisher` source may have no listeners at all, so a publisher gate inside the helper would have silently dropped network tags from every connection span. The three connection-side callers (`ConnectionFactory.cs:572`, `AutorecoveringConnection.cs:98`, `AutorecoveringConnection.Recovery.cs:263`) therefore call it directly on a known-owned activity, and only `Connection.WriteAsync` retains a `PublisherHasListeners` test. + +That is why `TestCreateConnectionRegisterAnActivity` passes while subscribing to `RabbitMQ.Client.Connection` alone. It is working by design, not by luck. Do not "restore" the publisher gate inside `SetNetworkTags`. + +Group A kept that shape. The publisher gate now lives in `SetNetworkTagsOnAmbientPublisherActivity`, a separate wrapper for the one ambient caller, so `SetNetworkTags` itself stays usable by the connection sites on a known-owned activity. + +Group A also made `OpenTcpConnection` take its `AmqpTcpEndpoint` and set the server tags itself. It was the only activity factory returning a bare activity and relying on its call site to tag it, which made it the one place a new caller could silently produce an untagged span. Covered by `TestTcpConnectionActivityHasServerTags_GH1967`. + +## Defect: ambient-span pollution (`Activity.Current`) - FIXED + +There are exactly three `Activity.Current` reads in the client: + +``` +projects/RabbitMQ.Client/Impl/SessionBase.cs:133 PopulateMessageEnvelopeSize(Activity.Current, bytes.Size) +projects/RabbitMQ.Client/Impl/SessionBase.cs:160 PopulateMessageEnvelopeSize(Activity.Current, bytes.Size) +projects/RabbitMQ.Client/Impl/Connection.cs:558 Activity.Current.SetNetworkTags(_frameHandler) +``` + +None of them checks whether the ambient activity belongs to this library. The intent is to decorate the `publish` span, which *is* `Activity.Current` at the moment its frames are transmitted. But `SessionBase.TransmitAsync` is on the path of **every** AMQP method, not just `basic.publish`. + +**Verified.** With the `Publisher` source listened and an application-owned `ActivitySource("MyApp")` span current, each of `QueueDeclarePassiveAsync`, `ExchangeDeclareAsync`, `BasicQosAsync`, `QueueBindAsync`, `BasicGetAsync`, and `BasicAckAsync` wrote ten tags onto the caller's span: + +``` +messaging.message.envelope.size +network.type +server.address, server.port +network.peer.address, network.peer.port +client.address, client.port +network.local.address, network.local.port +``` + +The same mechanism puts `messaging.message.envelope.size` on the client's own `connection attempt` span, since that span is current while the handshake frames go out - a messaging attribute on a connection span. + +With no listener on the `Publisher` source the caller's span stays clean, so the blast radius is exactly "applications that enable publisher tracing", which is to say all of them. + +This is **pre-existing**, not a regression from the connection-tracing work. `v7.2.1` already had `Activity.Current.SetNetworkTags(_frameHandler)` unconditionally in `Connection.WriteAsync` and both `Activity.Current` reads in `SessionBase`. + +The fix has to distinguish "this is my span" from "this is the caller's span". Passing the owned activity down from `Channel.BasicPublish.cs` is the direct route; checking `activity.Source` against the client's sources is the cheaper one. + +**Fixed** by the cheaper route: `RabbitMQActivitySource.IsPublisherActivity` tests `ReferenceEquals(activity.Source, s_publisherSource)`, and the two ambient call sites now go through `SetNetworkTagsOnAmbientPublisherActivity` / `PopulateMessageEnvelopeSizeOnAmbientPublisherActivity`, which read `Activity.Current` internally so the cheap `HasListeners()` test guards the `AsyncLocal` read. + +The check is against the **publisher source specifically**, not "any activity from this library". That distinction is load-bearing: the connection spans are ours too, but they are not publish operations, so gating on the library as a whole would leave `connection attempt` still carrying `messaging.message.envelope.size` from the handshake frames. Connection spans keep getting their network tags from the direct `SetNetworkTags` calls described above. + +Regression coverage: `TestAmqpOperationsDoNotTagAnUnrelatedAmbientActivity_GH1967` (asserts 11 absent tags on an app-owned span, then asserts the library's own `publish` span still gets them - this is an ownership check, not a blanket removal) and `TestConnectionActivityHasNoMessagingEnvelopeSize_GH1967`. + +## Defect: failed operations never record an error - FIXED + +Activity **disposal** is correct everywhere - every creation site uses `using`, including the error paths through `BasicPublishCoreAsync`. + +Activity **status** is not. No publisher or subscriber span ever records an error, because the `catch` blocks sit outside the activity's `using` scope: + +- `AsyncConsumerDispatcher.cs` - `using (Activity? activity = ...Deliver(...))` closes at line 39; the `catch (Exception e)` that reports to `CallbackExceptionAsync` is at line 59. +- `Channel.BasicPublish.cs` - `using Activity? sendActivity` at line 107 is scoped to the inner `try`; the `catch (Exception ex)` at line 126 cannot see it. + +**Verified.** A mandatory publish to an exchange with no matching queue raises `PublishReturnException` to the caller, and its span ends `status=Unset`, `StatusDescription=null`, zero events. A consumer `ReceivedAsync` handler that throws `InvalidOperationException` is reported through `CallbackExceptionAsync`, and its `deliver` span ends `status=Unset` with zero events. + +The connection spans get this right - `IEndpointResolverExtensions.cs:40-95` calls both `AddException` and `SetStatus(ActivityStatusCode.Error)` - so the inconsistency is internal to one implementation. + +Related: `error.type` is **Stable** in the RabbitMQ semantic convention and is Conditionally Required "if and only if the messaging operation has failed". The client sets it nowhere. + +**Fixed** via a single `SetActivityError(this Activity?, Exception)` helper that sets all three of the exception event, `SetStatus(Error)`, and `error.type`. All three are needed: a tracing backend reads an unset status as success, so a span carrying only an exception event still counts as a successful operation in error-rate queries. The helper also replaced the five hand-rolled `AddException` + `SetStatus` pairs on the connection spans, so publisher, subscriber and connection spans now report failures uniformly. + +This set is not just defensible, it is what the spec prescribes. `docs/general/recording-errors.md` is the governing document, and `messaging-spans.md` defers to it ("Span status SHOULD follow the Recording Errors document"). Verbatim from `recording-errors.md`: + +> [Span Status Code] MUST be left unset if the instrumented operation has ended without any errors. +> +> When the operation ends with an error, instrumentation: +> +> - SHOULD set the span status code to `Error` +> - SHOULD set the `error.type` attribute +> - SHOULD set the span status description when it has additional information about the error [...] When the operation fails with an exception, the span status description SHOULD be set to the exception message. + +`SetActivityError` does exactly those three, including passing `exception.Message` as the status description. The MUST-leave-unset clause is also why the helper is only ever called from failure paths, and why nothing sets a status on success. + +Two subtleties found while fixing this: + +- **The publish failure surfaces from the `finally`, not the `try`.** `PublishReturnException` for an unroutable mandatory publish comes out of `MaybeEndPublisherConfirmationTrackingAsync` via `MaybeWaitForConfirmationAsync`, which runs in the `finally` because the confirmation is only awaited once the send has been issued. Moving the `using` out of the inner try and adding a `catch` is therefore *not* sufficient on its own - the confirmation await needs its own try/catch. This was the primary verified case, so a fix without it would have looked complete and covered nothing. +- **A handled exception still marks the span `Error`.** When `MaybeHandleExceptionWithEnabledPublisherConfirmations` returns `true`, `SetActivityError` has already run. "Handled" there does not mean swallowed: the exception is routed onto the publisher confirmation task, and the `finally` awaits that task and re-raises the same instance to the caller. The publish did fail; it was reported through the confirmation channel instead of a throw from the send site. Revisit if that proves noisy in practice. + + There is spec text cutting against this, and it should be argued rather than ignored. `recording-errors.md` says "Errors that were retried or handled (allowing an operation to complete gracefully) SHOULD NOT be recorded on spans or metrics that describe this operation", and the deprecation of `exception.escaped` rests on the same reasoning ("It's no longer recommended to record exceptions that are handled and do not escape the scope of a span"). The argument for keeping the `Error` status is the parenthetical: the operation did *not* complete gracefully. A nacked or returned publish is a failed publish, and the caller learns about it through the confirmation task rather than through a throw from `BasicPublishAsync`. That is unlike the spec's own example, where `ResourceAlreadyExistsException` means the resource exists and `createIfNotExists` succeeded in its contract. The failure here is real and the reporting channel is the only thing that differs, so `error.type` on the span is the accurate signal. So this is arguably not the exemption case at all: the invariant is that every path which records `Error` is one the caller observes as a failure, and routing the exception through the confirmation task changes the channel, not the outcome. A conservative reading rather than a conformance defect, and worth revisiting only if it proves noisy. + +### Caller-initiated cancellation is not an error + +Contributed by @danielmarbach in #1982 and merged into the Group A branch. When the caller's own token is what cancelled the operation, the span rethrows without a status, exception event, or `error.type`. His argument was a comparison with ASP.NET Core's hosting layer, which treats every exception identically because it has no token to consult; this client does have the token, so it can make the distinction. + +The specification independently supports this, from a source neither of us cited at the time. `exceptions-logs.md` assigns DEBUG severity to "exceptions that don't indicate an actual issue", and its worked example is exactly this case: "an exception indicating that a request was cancelled on the client side". An operation the caller cancelled on purpose is not a failure of the operation. + +Note this is the *opposite* judgment from the handled-exception call above, and the two are consistent: cancellation means no failure occurred, while a nack means a failure occurred and was reported through a different channel. + +### The two catches recorded one failure twice + +The first version of the fix, having established that both the inner `catch` and the `finally` need to record, then double-recorded on a common path. Publishing on a **closed connection** with confirmations and tracking both enabled produces `events=2`, the same `AlreadyClosedException` twice, because: + +1. The send throws `AlreadyClosedException`; the inner `catch` records it. +2. `MaybeHandleException` calls `_publisherConfirmationTcs.SetException(ex)` and returns `true`, so the publish counts as handled and does *not* rethrow there. +3. The `finally` awaits that same TCS, which re-raises the identical exception instance (`TaskCompletionSource` rethrows through `ExceptionDispatchInfo`), and the `finally`'s `catch` records it again. + +**Verified.** Closed-connection publish gives `events=2` with `AlreadyClosedException` twice; an unroutable mandatory publish correctly gives `events=1`, because there the exception originates in the confirmation await and the inner catch never sees it. + +Both paths therefore have to stay, but the duplicate has to be suppressed. **Fixed** by tracking the exception the inner catch recorded and comparing by reference in the `finally`: `ReferenceEquals` is exactly right here because the TCS re-raises the same instance, so it suppresses the duplicate while still recording a genuinely different exception from the confirmation await. + +The tests did not catch this, because `ActivityAssert.HasRecordedException` only inspected `Events.First()`. `HasRecordedException` now asserts `Assert.Single` over the `"exception"` events before checking the type, so all five failure tests assert the count, not just the first event. + +### `SetActivityError` gated its three signals inconsistently + +`AddException` and `SetStatus` were unconditional while `error.type` sat behind `IsAllDataRequested`. **Verified** with a listener sampling `PropagationData`: the span got the exception event and the `Error` status but no `error.type`. A `PropagationData` span (`IsAllDataRequested=false`, `Recorded=false`) is still delivered to `ActivityStopped` with those signals intact - sampling suppresses neither tags nor exception events - so the gate dropped `error.type` from a span the listener did receive. + +`error.type` is Stable in the messaging convention and is what error-rate queries key off, so dropping it is the wrong signal to lose, and it is the cheap one: `AddException` allocates an `ActivityEvent` with a tag list, while `error.type` is one string already in hand. **Fixed** by making all three unconditional, gated only on `activity is null` - they now stay consistent across sampling levels, and the whole helper is skipped upstream (via `HasListeners` / `IsAllDataRequested` at the call sites) when nothing is recording. + +Note that `error.type` is not the *only* Stable attribute in the messaging convention, as an earlier draft of this document and of the code comment claimed. The producer-span table in `messaging-spans.md` also marks `server.address` and `server.port` Stable, and the consumer tables add `network.peer.address` and `network.peer.port`. The conclusion is unchanged - `error.type` is Stable, is `Conditionally Required` "if and only if the messaging operation has failed", and was set nowhere before this work - but the "only Stable attribute" phrasing was wrong and is corrected here and in `RabbitMQActivitySource`. + +Regression coverage: `TestPublishFailureIsRecordedOnTheSendActivity_GH1967` (mandatory publish to an exchange with no matching binding, which specifically exercises the `finally` path), `TestPublishFailureIsRecordedOnceWhenHandledByConfirmations_GH1967` (closed-connection publish, the duplicate-event path - requires *both* confirmations and tracking, or the exception is never handled and never resurfaces), and `TestConsumerFailureIsRecordedOnTheDeliverActivity_GH1967`. All go through `ActivityAssert.RecordsFailure`, which asserts all three signals plus the event count. + +## Defect: tracing configuration is process-global, last writer wins + +`RabbitMQActivitySource.TracingOptions` is a public settable static holding a mutable object, and `AddRabbitMQInstrumentation` replaces it wholesale while also overwriting both propagation delegates: + +```csharp +// projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs +RabbitMQActivitySource.TracingOptions = options; +RabbitMQActivitySource.ContextExtractor = OpenTelemetryContextExtractor; +RabbitMQActivitySource.ContextInjector = OpenTelemetryContextInjector; +``` + +**Verified.** Two independent `TracerProvider`s, each calling `AddRabbitMQInstrumentation` with different options: after the second call the first provider's configuration is silently gone, and *both* exporters receive spans shaped by the second (named `publish` / `fetch`, with no routing key). Disposing the second provider restores nothing - `ContextInjector` stays pointed at the OpenTelemetry implementation and `TracingOptions` keeps its values for the life of the process. + +This is not a memory-safety problem. 181,425 publishes with a concurrent writer swapping the options object produced zero exceptions, because reference assignment is atomic. The defect is the ownership model: per-provider configuration expressed as process-global mutable state. + +The statics are also unvalidated, so `ContextInjector = null` makes every subsequent publish throw `NullReferenceException` from inside the client. + +Because these members shipped in 7.2.1, removing them is a breaking change. Adding a per-provider path alongside them is not. + +## Exception events are on a deprecation path + +Raised by @tmasternak on #1978 and verified against the raw specification markdown, not the rendered site. + +`docs/exceptions/exceptions-spans.md` is marked **Status: Deprecated**, directing readers to `exceptions-logs.md`. `exception.escaped` is deprecated outright. Both documents carry the same normative migration block for existing instrumentations that record exceptions as span events: + +> - SHOULD introduce an environment variable `OTEL_SEMCONV_EXCEPTION_SIGNAL_OPT_IN` supporting the following values: +> - `logs` - emit exceptions as logs only. +> - `logs/dup` - emit both span events and logs, allowing for a phased rollout. +> - The default behavior (in the absence of one of these values) is to continue emitting exceptions as span events (existing behavior). +> - SHOULD maintain (security patching at a minimum) their existing major version for at least six months after it starts emitting both sets of conventions. +> - MAY drop the environment variable in their next major version and emit exceptions as logs only. + +Scope of the impact on this client: of the three signals `SetActivityError` sets, only `AddException` is affected. The `Error` status and `error.type` are prescribed by `recording-errors.md` and are unaffected. The deprecation is of the *recording API*, not of the ability to view events on spans - the SDK is expected to offer routing log-based events back onto spans. + +Nothing in the .NET stack supports this yet, verified rather than assumed: + +- On `System.Diagnostics.DiagnosticSource` 9.0.4, which this client references, none of `Activity.AddException`, `Activity.AddEvent` or `Activity.SetStatus` carries `[Obsolete]` or an experimental attribute. `RecordException` does not exist in .NET. +- `OTEL_SEMCONV_EXCEPTION_SIGNAL_OPT_IN` appears in no OpenTelemetry .NET assembly, including `OpenTelemetry.Api` 1.17.0, the current release. +- `OpenTelemetry.Instrumentation.AspNetCore` 1.17.0 still records exceptions via `Activity.AddException` behind its `RecordException` option. + +So 7.3.0 keeps span events. `RabbitMQ.Client` 7.x is a stable major version and the guidance for those is to stay behaviorally compatible for now. + +### Where the migration will actually be difficult + +Two consequences, the second of which is a real defect in this client: + +**No logging seam in the core.** `RabbitMQ.Client` has no `ILogger` dependency and logs through `EventSource`, which is not an OTel Logs API bridge. The core does, however, already have the right pattern: `RabbitMQActivitySource.ContextExtractor` and `ContextInjector` are settable static delegates the core calls and `AddRabbitMQInstrumentation` populates. An exception-recording delegate would follow that precedent, keeping the core free of a logging abstraction and letting the OpenTelemetry package decide between a span event and a `LogRecord`. That shape is easier to settle before `RabbitMQ.Client.OpenTelemetry` ships 1.0.0 (#1728) than after. + +**The consumer callback exception is reported outside the `deliver` span's scope.** `exceptions-logs.md` states that "Exception events emitted by instrumentations that also record spans for the same operation MUST be associated with the corresponding span context." In `AsyncConsumerDispatcher`, the `deliver` activity's `using` scope closes at line 55, but `OnCallbackExceptionAsync` - the user-facing report of that same exception - fires at line 77, outside it. **Verified** with a probe reproducing that exact nesting: at the `OnCallbackExceptionAsync` call site, `Activity.Current` has reverted to the enclosing activity, and the `deliver` span is already stopped. + +This costs nothing today, because the span event is added inside the scope at line 52 while the activity is still current. It becomes a MUST violation the moment exceptions are emitted as log records, because an application logging from its `CallbackException` handler would stamp the wrong `SpanId`, or none. The fix is to move the reporting `catch` inside the activity scope. Deliberately *not* done in the Group A PR: it is a scope change with no present-day symptom, so it belongs with the migration work and its own test. + +Tracked in #1992. + +## Semantic-convention gaps + +Checked against the specification at `main`: `model/messaging/registry.yaml`, `docs/messaging/messaging-spans.md`, `docs/messaging/rabbitmq.md`, `docs/general/recording-errors.md`, `docs/exceptions/exceptions-spans.md`, `docs/exceptions/exceptions-logs.md`. + +Stability context: every `messaging.*` attribute is **Development**, none is Stable. The Stable attributes appearing in the messaging span tables are all borrowed from other registries: `error.type`, `server.address`, `server.port`, and on the consumer tables `network.peer.address` and `network.peer.port`. So `messaging.*`-level changes are low-risk from the specification's own standpoint, and the client's attribute-name constants are `internal`. + +### Span kind for `receive` + +`messaging-spans.md` maps operation types to span kinds: + +| Operation type | Span kind | +|---|---| +| `create` | `PRODUCER` | +| `send` | `PRODUCER` if the send span's context is the creation context, otherwise `CLIENT` | +| `receive` | `CLIENT` | +| `process` | `CONSUMER` | +| `settle` | `CLIENT` | + +`BasicGet` and `BasicGetEmpty` both set `messaging.operation.type = receive` with `ActivityKind.Consumer`. They should be `ActivityKind.Client`. `Deliver` (`process` -> `Consumer`) and `BasicPublish` (`send` -> `Producer`, and its context is what gets injected) are both correct. + +### `messaging.rabbitmq.delivery_tag` is not a registry attribute + +The client emits `messaging.rabbitmq.delivery_tag`. The registry defines `messaging.rabbitmq.message.delivery_tag`. The emitted name matches nothing in the convention, so any consumer keying off it drops the value. + +### `messaging.destination.name` does not follow the RabbitMQ convention + +`rabbitmq.md` note [1] specifies `{exchange}:{routing key}` on the producer side when both are present and non-empty, only the available one when just one is, and `amq.default` only when the default exchange is used *and* no routing key is provided. The consumer side is `{exchange}:{routing key}:{queue}`. + +The client sets the bare exchange name, or the literal `amq.default` whenever the exchange is empty regardless of routing key. + +`BasicGetEmpty` is worse than non-conforming - it is wrong. It hardcodes `amq.default` even though the queue is known. **Verified** with a named exchange `probe-ex`, routing key `warning`, queue `probe-q`: + +``` +span "publish warning" kind=Producer messaging.destination.name = probe-ex + (convention: probe-ex:warning) +span "fetch warning" kind=Consumer messaging.destination.name = probe-ex + (convention: probe-ex:warning:probe-q) +span "fetch (empty) probe-q" kind=Consumer messaging.destination.name = amq.default + (the fetch never touched the default exchange) +error.type absent on all three. +``` + +### Span names use the routing key, not `{destination}` + +The convention is `{messaging.operation.name} {destination}`, where `{destination}` prefers `messaging.destination.template`, then `messaging.destination.name`, then `server.address:server.port`. The client appends the routing key. For server-named queues that also makes the span name high-cardinality, which the guidance on temporary and anonymous destinations warns against specifically. + +### `fetch (empty)` is not a valid operation name + +`messaging.operation.name = "fetch (empty)"` encodes an outcome into the operation name. `rabbitmq.md` gives `receive` and `poll` as receive-span examples. An empty result is representable without a distinct operation name. + +### `messaging.message.envelope.size` and `body.size` are Opt-In + +Opt-In means "SHOULD NOT be collected by default". The client always emits both when sampling. Defensible for a client library, but worth knowing. + +## Context propagation: no defects found + +**Verified** against a live broker with the `Publisher` source deliberately left unlistened so hand-planted headers survive the injector. Every one of these produced an unparented `fetch` span with zero links and no exception: no headers at all, an unrelated header only, a malformed `traceparent` as bytes, `traceparent = null`, `traceparent` as an `int`, `tracestate` with no `traceparent`, an empty-string `traceparent`, and a legacy `Request-Id` only. A well-formed `traceparent` parsed correctly as `byte[]` and as `string`, producing both a parent and a link. + +`DefaultContextGetter` handles only `byte[]`, which is *not* a defect: the broker returns header values as `byte[]` on the wire. It would only matter for an in-process carrier, which does not arise. + +Two cosmetic notes, both **fixed** in Group A: + +- `OpenTelemetryContextExtractor` passed `props.Headers` to `Propagators.DefaultTextMapPropagator.Extract` with no null check, unlike `DefaultContextExtractor` which returns early. With null headers the getter dereferenced null once per propagator field; the outcome was correct only because `catch (Exception)` swallowed it. That blanket catch was load-bearing rather than defensive, and its logger line was commented out, so a genuine extraction failure was silent. Now: an early return for null `Headers`, a `carrier != null` guard in the getter, and a comment explaining that the catch is now defensive-only (a custom `IDictionary` reaching the header table could still throw from `TryGetValue`, and a failed extraction must not fail the delivery). No test - the pre-existing catch already produced the right outcome, so there is no observable behaviour to assert. +- `DefaultContextSetter`'s comment said "Only propagate headers if they haven't already been set"; the code assigns unconditionally. The overwrite is the right behaviour - the comment was wrong, and now says so. + +## Consumer concurrency: no defects found + +**Verified** with 40 messages, `ConsumerDispatchConcurrency = 8` on both the factory and the channel, and a random 1-15 ms delay inside each callback to force interleaving: + +``` +publish spans 40, deliver spans 40 +deliver spans parented to a span in the publish set 40/40 +distinct deliver parent ids 40 +distinct deliver trace ids 40 +parent ids shared by more than one deliver span 0 +Activity.Current inside the callback was that message's own deliver span 40/40 +``` + +No mis-parenting, no context bleed across dispatch slots. + +## Test coverage gaps + +`TestActivitySource.cs` and `TestOpenTelemetry.cs` (4 tests) both live in `SequentialIntegration`, because `ActivityRecorder` and the activity sources are process-global. `TestOpenTelemetry.cs` drives the real SDK (`Sdk.CreateTracerProviderBuilder`, `AddRabbitMQInstrumentation`, `AddInMemoryExporter`); `TestActivitySource.cs` uses a bare `ActivityListener`. + +Group A closed the first two gaps, adding six `_GH1967` tests to `TestActivitySource.cs`, one to `TestOpenTelemetry.cs`, and the `ActivityAssert.RecordsFailure` / `HasRecordedExceptionOnce` helpers: + +- ~~**Error status** is asserted only on connection spans.~~ Now asserted on `publish` and `deliver`. +- ~~**Ambient-span pollution** has no coverage at all. `ActivityAssert.HasNoTag` exists and is used nowhere.~~ `HasNoTag` is now used for 11 tags in the ambient test, for three more after a publish in the same ambient scope, and for `messaging.message.envelope.size` on the connection span. + +Three gaps found by the review of the Group A branch itself, all now closed: + +- **Exception event *count* was unasserted,** so the duplicate-recording defect above went unnoticed. `HasRecordedExceptionOnce` fixes this; see that section. +- **Span parenting was unasserted.** All the new recorders set `VerifyParent = false`, which is unavoidable through the recorder - `ExpectedParent` has to be set before the recorder sees anything, and the ambient activity does not exist that early. `TestAmqpOperationsDoNotTagAnUnrelatedAmbientActivity_GH1967` now asserts `Assert.Same(appActivity, publishActivity.Parent)` directly instead, so scoping the tags to the publisher source cannot silently detach the publish span from the caller's trace. A detached span would show `Parent is null`, so the assertion is not vacuous. +- **The null-`Headers` extractor guard had no test.** `TestContextExtractorHandlesPropertiesWithNoHeaders_GH1967` pins the observable contract (no headers extracts to `default`, without throwing). Note this test would also have passed *before* the fix, because swallowing the `NullReferenceException` reached the same result. What it protects is the outcome if someone later narrows or removes that blanket `catch`, which the fix makes safe to do. + +Still open, and both belong to Group B: + +- **Two assertions lock in the span-kind gap.** `TestOpenTelemetry.cs` and `TestActivitySource.cs` both assert `ActivityKind.Consumer` for the `fetch` span. Fixing the span kind requires updating them. +- **One assertion locks in the destination gap.** `TestActivitySource.cs` asserts `messaging.destination.name == "amq.default"` for the default-exchange case. + +`ActivityRecorder.ShouldListenTo` is an exact source-name match, so a recorder constructed with `ConnectionSourceName` cannot see publisher or subscriber spans. Keep that in mind when reasoning about which tests would catch which regression. + +### `ActivityRecorder` matches on span name, and the routing key is in it by default + +`UseRoutingKeyAsOperationName` defaults to **`true`**, so a publish span is named `publish `, not `publish`. `ActivityRecorder` matches `activity.OperationName` exactly, so a recorder built for `"publish"` records **zero** activities under the default configuration and fails with `Expected: 1 / Actual: 0` - no hint that the name is the problem. + +Three of the five new tests hit this. Any new test that constructs a recorder with a bare operation name needs `TestActivitySource.PlainOperationNames`, a `using` scope that sets the flag false and **restores the previous value on dispose**. + +The restore matters beyond politeness. None of the pre-existing tests restore this flag or `TracingOptions` after mutating them, so test outcomes depend on execution order, and `TestOpenTelemetry.TestDefaultTracingOptions` asserts the default is `true` - it would fail if an earlier test left it `false`. That is Group C's defect (process-global configuration, last writer wins) reproducing inside our own test suite, which is a reasonable argument for fixing it. + +## Documentation gap + +`messaging-spans.md` states that an instrumentation using the message creation context as the parent of `process` spans SHOULD document that it does so, and MAY offer a configuration option. This client does exactly that by default (`UsePublisherAsParent = true`) and the option exists, but the `RabbitMQ.Client.OpenTelemetry` README documents only SDK wiring - not the trace structure, the span names, the attributes emitted, or either option. + +## What was checked and found clean + +- Activity disposal on every path, including the error branches through `BasicPublishCoreAsync`. +- `HasListeners()` / `IsAllDataRequested` guard pairing at all activity-creation sites. +- Exception recording on the connection spans (`IEndpointResolverExtensions.cs` sets both the event and the status, and deliberately leaves the parent connection activity alone when a later endpoint succeeds - Group A preserved that, routing it through `SetActivityError` and picking up `error.type` in the process). +- Public API surface parity between `net8.0` and `netstandard2.0`. +- `RabbitMQ.Client.OpenTelemetry` packaging: TFMs, signing, SourceLink, and the `otel-` MinVer prefix for independent versioning. +- The `OpenTelemetry.Api` 1.15.3 pin, which is the oldest version without GHSA-g94r-2vxg-569j. diff --git a/projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs b/projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs index 643c8f023..d8ef45e6c 100644 --- a/projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs +++ b/projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs @@ -30,6 +30,18 @@ public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProvid private static ActivityContext OpenTelemetryContextExtractor(IReadOnlyBasicProperties props) { + /* + * A message with no headers at all has nothing to extract. Returning early + * matters: without it the getter below is called once per propagator field + * with a null carrier, and the correct result depends entirely on its + * catch block swallowing a NullReferenceException. This mirrors the + * null check in RabbitMQActivitySource.DefaultContextExtractor. + */ + if (props.Headers is null) + { + return default; + } + // Extract the PropagationContext of the upstream parent from the message headers. var parentContext = Propagators.DefaultTextMapPropagator.Extract(default, props.Headers, OpenTelemetryContextGetter); Baggage.Current = parentContext.Baggage; @@ -38,16 +50,25 @@ private static ActivityContext OpenTelemetryContextExtractor(IReadOnlyBasicPrope private static IEnumerable OpenTelemetryContextGetter(IDictionary carrier, string key) { + /* + * Defensive only. The caller null-checks Headers, and a malformed value is + * handled by the `is byte[]` test rather than by throwing, so this catch is + * no longer load-bearing for any known input. It stays because a custom + * IDictionary implementation supplied through a header table could throw + * from TryGetValue, and a failed context extraction must not fail the + * delivery. + */ try { - if (carrier.TryGetValue(key, out object value) && value is byte[] bytes) + if (carrier != null && carrier.TryGetValue(key, out object value) && value is byte[] bytes) { return new[] { Encoding.UTF8.GetString(bytes) }; } } catch (Exception) { - //this.logger.LogError(ex, "Failed to extract trace context."); + // Ignored: an unparseable carrier yields an unparented span, which is + // strictly better than propagating the failure to the consumer. } return Enumerable.Empty(); diff --git a/projects/RabbitMQ.Client/ConnectionFactory.cs b/projects/RabbitMQ.Client/ConnectionFactory.cs index dd8519b6a..9147140b2 100644 --- a/projects/RabbitMQ.Client/ConnectionFactory.cs +++ b/projects/RabbitMQ.Client/ConnectionFactory.cs @@ -575,24 +575,20 @@ public async Task CreateConnectionAsync(IEndpointResolver endpointR .ConfigureAwait(false); } } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Caller-initiated cancellation is not a connection failure. + throw; + } catch (OperationCanceledException ex) { - connectionActivity?.SetStatus(ActivityStatusCode.Error); - connectionActivity?.AddException(ex); - if (cancellationToken.IsCancellationRequested) - { - throw; - } - else - { - throw new BrokerUnreachableException(ex); - } + connectionActivity.SetActivityError(ex); + throw new BrokerUnreachableException(ex); } catch (Exception ex) { var brokerUnreachableException = new BrokerUnreachableException(ex); - connectionActivity?.SetStatus(ActivityStatusCode.Error); - connectionActivity?.AddException(brokerUnreachableException); + connectionActivity.SetActivityError(brokerUnreachableException); throw brokerUnreachableException; } } diff --git a/projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs b/projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs index cfbf2c546..100875079 100644 --- a/projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs +++ b/projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs @@ -32,10 +32,26 @@ protected override async Task ProcessChannelAsync() using (Activity? activity = RabbitMQActivitySource.Deliver(work.RoutingKey!, work.Exchange!, work.DeliveryTag, work.BasicProperties!, work.Body.Size)) { - await work.Consumer.HandleBasicDeliverAsync( - work.ConsumerTag!, work.DeliveryTag, work.Redelivered, - work.Exchange!, work.RoutingKey!, work.BasicProperties!, work.Body.Memory, work.CancellationToken) - .ConfigureAwait(false); + /* + * Record a throwing consumer callback on the deliver span + * before rethrowing to the reporting catch below. Without + * this the span is disposed on the way out and ends + * status=Unset with no exception event, so a consumer that + * throws on every message still traces as fully + * successful. See issue #1967. + */ + try + { + await work.Consumer.HandleBasicDeliverAsync( + work.ConsumerTag!, work.DeliveryTag, work.Redelivered, + work.Exchange!, work.RoutingKey!, work.BasicProperties!, work.Body.Memory, work.CancellationToken) + .ConfigureAwait(false); + } + catch (Exception e) + { + activity.SetActivityError(e); + throw; + } } break; case WorkType.Cancel: diff --git a/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs b/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs index 1173e0d08..1b2f43c63 100644 --- a/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs +++ b/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs @@ -46,16 +46,17 @@ public static async Task SelectOneAsync(this IEndpointResolver resolver, foreach (AmqpTcpEndpoint ep in resolver.All()) { cancellationToken.ThrowIfCancellationRequested(); - using Activity? tcpConnection = RabbitMQActivitySource.OpenTcpConnection(); - if (tcpConnection is { IsAllDataRequested: true }) - { - tcpConnection.SetServerTags(ep); - } + using Activity? tcpConnection = RabbitMQActivitySource.OpenTcpConnection(ep); try { return await selector(ep, cancellationToken).ConfigureAwait(false); } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Caller-initiated cancellation is not a connection attempt failure. + throw; + } catch (OperationCanceledException ex) { /* @@ -67,19 +68,12 @@ public static async Task SelectOneAsync(this IEndpointResolver resolver, * later endpoint succeeds, the overall operation succeeded, and only the * individual attempt failed. */ - tcpConnection?.AddException(ex); - tcpConnection?.SetStatus(ActivityStatusCode.Error); - if (cancellationToken.IsCancellationRequested) - { - throw; - } - + tcpConnection.SetActivityError(ex); exceptions.Add(ex); } catch (Exception e) { - tcpConnection?.AddException(e); - tcpConnection?.SetStatus(ActivityStatusCode.Error); + tcpConnection.SetActivityError(e); exceptions.Add(e); } } diff --git a/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.Recovery.cs b/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.Recovery.cs index 5f1b44451..89958ef91 100644 --- a/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.Recovery.cs +++ b/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.Recovery.cs @@ -324,8 +324,7 @@ await maybeNewInnerConnection.OpenAsync(cancellationToken) } catch (Exception e) { - connectionActivity?.AddException(e); - connectionActivity?.SetStatus(ActivityStatusCode.Error); + connectionActivity.SetActivityError(e); ESLog.Error("Connection recovery exception.", e); // Trigger recovery error events if (!_connectionRecoveryErrorAsyncWrapper.IsEmpty) diff --git a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs index a14be961d..4c523fe9d 100644 --- a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs +++ b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs @@ -97,6 +97,26 @@ private async ValueTask BasicPublishCoreAsync( RateLimitLease? lease = await MaybeAcquirePublisherConfirmationLockAsync(cancellationToken) .ConfigureAwait(false); + /* + * The send activity is declared out here, rather than inside the try + * below, so the catch can record the failure on it. With the `using` + * scoped to the inner try the span was already disposed by the time + * the catch ran, so no publish failure was ever reported: the span + * ended status=Unset with no exception event, which tracing backends + * read as a successful publish. See issue #1967. + */ + using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners + ? RabbitMQActivitySource.BasicPublish(routingKey, exchange, body.Length, basicProperties) + : default; + /* + * Tracks the exception (if any) already recorded on sendActivity by the + * catch below, so the finally's confirmation-await catch does not record + * the same instance twice. When MaybeHandleExceptionWithEnabledPublisherConfirmations + * faults the confirm TCS, the finally's await re-raises that exception; + * without this guard a publish whose send failed (e.g. on a closed + * connection) recorded the same exception twice. See issue #1967. + */ + Exception? recordedSendError = null; try { publisherConfirmationInfo = MaybeStartPublisherConfirmationTracking(); @@ -104,10 +124,6 @@ await MaybeAcquirePublisherConfirmationLockAsync(cancellationToken) await MaybeEnforceFlowControlAsync(cancellationToken) .ConfigureAwait(false); - using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners - ? RabbitMQActivitySource.BasicPublish(routingKey, exchange, body.Length, basicProperties) - : default; - ulong publishSequenceNumber = publisherConfirmationInfo?.PublishSequenceNumber ?? 0; BasicProperties? props = PopulateBasicPropertiesHeaders(basicProperties, sendActivity, publishSequenceNumber); @@ -125,6 +141,34 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) } catch (Exception ex) { + /* + * Caller-initiated cancellation is not a publish failure, so it is + * not recorded on the span. Confirmation tracking still needs the + * cleanup below (faulting the TCS, decrementing the sequence number), + * which is why this is an inline guard rather than a `when` filter: + * a filter that skipped this catch would skip the cleanup too. + * See issue #1967. + */ + bool isCallerCancellation = + ex is OperationCanceledException && cancellationToken.IsCancellationRequested; + if (!isCallerCancellation) + { + sendActivity.SetActivityError(ex); + recordedSendError = ex; + } + + /* + * "Handled" here means the exception was routed onto the publisher + * confirmation task, not that it was swallowed: the finally below + * awaits that task and re-raises the same instance to the caller. So + * recording the error above is correct even when exceptionWasHandled + * is true - the publish failed and the caller sees it, just through + * the confirmation channel rather than a throw from here. This is not + * the spec's "handled or retried and completed gracefully" exemption, + * which is for operations that recover; a faulted publish never does. + * Every path that records Error is one the caller observes as a + * failure. See issue #1967. + */ bool exceptionWasHandled = MaybeHandleExceptionWithEnabledPublisherConfirmations(publisherConfirmationInfo, ex); if (!exceptionWasHandled) @@ -135,8 +179,30 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) finally { MaybeReleasePublisherConfirmationLock(lease); - await MaybeEndPublisherConfirmationTrackingAsync(publisherConfirmationInfo, cancellationToken) - .ConfigureAwait(false); + + /* + * This await is the one that surfaces a nack or an unroutable + * mandatory publish (PublishException), so it is a publish failure + * like any other and belongs on the span. It cannot simply be + * wrapped by the catch above, because it runs in the finally: the + * confirmation is only awaited once the send has been issued. + */ + try + { + await MaybeEndPublisherConfirmationTrackingAsync(publisherConfirmationInfo, cancellationToken) + .ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Caller-initiated cancellation during the confirmation await is + // not a publish failure. See issue #1967. + throw; + } + catch (Exception ex) when (!ReferenceEquals(ex, recordedSendError)) + { + sendActivity.SetActivityError(ex); + throw; + } } } finally diff --git a/projects/RabbitMQ.Client/Impl/Connection.cs b/projects/RabbitMQ.Client/Impl/Connection.cs index 20892898b..f80b229e9 100644 --- a/projects/RabbitMQ.Client/Impl/Connection.cs +++ b/projects/RabbitMQ.Client/Impl/Connection.cs @@ -31,7 +31,6 @@ using System; using System.Collections.Generic; -using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.IO; using System.Runtime.CompilerServices; @@ -553,10 +552,7 @@ internal Task OnCallbackExceptionAsync(CallbackExceptionEventArgs args) internal ValueTask WriteAsync(OutgoingFrame frames, CancellationToken cancellationToken) { - if (RabbitMQActivitySource.PublisherHasListeners) - { - Activity.Current.SetNetworkTags(_frameHandler); - } + RabbitMQActivitySource.SetNetworkTagsOnAmbientPublisherActivity(_frameHandler); return _frameHandler.WriteAsync(frames, cancellationToken); } diff --git a/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs b/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs index ce635dc93..27edc115f 100644 --- a/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs +++ b/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs @@ -33,6 +33,10 @@ public static class RabbitMQActivitySource internal const string ProtocolVersion = "network.protocol.version"; internal const string RabbitMQDeliveryTag = "messaging.rabbitmq.delivery_tag"; + // error.type is Stable in the messaging convention, and is Conditionally + // Required "if and only if the messaging operation has failed". + internal const string ErrorType = "error.type"; + // These constants are specific to this client - the OpenTelemetry messaging // conventions do not (yet) cover connection establishment. internal const string RabbitMQConnectionIsReconnection = "messaging.rabbitmq.connection.is_reconnection"; @@ -69,6 +73,27 @@ public static bool UseRoutingKeyAsOperationName public static RabbitMQTracingOptions TracingOptions { get; set; } = new RabbitMQTracingOptions(); internal static bool PublisherHasListeners => s_publisherSource.HasListeners(); + /* + * Both PopulateMessageEnvelopeSize and Connection.WriteAsync tag whatever + * Activity.Current happens to be, because the frame-writing path has no + * reference to the publish activity it belongs to. Without this check, any + * AMQP method issued inside an unrelated ambient activity stamps messaging + * and network tags onto a span this library does not own - an app's own + * span, or an ASP.NET request span, picking up server.address and friends + * from an incidental QueueDeclare. + * + * The test is publisher-source ownership specifically, not "any activity + * from this library". The connection spans are ours too, but they are not + * publish operations: gating on the library as a whole would leave the + * "connection attempt" span carrying messaging.message.envelope.size from + * the handshake frames. Connection spans get their network tags from the + * direct SetNetworkTags calls at the three connection call sites. + */ + private static bool IsPublisherActivity(Activity? activity) + { + return activity is not null && ReferenceEquals(activity.Source, s_publisherSource); + } + internal static readonly IEnumerable> CreationTags = new[] { new KeyValuePair(MessagingSystem, "rabbitmq"), @@ -93,14 +118,21 @@ public static bool UseRoutingKeyAsOperationName return connectionActivity; } - internal static Activity? OpenTcpConnection() + internal static Activity? OpenTcpConnection(AmqpTcpEndpoint endpoint) { if (!s_connectionSource.HasListeners()) { return null; } - return s_connectionSource.StartRabbitMQActivity("tcp connection attempt", ActivityKind.Client); + Activity? activity = + s_connectionSource.StartRabbitMQActivity("tcp connection attempt", ActivityKind.Client); + if (activity is { IsAllDataRequested: true }) + { + activity.SetServerTags(endpoint); + } + + return activity; } internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize, IReadOnlyBasicProperties basicProperties, @@ -250,14 +282,102 @@ private static void PopulateMessagingTags(string operationType, string operation } } - internal static void PopulateMessageEnvelopeSize(Activity? activity, int size) + /* + * As with SetNetworkTagsOnAmbientPublisherActivity, this reads Activity.Current + * itself so the cheap HasListeners() test guards the AsyncLocal read on a path + * that runs for every AMQP method transmitted. + */ + internal static void PopulateMessageEnvelopeSizeOnAmbientPublisherActivity(int size) { - if (activity != null && activity.IsAllDataRequested && PublisherHasListeners) + if (!PublisherHasListeners) + { + return; + } + + Activity? activity = Activity.Current; + if (activity != null && activity.IsAllDataRequested && IsPublisherActivity(activity)) { activity.SetTag(MessagingEnvelopeSize, size); } } + /* + * Tag the ambient activity from the frame-writing path. Unlike SetNetworkTags, + * this must not touch an activity this library does not own. See + * IsPublisherActivity. + * + * This reads Activity.Current itself rather than taking it as an argument: + * that is an AsyncLocal read on a path that runs for every frame written, so + * the cheap HasListeners() test comes first. With no publisher listeners the + * source cannot have produced the ambient activity anyway. + */ + internal static void SetNetworkTagsOnAmbientPublisherActivity(IFrameHandler frameHandler) + { + if (!PublisherHasListeners) + { + return; + } + + Activity? activity = Activity.Current; + if (IsPublisherActivity(activity)) + { + activity.SetNetworkTags(frameHandler); + } + } + + /* + * Record a failed messaging operation on its span. + * + * The OpenTelemetry "Recording errors" document prescribes exactly this set + * for an operation that ends with an error: set the span status code to + * Error, set error.type, and set the status description to the exception + * message when the failure is an exception. The messaging conventions defer + * to it ("Span status SHOULD follow the Recording Errors document"), and the + * status code MUST be left unset when the operation succeeded, which is why + * this is only called on failure paths. + * + * Tracing backends treat an unset status as success, so a span that merely + * carries an exception event still reads as a successful operation in + * error-rate queries. error.type is Stable in the messaging convention and + * is what error-rate queries key off; it is set to the fully-qualified + * exception type name, which is what the convention prescribes when there + * is no lower-cardinality domain-specific value to use. The connection spans + * use this same helper, so publisher, subscriber and connection spans report + * failures uniformly. + * + * All three signals fire together so they stay consistent across sampling + * levels, gated only on a null activity. AddException and SetStatus already + * execute when IsAllDataRequested is false - a listener sampling + * PropagationData still receives the event and the status - so gating + * error.type alone, as an earlier version of this helper did, recorded the + * expensive signals and dropped the cheap one that queries actually use. + * + * AddException is the only allocating signal (it builds an ActivityEvent) and + * fires even on a PropagationData-sampled span. That is deliberate, not an + * oversight: it is left unguarded rather than placed behind IsAllDataRequested + * because failure paths are not hot, so the allocation never lands on a hot + * path, and splitting it out would reintroduce the inconsistency above and turn + * the logs migration below into a three-site edit instead of one. + * + * The exception event is the one signal here on a deprecation path: the + * exceptions-on-spans convention is deprecated in favour of recording + * exceptions as log records, and Activity.AddException is expected to follow. + * The status and error.type are unaffected by that change. Keeping all three + * in one helper is what makes the eventual migration a single edit. See + * issues #1967 and #1992. + */ + internal static void SetActivityError(this Activity? activity, Exception exception) + { + if (activity is null) + { + return; + } + + activity.AddException(exception); + activity.SetStatus(ActivityStatusCode.Error, exception.Message); + activity.SetTag(ErrorType, exception.GetType().FullName); + } + internal static void SetNetworkTags(this Activity? activity, IFrameHandler frameHandler) { if (activity?.IsAllDataRequested ?? false) @@ -350,7 +470,11 @@ private static void DefaultContextSetter(object? carrier, string name, string va return; } - // Only propagate headers if they haven't already been set + /* + * Overwrite unconditionally. The client's own context is the authoritative + * one for the span it just created, so a caller-supplied traceparent in the + * same header table is replaced rather than preserved. + */ carrierDictionary[name] = value; } diff --git a/projects/RabbitMQ.Client/Impl/SessionBase.cs b/projects/RabbitMQ.Client/Impl/SessionBase.cs index 0248cdd62..d47db226d 100644 --- a/projects/RabbitMQ.Client/Impl/SessionBase.cs +++ b/projects/RabbitMQ.Client/Impl/SessionBase.cs @@ -31,7 +31,6 @@ using System; using System.Buffers; -using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Runtime.CompilerServices; using System.Threading; @@ -130,7 +129,7 @@ public virtual ValueTask TransmitAsync(in T cmd, CancellationToken cancellati } OutgoingFrame bytes = Framing.SerializeToFrames(ref Unsafe.AsRef(in cmd), ChannelNumber); - RabbitMQActivitySource.PopulateMessageEnvelopeSize(Activity.Current, bytes.Size); + RabbitMQActivitySource.PopulateMessageEnvelopeSizeOnAmbientPublisherActivity(bytes.Size); return Connection.WriteAsync(bytes, cancellationToken); } @@ -151,13 +150,14 @@ public ValueTask TransmitAsync(in TMethod cmd, in THeader head // // If SerializeToFrames throws, `bytes` is still the default OutgoingFrame (Size == 0); // we must dispose `bodyOwner` directly because it was never captured. - // If PopulateMessageEnvelopeSize or a synchronous fault inside Connection.WriteAsync - // throws, `bytes` already owns `bodyOwner`; disposing the frame releases both. + // If PopulateMessageEnvelopeSizeOnAmbientPublisherActivity or a synchronous + // fault inside Connection.WriteAsync throws, `bytes` already owns + // `bodyOwner`; disposing the frame releases both. OutgoingFrame bytes = default; try { bytes = Framing.SerializeToFrames(ref Unsafe.AsRef(in cmd), ref Unsafe.AsRef(in header), body, bodyOwner, ChannelNumber, Connection.MaxPayloadSize); - RabbitMQActivitySource.PopulateMessageEnvelopeSize(Activity.Current, bytes.Size); + RabbitMQActivitySource.PopulateMessageEnvelopeSizeOnAmbientPublisherActivity(bytes.Size); return Connection.WriteAsync(bytes, cancellationToken); } catch diff --git a/projects/Test/Common/ActivityRecorder.cs b/projects/Test/Common/ActivityRecorder.cs index b1ddaa2ac..b7e159fde 100644 --- a/projects/Test/Common/ActivityRecorder.cs +++ b/projects/Test/Common/ActivityRecorder.cs @@ -147,8 +147,16 @@ public static void HasRecordedException(this Activity activity, Exception except public static void HasRecordedException(this Activity activity, string exceptionTypeName) { - var exceptionEvent = activity.Events.First(); - Assert.Equal("exception", exceptionEvent.Name); + /* + * Assert exactly one exception event so duplicate recordings are caught. + * A publish whose send failed on a closed connection used to record the + * same exception twice (once in the send catch, once when the confirmation + * await re-raised it), which Events.First() alone does not detect. + * See issue #1967. + */ + var exceptionEvents = activity.Events.Where(e => e.Name == "exception").ToList(); + Assert.Single(exceptionEvents); + ActivityEvent exceptionEvent = exceptionEvents[0]; Assert.Equal(exceptionTypeName, exceptionEvent.Tags.SingleOrDefault(t => t.Key == "exception.type").Value); } @@ -158,6 +166,20 @@ public static void IsInError(this Activity activity) Assert.Equal(ActivityStatusCode.Error, activity.Status); } + /// + /// Assert that a failed operation is fully reported: exactly one exception + /// event, an Error status, and error.type. A tracing backend treats an unset + /// status as success, so all three are needed for the failure to be visible, + /// and one failure should appear once. + /// See rabbitmq/rabbitmq-dotnet-client#1967. + /// + public static void RecordsFailure(this Activity activity, Type exceptionType) + { + activity.HasRecordedException(exceptionType.ToString()); + activity.IsInError(); + activity.HasTag("error.type", exceptionType.FullName); + } + public static void HasNoTag(this Activity activity, string name) { bool contains = activity.TagObjects.Any(t => t.Key == name); diff --git a/projects/Test/SequentialIntegration/TestActivitySource.cs b/projects/Test/SequentialIntegration/TestActivitySource.cs index 5b4a4b2aa..7c9b7744d 100644 --- a/projects/Test/SequentialIntegration/TestActivitySource.cs +++ b/projects/Test/SequentialIntegration/TestActivitySource.cs @@ -34,6 +34,7 @@ using System.Diagnostics; using System.Linq; using System.Text; +using System.Threading; using System.Threading.Tasks; using RabbitMQ.Client; @@ -305,6 +306,357 @@ public async Task TestPublisherWithPublicationAddressAndBasicGetActivityTagsAsyn } } + /// + /// Scopes UseRoutingKeyAsOperationName to false, restoring it on dispose. + /// + /// + /// It defaults to true, which appends the routing key to the span name. The tests + /// below match spans by name via , so they need the + /// plain name. Restoring on dispose keeps this from becoming one more test that + /// leaves process-global tracing state mutated for whatever runs next - see the + /// public-API discussion on rabbitmq/rabbitmq-dotnet-client#1967. + /// + private sealed class PlainOperationNames : IDisposable + { + private readonly bool _previous; + + public PlainOperationNames() + { + _previous = RabbitMQActivitySource.UseRoutingKeyAsOperationName; + RabbitMQActivitySource.UseRoutingKeyAsOperationName = false; + } + + public void Dispose() => RabbitMQActivitySource.UseRoutingKeyAsOperationName = _previous; + } + + [Fact] + public async Task TestPublishFailureIsRecordedOnTheSendActivity_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * The `using Activity? sendActivity` in BasicPublishCoreAsync used to be + * scoped to the inner try, so both the catch and the confirmation await in + * the finally ran after the span was disposed. Publish failures were + * therefore never recorded: the span ended status=Unset with no exception + * event, which every tracing backend reads as a successful publish. + * + * A mandatory publish to an exchange with no matching binding is the + * cleanest trigger, and it specifically covers the finally path, since + * PublishException surfaces from the confirmation await rather than from + * the send itself. + */ + using var plainNames = new PlainOperationNames(); + + using ActivityRecorder publishRecorder = + new(RabbitMQActivitySource.PublisherSourceName, "publish"); + publishRecorder.VerifyParent = false; + + string exchange = $"exchange-{Guid.NewGuid()}"; + await _channel.ExchangeDeclareAsync(exchange, ExchangeType.Direct, autoDelete: true); + + try + { + await Assert.ThrowsAsync(() => + _channel.BasicPublishAsync(exchange, "no-such-routing-key", mandatory: true, + Encoding.UTF8.GetBytes("unroutable")).AsTask()); + + Activity publishActivity = publishRecorder.VerifyActivityRecordedOnce(); + publishActivity.RecordsFailure(typeof(PublishReturnException)); + } + finally + { + await _channel.ExchangeDeleteAsync(exchange); + } + } + + [Fact] + public async Task TestPublishFailureIsRecordedOnceWhenHandledByConfirmations_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * Recording the failure in both the inner catch and the one in the finally + * used to double-record it on this path. Publishing on a closed connection + * throws from the send, the inner catch hands the exception to the + * confirmation task (so the publish counts as handled and does not + * rethrow there), and awaiting that task in the finally re-raises the same + * instance - one failure, recorded twice. + * + * Publisher confirmations *and* tracking are both required to reproduce: + * without tracking there is no task to store the exception on, so it is + * never handled and never resurfaces. + */ + using var plainNames = new PlainOperationNames(); + + using ActivityRecorder publishRecorder = + new(RabbitMQActivitySource.PublisherSourceName, "publish"); + publishRecorder.VerifyParent = false; + + ConnectionFactory cf = CreateConnectionFactory(); + cf.AutomaticRecoveryEnabled = false; + + var channelOptions = new CreateChannelOptions( + publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true); + + await using IConnection conn = await cf.CreateConnectionAsync(); + await using IChannel ch = await conn.CreateChannelAsync(channelOptions); + + await conn.CloseAsync(); + + await Assert.ThrowsAsync(() => + ch.BasicPublishAsync("", "no-such-queue", true, + Encoding.UTF8.GetBytes("after close")).AsTask()); + + Activity publishActivity = publishRecorder.VerifyActivityRecordedOnce(); + publishActivity.RecordsFailure(typeof(AlreadyClosedException)); + } + + [Fact] + public async Task TestCallerCancellationIsNotRecordedAsPublishFailure_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * Once publish failures are recorded on the send activity, a publish the + * caller cancels is not a failure of the publish and must not be recorded + * as one. Without the guard the cancelled publish ended status=Error with a + * TaskCanceledException event, so an app cancelling its own publishes traced + * as a stream of publish errors. + * + * With confirmations enabled the publish parks awaiting the broker's + * confirmation, which BasicPublishCoreAsync awaits in its finally - after the + * send activity has been created. Blocking the connection holds off that + * confirmation, so cancelling the token throws OperationCanceledException + * from that await deterministically, with no dependence on broker timing. + * That is the window the finally's cancellation guard covers; removing it + * fails this test with an Error status on the span. + */ + using var plainNames = new PlainOperationNames(); + + using ActivityRecorder publishRecorder = + new(RabbitMQActivitySource.PublisherSourceName, "publish"); + publishRecorder.VerifyParent = false; + + using var cts = new CancellationTokenSource(); + + try + { + await BlockAsync(); + + ValueTask publishTask = _channel.BasicPublishAsync("", "no-such-queue", true, + Encoding.UTF8.GetBytes("cancel me"), cts.Token); + + // The publish is now parked in the confirmation await with its span open. + cts.Cancel(); + + await Assert.ThrowsAnyAsync(() => publishTask.AsTask()); + } + finally + { + await UnblockAsync(); + } + + Activity publishActivity = publishRecorder.VerifyActivityRecordedOnce(); + Assert.NotEqual(ActivityStatusCode.Error, publishActivity.Status); + Assert.DoesNotContain(publishActivity.Events, e => e.Name == "exception"); + publishActivity.HasNoTag("error.type"); + } + + [Fact] + public async Task TestConsumerFailureIsRecordedOnTheDeliverActivity_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * Same defect on the consume side: AsyncConsumerDispatcher's reporting + * catch sits outside the deliver activity's `using`, so a consumer callback + * that threw was reported via CallbackExceptionAsync but left the deliver + * span status=Unset with no exception event. A consumer failing on every + * message traced as completely healthy. + */ + using var plainNames = new PlainOperationNames(); + + using ActivityRecorder deliverRecorder = + new(RabbitMQActivitySource.SubscriberSourceName, "deliver"); + deliverRecorder.VerifyParent = false; + + string queue = $"queue-{Guid.NewGuid()}"; + await _channel.QueueDeclareAsync(queue, false, true, false, null); + + var callbackExceptionTcs = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _channel.CallbackExceptionAsync += (_, _) => + { + callbackExceptionTcs.TrySetResult(true); + return Task.CompletedTask; + }; + + var consumer = new AsyncEventingBasicConsumer(_channel); + consumer.ReceivedAsync += (_, _) => + throw new InvalidOperationException("consumer callback failed on purpose"); + + await _channel.BasicConsumeAsync(queue, autoAck: true, consumer: consumer); + await _channel.BasicPublishAsync("", queue, true, Encoding.UTF8.GetBytes("hi")); + + await callbackExceptionTcs.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + Activity deliverActivity = deliverRecorder.VerifyActivityRecordedOnce(); + deliverActivity.RecordsFailure(typeof(InvalidOperationException)); + } + + [Fact] + public async Task TestAmqpOperationsDoNotTagAnUnrelatedAmbientActivity_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * SessionBase.TransmitAsync and Connection.WriteAsync tag whatever + * Activity.Current happens to be, because the frame-writing path has no + * reference to the publish activity it belongs to. With no ownership check, + * any AMQP method issued inside an unrelated ambient activity stamped + * messaging and network tags onto a span this library does not own - an + * app's own span, or an ASP.NET request span, picking up server.address and + * messaging.message.envelope.size from an incidental QueueDeclare. + * + * The publisher source must have listeners for this to reproduce, which is + * what the recorder here provides. + */ + using var plainNames = new PlainOperationNames(); + + using ActivityRecorder publishRecorder = + new(RabbitMQActivitySource.PublisherSourceName, "publish"); + publishRecorder.VerifyParent = false; + + using var appSource = new ActivitySource("TestApp.GH1967"); + using var appListener = new ActivityListener + { + ShouldListenTo = source => source.Name == "TestApp.GH1967", + Sample = (ref ActivityCreationOptions _) => + ActivitySamplingResult.AllDataAndRecorded + }; + ActivitySource.AddActivityListener(appListener); + + string queue = $"queue-{Guid.NewGuid()}"; + + // Kept alive past the AMQP calls so the parenting assertion at the end can + // compare against it. + using Activity appActivity = appSource.StartActivity("app-operation"); + Assert.NotNull(appActivity); + Assert.Same(appActivity, Activity.Current); + + await _channel.QueueDeclareAsync(queue, false, true, false, null); + await _channel.QueueDeclarePassiveAsync(queue); + await _channel.BasicQosAsync(0, 1, false); + + /* + * Every tag either path would have written. The network tags come from + * Connection.WriteAsync, the envelope size from SessionBase. + */ + appActivity.HasNoTag("messaging.message.envelope.size"); + appActivity.HasNoTag("messaging.system"); + appActivity.HasNoTag("network.type"); + appActivity.HasNoTag("server.address"); + appActivity.HasNoTag("server.port"); + appActivity.HasNoTag("network.peer.address"); + appActivity.HasNoTag("network.peer.port"); + appActivity.HasNoTag("client.address"); + appActivity.HasNoTag("client.port"); + appActivity.HasNoTag("network.local.address"); + appActivity.HasNoTag("network.local.port"); + + /* + * Publish inside the same ambient scope. The library's own publish span must + * still get the tags - this is an ownership check, not a blanket removal - + * and the app's span must still come out clean, even though a publish is + * exactly the operation whose tags it was previously stealing. + */ + await _channel.BasicPublishAsync("", queue, true, Encoding.UTF8.GetBytes("hi")); + + appActivity.HasNoTag("messaging.message.envelope.size"); + appActivity.HasNoTag("server.port"); + appActivity.HasNoTag("network.peer.address"); + + Activity publishActivity = publishRecorder.VerifyActivityRecordedOnce(); + publishActivity.HasTag("messaging.message.envelope.size"); + publishActivity.HasTag("server.port"); + publishActivity.HasTag("network.peer.address"); + + /* + * Parenting is asserted here rather than through the recorder's + * VerifyParent, because ExpectedParent has to be set before the recorder + * sees anything and the ambient activity does not exist that early. The + * publish span is started while appActivity is current, so it must be its + * child: scoping the tags to the publisher source must not also detach the + * span from the caller's trace. See issue #1967. + */ + Assert.Same(appActivity, publishActivity.Parent); + } + + [Fact] + public async Task TestConnectionActivityHasNoMessagingEnvelopeSize_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * The client's own "connection attempt" span used to pick up + * messaging.message.envelope.size from the handshake frames, because it was + * Activity.Current while they were transmitted. It is a connection span, + * not a publish operation, so a messaging attribute has no business on it. + * + * This is why the ownership check tests the publisher source specifically + * rather than "any activity from this library". + */ + using ActivityRecorder connectionRecorder = + new(RabbitMQActivitySource.ConnectionSourceName, "connection attempt"); + connectionRecorder.VerifyParent = false; + + // The publisher source must have listeners, or the tagging path is skipped + // entirely and the test would pass without exercising the check. + using ActivityRecorder publishRecorder = + new(RabbitMQActivitySource.PublisherSourceName, "publish"); + publishRecorder.VerifyParent = false; + + ConnectionFactory cf = CreateConnectionFactory(); + await using (IConnection conn = await cf.CreateConnectionAsync()) + { + await conn.CloseAsync(); + } + + Activity connectionActivity = connectionRecorder.VerifyActivityRecordedOnce(); + connectionActivity.HasNoTag("messaging.message.envelope.size"); + + // Network tags still belong on it, from the direct SetNetworkTags call. + connectionActivity.HasTag("server.port"); + } + + [Fact] + public async Task TestTcpConnectionActivityHasServerTags_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * OpenTcpConnection was the only activity factory returning a bare activity + * with no tag block; the server tags were set by the call site instead. + * Folding them into the factory keeps every factory in this file + * self-consistent, so a new call site cannot forget them. + */ + using ActivityRecorder tcpConnectionRecorder = + new(RabbitMQActivitySource.ConnectionSourceName, "tcp connection attempt"); + tcpConnectionRecorder.VerifyParent = false; + + ConnectionFactory cf = CreateConnectionFactory(); + await using (IConnection conn = await cf.CreateConnectionAsync()) + { + await conn.CloseAsync(); + } + + Activity tcpActivity = tcpConnectionRecorder.VerifyActivityRecordedOnce(); + // cf.Port is still the UseDefaultPort sentinel; Endpoint.Port resolves it. + tcpActivity.HasTag("server.port", cf.Endpoint.Port); + tcpActivity.HasTag("messaging.system", "rabbitmq"); + } + private static ActivityListener StartActivityListener(List activities) { ActivityListener activityListener = new ActivityListener(); diff --git a/projects/Test/SequentialIntegration/TestOpenTelemetry.cs b/projects/Test/SequentialIntegration/TestOpenTelemetry.cs index c4f1d5d42..619baa2eb 100644 --- a/projects/Test/SequentialIntegration/TestOpenTelemetry.cs +++ b/projects/Test/SequentialIntegration/TestOpenTelemetry.cs @@ -94,6 +94,36 @@ public void TestDefaultTracingOptions() Assert.True(RabbitMQActivitySource.TracingOptions.UsePublisherAsParent); } + [Fact] + public void TestContextExtractorHandlesPropertiesWithNoHeaders_GH1967() + { + /* + * rabbitmq/rabbitmq-dotnet-client#1967 + * + * OpenTelemetryContextExtractor passed props.Headers straight to the + * propagator, so a message published with no headers at all called the + * getter once per propagator field with a null carrier. It worked only + * because the getter's blanket catch swallowed the resulting + * NullReferenceException. + * + * This pins the observable contract - no headers extracts to no context, + * without throwing - rather than the mechanism. It would also have passed + * before the fix, because swallowing the NRE reached the same result. What + * it protects is the outcome if someone later narrows or removes that + * catch, which the fix makes safe to do. + */ + using var tracer = Sdk.CreateTracerProviderBuilder() + .AddRabbitMQInstrumentation() + .Build(); + + var propsWithNoHeaders = new BasicProperties(); + Assert.Null(propsWithNoHeaders.Headers); + + ActivityContext extracted = RabbitMQActivitySource.ContextExtractor(propsWithNoHeaders); + + Assert.Equal(default, extracted); + } + [Theory] [InlineData(true, true)] [InlineData(true, false)]