diff --git a/Cargo.lock b/Cargo.lock index 9391804..8c321b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1339,6 +1339,36 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.5", + "thiserror", + "tokio", +] + [[package]] name = "os_str_bytes" version = "6.6.1" @@ -2659,6 +2689,8 @@ dependencies = [ "lazy_static", "names", "nom", + "opentelemetry", + "opentelemetry_sdk", "prost", "prost-types", "rand 0.9.5", diff --git a/trogon-eventstore/Cargo.toml b/trogon-eventstore/Cargo.toml index 8d78df4..77befe6 100755 --- a/trogon-eventstore/Cargo.toml +++ b/trogon-eventstore/Cargo.toml @@ -37,6 +37,7 @@ hyper-util = { version = "0.1", features = ["client-legacy", "http2"] } hyper-rustls = { version = "0.27", features = ["rustls-native-certs", "http2"] } tracing = "0.1" nom = "7" +opentelemetry = { version = "0.32", default-features = false, features = ["trace"] } prost = "0.13" prost-types = "0.13" rand = { version = "0.9", features = ["small_rng"] } @@ -68,6 +69,7 @@ name = "integration" [dev-dependencies] names = "0.14" +opentelemetry_sdk = { version = "0.32", default-features = false, features = ["testing", "trace"] } serde = { version = "1", features = ["derive"] } testcontainers = "0.23" tokio = { version = "1", default-features = false, features = [ diff --git a/trogon-eventstore/src/batch.rs b/trogon-eventstore/src/batch.rs index 084e778..6680848 100644 --- a/trogon-eventstore/src/batch.rs +++ b/trogon-eventstore/src/batch.rs @@ -1,3 +1,4 @@ +use crate::observability::{client_operation, operation}; use crate::{EventData, Position, StreamState}; use tokio::sync::{ mpsc::{UnboundedReceiver, UnboundedSender}, @@ -140,29 +141,70 @@ impl BatchAppendClient { stream_state: StreamState, events: Vec, ) -> crate::Result { - let (sender, receiver) = oneshot::channel(); - let req = Req { - id: uuid::Uuid::new_v4(), - stream_name: stream_name.as_ref().to_string(), - events, - expected_revision: stream_state, - }; - - let req = In { sender, req }; - - if let Err(e) = self.sender.send(BatchMsg::In(req)) { - error!("[sending-end] Batch-append stream is closed: {}", e); - - let status = tonic::Status::cancelled("Batch-append stream has been closed"); - return Err(crate::Error::ServerError(status.to_string())); - } + client_operation(operation::BATCH_APPEND_TO_STREAM, async { + let (sender, receiver) = oneshot::channel(); + let req = Req { + id: uuid::Uuid::new_v4(), + stream_name: stream_name.as_ref().to_string(), + events, + expected_revision: stream_state, + }; + + let req = In { sender, req }; + + if let Err(e) = self.sender.send(BatchMsg::In(req)) { + error!("[sending-end] Batch-append stream is closed: {}", e); + + let status = tonic::Status::cancelled("Batch-append stream has been closed"); + return Err(crate::Error::ServerError(status.to_string())); + } - receiver.await.unwrap_or_else(|e| { - error!("[receiving-end] Batch-append stream is closed: {}", e); + receiver.await.unwrap_or_else(|e| { + error!("[receiving-end] Batch-append stream is closed: {}", e); - let status = tonic::Status::cancelled("Batch-append stream has been closed"); + let status = tonic::Status::cancelled("Batch-append stream has been closed"); - Err(crate::Error::ServerError(status.to_string())) + Err(crate::Error::ServerError(status.to_string())) + }) }) + .await + } +} + +#[cfg(test)] +mod tests { + use super::BatchAppendClient; + use crate::StreamState; + use crate::observability::operation; + use opentelemetry::global; + use opentelemetry::trace::{Status, noop::NoopTracerProvider}; + use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; + + #[tokio::test] + async fn batch_append_to_stream_emits_a_logical_client_span() { + let _guard = crate::observability::TEST_GLOBALS.lock().unwrap(); + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + global::set_tracer_provider(provider.clone()); + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + drop(receiver); + let client = BatchAppendClient { sender }; + + let result = client + .append_to_stream("stream", StreamState::Any, Vec::new()) + .await; + assert!(result.is_err()); + provider.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name == operation::BATCH_APPEND_TO_STREAM.span_name()) + .expect("batch append client span"); + assert!(matches!(span.status, Status::Error { .. })); + + global::set_tracer_provider(NoopTracerProvider::new()); } } diff --git a/trogon-eventstore/src/client.rs b/trogon-eventstore/src/client.rs index d45c02e..1a31817 100644 --- a/trogon-eventstore/src/client.rs +++ b/trogon-eventstore/src/client.rs @@ -1,5 +1,6 @@ use crate::batch::BatchAppendClient; use crate::grpc::{ClientSettings, GrpcClient}; +use crate::observability::{client_operation, infallible_client_operation, operation}; use crate::options::batch_append::BatchAppendOptions; use crate::options::persistent_subscription::PersistentSubscriptionOptions; use crate::options::read_all::ReadAllOptions; @@ -78,7 +79,11 @@ impl Client { where Events: ToEvents, { - commands::append_to_stream(&self.client, stream_name, options, events.into_events()).await + client_operation( + operation::APPEND_TO_STREAM, + commands::append_to_stream(&self.client, stream_name, options, events.into_events()), + ) + .await } // Sets a stream metadata. @@ -88,11 +93,19 @@ impl Client { options: &AppendToStreamOptions, metadata: &StreamMetadata, ) -> crate::Result { - let event = EventData::json("$metadata", metadata) - .map_err(|e| crate::Error::InternalParsingError(e.to_string()))?; - - self.append_to_stream(name.into_metadata_stream_name(), options, event) + client_operation(operation::SET_STREAM_METADATA, async { + let event = EventData::json("$metadata", metadata) + .map_err(|e| crate::Error::InternalParsingError(e.to_string()))?; + + commands::append_to_stream( + &self.client, + name.into_metadata_stream_name(), + options, + event.into_events(), + ) .await + }) + .await } // Creates a batch-append client. @@ -100,7 +113,11 @@ impl Client { &self, options: &BatchAppendOptions, ) -> crate::Result { - commands::batch_append(&self.client, options).await + client_operation( + operation::BATCH_APPEND, + commands::batch_append(&self.client, options), + ) + .await } /// Reads events from a given stream. The reading can be done forward and @@ -110,11 +127,14 @@ impl Client { stream_name: impl StreamName, options: &ReadStreamOptions, ) -> crate::Result { - commands::read_stream( - self.client.clone(), - options, - stream_name, - options.max_count as u64, + client_operation( + operation::READ_STREAM, + commands::read_stream( + self.client.clone(), + options, + stream_name, + options.max_count as u64, + ), ) .await } @@ -122,7 +142,11 @@ impl Client { /// Reads events for the system stream `$all`. The reading can be done /// forward and backward. pub async fn read_all(&self, options: &ReadAllOptions) -> crate::Result { - commands::read_all(self.client.clone(), options, options.max_count as u64).await + client_operation( + operation::READ_ALL, + commands::read_all(self.client.clone(), options, options.max_count as u64), + ) + .await } /// Reads a stream metadata. @@ -131,32 +155,39 @@ impl Client { name: impl MetadataStreamName, options: &ReadStreamOptions, ) -> crate::Result { - let mut stream = self - .read_stream(name.into_metadata_stream_name(), options) + client_operation(operation::GET_STREAM_METADATA, async { + let mut stream = commands::read_stream( + self.client.clone(), + options, + name.into_metadata_stream_name(), + options.max_count as u64, + ) .await?; - match stream.next().await { - Ok(event) => { - let event = event.expect("to be defined"); - let metadata = event - .get_original_event() - .as_json::() - .map_err(|e| crate::Error::InternalParsingError(e.to_string()))?; - - let metadata = VersionedMetadata { - stream: event.get_original_stream_id().to_string(), - version: event.get_original_event().revision, - metadata, - }; - - Ok(StreamMetadataResult::Success(Box::new(metadata))) + match stream.next().await { + Ok(event) => { + let event = event.expect("to be defined"); + let metadata = event + .get_original_event() + .as_json::() + .map_err(|e| crate::Error::InternalParsingError(e.to_string()))?; + + let metadata = VersionedMetadata { + stream: event.get_original_stream_id().to_string(), + version: event.get_original_event().revision, + metadata, + }; + + Ok(StreamMetadataResult::Success(Box::new(metadata))) + } + Err(e) => match e { + crate::Error::ResourceNotFound => Ok(StreamMetadataResult::NotFound), + crate::Error::ResourceDeleted => Ok(StreamMetadataResult::Deleted), + other => Err(other), + }, } - Err(e) => match e { - crate::Error::ResourceNotFound => Ok(StreamMetadataResult::NotFound), - crate::Error::ResourceDeleted => Ok(StreamMetadataResult::Deleted), - other => Err(other), - }, - } + }) + .await } /// Soft deletes a given stream. @@ -170,7 +201,11 @@ impl Client { stream_name: impl StreamName, options: &DeleteStreamOptions, ) -> crate::Result> { - commands::delete_stream(&self.client, stream_name, options).await + client_operation( + operation::DELETE_STREAM, + commands::delete_stream(&self.client, stream_name, options), + ) + .await } /// Hard deletes a given stream. @@ -183,7 +218,11 @@ impl Client { stream_name: impl StreamName, options: &TombstoneStreamOptions, ) -> crate::Result> { - commands::tombstone_stream(&self.client, stream_name, options).await + client_operation( + operation::TOMBSTONE_STREAM, + commands::tombstone_stream(&self.client, stream_name, options), + ) + .await } /// Subscribes to a given stream. This kind of subscription specifies a @@ -205,14 +244,20 @@ impl Client { stream_name: impl StreamName, options: &SubscribeToStreamOptions, ) -> Subscription { - commands::subscribe_to_stream(self.client.clone(), stream_name, options) + infallible_client_operation(operation::SUBSCRIBE_TO_STREAM, async { + commands::subscribe_to_stream(self.client.clone(), stream_name, options) + }) + .await } /// Like [`subscribe_to_stream`] but specific to system `$all` stream. /// /// [`subscribe_to_stream`]: #method.subscribe_to_stream pub async fn subscribe_to_all(&self, options: &SubscribeToAllOptions) -> Subscription { - commands::subscribe_to_all(self.client.clone(), options) + infallible_client_operation(operation::SUBSCRIBE_TO_ALL, async { + commands::subscribe_to_all(self.client.clone(), options) + }) + .await } /// Creates a persistent subscription group on a stream. @@ -227,11 +272,14 @@ impl Client { group_name: impl AsRef, options: &PersistentSubscriptionOptions, ) -> crate::Result<()> { - commands::create_persistent_subscription( - &self.client, - stream_name, - group_name.as_ref(), - options, + client_operation( + operation::CREATE_PERSISTENT_SUBSCRIPTION, + commands::create_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + ), ) .await } @@ -242,8 +290,16 @@ impl Client { group_name: impl AsRef, options: &PersistentSubscriptionToAllOptions, ) -> crate::Result<()> { - commands::create_persistent_subscription(&self.client, "", group_name.as_ref(), options) - .await + client_operation( + operation::CREATE_PERSISTENT_SUBSCRIPTION_TO_ALL, + commands::create_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + ), + ) + .await } /// Updates a persistent subscription group on a stream. @@ -253,11 +309,14 @@ impl Client { group_name: impl AsRef, options: &PersistentSubscriptionOptions, ) -> crate::Result<()> { - commands::update_persistent_subscription( - &self.client, - stream_name, - group_name.as_ref(), - options, + client_operation( + operation::UPDATE_PERSISTENT_SUBSCRIPTION, + commands::update_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + ), ) .await } @@ -268,8 +327,16 @@ impl Client { group_name: impl AsRef, options: &PersistentSubscriptionToAllOptions, ) -> crate::Result<()> { - commands::update_persistent_subscription(&self.client, "", group_name.as_ref(), options) - .await + client_operation( + operation::UPDATE_PERSISTENT_SUBSCRIPTION_TO_ALL, + commands::update_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + ), + ) + .await } /// Deletes a persistent subscription group on a stream. @@ -279,12 +346,15 @@ impl Client { group_name: impl AsRef, options: &DeletePersistentSubscriptionOptions, ) -> crate::Result<()> { - commands::delete_persistent_subscription( - &self.client, - stream_name, - group_name.as_ref(), - options, - false, + client_operation( + operation::DELETE_PERSISTENT_SUBSCRIPTION, + commands::delete_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + false, + ), ) .await } @@ -295,12 +365,15 @@ impl Client { group_name: impl AsRef, options: &DeletePersistentSubscriptionOptions, ) -> crate::Result<()> { - commands::delete_persistent_subscription( - &self.client, - "", - group_name.as_ref(), - options, - true, + client_operation( + operation::DELETE_PERSISTENT_SUBSCRIPTION_TO_ALL, + commands::delete_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + true, + ), ) .await } @@ -312,12 +385,15 @@ impl Client { group_name: impl AsRef, options: &SubscribeToPersistentSubscriptionOptions, ) -> crate::Result { - commands::subscribe_to_persistent_subscription( - &self.client, - stream_name, - group_name.as_ref(), - options, - false, + client_operation( + operation::SUBSCRIBE_TO_PERSISTENT_SUBSCRIPTION, + commands::subscribe_to_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + false, + ), ) .await } @@ -328,12 +404,15 @@ impl Client { group_name: impl AsRef, options: &SubscribeToPersistentSubscriptionOptions, ) -> crate::Result { - commands::subscribe_to_persistent_subscription( - &self.client, - "", - group_name.as_ref(), - options, - true, + client_operation( + operation::SUBSCRIBE_TO_PERSISTENT_SUBSCRIPTION_TO_ALL, + commands::subscribe_to_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + true, + ), ) .await } @@ -345,12 +424,15 @@ impl Client { group_name: impl AsRef, options: &ReplayParkedMessagesOptions, ) -> crate::Result<()> { - commands::replay_parked_messages( - &self.client, - &self.http_client, - commands::RegularStream(stream_name.as_ref().to_string()), - group_name, - options, + client_operation( + operation::REPLAY_PARKED_MESSAGES, + commands::replay_parked_messages( + &self.client, + &self.http_client, + commands::RegularStream(stream_name.as_ref().to_string()), + group_name, + options, + ), ) .await } @@ -361,12 +443,15 @@ impl Client { group_name: impl AsRef, options: &ReplayParkedMessagesOptions, ) -> crate::Result<()> { - commands::replay_parked_messages( - &self.client, - &self.http_client, - commands::AllStream, - group_name, - options, + client_operation( + operation::REPLAY_PARKED_MESSAGES_TO_ALL, + commands::replay_parked_messages( + &self.client, + &self.http_client, + commands::AllStream, + group_name, + options, + ), ) .await } @@ -376,7 +461,11 @@ impl Client { &self, options: &ListPersistentSubscriptionsOptions, ) -> crate::Result>> { - commands::list_all_persistent_subscriptions(&self.client, &self.http_client, options).await + client_operation( + operation::LIST_ALL_PERSISTENT_SUBSCRIPTIONS, + commands::list_all_persistent_subscriptions(&self.client, &self.http_client, options), + ) + .await } /// List all persistent subscriptions of a specific stream. @@ -385,11 +474,14 @@ impl Client { stream_name: impl AsRef, options: &ListPersistentSubscriptionsOptions, ) -> crate::Result>> { - commands::list_persistent_subscriptions_for_stream( - &self.client, - &self.http_client, - commands::RegularStream(stream_name.as_ref().to_string()), - options, + client_operation( + operation::LIST_PERSISTENT_SUBSCRIPTIONS_FOR_STREAM, + commands::list_persistent_subscriptions_for_stream( + &self.client, + &self.http_client, + commands::RegularStream(stream_name.as_ref().to_string()), + options, + ), ) .await } @@ -399,11 +491,14 @@ impl Client { &self, options: &ListPersistentSubscriptionsOptions, ) -> crate::Result>> { - commands::list_persistent_subscriptions_for_stream( - &self.client, - &self.http_client, - commands::AllStream, - options, + client_operation( + operation::LIST_PERSISTENT_SUBSCRIPTIONS_TO_ALL, + commands::list_persistent_subscriptions_for_stream( + &self.client, + &self.http_client, + commands::AllStream, + options, + ), ) .await } @@ -415,12 +510,15 @@ impl Client { group_name: impl AsRef, options: &GetPersistentSubscriptionInfoOptions, ) -> crate::Result> { - commands::get_persistent_subscription_info( - &self.client, - &self.http_client, - commands::RegularStream(stream_name.as_ref().to_string()), - group_name, - options, + client_operation( + operation::GET_PERSISTENT_SUBSCRIPTION_INFO, + commands::get_persistent_subscription_info( + &self.client, + &self.http_client, + commands::RegularStream(stream_name.as_ref().to_string()), + group_name, + options, + ), ) .await } @@ -431,12 +529,15 @@ impl Client { group_name: impl AsRef, options: &GetPersistentSubscriptionInfoOptions, ) -> crate::Result> { - commands::get_persistent_subscription_info( - &self.client, - &self.http_client, - commands::AllStream, - group_name, - options, + client_operation( + operation::GET_PERSISTENT_SUBSCRIPTION_INFO_TO_ALL, + commands::get_persistent_subscription_info( + &self.client, + &self.http_client, + commands::AllStream, + group_name, + options, + ), ) .await } @@ -446,10 +547,13 @@ impl Client { &self, options: &RestartPersistentSubscriptionSubsystem, ) -> crate::Result<()> { - commands::restart_persistent_subscription_subsystem( - &self.client, - &self.http_client, - options, + client_operation( + operation::RESTART_PERSISTENT_SUBSCRIPTION_SUBSYSTEM, + commands::restart_persistent_subscription_subsystem( + &self.client, + &self.http_client, + options, + ), ) .await } diff --git a/trogon-eventstore/src/lib.rs b/trogon-eventstore/src/lib.rs index defb09e..5c57f2a 100755 --- a/trogon-eventstore/src/lib.rs +++ b/trogon-eventstore/src/lib.rs @@ -61,6 +61,7 @@ mod commands; mod event_store; mod grpc; mod http; +mod observability; pub mod operations; mod options; mod private; diff --git a/trogon-eventstore/src/observability.rs b/trogon-eventstore/src/observability.rs new file mode 100644 index 0000000..9845fdc --- /dev/null +++ b/trogon-eventstore/src/observability.rs @@ -0,0 +1,247 @@ +use opentelemetry::trace::{FutureExt, SpanKind, Status, TraceContextExt, Tracer}; +use opentelemetry::{Context, InstrumentationScope, KeyValue, global}; +use std::future::Future; + +#[cfg(test)] +pub(crate) static TEST_GLOBALS: std::sync::Mutex<()> = std::sync::Mutex::new(()); + +#[derive(Clone, Copy)] +pub(crate) struct InstrumentationScopeIdentity { + pub(crate) name: &'static str, + pub(crate) version: &'static str, +} + +pub(crate) const INSTRUMENTATION_SCOPE: InstrumentationScopeIdentity = + InstrumentationScopeIdentity { + name: env!("CARGO_PKG_NAME"), + version: env!("CARGO_PKG_VERSION"), + }; + +#[derive(Clone, Copy)] +pub(crate) struct ClientOperation { + span_name: &'static str, + operation_name: &'static str, +} + +impl ClientOperation { + const fn new(span_name: &'static str, operation_name: &'static str) -> Self { + Self { + span_name, + operation_name, + } + } + + pub(crate) const fn span_name(self) -> &'static str { + self.span_name + } +} + +pub(crate) mod operation { + use super::ClientOperation; + + macro_rules! client_operation { + ($constant:ident, $name:literal) => { + pub(crate) const $constant: ClientOperation = + ClientOperation::new(concat!("trogon_eventstore.", $name), $name); + }; + } + + client_operation!(APPEND_TO_STREAM, "append_to_stream"); + client_operation!(SET_STREAM_METADATA, "set_stream_metadata"); + client_operation!(BATCH_APPEND, "batch_append"); + client_operation!(BATCH_APPEND_TO_STREAM, "batch_append_to_stream"); + client_operation!(READ_STREAM, "read_stream"); + client_operation!(READ_ALL, "read_all"); + client_operation!(GET_STREAM_METADATA, "get_stream_metadata"); + client_operation!(DELETE_STREAM, "delete_stream"); + client_operation!(TOMBSTONE_STREAM, "tombstone_stream"); + client_operation!(SUBSCRIBE_TO_STREAM, "subscribe_to_stream"); + client_operation!(SUBSCRIBE_TO_ALL, "subscribe_to_all"); + client_operation!( + CREATE_PERSISTENT_SUBSCRIPTION, + "create_persistent_subscription" + ); + client_operation!( + CREATE_PERSISTENT_SUBSCRIPTION_TO_ALL, + "create_persistent_subscription_to_all" + ); + client_operation!( + UPDATE_PERSISTENT_SUBSCRIPTION, + "update_persistent_subscription" + ); + client_operation!( + UPDATE_PERSISTENT_SUBSCRIPTION_TO_ALL, + "update_persistent_subscription_to_all" + ); + client_operation!( + DELETE_PERSISTENT_SUBSCRIPTION, + "delete_persistent_subscription" + ); + client_operation!( + DELETE_PERSISTENT_SUBSCRIPTION_TO_ALL, + "delete_persistent_subscription_to_all" + ); + client_operation!( + SUBSCRIBE_TO_PERSISTENT_SUBSCRIPTION, + "subscribe_to_persistent_subscription" + ); + client_operation!( + SUBSCRIBE_TO_PERSISTENT_SUBSCRIPTION_TO_ALL, + "subscribe_to_persistent_subscription_to_all" + ); + client_operation!(REPLAY_PARKED_MESSAGES, "replay_parked_messages"); + client_operation!( + REPLAY_PARKED_MESSAGES_TO_ALL, + "replay_parked_messages_to_all" + ); + client_operation!( + LIST_ALL_PERSISTENT_SUBSCRIPTIONS, + "list_all_persistent_subscriptions" + ); + client_operation!( + LIST_PERSISTENT_SUBSCRIPTIONS_FOR_STREAM, + "list_persistent_subscriptions_for_stream" + ); + client_operation!( + LIST_PERSISTENT_SUBSCRIPTIONS_TO_ALL, + "list_persistent_subscriptions_to_all" + ); + client_operation!( + GET_PERSISTENT_SUBSCRIPTION_INFO, + "get_persistent_subscription_info" + ); + client_operation!( + GET_PERSISTENT_SUBSCRIPTION_INFO_TO_ALL, + "get_persistent_subscription_info_to_all" + ); + client_operation!( + RESTART_PERSISTENT_SUBSCRIPTION_SUBSYSTEM, + "restart_persistent_subscription_subsystem" + ); +} + +fn instrumentation_scope() -> InstrumentationScope { + InstrumentationScope::builder(INSTRUMENTATION_SCOPE.name) + .with_version(INSTRUMENTATION_SCOPE.version) + .build() +} + +fn start_client_operation(operation: ClientOperation) -> Context { + let tracer = global::tracer_with_scope(instrumentation_scope()); + let span = tracer + .span_builder(operation.span_name()) + .with_kind(SpanKind::Client) + .with_attributes(vec![ + KeyValue::new("db.system.name", "trogon_eventstore"), + KeyValue::new("db.operation.name", operation.operation_name), + ]) + .start(&tracer); + + Context::current_with_span(span) +} + +pub(crate) async fn client_operation( + operation: ClientOperation, + future: F, +) -> crate::Result +where + F: Future>, +{ + let context = start_client_operation(operation); + let result = future.with_context(context.clone()).await; + + if let Err(error) = &result { + context.span().set_status(Status::error(error.to_string())); + } + context.span().end(); + + result +} + +pub(crate) async fn infallible_client_operation(operation: ClientOperation, future: F) -> T +where + F: Future, +{ + let context = start_client_operation(operation); + let output = future.with_context(context.clone()).await; + context.span().end(); + + output +} + +#[cfg(test)] +mod tests { + use super::{INSTRUMENTATION_SCOPE, client_operation, operation::APPEND_TO_STREAM}; + use opentelemetry::Context; + use opentelemetry::global; + use opentelemetry::trace::{SpanKind, Status, TraceContextExt, noop::NoopTracerProvider}; + use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; + + #[tokio::test] + async fn client_operation_emits_a_logical_client_span() { + let _guard = super::TEST_GLOBALS.lock().unwrap(); + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + global::set_tracer_provider(provider.clone()); + + let current_span_id = client_operation(APPEND_TO_STREAM, async { + Ok::<_, crate::Error>(Context::current().span().span_context().span_id()) + }) + .await + .unwrap(); + provider.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name == APPEND_TO_STREAM.span_name()) + .expect("client span"); + + assert_eq!(span.span_kind, SpanKind::Client); + assert_eq!(span.span_context.span_id(), current_span_id); + assert_eq!(span.status, Status::Unset); + assert_eq!( + span.instrumentation_scope.name(), + INSTRUMENTATION_SCOPE.name + ); + assert_eq!( + span.instrumentation_scope.version(), + Some(INSTRUMENTATION_SCOPE.version) + ); + assert!(span.attributes.iter().any(|attribute| { + attribute.key.as_str() == "db.system.name" + && attribute.value.to_string() == "trogon_eventstore" + })); + + global::set_tracer_provider(NoopTracerProvider::new()); + } + + #[tokio::test] + async fn client_operation_marks_failures_as_errors() { + let _guard = super::TEST_GLOBALS.lock().unwrap(); + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + global::set_tracer_provider(provider.clone()); + + let result = client_operation::<(), _>(APPEND_TO_STREAM, async { + Err(crate::Error::ConnectionClosed) + }) + .await; + assert!(result.is_err()); + provider.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name == APPEND_TO_STREAM.span_name()) + .expect("client span"); + + assert!(matches!(span.status, Status::Error { .. })); + + global::set_tracer_provider(NoopTracerProvider::new()); + } +} diff --git a/trogon-eventstore/src/request.rs b/trogon-eventstore/src/request.rs index 4bc51af..1139312 100644 --- a/trogon-eventstore/src/request.rs +++ b/trogon-eventstore/src/request.rs @@ -1,7 +1,31 @@ use crate::options::CommonOperationOptions; use crate::{Authentication, ClientSettings, Credentials, NodePreference}; use base64::Engine; +use opentelemetry::Context; +use opentelemetry::global; +use opentelemetry::propagation::Injector; use std::borrow::Cow; +use tonic::metadata::{Ascii, MetadataKey, MetadataMap, MetadataValue}; + +struct MetadataInjector<'a>(&'a mut MetadataMap); + +impl Injector for MetadataInjector<'_> { + fn set(&mut self, key: &str, value: String) { + let Ok(key) = MetadataKey::::from_bytes(key.as_bytes()) else { + tracing::warn!(key, "propagator produced an invalid gRPC metadata key"); + return; + }; + let Ok(value) = MetadataValue::::try_from(value.as_str()) else { + tracing::warn!( + key = key.as_str(), + "propagator produced an invalid gRPC metadata value" + ); + return; + }; + + self.0.insert(key, value); + } +} pub(crate) fn build_request_metadata( settings: &ClientSettings, @@ -9,9 +33,10 @@ pub(crate) fn build_request_metadata( ) -> tonic::metadata::MetadataMap where { - use tonic::metadata::MetadataValue; - let mut metadata = tonic::metadata::MetadataMap::new(); + global::get_text_map_propagator(|propagator| { + propagator.inject_context(&Context::current(), &mut MetadataInjector(&mut metadata)); + }); let authentication: Option> = options .authentication .as_ref() @@ -79,7 +104,12 @@ fn build_authorization_header( mod auth_tests { use super::*; use crate::AppendToStreamOptions; + use crate::observability::{client_operation, operation}; use crate::options::Options; + use opentelemetry::global; + use opentelemetry::trace::noop::{NoopTextMapPropagator, NoopTracerProvider}; + use opentelemetry_sdk::propagation::TraceContextPropagator; + use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; fn settings_from(connection_string: &str) -> ClientSettings { connection_string @@ -154,6 +184,45 @@ mod auth_tests { ); } + #[tokio::test] + async fn build_request_metadata_injects_the_current_client_context() { + let _guard = crate::observability::TEST_GLOBALS.lock().unwrap(); + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + global::set_tracer_provider(provider.clone()); + global::set_text_map_propagator(TraceContextPropagator::new()); + let settings = settings_from("esdb://localhost:2113?tls=false"); + let options = AppendToStreamOptions::default(); + + let metadata = client_operation(operation::APPEND_TO_STREAM, async { + Ok(build_request_metadata( + &settings, + options.common_operation_options(), + )) + }) + .await + .unwrap(); + provider.force_flush().unwrap(); + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name == operation::APPEND_TO_STREAM.span_name()) + .expect("client span"); + + assert_eq!( + metadata.get("traceparent").unwrap().to_str().unwrap(), + format!( + "00-{}-{}-01", + span.span_context.trace_id(), + span.span_context.span_id() + ) + ); + global::set_tracer_provider(NoopTracerProvider::new()); + global::set_text_map_propagator(NoopTextMapPropagator::new()); + } + #[test] fn authenticated_builder_accepts_credentials_directly() { let settings = settings_from("esdb://localhost:2113?tls=false");