diff --git a/README.md b/README.md index 7e5e6d9a..619e683b 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,22 @@ Important note: It is **not recommended or supported** to use both the raw and t | `send_heart_beat` | | `u64` (TTL) | | | `gc` | `Timestamp` | `bool` | Returns true if the latest safepoint in PD equals the parameter | +## Logging + +The client emits logs through the [`log`](https://docs.rs/log) facade. It does +not install a logger itself — pick any `log`-compatible backend in your +application (the examples in `examples/` use [`env_logger`](https://docs.rs/env_logger)). +With `env_logger`, set `RUST_LOG` to control verbosity, e.g.: + +```sh +RUST_LOG=tikv_client=debug cargo run --example transaction +``` + +`debug` traces the transaction lifecycle with each transaction's `start_ts` +(begin, commit/rollback and their outcomes, prewrite, pessimistic lock, drop), +plus region/retry activity; `info` and above surface only infrequent +operational and error events. + # Development and contributing We welcome your contributions! Contributing code is great, we also appreciate filing [issues](https://github.com/tikv/client-rust/issues/new) to identify bugs and provide feedback, adding tests or examples, and improvements to documentation. diff --git a/src/raw/client.rs b/src/raw/client.rs index f8991e48..d75d20bb 100644 --- a/src/raw/client.rs +++ b/src/raw/client.rs @@ -591,7 +591,7 @@ impl Client { range: impl Into, limit: u32, ) -> Result> { - debug!("invoking raw scan_keys request"); + debug!("invoking raw scan_keys_reverse request"); Ok(self .scan_inner(range, limit, true, true) .await? diff --git a/src/region_cache.rs b/src/region_cache.rs index 70d17289..2a4b0fbb 100644 --- a/src/region_cache.rs +++ b/src/region_cache.rs @@ -5,6 +5,7 @@ use std::collections::HashMap; use std::collections::HashSet; use std::sync::Arc; +use log::debug; use tokio::sync::Notify; use tokio::sync::RwLock; @@ -216,6 +217,7 @@ impl RegionCache { let region_entry = cache.ver_id_to_region.get_mut(&ver_id); if let Some(region) = region_entry { region.leader = Some(leader); + debug!("updated cached region leader, region: {:?}", ver_id); } Ok(()) @@ -230,12 +232,17 @@ impl RegionCache { cache.ver_id_to_region.remove(&ver_id); cache.id_to_ver_id.remove(&id); cache.key_to_ver_id.remove(&start_key); + debug!("invalidated region cache entry, region: {:?}", ver_id); } } pub async fn invalidate_store_cache(&self, store_id: StoreId) -> Option { let mut cache = self.store_cache.write().await; - cache.remove(&store_id) + let removed = cache.remove(&store_id); + if removed.is_some() { + debug!("invalidated store cache entry, store: {:?}", store_id); + } + removed } pub async fn read_through_all_stores(&self) -> Result> { diff --git a/src/request/plan.rs b/src/request/plan.rs index c8acb5da..ffc949e3 100644 --- a/src/request/plan.rs +++ b/src/request/plan.rs @@ -9,6 +9,7 @@ use futures::future::try_join_all; use futures::prelude::*; use log::debug; use log::info; +use log::warn; use tokio::sync::Semaphore; use tokio::time::sleep; @@ -162,9 +163,12 @@ where permits: Arc, preserve_region_results: bool, ) -> Result<::Result> { - debug!("single_shard_handler"); let region_ver_id = region.ver_id(); let store_id = region.get_store_id().ok(); + debug!( + "single_shard_handler, region: {:?}, store: {:?}", + region_ver_id, store_id + ); let region_store = match pd_client .clone() .map_region_to_store(region) @@ -221,7 +225,10 @@ where debug!("single_shard_handler:execute: key errors: {:?}", e); Ok(vec![Err(Error::MultipleKeyErrors(e))]) } else if let Some(e) = resp.region_error() { - debug!("single_shard_handler:execute: region error: {:?}", e); + debug!( + "single_shard_handler:execute: region error: {:?}, region: {:?}", + e, region_ver_id + ); match backoff.next_delay_duration() { Some(duration) => { let region_error_resolved = @@ -239,7 +246,13 @@ where ) .await } - None => Err(Error::RegionError(Box::new(e))), + None => { + warn!( + "giving up after exhausting retries on region error, region: {:?}", + region_ver_id + ); + Err(Error::RegionError(Box::new(e))) + } } } else { Ok(vec![Ok(resp)]) @@ -290,9 +303,9 @@ pub(crate) async fn handle_region_error( e: errorpb::Error, region_store: RegionStore, ) -> Result { - debug!("handle_region_error: {:?}", e); let ver_id = region_store.region_with_leader.ver_id(); let store_id = region_store.region_with_leader.get_store_id(); + debug!("handling region error: {:?}, region: {:?}", e, ver_id); if let Some(not_leader) = e.not_leader { if let Some(leader) = not_leader.leader { match pd_client @@ -330,8 +343,10 @@ pub(crate) async fn handle_region_error( { Err(Error::RegionError(Box::new(e))) } else { - // TODO: pass the logger around - // info!("unknwon region error: {:?}", e); + debug!( + "unknown region error, invalidating region and store caches, region: {:?}: {:?}", + ver_id, e + ); pd_client.invalidate_region_cache(ver_id).await; if let Ok(store_id) = store_id { pd_client.invalidate_store_cache(store_id).await; diff --git a/src/transaction/client.rs b/src/transaction/client.rs index a40044b7..62f24859 100644 --- a/src/transaction/client.rs +++ b/src/transaction/client.rs @@ -85,7 +85,6 @@ impl Client { /// # }); /// ``` pub async fn new>(pd_endpoints: Vec) -> Result { - // debug!("creating transactional client"); Self::new_with_config(pd_endpoints, Config::default()).await } @@ -173,8 +172,11 @@ impl Client { /// # }); /// ``` pub async fn begin_optimistic(&self) -> Result { - debug!("creating new optimistic transaction"); let timestamp = self.current_timestamp().await?; + debug!( + "began optimistic transaction, start_ts: {}", + timestamp.version() + ); Ok(self.new_transaction(timestamp, TransactionOptions::new_optimistic())) } @@ -196,8 +198,11 @@ impl Client { /// # }); /// ``` pub async fn begin_pessimistic(&self) -> Result { - debug!("creating new pessimistic transaction"); let timestamp = self.current_timestamp().await?; + debug!( + "began pessimistic transaction, start_ts: {}", + timestamp.version() + ); Ok(self.new_transaction(timestamp, TransactionOptions::new_pessimistic())) } @@ -219,8 +224,8 @@ impl Client { /// # }); /// ``` pub async fn begin_with_options(&self, options: TransactionOptions) -> Result { - debug!("creating new customized transaction"); let timestamp = self.current_timestamp().await?; + debug!("began transaction, start_ts: {}", timestamp.version()); Ok(self.new_transaction(timestamp, options)) } @@ -276,7 +281,10 @@ impl Client { .update_safepoint(safepoint.version()) .await?; if !res { - info!("new safepoint != user-specified safepoint"); + info!( + "GC safepoint not updated: PD already holds a safepoint newer than the requested {}", + safepoint.version() + ); } Ok(res) } diff --git a/src/transaction/lock.rs b/src/transaction/lock.rs index 5c5d2513..2e662b1c 100644 --- a/src/transaction/lock.rs +++ b/src/transaction/lock.rs @@ -37,7 +37,7 @@ use crate::transaction::requests::TransactionStatusKind; use crate::Error; use crate::Result; -fn format_key_for_log(key: &[u8]) -> String { +pub(crate) fn format_key_for_log(key: &[u8]) -> String { let prefix_len = key.len().min(16); format!("len={}, prefix={:?}", key.len(), &key[..prefix_len]) } diff --git a/src/transaction/transaction.rs b/src/transaction/transaction.rs index 27e465e5..1b715a7a 100644 --- a/src/transaction/transaction.rs +++ b/src/transaction/transaction.rs @@ -14,7 +14,6 @@ use tokio::time::Duration; use crate::backoff::Backoff; use crate::backoff::DEFAULT_REGION_BACKOFF; -use crate::kv::HexRepr; use crate::pd::PdClient; use crate::pd::PdRpcClient; use crate::proto::kvrpcpb; @@ -32,6 +31,7 @@ use crate::request::RetryOptions; use crate::request::TruncateKeyspace; use crate::timestamp::TimestampExt; use crate::transaction::buffer::Buffer; +use crate::transaction::lock::format_key_for_log; use crate::transaction::lowering::*; use crate::BoundRange; use crate::Error; @@ -642,7 +642,10 @@ impl Transaction { /// # }); /// ``` pub async fn commit(&mut self) -> Result> { - debug!("commiting transaction"); + debug!( + "committing transaction, start_ts: {}", + self.timestamp.version() + ); if !self.transit_status( |status| { matches!( @@ -677,8 +680,14 @@ impl Transaction { .commit() .await; - if res.is_ok() { + if let Ok(commit_ts) = &res { self.set_status(TransactionStatus::Committed); + debug!( + "transaction committed, start_ts: {}, commit_ts: {:?}, elapsed: {:?}", + self.timestamp.version(), + commit_ts.as_ref().map(|ts| ts.version()), + self.start_instant.elapsed(), + ); } res } @@ -700,7 +709,10 @@ impl Transaction { /// # }); /// ``` pub async fn rollback(&mut self) -> Result<()> { - debug!("rolling back transaction"); + debug!( + "rolling back transaction, start_ts: {}", + self.timestamp.version() + ); if !self.transit_status( |status| { matches!( @@ -732,6 +744,10 @@ impl Transaction { if res.is_ok() { self.set_status(TransactionStatus::Rolledback); + debug!( + "transaction rolled back, start_ts: {}", + self.timestamp.version() + ); } res } @@ -746,7 +762,7 @@ impl Transaction { /// Returns the TTL set on the transaction's locks by TiKV. #[doc(hidden)] pub async fn send_heart_beat(&mut self) -> Result { - debug!("sending heart_beat"); + debug!("sending heartbeat, start_ts: {}", self.timestamp.version()); self.check_allow_operation().await?; let primary_key = match self.buffer.get_primary_key() { Some(k) => k, @@ -825,7 +841,6 @@ impl Transaction { keys: impl IntoIterator, need_value: bool, ) -> Result> { - debug!("acquiring pessimistic lock"); assert!( matches!(self.options.kind, TransactionKind::Pessimistic(_)), "`pessimistic_lock` is only valid to use with pessimistic transactions" @@ -835,6 +850,12 @@ impl Transaction { if keys.is_empty() { return Ok(vec![]); } + debug!( + "acquiring pessimistic lock, start_ts: {}, keys: {}, need_value: {}", + self.timestamp.version(), + keys.len(), + need_value, + ); let first_key = keys[0].clone().key(); // we do not set the primary key here, because pessimistic lock request @@ -871,6 +892,12 @@ impl Transaction { inner, success_keys, } if !success_keys.is_empty() => { + debug!( + "pessimistic lock failed, rolling back {} partially-acquired lock(s), start_ts: {}, for_update_ts: {}", + success_keys.len(), + self.timestamp.version(), + for_update_ts.version(), + ); let keys = success_keys.into_iter().map(Key::from); self.pessimistic_lock_rollback(keys, self.timestamp.clone(), for_update_ts) .await?; @@ -899,12 +926,16 @@ impl Transaction { start_version: Timestamp, for_update_ts: Timestamp, ) -> Result<()> { - debug!("rollback pessimistic lock"); - let keys: Vec<_> = keys.into_iter().collect(); if keys.is_empty() { return Ok(()); } + debug!( + "rolling back pessimistic lock, start_ts: {}, for_update_ts: {}, keys: {}", + start_version.version(), + for_update_ts.version(), + keys.len(), + ); let req = new_pessimistic_rollback_request( keys.clone().into_iter(), @@ -945,7 +976,6 @@ impl Transaction { } async fn start_auto_heartbeat(&mut self) { - debug!("starting auto_heartbeat"); if !self.options.heartbeat_option.is_auto_heartbeat() || self.is_heartbeat_started { return; } @@ -965,6 +995,11 @@ impl Transaction { }; let start_instant = self.start_instant; let keyspace = self.keyspace; + debug!( + "starting auto-heartbeat, start_ts: {}, interval: {:?}", + self.timestamp.version(), + heartbeat_interval, + ); let heartbeat_task = async move { loop { @@ -994,9 +1029,14 @@ impl Transaction { Ok::<(), Error>(()) }; - tokio::spawn(async { + let start_ts_for_log = self.timestamp.version(); + tokio::spawn(async move { if let Err(err) = heartbeat_task.await { - log::error!("Error: While sending heartbeat. {}", err); + log::error!( + "auto-heartbeat task terminated, start_ts: {}: {}", + start_ts_for_log, + err + ); } }); } @@ -1034,20 +1074,28 @@ impl Transaction { impl Drop for Transaction { fn drop(&mut self) { - debug!("dropping transaction"); + debug!( + "dropping transaction, start_ts: {}, status: {:?}", + self.timestamp.version(), + self.get_status() + ); if std::thread::panicking() { return; } if self.get_status() == TransactionStatus::Active { + let start_ts = self.timestamp.version(); match self.options.check_level { CheckLevel::Panic => { - panic!("Dropping an active transaction. Consider commit or rollback it.") + panic!("dropping an active transaction (start_ts: {start_ts}). Consider commit or rollback it.") } CheckLevel::Warn => { - warn!("Dropping an active transaction. Consider commit or rollback it.") + warn!("dropping an active transaction, start_ts: {start_ts}. Consider commit or rollback it.") + } + // Even with the drop check disabled, leave a debug breadcrumb so + // an unfinished transaction is not completely silent. + CheckLevel::None => { + debug!("dropping an active transaction (drop check disabled), start_ts: {start_ts}") } - - CheckLevel::None => {} } } self.set_status(TransactionStatus::Dropped); @@ -1270,7 +1318,10 @@ struct Committer { impl Committer { async fn commit(mut self) -> Result> { - debug!("committing"); + debug!( + "committing (2pc), start_ts: {}", + self.start_version.version() + ); let min_commit_ts = self.prewrite().await?; @@ -1309,7 +1360,11 @@ impl Committer { } async fn prewrite(&mut self) -> Result> { - debug!("prewriting"); + debug!( + "prewriting, start_ts: {}, mutations: {}", + self.start_version.version(), + self.mutations.len() + ); let primary_lock = self.primary_key.clone().unwrap(); let elapsed = self.start_instant.elapsed().as_millis() as u64; let lock_ttl = self.calc_txn_lock_ttl(); @@ -1375,7 +1430,10 @@ impl Committer { /// Commits the primary key and returns the commit version async fn commit_primary(&mut self) -> Result { - debug!("committing primary"); + debug!( + "committing primary, start_ts: {}", + self.start_version.version() + ); let primary_key = self.primary_key.clone().into_iter(); let commit_version = self.rpc.clone().get_timestamp().await?; let req = new_commit_request( @@ -1424,8 +1482,8 @@ impl Committer { let primary_key = self.primary_key.as_ref().unwrap(); if primary_key != expired.key.as_ref() { - error!("2PC commit_ts rejected by TiKV, but the key is not the primary key, start_ts: {}, key: {}, primary: {:?}", - self.start_version.version(), HexRepr(&expired.key), primary_key); + error!("2PC commit_ts rejected by TiKV, but the key is not the primary key, start_ts: {}, key: {}, primary: {}", + self.start_version.version(), format_key_for_log(&expired.key), format_key_for_log(primary_key)); return Err(Error::StringError("2PC commitTS rejected by TiKV, but the key is not the primary key".to_string())); } @@ -1454,7 +1512,11 @@ impl Committer { } async fn commit_secondary(self, commit_version: Timestamp) -> Result<()> { - debug!("committing secondary"); + debug!( + "committing secondary keys, start_ts: {}, mutations: {}", + self.start_version.version(), + self.mutations.len() + ); let start_version = self.start_version.clone(); let mutations_len = self.mutations.len(); let primary_only = mutations_len == 1; @@ -1514,7 +1576,10 @@ impl Committer { } async fn rollback(self) -> Result<()> { - debug!("rolling back"); + debug!( + "rolling back (2pc), start_ts: {}", + self.start_version.version() + ); if self.options.kind == TransactionKind::Optimistic && self.mutations.is_empty() { return Ok(()); } @@ -1566,7 +1631,7 @@ impl Committer { } } -#[derive(PartialEq, Eq, Clone, Copy)] +#[derive(Debug, PartialEq, Eq, Clone, Copy)] #[repr(u8)] enum TransactionStatus { /// The transaction is read-only [`Snapshot`](super::Snapshot), no need to commit or rollback or panic on drop. @@ -1607,6 +1672,8 @@ mod tests { use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::sync::Arc; + use std::sync::Mutex; + use std::sync::Once; use std::time::Duration; use fail::FailScenario; @@ -1617,6 +1684,7 @@ mod tests { use crate::proto::pdpb::Timestamp; use crate::request::Keyspace; use crate::transaction::HeartbeatOption; + use crate::TimestampExt; use crate::Transaction; use crate::TransactionOptions; @@ -1702,4 +1770,71 @@ mod tests { heartbeat_txn_handle.await.unwrap(); Ok(()) } + + // A minimal capturing logger (no extra dependency) used to assert that + // transaction lifecycle logs carry the txn's `start_ts`. + static CAPTURED_LOGS: Mutex> = Mutex::new(Vec::new()); + static LOGGER: CaptureLogger = CaptureLogger; + static LOGGER_INIT: Once = Once::new(); + + struct CaptureLogger; + + impl log::Log for CaptureLogger { + fn enabled(&self, _metadata: &log::Metadata) -> bool { + true + } + + fn log(&self, record: &log::Record) { + if let Ok(mut logs) = CAPTURED_LOGS.lock() { + logs.push(record.args().to_string()); + } + } + + fn flush(&self) {} + } + + fn install_capture_logger() { + LOGGER_INIT.call_once(|| { + // Ignore the error if another logger is already installed in this + // process; the assertion below only checks for presence of a unique + // marker, so foreign records are harmless. + let _ = log::set_logger(&LOGGER); + log::set_max_level(log::LevelFilter::Debug); + }); + } + + #[tokio::test] + async fn commit_logs_start_ts() { + install_capture_logger(); + CAPTURED_LOGS.lock().unwrap().clear(); + + let pd_client = Arc::new(MockPdClient::new(MockKvClient::with_dispatch_hook( + |req: &dyn Any| { + if req.downcast_ref::().is_some() { + Ok(Box::::default() as Box) + } else { + Ok(Box::::default() as Box) + } + }, + ))); + + // A unique start_ts so the assertion cannot be satisfied by any other + // test's log records if the process is shared (e.g. plain `cargo test`). + let start_ts = 424242; + let mut txn = Transaction::new( + Timestamp::from_version(start_ts), + pd_client, + TransactionOptions::new_optimistic(), + Keyspace::Disable, + ); + txn.put("key1".to_owned(), "value").await.unwrap(); + txn.commit().await.unwrap(); + + let logs = CAPTURED_LOGS.lock().unwrap(); + assert!( + logs.iter() + .any(|line| line.contains("start_ts") && line.contains(&start_ts.to_string())), + "expected a lifecycle log carrying start_ts {start_ts}; captured: {logs:?}" + ); + } }