From c778523910d489c1c1c1e34ca552c174f52e17a2 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Thu, 4 Jun 2026 14:34:05 -0400 Subject: [PATCH 1/2] telemetry: Reduce cardinality of some otel span names I tried to follow OpenTelemetry naming guidance, e.g. https://opentelemetry.io/docs/specs/semconv/db/database-spans/#name --- Cargo.lock | 1 + crates/factor-outbound-mqtt/src/host.rs | 18 +- crates/factor-outbound-mysql/src/host.rs | 9 +- crates/factor-outbound-pg/src/host.rs | 36 ++-- crates/factor-outbound-redis/src/host.rs | 60 ++++--- crates/factor-sqlite/Cargo.toml | 1 + crates/factor-sqlite/src/host.rs | 20 ++- crates/telemetry/src/db.rs | 207 +++++++++++++++++++++++ crates/telemetry/src/lib.rs | 1 + 9 files changed, 310 insertions(+), 43 deletions(-) create mode 100644 crates/telemetry/src/db.rs diff --git a/Cargo.lock b/Cargo.lock index b94bde3868..a1cf2a5136 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9231,6 +9231,7 @@ dependencies = [ "spin-factors-test", "spin-locked-app", "spin-resource-table", + "spin-telemetry", "spin-wasi-async", "spin-world", "tokio", diff --git a/crates/factor-outbound-mqtt/src/host.rs b/crates/factor-outbound-mqtt/src/host.rs index 5a090f3772..068afd9912 100644 --- a/crates/factor-outbound-mqtt/src/host.rs +++ b/crates/factor-outbound-mqtt/src/host.rs @@ -163,8 +163,13 @@ impl v3::HostConnectionWithStore for crate::MqttFactorData { } #[instrument(name = "spin_outbound_mqtt.publish", skip(accessor, connection, payload), err(level = Level::INFO), - fields(otel.kind = "producer", otel.name = format!("{} publish", topic), messaging.operation = "publish", - messaging.system = "mqtt"))] + fields( + otel.kind = "producer", + otel.name = "publish", + messaging.operation = "publish", + messaging.system = "mqtt", + messaging.destination.name = topic, + ))] async fn publish( accessor: &Accessor, connection: Resource, @@ -236,8 +241,13 @@ impl v2::HostConnection for InstanceState { /// current trace context into the payload yourself. /// https://w3c.github.io/trace-context-mqtt/#mqtt-v3-recommendation. #[instrument(name = "spin_outbound_mqtt.publish", skip(self, connection, payload), err(level = Level::INFO), - fields(otel.kind = "producer", otel.name = format!("{} publish", topic), messaging.operation = "publish", - messaging.system = "mqtt"))] + fields( + otel.kind = "producer", + otel.name = "publish", + messaging.operation = "publish", + messaging.system = "mqtt", + messaging.destination.name = topic, + ))] async fn publish( &mut self, connection: Resource, diff --git a/crates/factor-outbound-mysql/src/host.rs b/crates/factor-outbound-mysql/src/host.rs index 0e3fdf512c..e2d6ee5f55 100644 --- a/crates/factor-outbound-mysql/src/host.rs +++ b/crates/factor-outbound-mysql/src/host.rs @@ -178,7 +178,8 @@ impl v3::HostConnectionWithStore for MysqlFactorData { impl v2::Host for InstanceState {} impl v2::HostConnection for InstanceState { - #[instrument(name = "spin_outbound_mysql.open", skip(self, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] + #[instrument(name = "spin_outbound_mysql.open", skip(self, address), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] async fn open(&mut self, address: String) -> Result, v2::Error> { let permit = self .semaphore @@ -193,7 +194,8 @@ impl v2::HostConnection for InstanceState { .map(Resource::new_own) } - #[instrument(name = "spin_outbound_mysql.execute", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = statement))] + #[instrument(name = "spin_outbound_mysql.execute", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn execute( &mut self, connection: Resource, @@ -211,7 +213,8 @@ impl v2::HostConnection for InstanceState { .map_err(track_db_error_on_span) } - #[instrument(name = "spin_outbound_mysql.query", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = statement))] + #[instrument(name = "spin_outbound_mysql.query", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn query( &mut self, connection: Resource, diff --git a/crates/factor-outbound-pg/src/host.rs b/crates/factor-outbound-pg/src/host.rs index 34b66249f0..af762cb355 100644 --- a/crates/factor-outbound-pg/src/host.rs +++ b/crates/factor-outbound-pg/src/host.rs @@ -92,7 +92,8 @@ fn v3_params_to_v4(params: Vec) -> Vec { } impl v3::HostConnection for InstanceState { - #[instrument(name = "spin_outbound_pg.open", skip(self, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] + #[instrument(name = "spin_outbound_pg.open", skip(self, address), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] async fn open(&mut self, address: String) -> Result, v3::Error> { spin_factor_outbound_networking::record_address_fields(&address); @@ -106,7 +107,8 @@ impl v3::HostConnection for InstanceState { .map_err(v3::Error::from) } - #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn execute( &mut self, connection: Resource, @@ -122,7 +124,8 @@ impl v3::HostConnection for InstanceState { .map_err(track_db_error_on_span_v3) } - #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn query( &mut self, connection: Resource, @@ -199,7 +202,8 @@ impl v4::HostConnectionBuilder for InstanceState { } impl v4::HostConnection for InstanceState { - #[instrument(name = "spin_outbound_pg.open", skip(self, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] + #[instrument(name = "spin_outbound_pg.open", skip(self, address), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] async fn open(&mut self, address: String) -> Result, v4::Error> { spin_factor_outbound_networking::record_address_fields(&address); @@ -210,7 +214,8 @@ impl v4::HostConnection for InstanceState { self.open_connection(&address, None).await } - #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn execute( &mut self, connection: Resource, @@ -224,7 +229,8 @@ impl v4::HostConnection for InstanceState { .map_err(track_db_error_on_span_v4) } - #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn query( &mut self, connection: Resource, @@ -247,7 +253,8 @@ impl v4::HostConnection for InstanceState { impl spin_world::spin::postgres4_2_0::postgres::HostConnectionWithStore for crate::PgFactorData { - #[instrument(name = "spin_outbound_pg.open_async", skip(accessor, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] + #[instrument(name = "spin_outbound_pg.open_async", skip(accessor, address), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] async fn open_async( accessor: &Accessor, address: String, @@ -260,7 +267,8 @@ impl spin_world::spin::postgres4_2_0::postgres::HostConnectio Self::open_connection_async(accessor, &address, None).await } - #[instrument(name = "spin_outbound_pg.execute", skip(accessor, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.execute", skip(accessor, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn execute_async( accessor: &Accessor, connection: Resource, @@ -282,7 +290,8 @@ impl spin_world::spin::postgres4_2_0::postgres::HostConnectio } #[allow(clippy::type_complexity)] // blame bindgen, clippy, blame bindgen - #[instrument(name = "spin_outbound_pg.query_async", skip(accessor, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.query_async", skip(accessor, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn query_async( accessor: &Accessor, connection: Resource, @@ -472,7 +481,8 @@ macro_rules! delegate { impl v2::Host for InstanceState {} impl v2::HostConnection for InstanceState { - #[instrument(name = "spin_outbound_pg.open", skip(self, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] + #[instrument(name = "spin_outbound_pg.open", skip(self, address), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] async fn open(&mut self, address: String) -> Result, v2::Error> { self.otel.reparent_tracing_span(); spin_factor_outbound_networking::record_address_fields(&address); @@ -486,7 +496,8 @@ impl v2::HostConnection for InstanceState { .map_err(v2::Error::from) } - #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn execute( &mut self, connection: Resource, @@ -506,7 +517,8 @@ impl v2::HostConnection for InstanceState { .map_err(track_db_error_on_span_v2) } - #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = statement))] + #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] async fn query( &mut self, connection: Resource, diff --git a/crates/factor-outbound-redis/src/host.rs b/crates/factor-outbound-redis/src/host.rs index f349cf1992..6414a8e922 100644 --- a/crates/factor-outbound-redis/src/host.rs +++ b/crates/factor-outbound-redis/src/host.rs @@ -236,7 +236,8 @@ impl crate::RedisFactorData { } impl v3::HostConnectionWithStore for crate::RedisFactorData { - #[instrument(name = "spin_outbound_redis.open_connection", skip(accessor, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] + #[instrument(name = "spin_outbound_redis.open_connection", skip(accessor, address), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] async fn open( accessor: &Accessor, address: String, @@ -281,7 +282,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { }) } - #[instrument(name = "spin_outbound_redis.publish", skip(accessor, connection, payload), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("PUBLISH {}", channel)))] + #[instrument(name = "spin_outbound_redis.publish", skip(accessor, connection, payload), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "PUBLISH"))] async fn publish( accessor: &Accessor, connection: Resource, @@ -292,7 +294,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::publish(&mut conn, channel, payload).await } - #[instrument(name = "spin_outbound_redis.get", skip(accessor, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("GET {}", key)))] + #[instrument(name = "spin_outbound_redis.get", skip(accessor, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "GET"))] async fn get( accessor: &Accessor, connection: Resource, @@ -302,7 +305,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::get(&mut conn, key).await } - #[instrument(name = "spin_outbound_redis.set", skip(accessor, connection, value), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SET {}", key)))] + #[instrument(name = "spin_outbound_redis.set", skip(accessor, connection, value), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SET"))] async fn set( accessor: &Accessor, connection: Resource, @@ -313,7 +317,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::set(&mut conn, key, value).await } - #[instrument(name = "spin_outbound_redis.incr", skip(accessor, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("INCRBY {} 1", key)))] + #[instrument(name = "spin_outbound_redis.incr", skip(accessor, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "INCRBY"))] async fn incr( accessor: &Accessor, connection: Resource, @@ -323,7 +328,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::incr(&mut conn, key).await } - #[instrument(name = "spin_outbound_redis.del", skip(accessor, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("DEL {}", keys.join(" "))))] + #[instrument(name = "spin_outbound_redis.del", skip(accessor, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "DEL"))] async fn del( accessor: &Accessor, connection: Resource, @@ -333,7 +339,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::del(&mut conn, keys).await } - #[instrument(name = "spin_outbound_redis.sadd", skip(accessor, connection, values), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SADD {} {}", key, values.join(" "))))] + #[instrument(name = "spin_outbound_redis.sadd", skip(accessor, connection, values), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SADD"))] async fn sadd( accessor: &Accessor, connection: Resource, @@ -344,7 +351,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::sadd(&mut conn, key, values).await } - #[instrument(name = "spin_outbound_redis.smembers", skip(accessor, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SMEMBERS {}", key)))] + #[instrument(name = "spin_outbound_redis.smembers", skip(accessor, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SMEMBERS"))] async fn smembers( accessor: &Accessor, connection: Resource, @@ -354,7 +362,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::smembers(&mut conn, key).await } - #[instrument(name = "spin_outbound_redis.srem", skip(accessor, connection, values), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SREM {} {}", key, values.join(" "))))] + #[instrument(name = "spin_outbound_redis.srem", skip(accessor, connection, values), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SREM"))] async fn srem( accessor: &Accessor, connection: Resource, @@ -365,7 +374,8 @@ impl v3::HostConnectionWithStore for crate::RedisFactorData { operations::srem(&mut conn, key, values).await } - #[instrument(name = "spin_outbound_redis.execute", skip(accessor, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("{}", command)))] + #[instrument(name = "spin_outbound_redis.execute", skip(accessor, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("{}", command)))] async fn execute( accessor: &Accessor, connection: Resource, @@ -386,7 +396,8 @@ impl v2::Host for crate::InstanceState { } impl v2::HostConnection for crate::InstanceState { - #[instrument(name = "spin_outbound_redis.open_connection", skip(self, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] + #[instrument(name = "spin_outbound_redis.open_connection", skip(self, address), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))] async fn open(&mut self, address: String) -> Result, v2::Error> { self.otel.reparent_tracing_span(); if !self @@ -400,7 +411,8 @@ impl v2::HostConnection for crate::InstanceState { self.establish_connection(address).await } - #[instrument(name = "spin_outbound_redis.publish", skip(self, connection, payload), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("PUBLISH {}", channel)))] + #[instrument(name = "spin_outbound_redis.publish", skip(self, connection, payload), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "PUBLISH"))] async fn publish( &mut self, connection: Resource, @@ -414,7 +426,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::publish(conn, channel, payload).await?) } - #[instrument(name = "spin_outbound_redis.get", skip(self, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("GET {}", key)))] + #[instrument(name = "spin_outbound_redis.get", skip(self, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "GET"))] async fn get( &mut self, connection: Resource, @@ -427,7 +440,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::get(conn, key).await?) } - #[instrument(name = "spin_outbound_redis.set", skip(self, connection, value), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SET {}", key)))] + #[instrument(name = "spin_outbound_redis.set", skip(self, connection, value), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SET"))] async fn set( &mut self, connection: Resource, @@ -440,7 +454,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::set(conn, key, value).await?) } - #[instrument(name = "spin_outbound_redis.incr", skip(self, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("INCRBY {} 1", key)))] + #[instrument(name = "spin_outbound_redis.incr", skip(self, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "INCRBY"))] async fn incr( &mut self, connection: Resource, @@ -452,7 +467,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::incr(conn, key).await?) } - #[instrument(name = "spin_outbound_redis.del", skip(self, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("DEL {}", keys.join(" "))))] + #[instrument(name = "spin_outbound_redis.del", skip(self, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "DEL"))] async fn del( &mut self, connection: Resource, @@ -464,7 +480,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::del(conn, keys).await?) } - #[instrument(name = "spin_outbound_redis.sadd", skip(self, connection, values), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SADD {} {}", key, values.join(" "))))] + #[instrument(name = "spin_outbound_redis.sadd", skip(self, connection, values), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SADD"))] async fn sadd( &mut self, connection: Resource, @@ -477,7 +494,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::sadd(conn, key, values).await?) } - #[instrument(name = "spin_outbound_redis.smembers", skip(self, connection), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SMEMBERS {}", key)))] + #[instrument(name = "spin_outbound_redis.smembers", skip(self, connection), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SMEMBERS"))] async fn smembers( &mut self, connection: Resource, @@ -489,7 +507,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::smembers(conn, key).await?) } - #[instrument(name = "spin_outbound_redis.srem", skip(self, connection, values), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("SREM {} {}", key, values.join(" "))))] + #[instrument(name = "spin_outbound_redis.srem", skip(self, connection, values), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SREM"))] async fn srem( &mut self, connection: Resource, @@ -502,7 +521,8 @@ impl v2::HostConnection for crate::InstanceState { Ok(operations::srem(conn, key, values).await?) } - #[instrument(name = "spin_outbound_redis.execute", skip(self, connection, arguments), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("{}", command)))] + #[instrument(name = "spin_outbound_redis.execute", skip(self, connection, arguments), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = command))] async fn execute( &mut self, connection: Resource, diff --git a/crates/factor-sqlite/Cargo.toml b/crates/factor-sqlite/Cargo.toml index fa98e585ba..4774898d56 100644 --- a/crates/factor-sqlite/Cargo.toml +++ b/crates/factor-sqlite/Cargo.toml @@ -16,6 +16,7 @@ spin-factor-otel = { path = "../factor-otel" } spin-factors = { path = "../factors" } spin-locked-app = { path = "../locked-app" } spin-resource-table = { path = "../table" } +spin-telemetry = { path = "../telemetry" } spin-wasi-async = { path = "../wasi-async" } spin-world = { path = "../world" } tokio = { workspace = true, features = ["sync"] } diff --git a/crates/factor-sqlite/src/host.rs b/crates/factor-sqlite/src/host.rs index 3833984942..3546a0512b 100644 --- a/crates/factor-sqlite/src/host.rs +++ b/crates/factor-sqlite/src/host.rs @@ -102,12 +102,17 @@ impl v3::Host for InstanceState { } impl v3::HostConnection for InstanceState { - #[instrument(name = "spin_sqlite.open", skip(self), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", sqlite.backend = Empty))] + #[instrument(name = "spin_sqlite.open", skip(self), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", sqlite.backend = Empty))] async fn open(&mut self, database: String) -> Result, v3::Error> { self.open_impl(database).await } - #[instrument(name = "spin_sqlite.execute", skip(self, connection, parameters), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", otel.name = query, sqlite.backend = Empty))] + #[instrument(name = "spin_sqlite.execute", skip(self, connection, parameters), err(level = Level::INFO), + fields( + otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", + otel.name = spin_telemetry::db::sql_span_name(&query), sqlite.backend = Empty, + ))] async fn execute( &mut self, connection: Resource, @@ -275,13 +280,20 @@ impl v2::Host for InstanceState { } impl v2::HostConnection for InstanceState { - #[instrument(name = "spin_sqlite.open", skip(self), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", sqlite.backend = Empty))] + #[instrument(name = "spin_sqlite.open", skip(self), err(level = Level::INFO), + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", sqlite.backend = Empty))] async fn open(&mut self, database: String) -> Result, v2::Error> { self.otel.reparent_tracing_span(); self.open_impl(database).await.map_err(to_v2_error) } - #[instrument(name = "spin_sqlite.execute", skip(self, connection, parameters), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", otel.name = query, sqlite.backend = Empty))] + #[instrument(name = "spin_sqlite.execute", skip(self, connection, parameters), err(level = Level::INFO), + fields( + otel.kind = "client", + {otel_attribute::DB_SYSTEM_NAME} = "sqlite", + otel.name = spin_telemetry::db::sql_span_name(&query), + sqlite.backend = Empty, + ))] async fn execute( &mut self, connection: Resource, diff --git a/crates/telemetry/src/db.rs b/crates/telemetry/src/db.rs new file mode 100644 index 0000000000..dce99313e7 --- /dev/null +++ b/crates/telemetry/src/db.rs @@ -0,0 +1,207 @@ +//! Helpers for constructing low-cardinality OpenTelemetry span names for +//! database operations. + +/// Constructs a low-cardinality span name from an arbitrary SQL statement, +/// following the OpenTelemetry database span name conventions. +/// +/// The full text of a SQL statement has very high cardinality (it embeds +/// literals, parameter values, etc.) and so makes a poor span name. Instead the +/// conventions recommend a `{operation} {target}` summary, e.g. `SELECT users`, +/// falling back to just `{operation}` when no single target can be determined. +/// +/// See the OpenTelemetry guidance on [database span names] and [generating a +/// query summary]. This implementation is a lightweight, dependency-free +/// approximation modeled on the [opentelemetry-go-instrumentation SQL probe]. +/// +/// If no operation can be extracted (e.g. the statement is empty) this returns +/// `"SQL"` so that the span always has a stable, low-cardinality name. +/// +/// [database span names]: https://opentelemetry.io/docs/specs/semconv/db/database-spans/#name +/// [generating a query summary]: https://opentelemetry.io/docs/specs/semconv/db/database-spans/#generating-a-summary-of-the-query +/// [opentelemetry-go-instrumentation SQL probe]: https://github.com/open-telemetry/opentelemetry-go-instrumentation/blob/main/internal/pkg/instrumentation/bpf/database/sql/probe.go +pub fn sql_span_name(statement: &str) -> String { + let mut tokens = SqlTokens::new(statement); + + let Some(operation) = tokens.next() else { + return "SQL".to_string(); + }; + // SQL keywords are case-insensitive; normalize the operation to reduce + // cardinality (e.g. `select` and `SELECT` should map to the same name). + let operation = operation.to_ascii_uppercase(); + + let target = match operation.as_str() { + "SELECT" | "DELETE" => tokens.target_after("FROM"), + "INSERT" | "REPLACE" => tokens.target_after("INTO"), + "UPDATE" => tokens.next_target(), + // For other statements (DDL, transaction control, etc.) the operation + // alone is a sufficiently descriptive, low-cardinality summary. + _ => None, + }; + + match target { + Some(target) => format!("{operation} {target}"), + None => operation, + } +} + +/// A minimal tokenizer over a SQL statement. +/// +/// It yields whitespace-delimited "words", treating `(`, `)`, `,` and `;` as +/// their own single-character tokens so that constructs like `FROM(SELECT ...` +/// and `FROM a, b` are split correctly. +struct SqlTokens<'a> { + rest: &'a str, +} + +impl<'a> SqlTokens<'a> { + fn new(statement: &'a str) -> Self { + Self { rest: statement } + } + + /// Returns the next token following the first occurrence of `keyword` + /// (matched case-insensitively), cleaned up for use as a span target. + /// + /// Returns `None` if the keyword is not found, or if the entity following it + /// is an anonymous/derived table (e.g. a subquery in parentheses), in which + /// case the OpenTelemetry conventions recommend omitting the target. + fn target_after(&mut self, keyword: &str) -> Option { + while let Some(token) = self.next() { + if token.eq_ignore_ascii_case(keyword) { + return self.next_target(); + } + } + None + } + + /// Returns the next token cleaned up for use as a span target, or `None` if + /// there is no suitable target (e.g. a subquery or an empty token). + fn next_target(&mut self) -> Option { + let token = self.next()?; + // A `(` indicates an anonymous/derived table; omit the target. + if token == "(" { + return None; + } + let cleaned = token.trim_matches(|c| matches!(c, '"' | '`' | '\'' | '[' | ']')); + if cleaned.is_empty() { + None + } else { + Some(cleaned.to_string()) + } + } +} + +impl<'a> Iterator for SqlTokens<'a> { + type Item = &'a str; + + fn next(&mut self) -> Option<&'a str> { + let rest = std::mem::take(&mut self.rest).trim_start(); + if rest.is_empty() { + return None; + } + // Single ascii char tokens + let token_end = if rest.starts_with(['(', ')', ',', ';']) { + Some(1) + } else { + rest.find(|c: char| c.is_whitespace() || matches!(c, '(' | ')' | ',' | ';')) + } + .unwrap_or(rest.len()); + + let (token, rest) = rest.split_at(token_end); + self.rest = rest; + Some(token) + } +} + +#[cfg(test)] +mod tests { + use super::sql_span_name; + + #[test] + fn select_with_table() { + assert_eq!( + sql_span_name("SELECT * FROM users WHERE id = ?"), + "SELECT users" + ); + } + + #[test] + fn select_is_case_insensitive() { + assert_eq!(sql_span_name("select id from Users"), "SELECT Users"); + } + + #[test] + fn select_multiline() { + let query = "SELECT *\n FROM wuser_table\n WHERE username = ?"; + assert_eq!(sql_span_name(query), "SELECT wuser_table"); + } + + #[test] + fn select_multiple_tables_uses_first() { + assert_eq!( + sql_span_name("SELECT * FROM songs, artists"), + "SELECT songs" + ); + } + + #[test] + fn select_anonymous_table_has_no_target() { + assert_eq!( + sql_span_name("SELECT * FROM (SELECT * FROM orders) t"), + "SELECT" + ); + } + + #[test] + fn insert_into_table() { + assert_eq!( + sql_span_name("INSERT INTO shipping_details (order_id) VALUES (?)"), + "INSERT shipping_details" + ); + } + + #[test] + fn update_table() { + assert_eq!( + sql_span_name("UPDATE users SET name = ? WHERE id = ?"), + "UPDATE users" + ); + } + + #[test] + fn delete_from_table() { + assert_eq!( + sql_span_name("DELETE FROM users WHERE id = ?"), + "DELETE users" + ); + } + + #[test] + fn quoted_identifiers_are_cleaned() { + assert_eq!(sql_span_name("SELECT * FROM `my_table`"), "SELECT my_table"); + assert_eq!( + sql_span_name("SELECT * FROM \"my_table\""), + "SELECT my_table" + ); + } + + #[test] + fn schema_qualified_target_preserved() { + assert_eq!( + sql_span_name("SELECT * FROM public.users"), + "SELECT public.users" + ); + } + + #[test] + fn ddl_uses_operation_only() { + assert_eq!( + sql_span_name("CREATE TABLE users (id INTEGER PRIMARY KEY)"), + "CREATE" + ); + } + + #[test] + fn empty_statement_falls_back() { + assert_eq!(sql_span_name(" "), "SQL"); + } +} diff --git a/crates/telemetry/src/lib.rs b/crates/telemetry/src/lib.rs index 8eaacbc9b6..a9f9e690d5 100644 --- a/crates/telemetry/src/lib.rs +++ b/crates/telemetry/src/lib.rs @@ -8,6 +8,7 @@ use opentelemetry_sdk::propagation::TraceContextPropagator; use tracing_subscriber::{EnvFilter, Layer, fmt, prelude::*, registry}; mod alert_in_dev; +pub mod db; pub mod detector; pub mod env; pub mod logs; From 3a36b3b74d3a0638b1c386f76ee9ca3b2f5add60 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Tue, 9 Jun 2026 14:47:38 -0400 Subject: [PATCH 2/2] Remove some otel.name overrides --- crates/factor-outbound-mqtt/src/host.rs | 2 - crates/factor-outbound-mysql/src/host.rs | 4 +- crates/factor-outbound-pg/src/host.rs | 16 +- crates/factor-sqlite/src/host.rs | 12 +- crates/telemetry/src/db.rs | 207 ----------------------- crates/telemetry/src/lib.rs | 1 - 6 files changed, 12 insertions(+), 230 deletions(-) delete mode 100644 crates/telemetry/src/db.rs diff --git a/crates/factor-outbound-mqtt/src/host.rs b/crates/factor-outbound-mqtt/src/host.rs index 068afd9912..63d2932a90 100644 --- a/crates/factor-outbound-mqtt/src/host.rs +++ b/crates/factor-outbound-mqtt/src/host.rs @@ -165,7 +165,6 @@ impl v3::HostConnectionWithStore for crate::MqttFactorData { #[instrument(name = "spin_outbound_mqtt.publish", skip(accessor, connection, payload), err(level = Level::INFO), fields( otel.kind = "producer", - otel.name = "publish", messaging.operation = "publish", messaging.system = "mqtt", messaging.destination.name = topic, @@ -243,7 +242,6 @@ impl v2::HostConnection for InstanceState { #[instrument(name = "spin_outbound_mqtt.publish", skip(self, connection, payload), err(level = Level::INFO), fields( otel.kind = "producer", - otel.name = "publish", messaging.operation = "publish", messaging.system = "mqtt", messaging.destination.name = topic, diff --git a/crates/factor-outbound-mysql/src/host.rs b/crates/factor-outbound-mysql/src/host.rs index e2d6ee5f55..79d94186bb 100644 --- a/crates/factor-outbound-mysql/src/host.rs +++ b/crates/factor-outbound-mysql/src/host.rs @@ -195,7 +195,7 @@ impl v2::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_mysql.execute", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql"))] async fn execute( &mut self, connection: Resource, @@ -214,7 +214,7 @@ impl v2::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_mysql.query", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql"))] async fn query( &mut self, connection: Resource, diff --git a/crates/factor-outbound-pg/src/host.rs b/crates/factor-outbound-pg/src/host.rs index af762cb355..09f1951f80 100644 --- a/crates/factor-outbound-pg/src/host.rs +++ b/crates/factor-outbound-pg/src/host.rs @@ -108,7 +108,7 @@ impl v3::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn execute( &mut self, connection: Resource, @@ -125,7 +125,7 @@ impl v3::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn query( &mut self, connection: Resource, @@ -215,7 +215,7 @@ impl v4::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn execute( &mut self, connection: Resource, @@ -230,7 +230,7 @@ impl v4::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn query( &mut self, connection: Resource, @@ -268,7 +268,7 @@ impl spin_world::spin::postgres4_2_0::postgres::HostConnectio } #[instrument(name = "spin_outbound_pg.execute", skip(accessor, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn execute_async( accessor: &Accessor, connection: Resource, @@ -291,7 +291,7 @@ impl spin_world::spin::postgres4_2_0::postgres::HostConnectio #[allow(clippy::type_complexity)] // blame bindgen, clippy, blame bindgen #[instrument(name = "spin_outbound_pg.query_async", skip(accessor, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn query_async( accessor: &Accessor, connection: Resource, @@ -497,7 +497,7 @@ impl v2::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_pg.execute", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn execute( &mut self, connection: Resource, @@ -518,7 +518,7 @@ impl v2::HostConnection for InstanceState { } #[instrument(name = "spin_outbound_pg.query", skip(self, connection, params), err(level = Level::INFO), - fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql", otel.name = spin_telemetry::db::sql_span_name(&statement)))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "postgresql"))] async fn query( &mut self, connection: Resource, diff --git a/crates/factor-sqlite/src/host.rs b/crates/factor-sqlite/src/host.rs index 3546a0512b..e4e321c24d 100644 --- a/crates/factor-sqlite/src/host.rs +++ b/crates/factor-sqlite/src/host.rs @@ -109,10 +109,7 @@ impl v3::HostConnection for InstanceState { } #[instrument(name = "spin_sqlite.execute", skip(self, connection, parameters), err(level = Level::INFO), - fields( - otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", - otel.name = spin_telemetry::db::sql_span_name(&query), sqlite.backend = Empty, - ))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", sqlite.backend = Empty))] async fn execute( &mut self, connection: Resource, @@ -288,12 +285,7 @@ impl v2::HostConnection for InstanceState { } #[instrument(name = "spin_sqlite.execute", skip(self, connection, parameters), err(level = Level::INFO), - fields( - otel.kind = "client", - {otel_attribute::DB_SYSTEM_NAME} = "sqlite", - otel.name = spin_telemetry::db::sql_span_name(&query), - sqlite.backend = Empty, - ))] + fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "sqlite", sqlite.backend = Empty))] async fn execute( &mut self, connection: Resource, diff --git a/crates/telemetry/src/db.rs b/crates/telemetry/src/db.rs deleted file mode 100644 index dce99313e7..0000000000 --- a/crates/telemetry/src/db.rs +++ /dev/null @@ -1,207 +0,0 @@ -//! Helpers for constructing low-cardinality OpenTelemetry span names for -//! database operations. - -/// Constructs a low-cardinality span name from an arbitrary SQL statement, -/// following the OpenTelemetry database span name conventions. -/// -/// The full text of a SQL statement has very high cardinality (it embeds -/// literals, parameter values, etc.) and so makes a poor span name. Instead the -/// conventions recommend a `{operation} {target}` summary, e.g. `SELECT users`, -/// falling back to just `{operation}` when no single target can be determined. -/// -/// See the OpenTelemetry guidance on [database span names] and [generating a -/// query summary]. This implementation is a lightweight, dependency-free -/// approximation modeled on the [opentelemetry-go-instrumentation SQL probe]. -/// -/// If no operation can be extracted (e.g. the statement is empty) this returns -/// `"SQL"` so that the span always has a stable, low-cardinality name. -/// -/// [database span names]: https://opentelemetry.io/docs/specs/semconv/db/database-spans/#name -/// [generating a query summary]: https://opentelemetry.io/docs/specs/semconv/db/database-spans/#generating-a-summary-of-the-query -/// [opentelemetry-go-instrumentation SQL probe]: https://github.com/open-telemetry/opentelemetry-go-instrumentation/blob/main/internal/pkg/instrumentation/bpf/database/sql/probe.go -pub fn sql_span_name(statement: &str) -> String { - let mut tokens = SqlTokens::new(statement); - - let Some(operation) = tokens.next() else { - return "SQL".to_string(); - }; - // SQL keywords are case-insensitive; normalize the operation to reduce - // cardinality (e.g. `select` and `SELECT` should map to the same name). - let operation = operation.to_ascii_uppercase(); - - let target = match operation.as_str() { - "SELECT" | "DELETE" => tokens.target_after("FROM"), - "INSERT" | "REPLACE" => tokens.target_after("INTO"), - "UPDATE" => tokens.next_target(), - // For other statements (DDL, transaction control, etc.) the operation - // alone is a sufficiently descriptive, low-cardinality summary. - _ => None, - }; - - match target { - Some(target) => format!("{operation} {target}"), - None => operation, - } -} - -/// A minimal tokenizer over a SQL statement. -/// -/// It yields whitespace-delimited "words", treating `(`, `)`, `,` and `;` as -/// their own single-character tokens so that constructs like `FROM(SELECT ...` -/// and `FROM a, b` are split correctly. -struct SqlTokens<'a> { - rest: &'a str, -} - -impl<'a> SqlTokens<'a> { - fn new(statement: &'a str) -> Self { - Self { rest: statement } - } - - /// Returns the next token following the first occurrence of `keyword` - /// (matched case-insensitively), cleaned up for use as a span target. - /// - /// Returns `None` if the keyword is not found, or if the entity following it - /// is an anonymous/derived table (e.g. a subquery in parentheses), in which - /// case the OpenTelemetry conventions recommend omitting the target. - fn target_after(&mut self, keyword: &str) -> Option { - while let Some(token) = self.next() { - if token.eq_ignore_ascii_case(keyword) { - return self.next_target(); - } - } - None - } - - /// Returns the next token cleaned up for use as a span target, or `None` if - /// there is no suitable target (e.g. a subquery or an empty token). - fn next_target(&mut self) -> Option { - let token = self.next()?; - // A `(` indicates an anonymous/derived table; omit the target. - if token == "(" { - return None; - } - let cleaned = token.trim_matches(|c| matches!(c, '"' | '`' | '\'' | '[' | ']')); - if cleaned.is_empty() { - None - } else { - Some(cleaned.to_string()) - } - } -} - -impl<'a> Iterator for SqlTokens<'a> { - type Item = &'a str; - - fn next(&mut self) -> Option<&'a str> { - let rest = std::mem::take(&mut self.rest).trim_start(); - if rest.is_empty() { - return None; - } - // Single ascii char tokens - let token_end = if rest.starts_with(['(', ')', ',', ';']) { - Some(1) - } else { - rest.find(|c: char| c.is_whitespace() || matches!(c, '(' | ')' | ',' | ';')) - } - .unwrap_or(rest.len()); - - let (token, rest) = rest.split_at(token_end); - self.rest = rest; - Some(token) - } -} - -#[cfg(test)] -mod tests { - use super::sql_span_name; - - #[test] - fn select_with_table() { - assert_eq!( - sql_span_name("SELECT * FROM users WHERE id = ?"), - "SELECT users" - ); - } - - #[test] - fn select_is_case_insensitive() { - assert_eq!(sql_span_name("select id from Users"), "SELECT Users"); - } - - #[test] - fn select_multiline() { - let query = "SELECT *\n FROM wuser_table\n WHERE username = ?"; - assert_eq!(sql_span_name(query), "SELECT wuser_table"); - } - - #[test] - fn select_multiple_tables_uses_first() { - assert_eq!( - sql_span_name("SELECT * FROM songs, artists"), - "SELECT songs" - ); - } - - #[test] - fn select_anonymous_table_has_no_target() { - assert_eq!( - sql_span_name("SELECT * FROM (SELECT * FROM orders) t"), - "SELECT" - ); - } - - #[test] - fn insert_into_table() { - assert_eq!( - sql_span_name("INSERT INTO shipping_details (order_id) VALUES (?)"), - "INSERT shipping_details" - ); - } - - #[test] - fn update_table() { - assert_eq!( - sql_span_name("UPDATE users SET name = ? WHERE id = ?"), - "UPDATE users" - ); - } - - #[test] - fn delete_from_table() { - assert_eq!( - sql_span_name("DELETE FROM users WHERE id = ?"), - "DELETE users" - ); - } - - #[test] - fn quoted_identifiers_are_cleaned() { - assert_eq!(sql_span_name("SELECT * FROM `my_table`"), "SELECT my_table"); - assert_eq!( - sql_span_name("SELECT * FROM \"my_table\""), - "SELECT my_table" - ); - } - - #[test] - fn schema_qualified_target_preserved() { - assert_eq!( - sql_span_name("SELECT * FROM public.users"), - "SELECT public.users" - ); - } - - #[test] - fn ddl_uses_operation_only() { - assert_eq!( - sql_span_name("CREATE TABLE users (id INTEGER PRIMARY KEY)"), - "CREATE" - ); - } - - #[test] - fn empty_statement_falls_back() { - assert_eq!(sql_span_name(" "), "SQL"); - } -} diff --git a/crates/telemetry/src/lib.rs b/crates/telemetry/src/lib.rs index a9f9e690d5..8eaacbc9b6 100644 --- a/crates/telemetry/src/lib.rs +++ b/crates/telemetry/src/lib.rs @@ -8,7 +8,6 @@ use opentelemetry_sdk::propagation::TraceContextPropagator; use tracing_subscriber::{EnvFilter, Layer, fmt, prelude::*, registry}; mod alert_in_dev; -pub mod db; pub mod detector; pub mod env; pub mod logs;