feat(observability): preserve cross-client trace continuity (#10)

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

Cargo.lock

@@ -3,6 +3,11 @@ groups:
     type: attribute_group
     brief: TrogonEventStore client attributes.
     attributes:
+      - id: trogon.eventstore.event.type
+        type: string
+        stability: development
+        brief: The type of event received by the client.
+        examples: [order-created]
       - id: trogon.eventstore.batch.correlation_id
         type: string
         stability: development
@@ -28,6 +33,28 @@ groups:
         requirement_level:
           conditionally_required: If and only if the operation failed.
 
+  - id: span.trogon.eventstore.client.receive
+    type: span
+    span_kind: client
+    stability: development
+    brief: Receives an event delivered by a subscription.
+    attributes:
+      - ref: messaging.system
+        requirement_level: required
+      - ref: messaging.operation.name
+        requirement_level: required
+      - ref: messaging.operation.type
+        requirement_level: required
+      - ref: messaging.destination.name
+        requirement_level: required
+      - ref: messaging.message.id
+        requirement_level: recommended
+      - ref: messaging.consumer.group.name
+        requirement_level:
+          conditionally_required: If the event was delivered by a persistent subscription.
+      - ref: trogon.eventstore.event.type
+        requirement_level: recommended
+
   - id: span.trogon.eventstore.client.append_to_stream
     type: span
     extends: span.trogon.eventstore.client

otel/semconv/registry/trogon/eventstore/client-spans.yaml

@@ -8,6 +8,7 @@ pub(crate) const {{ attribute.name | screaming_snake_case }}: &str = "{{ attribu
 {% endfor %}
 
 pub(crate) const CLIENT_SPAN_KIND: SpanKind = SpanKind::{{ ctx.base.span_kind | pascal_case }};
+pub(crate) const RECEIVE_SPAN_KIND: SpanKind = SpanKind::{{ ctx.receive.span_kind | pascal_case }};
 
 pub(crate) mod operation {
     use super::ClientOperation;

otel/semconv/templates/registry/rust/observability.rs.j2

@@ -7,12 +7,23 @@ templates:
     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."))],
+        receive: [semconv_grouped_spans[].spans[] | select(.id == "span.trogon.eventstore.client.receive")][0],
+        operations: [semconv_grouped_spans[].spans[] | select(
+          (.id | startswith("span.trogon.eventstore.client.")) and
+          .annotations.code_generation.operation_name != null
+        )],
         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 == "messaging.consumer.group.name" or
+          .name == "messaging.destination.name" or
+          .name == "messaging.message.id" or
+          .name == "messaging.operation.name" or
+          .name == "messaging.operation.type" or
+          .name == "messaging.system" or
+          .name == "trogon.eventstore.event.type" or
           .name == "trogon.eventstore.batch.correlation_id"
         )] | unique_by(.name)
       }

otel/semconv/templates/registry/rust/weaver.yaml

@@ -69,6 +69,7 @@ name = "integration"
 
 [dev-dependencies]
 names = "0.14"
+opentelemetry-otlp = { version = "0.32", default-features = false, features = ["grpc-tonic", "trace"] }
 opentelemetry_sdk = { version = "0.32", default-features = false, features = ["testing", "trace"] }
 serde = { version = "1", features = ["derive"] }
 testcontainers = "0.23"

trogon-eventstore/Cargo.toml

@@ -21,6 +21,7 @@ pub(crate) struct Req {
     pub(crate) stream_name: String,
     pub(crate) events: Vec<EventData>,
     pub(crate) expected_revision: StreamState,
+    pub(crate) context: Context,
 }
 
 impl Req {
@@ -30,6 +31,7 @@ impl Req {
             stream_name,
             events,
             expected_revision,
+            context: Context::current(),
         }
     }
 }
@@ -200,7 +202,7 @@ mod tests {
         DB_COLLECTION_NAME, DB_OPERATION_NAME, TROGON_EVENTSTORE_BATCH_CORRELATION_ID,
     };
     use opentelemetry::global;
-    use opentelemetry::trace::{Status, noop::NoopTracerProvider};
+    use opentelemetry::trace::{Status, TraceContextExt, noop::NoopTracerProvider};
     use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider};
 
     #[tokio::test]
@@ -223,6 +225,7 @@ mod tests {
             panic!("expected an inbound batch request");
         };
         let correlation_id = req.id.to_string();
+        let captured_span_id = req.context.span().span_context().span_id();
         sender
             .send(Ok(BatchWriteResult::new(
                 "stream".to_string(),
@@ -239,6 +242,7 @@ mod tests {
             .iter()
             .find(|span| span.name == "batch_append_to_stream stream")
             .expect("batch append client span");
+        assert_eq!(span.span_context.span_id(), captured_span_id);
         assert!(span.attributes.iter().any(|attribute| {
             attribute.key.as_str() == TROGON_EVENTSTORE_BATCH_CORRELATION_ID
                 && attribute.value.to_string() == correlation_id