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    /// Set the maximum size in bytes of a gRPC response that the etcd client decodes.
118    ///
119    /// default is None, which keeps the etcd client default of 4 MiB
120    pub fn max_decoding_message_size(mut self, size: usize) -> Self {
121        self.config.max_decoding_message_size = Some(size);
122        self
123    }
124}
125
126impl Builder for EtcdBuilder {
127    type Config = EtcdConfig;
128
129    fn build(self) -> Result<impl Service> {
130        let endpoints = self
131            .config
132            .endpoints
133            .clone()
134            .unwrap_or_else(|| DEFAULT_ETCD_ENDPOINTS.to_string());
135
136        let endpoints: Vec<String> = endpoints.split(',').map(|s| s.to_string()).collect();
137
138        let mut options = ConnectOptions::new();
139
140        if self.config.ca_path.is_some()
141            && self.config.cert_path.is_some()
142            && self.config.key_path.is_some()
143        {
144            let ca = self.load_pem(self.config.ca_path.clone().unwrap().as_str())?;
145            let key = self.load_pem(self.config.key_path.clone().unwrap().as_str())?;
146            let cert = self.load_pem(self.config.cert_path.clone().unwrap().as_str())?;
147
148            let tls_options = TlsOptions::default()
149                .ca_certificate(Certificate::from_pem(ca))
150                .identity(Identity::from_pem(cert, key));
151            options = options.with_tls(tls_options);
152        }
153
154        if let Some(username) = self.config.username.clone() {
155            options = options.with_user(
156                username,
157                self.config.password.clone().unwrap_or("".to_string()),
158            );
159        }
160
161        let root = normalize_root(
162            self.config
163                .root
164                .clone()
165                .unwrap_or_else(|| "/".to_string())
166                .as_str(),
167        );
168
169        let core = EtcdCore::new(endpoints, options, self.config.max_decoding_message_size);
170        Ok(EtcdBackend::new(core, &root))
171    }
172}
173
174impl EtcdBuilder {
175    fn load_pem(&self, path: &str) -> Result<String> {
176        std::fs::read_to_string(path)
177            .map_err(|err| Error::new(ErrorKind::Unexpected, "invalid file path").set_source(err))
178    }
179}
180
181#[derive(Debug, Clone)]
182pub struct EtcdBackend {
183    pub(crate) core: Arc<EtcdCore>,
184    pub(crate) info: ServiceInfo,
185    pub(crate) capability: Capability,
186}
187
188impl EtcdBackend {
189    fn new(core: EtcdCore, root: &str) -> Self {
190        let info = ServiceInfo::new(ETCD_SCHEME, root, "etcd");
191        let capability = Capability {
192            read: true,
193
194            write: true,
195            write_can_empty: true,
196
197            delete: true,
198            stat: true,
199            list: true,
200
201            shared: true,
202
203            ..Default::default()
204        };
205
206        Self {
207            core: Arc::new(core),
208            info,
209            capability,
210        }
211    }
212}
213
214impl Service for EtcdBackend {
215    type Reader = oio::StreamReader<EtcdReader>;
216    type Writer = EtcdWriter;
217    type Lister = oio::HierarchyLister<EtcdLazyLister>;
218    type Deleter = oio::OneShotDeleter<EtcdDeleter>;
219    type Copier = ();
220    type Composer = ();
221
222    fn info(&self) -> ServiceInfo {
223        self.info.clone()
224    }
225
226    fn capability(&self) -> Capability {
227        self.capability
228    }
229
230    async fn create_dir(
231        &self,
232        _ctx: &OperationContext,
233        path: &str,
234        _args: OpCreateDir,
235    ) -> Result<RpCreateDir> {
236        let abs_path = build_abs_path(&self.info.root(), path);
237
238        // In etcd, we simulate directory creation by storing an empty value
239        // with the directory path (ensuring it ends with '/')
240        let dir_path = if abs_path.ends_with('/') {
241            abs_path
242        } else {
243            format!("{abs_path}/")
244        };
245
246        // Store an empty buffer to represent the directory
247        self.core.set(&dir_path, Buffer::new()).await?;
248
249        Ok(RpCreateDir::default())
250    }
251
252    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
253        let abs_path = build_abs_path(&self.info.root(), path);
254
255        // First check if it's a direct key
256        match self.core.get(&abs_path).await? {
257            Some(buffer) => {
258                let metadata = if abs_path.ends_with('/') {
259                    MetadataBuilder::dir()
260                } else {
261                    MetadataBuilder::file(buffer.len() as u64)
262                };
263                Ok(RpStat::new(metadata.build()))
264            }
265            None => {
266                // Check if it's a directory by looking for keys with this prefix
267                let prefix = if abs_path.ends_with('/') {
268                    abs_path
269                } else {
270                    format!("{abs_path}/")
271                };
272
273                // Use etcd prefix query to check if any keys exist with this prefix
274                let has_children = self.core.has_prefix(&prefix).await?;
275                if has_children {
276                    // Has children, it's a directory
277                    let metadata = MetadataBuilder::dir().build();
278                    Ok(RpStat::new(metadata))
279                } else {
280                    Err(Error::new(ErrorKind::NotFound, "path not found"))
281                }
282            }
283        }
284    }
285    fn read(&self, _ctx: &OperationContext, path: &str, op: OpRead) -> Result<Self::Reader> {
286        let output: oio::StreamReader<EtcdReader> = {
287            Ok(oio::StreamReader::new(EtcdReader::new(
288                self.clone(),
289                path,
290                op,
291            )))
292        }?;
293
294        Ok(output)
295    }
296
297    fn write(&self, _ctx: &OperationContext, path: &str, _op: OpWrite) -> Result<Self::Writer> {
298        let output: EtcdWriter = {
299            let abs_path = build_abs_path(&self.info.root(), path);
300            let writer = EtcdWriter::new(self.core.clone(), abs_path);
301            Ok(writer)
302        }?;
303
304        Ok(output)
305    }
306
307    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
308        let output: oio::OneShotDeleter<EtcdDeleter> = {
309            let deleter = oio::OneShotDeleter::new(EtcdDeleter::new(
310                self.core.clone(),
311                self.info.root().to_string(),
312            ));
313            Ok(deleter)
314        }?;
315
316        Ok(output)
317    }
318
319    fn list(&self, _ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
320        let output: oio::HierarchyLister<EtcdLazyLister> = {
321            let lister = EtcdLazyLister::new(
322                self.core.clone(),
323                self.info.root().to_string(),
324                path.to_string(),
325            );
326            let lister = oio::HierarchyLister::new(lister, path, args.recursive());
327            Ok(lister)
328        }?;
329
330        Ok(output)
331    }
332
333    fn copy(
334        &self,
335        _ctx: &OperationContext,
336        _from: &str,
337        _to: &str,
338        _args: OpCopy,
339    ) -> Result<Self::Copier> {
340        Err(Error::new(
341            ErrorKind::Unsupported,
342            "operation is not supported",
343        ))
344    }
345
346    async fn rename(
347        &self,
348        _ctx: &OperationContext,
349        _from: &str,
350        _to: &str,
351        _args: OpRename,
352    ) -> Result<RpRename> {
353        Err(Error::new(
354            ErrorKind::Unsupported,
355            "operation is not supported",
356        ))
357    }
358
359    async fn presign(
360        &self,
361        _ctx: &OperationContext,
362        _path: &str,
363        _args: OpPresign,
364    ) -> Result<RpPresign> {
365        Err(Error::new(
366            ErrorKind::Unsupported,
367            "operation is not supported",
368        ))
369    }
370}
opendal_service_etcd/backend.rs

opendal_service_etcd/
backend.rs
