Support kerberos authentication for Kudu connector

presto-kudu/pom.xml

@@ -23,6 +23,11 @@
             <version>${kudu.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.facebook.presto.hadoop</groupId>
+            <artifactId>hadoop-apache2</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>com.facebook.airlift</groupId>
             <artifactId>bootstrap</artifactId>

presto-kudu/src/main/java/com/facebook/presto/kudu/KuduClientConfig.java

@@ -40,6 +40,10 @@ public class KuduClientConfig
     private boolean disableStatistics;
     private boolean schemaEmulationEnabled;
     private String schemaEmulationPrefix = "presto::";
+    private boolean kerberosAuthEnabled;
+    private String kerberosPrincipal;
+    private String kerberosKeytab;
+    private boolean kerberosAuthDebugEnabled;
 
     @NotNull
     @Size(min = 1)
@@ -138,4 +142,52 @@ public KuduClientConfig setSchemaEmulationEnabled(boolean enabled)
         this.schemaEmulationEnabled = enabled;
         return this;
     }
+
+    public boolean isKerberosAuthEnabled()
+    {
+        return kerberosAuthEnabled;
+    }
+
+    @Config("kudu.kerberos-auth.enabled")
+    public KuduClientConfig setKerberosAuthEnabled(boolean enabled)
+    {
+        this.kerberosAuthEnabled = enabled;
+        return this;
+    }
+
+    public String getKerberosPrincipal()
+    {
+        return kerberosPrincipal;
+    }
+
+    @Config("kudu.kerberos-auth.principal")
+    public KuduClientConfig setKerberosPrincipal(String principal)
+    {
+        this.kerberosPrincipal = principal;
+        return this;
+    }
+
+    public String getKerberosKeytab()
+    {
+        return kerberosKeytab;
+    }
+
+    @Config("kudu.kerberos-auth.keytab")
+    public KuduClientConfig setKerberosKeytab(String keytab)
+    {
+        this.kerberosKeytab = keytab;
+        return this;
+    }
+
+    public boolean isKerberosAuthDebugEnabled()
+    {
+        return kerberosAuthDebugEnabled;
+    }
+
+    @Config("kudu.kerberos-auth.debug.enabled")
+    public KuduClientConfig setKerberosAuthDebugEnabled(boolean enabled)
+    {
+        this.kerberosAuthDebugEnabled = enabled;
+        return this;
+    }
 }

presto-kudu/src/main/java/com/facebook/presto/kudu/KuduClientSession.java

@@ -64,6 +64,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static com.facebook.presto.kudu.KuduUtil.reTryKerberos;
 import static com.facebook.presto.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static com.facebook.presto.spi.StandardErrorCode.QUERY_REJECTED;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -75,16 +76,19 @@ public class KuduClientSession
     private final KuduConnectorId connectorId;
     private final KuduClient client;
     private final SchemaEmulation schemaEmulation;
+    private final boolean kerberosAuthEnabled;
 
-    public KuduClientSession(KuduConnectorId connectorId, KuduClient client, SchemaEmulation schemaEmulation)
+    public KuduClientSession(KuduConnectorId connectorId, KuduClient client, SchemaEmulation schemaEmulation, boolean kerberosAuthEnabled)
     {
         this.connectorId = connectorId;
         this.client = client;
         this.schemaEmulation = schemaEmulation;
+        this.kerberosAuthEnabled = kerberosAuthEnabled;
     }
 
     public List<String> listSchemaNames()
     {
+        reTryKerberos(kerberosAuthEnabled);
         return schemaEmulation.listSchemaNames(client);
     }
 
@@ -105,6 +109,7 @@ private List<String> internalListTables(String prefix)
 
     public List<SchemaTableName> listTables(Optional<String> optSchemaName)
     {
+        reTryKerberos(kerberosAuthEnabled);
         if (optSchemaName.isPresent()) {
             return listTablesSingleSchema(optSchemaName.get());
         }
@@ -130,18 +135,21 @@ private List<SchemaTableName> listTablesSingleSchema(String schemaName)
 
     public Schema getTableSchema(KuduTableHandle tableHandle)
     {
+        reTryKerberos(kerberosAuthEnabled);
         KuduTable table = tableHandle.getTable(this);
         return table.getSchema();
     }
 
     public Map<String, Object> getTableProperties(KuduTableHandle tableHandle)
     {
+        reTryKerberos(kerberosAuthEnabled);
         KuduTable table = tableHandle.getTable(this);
         return KuduTableProperties.toMap(table);
     }
 
     public List<KuduSplit> buildKuduSplits(KuduTableLayoutHandle layoutHandle)
     {
+        reTryKerberos(kerberosAuthEnabled);
         KuduTableHandle tableHandle = layoutHandle.getTableHandle();
         KuduTable table = tableHandle.getTable(this);
         final int primaryKeyColumnCount = table.getSchema().getPrimaryKeyColumnCount();
@@ -185,6 +193,7 @@ public List<KuduSplit> buildKuduSplits(KuduTableLayoutHandle layoutHandle)
 
     public KuduScanner createScanner(KuduSplit kuduSplit)
     {
+        reTryKerberos(kerberosAuthEnabled);
         try {
             return KuduScanToken.deserializeIntoScanner(kuduSplit.getSerializedScanToken(), client);
         }
@@ -195,6 +204,7 @@ public KuduScanner createScanner(KuduSplit kuduSplit)
 
     public KuduTable openTable(SchemaTableName schemaTableName)
     {
+        reTryKerberos(kerberosAuthEnabled);
         String rawName = schemaEmulation.toRawName(schemaTableName);
         try {
             return client.openTable(rawName);
@@ -210,21 +220,25 @@ public KuduTable openTable(SchemaTableName schemaTableName)
 
     public KuduSession newSession()
     {
+        reTryKerberos(kerberosAuthEnabled);
         return client.newSession();
     }
 
     public void createSchema(String schemaName)
     {
+        reTryKerberos(kerberosAuthEnabled);
         schemaEmulation.createSchema(client, schemaName);
     }
 
     public void dropSchema(String schemaName)
     {
+        reTryKerberos(kerberosAuthEnabled);
         schemaEmulation.dropSchema(client, schemaName);
     }
 
     public void dropTable(SchemaTableName schemaTableName)
     {
+        reTryKerberos(kerberosAuthEnabled);
         try {
             String rawName = schemaEmulation.toRawName(schemaTableName);
             client.deleteTable(rawName);
@@ -236,6 +250,7 @@ public void dropTable(SchemaTableName schemaTableName)
 
     public void renameTable(SchemaTableName schemaTableName, SchemaTableName newSchemaTableName)
     {
+        reTryKerberos(kerberosAuthEnabled);
         try {
             String rawName = schemaEmulation.toRawName(schemaTableName);
             String newRawName = schemaEmulation.toRawName(newSchemaTableName);
@@ -250,6 +265,7 @@ public void renameTable(SchemaTableName schemaTableName, SchemaTableName newSche
 
     public KuduTable createTable(ConnectorTableMetadata tableMetadata, boolean ignoreExisting)
     {
+        reTryKerberos(kerberosAuthEnabled);
         try {
             String rawName = schemaEmulation.toRawName(tableMetadata.getTable());
             if (ignoreExisting) {
@@ -276,6 +292,7 @@ public KuduTable createTable(ConnectorTableMetadata tableMetadata, boolean ignor
 
     public void addColumn(SchemaTableName schemaTableName, ColumnMetadata column)
     {
+        reTryKerberos(kerberosAuthEnabled);
         try {
             String rawName = schemaEmulation.toRawName(schemaTableName);
             AlterTableOptions alterOptions = new AlterTableOptions();
@@ -290,6 +307,7 @@ public void addColumn(SchemaTableName schemaTableName, ColumnMetadata column)
 
     public void dropColumn(SchemaTableName schemaTableName, String name)
     {
+        reTryKerberos(kerberosAuthEnabled);
         try {
             String rawName = schemaEmulation.toRawName(schemaTableName);
             AlterTableOptions alterOptions = new AlterTableOptions();
@@ -303,6 +321,7 @@ public void dropColumn(SchemaTableName schemaTableName, String name)
 
     public void renameColumn(SchemaTableName schemaTableName, String oldName, String newName)
     {
+        reTryKerberos(kerberosAuthEnabled);
         try {
             String rawName = schemaEmulation.toRawName(schemaTableName);
             AlterTableOptions alterOptions = new AlterTableOptions();
@@ -316,11 +335,13 @@ public void renameColumn(SchemaTableName schemaTableName, String oldName, String
 
     public void addRangePartition(SchemaTableName schemaTableName, RangePartition rangePartition)
     {
+        reTryKerberos(kerberosAuthEnabled);
         changeRangePartition(schemaTableName, rangePartition, RangePartitionChange.ADD);
     }
 
     public void dropRangePartition(SchemaTableName schemaTableName, RangePartition rangePartition)
     {
+        reTryKerberos(kerberosAuthEnabled);
         changeRangePartition(schemaTableName, rangePartition, RangePartitionChange.DROP);
     }
 

presto-kudu/src/main/java/com/facebook/presto/kudu/KuduModule.java

@@ -91,14 +91,14 @@ KuduClientSession createKuduClientSession(
     {
         requireNonNull(config, "config is null");
 
-        KuduClient.KuduClientBuilder builder = new KuduClient.KuduClientBuilder(config.getMasterAddresses());
-        builder.defaultAdminOperationTimeoutMs(config.getDefaultAdminOperationTimeout().toMillis());
-        builder.defaultOperationTimeoutMs(config.getDefaultOperationTimeout().toMillis());
-        builder.defaultSocketReadTimeoutMs(config.getDefaultSocketReadTimeout().toMillis());
-        if (config.isDisableStatistics()) {
-            builder.disableStatistics();
+        KuduClient client;
+        if (!config.isKerberosAuthEnabled()) {
+            client = KuduUtil.createKuduClient(config);
+        }
+        else {
+            KuduUtil.initKerberosENV(config.getKerberosPrincipal(), config.getKerberosKeytab(), config.isKerberosAuthDebugEnabled());
+            client = KuduUtil.createKuduKerberosClient(config);
         }
-        KuduClient client = builder.build();
 
         SchemaEmulation strategy;
         if (config.isSchemaEmulationEnabled()) {
@@ -107,6 +107,6 @@ KuduClientSession createKuduClientSession(
         else {
             strategy = new NoSchemaEmulation();
         }
-        return new KuduClientSession(connectorId, client, strategy);
+        return new KuduClientSession(connectorId, client, strategy, config.isKerberosAuthEnabled());
     }
 }

presto-kudu/src/main/java/com/facebook/presto/kudu/KuduUtil.java

@@ -0,0 +1,98 @@
+/*
+ * 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.
+ */
+package com.facebook.presto.kudu;
+
+import com.facebook.airlift.log.Logger;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.kudu.client.KuduClient;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+
+public class KuduUtil
+{
+    private static final Logger log = Logger.get(KuduUtil.class);
+
+    private KuduUtil()
+    {
+        // not allowed to be called to initialize instance
+    }
+
+    /**
+     * Initialize kerberos authentication
+     */
+    static void initKerberosENV(String principal, String keytab, boolean debugEnabled)
+    {
+        try {
+            Configuration conf = new Configuration();
+            conf.set("hadoop.security.authentication", "kerberos");
+            if (debugEnabled) {
+                System.setProperty("sun.security.krb5.debug", "true");
+            }
+            UserGroupInformation.setConfiguration(conf);
+            UserGroupInformation.loginUserFromKeytab(principal, keytab);
+            log.info("Connecting to kudu with kerberos authentication");
+            log.info("Current user: " + UserGroupInformation.getCurrentUser());
+            log.info("Login user: " + UserGroupInformation.getLoginUser());
+        }
+        catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    static KuduClient createKuduKerberosClient(KuduClientConfig config)
+    {
+        KuduClient client = null;
+        try {
+            reTryKerberos(true);
+            client = UserGroupInformation.getLoginUser().doAs(
+                    (PrivilegedExceptionAction<KuduClient>) () -> createKuduClient(config));
+            return client;
+        }
+        catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    static KuduClient createKuduClient(KuduClientConfig config)
+    {
+        KuduClient.KuduClientBuilder builder = new KuduClient.KuduClientBuilder(config.getMasterAddresses());
+        builder.defaultAdminOperationTimeoutMs(config.getDefaultAdminOperationTimeout().toMillis());
+        builder.defaultOperationTimeoutMs(config.getDefaultOperationTimeout().toMillis());
+        builder.defaultSocketReadTimeoutMs(config.getDefaultSocketReadTimeout().toMillis());
+        if (config.isDisableStatistics()) {
+            builder.disableStatistics();
+        }
+        return builder.build();
+    }
+
+    static void reTryKerberos(boolean enabled)
+    {
+        if (enabled) {
+            log.debug("Try relogin kerberos at first!");
+            try {
+                if (UserGroupInformation.isLoginKeytabBased()) {
+                    UserGroupInformation.getLoginUser().reloginFromKeytab();
+                }
+                else if (UserGroupInformation.isLoginTicketBased()) {
+                    UserGroupInformation.getLoginUser().reloginFromTicketCache();
+                }
+            }
+            catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+}