feat: Implement services webhdfs

.env.example

@@ -73,3 +73,10 @@ OPENDAL_MEMCACHED_ROOT=/
 OPENDAL_WEBDAV_TEST=false
 OPENDAL_WEBDAV_ROOT=/tmp/opendal/
 OPENDAL_WEBDAV_ENDPOINT=http://127.0.0.1:8080
+# webhdfs
+OPENDAL_WEBHDFS_TEST=false
+OPENDAL_WEBHDFS_ROOT=/tmp/opendal/
+OPENDAL_WEBHDFS_ENDPOINT=http://127.0.0.1:9870
+OPENDAL_WEBHDFS_DELEGATION=<delegation>
+OPENDAL_WEBHDFS_USER=<username>
+OPENDAL_WEBHDFS_DOAS=<proxy username>

.github/workflows/service_test_hdfs.yml → .github/workflows/service_test_hdfs_and_webhdfs.yml

@@ -1,4 +1,4 @@
-name: Service Test HDFS
+name: Service Test HDFS and WebHDFS
 
 on:
   push:
@@ -41,11 +41,16 @@ jobs:
       - name: Test
         shell: bash
         continue-on-error: true
-        run: cargo test hdfs --features compress,services-hdfs -- --show-output
+        run:
+          # this command will test both hdfs and webhdfs, make use of it
+          cargo test hdfs --features compress,services-hdfs -- --show-output
         env:
           RUST_BACKTRACE: full
           RUST_LOG: debug
           OPENDAL_HDFS_TEST: on
           OPENDAL_HDFS_ROOT: /
           OPENDAL_HDFS_NAME_NODE: hdfs://${{ env.HDFS_NAMENODE_ADDR }}
           LD_LIBRARY_PATH: ${{ env.JAVA_HOME }}/lib/server:${{ env.LD_LIBRARY_PATH }}
+          OPENDAL_WEBHDFS_TEST: on
+          OPENDAL_WEBHDFS_ROOT: /
+          OPENDAL_WEBHDFS_ENDPOINT: http://${{ env.WEBHDFS_NAMENODE_ADDR }}

README.md

@@ -33,6 +33,7 @@
 - [rocksdb](https://opendal.databend.rs/opendal/services/rocksdb/index.html): [RocksDB](http://rocksdb.org/) services support.
 - [s3](https://opendal.databend.rs/opendal/services/s3/index.html): [AWS S3](https://aws.amazon.com/s3/) alike services.
 - [webdav](https://opendal.databend.rs/opendal/services/webdav/index.html): [WebDAV](https://datatracker.ietf.org/doc/html/rfc4918) Service Support.
+- [webhdfs](https://opendal.databend.rs/opendal/services/webhdfs/index.html): [WebHDFS](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/WebHDFS.html) Service Support.
 
 ## Features
 

src/operator.rs

@@ -127,6 +127,7 @@ impl Operator {
         let op = match scheme {
             Scheme::Azblob => services::azblob::Builder::from_iter(it).build()?.into(),
             Scheme::Azdfs => services::azdfs::Builder::from_iter(it).build()?.into(),
+            Scheme::WebHdfs => services::webhdfs::Builder::from_iter(it).build()?.into(),
             Scheme::Fs => services::fs::Builder::from_iter(it).build()?.into(),
             #[cfg(feature = "services-ftp")]
             Scheme::Ftp => services::ftp::Builder::from_iter(it).build()?.into(),

src/raw/rps.rs

@@ -16,11 +16,11 @@ use http::Request;
 
 use crate::*;
 
-/// Reply fro `create` operation
+/// Reply for `create` operation
 #[derive(Debug, Clone, Default)]
 pub struct RpCreate {}
 
-/// Reply fro `delete` operation
+/// Reply for `delete` operation
 #[derive(Debug, Clone, Default)]
 pub struct RpDelete {}
 

src/scheme.rs

@@ -71,6 +71,8 @@ pub enum Scheme {
     S3,
     /// [oss][crate::services::oss]: Aliyun Object Storage Services
     Oss,
+    /// [webhdfs][crate::services::webhdfs]: WebHDFS RESTful API Services
+    WebHdfs,
     /// [webdav][crate::services::webdav]: WebDAV support.
     Webdav,
     /// Custom that allow users to implement services outside of OpenDAL.
@@ -132,6 +134,7 @@ impl FromStr for Scheme {
             "rocksdb" => Ok(Scheme::Rocksdb),
             "s3" => Ok(Scheme::S3),
             "oss" => Ok(Scheme::Oss),
+            "webhdfs" => Ok(Scheme::WebHdfs),
             "webdav" => Ok(Scheme::Webdav),
             _ => Ok(Scheme::Custom(Box::leak(s.into_boxed_str()))),
         }
@@ -166,6 +169,7 @@ impl From<Scheme> for &'static str {
             Scheme::Rocksdb => "service-rocksdb",
             Scheme::S3 => "s3",
             Scheme::Oss => "oss",
+            Scheme::WebHdfs => "webhdfs",
             Scheme::Webdav => "webdav",
             Scheme::Custom(v) => v,
         }

src/services/mod.rs

@@ -45,3 +45,4 @@ pub mod redis;
 pub mod rocksdb;
 pub mod s3;
 pub mod webdav;
+pub mod webhdfs;

src/services/webhdfs/auth.rs

@@ -0,0 +1,63 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed 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.
+
+//! This module contains some naive utilities
+//! for WebHDFS authentication.
+
+/// a naive WebHDFS auth implementation
+/// # Note
+/// WebHDFS supports using delegation token, or user and proxy user, for authentication.
+/// So we use an enum to represent the signer.
+#[derive(Clone, Debug)]
+pub(super) enum WebHdfsAuth {
+    /// Delegation token authentication
+    Token(String),
+    /// User name and proxy user authentication
+    User(String, String),
+}
+
+impl WebHdfsAuth {
+    /// create a new signer with delegation token
+    pub fn new_delegation(delegation_token: &str) -> Self {
+        Self::Token(delegation_token.to_string())
+    }
+    /// create a new signer with user name and proxy user
+    ///
+    /// # example
+    ///
+    /// Signer::new_user("user", "proxy_user") will make us login
+    /// with identity of "user" but do as "proxy_user".
+    ///
+    /// Signer::new_user("user", "") will make us login with identity
+    /// and on behave of nobody.
+    pub fn new_user(username: &str, doas: &str) -> Self {
+        Self::User(username.to_string(), doas.to_string())
+    }
+
+    /// sign a request
+    pub fn auth_str(&self) -> String {
+        match self {
+            Self::Token(token) => {
+                format!("delegation={token}")
+            }
+            Self::User(username, doas) => {
+                if !doas.is_empty() {
+                    format!("user.name={username}&doas={doas}")
+                } else {
+                    format!("user.name={username}")
+                }
+            }
+        }
+    }
+}