diff --git a/.config/mise/tasks/semconv/check b/.config/mise/tasks/semconv/check new file mode 100755 index 0000000..38299fc --- /dev/null +++ b/.config/mise/tasks/semconv/check @@ -0,0 +1,12 @@ +#!/bin/sh +#MISE description="Validate the semantic convention registry and generated contract" + +set -eu + +root=$(CDPATH='' cd -- "$(dirname -- "$0")/../../../.." && pwd) +expected="$root/trogon-eventstore/src/observability" +generated=$(mktemp -d) +trap 'rm -rf "$generated"' EXIT HUP INT TERM + +"$root/.config/mise/tasks/semconv/generate" "$generated" +diff -u "$expected/generated.rs" "$generated/generated.rs" diff --git a/.config/mise/tasks/semconv/generate b/.config/mise/tasks/semconv/generate new file mode 100755 index 0000000..45bcdad --- /dev/null +++ b/.config/mise/tasks/semconv/generate @@ -0,0 +1,28 @@ +#!/bin/sh +#MISE description="Generate the OpenTelemetry semantic convention contract" + +set -eu + +root=$(CDPATH='' cd -- "$(dirname -- "$0")/../../../.." && pwd) +output=${1:-"$root/trogon-eventstore/src/observability"} +registry_version=$(sed -n '1p' "$root/otel/semconv/registry-version") +registry="$root/otel/semconv/registry" +official_registry="https://github.com/open-telemetry/semantic-conventions@${registry_version}[model]" +staging=$(mktemp -d) +trap 'rm -rf "$staging"' EXIT HUP INT TERM + +grep -Fqx " registry_path: $official_registry" "$registry/manifest.yaml" + +weaver registry check \ + --future \ + --registry "$registry" + +weaver registry generate rust "$staging" \ + --future \ + --registry "$registry" \ + --templates "$root/otel/semconv/templates" + +test -f "$staging/observability.rs" +rustfmt --edition 2024 "$staging/observability.rs" +mkdir -p "$output" +mv "$staging/observability.rs" "$output/generated.rs" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 702a27e..3fddd1d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,6 +58,14 @@ jobs: rustup update stable rustup default stable rustup component add clippy rustfmt + - name: Install repository tools + uses: jdx/mise-action@9e7f7633ff6f6d6048a9418a68d48f288f50eb14 # v4.2.3 + with: + version: 2026.8.2 + install_args: github:open-telemetry/weaver + cache: false + - name: Verify semantic conventions + run: mise run --skip-tools semconv:check - name: Check formatting run: cargo fmt --all -- --check - name: Run Clippy 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/mise.toml b/mise.toml new file mode 100644 index 0000000..8d9f559 --- /dev/null +++ b/mise.toml @@ -0,0 +1,2 @@ +[tools] +"github:open-telemetry/weaver" = "0.24.2" diff --git a/otel/semconv/registry-version b/otel/semconv/registry-version new file mode 100644 index 0000000..518f2ea --- /dev/null +++ b/otel/semconv/registry-version @@ -0,0 +1 @@ +v1.43.0 diff --git a/otel/semconv/registry/manifest.yaml b/otel/semconv/registry/manifest.yaml new file mode 100644 index 0000000..ab2efcf --- /dev/null +++ b/otel/semconv/registry/manifest.yaml @@ -0,0 +1,6 @@ +name: trogon_eventstore_client +description: Semantic conventions for TrogonEventStore client telemetry. +schema_url: https://trogondb.com/schemas/0.1.0 +dependencies: + - schema_url: https://opentelemetry.io/schemas/1.43.0 + registry_path: https://github.com/open-telemetry/semantic-conventions@v1.43.0[model] diff --git a/otel/semconv/registry/trogon/eventstore/client-spans.yaml b/otel/semconv/registry/trogon/eventstore/client-spans.yaml new file mode 100644 index 0000000..9354d5a --- /dev/null +++ b/otel/semconv/registry/trogon/eventstore/client-spans.yaml @@ -0,0 +1,302 @@ +groups: + - id: registry.trogon.eventstore.client.attributes + type: attribute_group + brief: TrogonEventStore client attributes. + attributes: + - id: trogon.eventstore.batch.correlation_id + type: string + stability: development + brief: The identifier used to correlate a batch append request with its response. + examples: ["00000000-0000-0000-0000-000000000000"] + + - id: span.trogon.eventstore.client + type: span + extends: span.db.client + span_kind: client + stability: development + brief: A logical TrogonEventStore client operation. + attributes: + - ref: db.system.name + requirement_level: required + note: The value MUST be `trogoneventstore` and SHOULD be set when the span is created. + - ref: db.operation.name + requirement_level: required + - ref: db.collection.name + requirement_level: + conditionally_required: If the operation targets a single stream. + - ref: error.type + requirement_level: + conditionally_required: If and only if the operation failed. + + - id: span.trogon.eventstore.client.append_to_stream + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Appends events to a stream. + annotations: + code_generation: + operation_name: append_to_stream + + - id: span.trogon.eventstore.client.set_stream_metadata + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Sets metadata for a stream. + annotations: + code_generation: + operation_name: set_stream_metadata + + - id: span.trogon.eventstore.client.batch_append + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Starts a batch append session. + annotations: + code_generation: + operation_name: batch_append + + - id: span.trogon.eventstore.client.batch_append_to_stream + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Appends a batch of events to a stream. + attributes: + - ref: trogon.eventstore.batch.correlation_id + requirement_level: required + annotations: + code_generation: + operation_name: batch_append_to_stream + + - id: span.trogon.eventstore.client.read_stream + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Reads events from a stream. + annotations: + code_generation: + operation_name: read_stream + + - id: span.trogon.eventstore.client.read_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Reads events from the all stream. + annotations: + code_generation: + operation_name: read_all + + - id: span.trogon.eventstore.client.get_stream_metadata + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Reads metadata for a stream. + annotations: + code_generation: + operation_name: get_stream_metadata + + - id: span.trogon.eventstore.client.delete_stream + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Soft deletes a stream. + annotations: + code_generation: + operation_name: delete_stream + + - id: span.trogon.eventstore.client.tombstone_stream + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Permanently deletes a stream. + annotations: + code_generation: + operation_name: tombstone_stream + + - id: span.trogon.eventstore.client.subscribe_to_stream + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Starts a volatile stream subscription. + annotations: + code_generation: + operation_name: subscribe_to_stream + + - id: span.trogon.eventstore.client.subscribe_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Starts a volatile subscription to the all stream. + annotations: + code_generation: + operation_name: subscribe_to_all + + - id: span.trogon.eventstore.client.create_persistent_subscription + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Creates a persistent subscription for a stream. + annotations: + code_generation: + operation_name: create_persistent_subscription + + - id: span.trogon.eventstore.client.create_persistent_subscription_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Creates a persistent subscription for the all stream. + annotations: + code_generation: + operation_name: create_persistent_subscription_to_all + + - id: span.trogon.eventstore.client.update_persistent_subscription + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Updates a persistent subscription for a stream. + annotations: + code_generation: + operation_name: update_persistent_subscription + + - id: span.trogon.eventstore.client.update_persistent_subscription_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Updates a persistent subscription for the all stream. + annotations: + code_generation: + operation_name: update_persistent_subscription_to_all + + - id: span.trogon.eventstore.client.delete_persistent_subscription + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Deletes a persistent subscription for a stream. + annotations: + code_generation: + operation_name: delete_persistent_subscription + + - id: span.trogon.eventstore.client.delete_persistent_subscription_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Deletes a persistent subscription for the all stream. + annotations: + code_generation: + operation_name: delete_persistent_subscription_to_all + + - id: span.trogon.eventstore.client.subscribe_to_persistent_subscription + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Connects to a persistent subscription for a stream. + annotations: + code_generation: + operation_name: subscribe_to_persistent_subscription + + - id: span.trogon.eventstore.client.subscribe_to_persistent_subscription_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Connects to a persistent subscription for the all stream. + annotations: + code_generation: + operation_name: subscribe_to_persistent_subscription_to_all + + - id: span.trogon.eventstore.client.replay_parked_messages + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Replays parked messages for a stream. + annotations: + code_generation: + operation_name: replay_parked_messages + + - id: span.trogon.eventstore.client.replay_parked_messages_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Replays parked messages for the all stream. + annotations: + code_generation: + operation_name: replay_parked_messages_to_all + + - id: span.trogon.eventstore.client.list_all_persistent_subscriptions + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Lists all persistent subscriptions. + annotations: + code_generation: + operation_name: list_all_persistent_subscriptions + + - id: span.trogon.eventstore.client.list_persistent_subscriptions_for_stream + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Lists persistent subscriptions for a stream. + annotations: + code_generation: + operation_name: list_persistent_subscriptions_for_stream + + - id: span.trogon.eventstore.client.list_persistent_subscriptions_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Lists persistent subscriptions for the all stream. + annotations: + code_generation: + operation_name: list_persistent_subscriptions_to_all + + - id: span.trogon.eventstore.client.get_persistent_subscription_info + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Gets persistent subscription information for a stream. + annotations: + code_generation: + operation_name: get_persistent_subscription_info + + - id: span.trogon.eventstore.client.get_persistent_subscription_info_to_all + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Gets persistent subscription information for the all stream. + annotations: + code_generation: + operation_name: get_persistent_subscription_info_to_all + + - id: span.trogon.eventstore.client.restart_persistent_subscription_subsystem + type: span + extends: span.trogon.eventstore.client + span_kind: client + stability: development + brief: Restarts the persistent subscription subsystem. + annotations: + code_generation: + operation_name: restart_persistent_subscription_subsystem diff --git a/otel/semconv/templates/registry/rust/observability.rs.j2 b/otel/semconv/templates/registry/rust/observability.rs.j2 new file mode 100644 index 0000000..bb70c25 --- /dev/null +++ b/otel/semconv/templates/registry/rust/observability.rs.j2 @@ -0,0 +1,19 @@ +// + +use super::ClientOperation; +use opentelemetry::trace::SpanKind; + +{% for attribute in ctx.attributes | sort(attribute="name") %} +pub(crate) const {{ attribute.name | screaming_snake_case }}: &str = "{{ attribute.name }}"; +{% endfor %} + +pub(crate) const CLIENT_SPAN_KIND: SpanKind = SpanKind::{{ ctx.base.span_kind | pascal_case }}; + +pub(crate) mod operation { + use super::ClientOperation; + +{% for span in ctx.operations | sort(attribute="id") %} + pub(crate) const {{ span.id | replace("span.trogon.eventstore.client.", "") | screaming_snake_case }}: ClientOperation = + ClientOperation::new("{{ span.annotations.code_generation.operation_name }}"); +{% endfor %} +} diff --git a/otel/semconv/templates/registry/rust/weaver.yaml b/otel/semconv/templates/registry/rust/weaver.yaml new file mode 100644 index 0000000..48e9288 --- /dev/null +++ b/otel/semconv/templates/registry/rust/weaver.yaml @@ -0,0 +1,20 @@ +whitespace_control: + trim_blocks: true + lstrip_blocks: true + +templates: + - template: observability.rs.j2 + filter: > + { + base: [semconv_grouped_spans[].spans[] | select(.id == "span.trogon.eventstore.client")][0], + operations: [semconv_grouped_spans[].spans[] | select(.id | startswith("span.trogon.eventstore.client."))], + attributes: [semconv_grouped_spans[].spans[].attributes[] | select( + .name == "db.collection.name" or + .name == "db.operation.name" or + .name == "db.system.name" or + .name == "error.type" or + .name == "trogon.eventstore.batch.correlation_id" + )] | unique_by(.name) + } + application_mode: single + file_name: observability.rs 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..16c1c32 100644 --- a/trogon-eventstore/src/batch.rs +++ b/trogon-eventstore/src/batch.rs @@ -1,4 +1,8 @@ +use crate::observability::{TROGON_EVENTSTORE_BATCH_CORRELATION_ID, client_operation, operation}; use crate::{EventData, Position, StreamState}; +use opentelemetry::Context; +use opentelemetry::KeyValue; +use opentelemetry::trace::TraceContextExt; use tokio::sync::{ mpsc::{UnboundedReceiver, UnboundedSender}, oneshot, @@ -19,6 +23,17 @@ pub(crate) struct Req { pub(crate) expected_revision: StreamState, } +impl Req { + fn new(stream_name: String, events: Vec, expected_revision: StreamState) -> Self { + Self { + id: uuid::Uuid::new_v4(), + stream_name, + events, + expected_revision, + } + } +} + #[derive(Debug)] pub(crate) struct Out { pub(crate) correlation_id: uuid::Uuid, @@ -140,29 +155,174 @@ 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 stream_name = stream_name.as_ref().to_string(); + client_operation( + operation::BATCH_APPEND_TO_STREAM.on_collection(stream_name.clone()), + async { + let (sender, receiver) = oneshot::channel(); + let req = Req::new(stream_name, events, stream_state); + let context = Context::current(); + let span = context.span(); + if span.is_recording() { + span.set_attribute(KeyValue::new( + TROGON_EVENTSTORE_BATCH_CORRELATION_ID, + req.id.to_string(), + )); + } + + 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); + + let status = tonic::Status::cancelled("Batch-append stream has been closed"); + + Err(crate::Error::ServerError(status.to_string())) + }) + }, + ) + .await + } +} + +#[cfg(test)] +mod tests { + use super::{BatchAppendClient, BatchMsg, BatchWriteResult, In}; + use crate::StreamState; + use crate::observability::{ + DB_COLLECTION_NAME, DB_OPERATION_NAME, TROGON_EVENTSTORE_BATCH_CORRELATION_ID, + }; + use opentelemetry::global; + use opentelemetry::trace::{Status, noop::NoopTracerProvider}; + use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; + + #[tokio::test] + async fn batch_append_to_stream_records_its_protocol_correlation_id() { + let _guard = crate::observability::TEST_GLOBALS.lock().await; + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + global::set_tracer_provider(provider.clone()); + let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel(); + let client = BatchAppendClient { sender }; + + let operation = tokio::spawn(async move { + client + .append_to_stream("stream", StreamState::Any, Vec::new()) + .await + }); + let BatchMsg::In(In { req, sender }) = receiver.recv().await.expect("batch request") else { + panic!("expected an inbound batch request"); }; + let correlation_id = req.id.to_string(); + sender + .send(Ok(BatchWriteResult::new( + "stream".to_string(), + None, + None, + Some(StreamState::Any), + ))) + .expect("batch response receiver"); + operation.await.expect("batch task").expect("batch result"); + provider.force_flush().unwrap(); - let req = In { sender, req }; + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name == "batch_append_to_stream stream") + .expect("batch append client span"); + assert!(span.attributes.iter().any(|attribute| { + attribute.key.as_str() == TROGON_EVENTSTORE_BATCH_CORRELATION_ID + && attribute.value.to_string() == correlation_id + })); - if let Err(e) = self.sender.send(BatchMsg::In(req)) { - error!("[sending-end] Batch-append stream is closed: {}", e); + global::set_tracer_provider(NoopTracerProvider::new()); + } - let status = tonic::Status::cancelled("Batch-append stream has been closed"); - return Err(crate::Error::ServerError(status.to_string())); - } + #[tokio::test] + async fn concurrent_appends_route_reversed_responses_by_correlation_id() { + let _guard = crate::observability::TEST_GLOBALS.lock().await; + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + let response_sender = sender.clone(); + let (forward, mut forwarded) = tokio::sync::mpsc::unbounded_channel(); + let client = BatchAppendClient::new(sender, receiver, forward); + + let first = client.append_to_stream("first", StreamState::Any, Vec::new()); + let second = client.append_to_stream("second", StreamState::Any, Vec::new()); + let responses = async move { + let first = forwarded.recv().await.expect("first request"); + let second = forwarded.recv().await.expect("second request"); + assert_ne!(first.id, second.id); + + response_sender + .send(BatchMsg::Out(super::Out { + correlation_id: second.id, + result: Ok(BatchWriteResult::new( + second.stream_name, + None, + None, + Some(StreamState::Any), + )), + })) + .expect("second response"); + response_sender + .send(BatchMsg::Out(super::Out { + correlation_id: first.id, + result: Ok(BatchWriteResult::new( + first.stream_name, + None, + None, + Some(StreamState::Any), + )), + })) + .expect("first response"); + }; + + let (first, second, ()) = tokio::join!(first, second, responses); + assert_eq!(first.expect("first result").stream_name(), "first"); + assert_eq!(second.expect("second result").stream_name(), "second"); + } + + #[tokio::test] + async fn batch_append_to_stream_emits_a_logical_client_span() { + let _guard = crate::observability::TEST_GLOBALS.lock().await; + 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 }; - receiver.await.unwrap_or_else(|e| { - error!("[receiving-end] Batch-append stream is closed: {}", e); + let result = client + .append_to_stream("stream", StreamState::Any, Vec::new()) + .await; + assert!(result.is_err()); + provider.force_flush().unwrap(); - let status = tonic::Status::cancelled("Batch-append stream has been closed"); + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name == "batch_append_to_stream stream") + .expect("batch append client span"); + assert!(matches!(span.status, Status::Error { .. })); + assert!(span.attributes.iter().any(|attribute| { + attribute.key.as_str() == DB_OPERATION_NAME + && attribute.value.to_string() == "batch_append_to_stream" + })); + assert!(span.attributes.iter().any(|attribute| { + attribute.key.as_str() == DB_COLLECTION_NAME && attribute.value.to_string() == "stream" + })); - Err(crate::Error::ServerError(status.to_string())) - }) + global::set_tracer_provider(NoopTracerProvider::new()); } } diff --git a/trogon-eventstore/src/client.rs b/trogon-eventstore/src/client.rs index d45c02e..0765224 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,12 @@ impl Client { where Events: ToEvents, { - commands::append_to_stream(&self.client, stream_name, options, events.into_events()).await + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::APPEND_TO_STREAM.on_stream(&stream_name), + commands::append_to_stream(&self.client, stream_name, options, events.into_events()), + ) + .await } // Sets a stream metadata. @@ -88,11 +94,18 @@ impl Client { options: &AppendToStreamOptions, metadata: &StreamMetadata, ) -> crate::Result { - let event = EventData::json("$metadata", metadata) - .map_err(|e| crate::Error::InternalParsingError(e.to_string()))?; + let stream_name = name.into_metadata_stream_name(); + client_operation( + operation::SET_STREAM_METADATA.on_stream(&stream_name), + async { + 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) - .await + commands::append_to_stream(&self.client, 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,15 @@ impl Client { stream_name: impl StreamName, options: &ReadStreamOptions, ) -> crate::Result { - commands::read_stream( - self.client.clone(), - options, - stream_name, - options.max_count as u64, + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::READ_STREAM.on_stream(&stream_name), + commands::read_stream( + self.client.clone(), + options, + stream_name, + options.max_count as u64, + ), ) .await } @@ -122,7 +143,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.on_collection("$all"), + commands::read_all(self.client.clone(), options, options.max_count as u64), + ) + .await } /// Reads a stream metadata. @@ -131,32 +156,43 @@ impl Client { name: impl MetadataStreamName, options: &ReadStreamOptions, ) -> crate::Result { - let mut stream = self - .read_stream(name.into_metadata_stream_name(), options) - .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))) - } - Err(e) => match e { - crate::Error::ResourceNotFound => Ok(StreamMetadataResult::NotFound), - crate::Error::ResourceDeleted => Ok(StreamMetadataResult::Deleted), - other => Err(other), + let stream_name = name.into_metadata_stream_name(); + client_operation( + operation::GET_STREAM_METADATA.on_stream(&stream_name), + async { + let mut stream = commands::read_stream( + self.client.clone(), + options, + 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))) + } + 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 +206,12 @@ impl Client { stream_name: impl StreamName, options: &DeleteStreamOptions, ) -> crate::Result> { - commands::delete_stream(&self.client, stream_name, options).await + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::DELETE_STREAM.on_stream(&stream_name), + commands::delete_stream(&self.client, stream_name, options), + ) + .await } /// Hard deletes a given stream. @@ -183,7 +224,12 @@ impl Client { stream_name: impl StreamName, options: &TombstoneStreamOptions, ) -> crate::Result> { - commands::tombstone_stream(&self.client, stream_name, options).await + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::TOMBSTONE_STREAM.on_stream(&stream_name), + commands::tombstone_stream(&self.client, stream_name, options), + ) + .await } /// Subscribes to a given stream. This kind of subscription specifies a @@ -205,14 +251,22 @@ impl Client { stream_name: impl StreamName, options: &SubscribeToStreamOptions, ) -> Subscription { - commands::subscribe_to_stream(self.client.clone(), stream_name, options) + let stream_name = stream_name.into_stream_name(); + infallible_client_operation( + operation::SUBSCRIBE_TO_STREAM.on_stream(&stream_name), + 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.on_collection("$all"), async { + commands::subscribe_to_all(self.client.clone(), options) + }) + .await } /// Creates a persistent subscription group on a stream. @@ -227,11 +281,15 @@ impl Client { group_name: impl AsRef, options: &PersistentSubscriptionOptions, ) -> crate::Result<()> { - commands::create_persistent_subscription( - &self.client, - stream_name, - group_name.as_ref(), - options, + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::CREATE_PERSISTENT_SUBSCRIPTION.on_stream(&stream_name), + commands::create_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + ), ) .await } @@ -242,8 +300,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.on_collection("$all"), + commands::create_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + ), + ) + .await } /// Updates a persistent subscription group on a stream. @@ -253,11 +319,15 @@ impl Client { group_name: impl AsRef, options: &PersistentSubscriptionOptions, ) -> crate::Result<()> { - commands::update_persistent_subscription( - &self.client, - stream_name, - group_name.as_ref(), - options, + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::UPDATE_PERSISTENT_SUBSCRIPTION.on_stream(&stream_name), + commands::update_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + ), ) .await } @@ -268,8 +338,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.on_collection("$all"), + commands::update_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + ), + ) + .await } /// Deletes a persistent subscription group on a stream. @@ -279,12 +357,16 @@ impl Client { group_name: impl AsRef, options: &DeletePersistentSubscriptionOptions, ) -> crate::Result<()> { - commands::delete_persistent_subscription( - &self.client, - stream_name, - group_name.as_ref(), - options, - false, + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::DELETE_PERSISTENT_SUBSCRIPTION.on_stream(&stream_name), + commands::delete_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + false, + ), ) .await } @@ -295,12 +377,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.on_collection("$all"), + commands::delete_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + true, + ), ) .await } @@ -312,12 +397,16 @@ 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, + let stream_name = stream_name.into_stream_name(); + client_operation( + operation::SUBSCRIBE_TO_PERSISTENT_SUBSCRIPTION.on_stream(&stream_name), + commands::subscribe_to_persistent_subscription( + &self.client, + stream_name, + group_name.as_ref(), + options, + false, + ), ) .await } @@ -328,12 +417,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.on_collection("$all"), + commands::subscribe_to_persistent_subscription( + &self.client, + "", + group_name.as_ref(), + options, + true, + ), ) .await } @@ -345,12 +437,16 @@ 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, + let stream_name = stream_name.as_ref().to_string(); + client_operation( + operation::REPLAY_PARKED_MESSAGES.on_collection(stream_name.clone()), + commands::replay_parked_messages( + &self.client, + &self.http_client, + commands::RegularStream(stream_name), + group_name, + options, + ), ) .await } @@ -361,12 +457,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.on_collection("$all"), + commands::replay_parked_messages( + &self.client, + &self.http_client, + commands::AllStream, + group_name, + options, + ), ) .await } @@ -376,7 +475,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 +488,15 @@ 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, + let stream_name = stream_name.as_ref().to_string(); + client_operation( + operation::LIST_PERSISTENT_SUBSCRIPTIONS_FOR_STREAM.on_collection(stream_name.clone()), + commands::list_persistent_subscriptions_for_stream( + &self.client, + &self.http_client, + commands::RegularStream(stream_name), + options, + ), ) .await } @@ -399,11 +506,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.on_collection("$all"), + commands::list_persistent_subscriptions_for_stream( + &self.client, + &self.http_client, + commands::AllStream, + options, + ), ) .await } @@ -415,12 +525,16 @@ 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, + let stream_name = stream_name.as_ref().to_string(); + client_operation( + operation::GET_PERSISTENT_SUBSCRIPTION_INFO.on_collection(stream_name.clone()), + commands::get_persistent_subscription_info( + &self.client, + &self.http_client, + commands::RegularStream(stream_name), + group_name, + options, + ), ) .await } @@ -431,12 +545,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.on_collection("$all"), + commands::get_persistent_subscription_info( + &self.client, + &self.http_client, + commands::AllStream, + group_name, + options, + ), ) .await } @@ -446,10 +563,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..72da08a --- /dev/null +++ b/trogon-eventstore/src/observability.rs @@ -0,0 +1,259 @@ +mod generated; + +use generated::CLIENT_SPAN_KIND; +pub(crate) use generated::{ + DB_COLLECTION_NAME, DB_OPERATION_NAME, DB_SYSTEM_NAME, ERROR_TYPE, + TROGON_EVENTSTORE_BATCH_CORRELATION_ID, +}; +use opentelemetry::trace::{FutureExt, Status, TraceContextExt, Tracer}; +use opentelemetry::{Context, InstrumentationScope, KeyValue, global}; +use std::borrow::Cow; +use std::future::Future; + +pub(crate) use generated::operation; + +#[cfg(test)] +pub(crate) static TEST_GLOBALS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_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 { + operation_name: &'static str, +} + +impl ClientOperation { + pub(super) const fn new(operation_name: &'static str) -> Self { + Self { operation_name } + } + + pub(crate) fn on_collection(self, collection_name: impl Into) -> ClientSpan { + ClientSpan { + operation: self, + collection_name: Some(collection_name.into()), + } + } + + pub(crate) fn on_stream(self, stream_name: &bytes::Bytes) -> ClientSpan { + self.on_collection(String::from_utf8_lossy(stream_name).into_owned()) + } + + pub(crate) const fn span_name(self) -> &'static str { + self.operation_name + } +} + +pub(crate) struct ClientSpan { + operation: ClientOperation, + collection_name: Option, +} + +impl ClientSpan { + fn span_name(&self) -> Cow<'static, str> { + match &self.collection_name { + Some(collection_name) => Cow::Owned(format!( + "{} {}", + self.operation.span_name(), + collection_name + )), + None => Cow::Borrowed(self.operation.span_name()), + } + } +} + +impl From for ClientSpan { + fn from(operation: ClientOperation) -> Self { + Self { + operation, + collection_name: None, + } + } +} + +fn instrumentation_scope() -> InstrumentationScope { + InstrumentationScope::builder(INSTRUMENTATION_SCOPE.name) + .with_version(INSTRUMENTATION_SCOPE.version) + .build() +} + +fn start_client_operation(operation: impl Into) -> Context { + let operation = operation.into(); + let tracer = global::tracer_with_scope(instrumentation_scope()); + let mut attributes = vec![ + KeyValue::new(DB_SYSTEM_NAME, "trogoneventstore"), + KeyValue::new(DB_OPERATION_NAME, operation.operation.span_name()), + ]; + if let Some(collection_name) = &operation.collection_name { + attributes.push(KeyValue::new(DB_COLLECTION_NAME, collection_name.clone())); + } + + let span = tracer + .span_builder(operation.span_name()) + .with_kind(CLIENT_SPAN_KIND) + .with_attributes(attributes) + .start(&tracer); + + Context::current_with_span(span) +} + +fn error_type(error: &crate::Error) -> &'static str { + match error { + crate::Error::ServerError(_) => "server_error", + crate::Error::NotLeaderException(_) => "not_leader", + crate::Error::ConnectionClosed => "connection_closed", + crate::Error::Grpc { .. } => "grpc", + crate::Error::GrpcConnectionError(_) => "grpc_connection", + crate::Error::InternalParsingError(_) => "internal_parsing", + crate::Error::AccessDenied => "access_denied", + crate::Error::ResourceAlreadyExists => "resource_already_exists", + crate::Error::ResourceNotFound => "resource_not_found", + crate::Error::ResourceDeleted => "resource_deleted", + crate::Error::UnsupportedFeature => "unsupported_feature", + crate::Error::InternalClientError => "internal_client", + crate::Error::DeadlineExceeded => "deadline_exceeded", + crate::Error::InitializationError(_) => "initialization", + crate::Error::IllegalStateError(_) => "illegal_state", + crate::Error::WrongExpectedVersion { .. } => "wrong_expected_version", + } +} + +pub(crate) async fn client_operation( + operation: impl Into, + 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 { + let error_type = error_type(error); + context + .span() + .set_attribute(KeyValue::new(ERROR_TYPE, error_type)); + context.span().set_status(Status::error(error_type)); + } + context.span().end(); + + result +} + +pub(crate) async fn infallible_client_operation( + operation: impl Into, + 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::{ + DB_COLLECTION_NAME, DB_OPERATION_NAME, DB_SYSTEM_NAME, ERROR_TYPE, 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_semantic_database_client_span() { + let _guard = super::TEST_GLOBALS.lock().await; + 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.on_collection("orders"), 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 orders") + .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_eq!(attribute(span, DB_SYSTEM_NAME), Some("trogoneventstore")); + assert_eq!(attribute(span, DB_OPERATION_NAME), Some("append_to_stream")); + assert_eq!(attribute(span, DB_COLLECTION_NAME), Some("orders")); + assert_eq!(attribute(span, ERROR_TYPE), None); + + global::set_tracer_provider(NoopTracerProvider::new()); + } + + #[tokio::test] + async fn client_operation_records_a_bounded_error_status() { + let _guard = super::TEST_GLOBALS.lock().await; + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + global::set_tracer_provider(provider.clone()); + + const ERROR_DETAIL: &str = "variable backend detail"; + let result = client_operation::<(), _>(APPEND_TO_STREAM, async { + Err(crate::Error::ServerError(ERROR_DETAIL.to_owned())) + }) + .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_eq!(span.status, Status::error("server_error")); + if let Status::Error { description } = &span.status { + assert!(!description.contains(ERROR_DETAIL)); + } + assert_eq!(attribute(span, ERROR_TYPE), Some("server_error")); + + global::set_tracer_provider(NoopTracerProvider::new()); + } + + fn attribute<'a>(span: &'a opentelemetry_sdk::trace::SpanData, key: &str) -> Option<&'a str> { + span.attributes + .iter() + .find(|attribute| attribute.key.as_str() == key) + .and_then(|attribute| match &attribute.value { + opentelemetry::Value::String(value) => Some(value.as_ref()), + _ => None, + }) + } +} diff --git a/trogon-eventstore/src/observability/generated.rs b/trogon-eventstore/src/observability/generated.rs new file mode 100644 index 0000000..a19b71e --- /dev/null +++ b/trogon-eventstore/src/observability/generated.rs @@ -0,0 +1,65 @@ +// + +use super::ClientOperation; +use opentelemetry::trace::SpanKind; + +pub(crate) const DB_COLLECTION_NAME: &str = "db.collection.name"; +pub(crate) const DB_OPERATION_NAME: &str = "db.operation.name"; +pub(crate) const DB_SYSTEM_NAME: &str = "db.system.name"; +pub(crate) const ERROR_TYPE: &str = "error.type"; +pub(crate) const TROGON_EVENTSTORE_BATCH_CORRELATION_ID: &str = + "trogon.eventstore.batch.correlation_id"; + +pub(crate) const CLIENT_SPAN_KIND: SpanKind = SpanKind::Client; + +pub(crate) mod operation { + use super::ClientOperation; + + pub(crate) const APPEND_TO_STREAM: ClientOperation = ClientOperation::new("append_to_stream"); + pub(crate) const BATCH_APPEND: ClientOperation = ClientOperation::new("batch_append"); + pub(crate) const BATCH_APPEND_TO_STREAM: ClientOperation = + ClientOperation::new("batch_append_to_stream"); + pub(crate) const CREATE_PERSISTENT_SUBSCRIPTION: ClientOperation = + ClientOperation::new("create_persistent_subscription"); + pub(crate) const CREATE_PERSISTENT_SUBSCRIPTION_TO_ALL: ClientOperation = + ClientOperation::new("create_persistent_subscription_to_all"); + pub(crate) const DELETE_PERSISTENT_SUBSCRIPTION: ClientOperation = + ClientOperation::new("delete_persistent_subscription"); + pub(crate) const DELETE_PERSISTENT_SUBSCRIPTION_TO_ALL: ClientOperation = + ClientOperation::new("delete_persistent_subscription_to_all"); + pub(crate) const DELETE_STREAM: ClientOperation = ClientOperation::new("delete_stream"); + pub(crate) const GET_PERSISTENT_SUBSCRIPTION_INFO: ClientOperation = + ClientOperation::new("get_persistent_subscription_info"); + pub(crate) const GET_PERSISTENT_SUBSCRIPTION_INFO_TO_ALL: ClientOperation = + ClientOperation::new("get_persistent_subscription_info_to_all"); + pub(crate) const GET_STREAM_METADATA: ClientOperation = + ClientOperation::new("get_stream_metadata"); + pub(crate) const LIST_ALL_PERSISTENT_SUBSCRIPTIONS: ClientOperation = + ClientOperation::new("list_all_persistent_subscriptions"); + pub(crate) const LIST_PERSISTENT_SUBSCRIPTIONS_FOR_STREAM: ClientOperation = + ClientOperation::new("list_persistent_subscriptions_for_stream"); + pub(crate) const LIST_PERSISTENT_SUBSCRIPTIONS_TO_ALL: ClientOperation = + ClientOperation::new("list_persistent_subscriptions_to_all"); + pub(crate) const READ_ALL: ClientOperation = ClientOperation::new("read_all"); + pub(crate) const READ_STREAM: ClientOperation = ClientOperation::new("read_stream"); + pub(crate) const REPLAY_PARKED_MESSAGES: ClientOperation = + ClientOperation::new("replay_parked_messages"); + pub(crate) const REPLAY_PARKED_MESSAGES_TO_ALL: ClientOperation = + ClientOperation::new("replay_parked_messages_to_all"); + pub(crate) const RESTART_PERSISTENT_SUBSCRIPTION_SUBSYSTEM: ClientOperation = + ClientOperation::new("restart_persistent_subscription_subsystem"); + pub(crate) const SET_STREAM_METADATA: ClientOperation = + ClientOperation::new("set_stream_metadata"); + pub(crate) const SUBSCRIBE_TO_ALL: ClientOperation = ClientOperation::new("subscribe_to_all"); + pub(crate) const SUBSCRIBE_TO_PERSISTENT_SUBSCRIPTION: ClientOperation = + ClientOperation::new("subscribe_to_persistent_subscription"); + pub(crate) const SUBSCRIBE_TO_PERSISTENT_SUBSCRIPTION_TO_ALL: ClientOperation = + ClientOperation::new("subscribe_to_persistent_subscription_to_all"); + pub(crate) const SUBSCRIBE_TO_STREAM: ClientOperation = + ClientOperation::new("subscribe_to_stream"); + pub(crate) const TOMBSTONE_STREAM: ClientOperation = ClientOperation::new("tombstone_stream"); + pub(crate) const UPDATE_PERSISTENT_SUBSCRIPTION: ClientOperation = + ClientOperation::new("update_persistent_subscription"); + pub(crate) const UPDATE_PERSISTENT_SUBSCRIPTION_TO_ALL: ClientOperation = + ClientOperation::new("update_persistent_subscription_to_all"); +} diff --git a/trogon-eventstore/src/request.rs b/trogon-eventstore/src/request.rs index 4bc51af..da04ff9 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().await; + 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");