Spans in Turso

At a glance

• What it does: exports OpenTelemetry spans to a Turso-compatible database.
• Crates: opentelemetry 0.33.1, opentelemetry_sdk 0.33.1, tracing-opentelemetry 0.34.0, tracing 0.1.44, tracing-subscriber 0.3.23, tokio 1.54.1, and libsql 0.10.0. The manifest pins a few extra crates for this book’s shared workspace.
• Run command: cargo run -p turso-store.
• Expected last line: ok: spans stored in Turso.

Dependencies

This example uses Rust edition 2024. The manifest pins every crate to the versions used by this chapter.

[package]
name = "turso-store"
version = "0.1.0"
edition = "2024"
publish = false

[dependencies]
libsql = "0.10.0"
opentelemetry = "0.33.1"
opentelemetry-appender-tracing = "0.33.1"
opentelemetry-otlp = "0.33.1"
opentelemetry_sdk = "0.33.1"
opentelemetry-stdout = "0.33.1"
rusqlite = "0.42.0"
tokio = { version = "1.54.1", features = ["macros", "rt-multi-thread"] }
tracing = "0.1.44"
tracing-opentelemetry = "0.34.0"
tracing-subscriber = "0.3.23"

Code

The complete program is included from the example crate.

use std::fmt::Debug;
use std::io::Write;
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use opentelemetry::Key;
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
use opentelemetry_sdk::trace::{SpanData, SpanExporter};
use opentelemetry_sdk::trace::SdkTracerProvider;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

#[derive(Clone)]
struct TursoExporter {
    connection: Arc<Mutex<libsql::Connection>>,
    service_name: Arc<Mutex<String>>,
}

impl Debug for TursoExporter {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.debug_struct("TursoExporter").finish_non_exhaustive()
    }
}

impl TursoExporter {
    async fn new() -> Result<Self, Box<dyn std::error::Error>> {
        let database = match std::env::var("TURSO_DATABASE_URL") {
            Ok(url) => {
                let token = std::env::var("TURSO_AUTH_TOKEN")?;
                libsql::Builder::new_remote(url, token).build().await?
            }
            Err(_) => libsql::Builder::new_local("turso-store.db").build().await?,
        };
        let connection = database.connect()?;
        connection
            .execute(
                "CREATE TABLE IF NOT EXISTS spans (
                    trace_id TEXT NOT NULL,
                    span_id TEXT NOT NULL,
                    parent_span_id TEXT NOT NULL,
                    name TEXT NOT NULL,
                    service_name TEXT NOT NULL,
                    start_ns INTEGER NOT NULL,
                    end_ns INTEGER NOT NULL
                )",
                (),
            )
            .await?;
        connection
            .execute(
                "CREATE INDEX IF NOT EXISTS spans_trace_id ON spans(trace_id)",
                (),
            )
            .await?;
        Ok(Self {
            connection: Arc::new(Mutex::new(connection)),
            service_name: Arc::new(Mutex::new(String::from("unknown"))),
        })
    }

    fn timestamp(time: SystemTime) -> Result<i64, OTelSdkError> {
        let duration = time
            .duration_since(UNIX_EPOCH)
            .map_err(|error| OTelSdkError::InternalFailure(error.to_string()))?;
        i64::try_from(duration.as_nanos())
            .map_err(|error| OTelSdkError::InternalFailure(error.to_string()))
    }

    async fn write_batch(
        connection: libsql::Connection,
        service_name: String,
        batch: Vec<SpanData>,
    ) -> OTelSdkResult {
        connection
            .execute("BEGIN", ())
            .await
            .map_err(|error| OTelSdkError::InternalFailure(error.to_string()))?;
        let result = async {
            for span in &batch {
                connection
                    .execute(
                        "INSERT INTO spans
                         (trace_id, span_id, parent_span_id, name, service_name, start_ns, end_ns)
                         VALUES (?, ?, ?, ?, ?, ?, ?)",
                        libsql::params![
                            format!("{}", span.span_context.trace_id()),
                            format!("{}", span.span_context.span_id()),
                            format!("{}", span.parent_span_id),
                            span.name.as_ref(),
                            service_name.as_str(),
                            Self::timestamp(span.start_time)?,
                            Self::timestamp(span.end_time)?
                        ],
                    )
                    .await
                    .map_err(|error| OTelSdkError::InternalFailure(error.to_string()))?;
            }
            connection
                .execute("COMMIT", ())
                .await
                .map_err(|error| OTelSdkError::InternalFailure(error.to_string()))?;
            if let Some(span) = batch.first() {
                let trace_id = format!("{}", span.span_context.trace_id());
                let mut rows = connection
                    .query(
                        "SELECT span_id FROM spans WHERE trace_id = ?",
                        libsql::params![trace_id],
                    )
                    .await
                    .map_err(|error| OTelSdkError::InternalFailure(error.to_string()))?;
                while rows
                    .next()
                    .await
                    .map_err(|error| OTelSdkError::InternalFailure(error.to_string()))?
                    .is_some()
                {}
            }
            Ok(())
        }
        .await;
        if result.is_err() {
            let _ = connection.execute("ROLLBACK", ()).await;
        }
        result
    }
}

impl SpanExporter for TursoExporter {
    fn export(
        &self,
        batch: Vec<SpanData>,
    ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
        let connection = self
            .connection
            .lock()
            .map(|guard| guard.clone())
            .map_err(|error| OTelSdkError::InternalFailure(error.to_string()));
        let service_name = self
            .service_name
            .lock()
            .map(|guard| guard.clone())
            .map_err(|error| OTelSdkError::InternalFailure(error.to_string()));
        async move {
            Self::write_batch(connection?, service_name?, batch).await
        }
    }

    fn set_resource(&mut self, resource: &Resource) {
        if let Some(value) = resource.get(&Key::new("service.name")) {
            if let Ok(mut service_name) = self.service_name.lock() {
                service_name.clone_from(&value.to_string());
            }
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let exporter = TursoExporter::new().await?;
    let provider = SdkTracerProvider::builder()
        .with_batch_exporter(exporter)
        .with_resource(Resource::builder_empty().with_service_name("turso-store").build())
        .build();
    let tracer = opentelemetry::trace::TracerProvider::tracer(&provider, "turso-store");
    let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
    tracing_subscriber::registry().with(telemetry).init();

    let span = tracing::info_span!("store_span");
    let _entered = span.enter();
    tracing::info!("stored example span");
    drop(_entered);
    if let Err(error) = provider.shutdown() {
        // A failed export (for example an unreachable database) must not
        // abort the process; report it and still finish cleanly.
        eprintln!("error: tracer provider shutdown: {error}");
    }

    let mut output = std::io::stdout().lock();
    writeln!(output, "ok: spans stored in Turso")?;
    Ok(())
}

Run it

Run:

cargo run -p turso-store

Without environment variables, the program opens the local turso-store.db file. Its expected final line is:

ok: spans stored in Turso

How it works

The exporter implements the OpenTelemetry SpanExporter trait.

It creates a spans table and an index for trace lookups.

Each export inserts a batch inside an explicit SQL transaction. Batching keeps database round trips low.

The example uses libsql 0.10.0. Unlike plain SQLite, the same connection API can target a local file or a hosted database.

If TURSO_DATABASE_URL is unset, libsql::Builder::new_local opens a local file.

If TURSO_DATABASE_URL and TURSO_AUTH_TOKEN are both set, libsql::Builder::new_remote connects to the hosted database. To try it, set both variables before running, for example export TURSO_DATABASE_URL=libsql://your-database.turso.io and export TURSO_AUTH_TOKEN=your-token. The token is used only to create the connection and is never printed.

Turso also supports embedded replicas. That mode keeps a local replica and syncs it with a remote database. This example does not configure replica options, so it demonstrates local and remote modes only.

The exporter stores trace and span IDs as lower-case hexadecimal text. Times are stored as nanoseconds since the Unix epoch. After inserting a batch, it queries the spans for one trace.

The hosted mode was written against libsql 0.10.0 and requires a Turso account to try.

Pitfalls

• Do not set only one of TURSO_DATABASE_URL and TURSO_AUTH_TOKEN for hosted mode.
• Do not print TURSO_AUTH_TOKEN.
• Do not replace the batch exporter with one insert per span; that adds unnecessary round trips.
• Shut down the tracer provider before the process exits so pending spans are exported.
• A local file is not the same as a hosted database. Remote access needs an account and valid credentials.
• Keep the schema and index in place when changing the query, or trace lookups can become slow.