YARN-2211. Persist AMRMToken master key in RMStateStore for RM recovery. Contributed by Xuan Gong

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1613515 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jian He 2014-07-25 20:42:37 +00:00
parent 10286e98ad
commit d4fec34933
23 changed files with 535 additions and 34 deletions

View File

@ -68,6 +68,9 @@ Release 2.6.0 - UNRELEASED
YARN-2214. FairScheduler: preemptContainerPreCheck() in FSParentQueue delays YARN-2214. FairScheduler: preemptContainerPreCheck() in FSParentQueue delays
convergence towards fairness. (Ashwin Shankar via kasha) convergence towards fairness. (Ashwin Shankar via kasha)
YARN-2211. Persist AMRMToken master key in RMStateStore for RM recovery.
(Xuan Gong via jianhe)
OPTIMIZATIONS OPTIMIZATIONS
BUG FIXES BUG FIXES

View File

@ -267,6 +267,7 @@ public abstract class ProtocolHATestBase extends ClientBaseWithFixes{
protected void startHACluster(int numOfNMs, boolean overrideClientRMService, protected void startHACluster(int numOfNMs, boolean overrideClientRMService,
boolean overrideRTS, boolean overrideApplicationMasterService) boolean overrideRTS, boolean overrideApplicationMasterService)
throws Exception { throws Exception {
conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
conf.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false); conf.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false);
cluster = cluster =
new MiniYARNClusterForHATesting(TestRMFailover.class.getName(), 2, new MiniYARNClusterForHATesting(TestRMFailover.class.getName(), 2,

View File

@ -54,11 +54,9 @@ public class TestApplicationMasterServiceOnHA extends ProtocolHATestBase{
amClient = ClientRMProxy amClient = ClientRMProxy
.createRMProxy(this.conf, ApplicationMasterProtocol.class); .createRMProxy(this.conf, ApplicationMasterProtocol.class);
AMRMTokenIdentifier id =
new AMRMTokenIdentifier(attemptId);
Token<AMRMTokenIdentifier> appToken = Token<AMRMTokenIdentifier> appToken =
new Token<AMRMTokenIdentifier>(id, this.cluster.getResourceManager() this.cluster.getResourceManager().getRMContext()
.getRMContext().getAMRMTokenSecretManager()); .getAMRMTokenSecretManager().createAndGetAMRMToken(attemptId);
appToken.setService(new Text("appToken service")); appToken.setService(new Text("appToken service"));
UserGroupInformation.setLoginUser(UserGroupInformation UserGroupInformation.setLoginUser(UserGroupInformation
.createRemoteUser(UserGroupInformation.getCurrentUser() .createRemoteUser(UserGroupInformation.getCurrentUser()

View File

@ -244,6 +244,37 @@
</execution> </execution>
</executions> </executions>
</plugin> </plugin>
<plugin>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-maven-plugins</artifactId>
<executions>
<execution>
<id>compile-protoc</id>
<phase>generate-sources</phase>
<goals>
<goal>protoc</goal>
</goals>
<configuration>
<protocVersion>${protobuf.version}</protocVersion>
<protocCommand>${protoc.path}</protocCommand>
<imports>
<param>${basedir}/../../../../hadoop-common-project/hadoop-common/src/main/proto</param>
<param>${basedir}/../../hadoop-yarn-api/src/main/proto</param>
<param>${basedir}/../hadoop-yarn-server-common/src/main/proto</param>
<param>${basedir}/src/main/proto</param>
</imports>
<source>
<directory>${basedir}/src/main/proto</directory>
<includes>
<include>yarn_server_resourcemanager_recovery.proto</include>
</includes>
</source>
<output>${project.build.directory}/generated-sources/java</output>
</configuration>
</execution>
</executions>
</plugin>
</plugins> </plugins>
</build> </build>

View File

@ -60,7 +60,7 @@ public class RMSecretManagerService extends AbstractService {
clientToAMSecretManager = createClientToAMTokenSecretManager(); clientToAMSecretManager = createClientToAMTokenSecretManager();
rmContext.setClientToAMTokenSecretManager(clientToAMSecretManager); rmContext.setClientToAMTokenSecretManager(clientToAMSecretManager);
amRmTokenSecretManager = createAMRMTokenSecretManager(conf); amRmTokenSecretManager = createAMRMTokenSecretManager(conf, this.rmContext);
rmContext.setAMRMTokenSecretManager(amRmTokenSecretManager); rmContext.setAMRMTokenSecretManager(amRmTokenSecretManager);
rmDTSecretManager = rmDTSecretManager =
@ -115,8 +115,8 @@ public class RMSecretManagerService extends AbstractService {
} }
protected AMRMTokenSecretManager createAMRMTokenSecretManager( protected AMRMTokenSecretManager createAMRMTokenSecretManager(
Configuration conf) { Configuration conf, RMContext rmContext) {
return new AMRMTokenSecretManager(conf); return new AMRMTokenSecretManager(conf, rmContext);
} }
protected ClientToAMTokenSecretManagerInRM createClientToAMTokenSecretManager() { protected ClientToAMTokenSecretManagerInRM createClientToAMTokenSecretManager() {

View File

@ -1026,6 +1026,9 @@ public class ResourceManager extends CompositeService implements Recoverable {
// recover RMdelegationTokenSecretManager // recover RMdelegationTokenSecretManager
rmContext.getRMDelegationTokenSecretManager().recover(state); rmContext.getRMDelegationTokenSecretManager().recover(state);
// recover AMRMTokenSecretManager
rmContext.getAMRMTokenSecretManager().recover(state);
// recover applications // recover applications
rmAppManager.recover(state); rmAppManager.recover(state);
} }

View File

@ -22,6 +22,7 @@ import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
import java.io.DataInputStream; import java.io.DataInputStream;
import java.io.DataOutputStream; import java.io.DataOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
@ -43,16 +44,18 @@ import org.apache.hadoop.security.token.delegation.DelegationKey;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerRecoveryProtos.AMRMTokenSecretManagerStateProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.EpochProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.EpochProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationAttemptStateDataProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationAttemptStateDataProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationStateDataProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationStateDataProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RMStateVersionProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RMStateVersionProto;
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.Epoch; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.Epoch;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.AMRMTokenSecretManagerStatePBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationAttemptStateDataPBImpl; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationAttemptStateDataPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationStateDataPBImpl; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationStateDataPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.EpochPBImpl; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.EpochPBImpl;
@ -76,6 +79,8 @@ public class FileSystemRMStateStore extends RMStateStore {
protected static final String ROOT_DIR_NAME = "FSRMStateRoot"; protected static final String ROOT_DIR_NAME = "FSRMStateRoot";
protected static final RMStateVersion CURRENT_VERSION_INFO = RMStateVersion protected static final RMStateVersion CURRENT_VERSION_INFO = RMStateVersion
.newInstance(1, 1); .newInstance(1, 1);
protected static final String AMRMTOKEN_SECRET_MANAGER_NODE =
"AMRMTokenSecretManagerNode";
protected FileSystem fs; protected FileSystem fs;
@ -89,6 +94,7 @@ public class FileSystemRMStateStore extends RMStateStore {
@VisibleForTesting @VisibleForTesting
Path fsWorkingPath; Path fsWorkingPath;
Path amrmTokenSecretManagerRoot;
@Override @Override
public synchronized void initInternal(Configuration conf) public synchronized void initInternal(Configuration conf)
throws Exception{ throws Exception{
@ -96,6 +102,8 @@ public class FileSystemRMStateStore extends RMStateStore {
rootDirPath = new Path(fsWorkingPath, ROOT_DIR_NAME); rootDirPath = new Path(fsWorkingPath, ROOT_DIR_NAME);
rmDTSecretManagerRoot = new Path(rootDirPath, RM_DT_SECRET_MANAGER_ROOT); rmDTSecretManagerRoot = new Path(rootDirPath, RM_DT_SECRET_MANAGER_ROOT);
rmAppRoot = new Path(rootDirPath, RM_APP_ROOT); rmAppRoot = new Path(rootDirPath, RM_APP_ROOT);
amrmTokenSecretManagerRoot =
new Path(rootDirPath, AMRMTOKEN_SECRET_MANAGER_ROOT);
} }
@Override @Override
@ -113,6 +121,7 @@ public class FileSystemRMStateStore extends RMStateStore {
fs = fsWorkingPath.getFileSystem(conf); fs = fsWorkingPath.getFileSystem(conf);
fs.mkdirs(rmDTSecretManagerRoot); fs.mkdirs(rmDTSecretManagerRoot);
fs.mkdirs(rmAppRoot); fs.mkdirs(rmAppRoot);
fs.mkdirs(amrmTokenSecretManagerRoot);
} }
@Override @Override
@ -180,9 +189,32 @@ public class FileSystemRMStateStore extends RMStateStore {
loadRMDTSecretManagerState(rmState); loadRMDTSecretManagerState(rmState);
// recover RM applications // recover RM applications
loadRMAppState(rmState); loadRMAppState(rmState);
// recover AMRMTokenSecretManager
loadAMRMTokenSecretManagerState(rmState);
return rmState; return rmState;
} }
private void loadAMRMTokenSecretManagerState(RMState rmState)
throws Exception {
checkAndResumeUpdateOperation(amrmTokenSecretManagerRoot);
Path amrmTokenSecretManagerStateDataDir =
new Path(amrmTokenSecretManagerRoot, AMRMTOKEN_SECRET_MANAGER_NODE);
FileStatus status;
try {
status = fs.getFileStatus(amrmTokenSecretManagerStateDataDir);
assert status.isFile();
} catch (FileNotFoundException ex) {
return;
}
byte[] data = readFile(amrmTokenSecretManagerStateDataDir, status.getLen());
AMRMTokenSecretManagerStatePBImpl stateData =
new AMRMTokenSecretManagerStatePBImpl(
AMRMTokenSecretManagerStateProto.parseFrom(data));
rmState.amrmTokenSecretManagerState =
AMRMTokenSecretManagerState.newInstance(
stateData.getCurrentMasterKey(), stateData.getNextMasterKey());
}
private void loadRMAppState(RMState rmState) throws Exception { private void loadRMAppState(RMState rmState) throws Exception {
try { try {
List<ApplicationAttemptState> attempts = List<ApplicationAttemptState> attempts =
@ -597,4 +629,25 @@ public class FileSystemRMStateStore extends RMStateStore {
return new Path(root, nodeName); return new Path(root, nodeName);
} }
@Override
public synchronized void storeOrUpdateAMRMTokenSecretManagerState(
AMRMTokenSecretManagerState amrmTokenSecretManagerState,
boolean isUpdate){
Path nodeCreatePath =
getNodePath(amrmTokenSecretManagerRoot, AMRMTOKEN_SECRET_MANAGER_NODE);
AMRMTokenSecretManagerState data =
AMRMTokenSecretManagerState.newInstance(amrmTokenSecretManagerState);
byte[] stateData = data.getProto().toByteArray();
try {
if (isUpdate) {
updateFile(nodeCreatePath, stateData);
} else {
writeFile(nodeCreatePath, stateData);
}
} catch (Exception ex) {
LOG.info("Error storing info for AMRMTokenSecretManager", ex);
notifyStoreOperationFailed(ex);
}
}
} }

View File

@ -32,6 +32,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion;
@ -72,6 +73,10 @@ public class MemoryRMStateStore extends RMStateStore {
state.rmSecretManagerState.getTokenState()); state.rmSecretManagerState.getTokenState());
returnState.rmSecretManagerState.dtSequenceNumber = returnState.rmSecretManagerState.dtSequenceNumber =
state.rmSecretManagerState.dtSequenceNumber; state.rmSecretManagerState.dtSequenceNumber;
returnState.amrmTokenSecretManagerState =
state.amrmTokenSecretManagerState == null ? null
: AMRMTokenSecretManagerState
.newInstance(state.amrmTokenSecretManagerState);
return returnState; return returnState;
} }
@ -267,6 +272,16 @@ public class MemoryRMStateStore extends RMStateStore {
return null; return null;
} }
@Override
public void storeOrUpdateAMRMTokenSecretManagerState(
AMRMTokenSecretManagerState amrmTokenSecretManagerState,
boolean isUpdate) {
if (amrmTokenSecretManagerState != null) {
state.amrmTokenSecretManagerState = AMRMTokenSecretManagerState
.newInstance(amrmTokenSecretManagerState);
}
}
@Override @Override
public void deleteStore() throws Exception { public void deleteStore() throws Exception {
} }

View File

@ -25,6 +25,7 @@ import org.apache.hadoop.security.token.delegation.DelegationKey;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion;
@ -138,6 +139,12 @@ public class NullRMStateStore extends RMStateStore {
return null; return null;
} }
@Override
public void storeOrUpdateAMRMTokenSecretManagerState(
AMRMTokenSecretManagerState state, boolean isUpdate) {
//DO Nothing
}
@Override @Override
public void deleteStore() throws Exception { public void deleteStore() throws Exception {
// Do nothing // Do nothing

View File

@ -45,16 +45,14 @@ import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationSubmissionContextPB
import org.apache.hadoop.yarn.event.AsyncDispatcher; import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.Dispatcher; import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler; import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier; import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.RMFatalEvent; import org.apache.hadoop.yarn.server.resourcemanager.RMFatalEvent;
import org.apache.hadoop.yarn.server.resourcemanager.RMFatalEventType; import org.apache.hadoop.yarn.server.resourcemanager.RMFatalEventType;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationAttemptStateDataPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationStateDataPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppNewSavedEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppNewSavedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
@ -85,6 +83,8 @@ public abstract class RMStateStore extends AbstractService {
protected static final String DELEGATION_TOKEN_PREFIX = "RMDelegationToken_"; protected static final String DELEGATION_TOKEN_PREFIX = "RMDelegationToken_";
protected static final String DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX = protected static final String DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX =
"RMDTSequenceNumber_"; "RMDTSequenceNumber_";
protected static final String AMRMTOKEN_SECRET_MANAGER_ROOT =
"AMRMTokenSecretManagerRoot";
protected static final String VERSION_NODE = "RMVersionNode"; protected static final String VERSION_NODE = "RMVersionNode";
protected static final String EPOCH_NODE = "EpochNode"; protected static final String EPOCH_NODE = "EpochNode";
@ -412,6 +412,8 @@ public abstract class RMStateStore extends AbstractService {
RMDTSecretManagerState rmSecretManagerState = new RMDTSecretManagerState(); RMDTSecretManagerState rmSecretManagerState = new RMDTSecretManagerState();
AMRMTokenSecretManagerState amrmTokenSecretManagerState = null;
public Map<ApplicationId, ApplicationState> getApplicationState() { public Map<ApplicationId, ApplicationState> getApplicationState() {
return appState; return appState;
} }
@ -419,6 +421,10 @@ public abstract class RMStateStore extends AbstractService {
public RMDTSecretManagerState getRMDTSecretManagerState() { public RMDTSecretManagerState getRMDTSecretManagerState() {
return rmSecretManagerState; return rmSecretManagerState;
} }
public AMRMTokenSecretManagerState getAMRMTokenSecretManagerState() {
return amrmTokenSecretManagerState;
}
} }
private Dispatcher rmDispatcher; private Dispatcher rmDispatcher;
@ -713,6 +719,14 @@ public abstract class RMStateStore extends AbstractService {
protected abstract void removeRMDTMasterKeyState(DelegationKey delegationKey) protected abstract void removeRMDTMasterKeyState(DelegationKey delegationKey)
throws Exception; throws Exception;
/**
* Blocking API Derived classes must implement this method to store or update
* the state of AMRMToken Master Key
*/
public abstract void storeOrUpdateAMRMTokenSecretManagerState(
AMRMTokenSecretManagerState amrmTokenSecretManagerState,
boolean isUpdate);
/** /**
* Non-blocking API * Non-blocking API
* ResourceManager services call this to remove an application from the state * ResourceManager services call this to remove an application from the state

View File

@ -44,18 +44,19 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.HAUtil; import org.apache.hadoop.yarn.conf.HAUtil;
import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerRecoveryProtos.AMRMTokenSecretManagerStateProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationAttemptStateDataProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationAttemptStateDataProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationStateDataProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.ApplicationStateDataProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RMStateVersionProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RMStateVersionProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.EpochProto; import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.EpochProto;
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.RMZKUtils; import org.apache.hadoop.yarn.server.resourcemanager.RMZKUtils;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.Epoch; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.Epoch;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.AMRMTokenSecretManagerStatePBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationAttemptStateDataPBImpl; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationAttemptStateDataPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationStateDataPBImpl; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationStateDataPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.EpochPBImpl; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.EpochPBImpl;
@ -128,6 +129,9 @@ public class ZKRMStateStore extends RMStateStore {
* | |----- Key_1 * | |----- Key_1
* | |----- Key_2 * | |----- Key_2
* .... * ....
* |--- AMRMTOKEN_SECRET_MANAGER_ROOT
* |----- currentMasterKey
* |----- nextMasterKey
* *
*/ */
private String zkRootNodePath; private String zkRootNodePath;
@ -136,6 +140,7 @@ public class ZKRMStateStore extends RMStateStore {
private String dtMasterKeysRootPath; private String dtMasterKeysRootPath;
private String delegationTokensRootPath; private String delegationTokensRootPath;
private String dtSequenceNumberPath; private String dtSequenceNumberPath;
private String amrmTokenSecretManagerRoot;
@VisibleForTesting @VisibleForTesting
protected String znodeWorkingPath; protected String znodeWorkingPath;
@ -255,6 +260,8 @@ public class ZKRMStateStore extends RMStateStore {
RM_DELEGATION_TOKENS_ROOT_ZNODE_NAME); RM_DELEGATION_TOKENS_ROOT_ZNODE_NAME);
dtSequenceNumberPath = getNodePath(rmDTSecretManagerRoot, dtSequenceNumberPath = getNodePath(rmDTSecretManagerRoot,
RM_DT_SEQUENTIAL_NUMBER_ZNODE_NAME); RM_DT_SEQUENTIAL_NUMBER_ZNODE_NAME);
amrmTokenSecretManagerRoot =
getNodePath(zkRootNodePath, AMRMTOKEN_SECRET_MANAGER_ROOT);
} }
@Override @Override
@ -275,6 +282,7 @@ public class ZKRMStateStore extends RMStateStore {
createRootDir(dtMasterKeysRootPath); createRootDir(dtMasterKeysRootPath);
createRootDir(delegationTokensRootPath); createRootDir(delegationTokensRootPath);
createRootDir(dtSequenceNumberPath); createRootDir(dtSequenceNumberPath);
createRootDir(amrmTokenSecretManagerRoot);
} }
private void createRootDir(final String rootPath) throws Exception { private void createRootDir(final String rootPath) throws Exception {
@ -427,9 +435,27 @@ public class ZKRMStateStore extends RMStateStore {
loadRMDTSecretManagerState(rmState); loadRMDTSecretManagerState(rmState);
// recover RM applications // recover RM applications
loadRMAppState(rmState); loadRMAppState(rmState);
// recover AMRMTokenSecretManager
loadAMRMTokenSecretManagerState(rmState);
return rmState; return rmState;
} }
private void loadAMRMTokenSecretManagerState(RMState rmState)
throws Exception {
byte[] data = getDataWithRetries(amrmTokenSecretManagerRoot, true);
if (data == null) {
LOG.warn("There is no data saved");
return;
}
AMRMTokenSecretManagerStatePBImpl stateData =
new AMRMTokenSecretManagerStatePBImpl(
AMRMTokenSecretManagerStateProto.parseFrom(data));
rmState.amrmTokenSecretManagerState =
AMRMTokenSecretManagerState.newInstance(
stateData.getCurrentMasterKey(), stateData.getNextMasterKey());
}
private synchronized void loadRMDTSecretManagerState(RMState rmState) private synchronized void loadRMDTSecretManagerState(RMState rmState)
throws Exception { throws Exception {
loadRMDelegationKeyState(rmState); loadRMDelegationKeyState(rmState);
@ -1112,4 +1138,19 @@ public class ZKRMStateStore extends RMStateStore {
return zk; return zk;
} }
@Override
public synchronized void storeOrUpdateAMRMTokenSecretManagerState(
AMRMTokenSecretManagerState amrmTokenSecretManagerState,
boolean isUpdate) {
AMRMTokenSecretManagerState data =
AMRMTokenSecretManagerState.newInstance(amrmTokenSecretManagerState);
byte[] stateData = data.getProto().toByteArray();
try {
setDataWithRetries(amrmTokenSecretManagerRoot, stateData, -1);
} catch (Exception ex) {
LOG.info("Error storing info for AMRMTokenSecretManager", ex);
notifyStoreOperationFailed(ex);
}
}
} }

View File

@ -0,0 +1,76 @@
/**
* 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 org.apache.hadoop.yarn.server.resourcemanager.recovery.records;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerRecoveryProtos.AMRMTokenSecretManagerStateProto;
import org.apache.hadoop.yarn.server.api.records.MasterKey;
import org.apache.hadoop.yarn.server.resourcemanager.security.AMRMTokenSecretManager;
import org.apache.hadoop.yarn.util.Records;
/**
* Contains all the state data that needs to be stored persistently
* for {@link AMRMTokenSecretManager}
*/
@Public
@Unstable
public abstract class AMRMTokenSecretManagerState {
public static AMRMTokenSecretManagerState newInstance(
MasterKey currentMasterKey, MasterKey nextMasterKey) {
AMRMTokenSecretManagerState data =
Records.newRecord(AMRMTokenSecretManagerState.class);
data.setCurrentMasterKey(currentMasterKey);
data.setNextMasterKey(nextMasterKey);
return data;
}
public static AMRMTokenSecretManagerState newInstance(
AMRMTokenSecretManagerState state) {
AMRMTokenSecretManagerState data =
Records.newRecord(AMRMTokenSecretManagerState.class);
data.setCurrentMasterKey(state.getCurrentMasterKey());
data.setNextMasterKey(state.getNextMasterKey());
return data;
}
/**
* {@link AMRMTokenSecretManager} current Master key
*/
@Public
@Unstable
public abstract MasterKey getCurrentMasterKey();
@Public
@Unstable
public abstract void setCurrentMasterKey(MasterKey currentMasterKey);
/**
* {@link AMRMTokenSecretManager} next Master key
*/
@Public
@Unstable
public abstract MasterKey getNextMasterKey();
@Public
@Unstable
public abstract void setNextMasterKey(MasterKey nextMasterKey);
public abstract AMRMTokenSecretManagerStateProto getProto();
}

View File

@ -0,0 +1,126 @@
/**
* 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 org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb;
import org.apache.hadoop.yarn.proto.YarnServerCommonProtos.MasterKeyProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerRecoveryProtos.AMRMTokenSecretManagerStateProto;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerRecoveryProtos.AMRMTokenSecretManagerStateProtoOrBuilder;
import org.apache.hadoop.yarn.server.api.records.MasterKey;
import org.apache.hadoop.yarn.server.api.records.impl.pb.MasterKeyPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
public class AMRMTokenSecretManagerStatePBImpl extends AMRMTokenSecretManagerState{
AMRMTokenSecretManagerStateProto proto =
AMRMTokenSecretManagerStateProto.getDefaultInstance();
AMRMTokenSecretManagerStateProto.Builder builder = null;
boolean viaProto = false;
private MasterKey currentMasterKey = null;
private MasterKey nextMasterKey = null;
public AMRMTokenSecretManagerStatePBImpl() {
builder = AMRMTokenSecretManagerStateProto.newBuilder();
}
public AMRMTokenSecretManagerStatePBImpl(AMRMTokenSecretManagerStateProto proto) {
this.proto = proto;
viaProto = true;
}
public AMRMTokenSecretManagerStateProto getProto() {
mergeLocalToProto();
proto = viaProto ? proto : builder.build();
viaProto = true;
return proto;
}
private void mergeLocalToBuilder() {
if (this.currentMasterKey != null) {
builder.setCurrentMasterKey(convertToProtoFormat(this.currentMasterKey));
}
if (this.nextMasterKey != null) {
builder.setNextMasterKey(convertToProtoFormat(this.nextMasterKey));
}
}
private void mergeLocalToProto() {
if (viaProto)
maybeInitBuilder();
mergeLocalToBuilder();
proto = builder.build();
viaProto = true;
}
private void maybeInitBuilder() {
if (viaProto || builder == null) {
builder = AMRMTokenSecretManagerStateProto.newBuilder(proto);
}
viaProto = false;
}
@Override
public MasterKey getCurrentMasterKey() {
AMRMTokenSecretManagerStateProtoOrBuilder p = viaProto ? proto : builder;
if (this.currentMasterKey != null) {
return this.currentMasterKey;
}
if (!p.hasCurrentMasterKey()) {
return null;
}
this.currentMasterKey = convertFromProtoFormat(p.getCurrentMasterKey());
return this.currentMasterKey;
}
@Override
public void setCurrentMasterKey(MasterKey currentMasterKey) {
maybeInitBuilder();
if (currentMasterKey == null)
builder.clearCurrentMasterKey();
this.currentMasterKey = currentMasterKey;
}
@Override
public MasterKey getNextMasterKey() {
AMRMTokenSecretManagerStateProtoOrBuilder p = viaProto ? proto : builder;
if (this.nextMasterKey != null) {
return this.nextMasterKey;
}
if (!p.hasNextMasterKey()) {
return null;
}
this.nextMasterKey = convertFromProtoFormat(p.getNextMasterKey());
return this.nextMasterKey;
}
@Override
public void setNextMasterKey(MasterKey nextMasterKey) {
maybeInitBuilder();
if (nextMasterKey == null)
builder.clearNextMasterKey();
this.nextMasterKey = nextMasterKey;
}
private MasterKeyProto convertToProtoFormat(MasterKey t) {
return ((MasterKeyPBImpl) t).getProto();
}
private MasterKeyPBImpl convertFromProtoFormat(MasterKeyProto p) {
return new MasterKeyPBImpl(p);
}
}

View File

@ -38,6 +38,10 @@ import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier; import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.server.api.records.MasterKey;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
import org.apache.hadoop.yarn.server.security.MasterKeyData; import org.apache.hadoop.yarn.server.security.MasterKeyData;
import com.google.common.annotations.VisibleForTesting; import com.google.common.annotations.VisibleForTesting;
@ -66,6 +70,7 @@ public class AMRMTokenSecretManager extends
private final Timer timer; private final Timer timer;
private final long rollingInterval; private final long rollingInterval;
private final long activationDelay; private final long activationDelay;
private RMContext rmContext;
private final Set<ApplicationAttemptId> appAttemptSet = private final Set<ApplicationAttemptId> appAttemptSet =
new HashSet<ApplicationAttemptId>(); new HashSet<ApplicationAttemptId>();
@ -73,7 +78,8 @@ public class AMRMTokenSecretManager extends
/** /**
* Create an {@link AMRMTokenSecretManager} * Create an {@link AMRMTokenSecretManager}
*/ */
public AMRMTokenSecretManager(Configuration conf) { public AMRMTokenSecretManager(Configuration conf, RMContext rmContext) {
this.rmContext = rmContext;
this.timer = new Timer(); this.timer = new Timer();
this.rollingInterval = this.rollingInterval =
conf conf
@ -98,6 +104,11 @@ public class AMRMTokenSecretManager extends
public void start() { public void start() {
if (this.currentMasterKey == null) { if (this.currentMasterKey == null) {
this.currentMasterKey = createNewMasterKey(); this.currentMasterKey = createNewMasterKey();
AMRMTokenSecretManagerState state =
AMRMTokenSecretManagerState.newInstance(
this.currentMasterKey.getMasterKey(), null);
rmContext.getStateStore().storeOrUpdateAMRMTokenSecretManagerState(state,
false);
} }
this.timer.scheduleAtFixedRate(new MasterKeyRoller(), rollingInterval, this.timer.scheduleAtFixedRate(new MasterKeyRoller(), rollingInterval,
rollingInterval); rollingInterval);
@ -130,6 +141,12 @@ public class AMRMTokenSecretManager extends
try { try {
LOG.info("Rolling master-key for amrm-tokens"); LOG.info("Rolling master-key for amrm-tokens");
this.nextMasterKey = createNewMasterKey(); this.nextMasterKey = createNewMasterKey();
AMRMTokenSecretManagerState state =
AMRMTokenSecretManagerState.newInstance(
this.currentMasterKey.getMasterKey(),
this.nextMasterKey.getMasterKey());
rmContext.getStateStore().storeOrUpdateAMRMTokenSecretManagerState(state,
true);
this.timer.schedule(new NextKeyActivator(), this.activationDelay); this.timer.schedule(new NextKeyActivator(), this.activationDelay);
} finally { } finally {
this.writeLock.unlock(); this.writeLock.unlock();
@ -225,8 +242,8 @@ public class AMRMTokenSecretManager extends
LOG.debug("Trying to retrieve password for " + applicationAttemptId); LOG.debug("Trying to retrieve password for " + applicationAttemptId);
} }
if (!appAttemptSet.contains(applicationAttemptId)) { if (!appAttemptSet.contains(applicationAttemptId)) {
throw new InvalidToken("Password not found for ApplicationAttempt " throw new InvalidToken(applicationAttemptId
+ applicationAttemptId); + " not found in AMRMTokenSecretManager.");
} }
if (identifier.getKeyId() == this.currentMasterKey.getMasterKey() if (identifier.getKeyId() == this.currentMasterKey.getMasterKey()
.getKeyId()) { .getKeyId()) {
@ -238,9 +255,7 @@ public class AMRMTokenSecretManager extends
return createPassword(identifier.getBytes(), return createPassword(identifier.getBytes(),
this.nextMasterKey.getSecretKey()); this.nextMasterKey.getSecretKey());
} }
throw new InvalidToken("Given AMRMToken for application : " throw new InvalidToken("Invalid AMRMToken from " + applicationAttemptId);
+ applicationAttemptId.toString()
+ " seems to have been generated illegally.");
} finally { } finally {
this.readLock.unlock(); this.readLock.unlock();
} }
@ -291,4 +306,25 @@ public class AMRMTokenSecretManager extends
this.readLock.unlock(); this.readLock.unlock();
} }
} }
public void recover(RMState state) {
if (state.getAMRMTokenSecretManagerState() != null) {
// recover the current master key
MasterKey currentKey =
state.getAMRMTokenSecretManagerState().getCurrentMasterKey();
this.currentMasterKey =
new MasterKeyData(currentKey, createSecretKey(currentKey.getBytes()
.array()));
// recover the next master key if not null
MasterKey nextKey =
state.getAMRMTokenSecretManagerState().getNextMasterKey();
if (nextKey != null) {
this.nextMasterKey =
new MasterKeyData(nextKey, createSecretKey(nextKey.getBytes()
.array()));
this.timer.schedule(new NextKeyActivator(), this.activationDelay);
}
}
}
} }

View File

@ -0,0 +1,30 @@
/**
* 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.
*/
option java_package = "org.apache.hadoop.yarn.proto";
option java_outer_classname = "YarnServerResourceManagerRecoveryProtos";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
package hadoop.yarn;
import "yarn_server_common_protos.proto";
message AMRMTokenSecretManagerStateProto {
optional MasterKeyProto current_master_key = 1;
optional MasterKeyProto next_master_key = 2;
}

View File

@ -1250,11 +1250,10 @@ public class TestRMRestart {
.getEncoded()); .getEncoded());
// assert AMRMTokenSecretManager also knows about the AMRMToken password // assert AMRMTokenSecretManager also knows about the AMRMToken password
// TODO: fix this on YARN-2211 Token<AMRMTokenIdentifier> amrmToken = loadedAttempt1.getAMRMToken();
// Token<AMRMTokenIdentifier> amrmToken = loadedAttempt1.getAMRMToken(); Assert.assertArrayEquals(amrmToken.getPassword(),
// Assert.assertArrayEquals(amrmToken.getPassword(), rm2.getRMContext().getAMRMTokenSecretManager().retrievePassword(
// rm2.getRMContext().getAMRMTokenSecretManager().retrievePassword( amrmToken.decodeIdentifier()));
// amrmToken.decodeIdentifier()));
rm1.stop(); rm1.stop();
rm2.stop(); rm2.stop();
} }

View File

@ -55,10 +55,12 @@ import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler; import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier; import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationAttemptState; import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationAttemptState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState; import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMDTSecretManagerState; import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMDTSecretManagerState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState; import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenSecretManagerState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
@ -176,8 +178,12 @@ public class RMStateStoreTestBase extends ClientBaseWithFixes{
TestDispatcher dispatcher = new TestDispatcher(); TestDispatcher dispatcher = new TestDispatcher();
store.setRMDispatcher(dispatcher); store.setRMDispatcher(dispatcher);
AMRMTokenSecretManager appTokenMgr = spy( RMContext rmContext = mock(RMContext.class);
new AMRMTokenSecretManager(conf)); when(rmContext.getStateStore()).thenReturn(store);
AMRMTokenSecretManager appTokenMgr =
spy(new AMRMTokenSecretManager(conf, rmContext));
MasterKeyData masterKeyData = appTokenMgr.createNewMasterKey(); MasterKeyData masterKeyData = appTokenMgr.createNewMasterKey();
when(appTokenMgr.getMasterKey()).thenReturn(masterKeyData); when(appTokenMgr.getMasterKey()).thenReturn(masterKeyData);
@ -576,4 +582,65 @@ public class RMStateStoreTestBase extends ClientBaseWithFixes{
} }
public void testAMRMTokenSecretManagerStateStore(
RMStateStoreHelper stateStoreHelper) throws Exception {
System.out.println("Start testing");
RMStateStore store = stateStoreHelper.getRMStateStore();
TestDispatcher dispatcher = new TestDispatcher();
store.setRMDispatcher(dispatcher);
RMContext rmContext = mock(RMContext.class);
when(rmContext.getStateStore()).thenReturn(store);
Configuration conf = new YarnConfiguration();
AMRMTokenSecretManager appTokenMgr =
new AMRMTokenSecretManager(conf, rmContext);
//create and save the first masterkey
MasterKeyData firstMasterKeyData = appTokenMgr.createNewMasterKey();
AMRMTokenSecretManagerState state1 =
AMRMTokenSecretManagerState.newInstance(
firstMasterKeyData.getMasterKey(), null);
rmContext.getStateStore().storeOrUpdateAMRMTokenSecretManagerState(state1,
false);
// load state
store = stateStoreHelper.getRMStateStore();
store.setRMDispatcher(dispatcher);
RMState state = store.loadState();
Assert.assertNotNull(state.getAMRMTokenSecretManagerState());
Assert.assertEquals(firstMasterKeyData.getMasterKey(), state
.getAMRMTokenSecretManagerState().getCurrentMasterKey());
Assert.assertNull(state
.getAMRMTokenSecretManagerState().getNextMasterKey());
//create and save the second masterkey
MasterKeyData secondMasterKeyData = appTokenMgr.createNewMasterKey();
AMRMTokenSecretManagerState state2 =
AMRMTokenSecretManagerState
.newInstance(firstMasterKeyData.getMasterKey(),
secondMasterKeyData.getMasterKey());
rmContext.getStateStore().storeOrUpdateAMRMTokenSecretManagerState(state2,
true);
// load state
store = stateStoreHelper.getRMStateStore();
store.setRMDispatcher(dispatcher);
RMState state_2 = store.loadState();
Assert.assertNotNull(state_2.getAMRMTokenSecretManagerState());
Assert.assertEquals(firstMasterKeyData.getMasterKey(), state_2
.getAMRMTokenSecretManagerState().getCurrentMasterKey());
Assert.assertEquals(secondMasterKeyData.getMasterKey(), state_2
.getAMRMTokenSecretManagerState().getNextMasterKey());
// re-create the masterKeyData based on the recovered masterkey
// should have the same secretKey
appTokenMgr.recover(state_2);
Assert.assertEquals(appTokenMgr.getCurrnetMasterKeyData().getSecretKey(),
firstMasterKeyData.getSecretKey());
Assert.assertEquals(appTokenMgr.getNextMasterKeyData().getSecretKey(),
secondMasterKeyData.getSecretKey());
store.close();
}
} }

View File

@ -38,7 +38,6 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.RMStateVersion;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationStateDataPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.RMStateVersionPBImpl; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.RMStateVersionPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
@ -161,6 +160,7 @@ public class TestFSRMStateStore extends RMStateStoreTestBase {
testEpoch(fsTester); testEpoch(fsTester);
testAppDeletion(fsTester); testAppDeletion(fsTester);
testDeleteStore(fsTester); testDeleteStore(fsTester);
testAMRMTokenSecretManagerStateStore(fsTester);
} finally { } finally {
cluster.shutdown(); cluster.shutdown();
} }

View File

@ -123,6 +123,7 @@ public class TestZKRMStateStore extends RMStateStoreTestBase {
testEpoch(zkTester); testEpoch(zkTester);
testAppDeletion(zkTester); testAppDeletion(zkTester);
testDeleteStore(zkTester); testDeleteStore(zkTester);
testAMRMTokenSecretManagerStateStore(zkTester);
} }
private Configuration createHARMConf( private Configuration createHARMConf(

View File

@ -193,7 +193,7 @@ public class TestRMAppTransitions {
this.rmContext = this.rmContext =
new RMContextImpl(rmDispatcher, new RMContextImpl(rmDispatcher,
containerAllocationExpirer, amLivelinessMonitor, amFinishingMonitor, containerAllocationExpirer, amLivelinessMonitor, amFinishingMonitor,
null, new AMRMTokenSecretManager(conf), null, new AMRMTokenSecretManager(conf, this.rmContext),
new RMContainerTokenSecretManager(conf), new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf), new NMTokenSecretManagerInRM(conf),
new ClientToAMTokenSecretManagerInRM(), new ClientToAMTokenSecretManagerInRM(),

View File

@ -134,7 +134,8 @@ public class TestRMAppAttemptTransitions {
private RMAppAttempt applicationAttempt; private RMAppAttempt applicationAttempt;
private Configuration conf = new Configuration(); private Configuration conf = new Configuration();
private AMRMTokenSecretManager amRMTokenManager = spy(new AMRMTokenSecretManager(conf)); private AMRMTokenSecretManager amRMTokenManager =
spy(new AMRMTokenSecretManager(conf, rmContext));
private ClientToAMTokenSecretManagerInRM clientToAMTokenManager = private ClientToAMTokenSecretManagerInRM clientToAMTokenManager =
spy(new ClientToAMTokenSecretManagerInRM()); spy(new ClientToAMTokenSecretManagerInRM());
private NMTokenSecretManagerInRM nmTokenManager = private NMTokenSecretManagerInRM nmTokenManager =

View File

@ -86,13 +86,12 @@ public class TestUtils {
Configuration conf = new Configuration(); Configuration conf = new Configuration();
RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class); RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
RMContext rmContext = RMContextImpl rmContext =
new RMContextImpl(nullDispatcher, cae, null, null, null, new RMContextImpl(nullDispatcher, cae, null, null, null,
new AMRMTokenSecretManager(conf), new AMRMTokenSecretManager(conf, null),
new RMContainerTokenSecretManager(conf), new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf), new NMTokenSecretManagerInRM(conf),
new ClientToAMTokenSecretManagerInRM(), writer); new ClientToAMTokenSecretManagerInRM(), writer);
return rmContext; return rmContext;
} }

View File

@ -184,8 +184,8 @@ public class TestAMRMTokens {
// The exception will still have the earlier appAttemptId as it picks it // The exception will still have the earlier appAttemptId as it picks it
// up from the token. // up from the token.
Assert.assertTrue(t.getCause().getMessage().contains( Assert.assertTrue(t.getCause().getMessage().contains(
"Password not found for ApplicationAttempt " + applicationAttemptId.toString()
applicationAttemptId.toString())); + " not found in AMRMTokenSecretManager."));
} }
} finally { } finally {