diff --git a/doc/DistributedTracing.md b/doc/DistributedTracing.md index b4f930a4fd..7902584c9d 100644 --- a/doc/DistributedTracing.md +++ b/doc/DistributedTracing.md @@ -102,6 +102,13 @@ clientOptions.Telemetry.ApplicationId = "MyApplication"; ServiceClient myServiceClient(clientOptions); ``` +Some clients do not derive their options from `Azure::Core::_internal::ClientOptions`, so those options have no `Telemetry` field. These clients declare a `TracingProvider` field at the top level of their own options structure. The Event Hubs clients work this way: + +```cpp +Azure::Messaging::EventHubs::ProducerClientOptions producerOptions; +producerOptions.TracingProvider = provider; +``` + ## Distributed Tracing Service Integration There are two steps needed to integrate Distributed Tracing with a Service Client. diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index 7fb2c1cc7a..da3b22f46b 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -34,6 +34,8 @@ ### Other Changes +- uAMQP connection and link failure logs now include the connection container ID or link name. Transitions to an error state use the warning log level. + ## 1.0.0-beta.12 (2026-05-14) ### Features Added diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp index 735a55b011..387756dcd0 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp @@ -356,8 +356,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { // When the connection transitions into the error or end state, it is no longer pollable. if (connection->m_options.EnableTrace) { - Log::Stream(Logger::Level::Verbose) - << "Connection " << connection->m_containerId << " state changed to " << newState; + Log::Stream( + newState == CONNECTION_STATE_ERROR ? Logger::Level::Warning : Logger::Level::Verbose) + << "AMQP connection '" << connection->m_containerId << "' state changed to " + << newState; } } // Nothing polls the connection after this point, so every operation that diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_receiver.cpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_receiver.cpp index 6e304edf9a..45dd793b03 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_receiver.cpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_receiver.cpp @@ -348,8 +348,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { if (receiver->m_options.EnableTrace) { - Log::Stream(Logger::Level::Verbose) - << "Message receiver state change " << oldState << " -> " << newState; + Log::Stream( + newState == MESSAGE_RECEIVER_STATE_ERROR ? Logger::Level::Warning + : Logger::Level::Verbose) + << "Message receiver link '" << receiver->m_options.Name << "' state change " << oldState + << " -> " << newState; } // If the message receiver isn't open, or if it's in the process of being destroyed, ignore // this notification. diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_sender.cpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_sender.cpp index d8d8c9fd94..d0ad08cd2a 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_sender.cpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/message_sender.cpp @@ -248,8 +248,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { sender->m_currentState = MessageSenderStateFromLowLevel(newState); if (sender->m_options.EnableTrace) { - Log::Stream(Logger::Level::Verbose) - << "Message sender state changed from " << oldState << " to " << newState << "."; + Log::Stream( + newState == MESSAGE_SENDER_STATE_ERROR ? Logger::Level::Warning + : Logger::Level::Verbose) + << "Message sender link '" << sender->m_options.Name << "' state changed from " + << oldState << " to " << newState << "."; } if (sender->m_events) { diff --git a/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md b/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md index 72f6ffa4c7..b0a7bc7b81 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md @@ -8,6 +8,9 @@ - [[#7295]](https://github.com/Azure/azure-sdk-for-cpp/issues/7295) Connection-string authentication now works on the Rust AMQP backend. `ProducerClient` and `ConsumerClient` no longer throw when the caller passes a connection string. - [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) `ProducerClient::Send` now builds a new sender on each retry attempt. A failed attempt discards the sender, the session, and the connection for that partition, so the next attempt builds all three again and authenticates with a current token. A send that a link detach ended previously failed for the life of the client. - [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) `PartitionClient::ReceiveEvents` now attaches a new receiver after a link fault, and it starts after the last event that it gave the caller. So the caller sees no duplicate event and no lost event. A permanent condition, for example `amqp:link:stolen`, still reaches the caller at once. A call that already holds events gives them back and recovers on the next call. +- [[#7336]](https://github.com/Azure/azure-sdk-for-cpp/issues/7336) Added distributed tracing. `ProducerClientOptions` and `ConsumerClientOptions` gained a `TracingProvider` field. `ProducerClient::Send` and `PartitionClient::ReceiveEvents` create one span per call when a tracing provider is set. Child spans measure AMQP link attachment, send disposition, and blocking receive latency. AMQP lifecycle logs identify the client, partition, component, and component generation for failures and rebuilds. + +### Breaking Changes ### Bugs Fixed diff --git a/sdk/eventhubs/azure-messaging-eventhubs/CMakeLists.txt b/sdk/eventhubs/azure-messaging-eventhubs/CMakeLists.txt index 618d19ce08..7c44fb7542 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/CMakeLists.txt +++ b/sdk/eventhubs/azure-messaging-eventhubs/CMakeLists.txt @@ -80,10 +80,14 @@ set( src/consumer_client.cpp src/event_data.cpp src/event_data_batch.cpp + src/eventhubs_diagnostics.cpp + src/eventhubs_tracing.cpp src/eventhubs_utilities.cpp src/partition_client.cpp src/partition_client_models.cpp src/private/eventhubs_constants.hpp + src/private/eventhubs_diagnostics.hpp + src/private/eventhubs_tracing.hpp src/private/eventhubs_utilities.hpp src/private/package_version.hpp src/private/processor_load_balancer.hpp diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index f33bcccac2..71596e6cfd 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -246,13 +246,102 @@ Azure::Messaging::EventHubs::PartitionClient partitionClient auto events = partitionClient.ReceiveEvents(1); ``` +## Distributed tracing + +The `ProducerClient` and the `PartitionClient` create distributed tracing spans through the Azure Core tracing API. This package does not depend on opentelemetry-cpp. The application creates the OpenTelemetry tracer provider and links the `azure-core-tracing-opentelemetry` package. + +To get the spans, set the `TracingProvider` field on the client options. The Event Hubs options structs declare this field at the top level: + +```cpp +#include +#include + +// Your Event Hubs namespace connection string is available in the Azure portal. +std::string connectionString = ""; +std::string eventHubName = ""; + +// Use the opentelemetry-cpp tracer provider of the application. +opentelemetry::nostd::shared_ptr tracerProvider + = opentelemetry::trace::Provider::GetTracerProvider(); + +std::shared_ptr provider + = Azure::Core::Tracing::OpenTelemetry::OpenTelemetryProvider::Create(tracerProvider); + +Azure::Messaging::EventHubs::ProducerClientOptions producerOptions; +producerOptions.TracingProvider = provider; +Azure::Messaging::EventHubs::ProducerClient producer( + connectionString, eventHubName, producerOptions); + +Azure::Messaging::EventHubs::ConsumerClientOptions consumerOptions; +consumerOptions.TracingProvider = provider; +Azure::Messaging::EventHubs::ConsumerClient consumer( + connectionString, + eventHubName, + Azure::Messaging::EventHubs::DefaultConsumerGroup, + consumerOptions); +``` + +The clients create these operation spans: + +| Span name | Span kind | Notes | +|---|---|---| +| `ProducerClient.Send` | Producer | One span for each `Send` call. The span covers all the retry attempts. The overloads that take events also create the batch inside the span. Batch and vector `ProducerClient.Send` operation spans receive the `messaging.batch.message_count` attribute. Single-event sends do not. | +| `PartitionClient.ReceiveEvents` | Client | One span for each `ReceiveEvents` call. | + +The clients also create child spans around calls into the AMQP transport: + +| Span name | What its duration measures | +|---|---| +| `ProducerClient.AmqpLink.Open` | Sender-link attachment or reattachment. On uAMQP, an initial attachment can also include lazy connection and session establishment. A retry can create another span. | +| `ProducerClient.AmqpSend` | The synchronous AMQP send through the service disposition. Every child span receives the internal batch count, including a single-event send. The `az.eventhubs.retry.attempt` attribute identifies the attempt. | +| `PartitionClient.AmqpLink.Open` | Receiver-link attachment or reattachment. On uAMQP, an initial attachment also includes lazy connection and session establishment. | +| `PartitionClient.AmqpReceive` | Time blocked in the AMQP transport waiting for a message or transport error. More than one can occur during one `ReceiveEvents` call. | + +The operation span duration is total SDK latency as observed by the caller. Subtracting the non-overlapping child-span durations from the operation duration gives the time spent in SDK work, retry delay, and locally queued message processing. The AMQP child duration is the client-observed service round-trip boundary; it does not isolate processing time inside the Event Hubs service. A receive can complete from AMQP prefetch, so `PartitionClient.AmqpReceive` can be shorter than a network round trip. + +The spans have these attributes. The names follow the OpenTelemetry semantic conventions version 1.17.0, which is the schema of the `azure-core-tracing-opentelemetry` package: + +| Attribute | Value | +|---|---| +| `az.namespace` | `Microsoft.EventHub` | +| `messaging.system` | `eventhubs` | +| `messaging.destination.name` | The Event Hub name on send spans. | +| `messaging.source.name` | The Event Hub name on receive spans. | +| `messaging.operation` | `publish` on a send span, `receive` on a receive span. | +| `messaging.batch.message_count` | The number of events in the operation. Batch and vector `ProducerClient.Send` spans receive this attribute. Every `ProducerClient.AmqpSend` child span receives the internal batch count. A `PartitionClient.ReceiveEvents` span gets this attribute when the call is successful. | +| `net.peer.name` | The fully qualified namespace. | + +AMQP child spans also have these Event Hubs diagnostic attributes: + +| Attribute | Value | +|---|---| +| `az.eventhubs.client.id` | A unique ID for the producer or consumer. It includes the configured client name when one exists and a generated UUID. | +| `az.eventhubs.partition.id` | The partition ID, or `` for the producer gateway link. | +| `az.eventhubs.amqp.component.type` | `link`. | +| `az.eventhubs.amqp.component.name` | The AMQP link name. | +| `az.eventhubs.amqp.component.id` | A unique ID composed from the client, partition, component generation, and type. | +| `az.eventhubs.amqp.component.generation` | Starts at 1 and increases when the component is recreated. | +| `az.eventhubs.retry.attempt` | The one-based retry or rebuild attempt, when applicable. | + +The instrumentation scope is `azure-messaging-eventhubs-cpp` with the package version. + +When the application does not set `TracingProvider`, the client creates no spans and records nothing. There is no global fallback provider. + +For the OpenTelemetry provider setup, see [Distributed Tracing in the C++ SDK][distributed_tracing]. # Troubleshooting ## Logging -The EventHubs SDK client uses the [Azure SDK log message](https://github.com/Azure/azure-sdk-for-cpp/tree/main/sdk/core/azure-core#sdk-log-messages) functionality to -enable diagnostics. +The EventHubs SDK client uses the [Azure SDK log message](https://github.com/Azure/azure-sdk-for-cpp/tree/main/sdk/core/azure-core#sdk-log-messages) functionality to enable diagnostics. + +AMQP lifecycle records start with `Event Hubs AMQP lifecycle:` and contain queryable `key='value'` fields. `client.id`, `partition.id`, `component.type`, `component.name`, `component.id`, and `component.generation` identify the exact connection, session, or link. The `event` field records creation, attachment, failure, discard, close, and recreation. Failure records use the warning level; successful recreations use the informational level; initial creation and normal close records use the verbose level. + +Failure records use the AMQP error-condition namespace to attribute `amqp:connection:*` and `amqp:session:*` failures to the connection or session. Other failures are attributed to the link operation where the client observed them. + +The producer gives every rebuilt connection, session, and link the same component generation. A receiver reattachment increases the link generation while keeping its owning connection and session. Low-level uAMQP connection and link failure records include the same connection container ID or link name, so they can be correlated with the Event Hubs lifecycle records. + +Azure Core does not currently expose a provider-neutral metrics API to service libraries. Applications can derive failure and recreation counters from lifecycle records, and latency histograms from the operation and AMQP child span durations. ## Contributing @@ -296,6 +385,7 @@ Azure SDK for C++ is licensed under the [MIT](https://github.com/Azure/azure-sdk [producer_client]: https://azuresdkdocs.z19.web.core.windows.net/cpp/azure-messaging-eventhubs/latest/class_azure_1_1_messaging_1_1_event_hubs_1_1_producer_client.html [source]: https://github.com/Azure/azure-sdk-for-cpp/tree/main/sdk/eventhubs +[distributed_tracing]: https://github.com/Azure/azure-sdk-for-cpp/blob/main/doc/DistributedTracing.md [azure_identity_pkg]: https://azuresdkdocs.z19.web.core.windows.net/cpp/azure-identity/latest/index.html [default_azure_credential]: https://azuresdkdocs.z19.web.core.windows.net/cpp/azure-identity/latest/index.html#defaultazurecredential diff --git a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp index ac3e2335c8..d7e99fefe4 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/consumer_client.hpp @@ -15,6 +15,8 @@ #include #include #include +#include +#include namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { class EventHubsPropertiesClient; @@ -41,6 +43,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { /** @brief Name of the consumer client. */ std::string Name{}; + /**@brief The tracer provider used to create distributed tracing spans. When this field is + * empty, the client creates no spans. + */ + std::shared_ptr TracingProvider; + private: // The friend declaration is needed so that ConsumerClient could access CppStandardVersion, // and it is not a struct's public field like the ones above to be set non-programmatically. @@ -238,6 +245,12 @@ namespace Azure { namespace Messaging { namespace EventHubs { /// @brief The options used to configure the consumer client. ConsumerClientOptions m_consumerClientOptions; + /// Correlates this client and its AMQP components across lifecycle logs and spans. + std::string m_clientIdentifier; + + /// @brief The factory used to create the distributed tracing spans of this client. + Azure::Core::Tracing::_internal::TracingContextFactory m_tracingFactory; + void EnsureConnection(std::string const& partitionId, Azure::Core::Context const& context); void EnsureSession(std::string const& partitionId, Azure::Core::Context const& context); Azure::Core::Amqp::_internal::Connection CreateConnection( diff --git a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/event_data_batch.hpp b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/event_data_batch.hpp index 7bc049fd24..6e8c014645 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/event_data_batch.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/event_data_batch.hpp @@ -55,7 +55,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { */ class EventDataBatch final { private: - std::mutex m_rwMutex; + mutable std::mutex m_rwMutex; std::string m_partitionId; std::string m_partitionKey; Azure::Nullable m_maxBytes; @@ -141,7 +141,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { /** @brief Gets the number of messages in the batch * */ - size_t NumberOfEvents() + size_t NumberOfEvents() const { std::lock_guard lock(m_rwMutex); return m_marshalledMessages.size(); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp index aa663fe2cc..ef699cd56b 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/partition_client.hpp @@ -9,8 +9,11 @@ #include #include #include +#include #include +#include + namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { class PartitionClientFactory; @@ -99,6 +102,13 @@ namespace Azure { namespace Messaging { namespace EventHubs { /// The link name of the receiver. A rebuild reuses this name. std::string m_receiverName; + /// Identifies the owning consumer and partition in lifecycle logs and spans. + std::string m_clientIdentifier; + std::string m_partitionId; + + /// Increases after each successful receiver reattach. + std::uint64_t m_receiverGeneration{1}; + /// The offset of the last event received. A rebuild starts just after it. Azure::Nullable m_lastReceivedOffset; @@ -116,26 +126,45 @@ namespace Azure { namespace Messaging { namespace EventHubs { */ Azure::Core::Http::Policies::RetryOptions m_retryOptions{}; + /// The factory used to create the distributed tracing spans of this client. + Azure::Core::Tracing::_internal::TracingContextFactory m_tracingFactory; + + /// The name of the Event Hub. + std::string m_eventHubName; + + /// The fully qualified namespace of the Event Hub. + std::string m_fullyQualifiedNamespace; + /** Creates a new PartitionClient * * @param messageReceiver Message Receiver for the partition client. * @param session The AMQP session that carries the message receiver. * @param partitionUrl The address of the partition. * @param receiverName The link name of the message receiver. + * @param clientIdentifier identifies the owning consumer in lifecycle logs and spans. + * @param partitionId identifies the partition in lifecycle logs and spans. * @param options options used to create the PartitionClient. * @param retryOptions controls how many times we should retry an operation in response to being * throttled or encountering a transient error. + * @param tracingFactory factory used to create the distributed tracing spans. + * @param eventHubName the name of the Event Hub. + * @param fullyQualifiedNamespace the fully qualified namespace of the Event Hub. */ PartitionClient( Azure::Core::Amqp::_internal::MessageReceiver const& messageReceiver, Azure::Core::Amqp::_internal::Session const& session, std::string partitionUrl, std::string receiverName, + std::string clientIdentifier, + std::string partitionId, PartitionClientOptions options, - Core::Http::Policies::RetryOptions retryOptions); + Core::Http::Policies::RetryOptions retryOptions, + Azure::Core::Tracing::_internal::TracingContextFactory tracingFactory, + std::string eventHubName, + std::string fullyQualifiedNamespace); /// Closes the faulted receiver and attaches a new one starting after the last offset. - void RebuildReceiver(Core::Context const& context); + void RebuildReceiver(std::uint64_t retryAttempt, Core::Context const& context); std::string GetStartExpression(Models::StartPosition const& startPosition); }; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp index a0d185668d..ae62ad3d56 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/inc/azure/messaging/eventhubs/producer_client.hpp @@ -12,7 +12,9 @@ #include #include #include +#include #include +#include #include #include @@ -48,6 +50,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { */ Azure::Nullable MaxMessageSize{}; + /**@brief The tracer provider used to create distributed tracing spans. When this field is + * empty, the client creates no spans. + */ + std::shared_ptr TracingProvider; + private: // The friend declaration is needed so that ProducerClient could access CppStandardVersion, // and it is not a struct's public field like the ones above to be set non-programmatically. @@ -214,6 +221,12 @@ namespace Azure { namespace Messaging { namespace EventHubs { ProducerClientOptions m_producerClientOptions{}; + /// Correlates this client and its AMQP components across lifecycle logs and spans. + std::string m_clientIdentifier; + + /// The factory used to create the distributed tracing spans of this client. + Azure::Core::Tracing::_internal::TracingContextFactory m_tracingFactory; + std::mutex m_propertiesClientLock; std::shared_ptr<_detail::EventHubsPropertiesClient> m_propertiesClient; @@ -236,6 +249,8 @@ namespace Azure { namespace Messaging { namespace EventHubs { { std::shared_timed_mutex stackLock; std::atomic generation{0}; + std::atomic nextStackId{0}; + std::atomic activeStackId{0}; }; // Protects m_partitionGuards. References into the map stay stable across inserts. @@ -245,6 +260,8 @@ namespace Azure { namespace Messaging { namespace EventHubs { PartitionGuard& GetPartitionGuard(std::string const& partitionId); Azure::Core::Amqp::_internal::Connection CreateConnection( + std::string const& partitionId, + std::uint64_t stackId, Azure::Core::Context const& context) const; Azure::Core::Amqp::_internal::Session CreateSession( std::string const& partitionId, @@ -275,7 +292,14 @@ namespace Azure { namespace Messaging { namespace EventHubs { void InvalidateSender( std::string const& partitionId, Azure::Nullable observedGeneration, - Azure::Core::Context const& context); + Azure::Core::Context const& context, + Azure::Nullable failureReason = {}); + + // Sends a batch inside the span that a public Send overload started. This method never + // starts a span, so one logical send makes one span. + void SendBatchInSpan( + EventDataBatch const& eventDataBatch, + Azure::Core::Tracing::_internal::TracingContextFactory::TracingContext& tracingContext); std::shared_ptr<_detail::EventHubsPropertiesClient> GetPropertiesClient( Azure::Core::Context const& context); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp index 12122175b4..5c88902143 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/consumer_client.cpp @@ -3,6 +3,8 @@ #include "private/best_effort_cleanup.hpp" #include "private/eventhubs_constants.hpp" +#include "private/eventhubs_diagnostics.hpp" +#include "private/eventhubs_tracing.hpp" #include "private/eventhubs_utilities.hpp" #include "private/package_version.hpp" @@ -16,6 +18,28 @@ using namespace Azure::Core::Diagnostics; using namespace Azure::Messaging::EventHubs::Models; using namespace Azure::Core::Amqp::_internal; +namespace { +Azure::Messaging::EventHubs::_detail::AmqpDiagnosticsContext CreateDiagnosticsContext( + std::string const& clientId, + std::string const& partitionId, + std::string componentType, + std::string componentName) +{ + return Azure::Messaging::EventHubs::_detail::AmqpDiagnosticsContext{ + clientId, partitionId, std::move(componentType), std::move(componentName), 1}; +} + +std::string CreateConnectionName( + std::string const& clientId, + std::string const& applicationId, + std::string const& partitionId) +{ + return clientId + (applicationId.empty() ? std::string{} : "/application/" + applicationId) + + "/partition/" + (partitionId.empty() ? std::string("") : partitionId) + + "/generation/1"; +} +} // namespace + namespace Azure { namespace Messaging { namespace EventHubs { ConsumerClient::ConsumerClient( @@ -24,7 +48,9 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::string const& consumerGroup, ConsumerClientOptions const& options) : m_connectionString{connectionString}, m_eventHub{eventHub}, m_consumerGroup{consumerGroup}, - m_consumerClientOptions(options) + m_consumerClientOptions(options), + m_clientIdentifier{_detail::CreateClientIdentifier("consumer", options.Name)}, + m_tracingFactory{_detail::CreateTracingContextFactory(options.TracingProvider)} { auto details = _detail::EventHubsUtilities::CreateConnectionStringDetails(connectionString, eventHub); @@ -34,6 +60,10 @@ namespace Azure { namespace Messaging { namespace EventHubs { m_targetPort = details.Port; m_hostUrl = details.ServiceScheme + m_fullyQualifiedNamespace + "/" + m_eventHub + _detail::EventHubsConsumerGroupsPath + m_consumerGroup; + if (m_consumerClientOptions.Name.empty()) + { + m_consumerClientOptions.Name = m_clientIdentifier; + } } ConsumerClient::ConsumerClient( @@ -43,10 +73,16 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::string const& consumerGroup, ConsumerClientOptions const& options) : m_fullyQualifiedNamespace{fullyQualifiedNamespace}, m_eventHub{eventHub}, - m_consumerGroup{consumerGroup}, m_credential{credential}, m_consumerClientOptions(options) + m_consumerGroup{consumerGroup}, m_credential{credential}, m_consumerClientOptions(options), + m_clientIdentifier{_detail::CreateClientIdentifier("consumer", options.Name)}, + m_tracingFactory{_detail::CreateTracingContextFactory(options.TracingProvider)} { m_hostUrl = _detail::EventHubsServiceScheme + m_fullyQualifiedNamespace + "/" + m_eventHub + _detail::EventHubsConsumerGroupsPath + m_consumerGroup; + if (m_consumerClientOptions.Name.empty()) + { + m_consumerClientOptions.Name = m_clientIdentifier; + } } ConsumerClient::~ConsumerClient() @@ -129,8 +165,8 @@ namespace Azure { namespace Messaging { namespace EventHubs { Azure::Core::Context const& context) const { ConnectionOptions connectOptions; - connectOptions.ContainerId - = "Consumer for " + m_consumerClientOptions.ApplicationID + " on " + partitionId; + connectOptions.ContainerId = CreateConnectionName( + m_clientIdentifier, m_consumerClientOptions.ApplicationID, partitionId); connectOptions.EnableTrace = _detail::EnableAmqpTrace; connectOptions.AuthenticationScopes = {"https://eventhubs.azure.net/.default"}; connectOptions.Port = m_targetPort; @@ -158,7 +194,24 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::unique_lock lock(m_sessionsLock); if (m_connections.find(partitionId) == m_connections.end()) { - m_connections.emplace(partitionId, CreateConnection(partitionId, context)); + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, + partitionId, + "connection", + CreateConnectionName( + m_clientIdentifier, m_consumerClientOptions.ApplicationID, partitionId)); + _detail::LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "creating"); + try + { + m_connections.emplace(partitionId, CreateConnection(partitionId, context)); + _detail::LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "created"); + } + catch (std::exception const& ex) + { + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "create_failed", ex.what()); + throw; + } } } @@ -186,7 +239,20 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::unique_lock lock(m_sessionsLock); if (m_sessions.find(partitionId) == m_sessions.end()) { - m_sessions.emplace(partitionId, CreateSession(partitionId, context)); + auto diagnosticsContext + = CreateDiagnosticsContext(m_clientIdentifier, partitionId, "session", partitionId); + _detail::LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "creating"); + try + { + m_sessions.emplace(partitionId, CreateSession(partitionId, context)); + _detail::LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "created"); + } + catch (std::exception const& ex) + { + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "create_failed", ex.what()); + throw; + } } } @@ -224,8 +290,13 @@ namespace Azure { namespace Messaging { namespace EventHubs { GetSession(partitionId), hostUrl, m_consumerClientOptions.Name, + m_clientIdentifier, + partitionId, options, m_consumerClientOptions.RetryOptions, + m_tracingFactory, + m_eventHub, + m_fullyQualifiedNamespace, context); } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/eventhubs_diagnostics.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/eventhubs_diagnostics.cpp new file mode 100644 index 0000000000..fb63df53cc --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/eventhubs_diagnostics.cpp @@ -0,0 +1,113 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#include "private/eventhubs_diagnostics.hpp" + +#include +#include + +#include + +namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { + + namespace { + std::string DisplayPartition(std::string const& partitionId) + { + return partitionId.empty() ? "" : partitionId; + } + + std::string DisplayComponentName(std::string const& componentName) + { + return componentName.empty() ? "" : componentName; + } + + std::string EscapeLogValue(std::string const& value) + { + std::string escaped; + escaped.reserve(value.size()); + for (auto const character : value) + { + switch (character) + { + case '\\': + escaped += "\\\\"; + break; + case '\'': + escaped += "\\'"; + break; + case '\n': + escaped += "\\n"; + break; + case '\r': + escaped += "\\r"; + break; + default: + escaped += character; + } + } + return escaped; + } + } // namespace + + std::string CreateClientIdentifier( + std::string const& clientType, + std::string const& configuredName) + { + return clientType + ":" + (configuredName.empty() ? std::string{} : configuredName + ":") + + Azure::Core::Uuid::CreateUuid().ToString(); + } + + std::string GetAmqpComponentIdentifier(AmqpDiagnosticsContext const& diagnosticsContext) + { + std::ostringstream identifier; + identifier << diagnosticsContext.ClientId << "/partition/" + << DisplayPartition(diagnosticsContext.PartitionId) << "/generation/" + << diagnosticsContext.ComponentGeneration << "/" << diagnosticsContext.ComponentType; + return identifier.str(); + } + + std::string GetAmqpFailureComponentType(std::string const& errorCondition) + { + if (errorCondition.find("amqp:connection:") == 0) + { + return "connection"; + } + if (errorCondition.find("amqp:session:") == 0) + { + return "session"; + } + return "link"; + } + + std::string FormatAmqpLifecycleEvent( + AmqpDiagnosticsContext const& diagnosticsContext, + std::string const& eventName, + std::string const& detail) + { + std::ostringstream message; + message << "Event Hubs AMQP lifecycle: event='" << EscapeLogValue(eventName) << "' client.id='" + << EscapeLogValue(diagnosticsContext.ClientId) << "' partition.id='" + << EscapeLogValue(DisplayPartition(diagnosticsContext.PartitionId)) + << "' component.type='" << EscapeLogValue(diagnosticsContext.ComponentType) + << "' component.name='" + << EscapeLogValue(DisplayComponentName(diagnosticsContext.ComponentName)) + << "' component.id='" << EscapeLogValue(GetAmqpComponentIdentifier(diagnosticsContext)) + << "' component.generation=" << diagnosticsContext.ComponentGeneration; + if (!detail.empty()) + { + message << " detail='" << EscapeLogValue(detail) << "'"; + } + return message.str(); + } + + void LogAmqpLifecycle( + Azure::Core::Diagnostics::Logger::Level level, + AmqpDiagnosticsContext const& diagnosticsContext, + std::string const& eventName, + std::string const& detail) + { + Azure::Core::Diagnostics::_internal::Log::Write( + level, FormatAmqpLifecycleEvent(diagnosticsContext, eventName, detail)); + } + +}}}} // namespace Azure::Messaging::EventHubs::_detail diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/eventhubs_tracing.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/eventhubs_tracing.cpp new file mode 100644 index 0000000000..7dc57aa62c --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/eventhubs_tracing.cpp @@ -0,0 +1,135 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#include "private/eventhubs_tracing.hpp" + +#include "private/package_version.hpp" + +#include + +#include + +namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { + + Azure::Core::Tracing::_internal::TracingContextFactory CreateTracingContextFactory( + std::shared_ptr const& tracingProvider) + { + Azure::Core::_internal::ClientOptions clientOptions; + clientOptions.Telemetry.TracingProvider = tracingProvider; + return Azure::Core::Tracing::_internal::TracingContextFactory{ + clientOptions, + "Microsoft.EventHub", + "azure-messaging-eventhubs-cpp", + PackageVersion::ToString()}; + } + + Azure::Core::Tracing::_internal::TracingContextFactory::TracingContext StartSpan( + Azure::Core::Tracing::_internal::TracingContextFactory const& tracingFactory, + std::string const& spanName, + Azure::Core::Tracing::_internal::SpanKind spanKind, + std::string const& operationName, + std::string const& eventHubName, + std::string const& fullyQualifiedNamespace, + Azure::Nullable messageCount, + Azure::Core::Context const& context, + MessagingEntityKind entityKind) + { + Azure::Core::Tracing::_internal::CreateSpanOptions createOptions; + createOptions.Kind = spanKind; + + // The az.namespace attribute comes from the tracing context factory. + auto tracingContext = tracingFactory.CreateTracingContext(spanName, createOptions, context); + // A factory with no tracer drops every attribute, so skip the string temporaries. + if (!tracingFactory.HasTracer()) + { + return tracingContext; + } + tracingContext.Span.AddAttribute("messaging.system", "eventhubs"); + tracingContext.Span.AddAttribute( + entityKind == MessagingEntityKind::Source ? "messaging.source.name" + : "messaging.destination.name", + eventHubName); + tracingContext.Span.AddAttribute("messaging.operation", operationName); + tracingContext.Span.AddAttribute( + Azure::Core::Tracing::_internal::TracingAttributes::NetPeerName.ToString(), + fullyQualifiedNamespace); + if (messageCount.HasValue()) + { + SetMessageCount(tracingFactory, tracingContext.Span, messageCount.Value()); + } + return tracingContext; + } + + Azure::Core::Tracing::_internal::TracingContextFactory::TracingContext StartAmqpSpan( + Azure::Core::Tracing::_internal::TracingContextFactory const& tracingFactory, + std::string const& spanName, + std::string const& operationName, + std::string const& eventHubName, + std::string const& fullyQualifiedNamespace, + Azure::Nullable messageCount, + AmqpDiagnosticsContext const& diagnosticsContext, + Azure::Nullable retryAttempt, + Azure::Core::Context const& context, + MessagingEntityKind entityKind) + { + auto tracingContext = StartSpan( + tracingFactory, + spanName, + Azure::Core::Tracing::_internal::SpanKind::Client, + operationName, + eventHubName, + fullyQualifiedNamespace, + messageCount, + context, + entityKind); + if (!tracingFactory.HasTracer()) + { + return tracingContext; + } + + auto attributes = tracingFactory.CreateAttributeSet(); + if (!attributes) + { + return tracingContext; + } + // The OpenTelemetry attribute set keeps non-owning string views until AddAttributes runs. + // Keep computed values alive through that call. + auto const partitionId = diagnosticsContext.PartitionId.empty() + ? std::string{""} + : diagnosticsContext.PartitionId; + auto const componentName = diagnosticsContext.ComponentName.empty() + ? std::string{""} + : diagnosticsContext.ComponentName; + auto const componentId = GetAmqpComponentIdentifier(diagnosticsContext); + attributes->AddAttribute("az.eventhubs.client.id", diagnosticsContext.ClientId); + attributes->AddAttribute("az.eventhubs.partition.id", partitionId); + attributes->AddAttribute("az.eventhubs.amqp.component.type", diagnosticsContext.ComponentType); + attributes->AddAttribute("az.eventhubs.amqp.component.name", componentName); + attributes->AddAttribute("az.eventhubs.amqp.component.id", componentId); + attributes->AddAttribute( + "az.eventhubs.amqp.component.generation", diagnosticsContext.ComponentGeneration); + if (retryAttempt.HasValue()) + { + attributes->AddAttribute("az.eventhubs.retry.attempt", retryAttempt.Value()); + } + tracingContext.Span.AddAttributes(*attributes); + return tracingContext; + } + + void SetMessageCount( + Azure::Core::Tracing::_internal::TracingContextFactory const& tracingFactory, + Azure::Core::Tracing::_internal::ServiceSpan& span, + size_t messageCount) + { + // A factory with no tracer returns a null attribute set, and that pointer is not null safe. + auto attributeSet = tracingFactory.CreateAttributeSet(); + if (!attributeSet) + { + return; + } + attributeSet->AddAttribute( + "messaging.batch.message_count", static_cast(messageCount)); + span.AddAttributes(*attributeSet); + } + +}}}} // namespace Azure::Messaging::EventHubs::_detail diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp index b8053a61d1..d7f5a54d74 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/partition_client.cpp @@ -5,6 +5,8 @@ #include "azure/messaging/eventhubs/eventhubs_exception.hpp" #include "private/eventhubs_constants.hpp" +#include "private/eventhubs_diagnostics.hpp" +#include "private/eventhubs_tracing.hpp" #include "private/eventhubs_utilities.hpp" #include "private/retry_operation.hpp" @@ -13,6 +15,7 @@ #include #include +#include #include using namespace Azure::Core::Diagnostics::_internal; @@ -36,6 +39,42 @@ namespace Azure { namespace Messaging { namespace EventHubs { FilterDescription SelectorFilter{"apache.org:selector-filter:string", 0x0000468c00000004}; + _detail::AmqpDiagnosticsContext CreateDiagnosticsContext( + std::string const& clientId, + std::string const& partitionId, + std::string const& receiverName, + std::uint64_t stackId) + { + return _detail::AmqpDiagnosticsContext{clientId, partitionId, "link", receiverName, stackId}; + } + + _detail::AmqpDiagnosticsContext CreateFailureDiagnosticsContext( + std::string const& clientId, + std::string const& partitionId, + std::string const& receiverName, + std::uint64_t receiverGeneration, + std::string const& errorCondition) + { + auto const componentType = _detail::GetAmqpFailureComponentType(errorCondition); + if (componentType == "session") + { + return _detail::AmqpDiagnosticsContext{ + clientId, partitionId, componentType, partitionId, 1}; + } + if (componentType == "connection") + { + return _detail::AmqpDiagnosticsContext{clientId, partitionId, componentType, {}, 1}; + } + return CreateDiagnosticsContext(clientId, partitionId, receiverName, receiverGeneration); + } + + std::string FormatAmqpError(Azure::Core::Amqp::Models::_internal::AmqpError const& error) + { + std::ostringstream message; + message << error; + return message.str(); + } + std::string GetStartExpression(Models::StartPosition const& startPosition) { Log::Stream(Logger::Level::Verbose) @@ -186,21 +225,55 @@ namespace Azure { namespace Messaging { namespace EventHubs { Azure::Core::Amqp::_internal::Session const& session, std::string const& partitionUrl, std::string const& receiverName, + std::string clientIdentifier, + std::string partitionId, PartitionClientOptions options, Azure::Core::Http::Policies::RetryOptions retryOptions, + Azure::Core::Tracing::_internal::TracingContextFactory tracingFactory, + std::string eventHubName, + std::string fullyQualifiedNamespace, Azure::Core::Context const& context) { - Azure::Core::Amqp::_internal::MessageReceiver messageReceiver{ - CreateMessageReceiver(session, partitionUrl, receiverName, options)}; - messageReceiver.Open(context); - - return PartitionClient( - std::move(messageReceiver), - session, - partitionUrl, - receiverName, - std::move(options), - std::move(retryOptions)); + auto diagnosticsContext + = CreateDiagnosticsContext(clientIdentifier, partitionId, receiverName, 1); + LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "attaching"); + auto amqpTracingContext = StartAmqpSpan( + tracingFactory, + "PartitionClient.AmqpLink.Open", + "receive", + eventHubName, + fullyQualifiedNamespace, + Azure::Nullable{}, + diagnosticsContext, + Azure::Nullable{}, + context, + MessagingEntityKind::Source); + try + { + Azure::Core::Amqp::_internal::MessageReceiver messageReceiver{ + CreateMessageReceiver(session, partitionUrl, receiverName, options)}; + messageReceiver.Open(amqpTracingContext.Context); + LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "attached"); + + return PartitionClient( + std::move(messageReceiver), + session, + partitionUrl, + receiverName, + std::move(clientIdentifier), + std::move(partitionId), + std::move(options), + std::move(retryOptions), + std::move(tracingFactory), + std::move(eventHubName), + std::move(fullyQualifiedNamespace)); + } + catch (std::exception const& ex) + { + amqpTracingContext.Span.AddEvent(ex); + LogAmqpLifecycle(Logger::Level::Warning, diagnosticsContext, "attach_failed", ex.what()); + throw; + } } /** Creates a new PartitionClient @@ -209,26 +282,55 @@ namespace Azure { namespace Messaging { namespace EventHubs { * @param options options used to create the PartitionClient. * @param retryOptions controls how many times we should retry an operation in response to being * throttled or encountering a transient error. + * @param tracingFactory factory used to create the distributed tracing spans. + * @param eventHubName the name of the Event Hub. + * @param fullyQualifiedNamespace the fully qualified namespace of the Event Hub. */ PartitionClient::PartitionClient( Azure::Core::Amqp::_internal::MessageReceiver const& messageReceiver, Azure::Core::Amqp::_internal::Session const& session, std::string partitionUrl, std::string receiverName, + std::string clientIdentifier, + std::string partitionId, PartitionClientOptions options, - Core::Http::Policies::RetryOptions retryOptions) + Core::Http::Policies::RetryOptions retryOptions, + Azure::Core::Tracing::_internal::TracingContextFactory tracingFactory, + std::string eventHubName, + std::string fullyQualifiedNamespace) : m_receiver{messageReceiver}, m_session{session}, m_partitionUrl{std::move(partitionUrl)}, - m_receiverName{std::move(receiverName)}, m_partitionOptions{options}, m_retryOptions{ - retryOptions} + m_receiverName{std::move(receiverName)}, m_clientIdentifier{std::move(clientIdentifier)}, + m_partitionId{std::move(partitionId)}, m_partitionOptions{options}, + m_retryOptions{retryOptions}, m_tracingFactory{std::move(tracingFactory)}, + m_eventHubName{std::move(eventHubName)}, m_fullyQualifiedNamespace{ + std::move(fullyQualifiedNamespace)} { } - void PartitionClient::Close(Core::Context const& context) { m_receiver.Close(context); } + void PartitionClient::Close(Core::Context const& context) + { + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, m_partitionId, m_receiverName, m_receiverGeneration); + _detail::LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "closing"); + try + { + m_receiver.Close(context); + _detail::LogAmqpLifecycle(Logger::Level::Verbose, diagnosticsContext, "closed"); + } + catch (std::exception const& ex) + { + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "close_failed", ex.what()); + throw; + } + } - void PartitionClient::RebuildReceiver(Core::Context const& context) + void PartitionClient::RebuildReceiver(std::uint64_t retryAttempt, Core::Context const& context) { - Log::Stream(Logger::Level::Informational) - << "Rebuild the message receiver for " << m_partitionUrl << "."; + auto const rebuiltGeneration = m_receiverGeneration + 1; + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, m_partitionId, m_receiverName, rebuiltGeneration); + _detail::LogAmqpLifecycle(Logger::Level::Informational, diagnosticsContext, "reattaching"); try { @@ -240,21 +342,41 @@ namespace Azure { namespace Messaging { namespace EventHubs { } catch (std::exception const& ex) { - Log::Stream(Logger::Level::Warning) - << "Exception while closing a faulted message receiver: " << ex.what(); + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "close_failed", ex.what()); } PartitionClientOptions options{m_partitionOptions}; options.StartPosition = _detail::ResumeStartPosition(m_partitionOptions.StartPosition, m_lastReceivedOffset); - Azure::Core::Amqp::_internal::MessageReceiver receiver{ - CreateMessageReceiver(m_session, m_partitionUrl, m_receiverName, options)}; - receiver.Open(context); - m_receiver = std::move(receiver); - - Log::Stream(Logger::Level::Informational) - << "The message receiver for " << m_partitionUrl << " is attached again."; + auto amqpTracingContext = _detail::StartAmqpSpan( + m_tracingFactory, + "PartitionClient.AmqpLink.Open", + "receive", + m_eventHubName, + m_fullyQualifiedNamespace, + Azure::Nullable{}, + diagnosticsContext, + retryAttempt, + context, + _detail::MessagingEntityKind::Source); + try + { + Azure::Core::Amqp::_internal::MessageReceiver receiver{ + CreateMessageReceiver(m_session, m_partitionUrl, m_receiverName, options)}; + receiver.Open(amqpTracingContext.Context); + m_receiver = std::move(receiver); + m_receiverGeneration = rebuiltGeneration; + _detail::LogAmqpLifecycle(Logger::Level::Informational, diagnosticsContext, "reattached"); + } + catch (std::exception const& ex) + { + amqpTracingContext.Span.AddEvent(ex); + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "reattach_failed", ex.what()); + throw; + } } PartitionClient::~PartitionClient() @@ -267,8 +389,12 @@ namespace Azure { namespace Messaging { namespace EventHubs { } catch (std::exception const& ex) { - Log::Stream(Logger::Level::Warning) - << "Exception in PartitionClient::~PartitionClient(): " << ex.what(); + _detail::LogAmqpLifecycle( + Logger::Level::Warning, + CreateDiagnosticsContext( + m_clientIdentifier, m_partitionId, m_receiverName, m_receiverGeneration), + "close_failed", + ex.what()); } } @@ -285,155 +411,215 @@ namespace Azure { namespace Messaging { namespace EventHubs { { std::vector> messages; - // RetryOperation::Execute's budget never resets, so this loop keeps its own counter. - Azure::Core::Http::Policies::RetryOptions retryOptions{m_retryOptions}; - _detail::RetryOperation retryOperation{retryOptions}; - int32_t rebuildAttempt = 0; + // The message count is known only when the loop ends, so the span gets it later. + auto tracingContext = _detail::StartSpan( + m_tracingFactory, + "PartitionClient.ReceiveEvents", + Azure::Core::Tracing::_internal::SpanKind::Client, + "receive", + m_eventHubName, + m_fullyQualifiedNamespace, + Azure::Nullable{}, + context, + _detail::MessagingEntityKind::Source); - // Keep the event, and record the offset a rebuild must start after. - auto keepMessage - = [&](std::shared_ptr const& message) { - auto eventData = std::make_shared(message); - if (eventData->Offset.HasValue()) + try + { + // RetryOperation::Execute's budget never resets, so this loop keeps its own counter. + Azure::Core::Http::Policies::RetryOptions retryOptions{m_retryOptions}; + _detail::RetryOperation retryOperation{retryOptions}; + int32_t rebuildAttempt = 0; + + // Keep the event, and record the offset a rebuild must start after. + auto keepMessage + = [&](std::shared_ptr const& message) { + auto eventData = std::make_shared(message); + if (eventData->Offset.HasValue()) + { + m_lastReceivedOffset = eventData->Offset.Value(); + } + rebuildAttempt = 0; + messages.push_back(eventData); + }; + + // True: the receiver works again. False: return the events held. Throws if none are held. + auto recover = [&](Azure::Core::Amqp::Models::_internal::AmqpError const& error) -> bool { + _detail::LogAmqpLifecycle( + Logger::Level::Warning, + CreateFailureDiagnosticsContext( + m_clientIdentifier, + m_partitionId, + m_receiverName, + m_receiverGeneration, + error.Condition.ToString()), + "failed", + FormatAmqpError(error)); + EventHubsException exception{ + _detail::EventHubsExceptionFactory::CreateEventHubsException(error)}; + // currentError tracks the last rebuild fault, not the first. m_pendingError takes + // this value below, so the next call recovers from the fault that stopped this + // loop, not from an earlier one that a later attempt already showed can be fixed. + Azure::Core::Amqp::Models::_internal::AmqpError currentError{error}; + // Preserves a rebuild failure whose original type carries more than the + // translated EventHubsException above, such as an AuthenticationException. The + // loop throws this one when it stops, so the caller sees the original type. + std::exception_ptr originalFailure{}; + + for (;;) + { + std::chrono::milliseconds retryAfter{}; + if (!_detail::ShouldRebuildReceiver(exception) + || !retryOperation.ShouldRetry(false, rebuildAttempt, retryAfter)) + { + if (!messages.empty()) { - m_lastReceivedOffset = eventData->Offset.Value(); + // The service will not send these again. The next call gets a new budget. + Log::Stream(Logger::Level::Warning) + << "Cannot rebuild the message receiver now. Return " << messages.size() + << " events and keep the error for the next call: " << exception.what(); + m_pendingError = currentError; + return false; } - rebuildAttempt = 0; - messages.push_back(eventData); - }; - - // True: the receiver works again. False: return the events held. Throws if none are held. - auto recover = [&](Azure::Core::Amqp::Models::_internal::AmqpError const& error) -> bool { - EventHubsException exception{ - _detail::EventHubsExceptionFactory::CreateEventHubsException(error)}; - // currentError tracks the last rebuild fault, not the first. m_pendingError takes - // this value below, so the next call recovers from the fault that stopped this - // loop, not from an earlier one that a later attempt already showed can be fixed. - Azure::Core::Amqp::Models::_internal::AmqpError currentError{error}; - // Preserves a rebuild failure whose original type carries more than the - // translated EventHubsException above, such as an AuthenticationException. The - // loop throws this one when it stops, so the caller sees the original type. - std::exception_ptr originalFailure{}; - - for (;;) - { - std::chrono::milliseconds retryAfter{}; - if (!_detail::ShouldRebuildReceiver(exception) - || !retryOperation.ShouldRetry(false, rebuildAttempt, retryAfter)) - { - if (!messages.empty()) + if (originalFailure) + { + std::rethrow_exception(originalFailure); + } + throw exception; + } + + rebuildAttempt++; + std::this_thread::sleep_for(retryAfter); + tracingContext.Context.ThrowIfCancelled(); + + try + { + RebuildReceiver(static_cast(rebuildAttempt), tracingContext.Context); + return true; + } + catch (Azure::Core::OperationCancelledException const&) + { + throw; + } + catch (EventHubsException const& rebuildFailure) { - // The service will not send these again. The next call gets a new budget. Log::Stream(Logger::Level::Warning) - << "Cannot rebuild the message receiver now. Return " << messages.size() - << " events and keep the error for the next call: " << exception.what(); - m_pendingError = currentError; - return false; + << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); + exception = rebuildFailure; + originalFailure = nullptr; + currentError.Condition = Azure::Core::Amqp::Models::_internal::AmqpErrorCondition{ + exception.ErrorCondition}; + currentError.Description = exception.ErrorDescription; } - if (originalFailure) + catch (Azure::Core::Credentials::AuthenticationException const& rebuildFailure) { - std::rethrow_exception(originalFailure); + Log::Stream(Logger::Level::Warning) + << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); + exception = _detail::TranslateAuthenticationFailure(rebuildFailure); + originalFailure = std::current_exception(); + currentError.Condition = Azure::Core::Amqp::Models::_internal::AmqpErrorCondition{}; + currentError.Description = exception.ErrorDescription; + } + catch (std::exception const& rebuildFailure) + { + Log::Stream(Logger::Level::Warning) + << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); + EventHubsException translated{rebuildFailure.what()}; + translated.IsTransient = true; + exception = translated; + originalFailure = nullptr; + currentError.Condition = Azure::Core::Amqp::Models::_internal::AmqpErrorCondition{}; + currentError.Description = translated.ErrorDescription; } - throw exception; } + }; + + // No event is held yet, so this recover either works or throws. + if (m_pendingError.HasValue()) + { + auto pendingError = m_pendingError.Value(); + m_pendingError.Reset(); + recover(pendingError); + } - rebuildAttempt++; - std::this_thread::sleep_for(retryAfter); - context.ThrowIfCancelled(); + while (messages.size() < maxMessages && !tracingContext.Context.IsCancelled()) + { + std::pair< + std::shared_ptr, + Azure::Core::Amqp::Models::_internal::AmqpError> + result; - try - { - RebuildReceiver(context); - return true; - } - catch (Azure::Core::OperationCancelledException const&) + // TryPeekForIncomingMessage will return two empty values if there is no data available. + result = m_receiver.TryWaitForIncomingMessage(); + if (result.first) { - throw; + keepMessage(result.first); } - catch (EventHubsException const& rebuildFailure) + else if (result.second) { - Log::Stream(Logger::Level::Warning) - << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); - exception = rebuildFailure; - originalFailure = nullptr; - currentError.Condition - = Azure::Core::Amqp::Models::_internal::AmqpErrorCondition{exception.ErrorCondition}; - currentError.Description = exception.ErrorDescription; + if (!recover(result.second)) + { + break; + } } - catch (Azure::Core::Credentials::AuthenticationException const& rebuildFailure) + // If we haven't gotten *any* messages, we're done. Otherwise, we'll wait for more. + else if (!messages.empty()) { - Log::Stream(Logger::Level::Warning) - << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); - exception = _detail::TranslateAuthenticationFailure(rebuildFailure); - originalFailure = std::current_exception(); - currentError.Condition = Azure::Core::Amqp::Models::_internal::AmqpErrorCondition{}; - currentError.Description = exception.ErrorDescription; + break; } - catch (std::exception const& rebuildFailure) + else { - Log::Stream(Logger::Level::Warning) - << "Rebuild attempt " << rebuildAttempt << " failed: " << rebuildFailure.what(); - EventHubsException translated{rebuildFailure.what()}; - translated.IsTransient = true; - exception = translated; - originalFailure = nullptr; - currentError.Condition = Azure::Core::Amqp::Models::_internal::AmqpErrorCondition{}; - currentError.Description = translated.ErrorDescription; + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, m_partitionId, m_receiverName, m_receiverGeneration); + { + auto amqpTracingContext = _detail::StartAmqpSpan( + m_tracingFactory, + "PartitionClient.AmqpReceive", + "receive", + m_eventHubName, + m_fullyQualifiedNamespace, + Azure::Nullable{}, + diagnosticsContext, + Azure::Nullable{}, + tracingContext.Context, + _detail::MessagingEntityKind::Source); + try + { + result = m_receiver.WaitForIncomingMessage(amqpTracingContext.Context); + if (result.second) + { + EventHubsException receiveFailure{ + _detail::EventHubsExceptionFactory::CreateEventHubsException(result.second)}; + amqpTracingContext.Span.AddEvent(receiveFailure); + } + } + catch (std::exception const& ex) + { + amqpTracingContext.Span.AddEvent(ex); + throw; + } + } + if (result.first) + { + Log::Stream(Logger::Level::Verbose) + << "Received message. Message count now " << messages.size(); + keepMessage(result.first); + } + else if (!recover(result.second)) + { + break; + } } } - }; + Log::Stream(Logger::Level::Verbose) + << "Receive Events. Return " << messages.size() << " messages."; - // No event is held yet, so this recover either works or throws. - if (m_pendingError.HasValue()) - { - auto pendingError = m_pendingError.Value(); - m_pendingError.Reset(); - recover(pendingError); + _detail::SetMessageCount(m_tracingFactory, tracingContext.Span, messages.size()); + return messages; } - - while (messages.size() < maxMessages && !context.IsCancelled()) + catch (std::exception const& ex) { - std::pair< - std::shared_ptr, - Azure::Core::Amqp::Models::_internal::AmqpError> - result; - - // TryPeekForIncomingMessage will return two empty values if there is no data available. - result = m_receiver.TryWaitForIncomingMessage(); - if (result.first) - { - keepMessage(result.first); - } - else if (result.second) - { - if (!recover(result.second)) - { - break; - } - } - // If we haven't gotten *any* messages, we're done. Otherwise, we'll wait for more. - else if (!messages.empty()) - { - break; - } - else - { - result = m_receiver.WaitForIncomingMessage(context); - if (result.first) - { - Log::Stream(Logger::Level::Verbose) - << "Received message. Message count now " << messages.size(); - keepMessage(result.first); - } - else if (!recover(result.second)) - { - break; - } - } + tracingContext.Span.AddEvent(ex); + throw; } - Log::Stream(Logger::Level::Verbose) - << "Receive Events. Return " << messages.size() << " messages."; - - return messages; } }}} // namespace Azure::Messaging::EventHubs diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_diagnostics.hpp b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_diagnostics.hpp new file mode 100644 index 0000000000..71a10aec03 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_diagnostics.hpp @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +// Helpers for identifying Event Hubs AMQP components in logs and tracing spans. +#pragma once + +#include + +#include +#include +#include + +namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { + + struct AmqpDiagnosticsContext final + { + std::string ClientId; + std::string PartitionId; + std::string ComponentType; + std::string ComponentName; + std::uint64_t ComponentGeneration{}; + }; + + // Includes the configured name when one exists and a UUID that remains stable for the lifetime + // of the client. + std::string CreateClientIdentifier( + std::string const& clientType, + std::string const& configuredName); + + // Returns the identifier shared by lifecycle logs and tracing spans. + std::string GetAmqpComponentIdentifier(AmqpDiagnosticsContext const& diagnosticsContext); + + // Uses the AMQP error-condition namespace to identify the failed protocol layer. Conditions + // without a connection or session scope are observed at the link operation. + std::string GetAmqpFailureComponentType(std::string const& errorCondition); + + // Formats a lifecycle record. Kept separate from LogAmqpLifecycle so its stable fields can be + // covered without replacing the process-wide logger listener in a unit test. + std::string FormatAmqpLifecycleEvent( + AmqpDiagnosticsContext const& diagnosticsContext, + std::string const& eventName, + std::string const& detail = {}); + + void LogAmqpLifecycle( + Azure::Core::Diagnostics::Logger::Level level, + AmqpDiagnosticsContext const& diagnosticsContext, + std::string const& eventName, + std::string const& detail = {}); + + template + void CloseAmqpComponent( + AmqpDiagnosticsContext const& diagnosticsContext, + bool discarded, + std::string const& detail, + CloseFunction close, + LogFunction log) + { + log(discarded ? Azure::Core::Diagnostics::Logger::Level::Informational + : Azure::Core::Diagnostics::Logger::Level::Verbose, + diagnosticsContext, + discarded ? "discarded" : "closing", + detail); + try + { + close(); + if (!discarded) + { + log(Azure::Core::Diagnostics::Logger::Level::Verbose, diagnosticsContext, "closed", {}); + } + } + catch (std::exception const& ex) + { + log(Azure::Core::Diagnostics::Logger::Level::Warning, + diagnosticsContext, + "close_failed", + ex.what()); + } + } + +}}}} // namespace Azure::Messaging::EventHubs::_detail diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_tracing.hpp b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_tracing.hpp new file mode 100644 index 0000000000..ed63a45fba --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_tracing.hpp @@ -0,0 +1,62 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +// Distributed tracing helpers shared by the Event Hubs clients. +#pragma once + +#include "eventhubs_diagnostics.hpp" + +#include +#include +#include + +#include +#include + +namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail { + + enum class MessagingEntityKind + { + Destination, + Source, + }; + + // Creates the tracing context factory of a client. The provider can be null. + Azure::Core::Tracing::_internal::TracingContextFactory CreateTracingContextFactory( + std::shared_ptr const& tracingProvider); + + // Starts a span for an operation and adds the messaging attributes. The message count is + // optional, because a receiver knows it only on return. + Azure::Core::Tracing::_internal::TracingContextFactory::TracingContext StartSpan( + Azure::Core::Tracing::_internal::TracingContextFactory const& tracingFactory, + std::string const& spanName, + Azure::Core::Tracing::_internal::SpanKind spanKind, + std::string const& operationName, + std::string const& eventHubName, + std::string const& fullyQualifiedNamespace, + Azure::Nullable messageCount, + Azure::Core::Context const& context, + MessagingEntityKind entityKind = MessagingEntityKind::Destination); + + // Starts a child span around one call into the AMQP stack. Its duration separates the remote + // operation or protocol handshake from the duration of the enclosing SDK operation span. + Azure::Core::Tracing::_internal::TracingContextFactory::TracingContext StartAmqpSpan( + Azure::Core::Tracing::_internal::TracingContextFactory const& tracingFactory, + std::string const& spanName, + std::string const& operationName, + std::string const& eventHubName, + std::string const& fullyQualifiedNamespace, + Azure::Nullable messageCount, + AmqpDiagnosticsContext const& diagnosticsContext, + Azure::Nullable retryAttempt, + Azure::Core::Context const& context, + MessagingEntityKind entityKind = MessagingEntityKind::Destination); + + // Adds the message count attribute to a span. The factory supplies the attribute set that + // carries the count as an unsigned integer. + void SetMessageCount( + Azure::Core::Tracing::_internal::TracingContextFactory const& tracingFactory, + Azure::Core::Tracing::_internal::ServiceSpan& span, + size_t messageCount); + +}}}} // namespace Azure::Messaging::EventHubs::_detail diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp index 78fd0d68ac..b207b975e1 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp @@ -136,8 +136,13 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail Azure::Core::Amqp::_internal::Session const& session, std::string const& partitionUrl, std::string const& receiverName, + std::string clientIdentifier, + std::string partitionId, PartitionClientOptions options, Azure::Core::Http::Policies::RetryOptions retryOptions, + Azure::Core::Tracing::_internal::TracingContextFactory tracingFactory, + std::string eventHubName, + std::string fullyQualifiedNamespace, Azure::Core::Context const& context); PartitionClientFactory() = delete; }; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp b/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp index 0fc527e9a3..49ac042db8 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/producer_client.cpp @@ -6,7 +6,10 @@ #include "azure/messaging/eventhubs/event_data_batch.hpp" #include "azure/messaging/eventhubs/eventhubs_exception.hpp" #include "private/eventhubs_constants.hpp" +#include "private/eventhubs_diagnostics.hpp" +#include "private/eventhubs_tracing.hpp" #include "private/eventhubs_utilities.hpp" +#include "private/package_version.hpp" #include "private/retry_operation.hpp" #include @@ -24,15 +27,63 @@ using namespace Azure::Core::Diagnostics::_internal; using namespace Azure::Core::Diagnostics; namespace { const std::string DefaultAuthScope = "https://eventhubs.azure.net/.default"; + +Azure::Messaging::EventHubs::_detail::AmqpDiagnosticsContext CreateDiagnosticsContext( + std::string const& clientId, + std::string const& partitionId, + std::string componentType, + std::string componentName, + std::uint64_t stackId) +{ + return Azure::Messaging::EventHubs::_detail::AmqpDiagnosticsContext{ + clientId, partitionId, std::move(componentType), std::move(componentName), stackId}; +} + +std::string CreateConnectionName( + std::string const& clientId, + std::string const& applicationId, + std::string const& partitionId, + std::uint64_t stackId) +{ + return clientId + (applicationId.empty() ? std::string{} : "/application/" + applicationId) + + "/partition/" + (partitionId.empty() ? std::string("") : partitionId) + + "/generation/" + std::to_string(stackId); } +Azure::Messaging::EventHubs::_detail::AmqpDiagnosticsContext CreateFailureDiagnosticsContext( + std::string const& clientId, + std::string const& partitionId, + std::string const& linkName, + std::string const& applicationId, + std::uint64_t stackId, + std::string const& errorCondition) +{ + auto const componentType + = Azure::Messaging::EventHubs::_detail::GetAmqpFailureComponentType(errorCondition); + auto componentName = linkName; + if (componentType == "connection") + { + componentName = CreateConnectionName(clientId, applicationId, partitionId, stackId); + } + else if (componentType == "session") + { + componentName = partitionId; + } + return CreateDiagnosticsContext( + clientId, partitionId, componentType, std::move(componentName), stackId); +} +} // namespace + namespace Azure { namespace Messaging { namespace EventHubs { ProducerClient::ProducerClient( std::string const& connectionString, std::string const& eventHub, Azure::Messaging::EventHubs::ProducerClientOptions options) - : m_connectionString{connectionString}, m_eventHub{eventHub}, m_producerClientOptions(options) + : m_connectionString{connectionString}, m_eventHub{eventHub}, + m_producerClientOptions(options), + m_clientIdentifier{_detail::CreateClientIdentifier("producer", options.Name)}, + m_tracingFactory{_detail::CreateTracingContextFactory(options.TracingProvider)} { auto details = _detail::EventHubsUtilities::CreateConnectionStringDetails(connectionString, eventHub); @@ -42,6 +93,10 @@ namespace Azure { namespace Messaging { namespace EventHubs { m_targetPort = details.Port; m_targetUrl = details.ServiceScheme + m_fullyQualifiedNamespace + ":" + std::to_string(m_targetPort) + "/" + m_eventHub; + if (m_producerClientOptions.Name.empty()) + { + m_producerClientOptions.Name = m_clientIdentifier; + } } ProducerClient::ProducerClient( @@ -51,8 +106,14 @@ namespace Azure { namespace Messaging { namespace EventHubs { Azure::Messaging::EventHubs::ProducerClientOptions options) : m_fullyQualifiedNamespace{fullyQualifiedNamespace}, m_eventHub{eventHub}, m_targetUrl{_detail::EventHubsServiceScheme + m_fullyQualifiedNamespace + "/" + m_eventHub}, - m_credential{credential}, m_producerClientOptions(options) + m_credential{credential}, m_producerClientOptions(options), + m_clientIdentifier{_detail::CreateClientIdentifier("producer", options.Name)}, + m_tracingFactory{_detail::CreateTracingContextFactory(options.TracingProvider)} { + if (m_producerClientOptions.Name.empty()) + { + m_producerClientOptions.Name = m_clientIdentifier; + } } ProducerClient::~ProducerClient() @@ -141,7 +202,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { << "Could not read the maximum message size from the cached sender for partition '" << (options.PartitionId.empty() ? std::string("") : options.PartitionId) << "'. Discard the stack and build it again: " << ex.what() << std::endl; - InvalidateSender(options.PartitionId, observedGeneration, context); + InvalidateSender(options.PartitionId, observedGeneration, context, std::string{ex.what()}); EstablishSenderWithRetry(options.PartitionId, context); std::uint64_t rebuiltGeneration = 0; optionsToUse.MaxBytes = readMaxMessageSize(rebuiltGeneration); @@ -151,8 +212,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { return _detail::EventDataBatchFactory::CreateEventDataBatch(optionsToUse); } - void ProducerClient::Send(EventDataBatch const& eventDataBatch, Core::Context const& context) + void ProducerClient::SendBatchInSpan( + EventDataBatch const& eventDataBatch, + Azure::Core::Tracing::_internal::TracingContextFactory::TracingContext& tracingContext) { + // The conversion stays inside the span scope, because an empty batch fails here. auto message = eventDataBatch.ToAmqpMessage(); Azure::Messaging::EventHubs::_detail::RetryOperation retryOp( @@ -161,10 +225,13 @@ namespace Azure { namespace Messaging { namespace EventHubs { // are exhausted, but if the lambda ever returns false directly the batch must not be // silently dropped. See issue #7130. auto const& partitionId = eventDataBatch.GetPartitionId(); + std::uint64_t retryAttempt = 0; if (!retryOp.Execute( [&]() -> bool { - EnsureSenderOrInvalidate(partitionId, context); + auto const currentAttempt = ++retryAttempt; + EnsureSenderOrInvalidate(partitionId, tracingContext.Context); std::uint64_t observedGeneration = 0; + std::uint64_t observedStackId = 0; auto& guard = GetPartitionGuard(partitionId); try { @@ -172,24 +239,49 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::shared_lock stackLock(guard.stackLock); auto sender = GetSender(partitionId); observedGeneration = guard.generation.load(); - auto result = sender.Send(message, context); -#if ENABLE_UAMQP - auto sendStatus = std::get<0>(result); - if (sendStatus == Azure::Core::Amqp::_internal::MessageSendStatus::Ok) + observedStackId = guard.activeStackId.load(); + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, + partitionId, + "link", + m_producerClientOptions.Name, + observedStackId); + auto amqpTracingContext = _detail::StartAmqpSpan( + m_tracingFactory, + "ProducerClient.AmqpSend", + "publish", + m_eventHub, + m_fullyQualifiedNamespace, + eventDataBatch.NumberOfEvents(), + diagnosticsContext, + currentAttempt, + tracingContext.Context); + try { + auto result = sender.Send(message, amqpTracingContext.Context); +#if ENABLE_UAMQP + auto sendStatus = std::get<0>(result); + if (sendStatus == Azure::Core::Amqp::_internal::MessageSendStatus::Ok) + { + return true; + } + // Throw an exception about the error we just received. + throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: + CreateEventHubsException(std::get<1>(result)); +#elif ENABLE_RUST_AMQP + if (result) + { + throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: + CreateEventHubsException(result); + } return true; +#endif } - // Throw an exception about the error we just received. - throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: - CreateEventHubsException(std::get<1>(result)); -#elif ENABLE_RUST_AMQP - if (result) + catch (std::exception const& ex) { - throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: - CreateEventHubsException(result); + amqpTracingContext.Span.AddEvent(ex); + throw; } - return true; -#endif } catch (Azure::Core::OperationCancelledException const&) { @@ -197,22 +289,51 @@ namespace Azure { namespace Messaging { namespace EventHubs { } catch (Azure::Messaging::EventHubs::EventHubsException const& ex) { - if (!context.IsCancelled() && _detail::ShouldInvalidateSender(ex)) + if (!tracingContext.Context.IsCancelled() && _detail::ShouldInvalidateSender(ex)) { - InvalidateSender(partitionId, observedGeneration, context); + _detail::LogAmqpLifecycle( + Logger::Level::Warning, + CreateFailureDiagnosticsContext( + m_clientIdentifier, + partitionId, + m_producerClientOptions.Name, + m_producerClientOptions.ApplicationID, + observedStackId, + ex.ErrorCondition), + "failed", + ex.what()); + InvalidateSender( + partitionId, + observedGeneration, + tracingContext.Context, + std::string{ex.what()}); } throw; } - catch (std::exception const&) + catch (std::exception const& ex) { - if (!context.IsCancelled()) + if (!tracingContext.Context.IsCancelled()) { - InvalidateSender(partitionId, observedGeneration, context); + _detail::LogAmqpLifecycle( + Logger::Level::Warning, + CreateDiagnosticsContext( + m_clientIdentifier, + partitionId, + "link", + m_producerClientOptions.Name, + observedStackId), + "failed", + ex.what()); + InvalidateSender( + partitionId, + observedGeneration, + tracingContext.Context, + std::string{ex.what()}); } throw; } }, - context)) + tracingContext.Context)) { std::string failureDetail = "ProducerClient::Send failed after exhausting " + std::to_string(m_producerClientOptions.RetryOptions.MaxRetries) @@ -227,36 +348,100 @@ namespace Azure { namespace Messaging { namespace EventHubs { } } + void ProducerClient::Send(EventDataBatch const& eventDataBatch, Core::Context const& context) + { + auto tracingContext = _detail::StartSpan( + m_tracingFactory, + "ProducerClient.Send", + Azure::Core::Tracing::_internal::SpanKind::Producer, + "publish", + m_eventHub, + m_fullyQualifiedNamespace, + eventDataBatch.NumberOfEvents(), + context); + + try + { + SendBatchInSpan(eventDataBatch, tracingContext); + } + catch (std::exception const& ex) + { + tracingContext.Span.AddEvent(ex); + throw; + } + } + void ProducerClient::Send(Models::EventData const& eventData, Core::Context const& context) { - auto batch = CreateBatch(EventDataBatchOptions{}, context); - if (!batch.TryAdd(eventData)) + // The batch creation opens the AMQP link, so it must run inside the span. + auto tracingContext = _detail::StartSpan( + m_tracingFactory, + "ProducerClient.Send", + Azure::Core::Tracing::_internal::SpanKind::Producer, + "publish", + m_eventHub, + m_fullyQualifiedNamespace, + Azure::Nullable{}, + context); + + try { - throw std::runtime_error("Could not add message to batch."); + auto batch = CreateBatch(EventDataBatchOptions{}, tracingContext.Context); + if (!batch.TryAdd(eventData)) + { + throw std::runtime_error("Could not add message to batch."); + } + SendBatchInSpan(batch, tracingContext); + } + catch (std::exception const& ex) + { + tracingContext.Span.AddEvent(ex); + throw; } - Send(batch, context); } void ProducerClient::Send( std::vector const& eventData, Core::Context const& context) { - auto batch = CreateBatch(EventDataBatchOptions{}, context); - for (const auto& data : eventData) + // The batch creation opens the AMQP link, so it must run inside the span. + auto tracingContext = _detail::StartSpan( + m_tracingFactory, + "ProducerClient.Send", + Azure::Core::Tracing::_internal::SpanKind::Producer, + "publish", + m_eventHub, + m_fullyQualifiedNamespace, + eventData.size(), + context); + + try { - if (!batch.TryAdd(data)) + auto batch = CreateBatch(EventDataBatchOptions{}, tracingContext.Context); + for (const auto& data : eventData) { - throw std::runtime_error("Could not add message to batch."); + if (!batch.TryAdd(data)) + { + throw std::runtime_error("Could not add message to batch."); + } } + SendBatchInSpan(batch, tracingContext); + } + catch (std::exception const& ex) + { + tracingContext.Span.AddEvent(ex); + throw; } - Send(batch, context); } Azure::Core::Amqp::_internal::Connection ProducerClient::CreateConnection( + std::string const& partitionId, + std::uint64_t stackId, Azure::Core::Context const& context) const { Azure::Core::Amqp::_internal::ConnectionOptions connectOptions; - connectOptions.ContainerId = m_producerClientOptions.ApplicationID; + connectOptions.ContainerId = CreateConnectionName( + m_clientIdentifier, m_producerClientOptions.ApplicationID, partitionId, stackId); connectOptions.EnableTrace = _detail::EnableAmqpTrace; connectOptions.AuthenticationScopes = {"https://eventhubs.azure.net/.default"}; connectOptions.Port = m_targetPort; @@ -283,7 +468,33 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::unique_lock lock(m_sessionsLock); if (m_connections.find(partitionId) == m_connections.end()) { - m_connections.emplace(partitionId, CreateConnection(context)); + auto& guard = GetPartitionGuard(partitionId); + auto const stackId = guard.nextStackId.fetch_add(1) + 1; + guard.activeStackId = stackId; + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, + partitionId, + "connection", + CreateConnectionName( + m_clientIdentifier, m_producerClientOptions.ApplicationID, partitionId, stackId), + stackId); + _detail::LogAmqpLifecycle( + Logger::Level::Verbose, diagnosticsContext, stackId == 1 ? "creating" : "recreating"); + try + { + m_connections.emplace(partitionId, CreateConnection(partitionId, stackId, context)); + _detail::LogAmqpLifecycle( + stackId == 1 ? Logger::Level::Verbose : Logger::Level::Informational, + diagnosticsContext, + stackId == 1 ? "created" : "recreated"); + } + catch (std::exception const& ex) + { + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "create_failed", ex.what()); + guard.activeStackId = 0; + throw; + } } } @@ -298,7 +509,25 @@ namespace Azure { namespace Messaging { namespace EventHubs { std::unique_lock lock(m_sessionsLock); if (m_sessions.find(partitionId) == m_sessions.end()) { - m_sessions.emplace(partitionId, CreateSession(partitionId, context)); + auto const stackId = GetPartitionGuard(partitionId).activeStackId.load(); + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, partitionId, "session", partitionId, stackId); + _detail::LogAmqpLifecycle( + Logger::Level::Verbose, diagnosticsContext, stackId == 1 ? "creating" : "recreating"); + try + { + m_sessions.emplace(partitionId, CreateSession(partitionId, context)); + _detail::LogAmqpLifecycle( + stackId == 1 ? Logger::Level::Verbose : Logger::Level::Informational, + diagnosticsContext, + stackId == 1 ? "created" : "recreated"); + } + catch (std::exception const& ex) + { + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "create_failed", ex.what()); + throw; + } } } @@ -328,19 +557,45 @@ namespace Azure { namespace Messaging { namespace EventHubs { senderOptions.EnableTrace = _detail::EnableAmqpTrace; senderOptions.MaxMessageSize = m_producerClientOptions.MaxMessageSize; - Azure::Core::Amqp::_internal::MessageSender sender - = GetSession(partitionId).CreateMessageSender(targetUrl, senderOptions); - auto openResult{sender.Open(context)}; - if (openResult) + auto const stackId = GetPartitionGuard(partitionId).activeStackId.load(); + auto diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, partitionId, "link", senderOptions.Name, stackId); + _detail::LogAmqpLifecycle( + Logger::Level::Verbose, diagnosticsContext, stackId == 1 ? "attaching" : "reattaching"); + auto amqpTracingContext = _detail::StartAmqpSpan( + m_tracingFactory, + "ProducerClient.AmqpLink.Open", + "publish", + m_eventHub, + m_fullyQualifiedNamespace, + Azure::Nullable{}, + diagnosticsContext, + Azure::Nullable{}, + context); + try { - Azure::Core::Diagnostics::_internal::Log::Stream( - Azure::Core::Diagnostics::Logger::Level::Error) - << "Failed to create message sender: " << openResult; - throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: - CreateEventHubsException(openResult); + Azure::Core::Amqp::_internal::MessageSender sender + = GetSession(partitionId).CreateMessageSender(targetUrl, senderOptions); + auto openResult{sender.Open(amqpTracingContext.Context)}; + if (openResult) + { + throw Azure::Messaging::EventHubs::_detail::EventHubsExceptionFactory:: + CreateEventHubsException(openResult); + } + m_senders.emplace(partitionId, std::move(sender)); + GetPartitionGuard(partitionId).generation.fetch_add(1); + _detail::LogAmqpLifecycle( + stackId == 1 ? Logger::Level::Verbose : Logger::Level::Informational, + diagnosticsContext, + stackId == 1 ? "attached" : "reattached"); + } + catch (std::exception const& ex) + { + amqpTracingContext.Span.AddEvent(ex); + _detail::LogAmqpLifecycle( + Logger::Level::Warning, diagnosticsContext, "attach_failed", ex.what()); + throw; } - m_senders.emplace(partitionId, std::move(sender)); - GetPartitionGuard(partitionId).generation.fetch_add(1); } } void ProducerClient::EnsureSenderOrInvalidate( @@ -357,12 +612,12 @@ namespace Azure { namespace Messaging { namespace EventHubs { { throw; } - catch (std::exception const&) + catch (std::exception const& ex) { // No exemption for AuthenticationException: on uAMQP it can mean a dead $cbs link (#7330). if (!context.IsCancelled()) { - InvalidateSender(partitionId, observedGeneration, context); + InvalidateSender(partitionId, observedGeneration, context, std::string{ex.what()}); } throw; } @@ -435,28 +690,28 @@ namespace Azure { namespace Messaging { namespace EventHubs { void ProducerClient::InvalidateSender( std::string const& partitionId, Azure::Nullable observedGeneration, - Azure::Core::Context const& context) + Azure::Core::Context const& context, + Azure::Nullable failureReason) { std::unique_ptr sender; std::unique_ptr session; std::unique_ptr connection; + std::uint64_t stackId = 0; { auto& guard = GetPartitionGuard(partitionId); std::unique_lock stackLock(guard.stackLock); + stackId = guard.activeStackId.load(); if (observedGeneration.HasValue() && guard.generation.load() != observedGeneration.Value()) { - Log::Stream(Logger::Level::Informational) - << "Skip the teardown for partition '" - << (partitionId.empty() ? std::string("") : partitionId) - << "': the cached stack changed." << std::endl; + _detail::LogAmqpLifecycle( + Logger::Level::Informational, + CreateDiagnosticsContext(m_clientIdentifier, partitionId, "stack", {}, stackId), + "teardown_skipped", + "the cached stack changed"); return; } - Log::Stream(Logger::Level::Informational) - << "Discard the sender stack for partition '" - << (partitionId.empty() ? std::string("") : partitionId) << "'." << std::endl; - std::lock_guard sendersLock(m_sendersLock); std::lock_guard sessionsLock(m_sessionsLock); // Test again under the map lock. EnsureSender bumps the generation under that @@ -465,10 +720,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { // the fresh stack instead of tearing it down. if (observedGeneration.HasValue() && guard.generation.load() != observedGeneration.Value()) { - Log::Stream(Logger::Level::Informational) - << "Skip the teardown for partition '" - << (partitionId.empty() ? std::string("") : partitionId) - << "': another thread cached a new stack." << std::endl; + _detail::LogAmqpLifecycle( + Logger::Level::Informational, + CreateDiagnosticsContext(m_clientIdentifier, partitionId, "stack", {}, stackId), + "teardown_skipped", + "another thread cached a new stack"); return; } auto senderIterator = m_senders.find(partitionId); @@ -495,46 +751,50 @@ namespace Azure { namespace Messaging { namespace EventHubs { if (sender || session || connection) { guard.generation.fetch_add(1); + guard.activeStackId = 0; } } - // The network closes run outside every lock, so no send waits on them. + // The network closes and their lifecycle logs run outside every lock, so no send waits on them. + auto const detail = failureReason.HasValue() ? failureReason.Value() : std::string{}; if (sender) { - try - { - sender->Close(context); - } - catch (std::exception const& ex) - { - Log::Stream(Logger::Level::Warning) - << "Exception while closing a faulted message sender: " << ex.what() << std::endl; - } + auto const diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, partitionId, "link", m_producerClientOptions.Name, stackId); + _detail::CloseAmqpComponent( + diagnosticsContext, + failureReason.HasValue(), + detail, + [&sender, &context]() { sender->Close(context); }, + _detail::LogAmqpLifecycle); } #if ENABLE_RUST_AMQP if (session) { - try - { - session->End(context); - } - catch (std::exception const& ex) - { - Log::Stream(Logger::Level::Warning) - << "Exception while ending a faulted session: " << ex.what() << std::endl; - } + auto const diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, partitionId, "session", partitionId, stackId); + _detail::CloseAmqpComponent( + diagnosticsContext, + failureReason.HasValue(), + detail, + [&session, &context]() { session->End(context); }, + _detail::LogAmqpLifecycle); } if (connection) { - try - { - connection->Close(context); - } - catch (std::exception const& ex) - { - Log::Stream(Logger::Level::Warning) - << "Exception while closing a faulted connection: " << ex.what() << std::endl; - } + auto const diagnosticsContext = CreateDiagnosticsContext( + m_clientIdentifier, + partitionId, + "connection", + CreateConnectionName( + m_clientIdentifier, m_producerClientOptions.ApplicationID, partitionId, stackId), + stackId); + _detail::CloseAmqpComponent( + diagnosticsContext, + failureReason.HasValue(), + detail, + [&connection, &context]() { connection->Close(context); }, + _detail::LogAmqpLifecycle); } #endif diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt index b21bcfe95b..c7fd374728 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt @@ -26,6 +26,8 @@ add_executable ( eventhubs_admin_client.cpp eventhubs_admin_client.hpp eventhubs_test_base.hpp + eventhubs_tracing_test.cpp + eventhubs_tracing_test_doubles.hpp processor_load_balancer_test.cpp processor_test.cpp producer_client_test.cpp diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/consumer_client_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/consumer_client_test.cpp index 282c998f42..35c3c988a5 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/consumer_client_test.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/consumer_client_test.cpp @@ -5,13 +5,16 @@ #include "eventhubs_admin_client.hpp" #include "eventhubs_test_base.hpp" +#include "eventhubs_tracing_test_doubles.hpp" #include #include +#include #include #include #include +#include #include #include @@ -88,6 +91,49 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { EXPECT_TRUE(events[0]->Offset.HasValue()); } + // The receive span needs a partition client, and a partition client needs a live AMQP link. + TEST_P(ConsumerClientTest, ReceiveEventsSpan_LIVEONLY_) + { + auto provider = std::make_shared(); + + Azure::Messaging::EventHubs::ConsumerClientOptions options; + options.ApplicationID + = std::string(testing::UnitTest::GetInstance()->current_test_info()->name()) + + " Application"; + options.Name = testing::UnitTest::GetInstance()->current_test_case()->name(); + options.TracingProvider = provider; + + auto client = CreateConsumerClient("", options); + + Azure::Messaging::EventHubs::PartitionClientOptions partitionOptions; + partitionOptions.StartPosition.Inclusive = true; + partitionOptions.StartPosition.Earliest = true; + + Azure::Messaging::EventHubs::PartitionClient partitionClient + = client->CreatePartitionClient("1", partitionOptions); + + auto events = partitionClient.ReceiveEvents(1); + ASSERT_FALSE(events.empty()); + + auto span = FindSpan(provider, "PartitionClient.ReceiveEvents"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("PartitionClient.ReceiveEvents", span->GetName()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Client, span->GetKind()); + + auto const& attributes = span->GetAttributes(); + ASSERT_EQ(1u, attributes.count("messaging.operation")); + EXPECT_EQ("receive", attributes.at("messaging.operation")); + ASSERT_EQ(1u, attributes.count("messaging.system")); + EXPECT_EQ("eventhubs", attributes.at("messaging.system")); + ASSERT_EQ(1u, attributes.count("az.namespace")); + EXPECT_EQ("Microsoft.EventHub", attributes.at("az.namespace")); + ASSERT_EQ(1u, attributes.count("messaging.source.name")); + EXPECT_EQ(GetEventHubName(), attributes.at("messaging.source.name")); + + ASSERT_EQ(1u, attributes.count("messaging.batch.message_count")); + EXPECT_EQ(std::to_string(events.size()), attributes.at("messaging.batch.message_count")); + } + TEST_P(ConsumerClientTest, GetEventHubProperties_LIVEONLY_) { std::string eventHubName{GetEventHubName()}; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/eventhubs_tracing_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/eventhubs_tracing_test.cpp new file mode 100644 index 0000000000..d59134a089 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/eventhubs_tracing_test.cpp @@ -0,0 +1,546 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#include "../src/private/eventhubs_diagnostics.hpp" +#include "../src/private/eventhubs_tracing.hpp" +#include "../src/private/eventhubs_utilities.hpp" +#include "../src/private/package_version.hpp" +#include "eventhubs_tracing_test_doubles.hpp" + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { + + namespace { + // The key below is fake text. cspell tokenizes it into fragments that are not words, so + // spell checking is disabled across this block. + // cspell: disable + constexpr const char* TestConnectionString + = "Endpoint=sb://fake.example.com/;SharedAccessKeyName=name;SharedAccessKey=key;" + "EntityPath=unit-test-eh"; + // cspell: enable + constexpr const char* TestEventHubName = "unit-test-eh"; + constexpr const char* TestFullyQualifiedNamespace = "fake.example.com"; + + EventDataBatch CreateTestBatch() + { + EventDataBatchOptions batchOptions; + batchOptions.MaxBytes = static_cast((std::numeric_limits::max)()); + EventDataBatch batch{_detail::EventDataBatchFactory::CreateEventDataBatch(batchOptions)}; + EXPECT_TRUE(batch.TryAdd(Models::EventData{"First message."})); + EXPECT_TRUE(batch.TryAdd(Models::EventData{"Second message."})); + return batch; + } + + EventDataBatch CreateEmptyTestBatch() + { + EventDataBatchOptions batchOptions; + batchOptions.MaxBytes = static_cast((std::numeric_limits::max)()); + return _detail::EventDataBatchFactory::CreateEventDataBatch(batchOptions); + } + + std::map ExpectedSendAttributes() + { + return std::map{ + {"az.namespace", "Microsoft.EventHub"}, + {"messaging.system", "eventhubs"}, + {"messaging.destination.name", TestEventHubName}, + {"messaging.operation", "publish"}, + {"messaging.batch.message_count", "2"}, + {"net.peer.name", TestFullyQualifiedNamespace}, + }; + } + } // namespace + + TEST(EventHubsTracingTest, ProducerOptionsTracingProviderDefaultsEmpty) + { + ProducerClientOptions options; + EXPECT_TRUE((std::is_same< + decltype(options.TracingProvider), + std::shared_ptr>::value)); + EXPECT_EQ(nullptr, options.TracingProvider); + + options.TracingProvider = std::make_shared(); + EXPECT_NE(nullptr, options.TracingProvider); + } + + TEST(EventHubsTracingTest, ConsumerOptionsTracingProviderDefaultsEmpty) + { + ConsumerClientOptions options; + EXPECT_TRUE((std::is_same< + decltype(options.TracingProvider), + std::shared_ptr>::value)); + EXPECT_EQ(nullptr, options.TracingProvider); + + options.TracingProvider = std::make_shared(); + EXPECT_NE(nullptr, options.TracingProvider); + } + + TEST(EventHubsTracingTest, ProducerBuildsTracerFromOptions) + { + auto provider = std::make_shared(); + ProducerClientOptions options; + options.TracingProvider = provider; + + ProducerClient client{TestConnectionString, TestEventHubName, options}; + + ASSERT_EQ(1u, provider->GetTracers().size()); + auto const& tracer = provider->GetTracers().front(); + EXPECT_EQ("azure-messaging-eventhubs-cpp", tracer->GetName()); + EXPECT_EQ(_detail::PackageVersion::ToString(), tracer->GetVersion()); + } + + TEST(EventHubsTracingTest, ConsumerBuildsTracerFromOptions) + { + auto provider = std::make_shared(); + ConsumerClientOptions options; + options.TracingProvider = provider; + + ConsumerClient client{TestConnectionString, TestEventHubName, DefaultConsumerGroup, options}; + + ASSERT_EQ(1u, provider->GetTracers().size()); + auto const& tracer = provider->GetTracers().front(); + EXPECT_EQ("azure-messaging-eventhubs-cpp", tracer->GetName()); + EXPECT_EQ(_detail::PackageVersion::ToString(), tracer->GetVersion()); + } + + TEST(EventHubsTracingTest, SendSpanOnCancelledContext) + { + auto provider = std::make_shared(); + ProducerClientOptions options; + options.TracingProvider = provider; + + ProducerClient client{TestConnectionString, TestEventHubName, options}; + EventDataBatch batch{CreateTestBatch()}; + + Azure::Core::Context cancelledContext; + cancelledContext.Cancel(); + EXPECT_THROW(client.Send(batch, cancelledContext), Azure::Core::OperationCancelledException); + + auto span = FindSpan(provider, "ProducerClient.Send"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("ProducerClient.Send", span->GetName()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Producer, span->GetKind()); + EXPECT_EQ(ExpectedSendAttributes(), span->GetAttributes()); + EXPECT_EQ(span->GetAttributes().end(), span->GetAttributes().find("server.address")); + + ASSERT_EQ(1u, span->GetEvents().size()); + EXPECT_NE(std::string::npos, span->GetEvents()[0].find("cancelled")); + + ASSERT_FALSE(span->GetStatuses().empty()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanStatus::Error, span->GetStatuses().back()); + } + + TEST(EventHubsTracingTest, SendSpanRecordsNonCancelException) + { + auto provider = std::make_shared(); + ProducerClientOptions options; + options.TracingProvider = provider; + + ProducerClient client{TestConnectionString, TestEventHubName, options}; + EventDataBatch batch{CreateTestBatch()}; + + EXPECT_THROW(client.Send(batch, Azure::Core::Context{}), std::runtime_error); + + auto span = FindSpan(provider, "ProducerClient.Send"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("ProducerClient.Send", span->GetName()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Producer, span->GetKind()); + EXPECT_EQ(ExpectedSendAttributes(), span->GetAttributes()); + + ASSERT_EQ(1u, span->GetEvents().size()); + EXPECT_FALSE(span->GetEvents()[0].empty()); + + ASSERT_FALSE(span->GetStatuses().empty()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanStatus::Error, span->GetStatuses().back()); + } + + // An empty batch fails in the conversion to an AMQP message. That failure must land in the + // send span like every other send failure. + TEST(EventHubsTracingTest, SendSpanRecordsEmptyBatchFailure) + { + auto provider = std::make_shared(); + ProducerClientOptions options; + options.TracingProvider = provider; + + ProducerClient client{TestConnectionString, TestEventHubName, options}; + EventDataBatch batch{CreateEmptyTestBatch()}; + + EXPECT_THROW(client.Send(batch, Azure::Core::Context{}), std::runtime_error); + + auto span = FindSpan(provider, "ProducerClient.Send"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("ProducerClient.Send", span->GetName()); + + ASSERT_EQ(1u, span->GetAttributes().count("messaging.batch.message_count")); + EXPECT_EQ("0", span->GetAttributes().at("messaging.batch.message_count")); + + ASSERT_EQ(1u, span->GetEvents().size()); + EXPECT_FALSE(span->GetEvents()[0].empty()); + + ASSERT_FALSE(span->GetStatuses().empty()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanStatus::Error, span->GetStatuses().back()); + } + + // The convenience overloads build the batch first, and that opens the AMQP link. A bad host + // fails there, before any batch exists, so the span must start before the batch. + TEST(EventHubsTracingTest, SendEventSpanRecordsCreateBatchFailure) + { + auto provider = std::make_shared(); + ProducerClientOptions options; + options.TracingProvider = provider; + + ProducerClient client{TestConnectionString, TestEventHubName, options}; + + EXPECT_THROW( + client.Send(Models::EventData{"Single message."}, Azure::Core::Context{}), + std::runtime_error); + + auto span = FindSpan(provider, "ProducerClient.Send"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("ProducerClient.Send", span->GetName()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Producer, span->GetKind()); + + EXPECT_EQ( + span->GetAttributes().end(), span->GetAttributes().find("messaging.batch.message_count")); + + ASSERT_EQ(1u, span->GetEvents().size()); + EXPECT_FALSE(span->GetEvents()[0].empty()); + + ASSERT_FALSE(span->GetStatuses().empty()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanStatus::Error, span->GetStatuses().back()); + } + + TEST(EventHubsTracingTest, SendEventVectorSpanRecordsCreateBatchFailure) + { + auto provider = std::make_shared(); + ProducerClientOptions options; + options.TracingProvider = provider; + + ProducerClient client{TestConnectionString, TestEventHubName, options}; + + std::vector const eventData{ + Models::EventData{"First message."}, + Models::EventData{"Second message."}, + Models::EventData{"Third message."}, + }; + + EXPECT_THROW(client.Send(eventData, Azure::Core::Context{}), std::runtime_error); + + auto span = FindSpan(provider, "ProducerClient.Send"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("ProducerClient.Send", span->GetName()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Producer, span->GetKind()); + + ASSERT_EQ(1u, span->GetAttributes().count("messaging.batch.message_count")); + EXPECT_EQ("3", span->GetAttributes().at("messaging.batch.message_count")); + + ASSERT_EQ(1u, span->GetAttributeTypes().count("messaging.batch.message_count")); + EXPECT_EQ(AttributeTypeUInt64, span->GetAttributeTypes().at("messaging.batch.message_count")); + + ASSERT_EQ(1u, span->GetEvents().size()); + EXPECT_FALSE(span->GetEvents()[0].empty()); + + ASSERT_FALSE(span->GetStatuses().empty()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanStatus::Error, span->GetStatuses().back()); + } + + // Characterization test. The convenience overloads must behave the same way when the caller + // supplies no tracing provider. + TEST(EventHubsTracingTest, SendEventWithoutProviderIsUnchanged) + { + ProducerClientOptions options; + ProducerClient client{TestConnectionString, TestEventHubName, options}; + + EXPECT_THROW( + client.Send(Models::EventData{"Single message."}, Azure::Core::Context{}), + std::runtime_error); + + std::vector const eventData{ + Models::EventData{"First message."}, + Models::EventData{"Second message."}, + Models::EventData{"Third message."}, + }; + EXPECT_THROW(client.Send(eventData, Azure::Core::Context{}), std::runtime_error); + } + + TEST(EventHubsTracingTest, ReceiveSpanShapeFromSharedHelper) + { + auto provider = std::make_shared(); + auto factory = _detail::CreateTracingContextFactory(provider); + + auto tracingContext = _detail::StartSpan( + factory, + "PartitionClient.ReceiveEvents", + Azure::Core::Tracing::_internal::SpanKind::Client, + "receive", + TestEventHubName, + TestFullyQualifiedNamespace, + Azure::Nullable{}, + Azure::Core::Context{}, + _detail::MessagingEntityKind::Source); + + auto span = FindSpan(provider, "PartitionClient.ReceiveEvents"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("PartitionClient.ReceiveEvents", span->GetName()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Client, span->GetKind()); + + std::map const expectedAttributes{ + {"az.namespace", "Microsoft.EventHub"}, + {"messaging.system", "eventhubs"}, + {"messaging.source.name", TestEventHubName}, + {"messaging.operation", "receive"}, + {"net.peer.name", TestFullyQualifiedNamespace}, + }; + EXPECT_EQ(expectedAttributes, span->GetAttributes()); + EXPECT_EQ(span->GetAttributes().end(), span->GetAttributes().find("server.address")); + EXPECT_EQ( + span->GetAttributes().end(), span->GetAttributes().find("messaging.batch.message_count")); + + _detail::SetMessageCount(factory, tracingContext.Span, 3); + ASSERT_EQ(1u, span->GetAttributes().count("messaging.batch.message_count")); + EXPECT_EQ("3", span->GetAttributes().at("messaging.batch.message_count")); + } + + // The OpenTelemetry semantic conventions define messaging.batch.message_count as an int, so + // the count must reach the span through a numeric overload. + TEST(EventHubsTracingTest, ReceiveSpanMessageCountIsUnsignedInteger) + { + auto provider = std::make_shared(); + auto factory = _detail::CreateTracingContextFactory(provider); + + auto tracingContext = _detail::StartSpan( + factory, + "PartitionClient.ReceiveEvents", + Azure::Core::Tracing::_internal::SpanKind::Client, + "receive", + TestEventHubName, + TestFullyQualifiedNamespace, + Azure::Nullable{}, + Azure::Core::Context{}); + + auto span = FindSpan(provider, "PartitionClient.ReceiveEvents"); + ASSERT_NE(nullptr, span); + + _detail::SetMessageCount(factory, tracingContext.Span, 3); + + ASSERT_EQ(1u, span->GetAttributes().count("messaging.batch.message_count")); + EXPECT_EQ("3", span->GetAttributes().at("messaging.batch.message_count")); + + ASSERT_EQ(1u, span->GetAttributeTypes().count("messaging.batch.message_count")); + EXPECT_EQ(AttributeTypeUInt64, span->GetAttributeTypes().at("messaging.batch.message_count")); + } + + TEST(EventHubsTracingTest, SendSpanMessageCountIsUnsignedInteger) + { + auto provider = std::make_shared(); + ProducerClientOptions options; + options.TracingProvider = provider; + + ProducerClient client{TestConnectionString, TestEventHubName, options}; + EventDataBatch batch{CreateTestBatch()}; + + Azure::Core::Context cancelledContext; + cancelledContext.Cancel(); + EXPECT_THROW(client.Send(batch, cancelledContext), Azure::Core::OperationCancelledException); + + auto span = FindSpan(provider, "ProducerClient.Send"); + ASSERT_NE(nullptr, span); + + ASSERT_EQ(1u, span->GetAttributes().count("messaging.batch.message_count")); + EXPECT_EQ("2", span->GetAttributes().at("messaging.batch.message_count")); + + ASSERT_EQ(1u, span->GetAttributeTypes().count("messaging.batch.message_count")); + EXPECT_EQ(AttributeTypeUInt64, span->GetAttributeTypes().at("messaging.batch.message_count")); + } + + TEST(EventHubsTracingTest, AmqpSpanIdentifiesComponentAndAttempt) + { + auto provider = std::make_shared(); + auto factory = _detail::CreateTracingContextFactory(provider); + auto operationContext = _detail::StartSpan( + factory, + "ProducerClient.Send", + Azure::Core::Tracing::_internal::SpanKind::Producer, + "publish", + TestEventHubName, + TestFullyQualifiedNamespace, + size_t{2}, + Azure::Core::Context{}); + + _detail::AmqpDiagnosticsContext diagnosticsContext{ + "producer:test-client", "2", "link", "sender-link", 3}; + auto amqpContext = _detail::StartAmqpSpan( + factory, + "ProducerClient.AmqpSend", + "publish", + TestEventHubName, + TestFullyQualifiedNamespace, + size_t{2}, + diagnosticsContext, + std::uint64_t{4}, + operationContext.Context); + + auto span = FindSpan(provider, "ProducerClient.AmqpSend"); + ASSERT_NE(nullptr, span); + EXPECT_TRUE(span->HasParent()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Client, span->GetKind()); + EXPECT_EQ("producer:test-client", span->GetAttributes().at("az.eventhubs.client.id")); + EXPECT_EQ("2", span->GetAttributes().at("az.eventhubs.partition.id")); + EXPECT_EQ("link", span->GetAttributes().at("az.eventhubs.amqp.component.type")); + EXPECT_EQ("sender-link", span->GetAttributes().at("az.eventhubs.amqp.component.name")); + EXPECT_EQ( + "producer:test-client/partition/2/generation/3/link", + span->GetAttributes().at("az.eventhubs.amqp.component.id")); + EXPECT_EQ("3", span->GetAttributes().at("az.eventhubs.amqp.component.generation")); + EXPECT_EQ("4", span->GetAttributes().at("az.eventhubs.retry.attempt")); + EXPECT_EQ( + AttributeTypeUInt64, + span->GetAttributeTypes().at("az.eventhubs.amqp.component.generation")); + EXPECT_EQ(AttributeTypeUInt64, span->GetAttributeTypes().at("az.eventhubs.retry.attempt")); + } + + TEST(EventHubsTracingTest, LifecycleLogIdentifiesGatewayComponent) + { + _detail::AmqpDiagnosticsContext diagnosticsContext{ + "producer:test-client", "", "connection", "connection-1", 7}; + + EXPECT_EQ( + "Event Hubs AMQP lifecycle: event='create_failed' client.id='producer:test-client' " + "partition.id='' component.type='connection' component.name='connection-1' " + "component.id='producer:test-client/partition//generation/7/connection' " + "component.generation=7 detail='socket closed'", + _detail::FormatAmqpLifecycleEvent(diagnosticsContext, "create_failed", "socket closed")); + } + + TEST(EventHubsTracingTest, LifecycleLogEscapesQuotedAndMultilineValues) + { + _detail::AmqpDiagnosticsContext diagnosticsContext{ + "producer:client's", "2", "link", "sender\\link", 1}; + + EXPECT_NE( + std::string::npos, + _detail::FormatAmqpLifecycleEvent( + diagnosticsContext, "failed", "can't send\nconnection closed") + .find("client.id='producer:client\\'s'")); + EXPECT_NE( + std::string::npos, + _detail::FormatAmqpLifecycleEvent( + diagnosticsContext, "failed", "can't send\nconnection closed") + .find("detail='can\\'t send\\nconnection closed'")); + } + + TEST(EventHubsTracingTest, CloseAmqpComponentLogsClosedAfterSuccessfulClose) + { + _detail::AmqpDiagnosticsContext diagnosticsContext{ + "producer:test-client", "2", "link", "sender-link", 7}; + std::vector events; + auto log = [&events]( + Azure::Core::Diagnostics::Logger::Level, + _detail::AmqpDiagnosticsContext const&, + std::string const& eventName, + std::string const&) { events.push_back(eventName); }; + + _detail::CloseAmqpComponent( + diagnosticsContext, false, {}, [&events]() { events.push_back("close"); }, log); + + EXPECT_EQ((std::vector{"closing", "close", "closed"}), events); + } + + TEST(EventHubsTracingTest, CloseAmqpComponentLogsFailureWithoutClosed) + { + _detail::AmqpDiagnosticsContext diagnosticsContext{ + "producer:test-client", "2", "link", "sender-link", 7}; + std::vector events; + auto log = [&events]( + Azure::Core::Diagnostics::Logger::Level, + _detail::AmqpDiagnosticsContext const&, + std::string const& eventName, + std::string const&) { events.push_back(eventName); }; + + _detail::CloseAmqpComponent( + diagnosticsContext, + false, + {}, + [&events]() { + events.push_back("close"); + throw std::runtime_error("socket closed"); + }, + log); + + EXPECT_EQ((std::vector{"closing", "close", "close_failed"}), events); + } + + TEST(EventHubsTracingTest, ClientIdentifiersAreUniqueWithTheSameConfiguredName) + { + auto const first = _detail::CreateClientIdentifier("producer", "orders"); + auto const second = _detail::CreateClientIdentifier("producer", "orders"); + + EXPECT_NE(first, second); + EXPECT_EQ(0u, first.find("producer:orders:")); + EXPECT_EQ(0u, second.find("producer:orders:")); + } + + TEST(EventHubsTracingTest, FailureComponentTypeUsesAmqpErrorScope) + { + EXPECT_EQ("connection", _detail::GetAmqpFailureComponentType("amqp:connection:forced")); + EXPECT_EQ("session", _detail::GetAmqpFailureComponentType("amqp:session:window-violation")); + EXPECT_EQ("link", _detail::GetAmqpFailureComponentType("amqp:link:detach-forced")); + EXPECT_EQ("link", _detail::GetAmqpFailureComponentType("amqp:unauthorized-access")); + } + + // Characterization test. A factory with no tracer returns a null attribute set, and that + // pointer is not null safe. The message count path must stay quiet in that case. + TEST(EventHubsTracingTest, SetMessageCountWithoutTracerIsSafe) + { + auto factory = _detail::CreateTracingContextFactory(nullptr); + ASSERT_FALSE(factory.HasTracer()); + EXPECT_EQ(nullptr, factory.CreateAttributeSet()); + + auto tracingContext = _detail::StartSpan( + factory, + "PartitionClient.ReceiveEvents", + Azure::Core::Tracing::_internal::SpanKind::Client, + "receive", + TestEventHubName, + TestFullyQualifiedNamespace, + Azure::Nullable{}, + Azure::Core::Context{}); + + _detail::SetMessageCount(factory, tracingContext.Span, 3); + SUCCEED(); + } + + // Characterization test. The producer must behave the same way when the caller supplies no + // tracing provider. + TEST(EventHubsTracingTest, SendWithoutProviderIsUnchanged) + { + ProducerClientOptions options; + ProducerClient client{TestConnectionString, TestEventHubName, options}; + EXPECT_EQ(TestEventHubName, client.GetEventHubName()); + + EventDataBatch batch{CreateTestBatch()}; + + Azure::Core::Context cancelledContext; + cancelledContext.Cancel(); + EXPECT_THROW(client.Send(batch, cancelledContext), Azure::Core::OperationCancelledException); + + EXPECT_THROW(client.Send(batch, Azure::Core::Context{}), std::runtime_error); + } + +}}}} // namespace Azure::Messaging::EventHubs::Test diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/eventhubs_tracing_test_doubles.hpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/eventhubs_tracing_test_doubles.hpp new file mode 100644 index 0000000000..c8ea778cd6 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/eventhubs_tracing_test_doubles.hpp @@ -0,0 +1,239 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +// Recording tracing test doubles shared by the offline and the live tracing tests. +#pragma once + +#include +#include + +#include +#include +#include +#include +#include + +#include + +namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { + + // The type tag of the overload that delivered an attribute. A string "3" and a uint64 3 both + // stringify to "3", so the value alone cannot show the type on the wire. + constexpr const char* AttributeTypeBool = "bool"; + constexpr const char* AttributeTypeInt32 = "int32"; + constexpr const char* AttributeTypeInt64 = "int64"; + constexpr const char* AttributeTypeUInt64 = "uint64"; + constexpr const char* AttributeTypeDouble = "double"; + constexpr const char* AttributeTypeCString = "cstring"; + constexpr const char* AttributeTypeString = "string"; + + class TestAttributeSet final : public Azure::Core::Tracing::_internal::AttributeSet { + std::map m_attributes; + std::map m_attributeTypes; + + public: + TestAttributeSet() : Azure::Core::Tracing::_internal::AttributeSet() {} + + void AddAttribute(std::string const& key, bool value) override + { + m_attributes[key] = std::to_string(value); + m_attributeTypes[key] = AttributeTypeBool; + } + void AddAttribute(std::string const& key, int32_t value) override + { + m_attributes[key] = std::to_string(value); + m_attributeTypes[key] = AttributeTypeInt32; + } + void AddAttribute(std::string const& key, int64_t value) override + { + m_attributes[key] = std::to_string(value); + m_attributeTypes[key] = AttributeTypeInt64; + } + void AddAttribute(std::string const& key, uint64_t value) override + { + m_attributes[key] = std::to_string(value); + m_attributeTypes[key] = AttributeTypeUInt64; + } + void AddAttribute(std::string const& key, double value) override + { + m_attributes[key] = std::to_string(value); + m_attributeTypes[key] = AttributeTypeDouble; + } + void AddAttribute(std::string const& key, const char* value) override + { + m_attributes[key] = std::string(value); + m_attributeTypes[key] = AttributeTypeCString; + } + void AddAttribute(std::string const& key, std::string const& value) override + { + m_attributes[key] = value; + m_attributeTypes[key] = AttributeTypeString; + } + + std::map const& GetAttributes() const { return m_attributes; } + std::map const& GetAttributeTypes() const { return m_attributeTypes; } + }; + + class TestSpan final : public Azure::Core::Tracing::_internal::Span { + std::string m_spanName; + Azure::Core::Tracing::_internal::SpanKind m_kind; + std::map m_attributes; + std::map m_attributeTypes; + std::vector m_events; + std::vector m_statuses; + bool m_hasParent; + + void Merge(TestAttributeSet const& attributes) + { + for (auto const& attribute : attributes.GetAttributes()) + { + m_attributes[attribute.first] = attribute.second; + } + for (auto const& attributeType : attributes.GetAttributeTypes()) + { + m_attributeTypes[attributeType.first] = attributeType.second; + } + } + + public: + TestSpan( + std::string const& spanName, + Azure::Core::Tracing::_internal::CreateSpanOptions const& options) + : Azure::Core::Tracing::_internal::Span(), m_spanName(spanName), m_kind(options.Kind), + m_hasParent(options.ParentSpan != nullptr) + { + if (options.Attributes) + { + Merge(*static_cast(options.Attributes.get())); + } + } + + void AddAttributes(Azure::Core::Tracing::_internal::AttributeSet const& attributeToAdd) override + { + Merge(static_cast(attributeToAdd)); + } + void AddAttribute(std::string const& attributeName, std::string const& attributeValue) override + { + m_attributes[attributeName] = attributeValue; + m_attributeTypes[attributeName] = AttributeTypeString; + } + void AddEvent( + std::string const& eventName, + Azure::Core::Tracing::_internal::AttributeSet const&) override + { + m_events.push_back(eventName); + } + void AddEvent(std::string const& eventName) override { m_events.push_back(eventName); } + void AddEvent(std::exception const& ex) override { m_events.push_back(ex.what()); } + void SetStatus(Azure::Core::Tracing::_internal::SpanStatus const& status, std::string const&) + override + { + m_statuses.push_back(status); + } + void End(Azure::Nullable) override {} + void PropagateToHttpHeaders(Azure::Core::Http::Request&) override {} + + std::string const& GetName() const { return m_spanName; } + Azure::Core::Tracing::_internal::SpanKind GetKind() const { return m_kind; } + bool HasParent() const { return m_hasParent; } + std::map const& GetAttributes() const { return m_attributes; } + std::map const& GetAttributeTypes() const { return m_attributeTypes; } + std::vector const& GetEvents() const { return m_events; } + std::vector const& GetStatuses() const + { + return m_statuses; + } + }; + + class TestTracer final : public Azure::Core::Tracing::_internal::Tracer { + std::string m_name; + std::string m_version; + mutable std::vector> m_spans; + + public: + TestTracer(std::string const& name, std::string const& version) + : Azure::Core::Tracing::_internal::Tracer(), m_name(name), m_version(version) + { + } + + std::shared_ptr CreateSpan( + std::string const& spanName, + Azure::Core::Tracing::_internal::CreateSpanOptions const& options) const override + { + auto span = std::make_shared(spanName, options); + m_spans.push_back(span); + return span; + } + + // azure-core dereferences the returned attribute set when a tracer exists, so this must + // never return null. See sdk/core/azure-core/src/tracing/tracing.cpp. + std::unique_ptr CreateAttributeSet() + const override + { + return std::make_unique(); + } + + std::string const& GetName() const { return m_name; } + std::string const& GetVersion() const { return m_version; } + std::vector> const& GetSpans() const { return m_spans; } + }; + + class TestTracingProvider final : public Azure::Core::Tracing::TracerProvider { + mutable std::list> m_tracers; + + public: + TestTracingProvider() : Azure::Core::Tracing::TracerProvider() {} + ~TestTracingProvider() override {} + + std::shared_ptr CreateTracer( + std::string const& name, + std::string const& version) const override + { + auto tracer = std::make_shared(name, version); + m_tracers.push_back(tracer); + return tracer; + } + + std::list> const& GetTracers() const { return m_tracers; } + }; + + inline std::shared_ptr SingleSpan(std::shared_ptr const& provider) + { + EXPECT_EQ(1u, provider->GetTracers().size()); + if (provider->GetTracers().size() != 1u) + { + return nullptr; + } + auto const& tracer = provider->GetTracers().front(); + EXPECT_EQ(1u, tracer->GetSpans().size()); + if (tracer->GetSpans().size() != 1u) + { + return nullptr; + } + return tracer->GetSpans().front(); + } + + inline std::shared_ptr FindSpan( + std::shared_ptr const& provider, + std::string const& spanName) + { + EXPECT_EQ(1u, provider->GetTracers().size()); + if (provider->GetTracers().size() != 1u) + { + return nullptr; + } + + std::shared_ptr match; + for (auto const& span : provider->GetTracers().front()->GetSpans()) + { + if (span->GetName() == spanName) + { + EXPECT_EQ(nullptr, match) << "More than one span named " << spanName; + match = span; + } + } + EXPECT_NE(nullptr, match) << "No span named " << spanName; + return match; + } + +}}}} // namespace Azure::Messaging::EventHubs::Test diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp index 57847877d8..2029015ef3 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp @@ -4,10 +4,12 @@ // cspell: words #include "eventhubs_test_base.hpp" +#include "eventhubs_tracing_test_doubles.hpp" #include #include #include +#include #include #include #include @@ -16,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -107,6 +110,32 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { client->Send({{12, 13, 14, 15}, {16, 17, 18, 19}}); } + // The send success path needs a live AMQP link. It shows a second span when the convenience + // overload starts one span and then calls the batch overload. + TEST_P(ProducerClientTest, SendEventSpan_LIVEONLY_) + { + auto provider = std::make_shared(); + + Azure::Messaging::EventHubs::ProducerClientOptions producerOptions; + producerOptions.Name = testing::UnitTest::GetInstance()->current_test_case()->name(); + producerOptions.ApplicationID + = std::string(testing::UnitTest::GetInstance()->current_test_info()->name()) + + " Application"; + producerOptions.TracingProvider = provider; + + auto client{CreateProducerClient("", producerOptions)}; + + client->Send(Azure::Messaging::EventHubs::Models::EventData{"Single span test message"}); + + auto span = FindSpan(provider, "ProducerClient.Send"); + ASSERT_NE(nullptr, span); + EXPECT_EQ("ProducerClient.Send", span->GetName()); + EXPECT_EQ(Azure::Core::Tracing::_internal::SpanKind::Producer, span->GetKind()); + + auto const& attributes = span->GetAttributes(); + EXPECT_EQ(attributes.end(), attributes.find("messaging.batch.message_count")); + } + TEST_P(ProducerClientTest, GetEventHubProperties_LIVEONLY_) { Azure::Messaging::EventHubs::ProducerClientOptions producerOptions;