Added encryption flags for configuration (#40)

* Added encryption flags for configuration * Added missing flags to example/test configuration
teragrep · Aug 20, 2024 · 220d0d5 · 220d0d5
1 parent 18de5a1
commit 220d0d5
Show file tree

Hide file tree

Showing 7 changed files with 38 additions and 4 deletions.
diff --git a/rpm/resources/application.properties b/rpm/resources/application.properties
@@ -47,4 +47,6 @@ dfs.namenode.kerberos.principal.pattern=test
 KerberosKeytabUser=test
 KerberosKeytabPath=test
 dfs.client.use.datanode.hostname=false
-kerberosLoginAutorenewal=true
+kerberosLoginAutorenewal=true
+dfs.data.transfer.protection=test
+dfs.encrypt.data.transfer.cipher.suites=test
diff --git a/src/main/java/com/teragrep/cfe_39/Config.java b/src/main/java/com/teragrep/cfe_39/Config.java
@@ -80,6 +80,8 @@ public class Config {
     private final long pruneOffset;
     private final boolean skipNonRFC5424Records;
     private final boolean skipEmptyRFC5424Records;
+    private final String dfsDataTransferProtection;
+    private final String dfsEncryptDataTransferCipherSuites;
 
     public Config() throws IOException {
         Properties properties = new Properties();
@@ -147,6 +149,15 @@ public Config() throws IOException {
         }
         this.kerberosTestMode = properties.getProperty("dfs.client.use.datanode.hostname", "false");
 
+        this.dfsDataTransferProtection = properties.getProperty("dfs.data.transfer.protection");
+        if (this.dfsDataTransferProtection == null) {
+            throw new IllegalArgumentException("dfsDataTransferProtection not set");
+        }
+        this.dfsEncryptDataTransferCipherSuites = properties.getProperty("dfs.encrypt.data.transfer.cipher.suites");
+        if (this.dfsEncryptDataTransferCipherSuites == null) {
+            throw new IllegalArgumentException("dfsEncryptDataTransferCipherSuites not set");
+        }
+
         // kafka
         this.queueTopicPattern = properties.getProperty("queueTopicPattern", "^.*$");
         this.numOfConsumers = Integer.parseInt(properties.getProperty("numOfConsumers", "1"));
@@ -274,4 +285,12 @@ public boolean getSkipEmptyRFC5424Records() {
     public String getKerberosLoginAutorenewal() {
         return kerberosLoginAutorenewal;
     }
+
+    public String getDfsDataTransferProtection() {
+        return dfsDataTransferProtection;
+    }
+
+    public String getDfsEncryptDataTransferCipherSuites() {
+        return dfsEncryptDataTransferCipherSuites;
+    }
 }
diff --git a/src/main/java/com/teragrep/cfe_39/consumers/kafka/HDFSWrite.java b/src/main/java/com/teragrep/cfe_39/consumers/kafka/HDFSWrite.java
@@ -131,6 +131,10 @@ These values should be fetched from config and other input parameters (topic+par
             // server principal, the kerberos principle that the namenode is using
             conf.set("dfs.namenode.kerberos.principal.pattern", config.getKerberosPrincipal());
 
+            // set sasl
+            conf.set("dfs.data.transfer.protection", config.getDfsDataTransferProtection());
+            conf.set("dfs.encrypt.data.transfer.cipher.suites", config.getDfsEncryptDataTransferCipherSuites());
+
             // filesystem for HDFS access is set here
             fs = FileSystem.get(conf);
         }

diff --git a/src/main/java/com/teragrep/cfe_39/consumers/kafka/HdfsDataIngestion.java b/src/main/java/com/teragrep/cfe_39/consumers/kafka/HdfsDataIngestion.java
@@ -133,6 +133,9 @@ public HdfsDataIngestion(Config config) throws IOException {
             /* server principal
              the kerberos principle that the namenode is using*/
             conf.set("dfs.namenode.kerberos.principal.pattern", config.getKerberosPrincipal());
+            // set sasl
+            conf.set("dfs.data.transfer.protection", config.getDfsDataTransferProtection());
+            conf.set("dfs.encrypt.data.transfer.cipher.suites", config.getDfsEncryptDataTransferCipherSuites());
             // set usergroup stuff
             UserGroupInformation.setConfiguration(conf);
             UserGroupInformation.loginUserFromKeytab(config.getKerberosKeytabUser(), config.getKerberosKeytabPath());

diff --git a/src/test/resources/broken.application.properties b/src/test/resources/broken.application.properties
@@ -39,4 +39,6 @@ dfs.namenode.kerberos.principal.pattern=test
 KerberosKeytabUser=test
 KerberosKeytabPath=test
 dfs.client.use.datanode.hostname=false
-kerberosLoginAutorenewal=true
+kerberosLoginAutorenewal=true
+dfs.data.transfer.protection=test
+dfs.encrypt.data.transfer.cipher.suites=test
diff --git a/src/test/resources/failProcessing.application.properties b/src/test/resources/failProcessing.application.properties
@@ -41,4 +41,6 @@ dfs.namenode.kerberos.principal.pattern=test
 KerberosKeytabUser=test
 KerberosKeytabPath=test
 dfs.client.use.datanode.hostname=false
-kerberosLoginAutorenewal=true
+kerberosLoginAutorenewal=true
+dfs.data.transfer.protection=test
+dfs.encrypt.data.transfer.cipher.suites=test
diff --git a/src/test/resources/valid.application.properties b/src/test/resources/valid.application.properties
@@ -41,4 +41,6 @@ dfs.namenode.kerberos.principal.pattern=test
 KerberosKeytabUser=test
 KerberosKeytabPath=test
 dfs.client.use.datanode.hostname=false
-kerberosLoginAutorenewal=true
+kerberosLoginAutorenewal=true
+dfs.data.transfer.protection=test
+dfs.encrypt.data.transfer.cipher.suites=test