You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org> on 2023/07/31 08:38:44 UTC
[GitHub] [incubator-opendal] Murphy-OrangeMud opened a new pull request, #2735: feat(services/zookeeper): Add zookeeper backend
Murphy-OrangeMud opened a new pull request, #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735
Add Zookeeper backend skeleton.
An issue to be discussed is that Zookeeper supports plugin Acl. In this implementation the authentication is similar to redis using username+password digest and the access control is consistent throughout the session. Maybe we could apply acl at a finer grain in the future.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308483882
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
Review Comment:
name is used to host namespace for current service, like `bucket` for `s3`.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308485115
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
Review Comment:
Seems a good idea.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308483882
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
Review Comment:
name is used to host namespace for current service, like `bucket` for `s3`. Users can use this field to identify different instances. The meaning could be different for services.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [incubator-opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1346134295
##########
.env.example:
##########
@@ -118,6 +118,13 @@ OPENDAL_WASABI_BUCKET=<bucket>
OPENDAL_WASABI_ENDPOINT=<endpoint>
OPENDAL_WASABI_ACCESS_KEY_ID=<ak>
OPENDAL_WASABI_SECRET_ACCESS_KEY=<sk>
+# atomicserver
+OPENDAL_ATOMICSERVER_TEST=false
Review Comment:
Those changes should not be part of this PR. Maybe you need to merge or rebase your branch with upstream?
##########
.github/workflows/bindings_cpp.yml:
##########
@@ -0,0 +1,70 @@
+# 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.
+
+name: Bindings Cpp CI
Review Comment:
The same. Those changes make this PR hard to review.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [incubator-opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1747225757
Running `cargo clippy --features services-zookeeper` to make clippy happy.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2735:
URL: https://github.com/apache/opendal/pull/2735#issuecomment-2044133668
Hi, @Murphy-OrangeMud, thanks for your contribution first. We've made numerous updates to our Rust core, rendering this PR quite outdated. I'll be closing it but feel free to open a new one. I'm here to offer any help you might need.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1300878871
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,246 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
Review Comment:
Seems we can't recover from broken connection. How about using bb8 to do this? We can take `memecache` service as a reference.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308483882
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
Review Comment:
name is used to host namespace for current service, like `bucket` for `s3`. Users can use this field to identify different instances.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Murphy-OrangeMud commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1300624322
##########
.env.example:
##########
@@ -59,7 +59,7 @@ OPENDAL_IPFS_TEST=false
OPENDAL_IPFS_ROOT=/ipfs/Qmxxxxxxxx
OPENDAL_IPFS_ENDPOINT=http://localhost:8080
# redis
-OPENDAL_REDIS_TEST=false
+OPENDAL_REDIS_TEST=on
Review Comment:
Done
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] kezhuw commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "kezhuw (via GitHub)" <gi...@apache.org>.
kezhuw commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1311988268
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
Review Comment:
I have release version `0.6.0` which has `mkdir` to create nodes up to given path, just like `mkdir -p` in filesystem. Hope it could help. @Murphy-OrangeMud
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308488705
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
Review Comment:
Yes, we will record the first limitation by adding a new capability with a maximum file size.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Murphy-OrangeMud commented on pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1696077835
https://github.com/kezhuw/zookeeper-client-rust/issues/13
It seems that a test has newly been added and this pr failed the test. Tested with the sdk and has opened an issue here.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1692624858
> Currently I've fixed the bugs and the tests has passed.
Thanks a lot for your effort!
> The`cargo clippy` test failed due to weird reason because I think the suggestions themselves correspond to the original code.
Don't worry about this. Rust 1.72 released! So part of our code didn't pass the clippy.
> About the multi cluster endpoints issue currently I didn't find a neat way to implement it but personally
That's ok to implement in the future.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [incubator-opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1862107993
Hi @Murphy-OrangeMud, are you still on this? This PR has been open for a while, and I believe we can resolve conflicts and workaround some issue so that we can merge it first.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo closed pull request #2735: feat(services/zookeeper): Add zookeeper backend
URL: https://github.com/apache/opendal/pull/2735
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Murphy-OrangeMud commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1300627991
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
Review Comment:
I have tried but the client sdk only supports single endpoint/single cluster and maintaining multiple endpoints and get connections several times when setting or getting data seems a bit weird. Haven't come up with a neat way to implement this yet. I'll continue to try this way currently.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Murphy-OrangeMud commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1300628834
##########
Cargo.lock:
##########
@@ -5777,10 +5906,11 @@ dependencies = [
[[package]]
name = "time"
-version = "0.3.22"
+version = "0.3.25"
Review Comment:
OK I'll look into this.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] kezhuw commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "kezhuw (via GitHub)" <gi...@apache.org>.
kezhuw commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308379479
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
Review Comment:
@Xuanwo What `kv::Metadata::name` represent for ? I saw "memcached", "TiKV", but I also saw `self.addr.as_str()` and `&self.endpoints.join(",")`.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
Review Comment:
I saw [Error::set_temporary](https://github.com/apache/incubator-opendal/blob/54ffc88aea5f546873735e3fe7fb9c4baa1c6ace/core/src/types/error.rs#L389). I think it is a natural fit for `zk::Error::ConnectionLoss`.
Besides this, is it better to enrich the error a bit ?
@Xuanwo Is it better to shape `Error::new` to accept `message: impl Into<String>` ? I saw `Error::new(_, &format!(""))`.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
Review Comment:
`username` ? For consistent with `etcd`, `redis`, `webdav` and etc. .
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ ..Default::default()
Review Comment:
How about `delete` and `list`(e.g. `Adapter::scan` ? @Xuanwo ) ? It would be frustrated to create things with no way to delete/inspect them.
Comparing to kv services, ZooKeeper is more like traditional file system as its [nodes are organized as tree](https://github.com/apache/zookeeper/blob/bc0e61854ca362261d899295422ac14dd5d82e59/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataNode.java#L62) and has [`Stat` for each node](https://github.com/apache/zookeeper/blob/bc0e61854ca362261d899295422ac14dd5d82e59/zookeeper-jute/src/main/resources/zookeeper.jute#L28).
* For `delete`, it has same limitation as filesystem, `delete` will fail on non empty directory.
* For `list`, ZooKeeper does not work well with [large number of children](https://issues.apache.org/jira/browse/ZOOKEEPER-272). I guess it won't be an issue here.
PS: I saw tikv overrides `delete` but without `delete: true`.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
Review Comment:
I guess it is this form cause of recursion. I have a [similar version using loop](https://github.com/kezhuw/zookeeper-client-rust/blob/04385bf927c5dca6fc0957d6805b3e956d462a95/src/client/mod.rs#L918C14-L918C38). Hope it could help @Murphy-OrangeMud .
PS: I think I should port some convenient methods from java world.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
Review Comment:
It is traditional for ZooKeeper to reach a cluster using "host1,host2". I will write a test cluster in "zookeeper-client" crate to assert this, I was fearing to bootstrap a cluster in test case 😮💨 .
I saw similar from [etcd with `endpoints`](https://github.com/apache/incubator-opendal/blob/54ffc88aea5f546873735e3fe7fb9c4baa1c6ace/core/src/services/etcd/backend.rs#L47) and [redis with `cluster_endpoints`](https://github.com/apache/incubator-opendal/blob/54ffc88aea5f546873735e3fe7fb9c4baa1c6ace/core/src/services/redis/backend.rs#L53). A rebalanced dns should also work in nature.
* ZooKeeper connection string: https://github.com/apache/zookeeper/blob/bc0e61854ca362261d899295422ac14dd5d82e59/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java#L761
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
Review Comment:
Shall we document the limitations of this backend ?
ZooKeeper, as an aged CP system, does not work well as file backend though its node hierarchy is shaped like a tree. I list some of notables here for what I know.
1. [Data node size limitation is 1MiB](https://github.com/apache/zookeeper/blob/bc0e61854ca362261d899295422ac14dd5d82e59/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java#L1393).
2. Does not work well when `list` directory with [large number of children](https://issues.apache.org/jira/browse/ZOOKEEPER-272).
The first might be crucial.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
Review Comment:
Besides this, I think we should support two more options for custom.
1. Session timeout is [mandatory in Java world](https://github.com/apache/zookeeper/blob/bc0e61854ca362261d899295422ac14dd5d82e59/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java#L447) and is important enough to gain attention as session expired is a terminal state. All operations will fail after session expired.
2. `root`. I saw it in `etcd` and `redis`. It is also attractive to have in zookeeper. Though, caller can specify a `chroot` in connection string, but it might be less well-known and `chroot` are [supposed to be exist](https://issues.apache.org/jira/browse/KAFKA-404).
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let path = build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(path));
+ match self.get_connection().await?.get_data(&path).await {
+ Ok(data) => Ok(Some(data.0)),
+ Err(e) => match e {
+ zk::Error::NoNode => Ok(None),
+ _ => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ },
+ }
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ let path = build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(path));
+ match self
+ .get_connection()
+ .await?
+ .set_data(&path, value, None)
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ return self.create_nested_node(&path, value).await;
+ }
+ _ => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ },
+ }
+ }
+
+ async fn delete(&self, path: &str) -> Result<()> {
Review Comment:
Hmm, I saw this, but not `delete: true`. Do we need `rm -r` semantic here @Xuanwo ?
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
Review Comment:
It is traditional for ZooKeeper to reach a cluster using "host1,host2". I will write a test cluster in "zookeeper-client" crate to assert this, I was fearing to bootstrap a cluster in test case 😮💨 .
I saw similar from [etcd with `endpoints`](https://github.com/apache/incubator-opendal/blob/54ffc88aea5f546873735e3fe7fb9c4baa1c6ace/core/src/services/etcd/backend.rs#L47) and [redis with `cluster_endpoints`](https://github.com/apache/incubator-opendal/blob/54ffc88aea5f546873735e3fe7fb9c4baa1c6ace/core/src/services/redis/backend.rs#L53). A rebalanced dns should also work in nature.
* ZooKeeper connection string: https://github.com/apache/zookeeper/blob/bc0e61854ca362261d899295422ac14dd5d82e59/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java#L761
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1278988339
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
Review Comment:
Is it possible for a zookeeper to need access to multiple endpoints?
##########
.env.example:
##########
@@ -59,7 +59,7 @@ OPENDAL_IPFS_TEST=false
OPENDAL_IPFS_ROOT=/ipfs/Qmxxxxxxxx
OPENDAL_IPFS_ENDPOINT=http://localhost:8080
# redis
-OPENDAL_REDIS_TEST=false
+OPENDAL_REDIS_TEST=on
Review Comment:
Please don't touch other services examples.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint)
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user)
+ }
+ if let Some(password) = self.password.clone() {
+ ds.field("password", &password)
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT,
+ Some(endpoint) => endpoint,
+ };
+
+ Ok(Backend::new(ZkAdapter {
+ endpoint: endpoint.to_string(),
+ user: self.user.clone(),
+ password: self.password.clone(),
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+pub type Backend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ user: Option<String>,
+ password: Option<String>,
+ client: OnceCell<zk::Client>,
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("user", &self.user);
+ ds.field("password", &self.password);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
Review Comment:
Maybe we can use `tokio::sync::OnceCell`.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint)
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user)
+ }
+ if let Some(password) = self.password.clone() {
+ ds.field("password", &password)
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT,
+ Some(endpoint) => endpoint,
+ };
+
+ Ok(Backend::new(ZkAdapter {
+ endpoint: endpoint.to_string(),
+ user: self.user.clone(),
+ password: self.password.clone(),
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+pub type Backend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ user: Option<String>,
+ password: Option<String>,
+ client: OnceCell<zk::Client>,
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("user", &self.user);
+ ds.field("password", &self.password);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(endpoint).await {
+ Ok(client) => {
+ match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes();
+ client.auth(ZOOKEEPER_AUTH_SCHEME.to_string(), auth.to_vec());
+ }
+ _ => log::warn!("username and password isn't set, default use `anyone` acl"),
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e),
+ }
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ blocking: false,
Review Comment:
Capability with `false` doesn't need to list here.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
Review Comment:
OpenDAL's MSRV is 1.65, and `OnceCell` is added in `1.70`, so we can't use this.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint)
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user)
+ }
+ if let Some(password) = self.password.clone() {
+ ds.field("password", &password)
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT,
+ Some(endpoint) => endpoint,
+ };
+
+ Ok(Backend::new(ZkAdapter {
+ endpoint: endpoint.to_string(),
+ user: self.user.clone(),
+ password: self.password.clone(),
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+pub type Backend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ user: Option<String>,
+ password: Option<String>,
+ client: OnceCell<zk::Client>,
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("user", &self.user);
+ ds.field("password", &self.password);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(endpoint).await {
+ Ok(client) => {
+ match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes();
+ client.auth(ZOOKEEPER_AUTH_SCHEME.to_string(), auth.to_vec());
+ }
+ _ => log::warn!("username and password isn't set, default use `anyone` acl"),
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e),
+ }
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ blocking: false,
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ Ok(Some(self.get_connection().await?.get_data(path).0))
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ match self
Review Comment:
OpenDAL prefer to have reabable code. For example, it's more readable like the following:
```rust
let conn = self.get_connection().await?;
let resp = conn.set_data(...).await
```
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint)
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user)
+ }
+ if let Some(password) = self.password.clone() {
+ ds.field("password", &password)
Review Comment:
Please do not expose sensitive information in `Debug`.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint)
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user)
+ }
+ if let Some(password) = self.password.clone() {
+ ds.field("password", &password)
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT,
+ Some(endpoint) => endpoint,
+ };
+
+ Ok(Backend::new(ZkAdapter {
+ endpoint: endpoint.to_string(),
+ user: self.user.clone(),
+ password: self.password.clone(),
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+pub type Backend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ user: Option<String>,
+ password: Option<String>,
+ client: OnceCell<zk::Client>,
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("user", &self.user);
+ ds.field("password", &self.password);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(endpoint).await {
+ Ok(client) => {
+ match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes();
+ client.auth(ZOOKEEPER_AUTH_SCHEME.to_string(), auth.to_vec());
+ }
+ _ => log::warn!("username and password isn't set, default use `anyone` acl"),
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e),
+ }
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ blocking: false,
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ Ok(Some(self.get_connection().await?.get_data(path).0))
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ match self
+ .get_connection()
+ .await?
+ .set_data(path, value, None)
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => self.get_connection().await?.create(
+ path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, zk::Acl::creator_all()),
Review Comment:
I prefer to store a `zk::Acl` in backend directly instead of checking everytime.
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,214 @@
+// 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::cell::OnceCell;
+use std::collections::HashMap;
+use zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+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;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint)
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user)
+ }
+ if let Some(password) = self.password.clone() {
+ ds.field("password", &password)
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT,
+ Some(endpoint) => endpoint,
+ };
+
+ Ok(Backend::new(ZkAdapter {
+ endpoint: endpoint.to_string(),
+ user: self.user.clone(),
+ password: self.password.clone(),
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+pub type Backend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ user: Option<String>,
+ password: Option<String>,
+ client: OnceCell<zk::Client>,
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("user", &self.user);
+ ds.field("password", &self.password);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(endpoint).await {
+ Ok(client) => {
+ match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes();
+ client.auth(ZOOKEEPER_AUTH_SCHEME.to_string(), auth.to_vec());
+ }
+ _ => log::warn!("username and password isn't set, default use `anyone` acl"),
Review Comment:
OpenDAL prefer to use `warn!()` instead of `log::warn!()`. And using `debug` level here is enough, it's not a warning that users need to take action.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308486697
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let path = build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(path));
+ match self.get_connection().await?.get_data(&path).await {
+ Ok(data) => Ok(Some(data.0)),
+ Err(e) => match e {
+ zk::Error::NoNode => Ok(None),
+ _ => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ },
+ }
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ let path = build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(path));
+ match self
+ .get_connection()
+ .await?
+ .set_data(&path, value, None)
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ return self.create_nested_node(&path, value).await;
+ }
+ _ => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ },
+ }
+ }
+
+ async fn delete(&self, path: &str) -> Result<()> {
Review Comment:
> Do we need `rm -r` semantic here
`delete` in opendal is just `unlink path` and `rm -r` is handled by upper application logic.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1312466783
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
Review Comment:
So nice!
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1692629914
> Don't worry about this. Rust 1.72 released!
Addressed in #2927. Please merge with main and try again!
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1295345689
##########
Cargo.lock:
##########
@@ -5777,10 +5906,11 @@ dependencies = [
[[package]]
name = "time"
-version = "0.3.22"
+version = "0.3.25"
Review Comment:
Seems we are trying to bump time to latest version. Can we avoid this?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Murphy-OrangeMud commented on pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1692604580
Hi, so sorry for the delay😔. Currently I've fixed the bugs and the tests has passed. The`cargo clippy` test failed due to weird reason because I think the suggestions themselves correspond to the original code. About the multi cluster endpoints issue currently I didn't find a neat way to implement it but personally I think according to the zookeeper characteristics it might be better to directly open more than one session for more than one endpoints (each session/endpoint has its own auths).
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1305025626
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ eprintln!("{path}");
Review Comment:
Please remove this
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
Review Comment:
Is it normal to use `digest_path` to connect zookeeper? How about passing `password` directly?
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
Review Comment:
Maybe we can use `async fn create_nested_node()` directly instead of playing with `BoxFuture`?
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ eprintln!("{path}");
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path = "/".to_string() + path.strip_suffix('/').unwrap_or(&path);
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let path = "/".to_string() + path.strip_suffix('/').unwrap_or(path);
Review Comment:
We can use https://opendal.apache.org/docs/rust/opendal/raw/fn.build_rooted_abs_path.html
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ eprintln!("{path}");
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path = "/".to_string() + path.strip_suffix('/').unwrap_or(&path);
+ }
+ match self.create_nested_node(&path, value).await {
Review Comment:
If it's required to create parent node first for zookeepr, I perfer to check if parent exist first instead handling error.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1308489147
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the password file of the user to connect to zookeeper service, default None
+ password: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the password of zookeeper service
+ pub fn password(&mut self, password: &str) -> &mut Self {
+ if !password.is_empty() {
+ self.password = Some(password.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("password").map(|v| builder.password(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.password.clone()) {
+ (Some(user), Some(password)) => {
+ let auth = format!("{user}:{password}").as_bytes().to_vec();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path =
+ build_rooted_abs_path("/", path.strip_suffix('/').unwrap_or(&path));
+ }
+ match self.create_nested_node(&path, value).await {
+ Ok(()) => {}
+ Err(e) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "error from zookeeper",
+ )
+ .set_source(e))
+ }
+ };
+ match self
+ .get_connection()
+ .await?
+ .create(
+ &tmpath,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper")
+ .set_source(e))
+ }
+ }
+ }
+ _ => {
+ Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e))
+ }
+ },
+ };
+ }
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for ZkAdapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::Zookeeper,
+ "ZooKeeper",
+ Capability {
+ read: true,
+ write: true,
+ ..Default::default()
Review Comment:
I'm guessing the `delete` is just missed here.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Murphy-OrangeMud commented on a diff in pull request #2735: feat(services/zookeeper): Add zookeeper backend
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1306839321
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,305 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::io;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
+ /// the path to the password file of the user to connect to zookeeper service, default None
+ digest_path: Option<String>,
+}
+
+impl ZookeeperBuilder {
+ /// Set the network addresses of zookeeper service
+ pub fn endpoint(&mut self, endpoint: &str) -> &mut Self {
+ if !endpoint.is_empty() {
+ self.endpoint = Some(endpoint.to_string());
+ }
+ self
+ }
+
+ /// Set the username of zookeeper service
+ pub fn user(&mut self, user: &str) -> &mut Self {
+ if !user.is_empty() {
+ self.user = Some(user.to_string());
+ }
+ self
+ }
+
+ /// Specify the auth digest path of zookeeper service
+ pub fn digest_path(&mut self, digest_path: &str) -> &mut Self {
+ if !digest_path.is_empty() {
+ self.digest_path = Some(digest_path.to_string());
+ }
+ self
+ }
+}
+
+impl Debug for ZookeeperBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Builder");
+ if let Some(endpoint) = self.endpoint.clone() {
+ ds.field("endpoint", &endpoint);
+ }
+ if let Some(user) = self.user.clone() {
+ ds.field("user", &user);
+ }
+ if let Some(digest_path) = self.digest_path.clone() {
+ ds.field("digest_path", &digest_path);
+ }
+ ds.finish()
+ }
+}
+
+impl Builder for ZookeeperBuilder {
+ const SCHEME: Scheme = Scheme::Zookeeper;
+ type Accessor = ZookeeperBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = ZookeeperBuilder::default();
+
+ map.get("endpoint").map(|v| builder.endpoint(v));
+ map.get("user").map(|v| builder.user(v));
+ map.get("digest_path").map(|v| builder.digest_path(v));
+
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let endpoint = match self.endpoint.clone() {
+ None => DEFAULT_ZOOKEEPER_ENDPOINT.to_string(),
+ Some(endpoint) => endpoint,
+ };
+ let (auth, acl) = match (self.user.clone(), self.digest_path.clone()) {
+ (Some(user), Some(digest_path)) => {
+ let password = std::fs::read(digest_path).map_err(parse_file_error)?;
+ let auth = [format!("{user}:").as_bytes().to_vec(), password].concat();
+ (auth, zk::Acl::creator_all())
+ }
+ _ => {
+ // TODO: change to warn
+ warn!("username and password isn't set, default use `anyone` acl");
+ (Vec::<u8>::new(), zk::Acl::anyone_all())
+ }
+ };
+
+ Ok(ZookeeperBackend::new(ZkAdapter {
+ endpoint,
+ auth,
+ acl,
+ client: OnceCell::new(),
+ }))
+ }
+}
+
+/// Backend for Zookeeper service
+pub type ZookeeperBackend = kv::Backend<ZkAdapter>;
+
+#[derive(Clone)]
+pub struct ZkAdapter {
+ endpoint: String,
+ auth: Vec<u8>,
+ client: OnceCell<zk::Client>,
+ acl: &'static [zk::Acl],
+}
+
+impl Debug for ZkAdapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut ds = f.debug_struct("Adapter");
+ ds.field("endpoint", &self.endpoint);
+ ds.field("acl", &self.acl);
+ ds.finish()
+ }
+}
+
+impl ZkAdapter {
+ async fn get_connection(&self) -> Result<zk::Client> {
+ if let Some(client) = self.client.get() {
+ return Ok(client.clone());
+ }
+ match zk::Client::connect(&self.endpoint.clone()).await {
+ Ok(client) => {
+ if !self.auth.is_empty() {
+ client
+ .auth(ZOOKEEPER_AUTH_SCHEME.to_string(), self.auth.clone())
+ .await
+ .map_err(parse_zookeeper_error)?;
+ }
+ self.client.set(client.clone()).ok();
+ Ok(client)
+ }
+ Err(e) => Err(Error::new(ErrorKind::Unexpected, "error from zookeeper").set_source(e)),
+ }
+ }
+
+ fn create_nested_node<'a>(
+ &'a self,
+ path: &'a str,
+ value: &'a [u8],
+ ) -> BoxFuture<'a, Result<()>> {
+ eprintln!("{path}");
+ let mut path = path.to_string();
+ async move {
+ return match self
+ .get_connection()
+ .await?
+ .create(
+ &path,
+ value,
+ &zk::CreateOptions::new(zk::CreateMode::Persistent, self.acl),
+ )
+ .await
+ {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ zk::Error::NoNode => {
+ let idx = path.rfind('/').unwrap();
+ let tmpath = path.clone();
+ path = path.to_string().substring(0, idx).to_string();
+ if path.as_bytes()[0] != b'/' {
+ path = "/".to_string() + path.strip_suffix('/').unwrap_or(&path);
+ }
+ match self.create_nested_node(&path, value).await {
Review Comment:
Yes, `create_nested_node()` is a recursive function that creates parent node recursively till a node exists. So when we find that the parent node doesn't exit (the NoNode error in zookeeper-client indicates this) we call this function to create the parent nodes till all the parent nodes on the path in the tree has created. Also it's because the fact that the function is both recursive and async that we use `BoxFuture`.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [incubator-opendal]
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on code in PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#discussion_r1348116126
##########
core/src/services/zookeeper/backend.rs:
##########
@@ -0,0 +1,297 @@
+// 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 zookeeper_client as zk;
+
+use crate::raw::adapters::kv;
+use crate::Scheme;
+use async_trait::async_trait;
+use tokio::sync::OnceCell;
+
+use crate::raw::build_rooted_abs_path;
+use crate::Builder;
+use crate::Error;
+use crate::ErrorKind;
+use crate::*;
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use futures::future::{BoxFuture, FutureExt};
+use substring::Substring;
+
+use log::warn;
+
+const DEFAULT_ZOOKEEPER_ENDPOINT: &str = "127.0.0.1:2181";
+/// The scheme for zookeeper authentication
+/// currently we do not support sasl authentication
+const ZOOKEEPER_AUTH_SCHEME: &str = "digest";
+
+/// Zookeeper backend builder
+#[derive(Clone, Default)]
+pub struct ZookeeperBuilder {
+ /// network address of the Zookeeper service
+ /// Default: 127.0.0.1:2181
+ endpoint: Option<String>,
+ /// the user to connect to zookeeper service, default None
+ user: Option<String>,
Review Comment:
Modified
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [incubator-opendal]
Posted by "Murphy-OrangeMud (via GitHub)" <gi...@apache.org>.
Murphy-OrangeMud commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1747847011
<img width="510" alt="image" src="https://github.com/apache/incubator-opendal/assets/46390322/690ce0d1-9e71-479a-8f24-85adae9ece7c">
It seems that this test （`test_copy_file_with_non_ascii_name`） will always fail. This is the screenshot using zookeeper commandline client with local zookeeper clusters.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/zookeeper): Add zookeeper backend [incubator-opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2735:
URL: https://github.com/apache/incubator-opendal/pull/2735#issuecomment-1748069682
> test_copy_file_with_non_ascii_name
You can skip this test for now like what we do for supabase.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org