Propagate opentelemetry spans across remote connections.

Cargo.toml

@@ -44,12 +44,17 @@ n0-future = { workspace = true }
 futures-util = { workspace = true, optional = true }
 # for the derive reexport/feature
 irpc-derive = { version = "0.10.0", path = "./irpc-derive", optional = true }
+# for remote span propagation when tracing-opentelemetry feature is enabled
+opentelemetry = { version = "0.31", optional = true }
+tracing-opentelemetry = { version = "0.32", optional = true }
 
 [target.'cfg(not(all(target_family = "wasm", target_os = "unknown")))'.dependencies]
 noq = { workspace = true, optional = true, features = ["runtime-tokio"] }
 
 [dev-dependencies]
 tracing-subscriber = { workspace = true, features = ["fmt"] }
+# pin lazy_static minimum for -Z minimal-versions compatibility (sharded-slab needs >= 1.4)
+lazy_static = { version = "1.4" }
 # just convenient for the enum definitions, in the manual example
 derive_more = { version = "2", features = ["from"] }
 # we need full for example main etc.
@@ -72,6 +77,8 @@ spans = ["dep:tracing"]
 stream = ["dep:futures-util"]
 derive = ["dep:irpc-derive"]
 varint-util = ["dep:postcard", "dep:smallvec", "tokio/io-util"]
+# enable OpenTelemetry span context propagation across remote connections
+tracing-opentelemetry = ["dep:opentelemetry", "dep:tracing-opentelemetry", "rpc"]
 default = ["rpc", "noq_endpoint_setup", "spans", "stream", "derive"]
 
 [[example]]
@@ -91,6 +98,7 @@ name = "storage"
 required-features = ["rpc", "noq_endpoint_setup"]
 
 [workspace]
+resolver = "2"
 members = ["irpc-derive", "irpc-iroh"]
 
 [package.metadata.docs.rs]
@@ -108,7 +116,7 @@ serde = { version = "1", default-features = false, features = ["derive"] }
 tracing = { version = "0.1.41", default-features = false }
 n0-future = { version = "0.3", default-features = false }
 n0-error = { version = "0.1" }
-tracing-subscriber = { version = "0.3.20" }
+tracing-subscriber = { version = "0.3.23" }
 iroh = { version = "0.97" }
 iroh-base = { version = "0.97" }
 noq = { version = "0.17.0", default-features = false }

irpc-derive/src/lib.rs

@@ -159,15 +159,21 @@ pub fn rpc_requests(attr: TokenStream, item: TokenStream) -> TokenStream {
         let message_from_impls =
             generate_message_enum_from_impls(message_enum_name, &variants_with_attr, enum_name);
 
+        let span_propagation = args.span_propagation;
         let service_impl = quote! {
             impl ::irpc::Service for #enum_name {
                 type Message = #message_enum_name;
+                const SPAN_PROPAGATION: bool = #span_propagation;
             }
         };
 
         let remote_service_impl = if !args.no_rpc {
-            let block =
-                generate_remote_service_impl(message_enum_name, enum_name, &variants_with_attr);
+            let block = generate_remote_service_impl(
+                message_enum_name,
+                enum_name,
+                &variants_with_attr,
+                args.span_propagation,
+            );
             quote! {
                 #cfg_feature_rpc
                 #block
@@ -287,17 +293,37 @@ fn generate_message_enum_from_impls(
 }
 
 /// Generate `RemoteService` impl for message enums.
+///
+/// When `span_propagation` is true, the generated code will create spans for each
+/// request and set their parent from the propagated remote context.
 fn generate_remote_service_impl(
     message_enum_name: &Ident,
     proto_enum_name: &Ident,
     variants_with_attr: &[(Ident, Type)],
+    span_propagation: bool,
 ) -> TokenStream2 {
     let variants = variants_with_attr
         .iter()
         .map(|(variant_name, _inner_type)| {
-            quote! {
-                #proto_enum_name::#variant_name(msg) => {
-                    #message_enum_name::from(::irpc::WithChannels::from((msg, tx, rx)))
+            let span_name = variant_name.to_string();
+
+            if span_propagation {
+                // When span_propagation is enabled, create spans and set parent from remote context
+                quote! {
+                    #proto_enum_name::#variant_name(msg) => {
+                        // Create a span for this specific RPC operation
+                        let span = ::tracing::info_span!(#span_name);
+                        // Set its parent to the propagated remote context if available
+                        ::irpc::span_propagation::set_span_parent_from_remote(&span);
+                        let _guard = span.enter();
+                        #message_enum_name::from(::irpc::WithChannels::from((msg, tx, rx)))
+                    }
+                }
+            } else {
+                quote! {
+                    #proto_enum_name::#variant_name(msg) => {
+                        #message_enum_name::from(::irpc::WithChannels::from((msg, tx, rx)))
+                    }
                 }
             }
         });
@@ -346,6 +372,8 @@ struct MacroArgs {
     rpc_feature: Option<String>,
     no_rpc: bool,
     no_spans: bool,
+    /// When true, includes span context in the wire format and enables span propagation.
+    span_propagation: bool,
 }
 
 impl Parse for MacroArgs {
@@ -381,6 +409,9 @@ impl Parse for MacroArgs {
                 "no_spans" => {
                     this.no_spans = true;
                 }
+                "span_propagation" => {
+                    this.span_propagation = true;
+                }
                 _ => {
                     return syn_err(arg.span(), format!("Unknown parameter: {arg}"));
                 }

irpc-iroh/Cargo.toml

@@ -35,3 +35,12 @@ futures-util.workspace = true
 hex = "0.4.3"
 rand = "0.9.2"
 anyhow = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+n0-tracing-test = "0.3.0"
+opentelemetry = "0.31.0"
+opentelemetry_sdk = { version = "0.31.0", features = ["testing"] }
+tracing-opentelemetry = "0.32.1"
+
+[features]
+default = []
+tracing-opentelemetry = ["irpc/tracing-opentelemetry"]