Add option to set config dir to load from (#284)

Author: Lostlitus

README.md

@@ -9,9 +9,11 @@
 `hdfs-native` is an HDFS client written natively in Rust. It supports nearly all major features of an HDFS client, and several key client configuration options listed below.
 
 ## Supported HDFS features
+
 Here is a list of currently supported and unsupported but possible future features.
 
 ### HDFS Operations
+
 - [x] Listing
 - [x] Reading
 - [x] Writing
@@ -24,14 +26,16 @@ Here is a list of currently supported and unsupported but possible future featur
 - [x] Set timestamps
 
 ### HDFS Features
+
 - [x] Name Services
 - [x] Observer reads
 - [x] ViewFS
 - [x] Router based federation
 - [x] Erasure coded reads and writes
-    - RS schema only, no support for RS-Legacy or XOR
+  - RS schema only, no support for RS-Legacy or XOR
 
 ### Security Features
+
 - [x] Kerberos authentication (GSSAPI SASL support) (requires libgssapi_krb5, see below)
 - [x] Token authentication (DIGEST-MD5 SASL support)
 - [x] NameNode SASL connection
@@ -40,47 +44,54 @@ Here is a list of currently supported and unsupported but possible future featur
 - [ ] Encryption at rest (KMS support)
 
 ### Kerberos Support
+
 Kerberos (SASL GSSAPI) mechanism is supported through a runtime dynamic link to `libgssapi_krb5`. This must be installed separately, but is likely already installed on your system. If not you can install it by:
 
 #### Debian-based systems
+
 ```bash
 apt-get install libgssapi-krb5-2
 ```
 
 #### RHEL-based systems
+
 ```bash
 yum install krb5-libs
 ```
 
 #### MacOS
+
 ```bash
 brew install krb5
 ```
 
 #### Windows
+
 Download and install the Microsoft Kerberos package from https://web.mit.edu/kerberos/dist/
 
 Copy the `<INSTALL FOLDER>\MIT\Kerberos\bin\gssapi64.dll` file to a folder in %PATH% and change the name to `gssapi_krb5.dll`
 
 ## Supported HDFS Settings
-The client will attempt to read Hadoop configs `core-site.xml` and `hdfs-site.xml` in the directories `$HADOOP_CONF_DIR` or if that doesn't exist, `$HADOOP_HOME/etc/hadoop`. Currently the supported configs that are used are:
+
+The client will attempt to read Hadoop configs `core-site.xml` and `hdfs-site.xml` in the directories `$HADOOP_CONF_DIR` or if that doesn't exist, `$HADOOP_HOME/etc/hadoop`. Passing configs in run time is supported as well via `client::ClientBuilder`. Currently the supported configs that are used are:
+
 - `fs.defaultFS` - Client::default() support
 - `dfs.ha.namenodes` - name service support
 - `dfs.namenode.rpc-address.*` - name service support
 - `dfs.client.failover.resolve-needed.*` - DNS based NameNode discovery
 - `dfs.client.failover.resolver.useFQDN.*` - DNS based NameNode discovery
 - `dfs.client.failover.random.order.*` - Randomize order of NameNodes to try
 - `dfs.client.failover.proxy.provider.*` - Supports the behavior of the following proxy providers. Any other values will default back to the `ConfiguredFailoverProxyProvider` behavior:
-    - `org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider`
-    - `org.apache.hadoop.hdfs.server.namenode.ha.ObserverReadProxyProvider`
-    - `org.apache.hadoop.hdfs.server.namenode.ha.RouterObserverReadConfiguredFailoverProxyProvider`
+  - `org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider`
+  - `org.apache.hadoop.hdfs.server.namenode.ha.ObserverReadProxyProvider`
+  - `org.apache.hadoop.hdfs.server.namenode.ha.RouterObserverReadConfiguredFailoverProxyProvider`
 - `dfs.client.block.write.replace-datanode-on-failure.enable`
 - `dfs.client.block.write.replace-datanode-on-failure.policy`
 - `dfs.client.block.write.replace-datanode-on-failure.best-effort`
 - `fs.viewfs.mounttable.*.link.*` - ViewFS links
 - `fs.viewfs.mounttable.*.linkFallback` - ViewFS link fallback
 
-All other settings are generally assumed to be the defaults currently. For instance, security is assumed to be enabled and SASL negotiation is always done, but on insecure clusters this will just do SIMPLE authentication. Any setups that require other customized Hadoop client configs may not work correctly. 
+All other settings are generally assumed to be the defaults currently. For instance, security is assumed to be enabled and SASL negotiation is always done, but on insecure clusters this will just do SIMPLE authentication. Any setups that require other customized Hadoop client configs may not work correctly.
 
 ## Building
 
@@ -89,19 +100,23 @@ cargo build
 ```
 
 ## Object store implementation
+
 An object_store implementation for HDFS is provided in the [hdfs-native-object-store](https://github.com/datafusion-contrib/hdfs-native-object-store) crate.
 
 ## Running tests
+
 The tests are mostly integration tests that utilize a small Java application in `rust/mindifs/` that runs a custom `MiniDFSCluster`. To run the tests, you need to have Java, Maven, Hadoop binaries, and Kerberos tools available and on your path. Any Java version between 8 and 17 should work.
 
 ```bash
-cargo test -p hdfs-native --features intergation-test
+cargo test -p hdfs-native --features integration-test
 ```
 
 ### Python tests
+
 See the [Python README](./python/README.md)
 
 ## Running benchmarks
+
 Some of the benchmarks compare performance to the JVM based client through libhdfs via the fs-hdfs3 crate. Because of that, some extra setup is required to run the benchmarks:
 
 ```bash
@@ -110,13 +125,15 @@ export CLASSPATH=$(hadoop classpath)
 ```
 
 then you can run the benchmarks with
+
 ```bash
 cargo bench -p hdfs-native --features benchmark
 ```
 
 The `benchmark` feature is required to expose `minidfs` and the internal erasure coding functions to benchmark.
 
 ## Running examples
+
 The examples make use of the `minidfs` module to create a simple HDFS cluster to run the example. This requires including the `integration-test` feature to enable the `minidfs` module. Alternatively, if you want to run the example against an existing HDFS cluster you can exclude the `integration-test` feature and make sure your `HADOOP_CONF_DIR` points to a directory with HDFS configs for talking to your cluster.
 
 ```bash

python/hdfs_native/__init__.py

@@ -131,8 +131,9 @@ def __init__(
         self,
         url: Optional[str] = None,
         config: Optional[Dict[str, str]] = None,
+        config_dir: Optional[str] = None,
     ):
-        self.inner = RawClient(url, config)
+        self.inner = RawClient(url, config, config_dir)
 
     def get_file_info(self, path: str) -> FileStatus:
         """Gets the file status for the file at `path`"""
@@ -362,8 +363,9 @@ def __init__(
         self,
         url: Optional[str] = None,
         config: Optional[Dict[str, str]] = None,
+        config_dir: Optional[str] = None,
     ):
-        self.inner = AsyncRawClient(url, config)
+        self.inner = AsyncRawClient(url, config, config_dir)
 
     async def get_file_info(self, path: str) -> FileStatus:
         """Gets the file status for the file at `path`"""

python/hdfs_native/_internal.pyi

@@ -98,6 +98,7 @@ class RawClient:
         self,
         url: Optional[str],
         config: Optional[Dict[str, str]],
+        config_dir: Optional[str],
     ) -> None: ...
     def get_file_info(self, path: str) -> FileStatus: ...
     def list_status(self, path: str, recursive: bool) -> Iterator[FileStatus]: ...
@@ -159,6 +160,7 @@ class AsyncRawClient:
         self,
         url: Optional[str],
         config: Optional[Dict[str, str]],
+        config_dir: Optional[str],
     ) -> None: ...
     async def get_file_info(self, path: str) -> FileStatus: ...
     def list_status(self, path: str, recursive: bool) -> AsyncIterator[FileStatus]: ...

python/src/lib.rs

@@ -413,8 +413,12 @@ struct RawClient {
 #[pymethods]
 impl RawClient {
     #[new]
-    #[pyo3(signature = (url, config))]
-    pub fn new(url: Option<&str>, config: Option<HashMap<String, String>>) -> PyResult<Self> {
+    #[pyo3(signature = (url, config, config_dir))]
+    pub fn new(
+        url: Option<&str>,
+        config: Option<HashMap<String, String>>,
+        config_dir: Option<&str>,
+    ) -> PyResult<Self> {
         // Initialize logging, ignore errors if this is called multiple times
         let _ = env_logger::try_init();
 
@@ -426,6 +430,10 @@ impl RawClient {
             builder = builder.with_url(url);
         }
 
+        if let Some(config_dir) = config_dir {
+            builder = builder.with_config_dir(config_dir);
+        }
+
         let inner = builder.build().map_err(PythonHdfsError::from)?;
 
         Ok(RawClient {
@@ -722,8 +730,12 @@ struct AsyncRawClient {
 #[pymethods]
 impl AsyncRawClient {
     #[new]
-    #[pyo3(signature = (url, config))]
-    pub fn new(url: Option<&str>, config: Option<HashMap<String, String>>) -> PyResult<Self> {
+    #[pyo3(signature = (url, config, config_dir))]
+    pub fn new(
+        url: Option<&str>,
+        config: Option<HashMap<String, String>>,
+        config_dir: Option<&str>,
+    ) -> PyResult<Self> {
         // Initialize logging, ignore errors if this is called multiple times
         let _ = env_logger::try_init();
 
@@ -735,6 +747,10 @@ impl AsyncRawClient {
             builder = builder.with_url(url);
         }
 
+        if let Some(config_dir) = config_dir {
+            builder = builder.with_config_dir(config_dir);
+        }
+
         let inner = builder.build().map_err(PythonHdfsError::from)?;
 
         Ok(Self { inner })

rust/build.rs

@@ -3,7 +3,9 @@ use std::io::Result;
 fn main() -> Result<()> {
     #[cfg(feature = "generate-protobuf")]
     {
-        std::env::set_var("PROTOC", protobuf_src::protoc());
+        unsafe {
+            std::env::set_var("PROTOC", protobuf_src::protoc());
+        }
 
         prost_build::compile_protos(
             &[

rust/src/client.rs

@@ -190,16 +190,41 @@ impl IORuntime {
     }
 }
 
-/// Builds a new [Client] instance. By default, configs will be loaded from the default config directories with the following precedence:
+/// Builds a new [Client] instance. Configs will be loaded with the following precedence:
+///
+/// - If method `ClientBuilder::with_config_dir` is invoked, configs will be loaded from `${config_dir}/{core,hdfs}-site.xml`
 /// - If the `HADOOP_CONF_DIR` environment variable is defined, configs will be loaded from `${HADOOP_CONF_DIR}/{core,hdfs}-site.xml`
 /// - If the `HADOOP_HOME` environment variable is defined, configs will be loaded from `${HADOOP_HOME}/etc/hadoop/{core,hdfs}-site.xml`
-/// - Otherwise no default configs are defined
+/// - Otherwise no configs are defined
+///
+/// Finally, configs set by `with_config` will override the configs loaded above.
 ///
 /// If no URL is defined, the `fs.defaultFS` config must be defined and is used as the URL.
 ///
 /// # Examples
 ///
+/// Create a new client with given config directory
+///
+/// ```rust,no_run
+/// # use hdfs_native::ClientBuilder;
+/// let client = ClientBuilder::new()
+///     .with_config_dir("/opt/hadoop/etc/hadoop")
+///     .build()
+///     .unwrap();
+/// ```
+///
+/// Create a new client with the environment variable
+///
+/// ```rust,no_run
+/// # use hdfs_native::ClientBuilder;
+/// unsafe { std::env::set_var("HADOOP_CONF_DIR", "/opt/hadoop/etc/hadoop") };
+/// let client = ClientBuilder::new()
+///     .build()
+///     .unwrap();
+/// ```
+///
 /// Create a new client using the fs.defaultFS config
+///
 /// ```rust
 /// # use hdfs_native::ClientBuilder;
 /// let client = ClientBuilder::new()
@@ -209,6 +234,7 @@ impl IORuntime {
 /// ```
 ///
 /// Create a new client connecting to a specific URL:
+///
 /// ```rust
 /// # use hdfs_native::ClientBuilder;
 /// let client = ClientBuilder::new()
@@ -218,6 +244,7 @@ impl IORuntime {
 /// ```
 ///
 /// Create a new client using a dedicated tokio runtime for spawned tasks and IO operations
+///
 /// ```rust
 /// # use hdfs_native::ClientBuilder;
 /// let client = ClientBuilder::new()
@@ -229,7 +256,8 @@ impl IORuntime {
 #[derive(Default)]
 pub struct ClientBuilder {
     url: Option<String>,
-    config: HashMap<String, String>,
+    config: Option<HashMap<String, String>>,
+    config_dir: Option<String>,
     runtime: Option<IORuntime>,
 }
 
@@ -245,15 +273,23 @@ impl ClientBuilder {
         self
     }
 
-    /// Set configs to use for the client. The provided configs will override any found in the default config files loaded
+    /// Set configs to use for the client. The provided configs will override any found in the config files loaded
     pub fn with_config(
         mut self,
         config: impl IntoIterator<Item = (impl Into<String>, impl Into<String>)>,
     ) -> Self {
-        self.config = config
-            .into_iter()
-            .map(|(k, v)| (k.into(), v.into()))
-            .collect();
+        self.config = Some(
+            config
+                .into_iter()
+                .map(|(k, v)| (k.into(), v.into()))
+                .collect(),
+        );
+        self
+    }
+
+    /// Set the configration directory path to read from. The provided path will override the one provided by environment variable.
+    pub fn with_config_dir(mut self, config_dir: impl Into<String>) -> Self {
+        self.config_dir = Some(config_dir.into());
         self
     }
 
@@ -266,7 +302,7 @@ impl ClientBuilder {
 
     /// Create the [Client] instance from the provided settings
     pub fn build(self) -> Result<Client> {
-        let config = Configuration::new_with_config(self.config)?;
+        let config = Configuration::new(self.config_dir, self.config)?;
         let url = if let Some(url) = self.url {
             Url::parse(&url)?
         } else {
@@ -323,18 +359,18 @@ impl Client {
     #[deprecated(since = "0.12.0", note = "Use ClientBuilder instead")]
     pub fn new(url: &str) -> Result<Self> {
         let parsed_url = Url::parse(url)?;
-        Self::build(&parsed_url, Configuration::new()?, None)
+        Self::build(&parsed_url, Configuration::new(None, None)?, None)
     }
 
     #[deprecated(since = "0.12.0", note = "Use ClientBuilder instead")]
     pub fn new_with_config(url: &str, config: HashMap<String, String>) -> Result<Self> {
         let parsed_url = Url::parse(url)?;
-        Self::build(&parsed_url, Configuration::new_with_config(config)?, None)
+        Self::build(&parsed_url, Configuration::new(None, Some(config))?, None)
     }
 
     #[deprecated(since = "0.12.0", note = "Use ClientBuilder instead")]
     pub fn default_with_config(config: HashMap<String, String>) -> Result<Self> {
-        let config = Configuration::new_with_config(config)?;
+        let config = Configuration::new(None, Some(config))?;
         Self::build(&Self::default_fs(&config)?, config, None)
     }
 
@@ -1138,7 +1174,7 @@ mod test {
     fn create_protocol(url: &str) -> Arc<NamenodeProtocol> {
         let proxy = NameServiceProxy::new(
             &Url::parse(url).unwrap(),
-            Arc::new(Configuration::new().unwrap()),
+            Arc::new(Configuration::new(None, None).unwrap()),
             RT.handle().clone(),
         )
         .unwrap();
@@ -1310,4 +1346,15 @@ mod test {
                 .is_ok()
         );
     }
+
+    #[test]
+    fn test_set_conf_dir() {
+        assert!(
+            ClientBuilder::new()
+                .with_url("hdfs://127.0.0.1:9000")
+                .with_config_dir("target/test")
+                .build()
+                .is_ok()
+        )
+    }
 }