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 2018/02/09 02:38:35 UTC
[5/6] storm git commit: STORM-2898: Support for WorkerToken
authentication
STORM-2898: Support for WorkerToken authentication
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9566a882
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9566a882
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9566a882
Branch: refs/heads/master
Commit: 9566a8823a61b753cb176f7394b7a37b825281c3
Parents: 5f05739
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Fri Jan 12 16:37:56 2018 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Thu Feb 8 10:22:39 2018 -0600
----------------------------------------------------------------------
storm-client/pom.xml | 2 +-
.../src/jvm/org/apache/storm/Config.java | 8 +
.../storm/cluster/ClusterStateContext.java | 22 +-
.../org/apache/storm/cluster/ClusterUtils.java | 58 +-
.../org/apache/storm/cluster/DaemonType.java | 90 ++-
.../storm/cluster/IStormClusterState.java | 67 +-
.../cluster/PaceMakerStateStorageFactory.java | 7 +-
.../storm/cluster/StateStorageFactory.java | 6 +-
.../storm/cluster/StormClusterStateImpl.java | 189 ++++-
.../apache/storm/cluster/ZKStateStorage.java | 4 +-
.../storm/cluster/ZKStateStorageFactory.java | 4 +-
.../org/apache/storm/daemon/worker/Worker.java | 10 +-
.../jvm/org/apache/storm/executor/Executor.java | 4 +-
.../storm/generated/PrivateWorkerKey.java | 615 ++++++++++++++++
.../org/apache/storm/generated/WorkerToken.java | 638 +++++++++++++++++
.../apache/storm/generated/WorkerTokenInfo.java | 701 +++++++++++++++++++
.../storm/generated/WorkerTokenServiceType.java | 62 ++
.../auth/AbstractSaslClientCallbackHandler.java | 76 --
.../auth/AbstractSaslServerCallbackHandler.java | 94 ---
.../apache/storm/security/auth/AuthUtils.java | 164 ++++-
.../storm/security/auth/ITransportPlugin.java | 12 +-
.../apache/storm/security/auth/ReqContext.java | 11 +
.../security/auth/SaslTransportPlugin.java | 187 -----
.../security/auth/SimpleTransportPlugin.java | 4 +-
.../storm/security/auth/ThriftClient.java | 2 +-
.../security/auth/ThriftConnectionType.java | 81 ++-
.../storm/security/auth/ThriftServer.java | 53 +-
.../auth/digest/ClientCallbackHandler.java | 58 --
.../auth/digest/DigestSaslTransportPlugin.java | 60 +-
.../auth/digest/JassPasswordProvider.java | 85 +++
.../auth/digest/ServerCallbackHandler.java | 85 ---
.../kerberos/KerberosSaslTransportPlugin.java | 73 +-
.../auth/kerberos/ServerCallbackHandler.java | 86 +--
.../auth/plain/PlainClientCallbackHandler.java | 16 +-
.../auth/plain/PlainSaslTransportPlugin.java | 18 +-
.../auth/plain/PlainServerCallbackHandler.java | 55 --
.../security/auth/plain/SaslPlainServer.java | 2 +
.../security/auth/sasl/PasswordProvider.java | 48 ++
.../security/auth/sasl/SaslTransportPlugin.java | 187 +++++
.../sasl/SimpleSaslClientCallbackHandler.java | 77 ++
.../sasl/SimpleSaslServerCallbackHandler.java | 189 +++++
.../auth/workertoken/WorkerTokenAuthorizer.java | 139 ++++
.../WorkerTokenClientCallbackHandler.java | 65 ++
.../auth/workertoken/WorkerTokenSigner.java | 63 ++
.../org/apache/storm/utils/ObjectReader.java | 12 +
.../src/jvm/org/apache/storm/utils/Utils.java | 39 +-
storm-client/src/py/storm/ttypes.py | 320 +++++++++
storm-client/src/storm.thrift | 41 +-
.../apache/storm/cluster/DaemonTypeTest.java | 78 +++
.../cluster/StormClusterStateImplTest.java | 18 +-
.../security/auth/SaslTransportPluginTest.java | 1 +
.../org/apache/storm/command/AdminCommands.java | 24 +-
.../org/apache/storm/command/Heartbeats.java | 2 +-
.../test/clj/org/apache/storm/cluster_test.clj | 10 +-
.../test/clj/org/apache/storm/nimbus_test.clj | 47 +-
.../apache/storm/security/auth/auth_test.clj | 474 -------------
.../security/auth/jaas_digest_bad_password.conf | 31 -
.../auth/jaas_digest_missing_client.conf | 23 -
.../security/auth/jaas_digest_unknown_user.conf | 31 -
.../storm/security/auth/nimbus_auth_test.clj | 5 +-
storm-server/pom.xml | 2 +-
.../java/org/apache/storm/DaemonConfig.java | 6 +
.../java/org/apache/storm/LocalCluster.java | 7 +-
.../src/main/java/org/apache/storm/Testing.java | 6 +-
.../org/apache/storm/daemon/nimbus/Nimbus.java | 76 +-
.../storm/daemon/supervisor/Supervisor.java | 24 +-
.../daemon/supervisor/SupervisorUtils.java | 7 -
.../auth/workertoken/WorkerTokenManager.java | 126 ++++
.../apache/storm/security/auth/AuthTest.java | 637 +++++++++++++++++
.../auth/workertoken/WorkerTokenTest.java | 159 +++++
.../src/test/resources/jaas_digest.conf | 30 +
.../resources/jaas_digest_bad_password.conf | 31 +
.../resources/jaas_digest_missing_client.conf | 23 +
.../resources/jaas_digest_unknown_user.conf | 31 +
74 files changed, 5317 insertions(+), 1481 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/pom.xml
----------------------------------------------------------------------
diff --git a/storm-client/pom.xml b/storm-client/pom.xml
index 22c8bcc..e0c9609 100644
--- a/storm-client/pom.xml
+++ b/storm-client/pom.xml
@@ -261,7 +261,7 @@
<!--Note - the version would be inherited-->
<configuration>
<excludes>**/generated/**</excludes>
- <maxAllowedViolations>10263</maxAllowedViolations>
+ <maxAllowedViolations>10079</maxAllowedViolations>
</configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java
index 7014dfd..d8bbcaa 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -1147,6 +1147,14 @@ public class Config extends HashMap<String, Object> {
public static final String STORM_ZOOKEEPER_SUPERACL = "storm.zookeeper.superACL";
/**
+ * The ACL of the drpc user in zookeeper so the drpc servers can verify worker tokens.
+ *
+ * Should be in the form 'scheme:acl' just like STORM_ZOOKEEPER_SUPERACL.
+ */
+ @isString
+ public static final String STORM_ZOOKEEPER_DRPC_ACL = "storm.zookeeper.drpcACL";
+
+ /**
* The topology Zookeeper authentication scheme to use, e.g. "digest". It is the internal config and user shouldn't set it.
*/
@isString
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/ClusterStateContext.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/ClusterStateContext.java b/storm-client/src/jvm/org/apache/storm/cluster/ClusterStateContext.java
index 9ad6a92..2c4351e 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/ClusterStateContext.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/ClusterStateContext.java
@@ -18,24 +18,40 @@
package org.apache.storm.cluster;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.storm.generated.WorkerTokenServiceType;
+import org.apache.zookeeper.data.ACL;
+
/**
* This class is intended to provide runtime-context to StateStorageFactory
* implementors, giving information such as what daemon is creating it.
*/
public class ClusterStateContext {
-
+
+ private final Map<String, Object> conf;
private DaemonType daemonType;
public ClusterStateContext() {
daemonType = DaemonType.UNKNOWN;
+ conf = new HashMap<>();
}
- public ClusterStateContext(DaemonType daemonType) {
+ public ClusterStateContext(DaemonType daemonType, Map<String, Object> conf) {
this.daemonType = daemonType;
+ this.conf = conf;
}
public DaemonType getDaemonType() {
return daemonType;
}
-
+
+ public List<ACL> getDefaultZkAcls() {
+ return daemonType.getDefaultZkAcls(conf);
+ }
+
+ public List<ACL> getZkSecretAcls(WorkerTokenServiceType type) {
+ return daemonType.getZkSecretAcls(type, conf);
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/ClusterUtils.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/ClusterUtils.java b/storm-client/src/jvm/org/apache/storm/cluster/ClusterUtils.java
index 0dc3e20..c61c104 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/ClusterUtils.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/ClusterUtils.java
@@ -22,6 +22,7 @@ import org.apache.storm.generated.ClusterWorkerHeartbeat;
import org.apache.storm.generated.ExecutorInfo;
import org.apache.storm.generated.ExecutorStats;
import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.generated.WorkerTokenServiceType;
import org.apache.storm.utils.Utils;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
@@ -55,6 +56,7 @@ public class ClusterUtils {
public static final String CREDENTIALS_ROOT = "credentials";
public static final String LOGCONFIG_ROOT = "logconfigs";
public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+ public static final String SECRET_KEYS_ROOT = "secretkeys";
public static final String ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
public static final String STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
@@ -68,6 +70,7 @@ public class ClusterUtils {
public static final String CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
public static final String LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
public static final String PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ public static final String SECRET_KEYS_SUBTREE = ZK_SEPERATOR + SECRET_KEYS_ROOT;
// A singleton instance allows us to mock delegated static methods in our
// tests by subclassing.
@@ -174,6 +177,11 @@ public class ClusterUtils {
return CREDENTIALS_SUBTREE + ZK_SEPERATOR + stormId;
}
+ /**
+ * Get the path to the log config for a topology.
+ * @param stormId the topology id.
+ * @return the path to the config.
+ */
public static String logConfigPath(String stormId) {
return LOGCONFIG_SUBTREE + ZK_SEPERATOR + stormId;
}
@@ -186,6 +194,36 @@ public class ClusterUtils {
return profilerConfigPath(stormId) + ZK_SEPERATOR + host + "_" + port + "_" + requestType;
}
+ /**
+ * Get the base path where secret keys are stored for a given service.
+ * @param type the service we are interested in.
+ * @return the path to that service root.
+ */
+ public static String secretKeysPath(WorkerTokenServiceType type) {
+ return SECRET_KEYS_SUBTREE + ZK_SEPERATOR + type.name();
+ }
+
+ /**
+ * Get the path to secret keys for a specific topology
+ * @param type the service the secret is for.
+ * @param topologyId the topology the secret is for.
+ * @return the path to the list of secret keys.
+ */
+ public static String secretKeysPath(WorkerTokenServiceType type, String topologyId) {
+ return secretKeysPath(type) + ZK_SEPERATOR + topologyId;
+ }
+
+ /**
+ * Get the path to a specific secret key.
+ * @param type the service the secret is for.
+ * @param topologyId the topology the secret is for.
+ * @param version the version the secret is for.
+ * @return the path to the secret.
+ */
+ public static String secretKeysPath(WorkerTokenServiceType type, String topologyId, long version) {
+ return secretKeysPath(type, topologyId) + ZK_SEPERATOR + version;
+ }
+
public static <T> T maybeDeserialize(byte[] serialized, Class<T> clazz) {
if (serialized != null) {
return Utils.deserialize(serialized, clazz);
@@ -214,16 +252,16 @@ public class ClusterUtils {
return executorWhb;
}
- public IStormClusterState mkStormClusterStateImpl(Object stateStorage, List<ACL> acls, ClusterStateContext context) throws Exception {
+ public IStormClusterState mkStormClusterStateImpl(Object stateStorage, ClusterStateContext context) throws Exception {
if (stateStorage instanceof IStateStorage) {
- return new StormClusterStateImpl((IStateStorage) stateStorage, acls, context, false);
+ return new StormClusterStateImpl((IStateStorage) stateStorage, context, false);
} else {
- IStateStorage Storage = _instance.mkStateStorageImpl((Map) stateStorage, (Map) stateStorage, acls, context);
- return new StormClusterStateImpl(Storage, acls, context, true);
+ IStateStorage Storage = _instance.mkStateStorageImpl((Map) stateStorage, (Map) stateStorage, context);
+ return new StormClusterStateImpl(Storage, context, true);
}
}
- public IStateStorage mkStateStorageImpl(Map<String, Object> config, Map auth_conf, List<ACL> acls, ClusterStateContext context) throws Exception {
+ public IStateStorage mkStateStorageImpl(Map<String, Object> config, Map auth_conf, ClusterStateContext context) throws Exception {
String className = null;
IStateStorage stateStorage = null;
if (config.get(Config.STORM_CLUSTER_STATE_STORE) != null) {
@@ -233,16 +271,16 @@ public class ClusterUtils {
}
Class clazz = Class.forName(className);
StateStorageFactory storageFactory = (StateStorageFactory) clazz.newInstance();
- stateStorage = storageFactory.mkStore(config, auth_conf, acls, context);
+ stateStorage = storageFactory.mkStore(config, auth_conf, context);
return stateStorage;
}
- public static IStateStorage mkStateStorage(Map<String, Object> config, Map auth_conf, List<ACL> acls, ClusterStateContext context) throws Exception {
- return _instance.mkStateStorageImpl(config, auth_conf, acls, context);
+ public static IStateStorage mkStateStorage(Map<String, Object> config, Map auth_conf, ClusterStateContext context) throws Exception {
+ return _instance.mkStateStorageImpl(config, auth_conf, context);
}
- public static IStormClusterState mkStormClusterState(Object StateStorage, List<ACL> acls, ClusterStateContext context) throws Exception {
- return _instance.mkStormClusterStateImpl(StateStorage, acls, context);
+ public static IStormClusterState mkStormClusterState(Object StateStorage, ClusterStateContext context) throws Exception {
+ return _instance.mkStormClusterStateImpl(StateStorage, context);
}
public static String stringifyError(Throwable error) {
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/DaemonType.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/DaemonType.java b/storm-client/src/jvm/org/apache/storm/cluster/DaemonType.java
index 4589be3..8593b7b 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/DaemonType.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/DaemonType.java
@@ -18,10 +18,92 @@
package org.apache.storm.cluster;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.generated.WorkerTokenServiceType;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The type of process/daemon that this server is running as. This is used with the IStormClusterState implementation to know
+ * how to properly secure data stored in it.
+ */
public enum DaemonType {
- SUPERVISOR,
- NIMBUS,
- WORKER,
+ SUPERVISOR {
+ @Override
+ public List<ACL> getDefaultZkAcls(Map<String, Object> conf) {
+ return getDefaultNimbusSupervisorZkAcls(conf);
+ }
+ },
+ NIMBUS {
+ @Override
+ public List<ACL> getDefaultZkAcls(Map<String, Object> conf) {
+ return getDefaultNimbusSupervisorZkAcls(conf);
+ }
+
+ @Override
+ public List<ACL> getZkSecretAcls(WorkerTokenServiceType type, Map<String, Object> conf) {
+ if (!Utils.isZkAuthenticationConfiguredStormServer(conf)) {
+ //This is here only for testing.
+ LOG.error("Will Store Worker Token Keys in ZK without ACLs. If you are not running tests STOP NOW!");
+ return null;
+ }
+ switch (type) {
+ case NIMBUS:
+ return ZooDefs.Ids.CREATOR_ALL_ACL;
+ case DRPC:
+ List<ACL> ret = new ArrayList<>(ZooDefs.Ids.CREATOR_ALL_ACL);
+ ret.add(new ACL(ZooDefs.Perms.READ,
+ Utils.parseZkId((String)conf.get(Config.STORM_ZOOKEEPER_DRPC_ACL), Config.STORM_ZOOKEEPER_DRPC_ACL)));
+ return ret;
+ default:
+ throw new IllegalStateException("WorkerTokens for " + type + " are not currently supported.");
+ }
+ }
+ },
+ WORKER {
+ @Override
+ public List<ACL> getDefaultZkAcls(Map<String, Object> conf) {
+ return Utils.getWorkerACL(conf);
+ }
+ },
PACEMAKER,
- UNKNOWN
+ UNKNOWN;
+
+ private static final Logger LOG = LoggerFactory.getLogger(DaemonType.class);
+ @VisibleForTesting
+ public static final List<ACL> NIMBUS_SUPERVISOR_ZK_ACLS = Arrays.asList(ZooDefs.Ids.CREATOR_ALL_ACL.get(0),
+ new ACL(ZooDefs.Perms.READ | ZooDefs.Perms.CREATE, ZooDefs.Ids.ANYONE_ID_UNSAFE));
+ private static List<ACL> getDefaultNimbusSupervisorZkAcls(Map<String, Object> conf) {
+ if (Utils.isZkAuthenticationConfiguredStormServer(conf)) {
+ return NIMBUS_SUPERVISOR_ZK_ACLS;
+ }
+ return null;
+ }
+
+ /**
+ * Get the default ZK ACLs that should be used for a given daemon type.
+ * @param conf the config used to help get the correct ACLs.
+ * @return the list of acls. null means no acls.
+ */
+ public List<ACL> getDefaultZkAcls(Map<String, Object> conf) {
+ return null;
+ }
+
+ /**
+ * Get the ACLs to store a secret for a given service.
+ * @param type the type of service the secret is for.
+ * @param conf the cluster configuration.
+ * @return the ACLs
+ */
+ public List<ACL> getZkSecretAcls(WorkerTokenServiceType type, Map<String, Object> conf) {
+ throw new IllegalArgumentException(name() + " does not support storing secrets.");
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/IStormClusterState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/IStormClusterState.java b/storm-client/src/jvm/org/apache/storm/cluster/IStormClusterState.java
index 051ccd5..adc434c 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/IStormClusterState.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/IStormClusterState.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.cluster;
import java.security.NoSuchAlgorithmException;
@@ -22,7 +23,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-
+import java.util.Set;
import org.apache.storm.generated.Assignment;
import org.apache.storm.generated.ClusterWorkerHeartbeat;
import org.apache.storm.generated.Credentials;
@@ -31,9 +32,11 @@ import org.apache.storm.generated.ExecutorInfo;
import org.apache.storm.generated.LogConfig;
import org.apache.storm.generated.NimbusSummary;
import org.apache.storm.generated.NodeInfo;
+import org.apache.storm.generated.PrivateWorkerKey;
import org.apache.storm.generated.ProfileRequest;
import org.apache.storm.generated.StormBase;
import org.apache.storm.generated.SupervisorInfo;
+import org.apache.storm.generated.WorkerTokenServiceType;
import org.apache.storm.nimbus.NimbusInfo;
public interface IStormClusterState {
@@ -140,9 +143,65 @@ public interface IStormClusterState {
public Credentials credentials(String stormId, Runnable callback);
public void disconnect();
-
+
/**
- * @return All of the supervisors with the ID as the key
+ * Get a private key used to validate a token is correct.
+ * This is expected to be called from a privileged daemon, and the ACLs should be set up to only
+ * allow nimbus and these privileged daemons access to these private keys.
+ * @param type the type of service the key is for.
+ * @param topologyId the topology id the key is for.
+ * @param keyVersion the version of the key this is for.
+ * @return the private key or null if it could not be found.
+ */
+ PrivateWorkerKey getPrivateWorkerKey(WorkerTokenServiceType type, String topologyId, long keyVersion);
+
+ /**
+ * Store a new version of a private key.
+ * This is expected to only ever be called from nimbus. All ACLs however need to be setup to allow
+ * the given services access to the stored information.
+ * @param type the type of service this key is for.
+ * @param topologyId the topology this key is for
+ * @param keyVersion the version of the key this is for.
+ * @param key the key to store.
+ */
+ void addPrivateWorkerKey(WorkerTokenServiceType type, String topologyId, long keyVersion, PrivateWorkerKey key);
+
+ /**
+ * Get the next key version number that should be used for this topology id.
+ * This is expected to only ever be called from nimbus, but it is acceptable if the ACLs are setup
+ * so that it can work from a privileged daemon for the given service.
+ * @param type the type of service this is for.
+ * @param topologyId the topology id this is for.
+ * @return the next version number. It should be 0 for a new topology id/service combination.
+ */
+ long getNextPrivateWorkerKeyVersion(WorkerTokenServiceType type, String topologyId);
+
+ /**
+ * Remove all keys for the given topology that have expired. The number of keys should be small enough
+ * that doing an exhaustive scan of them all is acceptable as there is no guarantee that expiration time
+ * and version number are related. This should be for all service types.
+ * This is expected to only ever be called from nimbus and some ACLs may be setup so being called from other
+ * daemons will cause it to fail.
+ * @param topologyId the id of the topology to scan.
+ */
+ void removeExpiredPrivateWorkerKeys(String topologyId);
+
+ /**
+ * Remove all of the worker keys for a given topology. Used to clean up after a topology finishes.
+ * This is expected to only ever be called from nimbus and ideally should only ever work from nimbus.
+ * @param topologyId the topology to clean up after.
+ */
+ void removeAllPrivateWorkerKeys(String topologyId);
+
+ /**
+ * Get a list of all topologyIds that currently have private worker keys stored, of any kind.
+ * This is expected to only ever be called from nimbus.
+ * @return the list of topology ids with any kind of private worker key stored.
+ */
+ Set<String> idsOfTopologiesWithPrivateWorkerKeys();
+
+ /**
+ * Get all of the supervisors with the ID as the key.
*/
default Map<String, SupervisorInfo> allSupervisorInfo() {
return allSupervisorInfo(null);
@@ -169,7 +228,7 @@ public interface IStormClusterState {
String ret = null;
for (String topoId: activeStorms()) {
StormBase base = stormBase(topoId, null);
- if(base != null && topologyName.equals(base.get_name())) {
+ if (base != null && topologyName.equals(base.get_name())) {
ret = topoId;
break;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java b/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
index e4e1552..065e0df 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
@@ -15,20 +15,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.cluster;
-import java.util.List;
import java.util.Map;
import org.apache.storm.pacemaker.PacemakerClientPool;
import org.apache.storm.utils.Utils;
-import org.apache.zookeeper.data.ACL;
public class PaceMakerStateStorageFactory implements StateStorageFactory {
@Override
- public IStateStorage mkStore(Map<String, Object> config, Map auth_conf, List<ACL> acls, ClusterStateContext context) {
+ public IStateStorage mkStore(Map<String, Object> config, Map auth_conf, ClusterStateContext context) {
try {
ZKStateStorageFactory zkfact = new ZKStateStorageFactory();
- IStateStorage zkState = zkfact.mkStore(config, auth_conf, acls, context);
+ IStateStorage zkState = zkfact.mkStore(config, auth_conf, context);
return new PaceMakerStateStorage(new PacemakerClientPool(config), zkState);
} catch (Exception e) {
throw Utils.wrapInRuntime(e);
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/StateStorageFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/StateStorageFactory.java b/storm-client/src/jvm/org/apache/storm/cluster/StateStorageFactory.java
index 2619104..584a2d8 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/StateStorageFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/StateStorageFactory.java
@@ -15,14 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.cluster;
-import java.util.List;
import java.util.Map;
-import org.apache.zookeeper.data.ACL;
-
public interface StateStorageFactory {
- IStateStorage mkStore(Map<String, Object> config, Map auth_conf, List<ACL> acls, ClusterStateContext context);
+ IStateStorage mkStore(Map<String, Object> config, Map auth_conf, ClusterStateContext context);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java b/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
index 7a52d86..11757c0 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
@@ -15,29 +15,50 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.cluster;
+import java.nio.ByteBuffer;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.state.*;
import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.storm.callback.ZKStateChangedCallback;
-import org.apache.storm.generated.*;
+import org.apache.storm.generated.Assignment;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.Credentials;
+import org.apache.storm.generated.DebugOptions;
+import org.apache.storm.generated.ErrorInfo;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.LogConfig;
+import org.apache.storm.generated.NimbusSummary;
+import org.apache.storm.generated.NodeInfo;
+import org.apache.storm.generated.PrivateWorkerKey;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.generated.ProfileRequest;
+import org.apache.storm.generated.StormBase;
+import org.apache.storm.generated.SupervisorInfo;
+import org.apache.storm.generated.WorkerTokenServiceType;
import org.apache.storm.nimbus.NimbusInfo;
-import org.apache.storm.utils.Utils;
import org.apache.storm.utils.Time;
+import org.apache.storm.utils.Utils;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.nio.ByteBuffer;
-import java.security.NoSuchAlgorithmException;
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicReference;
-
public class StormClusterStateImpl implements IStormClusterState {
private static Logger LOG = LoggerFactory.getLogger(StormClusterStateImpl.class);
@@ -48,7 +69,7 @@ public class StormClusterStateImpl implements IStormClusterState {
private ConcurrentHashMap<String, Runnable> assignmentInfoWithVersionCallback;
private ConcurrentHashMap<String, Runnable> assignmentVersionCallback;
private AtomicReference<Runnable> supervisorsCallback;
- // we want to reigister a topo directory getChildren callback for all workers of this dir
+ // we want to register a topo directory getChildren callback for all workers of this dir
private ConcurrentHashMap<String, Runnable> backPressureCallback;
private AtomicReference<Runnable> assignmentsCallback;
private ConcurrentHashMap<String, Runnable> stormBaseCallback;
@@ -56,15 +77,17 @@ public class StormClusterStateImpl implements IStormClusterState {
private ConcurrentHashMap<String, Runnable> credentialsCallback;
private ConcurrentHashMap<String, Runnable> logConfigCallback;
- private List<ACL> acls;
- private String stateId;
- private boolean solo;
+ private final List<ACL> defaultAcls;
+ private final String stateId;
+ private final boolean solo;
+ private final ClusterStateContext context;
- public StormClusterStateImpl(IStateStorage StateStorage, List<ACL> acls, ClusterStateContext context, boolean solo) throws Exception {
+ public StormClusterStateImpl(IStateStorage StateStorage, ClusterStateContext context, boolean solo) throws Exception {
this.stateStorage = StateStorage;
this.solo = solo;
- this.acls = acls;
+ this.defaultAcls = context.getDefaultZkAcls();
+ this.context = context;
assignmentInfoCallback = new ConcurrentHashMap<>();
assignmentInfoWithVersionCallback = new ConcurrentHashMap<>();
@@ -128,7 +151,7 @@ public class StormClusterStateImpl implements IStormClusterState {
ClusterUtils.LOGCONFIG_SUBTREE,
ClusterUtils.BACKPRESSURE_SUBTREE };
for (String path : pathlist) {
- this.stateStorage.mkdirs(path, acls);
+ this.stateStorage.mkdirs(path, defaultAcls);
}
}
@@ -221,13 +244,13 @@ public class StormClusterStateImpl implements IStormClusterState {
LOG.info("Connection state has changed to reconnected so setting nimbuses entry one more time");
// explicit delete for ephmeral node to ensure this session creates the entry.
stateStorage.delete_node(ClusterUtils.nimbusPath(nimbusId));
- stateStorage.set_ephemeral_node(ClusterUtils.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), acls);
+ stateStorage.set_ephemeral_node(ClusterUtils.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), defaultAcls);
}
}
});
- stateStorage.set_ephemeral_node(ClusterUtils.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), acls);
+ stateStorage.set_ephemeral_node(ClusterUtils.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), defaultAcls);
}
@Override
@@ -285,7 +308,7 @@ public class StormClusterStateImpl implements IStormClusterState {
String host = profileRequest.get_nodeInfo().get_node();
Long port = profileRequest.get_nodeInfo().get_port_iterator().next();
String path = ClusterUtils.profilerConfigPath(stormId, host, port, profileAction);
- stateStorage.set_data(path, Utils.serialize(profileRequest), acls);
+ stateStorage.set_data(path, Utils.serialize(profileRequest), defaultAcls);
}
@Override
@@ -343,7 +366,7 @@ public class StormClusterStateImpl implements IStormClusterState {
@Override
public void setupHeatbeats(String stormId) {
- stateStorage.mkdirs(ClusterUtils.workerbeatStormRoot(stormId), acls);
+ stateStorage.mkdirs(ClusterUtils.workerbeatStormRoot(stormId), defaultAcls);
}
@Override
@@ -391,7 +414,7 @@ public class StormClusterStateImpl implements IStormClusterState {
@Override
public void setTopologyLogConfig(String stormId, LogConfig logConfig) {
- stateStorage.set_data(ClusterUtils.logConfigPath(stormId), Utils.serialize(logConfig), acls);
+ stateStorage.set_data(ClusterUtils.logConfigPath(stormId), Utils.serialize(logConfig), defaultAcls);
}
@Override
@@ -407,7 +430,7 @@ public class StormClusterStateImpl implements IStormClusterState {
public void workerHeartbeat(String stormId, String node, Long port, ClusterWorkerHeartbeat info) {
if (info != null) {
String path = ClusterUtils.workerbeatPath(stormId, node, port);
- stateStorage.set_worker_hb(path, Utils.serialize(info), acls);
+ stateStorage.set_worker_hb(path, Utils.serialize(info), defaultAcls);
}
}
@@ -420,7 +443,7 @@ public class StormClusterStateImpl implements IStormClusterState {
@Override
public void supervisorHeartbeat(String supervisorId, SupervisorInfo info) {
String path = ClusterUtils.supervisorPath(supervisorId);
- stateStorage.set_ephemeral_node(path, Utils.serialize(info), acls);
+ stateStorage.set_ephemeral_node(path, Utils.serialize(info), defaultAcls);
}
/**
@@ -442,12 +465,12 @@ public class StormClusterStateImpl implements IStormClusterState {
stateStorage.delete_node(path);
} else {
byte[] data = ByteBuffer.allocate(Long.BYTES).putLong(timestamp).array();
- stateStorage.set_data(path, data, acls);
+ stateStorage.set_data(path, data, defaultAcls);
}
} else {
if (timestamp > 0) {
byte[] data = ByteBuffer.allocate(Long.BYTES).putLong(timestamp).array();
- stateStorage.set_ephemeral_node(path, data, acls);
+ stateStorage.set_ephemeral_node(path, data, defaultAcls);
}
}
}
@@ -485,7 +508,7 @@ public class StormClusterStateImpl implements IStormClusterState {
@Override
public void setupBackpressure(String stormId) {
- stateStorage.mkdirs(ClusterUtils.backpressureStormRoot(stormId), acls);
+ stateStorage.mkdirs(ClusterUtils.backpressureStormRoot(stormId), defaultAcls);
}
@Override
@@ -514,7 +537,7 @@ public class StormClusterStateImpl implements IStormClusterState {
@Override
public void activateStorm(String stormId, StormBase stormBase) {
String path = ClusterUtils.stormPath(stormId);
- stateStorage.set_data(path, Utils.serialize(stormBase), acls);
+ stateStorage.set_data(path, Utils.serialize(stormBase), defaultAcls);
}
@Override
@@ -596,7 +619,7 @@ public class StormClusterStateImpl implements IStormClusterState {
if (newElems.get_status() == null) {
newElems.set_status(stormBase.get_status());
}
- stateStorage.set_data(ClusterUtils.stormPath(stormId), Utils.serialize(newElems), acls);
+ stateStorage.set_data(ClusterUtils.stormPath(stormId), Utils.serialize(newElems), defaultAcls);
}
@Override
@@ -606,16 +629,16 @@ public class StormClusterStateImpl implements IStormClusterState {
@Override
public void setAssignment(String stormId, Assignment info) {
- stateStorage.set_data(ClusterUtils.assignmentPath(stormId), Utils.serialize(info), acls);
+ stateStorage.set_data(ClusterUtils.assignmentPath(stormId), Utils.serialize(info), defaultAcls);
}
@Override
public void setupBlobstore(String key, NimbusInfo nimbusInfo, Integer versionInfo) {
String path = ClusterUtils.blobstorePath(key) + ClusterUtils.ZK_SEPERATOR + nimbusInfo.toHostPortString() + "-" + versionInfo;
LOG.info("set-path: {}", path);
- stateStorage.mkdirs(ClusterUtils.blobstorePath(key), acls);
+ stateStorage.mkdirs(ClusterUtils.blobstorePath(key), defaultAcls);
stateStorage.delete_node_blobstore(ClusterUtils.blobstorePath(key), nimbusInfo.toHostPortString());
- stateStorage.set_ephemeral_node(path, null, acls);
+ stateStorage.set_ephemeral_node(path, null, defaultAcls);
}
@Override
@@ -663,9 +686,9 @@ public class StormClusterStateImpl implements IStormClusterState {
errorInfo.set_host(node);
errorInfo.set_port(port.intValue());
byte[] serData = Utils.serialize(errorInfo);
- stateStorage.mkdirs(path, acls);
- stateStorage.create_sequential(path + ClusterUtils.ZK_SEPERATOR + "e", serData, acls);
- stateStorage.set_data(lastErrorPath, serData, acls);
+ stateStorage.mkdirs(path, defaultAcls);
+ stateStorage.create_sequential(path + ClusterUtils.ZK_SEPERATOR + "e", serData, defaultAcls);
+ stateStorage.set_data(lastErrorPath, serData, defaultAcls);
List<String> childrens = stateStorage.get_children(path, false);
Collections.sort(childrens, new Comparator<String>() {
@@ -744,13 +767,106 @@ public class StormClusterStateImpl implements IStormClusterState {
@Override
public void disconnect() {
stateStorage.unregister(stateId);
- if (solo)
+ if (solo) {
stateStorage.close();
+ }
+ }
+
+ @Override
+ public PrivateWorkerKey getPrivateWorkerKey(WorkerTokenServiceType type, String topologyId, long keyVersion) {
+ String path = ClusterUtils.secretKeysPath(type, topologyId, keyVersion);
+ return ClusterUtils.maybeDeserialize(stateStorage.get_data(path, false), PrivateWorkerKey.class);
+ }
+
+ @Override
+ public void addPrivateWorkerKey(WorkerTokenServiceType type, String topologyId, long keyVersion, PrivateWorkerKey key) {
+ assert context.getDaemonType() == DaemonType.NIMBUS;
+ List<ACL> secretAcls = context.getZkSecretAcls(type);
+ String path = ClusterUtils.secretKeysPath(type, topologyId, keyVersion);
+ LOG.debug("Storing private key for {} connecting to a {} at {} with ACL {}\n\n", topologyId, type, path, secretAcls);
+ stateStorage.set_data(path, Utils.serialize(key), secretAcls);
+ }
+
+ @Override
+ public long getNextPrivateWorkerKeyVersion(WorkerTokenServiceType type, String topologyId) {
+ String path = ClusterUtils.secretKeysPath(type, topologyId);
+ try {
+ List<String> versions = stateStorage.get_children(path, false);
+ return versions.stream().mapToLong(Long::valueOf).max().orElse(0);
+ } catch (RuntimeException e) {
+ if (Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
+ //If the node does not exist, then the version must be 0
+ return 0;
+ }
+ throw e;
+ }
}
+ @Override
+ public void removeExpiredPrivateWorkerKeys(String topologyId) {
+ for (WorkerTokenServiceType type : WorkerTokenServiceType.values()) {
+ String basePath = ClusterUtils.secretKeysPath(type, topologyId);
+ try {
+ for (String version : stateStorage.get_children(basePath, false)) {
+ String fullPath = basePath + ClusterUtils.ZK_SEPERATOR + version;
+ try {
+ PrivateWorkerKey key = ClusterUtils.maybeDeserialize(stateStorage.get_data(fullPath, false), PrivateWorkerKey.class);
+ if (Time.currentTimeMillis() > key.get_expirationTimeMillis()) {
+ stateStorage.delete_node(fullPath);
+ }
+ } catch (RuntimeException e) {
+ //This should never happen because only the primary nimbus is active, but just in case
+ // declare the race safe, even if we lose it.
+ if (!Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
+ throw e;
+ }
+ }
+ }
+ } catch (RuntimeException e) {
+ //No node for basePath is OK, nothing to remove
+ if (!Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
+ throw e;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void removeAllPrivateWorkerKeys(String topologyId) {
+ for (WorkerTokenServiceType type : WorkerTokenServiceType.values()) {
+ String path = ClusterUtils.secretKeysPath(type, topologyId);
+ try {
+ stateStorage.delete_node(path);
+ } catch (RuntimeException e) {
+ //This should never happen because only the primary nimbus is active, but just in case
+ // declare the race safe, even if we lose it.
+ if (!Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
+ throw e;
+ }
+ }
+ }
+ }
+
+ @Override
+ public Set<String> idsOfTopologiesWithPrivateWorkerKeys() {
+ HashSet<String> ret = new HashSet<>();
+ for (WorkerTokenServiceType type : WorkerTokenServiceType.values()) {
+ String path = ClusterUtils.secretKeysPath(type);
+ try {
+ ret.addAll(stateStorage.get_children(path, false));
+ } catch (RuntimeException e) {
+ //If the node does not exist it is fine/expected...
+ if (!Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
+ throw e;
+ }
+ }
+ }
+ return ret;
+ }
+
private List<String> tokenizePath(String path) {
String[] toks = path.split("/");
- java.util.ArrayList<String> rtn = new ArrayList<String>();
+ java.util.ArrayList<String> rtn = new ArrayList<>();
for (String str : toks) {
if (!str.isEmpty()) {
rtn.add(str);
@@ -758,5 +874,4 @@ public class StormClusterStateImpl implements IStormClusterState {
}
return rtn;
}
-
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorage.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorage.java b/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorage.java
index a5aa8ef..c68011e 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorage.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorage.java
@@ -73,7 +73,7 @@ public class ZKStateStorage implements IStateStorage {
}
}
- public ZKStateStorage(Map<String, Object> conf, Map authConf, List<ACL> acls, ClusterStateContext context) throws Exception {
+ public ZKStateStorage(Map<String, Object> conf, Map authConf, ClusterStateContext context) throws Exception {
this.conf = conf;
this.authConf = authConf;
if (context.getDaemonType().equals(DaemonType.NIMBUS))
@@ -82,7 +82,7 @@ public class ZKStateStorage implements IStateStorage {
// just mkdir STORM_ZOOKEEPER_ROOT dir
CuratorFramework zkTemp = mkZk();
String rootPath = String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT));
- ClientZookeeper.mkdirs(zkTemp, rootPath, acls);
+ ClientZookeeper.mkdirs(zkTemp, rootPath, context.getDefaultZkAcls());
zkTemp.close();
active = new AtomicBoolean(true);
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorageFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorageFactory.java b/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorageFactory.java
index d38e211..987c2fd 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorageFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/ZKStateStorageFactory.java
@@ -26,9 +26,9 @@ import java.util.Map;
public class ZKStateStorageFactory implements StateStorageFactory {
@Override
- public IStateStorage mkStore(Map<String, Object> config, Map auth_conf, List<ACL> acls, ClusterStateContext context) {
+ public IStateStorage mkStore(Map<String, Object> config, Map auth_conf, ClusterStateContext context) {
try {
- return new ZKStateStorage(config, auth_conf, acls, context);
+ return new ZKStateStorage(config, auth_conf, context);
} catch (Exception e) {
throw Utils.wrapInRuntime(e);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java b/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java
index 4209422..57ceb39 100644
--- a/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java
+++ b/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java
@@ -72,7 +72,6 @@ import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
import org.apache.storm.utils.WorkerBackpressureCallback;
import org.apache.storm.utils.WorkerBackpressureThread;
-import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -141,11 +140,10 @@ public class Worker implements Shutdownable, DaemonCommon {
}
final Map<String, Object> topologyConf =
ConfigUtils.overrideLoginConfigWithSystemProperty(ConfigUtils.readSupervisorStormConf(conf, topologyId));
- List<ACL> acls = Utils.getWorkerACL(topologyConf);
- IStateStorage stateStorage =
- ClusterUtils.mkStateStorage(conf, topologyConf, acls, new ClusterStateContext(DaemonType.WORKER));
- IStormClusterState stormClusterState =
- ClusterUtils.mkStormClusterState(stateStorage, acls, new ClusterStateContext());
+ ClusterStateContext csContext = new ClusterStateContext(DaemonType.WORKER, topologyConf);
+ IStateStorage stateStorage = ClusterUtils.mkStateStorage(conf, topologyConf, csContext);
+ IStormClusterState stormClusterState = ClusterUtils.mkStormClusterState(stateStorage, csContext);
+
Credentials initialCredentials = stormClusterState.credentials(topologyId, null);
Map<String, String> initCreds = new HashMap<>();
if (initialCredentials != null) {
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/executor/Executor.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/executor/Executor.java b/storm-client/src/jvm/org/apache/storm/executor/Executor.java
index a90e16c..6b87aed 100644
--- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java
+++ b/storm-client/src/jvm/org/apache/storm/executor/Executor.java
@@ -146,8 +146,8 @@ public abstract class Executor implements Callable, EventHandler<Object> {
this.suicideFn = workerData.getSuicideCallback();
try {
- this.stormClusterState = ClusterUtils.mkStormClusterState(workerData.getStateStorage(), Utils.getWorkerACL(topoConf),
- new ClusterStateContext(DaemonType.WORKER));
+ this.stormClusterState = ClusterUtils.mkStormClusterState(workerData.getStateStorage(),
+ new ClusterStateContext(DaemonType.WORKER, topoConf));
} catch (Exception e) {
throw Utils.wrapInRuntime(e);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9566a882/storm-client/src/jvm/org/apache/storm/generated/PrivateWorkerKey.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/generated/PrivateWorkerKey.java b/storm-client/src/jvm/org/apache/storm/generated/PrivateWorkerKey.java
new file mode 100644
index 0000000..9884e09
--- /dev/null
+++ b/storm-client/src/jvm/org/apache/storm/generated/PrivateWorkerKey.java
@@ -0,0 +1,615 @@
+/**
+ * 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.
+ */
+/**
+ * Autogenerated by Thrift Compiler (0.9.3)
+ *
+ * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
+ * @generated
+ */
+package org.apache.storm.generated;
+
+import org.apache.thrift.scheme.IScheme;
+import org.apache.thrift.scheme.SchemeFactory;
+import org.apache.thrift.scheme.StandardScheme;
+
+import org.apache.thrift.scheme.TupleScheme;
+import org.apache.thrift.protocol.TTupleProtocol;
+import org.apache.thrift.protocol.TProtocolException;
+import org.apache.thrift.EncodingUtils;
+import org.apache.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.apache.thrift.server.AbstractNonblockingServer.*;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.EnumMap;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.EnumSet;
+import java.util.Collections;
+import java.util.BitSet;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import javax.annotation.Generated;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.3)")
+public class PrivateWorkerKey implements org.apache.thrift.TBase<PrivateWorkerKey, PrivateWorkerKey._Fields>, java.io.Serializable, Cloneable, Comparable<PrivateWorkerKey> {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("PrivateWorkerKey");
+
+ private static final org.apache.thrift.protocol.TField KEY_FIELD_DESC = new org.apache.thrift.protocol.TField("key", org.apache.thrift.protocol.TType.STRING, (short)1);
+ private static final org.apache.thrift.protocol.TField USER_NAME_FIELD_DESC = new org.apache.thrift.protocol.TField("userName", org.apache.thrift.protocol.TType.STRING, (short)2);
+ private static final org.apache.thrift.protocol.TField EXPIRATION_TIME_MILLIS_FIELD_DESC = new org.apache.thrift.protocol.TField("expirationTimeMillis", org.apache.thrift.protocol.TType.I64, (short)3);
+
+ private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
+ static {
+ schemes.put(StandardScheme.class, new PrivateWorkerKeyStandardSchemeFactory());
+ schemes.put(TupleScheme.class, new PrivateWorkerKeyTupleSchemeFactory());
+ }
+
+ private ByteBuffer key; // required
+ private String userName; // required
+ private long expirationTimeMillis; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+ KEY((short)1, "key"),
+ USER_NAME((short)2, "userName"),
+ EXPIRATION_TIME_MILLIS((short)3, "expirationTimeMillis");
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // KEY
+ return KEY;
+ case 2: // USER_NAME
+ return USER_NAME;
+ case 3: // EXPIRATION_TIME_MILLIS
+ return EXPIRATION_TIME_MILLIS;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+ private static final int __EXPIRATIONTIMEMILLIS_ISSET_ID = 0;
+ private byte __isset_bitfield = 0;
+ public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.KEY, new org.apache.thrift.meta_data.FieldMetaData("key", org.apache.thrift.TFieldRequirementType.REQUIRED,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING , true)));
+ tmpMap.put(_Fields.USER_NAME, new org.apache.thrift.meta_data.FieldMetaData("userName", org.apache.thrift.TFieldRequirementType.REQUIRED,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+ tmpMap.put(_Fields.EXPIRATION_TIME_MILLIS, new org.apache.thrift.meta_data.FieldMetaData("expirationTimeMillis", org.apache.thrift.TFieldRequirementType.REQUIRED,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I64)));
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(PrivateWorkerKey.class, metaDataMap);
+ }
+
+ public PrivateWorkerKey() {
+ }
+
+ public PrivateWorkerKey(
+ ByteBuffer key,
+ String userName,
+ long expirationTimeMillis)
+ {
+ this();
+ this.key = org.apache.thrift.TBaseHelper.copyBinary(key);
+ this.userName = userName;
+ this.expirationTimeMillis = expirationTimeMillis;
+ set_expirationTimeMillis_isSet(true);
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public PrivateWorkerKey(PrivateWorkerKey other) {
+ __isset_bitfield = other.__isset_bitfield;
+ if (other.is_set_key()) {
+ this.key = org.apache.thrift.TBaseHelper.copyBinary(other.key);
+ }
+ if (other.is_set_userName()) {
+ this.userName = other.userName;
+ }
+ this.expirationTimeMillis = other.expirationTimeMillis;
+ }
+
+ public PrivateWorkerKey deepCopy() {
+ return new PrivateWorkerKey(this);
+ }
+
+ @Override
+ public void clear() {
+ this.key = null;
+ this.userName = null;
+ set_expirationTimeMillis_isSet(false);
+ this.expirationTimeMillis = 0;
+ }
+
+ public byte[] get_key() {
+ set_key(org.apache.thrift.TBaseHelper.rightSize(key));
+ return key == null ? null : key.array();
+ }
+
+ public ByteBuffer buffer_for_key() {
+ return org.apache.thrift.TBaseHelper.copyBinary(key);
+ }
+
+ public void set_key(byte[] key) {
+ this.key = key == null ? (ByteBuffer)null : ByteBuffer.wrap(Arrays.copyOf(key, key.length));
+ }
+
+ public void set_key(ByteBuffer key) {
+ this.key = org.apache.thrift.TBaseHelper.copyBinary(key);
+ }
+
+ public void unset_key() {
+ this.key = null;
+ }
+
+ /** Returns true if field key is set (has been assigned a value) and false otherwise */
+ public boolean is_set_key() {
+ return this.key != null;
+ }
+
+ public void set_key_isSet(boolean value) {
+ if (!value) {
+ this.key = null;
+ }
+ }
+
+ public String get_userName() {
+ return this.userName;
+ }
+
+ public void set_userName(String userName) {
+ this.userName = userName;
+ }
+
+ public void unset_userName() {
+ this.userName = null;
+ }
+
+ /** Returns true if field userName is set (has been assigned a value) and false otherwise */
+ public boolean is_set_userName() {
+ return this.userName != null;
+ }
+
+ public void set_userName_isSet(boolean value) {
+ if (!value) {
+ this.userName = null;
+ }
+ }
+
+ public long get_expirationTimeMillis() {
+ return this.expirationTimeMillis;
+ }
+
+ public void set_expirationTimeMillis(long expirationTimeMillis) {
+ this.expirationTimeMillis = expirationTimeMillis;
+ set_expirationTimeMillis_isSet(true);
+ }
+
+ public void unset_expirationTimeMillis() {
+ __isset_bitfield = EncodingUtils.clearBit(__isset_bitfield, __EXPIRATIONTIMEMILLIS_ISSET_ID);
+ }
+
+ /** Returns true if field expirationTimeMillis is set (has been assigned a value) and false otherwise */
+ public boolean is_set_expirationTimeMillis() {
+ return EncodingUtils.testBit(__isset_bitfield, __EXPIRATIONTIMEMILLIS_ISSET_ID);
+ }
+
+ public void set_expirationTimeMillis_isSet(boolean value) {
+ __isset_bitfield = EncodingUtils.setBit(__isset_bitfield, __EXPIRATIONTIMEMILLIS_ISSET_ID, value);
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case KEY:
+ if (value == null) {
+ unset_key();
+ } else {
+ set_key((ByteBuffer)value);
+ }
+ break;
+
+ case USER_NAME:
+ if (value == null) {
+ unset_userName();
+ } else {
+ set_userName((String)value);
+ }
+ break;
+
+ case EXPIRATION_TIME_MILLIS:
+ if (value == null) {
+ unset_expirationTimeMillis();
+ } else {
+ set_expirationTimeMillis((Long)value);
+ }
+ break;
+
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case KEY:
+ return get_key();
+
+ case USER_NAME:
+ return get_userName();
+
+ case EXPIRATION_TIME_MILLIS:
+ return get_expirationTimeMillis();
+
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ case KEY:
+ return is_set_key();
+ case USER_NAME:
+ return is_set_userName();
+ case EXPIRATION_TIME_MILLIS:
+ return is_set_expirationTimeMillis();
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof PrivateWorkerKey)
+ return this.equals((PrivateWorkerKey)that);
+ return false;
+ }
+
+ public boolean equals(PrivateWorkerKey that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_key = true && this.is_set_key();
+ boolean that_present_key = true && that.is_set_key();
+ if (this_present_key || that_present_key) {
+ if (!(this_present_key && that_present_key))
+ return false;
+ if (!this.key.equals(that.key))
+ return false;
+ }
+
+ boolean this_present_userName = true && this.is_set_userName();
+ boolean that_present_userName = true && that.is_set_userName();
+ if (this_present_userName || that_present_userName) {
+ if (!(this_present_userName && that_present_userName))
+ return false;
+ if (!this.userName.equals(that.userName))
+ return false;
+ }
+
+ boolean this_present_expirationTimeMillis = true;
+ boolean that_present_expirationTimeMillis = true;
+ if (this_present_expirationTimeMillis || that_present_expirationTimeMillis) {
+ if (!(this_present_expirationTimeMillis && that_present_expirationTimeMillis))
+ return false;
+ if (this.expirationTimeMillis != that.expirationTimeMillis)
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ List<Object> list = new ArrayList<Object>();
+
+ boolean present_key = true && (is_set_key());
+ list.add(present_key);
+ if (present_key)
+ list.add(key);
+
+ boolean present_userName = true && (is_set_userName());
+ list.add(present_userName);
+ if (present_userName)
+ list.add(userName);
+
+ boolean present_expirationTimeMillis = true;
+ list.add(present_expirationTimeMillis);
+ if (present_expirationTimeMillis)
+ list.add(expirationTimeMillis);
+
+ return list.hashCode();
+ }
+
+ @Override
+ public int compareTo(PrivateWorkerKey other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+
+ lastComparison = Boolean.valueOf(is_set_key()).compareTo(other.is_set_key());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_key()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.key, other.key);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = Boolean.valueOf(is_set_userName()).compareTo(other.is_set_userName());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_userName()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.userName, other.userName);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = Boolean.valueOf(is_set_expirationTimeMillis()).compareTo(other.is_set_expirationTimeMillis());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_expirationTimeMillis()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.expirationTimeMillis, other.expirationTimeMillis);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("PrivateWorkerKey(");
+ boolean first = true;
+
+ sb.append("key:");
+ if (this.key == null) {
+ sb.append("null");
+ } else {
+ org.apache.thrift.TBaseHelper.toString(this.key, sb);
+ }
+ first = false;
+ if (!first) sb.append(", ");
+ sb.append("userName:");
+ if (this.userName == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.userName);
+ }
+ first = false;
+ if (!first) sb.append(", ");
+ sb.append("expirationTimeMillis:");
+ sb.append(this.expirationTimeMillis);
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ if (!is_set_key()) {
+ throw new org.apache.thrift.protocol.TProtocolException("Required field 'key' is unset! Struct:" + toString());
+ }
+
+ if (!is_set_userName()) {
+ throw new org.apache.thrift.protocol.TProtocolException("Required field 'userName' is unset! Struct:" + toString());
+ }
+
+ if (!is_set_expirationTimeMillis()) {
+ throw new org.apache.thrift.protocol.TProtocolException("Required field 'expirationTimeMillis' is unset! Struct:" + toString());
+ }
+
+ // check for sub-struct validity
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
+ __isset_bitfield = 0;
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private static class PrivateWorkerKeyStandardSchemeFactory implements SchemeFactory {
+ public PrivateWorkerKeyStandardScheme getScheme() {
+ return new PrivateWorkerKeyStandardScheme();
+ }
+ }
+
+ private static class PrivateWorkerKeyStandardScheme extends StandardScheme<PrivateWorkerKey> {
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot, PrivateWorkerKey struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField schemeField;
+ iprot.readStructBegin();
+ while (true)
+ {
+ schemeField = iprot.readFieldBegin();
+ if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (schemeField.id) {
+ case 1: // KEY
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.key = iprot.readBinary();
+ struct.set_key_isSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ case 2: // USER_NAME
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.userName = iprot.readString();
+ struct.set_userName_isSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ case 3: // EXPIRATION_TIME_MILLIS
+ if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
+ struct.expirationTimeMillis = iprot.readI64();
+ struct.set_expirationTimeMillis_isSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+ struct.validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot, PrivateWorkerKey struct) throws org.apache.thrift.TException {
+ struct.validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ if (struct.key != null) {
+ oprot.writeFieldBegin(KEY_FIELD_DESC);
+ oprot.writeBinary(struct.key);
+ oprot.writeFieldEnd();
+ }
+ if (struct.userName != null) {
+ oprot.writeFieldBegin(USER_NAME_FIELD_DESC);
+ oprot.writeString(struct.userName);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldBegin(EXPIRATION_TIME_MILLIS_FIELD_DESC);
+ oprot.writeI64(struct.expirationTimeMillis);
+ oprot.writeFieldEnd();
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ }
+
+ private static class PrivateWorkerKeyTupleSchemeFactory implements SchemeFactory {
+ public PrivateWorkerKeyTupleScheme getScheme() {
+ return new PrivateWorkerKeyTupleScheme();
+ }
+ }
+
+ private static class PrivateWorkerKeyTupleScheme extends TupleScheme<PrivateWorkerKey> {
+
+ @Override
+ public void write(org.apache.thrift.protocol.TProtocol prot, PrivateWorkerKey struct) throws org.apache.thrift.TException {
+ TTupleProtocol oprot = (TTupleProtocol) prot;
+ oprot.writeBinary(struct.key);
+ oprot.writeString(struct.userName);
+ oprot.writeI64(struct.expirationTimeMillis);
+ }
+
+ @Override
+ public void read(org.apache.thrift.protocol.TProtocol prot, PrivateWorkerKey struct) throws org.apache.thrift.TException {
+ TTupleProtocol iprot = (TTupleProtocol) prot;
+ struct.key = iprot.readBinary();
+ struct.set_key_isSet(true);
+ struct.userName = iprot.readString();
+ struct.set_userName_isSet(true);
+ struct.expirationTimeMillis = iprot.readI64();
+ struct.set_expirationTimeMillis_isSet(true);
+ }
+ }
+
+}
+