Adding hadoop kerberos authentification. (apache#3419)

* adding kerberos authentication

* make the 2 functions identical

docs/content/development/extensions-core/hdfs.md

@@ -14,8 +14,11 @@ Make sure to [include](../../operations/including-extensions.html) `druid-hdfs-s
 |--------|---------------|-----------|-------|
 |`druid.storage.type`|hdfs||Must be set.|
 |`druid.storage.storageDirectory`||Directory for storing segments.|Must be set.|
+|`druid.hadoop.security.kerberos.principal`|`[email protected]`| Principal user name |empty|
+|`druid.hadoop.security.kerberos.keytab`|`/etc/security/keytabs/druid.headlessUser.keytab`|Path to keytab file|empty|
 
-If you are using the Hadoop indexer, set your output directory to be a location on Hadoop and it will work
+If you are using the Hadoop indexer, set your output directory to be a location on Hadoop and it will work.
+If you want to eagerly authenticate against a secured hadoop/hdfs cluster you must set `druid.hadoop.security.kerberos.principal` and `druid.hadoop.security.kerberos.keytab`, this is an alternative to the cron job method that runs `kinit` command periodically.  
 
 ## Google Cloud Storage
 

docs/content/ingestion/batch-ingestion.md

@@ -287,6 +287,19 @@ classification=yarn-site,properties=[mapreduce.reduce.memory.mb=6144,mapreduce.r
 loads](../tutorials/cluster.html#configure-cluster-for-hadoop-data-loads)" using the XML files from
 `/etc/hadoop/conf` on your EMR master.
 
+### Secured Hadoop Cluster
+
+By default druid can use the exisiting TGT kerberos ticket available in local kerberos key cache.
+Although TGT ticket has a limited life cycle, 
+therefore you need to call `kinit` command periodically to ensure validity of TGT ticket.
+To avoid this extra external cron job script calling `kinit` periodically,
+ you can provide the principal name and keytab location and druid will do the authentication transparently at startup and job launching time.   
+
+|Property|Possible Values|Description|Default|
+|--------|---------------|-----------|-------|
+|`druid.hadoop.security.kerberos.principal`|`[email protected]`| Principal user name |empty|
+|`druid.hadoop.security.kerberos.keytab`|`/etc/security/keytabs/druid.headlessUser.keytab`|Path to keytab file|empty|
+
 #### Loading from S3 with EMR
 
 - In the `jobProperties` field in the `tuningConfig` section of your Hadoop indexing task, add:

extensions-core/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsKerberosConfig.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you 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 io.druid.storage.hdfs;
+
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class HdfsKerberosConfig
+{
+  @JsonProperty
+  private final String principal;
+  @JsonProperty
+  private final String keytab;
+
+  @JsonCreator
+  public HdfsKerberosConfig(@JsonProperty("principal") String principal,@JsonProperty("keytab") String keytab) {
+    this.principal = principal;
+    this.keytab = keytab;
+  }
+
+  @JsonProperty
+  public String getPrincipal()
+  {
+    return principal;
+  }
+
+  @JsonProperty
+  public String getKeytab()
+  {
+    return keytab;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof HdfsKerberosConfig)) {
+      return false;
+    }
+
+    HdfsKerberosConfig that = (HdfsKerberosConfig) o;
+
+    if (getPrincipal() != null ? !getPrincipal().equals(that.getPrincipal()) : that.getPrincipal() != null) {
+      return false;
+    }
+    return getKeytab() != null ? getKeytab().equals(that.getKeytab()) : that.getKeytab() == null;
+
+  }
+
+  @Override
+  public int hashCode()
+  {
+    int result = getPrincipal() != null ? getPrincipal().hashCode() : 0;
+    result = 31 * result + (getKeytab() != null ? getKeytab().hashCode() : 0);
+    return result;
+  }
+}

extensions-core/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsStorageAuthentication.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you 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 io.druid.storage.hdfs;
+
+
+import com.google.common.base.Strings;
+import com.google.inject.Inject;
+import com.metamx.common.ISE;
+import com.metamx.common.lifecycle.LifecycleStart;
+import com.metamx.common.lifecycle.LifecycleStop;
+import com.metamx.common.logger.Logger;
+import io.druid.guice.ManageLifecycle;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import java.io.IOException;
+
+@ManageLifecycle
+public class HdfsStorageAuthentication
+{
+  private static final Logger log = new Logger(HdfsStorageAuthentication.class);
+  private final HdfsKerberosConfig hdfsKerberosConfig;
+  private final Configuration hadoopConf;
+
+  @Inject
+  public HdfsStorageAuthentication(HdfsKerberosConfig hdfsKerberosConfig, Configuration hadoopConf)
+  {
+    this.hdfsKerberosConfig = hdfsKerberosConfig;
+    this.hadoopConf = hadoopConf;
+  }
+
+  /**
+   * Dose authenticate against a secured hadoop cluster
+   * In case of any bug fix make sure to fix the code in JobHelper#authenticate as well.
+   */
+  @LifecycleStart
+  public void authenticate()
+  {
+    String principal = hdfsKerberosConfig.getPrincipal();
+    String keytab = hdfsKerberosConfig.getKeytab();
+    if (!Strings.isNullOrEmpty(principal) && !Strings.isNullOrEmpty(keytab)) {
+      UserGroupInformation.setConfiguration(hadoopConf);
+      if (UserGroupInformation.isSecurityEnabled()) {
+        try {
+          if (UserGroupInformation.getCurrentUser().hasKerberosCredentials() == false
+              || !UserGroupInformation.getCurrentUser().getUserName().equals(principal)) {
+            log.info("Trying to authenticate user [%s] with keytab [%s]..", principal, keytab);
+            UserGroupInformation.loginUserFromKeytab(principal, keytab);
+          }
+        }
+        catch (IOException e) {
+          throw new ISE(e, "Failed to authenticate user principal [%s] with keytab [%s]", principal, keytab);
+        }
+      }
+    }
+  }
+
+  @LifecycleStop
+  public void stop()
+  {
+    //noop
+  }
+}

extensions-core/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsStorageDruidModule.java

@@ -26,10 +26,13 @@
 import com.google.inject.Binder;
 import com.google.inject.Inject;
 import com.google.inject.multibindings.MapBinder;
+import com.metamx.common.logger.Logger;
 import io.druid.data.SearchableVersionedDataFinder;
 import io.druid.guice.Binders;
 import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.LazySingleton;
+import io.druid.guice.LifecycleModule;
+import io.druid.guice.ManageLifecycle;
 import io.druid.initialization.DruidModule;
 import io.druid.storage.hdfs.tasklog.HdfsTaskLogs;
 import io.druid.storage.hdfs.tasklog.HdfsTaskLogsConfig;
@@ -44,6 +47,7 @@
  */
 public class HdfsStorageDruidModule implements DruidModule
 {
+  private static final Logger log = new Logger(HdfsStorageDruidModule.class);
   public static final String SCHEME = "hdfs";
   private Properties props = null;
 
@@ -126,5 +130,9 @@ public void configure(Binder binder)
     Binders.taskLogsBinder(binder).addBinding("hdfs").to(HdfsTaskLogs.class);
     JsonConfigProvider.bind(binder, "druid.indexer.logs", HdfsTaskLogsConfig.class);
     binder.bind(HdfsTaskLogs.class).in(LazySingleton.class);
+    JsonConfigProvider.bind(binder, "druid.hadoop.security.kerberos", HdfsKerberosConfig.class);
+    binder.bind(HdfsStorageAuthentication.class).in(ManageLifecycle.class);
+    LifecycleModule.register(binder, HdfsStorageAuthentication.class);
+
   }
 }

extensions-core/hdfs-storage/src/test/java/io/druid/storage/hdfs/HdfsKerberosConfigTest.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you 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 io.druid.storage.hdfs;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class HdfsKerberosConfigTest
+{
+  private final ObjectMapper mapper = new ObjectMapper();
+
+  @Test
+  public void testSerDesr() throws IOException
+  {
+    HdfsKerberosConfig hdfsKerberosConfig = new HdfsKerberosConfig("principal", "keytab");
+    Assert.assertEquals(
+        hdfsKerberosConfig,
+        mapper.reader(HdfsKerberosConfig.class).readValue(mapper.writeValueAsString(hdfsKerberosConfig))
+    );
+  }
+
+}

indexing-hadoop/src/main/java/io/druid/indexer/HadoopDruidIndexerConfig.java

@@ -91,6 +91,7 @@ public class HadoopDruidIndexerConfig
   public static final IndexIO INDEX_IO;
   public static final IndexMerger INDEX_MERGER;
   public static final IndexMergerV9 INDEX_MERGER_V9;
+  public static final HadoopKerberosConfig HADOOP_KERBEROS_CONFIG;
 
   private static final String DEFAULT_WORKING_PATH = "/tmp/druid-indexing";
 
@@ -106,6 +107,7 @@ public void configure(Binder binder)
                 JsonConfigProvider.bindInstance(
                     binder, Key.get(DruidNode.class, Self.class), new DruidNode("hadoop-indexer", null, null)
                 );
+                JsonConfigProvider.bind(binder, "druid.hadoop.security.kerberos", HadoopKerberosConfig.class);
               }
             },
             new IndexingHadoopModule()
@@ -115,6 +117,7 @@ public void configure(Binder binder)
     INDEX_IO = injector.getInstance(IndexIO.class);
     INDEX_MERGER = injector.getInstance(IndexMerger.class);
     INDEX_MERGER_V9 = injector.getInstance(IndexMergerV9.class);
+    HADOOP_KERBEROS_CONFIG = injector.getInstance(HadoopKerberosConfig.class);
   }
 
   public static enum IndexJobCounters

indexing-hadoop/src/main/java/io/druid/indexer/HadoopKerberosConfig.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you 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 io.druid.indexer;
+
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class HadoopKerberosConfig
+{
+  @JsonProperty
+  private final String principal;
+  @JsonProperty
+  private final String keytab;
+
+  @JsonCreator
+  public HadoopKerberosConfig(@JsonProperty("principal") String principal,@JsonProperty("keytab") String keytab) {
+    this.principal = principal;
+    this.keytab = keytab;
+  }
+
+  @JsonProperty
+  public String getPrincipal()
+  {
+    return principal;
+  }
+
+  @JsonProperty
+  public String getKeytab()
+  {
+    return keytab;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof HadoopKerberosConfig)) {
+      return false;
+    }
+
+    HadoopKerberosConfig that = (HadoopKerberosConfig) o;
+
+    if (getPrincipal() != null ? !getPrincipal().equals(that.getPrincipal()) : that.getPrincipal() != null) {
+      return false;
+    }
+    return getKeytab() != null ? getKeytab().equals(that.getKeytab()) : that.getKeytab() == null;
+
+  }
+
+  @Override
+  public int hashCode()
+  {
+    int result = getPrincipal() != null ? getPrincipal().hashCode() : 0;
+    result = 31 * result + (getKeytab() != null ? getKeytab().hashCode() : 0);
+    return result;
+  }
+}