transaction: add contextual logging with start_ts (#267)

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.

src/raw/client.rs

@@ -591,7 +591,7 @@ impl<PdC: PdClient> Client<PdC> {
         range: impl Into<BoundRange>,
         limit: u32,
     ) -> Result<Vec<Key>> {
-        debug!("invoking raw scan_keys request");
+        debug!("invoking raw scan_keys_reverse request");
         Ok(self
             .scan_inner(range, limit, true, true)
             .await?

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<C: RetryClientTrait> RegionCache<C> {
         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<C: RetryClientTrait> RegionCache<C> {
             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<Store> {
         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<Vec<Store>> {

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<Semaphore>,
         preserve_region_results: bool,
     ) -> Result<<Self as Plan>::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<PdC: PdClient>(
     e: errorpb::Error,
     region_store: RegionStore,
 ) -> Result<bool> {
-    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<PdC: PdClient>(
     {
         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;

src/transaction/client.rs

@@ -85,7 +85,6 @@ impl Client {
     /// # });
     /// ```
     pub async fn new<S: Into<String>>(pd_endpoints: Vec<S>) -> Result<Client> {
-        // 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<Transaction> {
-        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<Transaction> {
-        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<Transaction> {
-        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)
     }

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])
 }