grpc-io: upgrade to latest version - v0.13.0.

Cargo.toml

@@ -12,7 +12,13 @@ edition = "2021"
 publish = false
 
 [features]
-default = ["test-engine-kv-rocksdb", "test-engine-raft-raft-engine", "cloud-aws", "cloud-gcp", "cloud-azure"]
+default = [
+  "test-engine-kv-rocksdb",
+  "test-engine-raft-raft-engine",
+  "cloud-aws",
+  "cloud-gcp",
+  "cloud-azure",
+]
 trace-tablet-lifetime = ["engine_rocks/trace-lifetime"]
 tcmalloc = ["tikv_alloc/tcmalloc"]
 jemalloc = ["tikv_alloc/jemalloc", "engine_rocks/jemalloc"]
@@ -21,11 +27,24 @@ snmalloc = ["tikv_alloc/snmalloc"]
 portable = ["engine_rocks/portable"]
 sse = ["engine_rocks/sse"]
 mem-profiling = ["tikv_alloc/mem-profiling"]
-failpoints = ["fail/failpoints", "raftstore/failpoints", "tikv_util/failpoints", "engine_rocks/failpoints", "raft_log_engine/failpoints"]
+failpoints = [
+  "fail/failpoints",
+  "raftstore/failpoints",
+  "tikv_util/failpoints",
+  "engine_rocks/failpoints",
+  "raft_log_engine/failpoints",
+]
 cloud-aws = ["encryption_export/cloud-aws"]
 cloud-gcp = ["encryption_export/cloud-gcp"]
 cloud-azure = ["encryption_export/cloud-azure"]
-testexport = ["raftstore/testexport", "api_version/testexport", "causal_ts/testexport", "engine_traits/testexport", "engine_rocks/testexport", "engine_panic/testexport"]
+testexport = [
+  "raftstore/testexport",
+  "api_version/testexport",
+  "causal_ts/testexport",
+  "engine_traits/testexport",
+  "engine_rocks/testexport",
+  "engine_panic/testexport",
+]
 test-engine-kv-rocksdb = ["engine_test/test-engine-kv-rocksdb"]
 test-engine-raft-raft-engine = ["engine_test/test-engine-raft-raft-engine"]
 test-engines-rocksdb = ["engine_test/test-engines-rocksdb"]
@@ -83,7 +102,10 @@ flate2 = { version = "1.0", default-features = false, features = ["zlib"] }
 futures = { version = "0.3", features = ["thread-pool", "compat"] }
 futures-executor = "0.3.1"
 futures-timer = "3.0"
-futures-util = { version = "0.3.1", default-features = false, features = ["io", "async-await"] }
+futures-util = { version = "0.3.1", default-features = false, features = [
+  "io",
+  "async-await",
+] }
 fxhash = "0.2.1"
 getset = "0.1"
 grpcio = { workspace = true }
@@ -100,7 +122,10 @@ kvproto = { workspace = true }
 lazy_static = "1.3"
 libc = "0.2"
 libloading = "0.7"
-log = { version = "0.4", features = ["max_level_trace", "release_max_level_debug"] }
+log = { version = "0.4", features = [
+  "max_level_trace",
+  "release_max_level_debug",
+] }
 log_wrappers = { workspace = true }
 match-template = "0.0.1"
 memory_trace_macros = { workspace = true }
@@ -118,7 +143,10 @@ paste = "1.0"
 pd_client = { workspace = true }
 pin-project = "1.0"
 pnet_datalink = "0.23"
-pprof = { version = "0.11", default-features = false, features = ["flamegraph", "protobuf-codec"] }
+pprof = { version = "0.11", default-features = false, features = [
+  "flamegraph",
+  "protobuf-codec",
+] }
 prometheus = { version = "0.13", features = ["nightly"] }
 prometheus-static-metric = "0.5"
 protobuf = { version = "2.8", features = ["bytes"] }
@@ -171,7 +199,7 @@ yatp = { workspace = true }
 
 [dev-dependencies]
 api_version = { workspace = true, features = ["testexport"] }
-example_coprocessor_plugin = { workspace = true } # should be a binary dependency
+example_coprocessor_plugin = { workspace = true }                              # should be a binary dependency
 hyper-openssl = "0.9"
 panic_hook = { workspace = true }
 raftstore = { workspace = true, features = ["testexport"] }
@@ -205,7 +233,7 @@ fs2 = { git = "https://github.com/tabokie/fs2-rs", branch = "tikv" }
 # Remove this when a new version is release. We need to solve rust-lang/cmake-rs#143.
 cmake = { git = "https://github.com/rust-lang/cmake-rs" }
 
-sysinfo ={ git = "https://github.com/tikv/sysinfo", branch = "0.26-fix-cpu" }
+sysinfo = { git = "https://github.com/tikv/sysinfo", branch = "0.26-fix-cpu" }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 procinfo = { git = "https://github.com/tikv/procinfo-rs", rev = "7693954bd1dd86eb1709572fd7b62fd5f7ff2ea1" }
@@ -245,9 +273,9 @@ members = [
   "components/encryption",
   "components/encryption/export",
   "components/engine_rocks_helper",
-# Only enable tirocks in local development, otherwise it can slow down compilation.
-# TODO: always enable tirocks and remove engine_rocks.
-#  "components/engine_tirocks",
+  # Only enable tirocks in local development, otherwise it can slow down compilation.
+  # TODO: always enable tirocks and remove engine_rocks.
+  #  "components/engine_tirocks",
   "components/error_code",
   "components/external_storage",
   "components/file_system",
@@ -372,15 +400,26 @@ tipb_helper = { path = "components/tipb_helper" }
 tracker = { path = "components/tracker" }
 txn_types = { path = "components/txn_types" }
 # External libs
-raft = { version = "0.7.0", default-features = false, features = ["protobuf-codec"] }
-grpcio = { version = "0.10.4", default-features = false, features = ["openssl", "protobuf-codec", "nightly"] }
-grpcio-health = { version = "0.10.4", default-features = false, features = ["protobuf-codec"] }
+raft = { version = "0.7.0", default-features = false, features = [
+  "protobuf-codec",
+] }
+grpcio = { version = "0.13.0", default-features = false, features = [
+  "openssl",
+  "protobuf-codec",
+  "nightly",
+] }
+grpcio-health = { version = "0.13.0", default-features = false, features = [
+  "protobuf-codec",
+] }
 tipb = { git = "https://github.com/pingcap/tipb.git" }
 kvproto = { git = "https://github.com/pingcap/kvproto.git" }
 yatp = { git = "https://github.com/tikv/yatp.git", branch = "master" }
 tokio-timer = { git = "https://github.com/tikv/tokio", branch = "tokio-timer-hotfix" }
 tokio-executor = { git = "https://github.com/tikv/tokio", branch = "tokio-timer-hotfix" }
-slog = { version = "2.3", features = ["max_level_trace", "release_max_level_debug"] }
+slog = { version = "2.3", features = [
+  "max_level_trace",
+  "release_max_level_debug",
+] }
 slog-global = { version = "0.1", git = "https://github.com/breeswish/slog-global.git", rev = "d592f88e4dbba5eb439998463054f1a44fbf17b9" }
 openssl = "0.10"
 openssl-sys = "0.9"

components/backup-stream/tests/suite.rs

@@ -22,7 +22,7 @@ use backup_stream::{
     RegionSet, Service, Task,
 };
 use futures::{executor::block_on, AsyncWriteExt, Future, Stream, StreamExt};
-use grpcio::{ChannelBuilder, Server, ServerBuilder};
+use grpcio::{ChannelBuilder, Server, ServerBuilder, ServerCredentials};
 use kvproto::{
     brpb::{CompressionType, Local, Metadata, StorageBackend},
     kvrpcpb::*,
@@ -342,9 +342,11 @@ impl Suite {
         let serv = Service::new(endpoint.scheduler());
         let builder =
             ServerBuilder::new(self.env.clone()).register_service(create_log_backup(serv));
-        let mut server = builder.bind("127.0.0.1", 0).build().unwrap();
+        let mut server = builder.build().unwrap();
+        let port = server
+            .add_listening_port(format!("127.0.0.1:{}", 0), ServerCredentials::insecure())
+            .unwrap();
         server.start();
-        let (_, port) = server.bind_addrs().next().unwrap();
         let addr = format!("127.0.0.1:{}", port);
         let channel = ChannelBuilder::new(self.env.clone()).connect(&addr);
         println!("connecting channel to {} for store {}", addr, id);

components/backup/src/service.rs

@@ -158,9 +158,11 @@ mod tests {
         let backup_service = super::Service::<RocksEngine, RocksEngine>::new(scheduler);
         let builder =
             ServerBuilder::new(env.clone()).register_service(create_backup(backup_service));
-        let mut server = builder.bind("127.0.0.1", 0).build().unwrap();
-        server.start();
-        let (_, port) = server.bind_addrs().next().unwrap();
+        let mut server = builder.build().unwrap();
+        let addr = "127.0.0.1";
+        let port = server
+            .add_listening_port(format!("{}:{}", addr, 0), ServerCredentials::insecure())
+            .unwrap();
         let addr = format!("127.0.0.1:{}", port);
         let channel = ChannelBuilder::new(env).connect(&addr);
         let client = BackupClient::new(channel);

components/cdc/src/service.rs

@@ -525,7 +525,9 @@ mod tests {
     use std::{sync::Arc, time::Duration};
 
     use futures::{executor::block_on, SinkExt};
-    use grpcio::{self, ChannelBuilder, EnvBuilder, Server, ServerBuilder, WriteFlags};
+    use grpcio::{
+        self, ChannelBuilder, EnvBuilder, Server, ServerBuilder, ServerCredentials, WriteFlags,
+    };
     use kvproto::cdcpb::{create_change_data, ChangeDataClient, ResolvedTs};
     use tikv_util::future::block_on_timeout;
 
@@ -539,9 +541,11 @@ mod tests {
         let env = Arc::new(EnvBuilder::new().build());
         let builder =
             ServerBuilder::new(env.clone()).register_service(create_change_data(cdc_service));
-        let mut server = builder.bind("127.0.0.1", 0).build().unwrap();
-        server.start();
-        let (_, port) = server.bind_addrs().next().unwrap();
+        let mut server = builder.build().unwrap();
+        let addr = "127.0.0.1";
+        let port = server
+            .add_listening_port(format!("{}:{}", addr, 0), ServerCredentials::insecure())
+            .unwrap();
         let addr = format!("127.0.0.1:{}", port);
         let channel = ChannelBuilder::new(env).connect(&addr);
         let client = ChangeDataClient::new(channel);

components/resolved_ts/src/advance.rs

@@ -574,7 +574,7 @@ mod tests {
         time::Duration,
     };
 
-    use grpcio::{self, ChannelBuilder, EnvBuilder, Server, ServerBuilder};
+    use grpcio::{self, ChannelBuilder, EnvBuilder, Server, ServerBuilder, ServerCredentials};
     use kvproto::{metapb::Region, tikvpb::Tikv, tikvpb_grpc::create_tikv};
     use pd_client::PdClient;
     use raftstore::store::util::RegionReadProgress;
@@ -608,9 +608,11 @@ mod tests {
         let (tx, rx) = channel();
         let tikv_service = MockTikv { req_tx: tx };
         let builder = ServerBuilder::new(env.clone()).register_service(create_tikv(tikv_service));
-        let mut server = builder.bind("127.0.0.1", 0).build().unwrap();
-        server.start();
-        let (_, port) = server.bind_addrs().next().unwrap();
+        let mut server = builder.build().unwrap();
+        let addr = "127.0.0.1";
+        let port = server
+            .add_listening_port(format!("{}:{}", addr, 0), ServerCredentials::insecure())
+            .unwrap();
         let addr = format!("127.0.0.1:{}", port);
         let channel = ChannelBuilder::new(env).connect(&addr);
         let client = TikvClient::new(channel);

components/security/src/lib.rs

@@ -15,8 +15,8 @@ use collections::HashSet;
 use encryption::EncryptionConfig;
 use grpcio::{
     CertificateRequestType, Channel, ChannelBuilder, ChannelCredentialsBuilder, CheckResult,
-    RpcContext, RpcStatus, RpcStatusCode, ServerBuilder, ServerChecker, ServerCredentialsBuilder,
-    ServerCredentialsFetcher,
+    RpcContext, RpcStatus, RpcStatusCode, Server, ServerBuilder, ServerChecker, ServerCredentials,
+    ServerCredentialsBuilder, ServerCredentialsFetcher,
 };
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Default)]
@@ -163,33 +163,39 @@ impl SecurityManager {
                 .root_cert(ca)
                 .cert(cert, key)
                 .build();
-            cb.secure_connect(addr, cred)
+            cb.set_credentials(cred).connect(addr)
         }
     }
 
-    pub fn bind(&self, mut sb: ServerBuilder, addr: &str, port: u16) -> ServerBuilder {
+    pub fn gen_credentials(&self) -> ServerCredentials {
         if self.cfg.ca_path.is_empty() {
-            sb.bind(addr, port)
+            ServerCredentials::insecure()
         } else {
-            if !self.cfg.cert_allowed_cn.is_empty() {
-                let cn_checker = CnChecker {
-                    allowed_cn: Arc::new(self.cfg.cert_allowed_cn.clone()),
-                };
-                sb = sb.add_checker(cn_checker);
-            }
             let fetcher = Box::new(Fetcher {
                 cfg: self.cfg.clone(),
                 last_modified_time: Arc::new(Mutex::new(None)),
             });
-            sb.bind_with_fetcher(
-                addr,
-                port,
+            ServerCredentials::with_fetcher(
                 fetcher,
                 CertificateRequestType::RequestAndRequireClientCertificateAndVerify,
             )
         }
     }
 
+    pub fn bind(&self, mut sb: ServerBuilder, addr: &str, port: u16) -> (Server, u16) {
+        let addr = format!("{}:{}", addr, port);
+        let creds = self.gen_credentials();
+        if !self.cfg.ca_path.is_empty() && !self.cfg.cert_allowed_cn.is_empty() {
+            let cn_checker = CnChecker {
+                allowed_cn: Arc::new(self.cfg.cert_allowed_cn.clone()),
+            };
+            sb = sb.add_checker(cn_checker);
+        }
+        let mut server = sb.build().unwrap();
+        let port = server.add_listening_port(addr, creds).unwrap();
+        (server, port)
+    }
+
     pub fn get_config(&self) -> &SecurityConfig {
         &self.cfg
     }