opendal/services/hdfs/
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::fmt::Debug;
19use std::fmt::Formatter;
20use std::io;
21use std::io::SeekFrom;
22use std::path::PathBuf;
23use std::sync::Arc;
24
25use log::debug;
26use uuid::Uuid;
27
28use super::delete::HdfsDeleter;
29use super::lister::HdfsLister;
30use super::reader::HdfsReader;
31use super::writer::HdfsWriter;
32use crate::raw::*;
33use crate::services::HdfsConfig;
34use crate::*;
35
36impl Configurator for HdfsConfig {
37    type Builder = HdfsBuilder;
38    fn into_builder(self) -> Self::Builder {
39        HdfsBuilder { config: self }
40    }
41}
42
43#[doc = include_str!("docs.md")]
44#[derive(Default)]
45pub struct HdfsBuilder {
46    config: HdfsConfig,
47}
48
49impl Debug for HdfsBuilder {
50    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
51        f.debug_struct("HdfsBuilder")
52            .field("config", &self.config)
53            .finish()
54    }
55}
56
57impl HdfsBuilder {
58    /// Set root of this backend.
59    ///
60    /// All operations will happen under this root.
61    pub fn root(mut self, root: &str) -> Self {
62        self.config.root = if root.is_empty() {
63            None
64        } else {
65            Some(root.to_string())
66        };
67
68        self
69    }
70
71    /// Set name_node of this backend.
72    ///
73    /// Valid format including:
74    ///
75    /// - `default`: using the default setting based on hadoop config.
76    /// - `hdfs://127.0.0.1:9000`: connect to hdfs cluster.
77    pub fn name_node(mut self, name_node: &str) -> Self {
78        if !name_node.is_empty() {
79            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
80            self.config.name_node = Some(name_node.trim_end_matches('/').to_string())
81        }
82
83        self
84    }
85
86    /// Set kerberos_ticket_cache_path of this backend
87    ///
88    /// This should be configured when kerberos is enabled.
89    pub fn kerberos_ticket_cache_path(mut self, kerberos_ticket_cache_path: &str) -> Self {
90        if !kerberos_ticket_cache_path.is_empty() {
91            self.config.kerberos_ticket_cache_path = Some(kerberos_ticket_cache_path.to_string())
92        }
93        self
94    }
95
96    /// Set user of this backend
97    pub fn user(mut self, user: &str) -> Self {
98        if !user.is_empty() {
99            self.config.user = Some(user.to_string())
100        }
101        self
102    }
103
104    /// Enable append capacity of this backend.
105    ///
106    /// This should be disabled when HDFS runs in non-distributed mode.
107    pub fn enable_append(mut self, enable_append: bool) -> Self {
108        self.config.enable_append = enable_append;
109        self
110    }
111
112    /// Set temp dir for atomic write.
113    ///
114    /// # Notes
115    ///
116    /// - When append is enabled, we will not use atomic write
117    ///   to avoid data loss and performance issue.
118    pub fn atomic_write_dir(mut self, dir: &str) -> Self {
119        self.config.atomic_write_dir = if dir.is_empty() {
120            None
121        } else {
122            Some(String::from(dir))
123        };
124        self
125    }
126}
127
128impl Builder for HdfsBuilder {
129    const SCHEME: Scheme = Scheme::Hdfs;
130    type Config = HdfsConfig;
131
132    fn build(self) -> Result<impl Access> {
133        debug!("backend build started: {:?}", &self);
134
135        let name_node = match &self.config.name_node {
136            Some(v) => v,
137            None => {
138                return Err(Error::new(ErrorKind::ConfigInvalid, "name node is empty")
139                    .with_context("service", Scheme::Hdfs))
140            }
141        };
142
143        let root = normalize_root(&self.config.root.unwrap_or_default());
144        debug!("backend use root {}", root);
145
146        let mut builder = hdrs::ClientBuilder::new(name_node);
147        if let Some(ticket_cache_path) = &self.config.kerberos_ticket_cache_path {
148            builder = builder.with_kerberos_ticket_cache_path(ticket_cache_path.as_str());
149        }
150        if let Some(user) = &self.config.user {
151            builder = builder.with_user(user.as_str());
152        }
153
154        let client = builder.connect().map_err(new_std_io_error)?;
155
156        // Create root dir if not exist.
157        if let Err(e) = client.metadata(&root) {
158            if e.kind() == io::ErrorKind::NotFound {
159                debug!("root {} is not exist, creating now", root);
160
161                client.create_dir(&root).map_err(new_std_io_error)?
162            }
163        }
164
165        let atomic_write_dir = self.config.atomic_write_dir;
166
167        // If atomic write dir is not exist, we must create it.
168        if let Some(d) = &atomic_write_dir {
169            if let Err(e) = client.metadata(d) {
170                if e.kind() == io::ErrorKind::NotFound {
171                    client.create_dir(d).map_err(new_std_io_error)?
172                }
173            }
174        }
175
176        Ok(HdfsBackend {
177            info: {
178                let am = AccessorInfo::default();
179                am.set_scheme(Scheme::Hdfs)
180                    .set_root(&root)
181                    .set_native_capability(Capability {
182                        stat: true,
183                        stat_has_content_length: true,
184                        stat_has_last_modified: true,
185
186                        read: true,
187
188                        write: true,
189                        write_can_append: self.config.enable_append,
190
191                        create_dir: true,
192                        delete: true,
193
194                        list: true,
195                        list_has_content_length: true,
196                        list_has_last_modified: true,
197
198                        rename: true,
199
200                        shared: true,
201
202                        ..Default::default()
203                    });
204
205                am.into()
206            },
207            root,
208            atomic_write_dir,
209            client: Arc::new(client),
210        })
211    }
212}
213
214#[inline]
215fn tmp_file_of(path: &str) -> String {
216    let name = get_basename(path);
217    let uuid = Uuid::new_v4().to_string();
218
219    format!("{name}.{uuid}")
220}
221
222/// Backend for hdfs services.
223#[derive(Debug, Clone)]
224pub struct HdfsBackend {
225    pub info: Arc<AccessorInfo>,
226    pub root: String,
227    atomic_write_dir: Option<String>,
228    pub client: Arc<hdrs::Client>,
229}
230
231/// hdrs::Client is thread-safe.
232unsafe impl Send for HdfsBackend {}
233unsafe impl Sync for HdfsBackend {}
234
235impl Access for HdfsBackend {
236    type Reader = HdfsReader<hdrs::AsyncFile>;
237    type Writer = HdfsWriter<hdrs::AsyncFile>;
238    type Lister = Option<HdfsLister>;
239    type Deleter = oio::OneShotDeleter<HdfsDeleter>;
240
241    fn info(&self) -> Arc<AccessorInfo> {
242        self.info.clone()
243    }
244
245    async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
246        let p = build_rooted_abs_path(&self.root, path);
247
248        self.client.create_dir(&p).map_err(new_std_io_error)?;
249
250        Ok(RpCreateDir::default())
251    }
252
253    async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
254        let p = build_rooted_abs_path(&self.root, path);
255
256        let meta = self.client.metadata(&p).map_err(new_std_io_error)?;
257
258        let mode = if meta.is_dir() {
259            EntryMode::DIR
260        } else if meta.is_file() {
261            EntryMode::FILE
262        } else {
263            EntryMode::Unknown
264        };
265        let mut m = Metadata::new(mode);
266        m.set_content_length(meta.len());
267        m.set_last_modified(meta.modified().into());
268
269        Ok(RpStat::new(m))
270    }
271
272    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
273        let p = build_rooted_abs_path(&self.root, path);
274
275        let client = self.client.clone();
276        let mut f = client
277            .open_file()
278            .read(true)
279            .async_open(&p)
280            .await
281            .map_err(new_std_io_error)?;
282
283        if args.range().offset() != 0 {
284            use futures::AsyncSeekExt;
285
286            f.seek(SeekFrom::Start(args.range().offset()))
287                .await
288                .map_err(new_std_io_error)?;
289        }
290
291        Ok((
292            RpRead::new(),
293            HdfsReader::new(f, args.range().size().unwrap_or(u64::MAX) as _),
294        ))
295    }
296
297    async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> {
298        let target_path = build_rooted_abs_path(&self.root, path);
299        let mut initial_size = 0;
300        let target_exists = match self.client.metadata(&target_path) {
301            Ok(meta) => {
302                initial_size = meta.len();
303                true
304            }
305            Err(err) => {
306                if err.kind() != io::ErrorKind::NotFound {
307                    return Err(new_std_io_error(err));
308                }
309                false
310            }
311        };
312
313        let should_append = op.append() && target_exists;
314        let tmp_path = self.atomic_write_dir.as_ref().and_then(|atomic_write_dir| {
315            // If the target file exists, we should append to the end of it directly.
316            if should_append {
317                None
318            } else {
319                Some(build_rooted_abs_path(atomic_write_dir, &tmp_file_of(path)))
320            }
321        });
322
323        if !target_exists {
324            let parent = get_parent(&target_path);
325            self.client.create_dir(parent).map_err(new_std_io_error)?;
326        }
327        if !should_append {
328            initial_size = 0;
329        }
330
331        let mut open_options = self.client.open_file();
332        open_options.create(true);
333        if should_append {
334            open_options.append(true);
335        } else {
336            open_options.write(true);
337        }
338
339        let f = open_options
340            .async_open(tmp_path.as_ref().unwrap_or(&target_path))
341            .await
342            .map_err(new_std_io_error)?;
343
344        Ok((
345            RpWrite::new(),
346            HdfsWriter::new(
347                target_path,
348                tmp_path,
349                f,
350                Arc::clone(&self.client),
351                target_exists,
352                initial_size,
353            ),
354        ))
355    }
356
357    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
358        Ok((
359            RpDelete::default(),
360            oio::OneShotDeleter::new(HdfsDeleter::new(Arc::new(self.clone()))),
361        ))
362    }
363
364    async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
365        let p = build_rooted_abs_path(&self.root, path);
366
367        let f = match self.client.read_dir(&p) {
368            Ok(f) => f,
369            Err(e) => {
370                return if e.kind() == io::ErrorKind::NotFound {
371                    Ok((RpList::default(), None))
372                } else {
373                    Err(new_std_io_error(e))
374                }
375            }
376        };
377
378        let rd = HdfsLister::new(&self.root, f, path);
379
380        Ok((RpList::default(), Some(rd)))
381    }
382
383    async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
384        let from_path = build_rooted_abs_path(&self.root, from);
385        self.client.metadata(&from_path).map_err(new_std_io_error)?;
386
387        let to_path = build_rooted_abs_path(&self.root, to);
388        let result = self.client.metadata(&to_path);
389        match result {
390            Err(err) => {
391                // Early return if other error happened.
392                if err.kind() != io::ErrorKind::NotFound {
393                    return Err(new_std_io_error(err));
394                }
395
396                let parent = PathBuf::from(&to_path)
397                    .parent()
398                    .ok_or_else(|| {
399                        Error::new(
400                            ErrorKind::Unexpected,
401                            "path should have parent but not, it must be malformed",
402                        )
403                        .with_context("input", &to_path)
404                    })?
405                    .to_path_buf();
406
407                self.client
408                    .create_dir(&parent.to_string_lossy())
409                    .map_err(new_std_io_error)?;
410            }
411            Ok(metadata) => {
412                if metadata.is_file() {
413                    self.client
414                        .remove_file(&to_path)
415                        .map_err(new_std_io_error)?;
416                } else {
417                    return Err(Error::new(ErrorKind::IsADirectory, "path should be a file")
418                        .with_context("input", &to_path));
419                }
420            }
421        }
422
423        self.client
424            .rename_file(&from_path, &to_path)
425            .map_err(new_std_io_error)?;
426
427        Ok(RpRename::new())
428    }
429}
opendal/services/hdfs/backend.rs

opendal/services/hdfs/
backend.rs
