From b47c17f084dc3b1e36157cc749356a5943c24698 Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 8 May 2026 10:48:11 +0200 Subject: [PATCH] examples(irpc-iroh): expand span_propagation example to work with Jaeger --- Cargo.lock | 157 ++++++++++++++++++- irpc-iroh/Cargo.toml | 1 + irpc-iroh/examples/span_propagation.rs | 205 +++++++++++++++++++------ 3 files changed, 315 insertions(+), 48 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7bea929..9421044 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1646,7 +1646,7 @@ dependencies = [ "portable-atomic", "portmapper", "rand 0.10.1", - "reqwest", + "reqwest 0.13.3", "rustc-hash", "rustls", "rustls-pki-types", @@ -1757,7 +1757,7 @@ dependencies = [ "pin-project", "postcard", "rand 0.10.1", - "reqwest", + "reqwest 0.13.3", "rustls", "rustls-pki-types", "serde", @@ -1829,6 +1829,7 @@ dependencies = [ "n0-future", "n0-tracing-test", "opentelemetry", + "opentelemetry-otlp", "opentelemetry_sdk", "postcard", "rand 0.9.2", @@ -1845,6 +1846,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.17" @@ -2479,6 +2489,48 @@ dependencies = [ "tracing", ] +[[package]] +name = "opentelemetry-http" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d" +dependencies = [ + "async-trait", + "bytes", + "http", + "opentelemetry", + "reqwest 0.12.28", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.31.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest 0.12.28", + "thiserror", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + [[package]] name = "opentelemetry_sdk" version = "0.31.0" @@ -2770,6 +2822,29 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prost" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "quick-xml" version = "0.38.4" @@ -2886,6 +2961,40 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64", + "bytes", + "futures-channel", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "reqwest" version = "0.13.3" @@ -3191,6 +3300,18 @@ dependencies = [ "serde_core", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "sha1_smol" version = "1.0.1" @@ -3688,6 +3809,38 @@ version = "1.0.7+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f17aaa1c6e3dc22b1da4b6bba97d066e354c7945cac2f7852d4e4e7ca7a6b56d" +[[package]] +name = "tonic" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" +dependencies = [ + "async-trait", + "base64", + "bytes", + "http", + "http-body", + "http-body-util", + "percent-encoding", + "pin-project", + "sync_wrapper", + "tokio-stream", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-prost" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0" +dependencies = [ + "bytes", + "prost", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" diff --git a/irpc-iroh/Cargo.toml b/irpc-iroh/Cargo.toml index f31696f..06240fe 100644 --- a/irpc-iroh/Cargo.toml +++ b/irpc-iroh/Cargo.toml @@ -39,6 +39,7 @@ tokio = { workspace = true, features = ["full"] } n0-tracing-test = "0.3.0" opentelemetry = "0.31.0" opentelemetry_sdk = { version = "0.31.0", features = ["testing"] } +opentelemetry-otlp = { version = "0.31.0", default-features = false, features = ["http-proto", "reqwest-blocking-client", "trace"] } tracing-opentelemetry = "0.32.1" [features] diff --git a/irpc-iroh/examples/span_propagation.rs b/irpc-iroh/examples/span_propagation.rs index 88cc879..de8c0f1 100644 --- a/irpc-iroh/examples/span_propagation.rs +++ b/irpc-iroh/examples/span_propagation.rs @@ -1,32 +1,39 @@ -//! Demonstrates how to enable OpenTelemetry span context propagation across -//! irpc-iroh remote calls. +//! Distributed span propagation across irpc-iroh, visualized in Jaeger. //! -//! Run with: +//! Start Jaeger (UI on 16686, OTLP HTTP on 4318): //! //! ```sh -//! cargo run --example span_propagation -p irpc-iroh \ -//! --features tracing-opentelemetry +//! docker run --rm -p 16686:16686 -p 4318:4318 jaegertracing/jaeger:latest //! ``` //! -//! What this shows: +//! Server (prints its endpoint id): //! -//! 1. Setting up `tracing-opentelemetry` with a `TextMapPropagator` so that -//! trace context can be injected into / extracted from RPC payloads. -//! 2. Declaring a service with `#[rpc_requests(..., span_propagation)]`. Only -//! services that opt in pay the wire-format cost; everything else is -//! unchanged. -//! 3. Entering the per-request span that `WithChannels` carries. The macro -//! creates a span named after the protocol variant (here `"Get"`) and -//! attaches the propagated remote context as its parent. Entering it makes -//! any work the handler does — including child spans — part of the same -//! distributed trace. - -use std::sync::Arc; +//! ```sh +//! cargo run -p irpc-iroh --features tracing-opentelemetry --example span_propagation -- server +//! ``` +//! +//! Client: +//! +//! ```sh +//! cargo run -p irpc-iroh --features tracing-opentelemetry --example span_propagation -- client +//! ``` +//! +//! Open , pick the `example-client` service, and each +//! `req` trace will include the server's child `Get` span. +//! +//! After opening a trace, click "Trace Logs" in the view selector at the top right +//! to open a log view that assembles logs from both client and server. + +use std::{sync::Arc, time::Duration}; use anyhow::Result; -use iroh::{Endpoint, endpoint::presets, protocol::Router}; +use clap::{Parser, Subcommand}; +use iroh::{Endpoint, EndpointId, endpoint::presets, protocol::Router}; use irpc::{WithChannels, channel::oneshot, rpc::RemoteService, rpc_requests}; use irpc_iroh::IrohProtocol; +use opentelemetry::KeyValue; +use opentelemetry_otlp::WithExportConfig; +use opentelemetry_sdk::{Resource, trace::SdkTracerProvider}; use serde::{Deserialize, Serialize}; use tracing::{Instrument, info, info_span}; use tracing_subscriber::{EnvFilter, Layer, Registry, layer::SubscriberExt}; @@ -40,60 +47,165 @@ enum Proto { } const ALPN: &[u8] = b"irpc-iroh/span_propagation/1"; +const DEFAULT_OTLP_ENDPOINT: &str = "http://localhost:4318/v1/traces"; -#[tokio::main] -async fn main() -> Result<()> { - // (1) Install a W3C trace-context propagator. Without this, the carrier - // written by the client and read by the server has no fields to - // inject into / extract from. +#[derive(Parser, Debug)] +#[command(about = "Distributed span propagation demo over irpc-iroh")] +struct Cli { + /// OTLP HTTP/protobuf endpoint to export spans to. Defaults to Jaeger's standard port. + #[arg(long, global = true, default_value = DEFAULT_OTLP_ENDPOINT)] + otlp: String, + + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand, Debug)] +enum Command { + /// Run the server. Prints its endpoint id; pass that to the client. + Server, + /// Send some requests to a server identified by `endpoint_id`. + Client { + endpoint_id: EndpointId, + /// Number of requests to send. + #[arg(long, default_value_t = 3)] + count: u32, + }, +} + +fn init_tracing( + service_name: &'static str, + endpoint_id: EndpointId, + otlp_endpoint: &str, +) -> Result { + // (1) Install a W3C trace-context propagator. Without this, the + // `SpanContextCarrier` injected on the client and extracted on the + // server has no header keys to read or write, so trace ids never + // cross the wire. opentelemetry::global::set_text_map_propagator( opentelemetry_sdk::propagation::TraceContextPropagator::new(), ); - // A no-op global tracer is enough for this example. In a real service - // you'd wire this up to OTLP / Jaeger / etc. - let tracer = opentelemetry::global::tracer("irpc-iroh-example"); - let env = EnvFilter::try_from_default_env() + // (2) Build the OTLP HTTP/protobuf exporter pointed at the collector + // (Jaeger's OTLP ingestor by default). The `reqwest-blocking-client` + // transport lets the SDK's batch processor send from its own OS + // thread without a tokio runtime in scope. + let exporter = opentelemetry_otlp::SpanExporter::builder() + .with_http() + .with_endpoint(otlp_endpoint) + .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) + .build()?; + + // (3) Wire the exporter into a tracer provider. The `service.name` + // resource is what Jaeger lists in its service dropdown, so the + // client and server show up as separate boxes joined by trace id; + // `service.instance.id` carries the iroh endpoint id so multiple + // clients/servers on the same trace are distinguishable. + let provider = SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .with_resource( + Resource::builder() + .with_attribute(KeyValue::new("service.name", service_name)) + .with_attribute(KeyValue::new( + "service.instance.id", + endpoint_id.fmt_short().to_string(), + )) + .build(), + ) + .build(); + opentelemetry::global::set_tracer_provider(provider.clone()); + + // (4) Bridge `tracing` into OpenTelemetry. Every `tracing` span flows + // through the `tracing-opentelemetry` layer (which forwards to OTLP + // and is filtered to `info` so noisy debug spans don't ship to + // Jaeger), in parallel with a plain fmt layer that prints to stderr + // under `RUST_LOG`. + let tracer = opentelemetry::global::tracer(service_name); + let telemetry = tracing_opentelemetry::layer() + .with_tracer(tracer) + .with_filter(EnvFilter::new("info")); + let fmt_filter = EnvFilter::try_from_default_env() .unwrap_or_else(|_| EnvFilter::new("span_propagation=info")); - let telemetry = tracing_opentelemetry::layer().with_tracer(tracer); let subscriber = Registry::default().with(telemetry).with( tracing_subscriber::fmt::layer() .with_target(false) - .with_filter(env), + .with_filter(fmt_filter), ); tracing::subscriber::set_global_default(subscriber)?; + Ok(provider) +} + +#[tokio::main] +async fn main() -> Result<()> { + let cli = Cli::parse(); + match cli.command { + Command::Server => server(&cli.otlp).await, + Command::Client { endpoint_id, count } => client(&cli.otlp, endpoint_id, count).await, + } +} - // (2) Server: a handler that opts into span propagation via the - // `span_propagation` flag on `rpc_requests` (above). +async fn server(otlp_endpoint: &str) -> Result<()> { + // The protocol opts into span propagation via the `span_propagation` flag + // on `rpc_requests` above. let endpoint = Endpoint::bind(presets::N0).await?; + let provider = init_tracing("example-server", endpoint.id(), otlp_endpoint)?; let protocol = IrohProtocol::::new(Arc::new(|req, rx, tx| { Box::pin(async move { let msg: Message = ::with_remote_channels(req, rx, tx); match msg { Message::Get(msg) => { - // (3) `WithChannels` carries a `span` field. The derive macro - // created it as `info_span!("Get")` and set its parent - // from the propagated remote context. Entering it makes - // the rest of the handler — and any child spans it - // creates — part of the same distributed trace. + // `WithChannels` carries a `span` field that the derive + // macro builds as `info_span!("Get")` with its parent set + // from the propagated remote context, so the handler's + // work shows up under the originating client trace. let WithChannels { inner, tx, span, .. } = msg; - let _guard = span.enter(); - info!(?inner, "handling request"); - tx.send(inner.0.to_uppercase()).await.ok(); + // Hand the response off to a separate task to show that + // `.instrument(span)` keeps work hooked into the same + // trace even after the handler future returns. + tokio::spawn( + async move { + info!(?inner, "handling request"); + // Simulate async work. + tokio::time::sleep(Duration::from_millis(rand::random_range(20..200))) + .await; + let res = inner.0.to_uppercase(); + info!(?res, "generated response"); + tx.send(res).await.ok(); + info!("response sent"); + } + .instrument(span), + ); } } Ok(()) }) })); - let server = Router::builder(endpoint).accept(ALPN, protocol).spawn(); + let router = Router::builder(endpoint).accept(ALPN, protocol).spawn(); + + println!("server endpoint id: {}", router.endpoint().id()); + println!("run the client with:"); + println!( + " cargo run -p irpc-iroh --features tracing-opentelemetry --example span_propagation -- client {}", + router.endpoint().id() + ); + println!("press ctrl+c to stop"); + tokio::signal::ctrl_c().await?; + + router.shutdown().await?; + let _ = provider.force_flush(); + let _ = provider.shutdown(); + Ok(()) +} - // Client: every request lives inside a `req` span. +async fn client(otlp_endpoint: &str, endpoint_id: EndpointId, count: u32) -> Result<()> { + // Each request lives inside a `req` span; the `Get` span the server + // produces will hang off that one in Jaeger. let client_ep = Endpoint::bind(presets::N0).await?; - let server_addr = server.endpoint().addr(); - for req_id in 0..3 { - let client = irpc_iroh::client::(client_ep.clone(), server_addr.clone(), ALPN); + let provider = init_tracing("example-client", client_ep.id(), otlp_endpoint)?; + for req_id in 0..count { + let client = irpc_iroh::client::(client_ep.clone(), endpoint_id, ALPN); let payload = format!("hello-{req_id}"); async { info!(%payload, "sending request"); @@ -105,7 +217,8 @@ async fn main() -> Result<()> { .await?; } - server.shutdown().await?; client_ep.close().await; + let _ = provider.force_flush(); + let _ = provider.shutdown(); Ok(()) }