tracing::instrument a fair bit of the repo

sqlx-core/Cargo.toml

@@ -88,6 +88,7 @@ futures-util = { version = "0.3.32", default-features = false, features = ["allo
 log = { version = "0.4.18", default-features = false }
 memchr = { version = "2.5.0", default-features = false }
 percent-encoding = "2.3.0"
+pin-project-lite = "0.2.16"
 serde = { version = "1.0.219", features = ["derive", "rc"], optional = true }
 serde_json = { version = "1.0.142", features = ["raw_value"], optional = true }
 toml = { version = "0.8.16", optional = true }

sqlx-core/src/instrument_stream.rs

@@ -0,0 +1,52 @@
+//! Attach a [`tracing::Span`] to a [`Stream`] so the span stays open for the
+//! whole stream rather than just its construction.
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures_core::Stream;
+use pin_project_lite::pin_project;
+use tracing::Span;
+
+pin_project! {
+    /// A [`Stream`] adapter that enters `span` for the duration of every
+    /// [`poll_next`](Stream::poll_next).
+    ///
+    /// A plain `#[tracing::instrument]` on an `async fn` that *returns* a stream
+    /// only keeps its span entered while the stream is being built; the span is
+    /// then closed before the caller ever polls the stream. Wrapping the
+    /// returned stream with this adapter instead keeps the span open across row
+    /// fetching, so e.g. a `sqlx::query` span measures the whole query rather
+    /// than just its setup.
+    pub struct InstrumentedStream<S> {
+        #[pin]
+        stream: S,
+        span: Span,
+    }
+}
+
+impl<S: Stream> Stream for InstrumentedStream<S> {
+    type Item = S::Item;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
+        let _entered = this.span.enter();
+        this.stream.poll_next(cx)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
+}
+
+/// Extension trait for attaching a [`Span`] to a [`Stream`].
+pub trait InstrumentStream: Stream + Sized {
+    /// Wrap this stream so that `span` is entered every time it is polled.
+    ///
+    /// See [`InstrumentedStream`].
+    fn instrument_stream(self, span: Span) -> InstrumentedStream<Self> {
+        InstrumentedStream { stream: self, span }
+    }
+}
+
+impl<S: Stream> InstrumentStream for S {}

sqlx-core/src/lib.rs

@@ -66,6 +66,7 @@ pub mod describe;
 pub mod executor;
 pub mod from_row;
 pub mod fs;
+pub mod instrument_stream;
 pub mod io;
 pub mod logger;
 pub mod net;

sqlx-core/src/migrate/migrator.rs

@@ -224,6 +224,17 @@ impl Migrator {
 
     // Getting around the annoying "implementation of `Acquire` is not general enough" error
     #[doc(hidden)]
+    #[tracing::instrument(
+        target = "sqlx::migrate",
+        name = "migrate.run",
+        skip_all,
+        fields(
+            migrate.target = target,
+            migrate.skip = skip,
+            migrate.table = %self.table_name,
+            migrate.known = self.migrations.len(),
+        ),
+    )]
     pub async fn run_direct<C>(
         &self,
         target: Option<i64>,
@@ -235,6 +246,7 @@ impl Migrator {
     {
         // lock the database for exclusive access by the migrator
         if self.locking {
+            tracing::debug!(target: "sqlx::migrate", "acquiring migration lock");
             conn.lock().await?;
         }
 
@@ -277,8 +289,21 @@ impl Migrator {
                 }
                 None => {
                     if skip {
+                        tracing::info!(
+                            target: "sqlx::migrate",
+                            version = migration.version,
+                            description = %migration.description,
+                            "skipping migration (marking as applied without running)",
+                        );
                         conn.skip(&self.table_name, migration).await?;
                     } else {
+                        tracing::info!(
+                            target: "sqlx::migrate",
+                            version = migration.version,
+                            description = %migration.description,
+                            migration_type = ?migration.migration_type,
+                            "applying migration",
+                        );
                         conn.apply(&self.table_name, migration).await?;
                     }
                 }
@@ -288,6 +313,7 @@ impl Migrator {
         // unlock the migrator to allow other migrators to run
         // but do nothing as we already migrated
         if self.locking {
+            tracing::debug!(target: "sqlx::migrate", "releasing migration lock");
             conn.unlock().await?;
         }
 
@@ -311,6 +337,12 @@ impl Migrator {
     /// #     })
     /// # }
     /// ```
+    #[tracing::instrument(
+        target = "sqlx::migrate",
+        name = "migrate.undo",
+        skip_all,
+        fields(migrate.target = target, migrate.table = %self.table_name),
+    )]
     pub async fn undo<'a, A>(&self, migrator: A, target: i64) -> Result<(), MigrateError>
     where
         A: Acquire<'a>,
@@ -320,6 +352,7 @@ impl Migrator {
 
         // lock the database for exclusive access by the migrator
         if self.locking {
+            tracing::debug!(target: "sqlx::migrate", "acquiring migration lock");
             conn.lock().await?;
         }
 
@@ -347,12 +380,19 @@ impl Migrator {
             .filter(|m| applied_migrations.contains_key(&m.version))
             .filter(|m| m.version > target)
         {
+            tracing::info!(
+                target: "sqlx::migrate",
+                version = migration.version,
+                description = %migration.description,
+                "reverting migration",
+            );
             conn.revert(&self.table_name, migration).await?;
         }
 
         // unlock the migrator to allow other migrators to run
         // but do nothing as we already migrated
         if self.locking {
+            tracing::debug!(target: "sqlx::migrate", "releasing migration lock");
             conn.unlock().await?;
         }
 

sqlx-core/src/pool/inner.rs

@@ -243,6 +243,17 @@ impl<DB: Database> PoolInner<DB> {
         }
     }
 
+    #[tracing::instrument(
+        target = "sqlx::pool::acquire",
+        name = "pool.acquire",
+        skip_all,
+        fields(
+            pool.size = self.size(),
+            pool.idle = self.num_idle(),
+            pool.max = self.options.max_connections,
+        ),
+        level = "debug",
+    )]
     pub(super) async fn acquire(self: &Arc<Self>) -> Result<Floating<DB, Live<DB>>, Error> {
         if self.is_closed() {
             return Err(Error::PoolClosed);
@@ -322,6 +333,13 @@ impl<DB: Database> PoolInner<DB> {
         Ok(acquired)
     }
 
+    #[tracing::instrument(
+        target = "sqlx::pool::connect",
+        name = "pool.connect",
+        skip_all,
+        fields(pool.size = self.size()),
+        level = "debug",
+    )]
     pub(super) async fn connect(
         self: &Arc<Self>,
         deadline: Instant,

sqlx-core/src/transaction.rs

@@ -274,6 +274,11 @@ where
             // operation that will happen on the next asynchronous invocation of the underlying
             // connection (including if the connection is returned to a pool)
 
+            tracing::debug!(
+                target: "sqlx::transaction",
+                "transaction dropped without explicit commit/rollback; queueing implicit rollback",
+            );
+
             DB::TransactionManager::start_rollback(&mut self.connection);
         }
     }

sqlx-mysql/src/connection/establish.rs

@@ -12,6 +12,19 @@ use crate::protocol::Capabilities;
 use crate::{MySqlConnectOptions, MySqlConnection, MySqlSslMode};
 
 impl MySqlConnection {
+    #[tracing::instrument(
+        target = "sqlx::connect",
+        name = "mysql.establish",
+        skip_all,
+        fields(
+            db.system = "mysql",
+            server.address = %options.host,
+            server.port = options.port,
+            db.name = options.database.as_deref().unwrap_or_default(),
+            db.user = %options.username,
+        ),
+        level = "debug",
+    )]
     pub(crate) async fn establish(options: &MySqlConnectOptions) -> Result<Self, Error> {
         let do_handshake = DoHandshake::new(options)?;
 
@@ -22,6 +35,11 @@ impl MySqlConnection {
 
         let stream = handshake?;
 
+        tracing::debug!(
+            server_version = ?stream.server_version,
+            "MySQL connection established"
+        );
+
         Ok(Self {
             inner: Box::new(MySqlConnectionInner {
                 stream,

sqlx-mysql/src/connection/executor.rs

@@ -21,11 +21,20 @@ use futures_core::future::BoxFuture;
 use futures_core::stream::BoxStream;
 use futures_core::Stream;
 use futures_util::TryStreamExt;
+use sqlx_core::arguments::Arguments as _;
 use sqlx_core::column::{ColumnOrigin, TableColumn};
+use sqlx_core::instrument_stream::InstrumentStream;
 use sqlx_core::sql_str::SqlStr;
 use std::{pin::pin, sync::Arc};
 
 impl MySqlConnection {
+    #[tracing::instrument(
+        target = "sqlx::prepare",
+        name = "mysql.prepare",
+        skip_all,
+        fields(db.system = "mysql"),
+        level = "debug",
+    )]
     async fn prepare_statement(
         &mut self,
         sql: &str,
@@ -78,10 +87,13 @@ impl MySqlConnection {
         sql: &str,
     ) -> Result<(u32, MySqlStatementMetadata), Error> {
         if let Some(statement) = self.inner.cache_statement.get_mut(sql) {
+            tracing::trace!(target: "sqlx::prepare", "prepared statement cache hit");
             // <MySqlStatementMetadata> is internally reference-counted
             return Ok((*statement).clone());
         }
 
+        tracing::trace!(target: "sqlx::prepare", "prepared statement cache miss");
+
         let (id, metadata) = self.prepare_statement(sql).await?;
 
         // in case of the cache being full, close the least recently used statement
@@ -107,6 +119,17 @@ impl MySqlConnection {
         persistent: bool,
     ) -> Result<impl Stream<Item = Result<Either<MySqlQueryResult, MySqlRow>, Error>> + 'e, Error>
     {
+        // The span is attached to the returned stream (see `instrument_stream`)
+        // rather than via `#[tracing::instrument]` so it stays open while rows
+        // are fetched, not just while the query is set up.
+        let span = tracing::debug_span!(
+            target: "sqlx::query",
+            "mysql.run",
+            db.system = "mysql",
+            db.operation.parameters = arguments.as_ref().map_or(0, |a| a.len()),
+            db.mysql.prepared = arguments.is_some(),
+        );
+
         let mut logger = QueryLogger::new(sql, self.inner.log_settings.clone());
 
         self.inner.stream.wait_until_ready().await?;
@@ -266,7 +289,8 @@ impl MySqlConnection {
                     r#yield!(v);
                 }
             }
-        })
+        }
+        .instrument_stream(span))
     }
 }
 

sqlx-mysql/src/connection/mod.rs

@@ -70,6 +70,13 @@ impl Connection for MySqlConnection {
 
     type Options = MySqlConnectOptions;
 
+    #[tracing::instrument(
+        target = "sqlx::connect",
+        name = "mysql.close",
+        skip_all,
+        fields(db.system = "mysql"),
+        level = "debug",
+    )]
     async fn close(mut self) -> Result<(), Error> {
         self.inner.stream.send_packet(Quit).await?;
         self.inner.stream.shutdown().await?;
@@ -82,6 +89,13 @@ impl Connection for MySqlConnection {
         Ok(())
     }
 
+    #[tracing::instrument(
+        target = "sqlx::connect",
+        name = "mysql.ping",
+        skip_all,
+        fields(db.system = "mysql"),
+        level = "trace",
+    )]
     async fn ping(&mut self) -> Result<(), Error> {
         self.inner.stream.wait_until_ready().await?;
         self.inner.stream.send_packet(Ping).await?;

sqlx-mysql/src/transaction.rs

@@ -14,6 +14,13 @@ pub struct MySqlTransactionManager;
 impl TransactionManager for MySqlTransactionManager {
     type Database = MySql;
 
+    #[tracing::instrument(
+        target = "sqlx::transaction",
+        name = "mysql.transaction.begin",
+        skip_all,
+        fields(db.system = "mysql", depth = conn.inner.transaction_depth),
+        level = "debug",
+    )]
     async fn begin(conn: &mut MySqlConnection, statement: Option<SqlStr>) -> Result<(), Error> {
         let depth = conn.inner.transaction_depth;
 
@@ -30,9 +37,18 @@ impl TransactionManager for MySqlTransactionManager {
         }
         conn.inner.transaction_depth += 1;
 
+        tracing::debug!("transaction/savepoint opened");
+
         Ok(())
     }
 
+    #[tracing::instrument(
+        target = "sqlx::transaction",
+        name = "mysql.transaction.commit",
+        skip_all,
+        fields(db.system = "mysql", depth = conn.inner.transaction_depth),
+        level = "debug",
+    )]
     async fn commit(conn: &mut MySqlConnection) -> Result<(), Error> {
         let depth = conn.inner.transaction_depth;
 
@@ -44,6 +60,13 @@ impl TransactionManager for MySqlTransactionManager {
         Ok(())
     }
 
+    #[tracing::instrument(
+        target = "sqlx::transaction",
+        name = "mysql.transaction.rollback",
+        skip_all,
+        fields(db.system = "mysql", depth = conn.inner.transaction_depth),
+        level = "debug",
+    )]
     async fn rollback(conn: &mut MySqlConnection) -> Result<(), Error> {
         let depth = conn.inner.transaction_depth;
 
@@ -59,6 +82,11 @@ impl TransactionManager for MySqlTransactionManager {
         let depth = conn.inner.transaction_depth;
 
         if depth > 0 {
+            tracing::debug!(
+                target: "sqlx::transaction",
+                depth,
+                "queueing implicit rollback for unfinished transaction/savepoint on drop",
+            );
             conn.inner.stream.waiting.push_back(Waiting::Result);
             conn.inner.stream.sequence_id = 0;
             conn.inner