Merge branch 'master' into xlu/sync-api

Author: Xuhui Lu

Cargo.toml

@@ -31,6 +31,8 @@ rand = "0.8"
 regex = "1"
 serde = "1.0"
 serde_derive = "1.0"
+slog = { version = "2.3", features = ["max_level_trace", "release_max_level_debug"] }
+slog-term = { version = "2.4" }
 thiserror = "1"
 tokio = { version = "1.0", features = [ "sync", "time" ] }
 

README.md

@@ -33,7 +33,7 @@ Raw mode:
 ```rust
 use tikv_client::RawClient;
 
-let client = RawClient::new(vec!["127.0.0.1:2379"]).await?;
+let client = RawClient::new(vec!["127.0.0.1:2379"], None).await?;
 client.put("key".to_owned(), "value".to_owned()).await?;
 let value = client.get("key".to_owned()).await?;
 ```
@@ -43,7 +43,7 @@ Transactional mode:
 ```rust
 use tikv_client::TransactionClient;
 
-let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"]).await?;
+let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"], None).await?;
 let mut txn = txn_client.begin_optimistic().await?;
 txn.put("key".to_owned(), "value".to_owned()).await?;
 let value = txn.get("key".to_owned()).await?;

examples/pessimistic.rs

@@ -20,7 +20,7 @@ async fn main() {
     };
 
     // init
-    let client = Client::new_with_config(args.pd, config)
+    let client = Client::new_with_config(args.pd, config, None)
         .await
         .expect("Could not connect to tikv");
 

examples/raw.rs

@@ -26,7 +26,7 @@ async fn main() -> Result<()> {
 
     // When we first create a client we receive a `Connect` structure which must be resolved before
     // the client is actually connected and usable.
-    let client = Client::new_with_config(args.pd, config).await?;
+    let client = Client::new_with_config(args.pd, config, None).await?;
 
     // Requests are created from the connected client. These calls return structures which
     // implement `Future`. This means the `Future` must be resolved before the action ever takes

examples/transaction.rs

@@ -81,7 +81,7 @@ async fn main() {
         Config::default()
     };
 
-    let txn = Client::new_with_config(args.pd, config)
+    let txn = Client::new_with_config(args.pd, config, None)
         .await
         .expect("Could not connect to tikv");
 

getting-started.md

@@ -23,7 +23,7 @@ Raw mode:
 ```rust
 use tikv_client::RawClient;
 
-let client = RawClient::new(vec!["127.0.0.1:2379"]).await?;
+let client = RawClient::new(vec!["127.0.0.1:2379"], None).await?;
 client.put("key".to_owned(), "value".to_owned()).await?;
 let value = client.get("key".to_owned()).await?;
 ```
@@ -33,7 +33,7 @@ Transactional mode:
 ```rust
 use tikv_client::TransactionClient;
 
-let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"]).await?;
+let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"], None).await?;
 let mut txn = txn_client.begin_optimistic().await?;
 txn.put("key".to_owned(), "value".to_owned()).await?;
 let value = txn.get("key".to_owned()).await?;
@@ -46,7 +46,7 @@ To make an example which builds and runs,
 use tikv_client::{TransactionClient, Error};
 
 async fn run() -> Result<(), Error> {
-    let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"]).await?;
+    let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"], None).await?;
     let mut txn = txn_client.begin_optimistic().await?;
     txn.put("key".to_owned(), "value".to_owned()).await?;
     let value = txn.get("key".to_owned()).await?;

src/lib.rs

@@ -67,7 +67,7 @@
 //! # use futures::prelude::*;
 //! # fn main() -> Result<()> {
 //! # futures::executor::block_on(async {
-//! let client = RawClient::new(vec!["127.0.0.1:2379"]).await?;
+//! let client = RawClient::new(vec!["127.0.0.1:2379"], None).await?;
 //! client.put("key".to_owned(), "value".to_owned()).await?;
 //! let value = client.get("key".to_owned()).await?;
 //! # Ok(())
@@ -81,7 +81,7 @@
 //! # use futures::prelude::*;
 //! # fn main() -> Result<()> {
 //! # futures::executor::block_on(async {
-//! let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"]).await?;
+//! let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"], None).await?;
 //! let mut txn = txn_client.begin_optimistic().await?;
 //! txn.put("key".to_owned(), "value".to_owned()).await?;
 //! let value = txn.get("key".to_owned()).await?;
@@ -115,7 +115,8 @@ mod mock;
 mod proptests;
 
 #[macro_use]
-extern crate log;
+extern crate slog;
+extern crate slog_term;
 
 #[doc(inline)]
 pub use crate::backoff::Backoff;

src/pd/client.rs

@@ -11,6 +11,7 @@ use crate::{
 use async_trait::async_trait;
 use futures::{prelude::*, stream::BoxStream};
 use grpcio::{EnvBuilder, Environment};
+use slog::{Drain, Logger};
 use std::{
     collections::HashMap,
     sync::{Arc, RwLock},
@@ -207,6 +208,7 @@ pub struct PdRpcClient<KvC: KvConnect + Send + Sync + 'static = TikvConnect, Cl
     kv_connect: KvC,
     kv_client_cache: Arc<RwLock<HashMap<String, KvC::KvClient>>>,
     enable_codec: bool,
+    logger: Logger,
 }
 
 #[async_trait]
@@ -284,6 +286,9 @@ impl<KvC: KvConnect + Send + Sync + 'static, Cl> PdRpcClient<KvC, Cl> {
         MakeKvC: FnOnce(Arc<Environment>, Arc<SecurityManager>) -> KvC,
         MakePd: FnOnce(Arc<Environment>, Arc<SecurityManager>) -> PdFut,
     {
+        let plain = slog_term::PlainSyncDecorator::new(std::io::stdout());
+        let logger = Logger::root(slog_term::FullFormat::new(plain).build().fuse(), o!());
+        info!(logger, "Logging ready!");
         let env = Arc::new(
             EnvBuilder::new()
                 .cq_count(CQ_COUNT)
@@ -307,14 +312,15 @@ impl<KvC: KvConnect + Send + Sync + 'static, Cl> PdRpcClient<KvC, Cl> {
             kv_client_cache,
             kv_connect: kv_connect(env, security_mgr),
             enable_codec,
+            logger,
         })
     }
 
     fn kv_client(&self, address: &str) -> Result<KvC::KvClient> {
         if let Some(client) = self.kv_client_cache.read().unwrap().get(address) {
             return Ok(client.clone());
         };
-        info!("connect to tikv endpoint: {:?}", address);
+        info!(self.logger, "connect to tikv endpoint: {:?}", address);
         self.kv_connect.connect(address).map(|client| {
             self.kv_client_cache
                 .write()