Initial kerberos support

Cargo.toml

@@ -8,9 +8,10 @@ edition = "2021"
 [dependencies]
 log = "0.4"
 testcontainers = "0.15"
-
+project-root = "0.2"
 
 [dev-dependencies]
 env_logger = "0.10"
-tokio = { version = "1.15", features = ["rt", "macros", "time"]}
-hdfs-native = { git = "https://github.com/milenkovicm/hdfs-native.git"}
+tokio = { version = "1", features = ["rt", "macros", "time"]}
+hdfs-native = { git = "https://github.com/milenkovicm/hdfs-native.git"}
+serial_test = { version = "2.0"}

README.md

@@ -1,4 +1,4 @@
-# MiniDFS cluster
+# MiniDFS Cluster
 
 Testcontainer local HDFS cluster using hadoop mini dfs.
 
@@ -15,6 +15,30 @@ let hdfs_server_url = format!("hdfs://{}:{}/", "localhost", server_node.get_host
 
 HDFS name node should be available at `hdfs://localhost:9000` and name node http at `http://localhost:8020`.
 
-Limitations:
+## Kerberos Support
+
+MiniHDFS can be configured with kerberos enabled:
+
+```rust
+use testcontainers::clients;
+use testcontainers_minidfs_rs::*;
+
+let docker = clients::Cli::default();
+let container = MiniDFS::builder().with_kerberos(true).build();
+let server_node = docker.run(container);
+```
+
+MiniDFS will be configured to support kerberos and all required files will be exposed as a docker volume mounted in the target directory.
+
+```rust
+let container = MiniDFS::builder().with_kerberos(true).build();
+let kerberos_cache = container.inner().kerberos_cache();
+let kerberos_config = container.inner().kerberos_config();
+let hdfs_config = container.inner().hdfs_config();
+```
+
+All required files needed for hdfs client setup are exposed. (`kinit` will be executed by the container, kerberos cache will be produced).
+
+## Limitations
 
 - ports are hardcoded, thus only single instance per host is possible

src/lib.rs

@@ -1,3 +1,5 @@
+use log::info;
+use std::path::PathBuf;
 use testcontainers::{core::WaitFor, Image, ImageArgs, RunnableImage};
 
 pub const PORT_NAME_NODE: u16 = 9000;
@@ -7,27 +9,155 @@ const PORT_DATA_N0DE_0: u16 = 50010;
 const PORT_DATA_NODE_1: u16 = 50011;
 const PORT_DATA_NODE_2: u16 = 50012;
 const PORT_DATA_NODE_3: u16 = 50013;
+const PORT_KERBEROS: u16 = 62600;
 
 pub struct MiniDFS {
     tag: String,
+    config_path: Option<PathBuf>,
+    kerberos_enabled: bool,
+}
+
+pub struct MiniDFSBuilder {
+    tag: String,
+    kerberos_enabled: bool,
+    config_volume: bool,
+}
+
+impl MiniDFSBuilder {
+    pub fn with_tag(mut self, tag: &str) -> Self {
+        self.tag = tag.to_string();
+        self
+    }
+
+    pub fn with_config_volume(mut self, enabled: bool) -> Self {
+        self.config_volume = enabled;
+
+        self
+    }
+
+    pub fn with_kerberos(mut self, enabled: bool) -> Self {
+        self.kerberos_enabled = enabled;
+
+        self
+    }
+    /// local directory where kerberos configuration
+    /// directory should be mounted.
+    fn project_root_directory() -> PathBuf {
+        // TODO check override env as well
+        // CARGO_TARGET_DIR
+
+        // this one does not work as expected
+        // let mut cwd = std::env::current_dir().expect("cwd");
+        let mut root = project_root::get_project_root().expect("project root folder");
+
+        root.push("target");
+        root.push("HDFS");
+
+        root
+    }
+
+    pub fn build(self) -> RunnableImage<MiniDFS> {
+        let config_path = if self.kerberos_enabled || self.config_volume {
+            Some(Self::project_root_directory())
+        } else {
+            None
+        };
+
+        let mut image = RunnableImage::from(MiniDFS {
+            tag: self.tag,
+            config_path: config_path.clone(),
+            kerberos_enabled: self.kerberos_enabled,
+        })
+        .with_mapped_port((PORT_NAME_NODE, PORT_NAME_NODE))
+        .with_mapped_port((PORT_NAME_NODE_HTTP, PORT_NAME_NODE_HTTP))
+        .with_mapped_port((PORT_DATA_N0DE_0, PORT_DATA_N0DE_0))
+        .with_mapped_port((PORT_DATA_NODE_1, PORT_DATA_NODE_1))
+        .with_mapped_port((PORT_DATA_NODE_2, PORT_DATA_NODE_2))
+        .with_mapped_port((PORT_DATA_NODE_3, PORT_DATA_NODE_3))
+        .with_mapped_port((PORT_KERBEROS, PORT_KERBEROS));
+
+        if self.config_volume || self.kerberos_enabled {
+            let volume_kerberos = config_path.unwrap();
+            let volume_kerberos = volume_kerberos.to_string_lossy();
+            if self.kerberos_enabled {
+                info!(
+                    "Kerberos support has been enabled, testcontainer will mount container volume to local directory: [{}]",
+                volume_kerberos
+            );
+            } else {
+                info!(
+                    "Configuration support has been enabled, testcontainer will mount container volume to local directory: [{}]",
+                volume_kerberos
+            );
+            }
+
+            image = image.with_volume((volume_kerberos, "/tmp/HDFS"))
+        }
+
+        if self.kerberos_enabled {
+            image = image
+                .with_env_var(("KDC_ENABLED", "true"))
+                .with_env_var(("HOST_OS", std::env::consts::OS))
+        }
+
+        image
+    }
 }
 
 impl MiniDFS {
+    pub fn builder() -> MiniDFSBuilder {
+        MiniDFSBuilder {
+            tag: "latest".into(),
+            kerberos_enabled: false,
+            config_volume: false,
+        }
+    }
+
     /// Runnable docker image
     pub fn runnable() -> RunnableImage<MiniDFS> {
-        Self::runnable_from_tag("latest".into())
+        Self::builder().build()
     }
 
     /// Runnable docker image from a tag
-    pub fn runnable_from_tag(tag: String) -> RunnableImage<MiniDFS> {
-        RunnableImage::from(MiniDFS { tag })
-            .with_mapped_port((PORT_NAME_NODE, PORT_NAME_NODE))
-            .with_mapped_port((PORT_NAME_NODE_HTTP, PORT_NAME_NODE_HTTP))
-            .with_mapped_port((PORT_DATA_N0DE_0, PORT_DATA_N0DE_0))
-            .with_mapped_port((PORT_DATA_NODE_1, PORT_DATA_NODE_1))
-            .with_mapped_port((PORT_DATA_NODE_2, PORT_DATA_NODE_2))
-            .with_mapped_port((PORT_DATA_NODE_3, PORT_DATA_NODE_3))
-        //.with_container_name("minidfs") // we wont name container as failed or exited may occupy the name
+    pub fn runnable_from_tag(tag: &str) -> RunnableImage<MiniDFS> {
+        Self::builder().with_tag(tag).build()
+    }
+    // core-site.xml
+    pub fn kerberos_config(&self) -> Option<PathBuf> {
+        if !self.kerberos_enabled {
+            None
+        } else if std::env::consts::OS == "macos" {
+            // macos handles TCP protocol configuration a slightly
+            // different, thus in case of macos we use prepared
+            // configuration
+            self.config_path.clone().map(|mut p| {
+                p.push("macos_krb5.conf");
+                p
+            })
+        } else {
+            self.config_path.clone().map(|mut p| {
+                p.push("krb5.conf");
+                p
+            })
+        }
+    }
+
+    pub fn kerberos_cache(&self) -> Option<PathBuf> {
+        if !self.kerberos_enabled {
+            None
+        } else {
+            self.config_path.clone().map(|mut p| {
+                p.push("krb5_cache");
+                p
+            })
+        }
+    }
+
+    pub fn hdfs_config(&self) -> Option<PathBuf> {
+        self.config_path.clone().map(|mut p| {
+            p.push("core-site.xml");
+            p
+        })
     }
 }
 

tests/e2e.rs

@@ -6,6 +6,7 @@ use testcontainers_minidfs_rs::*;
 const DATA: &str = "1234567890";
 
 #[tokio::test]
+#[serial_test::serial]
 async fn e2e() {
     let _ = env_logger::builder().is_test(true).try_init();
     std::env::set_var("LIBHDFS3_CONF", "libhdfs3-hdfs-client.xml");
@@ -61,3 +62,51 @@ async fn e2e() {
 
     fs.delete(&test_dir, true).expect("directory to be deleted");
 }
+
+#[tokio::test]
+#[serial_test::serial]
+async fn e2e_kerberos() {
+    let _ = env_logger::builder().is_test(true).try_init();
+    let docker = clients::Cli::default();
+
+    let container = MiniDFS::builder().with_kerberos(true).build();
+    let kerberos_cache = container.inner().kerberos_cache();
+    let kerberos_config = container.inner().kerberos_config();
+    let hdfs_config = container.inner().hdfs_config();
+
+    let _server_node = docker.run(container);
+
+    assert!(kerberos_cache.is_some());
+    assert!(kerberos_config.is_some());
+    assert!(hdfs_config.is_some());
+
+    let kerberos_cache = kerberos_cache.unwrap();
+    let kerberos_config = kerberos_config.unwrap();
+    let hdfs_config = hdfs_config.unwrap();
+
+    info!("kerberos cache: {:?}", kerberos_cache);
+    info!("kerberos config: {:?}", kerberos_config);
+    info!("hdfs config: {:?}", hdfs_config);
+
+    assert!(kerberos_cache.exists());
+    assert!(kerberos_config.exists());
+    assert!(hdfs_config.exists());
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn e2e_config_volume() {
+    let _ = env_logger::builder().is_test(true).try_init();
+    let docker = clients::Cli::default();
+
+    let container = MiniDFS::builder().with_config_volume(true).build();
+
+    let hdfs_config = container.inner().hdfs_config();
+
+    let _server_node = docker.run(container);
+
+    assert!(hdfs_config.is_some());
+    let hdfs_config = hdfs_config.unwrap();
+    info!("hdfs config: {:?}", hdfs_config);
+    assert!(hdfs_config.exists());
+}