Spans in SQLite
At a glance
- What it does: writes finished OpenTelemetry spans to SQLite.
- Crates:
opentelemetry0.33.1,opentelemetry_sdk0.33.1,tracing-opentelemetry0.34.0,tracing0.1.44,tracing-subscriber0.3.23, andrusqlite0.42.0. - Run command:
cargo run -p sqlite-store - Expected last line:
ok: spans stored in SQLite
Dependencies
[package]
name = "sqlite-store"
version = "0.1.0"
edition = "2024"
publish = false
[dependencies]
opentelemetry = "=0.33.1"
opentelemetry_sdk = { version = "=0.33.1", features = ["rt-tokio"] }
opentelemetry-otlp = "=0.33.1"
opentelemetry-stdout = "=0.33.1"
opentelemetry-appender-tracing = "=0.33.1"
tracing-opentelemetry = "=0.34.0"
tracing = "=0.1.44"
tracing-subscriber = "=0.3.23"
tokio = { version = "=1.54.1", features = ["macros", "rt-multi-thread"] }
rusqlite = { version = "=0.42.0", features = ["bundled"] }
The manifest also pins opentelemetry-otlp, opentelemetry-stdout, opentelemetry-appender-tracing, and tokio for the book’s shared workspace.
Code
use opentelemetry::trace::{SpanKind, Status, TracerProvider as _};
use opentelemetry::{Key, KeyValue, Value};
use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
use opentelemetry_sdk::trace::{SpanData, SpanExporter};
use opentelemetry_sdk::trace::SdkTracerProvider;
use opentelemetry_sdk::Resource;
use rusqlite::Connection;
use std::fmt::Debug;
use std::sync::Mutex;
use std::time::UNIX_EPOCH;
use tracing_opentelemetry::OpenTelemetryLayer;
use tracing_subscriber::prelude::*;
use tracing_subscriber::Registry;
#[derive(Debug)]
struct SqliteExporter {
connection: Mutex<Connection>,
service_name: Mutex<String>,
}
impl SqliteExporter {
fn new(path: &str) -> Result<Self, Box<dyn std::error::Error>> {
let connection = Connection::open(path)?;
connection.execute_batch(
"PRAGMA journal_mode = WAL;
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,
kind TEXT NOT NULL,
start_unix_nanos INTEGER NOT NULL,
end_unix_nanos INTEGER NOT NULL,
status TEXT NOT NULL,
attributes TEXT NOT NULL,
service_name TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS spans_trace_id ON spans(trace_id);
CREATE VIRTUAL TABLE IF NOT EXISTS spans_fts
USING fts5(name, attributes, content='spans', content_rowid='rowid');",
)?;
Ok(Self {
connection: Mutex::new(connection),
service_name: Mutex::new("sqlite-store".to_string()),
})
}
fn write_batch(&self, batch: Vec<SpanData>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut connection = self
.connection
.lock()
.map_err(|_| std::io::Error::other("SQLite connection mutex was poisoned"))?;
let service_name = self
.service_name
.lock()
.map_err(|_| std::io::Error::other("service name mutex was poisoned"))?
.clone();
let transaction = connection.transaction()?;
for span in batch {
let start = i64::try_from(span.start_time.duration_since(UNIX_EPOCH)?.as_nanos())?;
let end = i64::try_from(span.end_time.duration_since(UNIX_EPOCH)?.as_nanos())?;
let attributes = attributes_json(&span.attributes);
let name = span.name.to_string();
let kind = span_kind_name(span.span_kind);
let status = status_name(&span.status);
transaction.execute(
"INSERT INTO spans
(trace_id, span_id, parent_span_id, name, kind, start_unix_nanos,
end_unix_nanos, status, attributes, service_name)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
rusqlite::params![
format!("{}", span.span_context.trace_id()),
format!("{}", span.span_context.span_id()),
format!("{}", span.parent_span_id),
name,
kind,
start,
end,
status,
attributes,
service_name,
],
)?;
let rowid = transaction.last_insert_rowid();
transaction.execute(
"INSERT INTO spans_fts(rowid, name, attributes) VALUES (?1, ?2, ?3)",
rusqlite::params![rowid, span.name.to_string(), attributes_json(&span.attributes)],
)?;
}
transaction.commit()?;
Ok(())
}
}
impl SpanExporter for SqliteExporter {
fn export(
&self,
batch: Vec<SpanData>,
) -> impl std::future::Future<Output = OTelSdkResult> + Send {
std::future::ready(self.write_batch(batch).map_err(|error| {
OTelSdkError::InternalFailure(format!("SQLite span export failed: {error}"))
}))
}
fn shutdown(&self) -> OTelSdkResult {
Ok(())
}
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());
}
}
}
}
fn json_escape(value: &str) -> String {
let mut escaped = String::new();
for character in value.chars() {
match character {
'"' => escaped.push_str("\\\""),
'\\' => escaped.push_str("\\\\"),
'\n' => escaped.push_str("\\n"),
'\r' => escaped.push_str("\\r"),
'\t' => escaped.push_str("\\t"),
character => escaped.push(character),
}
}
escaped
}
fn attributes_json(attributes: &[KeyValue]) -> String {
let mut output = String::from("{");
for attribute in attributes {
let value = match &attribute.value {
Value::Bool(value) => value.to_string(),
Value::I64(value) => value.to_string(),
Value::F64(value) => value.to_string(),
Value::String(value) => value.to_string(),
Value::Array(value) => format!("{value:?}"),
_ => continue,
};
if output.len() > 1 {
output.push(',');
}
output.push('"');
output.push_str(&json_escape(attribute.key.as_str()));
output.push_str("\":\"");
output.push_str(&json_escape(&value));
output.push('"');
}
output.push('}');
output
}
fn span_kind_name(kind: SpanKind) -> &'static str {
match kind {
SpanKind::Internal => "internal",
SpanKind::Server => "server",
SpanKind::Client => "client",
SpanKind::Producer => "producer",
SpanKind::Consumer => "consumer",
}
}
fn status_name(status: &Status) -> String {
match status {
Status::Unset => "unset".to_string(),
Status::Ok => "ok".to_string(),
Status::Error { description } => format!("error: {description}"),
}
}
fn main() -> Result<(), Box<dyn std::error::Error>> {
let exporter = SqliteExporter::new("spans.sqlite3")?;
let provider = SdkTracerProvider::builder()
.with_batch_exporter(exporter)
.build();
let tracer = provider.tracer("sqlite-store");
let layer: OpenTelemetryLayer<Registry, _> =
tracing_opentelemetry::layer().with_tracer(tracer);
let subscriber = tracing_subscriber::registry()
.with(layer)
.with(tracing_subscriber::fmt::layer());
tracing::subscriber::set_global_default(subscriber)?;
let root = tracing::info_span!("root");
let child = tracing::info_span!("child", task = "work").entered();
tracing::info!("nested work");
drop(child);
root.in_scope(|| tracing::info!("root work"));
provider.shutdown()?;
let connection = Connection::open("spans.sqlite3")?;
let trace_id: String = connection.query_row(
"SELECT trace_id FROM spans ORDER BY start_unix_nanos LIMIT 1",
[],
|row| row.get(0),
)?;
let mut rows = connection.prepare(
"SELECT span_id, parent_span_id, name, start_unix_nanos, end_unix_nanos
FROM spans WHERE trace_id = ?1 ORDER BY start_unix_nanos",
)?;
for row in rows.query_map([&trace_id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, i64>(4)?,
))
})? {
let (span_id, parent_span_id, name, start, end) = row?;
println!("trace {span_id} parent {parent_span_id} {name} {start}-{end}");
}
let mut slow = connection.prepare(
"SELECT name, end_unix_nanos - start_unix_nanos
FROM spans ORDER BY end_unix_nanos - start_unix_nanos DESC LIMIT 10",
)?;
for row in slow.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
})? {
let (name, duration) = row?;
println!("slow {name}: {duration}ns");
}
let mut search = connection.prepare(
"SELECT name, attributes FROM spans_fts WHERE spans_fts MATCH ?1",
)?;
for row in search.query_map(["work"], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})? {
let (name, attributes) = row?;
println!("search {name}: {attributes}");
}
println!("ok: spans stored in SQLite");
Ok(())
}Run it
cargo run -p sqlite-store
The output includes stored span rows, slow-span rows, and a full-text search result for the child span whose task attribute is "work". Its last line starts with:
ok: spans stored in SQLite
How it works
The exporter opens spans.sqlite3 and enables write-ahead logging.
It creates this schema:
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,
kind TEXT NOT NULL,
start_unix_nanos INTEGER NOT NULL,
end_unix_nanos INTEGER NOT NULL,
status TEXT NOT NULL,
attributes TEXT NOT NULL,
service_name TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS spans_trace_id ON spans(trace_id);
CREATE VIRTUAL TABLE IF NOT EXISTS spans_fts
USING fts5(name, attributes, content='spans', content_rowid='rowid');
SpanExporter receives finished spans in batches. The exporter keeps its SQLite connection behind a mutex because export takes &self.
Each batch is inserted in one transaction. The regular row and its FTS row are written in that same transaction. This keeps search data in step with the spans table.
The provider uses with_batch_exporter. Shutdown flushes pending spans before the program queries the database.
The trace query is:
SELECT span_id, parent_span_id, name, start_unix_nanos, end_unix_nanos
FROM spans
WHERE trace_id = ?1
ORDER BY start_unix_nanos;
The slow-span query is:
SELECT name, end_unix_nanos - start_unix_nanos AS duration_nanos
FROM spans
ORDER BY duration_nanos DESC
LIMIT 10;
The full-text query is:
SELECT name, attributes
FROM spans_fts
WHERE spans_fts MATCH ?1;
One SQLite writer avoids write contention. WAL lets readers work while the writer commits. For retention, periodically delete old rows, then rebuild or update the FTS table in the same maintenance transaction.
Pitfalls
- Do not query before provider shutdown; the batch processor may still hold spans.
- Do not update the FTS table in a separate transaction from the span insert.
- Do not use multiple concurrent SQLite writers for this simple store.
- Store timestamps as Unix nanoseconds consistently.
- Escape attribute text before storing it as JSON text.
- The batch exporter is version-specific to
opentelemetry_sdk0.33.1.