opendal_service_etcd/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::sync::Arc;
19
20use etcd_client::Certificate;
21use etcd_client::ConnectOptions;
22use etcd_client::Identity;
23use etcd_client::TlsOptions;
24
25use super::ETCD_SCHEME;
26use super::config::EtcdConfig;
27use super::core::EtcdCore;
28use super::core::constants::DEFAULT_ETCD_ENDPOINTS;
29use super::deleter::EtcdDeleter;
30use super::lister::EtcdLazyLister;
31use super::reader::*;
32use super::writer::EtcdWriter;
33use opendal_core::raw::*;
34use opendal_core::*;
35
36/// [Etcd](https://etcd.io/) services support.
37#[doc = include_str!("docs.md")]
38#[derive(Debug, Default)]
39pub struct EtcdBuilder {
40    pub(super) config: EtcdConfig,
41}
42
43impl EtcdBuilder {
44    /// set the network address of etcd service.
45    ///
46    /// default: "http://127.0.0.1:2379"
47    pub fn endpoints(mut self, endpoints: &str) -> Self {
48        if !endpoints.is_empty() {
49            self.config.endpoints = Some(endpoints.to_owned());
50        }
51        self
52    }
53
54    /// set the username for etcd
55    ///
56    /// default: no username
57    pub fn username(mut self, username: &str) -> Self {
58        if !username.is_empty() {
59            self.config.username = Some(username.to_owned());
60        }
61        self
62    }
63
64    /// set the password for etcd
65    ///
66    /// default: no password
67    pub fn password(mut self, password: &str) -> Self {
68        if !password.is_empty() {
69            self.config.password = Some(password.to_owned());
70        }
71        self
72    }
73
74    /// set the working directory, all operations will be performed under it.
75    ///
76    /// default: "/"
77    pub fn root(mut self, root: &str) -> Self {
78        self.config.root = if root.is_empty() {
79            None
80        } else {
81            Some(root.to_string())
82        };
83
84        self
85    }
86
87    /// Set the certificate authority file path.
88    ///
89    /// default is None
90    pub fn ca_path(mut self, ca_path: &str) -> Self {
91        if !ca_path.is_empty() {
92            self.config.ca_path = Some(ca_path.to_string())
93        }
94        self
95    }
96
97    /// Set the certificate file path.
98    ///
99    /// default is None
100    pub fn cert_path(mut self, cert_path: &str) -> Self {
101        if !cert_path.is_empty() {
102            self.config.cert_path = Some(cert_path.to_string())
103        }
104        self
105    }
106
107    /// Set the key file path.
108    ///
109    /// default is None
110    pub fn key_path(mut self, key_path: &str) -> Self {
111        if !key_path.is_empty() {
112            self.config.key_path = Some(key_path.to_string())
113        }
114        self
115    }
116}
117
118impl Builder for EtcdBuilder {
119    type Config = EtcdConfig;
120
121    fn build(self) -> Result<impl Service> {
122        let endpoints = self
123            .config
124            .endpoints
125            .clone()
126            .unwrap_or_else(|| DEFAULT_ETCD_ENDPOINTS.to_string());
127
128        let endpoints: Vec<String> = endpoints.split(',').map(|s| s.to_string()).collect();
129
130        let mut options = ConnectOptions::new();
131
132        if self.config.ca_path.is_some()
133            && self.config.cert_path.is_some()
134            && self.config.key_path.is_some()
135        {
136            let ca = self.load_pem(self.config.ca_path.clone().unwrap().as_str())?;
137            let key = self.load_pem(self.config.key_path.clone().unwrap().as_str())?;
138            let cert = self.load_pem(self.config.cert_path.clone().unwrap().as_str())?;
139
140            let tls_options = TlsOptions::default()
141                .ca_certificate(Certificate::from_pem(ca))
142                .identity(Identity::from_pem(cert, key));
143            options = options.with_tls(tls_options);
144        }
145
146        if let Some(username) = self.config.username.clone() {
147            options = options.with_user(
148                username,
149                self.config.password.clone().unwrap_or("".to_string()),
150            );
151        }
152
153        let root = normalize_root(
154            self.config
155                .root
156                .clone()
157                .unwrap_or_else(|| "/".to_string())
158                .as_str(),
159        );
160
161        let core = EtcdCore::new(endpoints, options);
162        Ok(EtcdBackend::new(core, &root))
163    }
164}
165
166impl EtcdBuilder {
167    fn load_pem(&self, path: &str) -> Result<String> {
168        std::fs::read_to_string(path)
169            .map_err(|err| Error::new(ErrorKind::Unexpected, "invalid file path").set_source(err))
170    }
171}
172
173#[derive(Debug, Clone)]
174pub struct EtcdBackend {
175    pub(crate) core: Arc<EtcdCore>,
176    pub(crate) info: ServiceInfo,
177    pub(crate) capability: Capability,
178}
179
180impl EtcdBackend {
181    fn new(core: EtcdCore, root: &str) -> Self {
182        let info = ServiceInfo::new(ETCD_SCHEME, root, "etcd");
183        let capability = Capability {
184            read: true,
185
186            write: true,
187            write_can_empty: true,
188
189            delete: true,
190            stat: true,
191            list: true,
192
193            shared: true,
194
195            ..Default::default()
196        };
197
198        Self {
199            core: Arc::new(core),
200            info,
201            capability,
202        }
203    }
204}
205
206impl Service for EtcdBackend {
207    type Reader = oio::StreamReader<EtcdReader>;
208    type Writer = EtcdWriter;
209    type Lister = oio::HierarchyLister<EtcdLazyLister>;
210    type Deleter = oio::OneShotDeleter<EtcdDeleter>;
211    type Copier = ();
212
213    fn info(&self) -> ServiceInfo {
214        self.info.clone()
215    }
216
217    fn capability(&self) -> Capability {
218        self.capability
219    }
220
221    async fn create_dir(
222        &self,
223        _ctx: &OperationContext,
224        path: &str,
225        _args: OpCreateDir,
226    ) -> Result<RpCreateDir> {
227        let abs_path = build_abs_path(&self.info.root(), path);
228
229        // In etcd, we simulate directory creation by storing an empty value
230        // with the directory path (ensuring it ends with '/')
231        let dir_path = if abs_path.ends_with('/') {
232            abs_path
233        } else {
234            format!("{abs_path}/")
235        };
236
237        // Store an empty buffer to represent the directory
238        self.core.set(&dir_path, Buffer::new()).await?;
239
240        Ok(RpCreateDir::default())
241    }
242
243    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
244        let abs_path = build_abs_path(&self.info.root(), path);
245
246        // First check if it's a direct key
247        match self.core.get(&abs_path).await? {
248            Some(buffer) => {
249                let mut metadata = Metadata::new(EntryMode::from_path(&abs_path));
250                metadata.set_content_length(buffer.len() as u64);
251                Ok(RpStat::new(metadata))
252            }
253            None => {
254                // Check if it's a directory by looking for keys with this prefix
255                let prefix = if abs_path.ends_with('/') {
256                    abs_path
257                } else {
258                    format!("{abs_path}/")
259                };
260
261                // Use etcd prefix query to check if any keys exist with this prefix
262                let has_children = self.core.has_prefix(&prefix).await?;
263                if has_children {
264                    // Has children, it's a directory
265                    let metadata = Metadata::new(EntryMode::DIR);
266                    Ok(RpStat::new(metadata))
267                } else {
268                    Err(Error::new(ErrorKind::NotFound, "path not found"))
269                }
270            }
271        }
272    }
273    fn read(&self, _ctx: &OperationContext, path: &str, op: OpRead) -> Result<Self::Reader> {
274        let output: oio::StreamReader<EtcdReader> = {
275            Ok(oio::StreamReader::new(EtcdReader::new(
276                self.clone(),
277                path,
278                op,
279            )))
280        }?;
281
282        Ok(output)
283    }
284
285    fn write(&self, _ctx: &OperationContext, path: &str, _op: OpWrite) -> Result<Self::Writer> {
286        let output: EtcdWriter = {
287            let abs_path = build_abs_path(&self.info.root(), path);
288            let writer = EtcdWriter::new(self.core.clone(), abs_path);
289            Ok(writer)
290        }?;
291
292        Ok(output)
293    }
294
295    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
296        let output: oio::OneShotDeleter<EtcdDeleter> = {
297            let deleter = oio::OneShotDeleter::new(EtcdDeleter::new(
298                self.core.clone(),
299                self.info.root().to_string(),
300            ));
301            Ok(deleter)
302        }?;
303
304        Ok(output)
305    }
306
307    fn list(&self, _ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
308        let output: oio::HierarchyLister<EtcdLazyLister> = {
309            let lister = EtcdLazyLister::new(
310                self.core.clone(),
311                self.info.root().to_string(),
312                path.to_string(),
313            );
314            let lister = oio::HierarchyLister::new(lister, path, args.recursive());
315            Ok(lister)
316        }?;
317
318        Ok(output)
319    }
320
321    fn copy(
322        &self,
323        _ctx: &OperationContext,
324        _from: &str,
325        _to: &str,
326        _args: OpCopy,
327        _opts: OpCopier,
328    ) -> Result<Self::Copier> {
329        Err(Error::new(
330            ErrorKind::Unsupported,
331            "operation is not supported",
332        ))
333    }
334
335    async fn rename(
336        &self,
337        _ctx: &OperationContext,
338        _from: &str,
339        _to: &str,
340        _args: OpRename,
341    ) -> Result<RpRename> {
342        Err(Error::new(
343            ErrorKind::Unsupported,
344            "operation is not supported",
345        ))
346    }
347
348    async fn presign(
349        &self,
350        _ctx: &OperationContext,
351        _path: &str,
352        _args: OpPresign,
353    ) -> Result<RpPresign> {
354        Err(Error::new(
355            ErrorKind::Unsupported,
356            "operation is not supported",
357        ))
358    }
359}
opendal_service_etcd/backend.rs

opendal_service_etcd/
backend.rs
