You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2014/08/07 17:10:57 UTC
[07/10] git commit: STORM-346: added AutoHDFS class that will get
hdfs delegation tokens on behalf of users,
push it to workers and renew the delegation tokens automatically.
INimbusCredentialPlugin is a plugin interfae that will be executed as part of
n
STORM-346: added AutoHDFS class that will get hdfs delegation tokens on behalf of users, push it to workers and renew the delegation tokens automatically. INimbusCredentialPlugin is a plugin interfae that will be executed as part of nimbu's submit topology operation and fetch credential from nimbus process on behalf of user. AutoHDFS will renew the tokens on nimbus host as long as credential has more then 24 hours on renew cycle, if tokens are going to expire in next 24 hours then new tokens will be fetched and pushed to all workers.
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/2c78f911
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/2c78f911
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/2c78f911
Branch: refs/heads/security
Commit: 2c78f911e495518b65241ee0904d37d0f854b8b7
Parents: 8973e30
Author: Parth Brahmbhatt <br...@gmail.com>
Authored: Fri Jul 11 11:50:07 2014 -0700
Committer: Parth Brahmbhatt <br...@gmail.com>
Committed: Tue Jul 22 09:42:52 2014 -0700
----------------------------------------------------------------------
.../src/clj/backtype/storm/daemon/nimbus.clj | 40 ++---
storm-core/src/jvm/backtype/storm/Config.java | 18 ++-
.../storm/security/INimbusCredentialPlugin.java | 39 +++++
.../backtype/storm/security/auth/AuthUtils.java | 23 +++
.../security/auth/ICredentialsRenewer.java | 3 +-
.../storm/security/auth/kerberos/AutoHDFS.java | 154 +++++++------------
.../storm/security/auth/kerberos/AutoTGT.java | 2 +-
7 files changed, 152 insertions(+), 127 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2c78f911/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
index e2952d0..3fe1a0a 100644
--- a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
@@ -857,24 +857,6 @@
(swap! (:heartbeats-cache nimbus) dissoc id))
))))
-(defn renew-credentials [nimbus]
- (let [storm-cluster-state (:storm-cluster-state nimbus)
- renewers (:cred-renewers nimbus)
- update-lock (:cred-update-lock nimbus)
- assigned-ids (set (.active-storms storm-cluster-state))]
- (when-not (empty? assigned-ids)
- (doseq [id assigned-ids]
- (locking update-lock
- (let [orig-creds (.credentials storm-cluster-state id nil)]
- (if orig-creds
- (let [new-creds (HashMap. orig-creds)]
- (doseq [renewer renewers]
- (log-message "Renewing Creds For " id " with " renewer)
- (.renew renewer new-creds))
- (when-not (= orig-creds new-creds)
- (.set-credentials! storm-cluster-state id new-creds)
- )))))))))
-
(defn- file-older-than? [now seconds file]
(<= (+ (.lastModified file) (to-millis seconds)) (to-millis now)))
@@ -952,6 +934,25 @@
)
)
+(defn renew-credentials [nimbus]
+ (let [storm-cluster-state (:storm-cluster-state nimbus)
+ renewers (:cred-renewers nimbus)
+ update-lock (:cred-update-lock nimbus)
+ assigned-ids (set (.active-storms storm-cluster-state))]
+ (when-not (empty? assigned-ids)
+ (doseq [id assigned-ids]
+ (locking update-lock
+ (let [orig-creds (.credentials storm-cluster-state id nil)
+ topology-conf (try-read-storm-conf (:conf nimbus) id)]
+ (if orig-creds
+ (let [new-creds (HashMap. orig-creds)]
+ (doseq [renewer renewers]
+ (log-message "Renewing Creds For " id " with " renewer)
+ (.renew renewer new-creds topology-conf))
+ (when-not (= orig-creds new-creds)
+ (.set-credentials! storm-cluster-state id new-creds)
+ )))))))))
+
(defn validate-topology-size [topo-conf nimbus-conf topology]
(let [workers-count (get topo-conf TOPOLOGY-WORKERS)
workers-allowed (get nimbus-conf NIMBUS-SLOTS-PER-TOPOLOGY)
@@ -1046,7 +1047,10 @@
(dissoc storm-conf STORM-ZOOKEEPER-TOPOLOGY-AUTH-SCHEME STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD))
total-storm-conf (merge conf storm-conf)
topology (normalize-topology total-storm-conf topology)
+ nimbus-autocred-plugins (AuthUtils/getNimbusAutoCredPlugins total-storm-conf)
storm-cluster-state (:storm-cluster-state nimbus)]
+ (doseq [nimbus-autocred-plugin nimbus-autocred-plugins]
+ (.populateCredentials nimbus-autocred-plugin credentials total-storm-conf))
(if (and (conf SUPERVISOR-RUN-WORKER-AS-USER) (or (nil? submitter-user) (.isEmpty (.trim submitter-user))))
(throw (AuthorizationException. "Could not determine the user to run this topology as.")))
(system-topology! total-storm-conf topology) ;; this validates the structure of the topology
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2c78f911/storm-core/src/jvm/backtype/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/Config.java b/storm-core/src/jvm/backtype/storm/Config.java
index 785d1e2..3c95f8a 100644
--- a/storm-core/src/jvm/backtype/storm/Config.java
+++ b/storm-core/src/jvm/backtype/storm/Config.java
@@ -426,6 +426,13 @@ public class Config extends HashMap<String, Object> {
public static final Object NIMBUS_CREDENTIAL_RENEWERS_SCHEMA = ConfigValidation.StringsValidator;
/**
+ * A list of plugins that nimbus should load during submit topology to populate
+ * credentials on user's behalf.
+ */
+ public static final String NIMBUS_AUTO_CRED_PLUGINS = "nimbus.autocredential.plugins.classes";
+ public static final Object NIMBUS_AUTO_CRED_PLUGINS_SCHEMA = ConfigValidation.StringsValidator;
+
+ /**
* Storm UI binds to this port.
*/
public static final String UI_PORT = "ui.port";
@@ -1190,15 +1197,10 @@ public class Config extends HashMap<String, Object> {
/**
* HDFS information, used to get the delegation token on behalf of the topology
* submitter user and renew the tokens. see {@link backtype.storm.security.auth.kerberos.AutoHDFS}
+ * kerberos principal name with realm should be provided.
*/
- public static final String HDFS_NAMENODE_URL = "topology.hdfs.namenodeURI";
- public static final Object HDFS_NAMENODE_URL_SCHEMA = String.class;
-
- public static final Object HDFS_USER = "topology.hdfs.user";
- public static final Object HDFS_USER_SCHEMA = String.class;
-
- public static final Object HDFS_USER_KEYTAB = "topology.hdfs.userKeyTab";
- public static final Object HDFS_USER_KEYTAB_SCHEMA = String.class;
+ public static final Object HDFS_PRINCIPAL = "topology.hdfs.user";
+ public static final Object HDFS_PRINCIPAL_SCHEMA = String.class;
public static void setDebug(Map conf, boolean isOn) {
conf.put(Config.TOPOLOGY_DEBUG, isOn);
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2c78f911/storm-core/src/jvm/backtype/storm/security/INimbusCredentialPlugin.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/security/INimbusCredentialPlugin.java b/storm-core/src/jvm/backtype/storm/security/INimbusCredentialPlugin.java
new file mode 100644
index 0000000..4c56863
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/security/INimbusCredentialPlugin.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 backtype.storm.security;
+
+import java.util.Map;
+
+/**
+ * Nimbus auto credential plugin that will be called on nimbus host
+ * during submit topology option. User can specify a list of implementation using config key
+ * nimbus.autocredential.plugins.classes.
+ */
+public interface INimbusCredentialPlugin {
+
+ /**
+ * Method that will be called on nimbus as part of submit topology. This plugin will be called
+ * at least once during the submit Topology action. It will be not be called during activate instead
+ * the credentials return by this method will be merged with the other credentials in the topology
+ * and stored in zookeeper.
+ * @param credentials credentials map where more credentials will be added.
+ * @param conf topology configuration
+ * @return
+ */
+ void populateCredentials(Map<String, String> credentials, Map conf);
+}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2c78f911/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java b/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java
index b5118c7..fc06698 100644
--- a/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java
+++ b/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java
@@ -23,6 +23,8 @@ import javax.security.auth.login.AppConfigurationEntry;
import javax.security.auth.Subject;
import java.security.NoSuchAlgorithmException;
import java.security.URIParameter;
+
+import backtype.storm.security.INimbusCredentialPlugin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
@@ -109,6 +111,27 @@ public class AuthUtils {
}
/**
+ * Get all the Nimbus Auto cred plugins that users want to use.
+ * @param topologyConf topologyConfiguration to use.
+ * @return nimbus auto credential plugins.
+ */
+ public static Collection<INimbusCredentialPlugin> getNimbusAutoCredPlugins(Map topologyConf) {
+ try {
+ Set<INimbusCredentialPlugin> ret = new HashSet<INimbusCredentialPlugin>();
+ Collection<String> clazzes = (Collection<String>)topologyConf.get(Config.NIMBUS_AUTO_CRED_PLUGINS);
+ if (clazzes != null) {
+ for (String clazz : clazzes) {
+ INimbusCredentialPlugin inst = (INimbusCredentialPlugin)Class.forName(clazz).newInstance();
+ ret.add(inst);
+ }
+ }
+ return ret;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
* Get all of the configured AutoCredential Plugins.
* @param storm_conf the storm configuration to use.
* @return the configured auto credentials.
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2c78f911/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java b/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java
index a84a9e8..3eaf6c4 100644
--- a/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java
+++ b/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java
@@ -35,6 +35,7 @@ public interface ICredentialsRenewer {
/**
* Renew any credentials that need to be renewed. (Update the credentials if needed)
* @param credentials the credentials that may have something to renew.
+ * @param topologyConf topology configuration.
*/
- public void renew(Map<String, String> credentials);
+ public void renew(Map<String, String> credentials, Map topologyConf);
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2c78f911/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoHDFS.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoHDFS.java b/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoHDFS.java
index 80b5916..7685fb1 100644
--- a/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoHDFS.java
+++ b/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoHDFS.java
@@ -19,6 +19,7 @@
package backtype.storm.security.auth.kerberos;
import backtype.storm.Config;
+import backtype.storm.security.INimbusCredentialPlugin;
import backtype.storm.security.auth.IAutoCredentials;
import backtype.storm.security.auth.ICredentialsRenewer;
import org.slf4j.Logger;
@@ -30,88 +31,28 @@ import java.io.*;
import java.lang.reflect.Method;
import java.net.URI;
import java.util.Collection;
+import java.util.HashMap;
import java.util.Map;
/**
* Automatically get HDFS delegation tokens and push it to user's topology. The class
* assumes that HDFS configuration files are in your class path.
*/
-public class AutoHDFS implements IAutoCredentials, ICredentialsRenewer {
+public class AutoHDFS implements IAutoCredentials, ICredentialsRenewer, INimbusCredentialPlugin {
private static final Logger LOG = LoggerFactory.getLogger(AutoHDFS.class);
public static final String HDFS_CREDENTIALS = "HDFS_CREDENTIALS";
- private static final String CONF_KEYTAB_KEY = "keytab";
- private static final String CONF_USER_KEY = "user";
-
- private Map conf;
public void prepare(Map conf) {
- this.conf = conf;
+ LOG.debug("no op.");
}
@SuppressWarnings("unchecked")
- private Object getConfiguration() {
- try {
- final String hdfsUser = (String) conf.get(Config.HDFS_USER);
- final String hdfsUserKeyTab = (String) conf.get(Config.HDFS_USER_KEYTAB);
-
- /**
- * Configuration configuration = new Configuration();
- * configuration.set(CONF_KEYTAB_KEY, hdfsUserKeyTab);
- * configuration.set(CONF_USER_KEY, hdfsUser);
- */
- Class configurationClass = Class.forName("org.apache.hadoop.conf.Configuration");
- Object configuration = configurationClass.newInstance();
-
- Method setMethod = configurationClass.getMethod("set", String.class, String.class);
- setMethod.invoke(configuration, CONF_KEYTAB_KEY, hdfsUserKeyTab);
- setMethod.invoke(configuration, CONF_USER_KEY, hdfsUser);
- /**
- * Following are the minimum set of configuration that needs to be set, users should have hdfs-site.xml
- * and core-site.xml in the class path which should set these configuration.
- * setMethod.invoke(configuration, "hadoop.security.authentication", "KERBEROS");
- * setMethod.invoke(configuration,"dfs.namenode.kerberos.principal",
- * "hdfs/zookeeper.witzend.com@WITZEND.COM");
- * setMethod.invoke(configuration, "hadoop.security.kerberos.ticket.cache.path", "/tmp/krb5cc_1002");
- */
-
- setMethod.invoke(configuration, "hadoop.security.authentication", "KERBEROS");
- setMethod.invoke(configuration, "dfs.namenode.kerberos.principal","hdfs/zookeeper.witzend.com@WITZEND.COM");
- setMethod.invoke(configuration, "hadoop.security.kerberos.ticket.cache.path", "/tmp/krb5cc_1002");
-
- //UserGroupInformation.setConfiguration(configuration);
- final Class ugiClass = Class.forName("org.apache.hadoop.security.UserGroupInformation");
- Method setConfigurationMethod = ugiClass.getMethod("setConfiguration", configurationClass);
- setConfigurationMethod.invoke(null, configuration);
- return configuration;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @SuppressWarnings("unchecked")
- private void login(Object configuration) {
- try {
- Class configurationClass = Class.forName("org.apache.hadoop.conf.Configuration");
- final Class securityUtilClass = Class.forName("org.apache.hadoop.security.SecurityUtil");
- Method loginMethod = securityUtilClass.getMethod("login", configurationClass, String.class, String.class);
- loginMethod.invoke(null, configuration, CONF_KEYTAB_KEY, CONF_USER_KEY);
- } catch (Exception e) {
- throw new RuntimeException("Failed to login to hdfs .", e);
- }
- }
-
- @SuppressWarnings("unchecked")
- private byte[] getHDFSCredsWithDelegationToken() throws Exception {
+ private byte[] getHDFSCredsWithDelegationToken(Map conf) throws Exception {
try {
/**
* What we want to do is following:
- * Configuration configuration = new Configuration();
- * configuration.set(CONF_KEYTAB_KEY, hdfsUserKeyTab);
- * configuration.set(CONF_USER_KEY, hdfsUser);
- * UserGroupInformation.setConfiguration(configuration);
* if(UserGroupInformation.isSecurityEnabled) {
- * SecurityUtil.login(configuration, CONF_KEYTAB_KEY, CONF_USER_KEY);
* FileSystem fs = FileSystem.get(nameNodeURI, configuration, topologySubmitterUser);
* UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
* UserGroupInformation proxyUser = UserGroupInformation.createProxyUser(topologySubmitterUser, ugi);
@@ -119,24 +60,32 @@ public class AutoHDFS implements IAutoCredentials, ICredentialsRenewer {
* fs.addDelegationToken(hdfsUser, credential);
* }
* and then return the credential object as a bytearray.
+ *
+ * Following are the minimum set of configuration that needs to be set, users should have hdfs-site.xml
+ * and core-site.xml in the class path which should set these configuration.
+ * configuration.set("hadoop.security.authentication", "KERBEROS");
+ * configuration.set("dfs.namenode.kerberos.principal",
+ * "hdfs/zookeeper.witzend.com@WITZEND.COM");
+ * configuration.set("hadoop.security.kerberos.ticket.cache.path", "/tmp/krb5cc_1002");
+ * anf the ticket cache must have the hdfs user's creds.
*/
- Object configuration = getConfiguration();
+ Class configurationClass = Class.forName("org.apache.hadoop.conf.Configuration");
+ Object configuration = configurationClass.newInstance();
+
+ //UserGroupInformation.isSecurityEnabled
final Class ugiClass = Class.forName("org.apache.hadoop.security.UserGroupInformation");
final Method isSecurityEnabledMethod = ugiClass.getDeclaredMethod("isSecurityEnabled");
boolean isSecurityEnabled = (Boolean)isSecurityEnabledMethod.invoke(null);
- if(isSecurityEnabled) {
- login(configuration);
- final URI nameNodeURI = URI.create((String) conf.get(Config.HDFS_NAMENODE_URL));
+ if(isSecurityEnabled) {
final String topologySubmitterUser = (String) conf.get(Config.TOPOLOGY_SUBMITTER_USER);
- final String hdfsUser = (String) conf.get(Config.HDFS_USER);
-
- Class configurationClass = Class.forName("org.apache.hadoop.conf.Configuration");
+ final String hdfsUser = (String) conf.get(Config.HDFS_PRINCIPAL);
//FileSystem fs = FileSystem.get(nameNodeURI, configuration, topologySubmitterUser);
Class fileSystemClass = Class.forName("org.apache.hadoop.fs.FileSystem");
+ Object defaultNameNodeURI = fileSystemClass.getMethod("getDefaultUri", configurationClass).invoke(null, configuration);
Method getMethod = fileSystemClass.getMethod("get", URI.class, configurationClass, String.class);
- Object fileSystem = getMethod.invoke(null, nameNodeURI, configuration, topologySubmitterUser);
+ Object fileSystem = getMethod.invoke(null, defaultNameNodeURI, configuration, topologySubmitterUser);
//UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
Method getCurrentUserMethod = ugiClass.getMethod("getCurrentUser");
@@ -164,7 +113,6 @@ public class AutoHDFS implements IAutoCredentials, ICredentialsRenewer {
out.flush();
out.close();
- LOG.info(bao.toString());
return bao.toByteArray();
} else {
throw new RuntimeException("Security is not enabled for HDFS");
@@ -175,14 +123,19 @@ public class AutoHDFS implements IAutoCredentials, ICredentialsRenewer {
}
@Override
- public void populateCredentials(Map<String, String> credentials) {
+ public void populateCredentials(Map<String, String> credentials, Map conf) {
try {
- credentials.put(HDFS_CREDENTIALS, DatatypeConverter.printBase64Binary( getHDFSCredsWithDelegationToken()));
+ credentials.put(HDFS_CREDENTIALS, DatatypeConverter.printBase64Binary( getHDFSCredsWithDelegationToken(conf)));
} catch (Exception e) {
LOG.warn("Could not populate HDFS credentials.", e);
}
}
+ @Override
+ public void populateCredentials(Map<String, String> credentials) {
+ LOG.debug("populateCredentials is a noop, nimbus should populate the crdes.");
+ }
+
/**
*
* @param credentials map with creds.
@@ -235,63 +188,66 @@ public class AutoHDFS implements IAutoCredentials, ICredentialsRenewer {
@Override
@SuppressWarnings("unchecked")
- public void renew(Map<String,String> credentials) {
+ public void renew(Map<String, String> credentials, Map topologyConf) {
Object credential = getHDFSCredential(credentials);
+ //maximum allowed expiration time until which tokens will keep renewing,
+ //currently set to 1 day.
+ final long MAX_ALLOWED_EXPIRATION_MILLIS = 24 * 60 * 60 * 1000;
+
/**
* We are trying to do the following :
* List<Token> tokens = credential.getAllTokens();
* for(Token token: tokens) {
- * token.renew(configuration);
+ * long expiration = token.renew(configuration);
* }
- * TODO: Need to talk to HDFS guys to check what is recommended way to identify a token that is beyond
- * renew cycle, once we can identified tokens are not renewable we can repopulate credentials by asking HDFS
- * to issue new tokens. Until we have a better way any exception during renewal would result in attempt to
- * get new delegation tokens.
*/
- if (credential != null) {
- try {
- Object configuration = getConfiguration();
-
+ try {
+ if (credential != null) {
Class configurationClass = Class.forName("org.apache.hadoop.conf.Configuration");
+ Object configuration = configurationClass.newInstance();
+
Class credentialClass = Class.forName("org.apache.hadoop.security.Credentials");
Class tokenClass = Class.forName("org.apache.hadoop.security.token.Token");
- Method renewMethod = tokenClass.getMethod("renew", configurationClass);
+ Method renewMethod = tokenClass.getMethod("renew", configurationClass);
Method getAllTokensMethod = credentialClass.getMethod("getAllTokens");
Collection<?> tokens = (Collection<?>) getAllTokensMethod.invoke(credential);
- for(Object token : tokens) {
- renewMethod.invoke(token, configuration);
+ for (Object token : tokens) {
+ long expiration = (Long) renewMethod.invoke(token, configuration);
+ if(expiration < MAX_ALLOWED_EXPIRATION_MILLIS) {
+ LOG.debug("expiration {} is less then MAX_ALLOWED_EXPIRATION_MILLIS {}, getting new tokens",
+ expiration, MAX_ALLOWED_EXPIRATION_MILLIS);
+ populateCredentials(credentials, topologyConf);
+ }
}
- } catch(Exception e) {
- LOG.warn("could not renew the credentials, one of the possible reason is tokens are beyond " +
- "renewal period so attempting to get new tokens.", e);
- populateCredentials(credentials);
}
+ } catch (Exception e) {
+ LOG.warn("could not renew the credentials, one of the possible reason is tokens are beyond " +
+ "renewal period so attempting to get new tokens.", e);
+ populateCredentials(credentials);
}
}
@SuppressWarnings("unchecked")
public static void main(String[] args) throws Exception {
Map conf = new java.util.HashMap();
- conf.put(Config.HDFS_NAMENODE_URL, args[0]);
- conf.put(Config.TOPOLOGY_SUBMITTER_USER, args[1]); //with realm e.g. storm@WITZEND.COM
- conf.put(Config.HDFS_USER, args[2]); //with realm e.g. hdfs@WITZEND.COM
- conf.put(Config.HDFS_USER_KEYTAB, args[3]);
+ conf.put(Config.TOPOLOGY_SUBMITTER_PRINCIPAL, args[0]); //with realm e.g. storm@WITZEND.COM
+ conf.put(Config.HDFS_PRINCIPAL, args[1]); //with realm e.g. hdfs@WITZEND.COM
AutoHDFS autoHDFS = new AutoHDFS();
autoHDFS.prepare(conf);
- Map<String,String> creds = new java.util.HashMap<String,String>();
- autoHDFS.populateCredentials(creds);
+ Map<String,String> creds = new HashMap<String, String>();
+ autoHDFS.populateCredentials(creds, conf);
LOG.info("Got HDFS credentials", AutoHDFS.getHDFSCredential(creds));
Subject s = new Subject();
autoHDFS.populateSubject(s, creds);
LOG.info("Got a Subject "+ s);
- autoHDFS.renew(creds);
+ autoHDFS.renew(creds, conf);
LOG.info("renewed credentials", AutoHDFS.getHDFSCredential(creds));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2c78f911/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoTGT.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoTGT.java b/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoTGT.java
index 52bf540..3bbb0d3 100644
--- a/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoTGT.java
+++ b/storm-core/src/jvm/backtype/storm/security/auth/kerberos/AutoTGT.java
@@ -236,7 +236,7 @@ public class AutoTGT implements IAutoCredentials, ICredentialsRenewer {
}
@Override
- public void renew(Map<String,String> credentials) {
+ public void renew(Map<String,String> credentials, Map topologyConf) {
KerberosTicket tgt = getTGT(credentials);
if (tgt != null) {
long refreshTime = getRefreshTime(tgt);