Xuanwo commented on code in PR #2565: URL: https://github.com/apache/incubator-opendal/pull/2565#discussion_r1262854863
########## core/src/services/mod.rs: ########## @@ -183,3 +183,8 @@ pub use vercel_artifacts::VercelArtifacts; mod redb; #[cfg(feature = "services-redb")] pub use self::redb::Redb; + +#[cfg(feature = "services-tikv")] +mod tikv; +#[cfg(feature = "services-tikv")] +pub use self::tikv::TiKV; Review Comment: Please follow the same naming style like other serivces: `Tikv`. ########## core/src/services/tikv/backend.rs: ########## @@ -0,0 +1,257 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; +use tikv_client::Config; +use tikv_client::RawClient; + +use crate::raw::adapters::kv; +use crate::Capability; +use crate::Scheme; +use async_trait::async_trait; + +use crate::Builder; +use crate::Error; +use crate::ErrorKind; +use crate::*; + +use std::fmt::Debug; +use std::fmt::Formatter; + +/// TiKV backend builder +#[derive(Clone, Default)] +pub struct TiKVBuilder { + /// network address of the TiKV service. + /// + /// default is "127.0.0.1:2379" + endpoints: Option<Vec<String>>, + /// whether using insecure connection to TiKV + insecure: bool, + /// certificate authority file path + ca_path: Option<String>, + /// cert path + cert_path: Option<String>, + /// key path + key_path: Option<String>, +} + +impl TiKVBuilder { + /// Set the network address of the TiKV service. + pub fn endpoints(&mut self, endpoints: impl Into<Vec<String>>) -> &mut Self { Review Comment: I don't like unneed convertion API like this. How about accepting `Vec<&str>` directly? ########## core/src/services/tikv/backend.rs: ########## @@ -0,0 +1,257 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; +use tikv_client::Config; +use tikv_client::RawClient; + +use crate::raw::adapters::kv; +use crate::Capability; +use crate::Scheme; +use async_trait::async_trait; + +use crate::Builder; +use crate::Error; +use crate::ErrorKind; +use crate::*; + +use std::fmt::Debug; +use std::fmt::Formatter; + +/// TiKV backend builder +#[derive(Clone, Default)] +pub struct TiKVBuilder { + /// network address of the TiKV service. + /// + /// default is "127.0.0.1:2379" + endpoints: Option<Vec<String>>, + /// whether using insecure connection to TiKV + insecure: bool, + /// certificate authority file path + ca_path: Option<String>, + /// cert path + cert_path: Option<String>, + /// key path + key_path: Option<String>, +} + +impl TiKVBuilder { + /// Set the network address of the TiKV service. + pub fn endpoints(&mut self, endpoints: impl Into<Vec<String>>) -> &mut Self { + let ep: Vec<String> = endpoints.into().into_iter().collect(); + if !ep.is_empty() { + self.endpoints = Some(ep) + } + self + } + + /// Set the insecure connection to TiKV. + pub fn insecure(&mut self) -> &mut Self { + self.insecure = false; + self + } + + /// Set the certificate authority file path. + pub fn ca_path(&mut self, ca_path: &str) -> &mut Self { + if !ca_path.is_empty() { + self.ca_path = Some(ca_path.to_string()) + } + self + } + + /// Set the certificate file path. + pub fn cert_path(&mut self, cert_path: &str) -> &mut Self { + if !cert_path.is_empty() { + self.cert_path = Some(cert_path.to_string()) + } + self + } + + /// Set the key file path. + pub fn key_path(&mut self, key_path: &str) -> &mut Self { + if !key_path.is_empty() { + self.key_path = Some(key_path.to_string()) + } + self + } +} + +impl Builder for TiKVBuilder { + const SCHEME: Scheme = Scheme::Redb; + type Accessor = Backend; + + fn from_map(map: HashMap<String, String>) -> Self { + let mut builder = TiKVBuilder::default(); + + map.get("endpoints") + .map(|v| v.split(',').map(|s| s.to_owned()).collect::<Vec<String>>()) + .map(|v| builder.endpoints(v)); + map.get("insecure") + .filter(|v| *v == "on" || *v == "true") + .map(|_| builder.insecure()); + map.get("ca_path").map(|v| builder.ca_path(v)); + map.get("cert_path").map(|v| builder.cert_path(v)); + map.get("key_path").map(|v| builder.key_path(v)); + + builder + } + + fn build(&mut self) -> Result<Self::Accessor> { + let endpoints = self.endpoints.take().ok_or_else(|| { + Error::new( + ErrorKind::ConfigInvalid, + "endpoints is required but not set", + ) + .with_context("service", Scheme::TiKV) + })?; + + if self.insecure { Review Comment: How about simplifying the logic by ignoring TLS configuration if `insecure` is set to true? ########## core/src/services/tikv/backend.rs: ########## @@ -0,0 +1,257 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; +use tikv_client::Config; +use tikv_client::RawClient; + +use crate::raw::adapters::kv; +use crate::Capability; +use crate::Scheme; +use async_trait::async_trait; + +use crate::Builder; +use crate::Error; +use crate::ErrorKind; +use crate::*; + +use std::fmt::Debug; +use std::fmt::Formatter; + +/// TiKV backend builder +#[derive(Clone, Default)] +pub struct TiKVBuilder { + /// network address of the TiKV service. + /// + /// default is "127.0.0.1:2379" + endpoints: Option<Vec<String>>, + /// whether using insecure connection to TiKV + insecure: bool, + /// certificate authority file path + ca_path: Option<String>, + /// cert path + cert_path: Option<String>, + /// key path + key_path: Option<String>, +} + +impl TiKVBuilder { + /// Set the network address of the TiKV service. + pub fn endpoints(&mut self, endpoints: impl Into<Vec<String>>) -> &mut Self { + let ep: Vec<String> = endpoints.into().into_iter().collect(); + if !ep.is_empty() { + self.endpoints = Some(ep) + } + self + } + + /// Set the insecure connection to TiKV. + pub fn insecure(&mut self) -> &mut Self { + self.insecure = false; + self + } + + /// Set the certificate authority file path. + pub fn ca_path(&mut self, ca_path: &str) -> &mut Self { + if !ca_path.is_empty() { + self.ca_path = Some(ca_path.to_string()) + } + self + } + + /// Set the certificate file path. + pub fn cert_path(&mut self, cert_path: &str) -> &mut Self { + if !cert_path.is_empty() { + self.cert_path = Some(cert_path.to_string()) + } + self + } + + /// Set the key file path. + pub fn key_path(&mut self, key_path: &str) -> &mut Self { + if !key_path.is_empty() { + self.key_path = Some(key_path.to_string()) + } + self + } +} + +impl Builder for TiKVBuilder { + const SCHEME: Scheme = Scheme::Redb; + type Accessor = Backend; + + fn from_map(map: HashMap<String, String>) -> Self { + let mut builder = TiKVBuilder::default(); + + map.get("endpoints") + .map(|v| v.split(',').map(|s| s.to_owned()).collect::<Vec<String>>()) + .map(|v| builder.endpoints(v)); + map.get("insecure") + .filter(|v| *v == "on" || *v == "true") + .map(|_| builder.insecure()); + map.get("ca_path").map(|v| builder.ca_path(v)); + map.get("cert_path").map(|v| builder.cert_path(v)); + map.get("key_path").map(|v| builder.key_path(v)); + + builder + } + + fn build(&mut self) -> Result<Self::Accessor> { + let endpoints = self.endpoints.take().ok_or_else(|| { + Error::new( + ErrorKind::ConfigInvalid, + "endpoints is required but not set", + ) + .with_context("service", Scheme::TiKV) + })?; + + if self.insecure { + if self.ca_path.is_some() || self.key_path.is_some() || self.cert_path.is_some() { + return Err( + Error::new(ErrorKind::ConfigInvalid, "invalid tls configuration") + .with_context("service", Scheme::TiKV) + .with_context("endpoints", format!("{:?}", endpoints)), + )?; + } + } else if !(self.ca_path.is_some() && self.key_path.is_some() && self.cert_path.is_some()) { + return Err( + Error::new(ErrorKind::ConfigInvalid, "invalid tls configuration") + .with_context("service", Scheme::TiKV) + .with_context("endpoints", format!("{:?}", endpoints)), + )?; + } + + Ok(Backend::new(Adapter { + client: None, + endpoints, + insecure: self.insecure, + ca_path: self.ca_path.clone(), + cert_path: self.cert_path.clone(), + key_path: self.key_path.clone(), + })) + } +} + +/// Backend for TiKV service +pub type Backend = kv::Backend<Adapter>; + +#[derive(Clone)] +pub struct Adapter { + client: Option<RawClient>, + endpoints: Vec<String>, + insecure: bool, + ca_path: Option<String>, + cert_path: Option<String>, + key_path: Option<String>, +} + +impl Debug for Adapter { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + let mut ds = f.debug_struct("Adapter"); + ds.field("endpoints", &self.endpoints); + ds.field("insecure", &self.insecure); + ds.field("ca_path", &self.ca_path); Review Comment: Please don't add those info in `Debug`. ########## core/src/services/tikv/backend.rs: ########## @@ -0,0 +1,257 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; +use tikv_client::Config; +use tikv_client::RawClient; + +use crate::raw::adapters::kv; +use crate::Capability; +use crate::Scheme; +use async_trait::async_trait; + +use crate::Builder; +use crate::Error; +use crate::ErrorKind; +use crate::*; + +use std::fmt::Debug; +use std::fmt::Formatter; + +/// TiKV backend builder +#[derive(Clone, Default)] +pub struct TiKVBuilder { + /// network address of the TiKV service. + /// + /// default is "127.0.0.1:2379" + endpoints: Option<Vec<String>>, + /// whether using insecure connection to TiKV + insecure: bool, + /// certificate authority file path + ca_path: Option<String>, + /// cert path + cert_path: Option<String>, + /// key path + key_path: Option<String>, +} + +impl TiKVBuilder { + /// Set the network address of the TiKV service. + pub fn endpoints(&mut self, endpoints: impl Into<Vec<String>>) -> &mut Self { + let ep: Vec<String> = endpoints.into().into_iter().collect(); + if !ep.is_empty() { + self.endpoints = Some(ep) + } + self + } + + /// Set the insecure connection to TiKV. + pub fn insecure(&mut self) -> &mut Self { + self.insecure = false; + self + } + + /// Set the certificate authority file path. + pub fn ca_path(&mut self, ca_path: &str) -> &mut Self { + if !ca_path.is_empty() { + self.ca_path = Some(ca_path.to_string()) + } + self + } + + /// Set the certificate file path. + pub fn cert_path(&mut self, cert_path: &str) -> &mut Self { + if !cert_path.is_empty() { + self.cert_path = Some(cert_path.to_string()) + } + self + } + + /// Set the key file path. + pub fn key_path(&mut self, key_path: &str) -> &mut Self { + if !key_path.is_empty() { + self.key_path = Some(key_path.to_string()) + } + self + } +} + +impl Builder for TiKVBuilder { + const SCHEME: Scheme = Scheme::Redb; + type Accessor = Backend; + + fn from_map(map: HashMap<String, String>) -> Self { + let mut builder = TiKVBuilder::default(); + + map.get("endpoints") + .map(|v| v.split(',').map(|s| s.to_owned()).collect::<Vec<String>>()) + .map(|v| builder.endpoints(v)); + map.get("insecure") + .filter(|v| *v == "on" || *v == "true") + .map(|_| builder.insecure()); + map.get("ca_path").map(|v| builder.ca_path(v)); + map.get("cert_path").map(|v| builder.cert_path(v)); + map.get("key_path").map(|v| builder.key_path(v)); + + builder + } + + fn build(&mut self) -> Result<Self::Accessor> { + let endpoints = self.endpoints.take().ok_or_else(|| { + Error::new( + ErrorKind::ConfigInvalid, + "endpoints is required but not set", + ) + .with_context("service", Scheme::TiKV) + })?; + + if self.insecure { + if self.ca_path.is_some() || self.key_path.is_some() || self.cert_path.is_some() { + return Err( + Error::new(ErrorKind::ConfigInvalid, "invalid tls configuration") + .with_context("service", Scheme::TiKV) + .with_context("endpoints", format!("{:?}", endpoints)), + )?; + } + } else if !(self.ca_path.is_some() && self.key_path.is_some() && self.cert_path.is_some()) { + return Err( + Error::new(ErrorKind::ConfigInvalid, "invalid tls configuration") + .with_context("service", Scheme::TiKV) + .with_context("endpoints", format!("{:?}", endpoints)), + )?; + } + + Ok(Backend::new(Adapter { + client: None, + endpoints, + insecure: self.insecure, + ca_path: self.ca_path.clone(), + cert_path: self.cert_path.clone(), + key_path: self.key_path.clone(), + })) + } +} + +/// Backend for TiKV service +pub type Backend = kv::Backend<Adapter>; + +#[derive(Clone)] +pub struct Adapter { + client: Option<RawClient>, + endpoints: Vec<String>, + insecure: bool, + ca_path: Option<String>, + cert_path: Option<String>, + key_path: Option<String>, +} + +impl Debug for Adapter { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + let mut ds = f.debug_struct("Adapter"); + ds.field("endpoints", &self.endpoints); + ds.field("insecure", &self.insecure); + ds.field("ca_path", &self.ca_path); + ds.field("cert_path", &self.cert_path); + ds.field("key_path", &self.key_path); + ds.finish() + } +} + +impl Adapter { + async fn get_connection(&self) -> Result<RawClient> { + if let Some(client) = self.client.clone() { Review Comment: self.client will always be `None`. ########## core/src/services/tikv/backend.rs: ########## @@ -0,0 +1,257 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; +use tikv_client::Config; +use tikv_client::RawClient; + +use crate::raw::adapters::kv; +use crate::Capability; +use crate::Scheme; +use async_trait::async_trait; + +use crate::Builder; +use crate::Error; +use crate::ErrorKind; +use crate::*; + +use std::fmt::Debug; +use std::fmt::Formatter; + +/// TiKV backend builder +#[derive(Clone, Default)] +pub struct TiKVBuilder { + /// network address of the TiKV service. + /// + /// default is "127.0.0.1:2379" + endpoints: Option<Vec<String>>, + /// whether using insecure connection to TiKV + insecure: bool, + /// certificate authority file path + ca_path: Option<String>, + /// cert path + cert_path: Option<String>, + /// key path + key_path: Option<String>, +} + +impl TiKVBuilder { + /// Set the network address of the TiKV service. + pub fn endpoints(&mut self, endpoints: impl Into<Vec<String>>) -> &mut Self { + let ep: Vec<String> = endpoints.into().into_iter().collect(); + if !ep.is_empty() { + self.endpoints = Some(ep) + } + self + } + + /// Set the insecure connection to TiKV. + pub fn insecure(&mut self) -> &mut Self { + self.insecure = false; + self + } + + /// Set the certificate authority file path. + pub fn ca_path(&mut self, ca_path: &str) -> &mut Self { + if !ca_path.is_empty() { + self.ca_path = Some(ca_path.to_string()) + } + self + } + + /// Set the certificate file path. + pub fn cert_path(&mut self, cert_path: &str) -> &mut Self { + if !cert_path.is_empty() { + self.cert_path = Some(cert_path.to_string()) + } + self + } + + /// Set the key file path. + pub fn key_path(&mut self, key_path: &str) -> &mut Self { + if !key_path.is_empty() { + self.key_path = Some(key_path.to_string()) + } + self + } +} + +impl Builder for TiKVBuilder { + const SCHEME: Scheme = Scheme::Redb; + type Accessor = Backend; + + fn from_map(map: HashMap<String, String>) -> Self { + let mut builder = TiKVBuilder::default(); + + map.get("endpoints") + .map(|v| v.split(',').map(|s| s.to_owned()).collect::<Vec<String>>()) + .map(|v| builder.endpoints(v)); + map.get("insecure") + .filter(|v| *v == "on" || *v == "true") + .map(|_| builder.insecure()); + map.get("ca_path").map(|v| builder.ca_path(v)); + map.get("cert_path").map(|v| builder.cert_path(v)); + map.get("key_path").map(|v| builder.key_path(v)); + + builder + } + + fn build(&mut self) -> Result<Self::Accessor> { + let endpoints = self.endpoints.take().ok_or_else(|| { + Error::new( + ErrorKind::ConfigInvalid, + "endpoints is required but not set", + ) + .with_context("service", Scheme::TiKV) + })?; + + if self.insecure { + if self.ca_path.is_some() || self.key_path.is_some() || self.cert_path.is_some() { + return Err( + Error::new(ErrorKind::ConfigInvalid, "invalid tls configuration") + .with_context("service", Scheme::TiKV) + .with_context("endpoints", format!("{:?}", endpoints)), + )?; + } + } else if !(self.ca_path.is_some() && self.key_path.is_some() && self.cert_path.is_some()) { + return Err( + Error::new(ErrorKind::ConfigInvalid, "invalid tls configuration") + .with_context("service", Scheme::TiKV) + .with_context("endpoints", format!("{:?}", endpoints)), + )?; + } + + Ok(Backend::new(Adapter { + client: None, + endpoints, + insecure: self.insecure, + ca_path: self.ca_path.clone(), + cert_path: self.cert_path.clone(), + key_path: self.key_path.clone(), + })) + } +} + +/// Backend for TiKV service +pub type Backend = kv::Backend<Adapter>; + +#[derive(Clone)] +pub struct Adapter { + client: Option<RawClient>, + endpoints: Vec<String>, + insecure: bool, + ca_path: Option<String>, + cert_path: Option<String>, + key_path: Option<String>, +} + +impl Debug for Adapter { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + let mut ds = f.debug_struct("Adapter"); + ds.field("endpoints", &self.endpoints); + ds.field("insecure", &self.insecure); + ds.field("ca_path", &self.ca_path); + ds.field("cert_path", &self.cert_path); + ds.field("key_path", &self.key_path); + ds.finish() + } +} + +impl Adapter { + async fn get_connection(&self) -> Result<RawClient> { + if let Some(client) = self.client.clone() { Review Comment: We can use `tokio::sync::OnceCell` here to make sure it init once. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
