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    type Composer = ();
213
214    fn info(&self) -> ServiceInfo {
215        self.info.clone()
216    }
217
218    fn capability(&self) -> Capability {
219        self.capability
220    }
221
222    async fn create_dir(
223        &self,
224        _ctx: &OperationContext,
225        path: &str,
226        _args: OpCreateDir,
227    ) -> Result<RpCreateDir> {
228        let abs_path = build_abs_path(&self.info.root(), path);
229
230        // In etcd, we simulate directory creation by storing an empty value
231        // with the directory path (ensuring it ends with '/')
232        let dir_path = if abs_path.ends_with('/') {
233            abs_path
234        } else {
235            format!("{abs_path}/")
236        };
237
238        // Store an empty buffer to represent the directory
239        self.core.set(&dir_path, Buffer::new()).await?;
240
241        Ok(RpCreateDir::default())
242    }
243
244    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
245        let abs_path = build_abs_path(&self.info.root(), path);
246
247        // First check if it's a direct key
248        match self.core.get(&abs_path).await? {
249            Some(buffer) => {
250                let metadata = if abs_path.ends_with('/') {
251                    MetadataBuilder::dir()
252                } else {
253                    MetadataBuilder::file(buffer.len() as u64)
254                };
255                Ok(RpStat::new(metadata.build()))
256            }
257            None => {
258                // Check if it's a directory by looking for keys with this prefix
259                let prefix = if abs_path.ends_with('/') {
260                    abs_path
261                } else {
262                    format!("{abs_path}/")
263                };
264
265                // Use etcd prefix query to check if any keys exist with this prefix
266                let has_children = self.core.has_prefix(&prefix).await?;
267                if has_children {
268                    // Has children, it's a directory
269                    let metadata = MetadataBuilder::dir().build();
270                    Ok(RpStat::new(metadata))
271                } else {
272                    Err(Error::new(ErrorKind::NotFound, "path not found"))
273                }
274            }
275        }
276    }
277    fn read(&self, _ctx: &OperationContext, path: &str, op: OpRead) -> Result<Self::Reader> {
278        let output: oio::StreamReader<EtcdReader> = {
279            Ok(oio::StreamReader::new(EtcdReader::new(
280                self.clone(),
281                path,
282                op,
283            )))
284        }?;
285
286        Ok(output)
287    }
288
289    fn write(&self, _ctx: &OperationContext, path: &str, _op: OpWrite) -> Result<Self::Writer> {
290        let output: EtcdWriter = {
291            let abs_path = build_abs_path(&self.info.root(), path);
292            let writer = EtcdWriter::new(self.core.clone(), abs_path);
293            Ok(writer)
294        }?;
295
296        Ok(output)
297    }
298
299    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
300        let output: oio::OneShotDeleter<EtcdDeleter> = {
301            let deleter = oio::OneShotDeleter::new(EtcdDeleter::new(
302                self.core.clone(),
303                self.info.root().to_string(),
304            ));
305            Ok(deleter)
306        }?;
307
308        Ok(output)
309    }
310
311    fn list(&self, _ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
312        let output: oio::HierarchyLister<EtcdLazyLister> = {
313            let lister = EtcdLazyLister::new(
314                self.core.clone(),
315                self.info.root().to_string(),
316                path.to_string(),
317            );
318            let lister = oio::HierarchyLister::new(lister, path, args.recursive());
319            Ok(lister)
320        }?;
321
322        Ok(output)
323    }
324
325    fn copy(
326        &self,
327        _ctx: &OperationContext,
328        _from: &str,
329        _to: &str,
330        _args: OpCopy,
331    ) -> Result<Self::Copier> {
332        Err(Error::new(
333            ErrorKind::Unsupported,
334            "operation is not supported",
335        ))
336    }
337
338    async fn rename(
339        &self,
340        _ctx: &OperationContext,
341        _from: &str,
342        _to: &str,
343        _args: OpRename,
344    ) -> Result<RpRename> {
345        Err(Error::new(
346            ErrorKind::Unsupported,
347            "operation is not supported",
348        ))
349    }
350
351    async fn presign(
352        &self,
353        _ctx: &OperationContext,
354        _path: &str,
355        _args: OpPresign,
356    ) -> Result<RpPresign> {
357        Err(Error::new(
358            ErrorKind::Unsupported,
359            "operation is not supported",
360        ))
361    }
362}
opendal_service_etcd/backend.rs

opendal_service_etcd/
backend.rs
