Example: distributed tracing

The client orders pizzas and the calls executed by the server appear as spans in the client's trace.

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.

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: 

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

The program

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

pizzeria

Defines the remote trait 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"] }

pizzeria/src/lib.rs

Defines the trait whose tracing argument makes the server create a span for processing each call.

//! This library crate defines the remote pizzeria service.
//!
//! 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,
}

/// 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.
    async fn order(&self, pizza: Pizza) -> Result<String, rtc::CallError>;
}

pizzeria-server

Prepares each ordered pizza in instrumented steps.

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"
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-server/src/main.rs

Implements the trait with #[instrument] steps and exports its spans when an OpenTelemetry endpoint is configured.

//! 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, TCP_PORT};

/// 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) -> 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.
        self.prepare_dough(pizza).await;
        self.add_toppings(pizza).await;
        self.bake().await;

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

/// 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.
fn init_tracing() -> Option<opentelemetry_sdk::trace::SdkTracerProvider> {
    use opentelemetry::trace::TracerProvider;
    use tracing_subscriber::{Layer, filter, 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("pizzeria-server").build(),
            )
            .build()
    });

    // The OpenTelemetry layer assigns globally meaningful identifiers to the
    // spans of the tracing crate and exports them.
    // Only spans of level info and above are exported; this also excludes the
    // spans of the HTTP/2 library that performs the export.
    let otel_layer = provider.as_ref().map(|p| {
        tracing_opentelemetry::layer()
            .with_tracer(p.tracer("pizzeria-server"))
            .with_filter(filter::LevelFilter::INFO)
    });

    // Log to the terminal at info level, overridable via `RUST_LOG`.
    let fmt_layer = tracing_subscriber::fmt::layer().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
}

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

    // 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();
            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"
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-client/src/main.rs

Places the orders concurrently within one span and exports the resulting trace.

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

use remoc::prelude::*;
use std::net::Ipv4Addr;
use tokio::net::TcpStream;
use tracing::{Instrument, info_span};

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

/// 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.
fn init_tracing() -> Option<opentelemetry_sdk::trace::SdkTracerProvider> {
    use opentelemetry::trace::TracerProvider;
    use tracing_subscriber::{Layer, filter, 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("pizzeria-client").build(),
            )
            .build()
    });

    // The OpenTelemetry layer assigns globally meaningful identifiers to the
    // spans of the tracing crate and exports them.
    // Only spans of level info and above are exported; this also excludes the
    // spans of the HTTP/2 library that performs the export.
    let otel_layer = provider.as_ref().map(|p| {
        tracing_opentelemetry::layer()
            .with_tracer(p.tracer("pizzeria-client"))
            .with_filter(filter::LevelFilter::INFO)
    });

    // Log to the terminal at info level, overridable via `RUST_LOG`.
    let fmt_layer = tracing_subscriber::fmt::layer().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
}

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

    // Establish TCP connection to server.
    let socket = TcpStream::connect((Ipv4Addr::LOCALHOST, TCP_PORT)).await.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 within a span, so that the spans of the calls executed by
    // the server are linked to it in one distributed trace.
    async {
        // 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");

        // 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))).await.unwrap();

        for delivery in deliveries {
            println!("Received: {delivery}");
        }
    }
    .instrument(info_span!("pizza order"))
    .await;

    // Send the spans of the order to the collector.
    if let Some(provider) = &provider {
        provider.force_flush().unwrap();
    }
}

Browse it on GitHub