Distributed tracing of remote trait calls: a Remoc example

The pizzeria trait specifies the tracing argument in its #[rtc::remote] attribute, so the client creates a span for each call and the server one for processing it. The tracing context travels with the call, linking the span of the server to the span of the call across the connection.

The server prepares each pizza in steps carrying the #[instrument] attribute of the tracing crate, whose spans nest within the span of the call. The client orders all pizzas at once, so their preparation runs in parallel on the server, which is visible in the timing of the spans.

The client also passes a remote callback with each order, through which the server reports the progress of the preparation. Calling it is a remote call in the opposite direction, from the server to the client, and its spans are linked into the same trace.

Running it

Clone the repository:

git clone https://github.com/remoc-rs/remoc
cd remoc

Run an OpenTelemetry collector, for example otel-tui, which displays the traces directly in the terminal:

otel-tui

Start the server:

OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \
    cargo run --manifest-path examples/tracing/Cargo.toml -p pizzeria-server

Then, in another terminal, start the client:

OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \
    cargo run --manifest-path examples/tracing/Cargo.toml -p pizzeria-client

The collector shows the trace of the pizza order, with the three orders executing in parallel on the server and the progress reports crossing back to the client:

[pizzeria-client] order_pizzas
├─ [pizzeria-client] Pizzeria::order
│  └─ [pizzeria-server] Pizzeria::order
│     ├─ prepare_dough
│     ├─ [pizzeria-server] progress
│     │  └─ [pizzeria-client] progress
│     ├─ add_toppings
│     ├─ [pizzeria-server] progress
│     │  └─ [pizzeria-client] progress
│     ├─ bake
│     └─ [pizzeria-server] progress
│        └─ [pizzeria-client] progress
├─ [pizzeria-client] Pizzeria::order
│  └─ …
└─ [pizzeria-client] Pizzeria::order
   └─ …

The program

The shared crate defines the remote trait, the progress callback and the tracing setup, and the other two crates provide the server and client binaries.

pizzeria

Defines the remote trait and the tracing setup shared by the client and server.

`pizzeria/Cargo.toml`

Its dependencies.

[package]
name = "pizzeria"
version = "0.1.0"
edition = "2024"

[dependencies]
remoc = { version = "0.20" }
serde = { version = "1.0", features = ["derive"] }
tracing-subscriber = { version = "0.3.22", features = ["env-filter"] }
tracing-opentelemetry = "0.33"
opentelemetry = "0.32"
opentelemetry_sdk = "0.32"
opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic"] }

`pizzeria/src/lib.rs`

Defines the trait whose tracing argument makes the server create a span for processing each call, the progress callback type and the tracing subscriber with optional span export.

//! This library crate defines the remote pizzeria service and provides the
//! tracing setup shared by client and server.
//!
//! The client and server depend on it.
#![warn(missing_docs)]

use remoc::prelude::*;

/// TCP port the server is listening on.
pub const TCP_PORT: u16 = 9873;

/// A pizza on the menu.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum Pizza {
    /// Tomatoes, mozzarella and basil.
    Margherita,
    /// Tomatoes, mozzarella and salami.
    Salami,
    /// Tomatoes, mozzarella, ham and pineapple.
    Hawaii,
}

/// Callback reporting the progress of an order.
///
/// The remote function is created by the client and called by the server
/// for each completed step of preparing the pizza. Its calls are traced as
/// well, when the client enables it on the function.
pub type Progress = rfn::RFn<(String,), ()>;

/// Remote pizzeria service.
///
/// The `tracing` argument makes the client create a span at info level for
/// each call and the server one for processing it, which is linked into the
/// trace of the client.
#[rtc::remote(server(Shared), tracing)]
pub trait Pizzeria {
    /// The pizzas on offer.
    ///
    /// Querying the menu is cheap and frequent, so no span is created for it.
    #[tracing(level = "off")]
    async fn menu(&self) -> Result<Vec<Pizza>, rtc::CallError>;

    /// Prepares and bakes the specified pizza, reporting each completed step
    /// through the progress callback.
    async fn order(&self, pizza: Pizza, progress: Progress) -> Result<String, rtc::CallError>;
}

/// Initializes logging to the terminal and, if the environment variable
/// `OTEL_EXPORTER_OTLP_ENDPOINT` is set, span export to the OpenTelemetry
/// collector at that endpoint.
///
/// The spans are exported under the specified service name.
pub fn init_tracing(service_name: &'static str) -> Option<opentelemetry_sdk::trace::SdkTracerProvider> {
    use opentelemetry::trace::TracerProvider;
    use tracing_subscriber::{
        Layer, filter, fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt,
    };

    let provider = std::env::var_os("OTEL_EXPORTER_OTLP_ENDPOINT").map(|_| {
        let exporter = opentelemetry_otlp::SpanExporter::builder().with_tonic().build().unwrap();
        opentelemetry_sdk::trace::SdkTracerProvider::builder()
            .with_batch_exporter(exporter)
            .with_resource(opentelemetry_sdk::Resource::builder().with_service_name(service_name).build())
            .build()
    });

    // The OpenTelemetry layer assigns globally meaningful identifiers to the
    // spans of the tracing crate and exports them.
    let otel_layer = provider.as_ref().map(|p| {
        tracing_opentelemetry::layer().with_tracer(p.tracer(service_name)).with_filter(filter::LevelFilter::INFO)
    });

    // Log to the terminal at info level, overridable via `RUST_LOG`.
    // Opening and closing spans are logged as well, so that the spans of the
    // calls and their identifiers are visible without a collector.
    let fmt_layer = tracing_subscriber::fmt::layer().with_span_events(FmtSpan::NEW | FmtSpan::CLOSE).with_filter(
        filter::EnvFilter::builder().with_default_directive(filter::LevelFilter::INFO.into()).from_env_lossy(),
    );

    tracing_subscriber::registry().with(fmt_layer).with(otel_layer).init();

    provider
}

pizzeria-server

Prepares each ordered pizza in instrumented steps, reporting progress to the client.

`pizzeria-server/Cargo.toml`

Its dependencies.

[package]
name = "pizzeria-server"
version = "0.1.0"
edition = "2024"

[dependencies]
pizzeria = { path = "../pizzeria" }
remoc = { version = "0.20" }
tokio = { version = "1", features = ["rt-multi-thread", "net", "signal", "sync", "time"] }
tracing = "0.1"

`pizzeria-server/src/main.rs`

Implements the trait with #[instrument] steps and calls the progress callback of the client after each of them.

//! This crate implements the server of the remote pizzeria service.
#![warn(missing_docs)]

use remoc::{codec, prelude::*};
use std::{net::Ipv4Addr, sync::Arc, time::Duration};
use tokio::{net::TcpListener, time::sleep};
use tracing::{Instrument, info_span, instrument};

use pizzeria::{Pizza, Pizzeria, PizzeriaServerShared, Progress, TCP_PORT, init_tracing};

/// Server object for the pizzeria service.
pub struct PizzeriaObj;

impl PizzeriaObj {
    /// Prepares the dough.
    #[instrument(skip(self))]
    async fn prepare_dough(&self, pizza: Pizza) {
        sleep(Duration::from_millis(300)).await;
        tracing::info!("the dough is ready");
    }

    /// Puts the toppings for the specified pizza onto the dough.
    #[instrument(skip(self))]
    async fn add_toppings(&self, pizza: Pizza) {
        sleep(Duration::from_millis(150)).await;
    }

    /// Bakes the pizza.
    #[instrument(skip(self))]
    async fn bake(&self) {
        sleep(Duration::from_millis(500)).await;
        tracing::info!("out of the oven");
    }
}

/// Implementation of the remote pizzeria service.
impl Pizzeria for PizzeriaObj {
    async fn menu(&self) -> Result<Vec<Pizza>, rtc::CallError> {
        Ok(vec![Pizza::Margherita, Pizza::Salami, Pizza::Hawaii])
    }

    async fn order(&self, pizza: Pizza, progress: Progress) -> Result<String, rtc::CallError> {
        // These methods carry the #[instrument] attribute of the tracing
        // crate, thus their spans become children of the span of this call.
        //
        // After each step the progress callback of the client is called.
        // This is a remote function call in the opposite direction, whose
        // spans are linked into the trace of this call as well.
        // Reporting is best-effort, so failures to reach the client are
        // ignored and the pizza is finished regardless.
        self.prepare_dough(pizza).await;
        let _ = progress.try_call(format!("{pizza:?}: dough is ready")).await;
        self.add_toppings(pizza).await;
        let _ = progress.try_call(format!("{pizza:?}: toppings are on")).await;
        self.bake().await;
        let _ = progress.try_call(format!("{pizza:?}: baked")).await;

        Ok(format!("{pizza:?} pizza, fresh out of the oven"))
    }
}

#[tokio::main]
async fn main() {
    // Initialize logging and trace export.
    let provider = init_tracing("pizzeria-server");

    // Create a pizzeria object that will be shared between all clients.
    let pizzeria_obj = Arc::new(PizzeriaObj);

    let serve = async move {
        // Listen to TCP connections using Tokio.
        // In reality you would probably use TLS or WebSockets over HTTPS.
        println!("Listening on port {}. Press Ctrl+C to exit.", TCP_PORT);
        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, TCP_PORT)).await.unwrap();

        loop {
            // Accept an incoming TCP connection.
            let (socket, addr) = listener.accept().await.unwrap();
            socket.set_nodelay(true).unwrap();
            let (socket_rx, socket_tx) = socket.into_split();
            println!("Accepted connection from {}", addr);

            // Create a new shared reference to the pizzeria object.
            let pizzeria_obj = pizzeria_obj.clone();

            // Spawn a task for each incoming connection.
            tokio::spawn(
                async move {
                    // Create a server proxy and client for the accepted connection.
                    //
                    // The server proxy executes all incoming method calls on the
                    // shared pizzeria_obj.
                    //
                    // Current limitations of the Rust compiler require that we
                    // explicitly specify the codec.
                    let (server, client) = PizzeriaServerShared::<_, codec::Default>::new(pizzeria_obj);

                    // Establish a Remoc connection with default configuration over
                    // the TCP connection and provide (i.e. send) the pizzeria
                    // client to the client.
                    remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx)
                        .provide(client)
                        .await
                        .unwrap();

                    // Serve incoming requests from the client on this task.
                    // Requests are executed in parallel, so several pizzas can be
                    // prepared at once.
                    server.serve().await.unwrap();
                }
                .instrument(info_span!("incoming", %addr)),
            );
        }
    };

    // Serve until Ctrl+C is pressed.
    tokio::select! {
        () = serve => (),
        _ = tokio::signal::ctrl_c() => (),
    }

    // Send the remaining spans to the collector.
    if let Some(provider) = provider {
        provider.shutdown().unwrap();
    }
}

pizzeria-client

Orders one of each pizza on the menu, all at once.

`pizzeria-client/Cargo.toml`

Its dependencies.

[package]
name = "pizzeria-client"
version = "0.1.0"
edition = "2024"

[dependencies]
pizzeria = { path = "../pizzeria" }
remoc = { version = "0.20" }
futures = "0.3"
tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "time"] }
tracing = "0.1"

`pizzeria-client/src/main.rs`

Places the orders concurrently within one span, passing a traced progress callback with each.

//! This crate implements the client of the remote pizzeria service.
#![warn(missing_docs)]

use remoc::prelude::*;
use std::{net::Ipv4Addr, time::Duration};
use tokio::net::TcpStream;
use tracing::{Span, instrument, level_filters::LevelFilter};

use pizzeria::{Pizzeria, PizzeriaClient, Progress, TCP_PORT, init_tracing};

/// Orders one of each pizza on the menu.
///
/// The `#[instrument]` attribute of the tracing crate creates a span for this
/// function, so that the spans of the calls executed by the server are
/// linked to it in one distributed trace.
#[instrument(skip(client))]
async fn order_pizzas(client: &PizzeriaClient) {
    // Query the menu. No span is created for this call, since the menu
    // method sets the tracing level "off".
    let menu = client.menu().await.unwrap();
    println!("On the menu today: {menu:?}\n");

    // Create the callback through which the server reports the progress of
    // the orders. It is executed here, at the client, whenever the server
    // calls it.
    //
    // Calls of a remote function are not traced by default. Enabling them at
    // info level makes the server create a span for each call and the client
    // one for processing it, which is linked into the trace of the order.
    // The name is recorded in these spans.
    //
    // The callback is only called while the orders are running, so its calls
    // are processed within the span of this function. The span then stays
    // open until the callback is dropped by the server and the client, which
    // happens shortly after the orders complete. To close it exactly when
    // this function returns, create the callback using `provided_1` and keep
    // its provider in this function instead.
    let mut progress = Progress::new_1(|step: String| async move {
        tracing::info!("progress report: {step}");
    });
    progress.set_name("progress");
    progress.set_tracing_level(LevelFilter::INFO);
    progress.set_span(Span::current());

    // Order one of each pizza, all at once.
    // The server prepares the pizzas in parallel, which is visible in
    // the timing of their spans within the trace.
    println!("Ordering one of each...");
    let deliveries =
        futures::future::try_join_all(menu.iter().map(|&pizza| client.order(pizza, progress.clone())))
            .await
            .unwrap();

    for delivery in deliveries {
        println!("Received: {delivery}");
    }
}

#[tokio::main]
async fn main() {
    // Initialize logging and trace export.
    let provider = init_tracing("pizzeria-client");

    // Establish TCP connection to server.
    let socket = TcpStream::connect((Ipv4Addr::LOCALHOST, TCP_PORT)).await.unwrap();
    socket.set_nodelay(true).unwrap();
    let (socket_rx, socket_tx) = socket.into_split();

    // Establish a Remoc connection with default configuration over the TCP
    // connection and consume (i.e. receive) the pizzeria client from the server.
    let client: PizzeriaClient =
        remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx).consume().await.unwrap();

    order_pizzas(&client).await;

    // Send the spans of the order to the collector, after giving the span
    // of the order a moment to close, see above.
    if let Some(provider) = &provider {
        tokio::time::sleep(Duration::from_millis(100)).await;
        provider.force_flush().unwrap();
    }
}

Browse it on GitHub