import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public void recover(RMState rmState) throws Exception {

  LOG.info("recovering RMDelegationTokenSecretManager.");
  // recover RMDTMasterKeys
  for (DelegationKey dtKey : rmState.getRMDTSecretManagerState()
    .getMasterKeyState()) {
    addKey(dtKey);
  }

  // recover RMDelegationTokens
  Map<RMDelegationTokenIdentifier, Long> rmDelegationTokens =
      rmState.getRMDTSecretManagerState().getTokenState();
  this.delegationTokenSequenceNumber =
      rmState.getRMDTSecretManagerState().getDTSequenceNumber();
  for (Map.Entry<RMDelegationTokenIdentifier, Long> entry : rmDelegationTokens
    .entrySet()) {
    addPersistedDelegationToken(entry.getKey(), entry.getValue());
  }
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public RenewDelegationTokenResponse renewDelegationToken(
    RenewDelegationTokenRequest request) throws YarnException {
  try {
    if (!isAllowedDelegationTokenOp()) {
      throw new IOException(
          "Delegation Token can be renewed only with kerberos authentication");
    }
    
    org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
    Token<RMDelegationTokenIdentifier> token = new Token<RMDelegationTokenIdentifier>(
        protoToken.getIdentifier().array(), protoToken.getPassword().array(),
        new Text(protoToken.getKind()), new Text(protoToken.getService()));

    String user = getRenewerForToken(token);
    long nextExpTime = rmDTSecretManager.renewToken(token, user);
    RenewDelegationTokenResponse renewResponse = Records
        .newRecord(RenewDelegationTokenResponse.class);
    renewResponse.setNextExpirationTime(nextExpTime);
    return renewResponse;
  } catch (IOException e) {
    throw RPCUtil.getRemoteException(e);
  }
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
public CancelDelegationTokenResponse cancelDelegationToken(
    CancelDelegationTokenRequest request) throws YarnException {
  try {
    if (!isAllowedDelegationTokenOp()) {
      throw new IOException(
          "Delegation Token can be cancelled only with kerberos authentication");
    }
    org.apache.hadoop.yarn.api.records.Token protoToken = request.getDelegationToken();
    Token<RMDelegationTokenIdentifier> token = new Token<RMDelegationTokenIdentifier>(
        protoToken.getIdentifier().array(), protoToken.getPassword().array(),
        new Text(protoToken.getKind()), new Text(protoToken.getService()));

    String user = UserGroupInformation.getCurrentUser().getUserName();
    rmDTSecretManager.cancelToken(token, user);
    return Records.newRecord(CancelDelegationTokenResponse.class);
  } catch (IOException e) {
    throw RPCUtil.getRemoteException(e);
  }
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected synchronized void removeRMDelegationTokenState(
    RMDelegationTokenIdentifier rmDTIdentifier) throws Exception {
  String nodeRemovePath =
      getNodePath(delegationTokensRootPath, DELEGATION_TOKEN_PREFIX
          + rmDTIdentifier.getSequenceNumber());
  if (LOG.isDebugEnabled()) {
    LOG.debug("Removing RMDelegationToken_"
        + rmDTIdentifier.getSequenceNumber());
  }
  if (existsWithRetries(nodeRemovePath, false) != null) {
    ArrayList<Op> opList = new ArrayList<Op>();
    opList.add(Op.delete(nodeRemovePath, -1));
    doDeleteMultiWithRetries(opList);
  } else {
    LOG.debug("Attempted to delete a non-existing znode " + nodeRemovePath);
  }
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected synchronized void updateRMDelegationTokenState(
    RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
    throws Exception {
  ArrayList<Op> opList = new ArrayList<Op>();
  String nodeRemovePath =
      getNodePath(delegationTokensRootPath, DELEGATION_TOKEN_PREFIX
          + rmDTIdentifier.getSequenceNumber());
  if (existsWithRetries(nodeRemovePath, false) == null) {
    // in case znode doesn't exist
    addStoreOrUpdateOps(opList, rmDTIdentifier, renewDate, false);
    LOG.debug("Attempted to update a non-existing znode " + nodeRemovePath);
  } else {
    // in case znode exists
    addStoreOrUpdateOps(opList, rmDTIdentifier, renewDate, true);
  }
  doStoreMultiWithRetries(opList);
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private void storeOrUpdateRMDT(RMDelegationTokenIdentifier rmDTIdentifier,
    Long renewDate, boolean isUpdate) throws Exception {
  Map<RMDelegationTokenIdentifier, Long> rmDTState =
      state.rmSecretManagerState.getTokenState();
  if (rmDTState.containsKey(rmDTIdentifier)) {
    IOException e = new IOException("RMDelegationToken: " + rmDTIdentifier
        + "is already stored.");
    LOG.info("Error storing info for RMDelegationToken: " + rmDTIdentifier, e);
    throw e;
  }
  rmDTState.put(rmDTIdentifier, renewDate);
  if(!isUpdate) {
    state.rmSecretManagerState.dtSequenceNumber = 
        rmDTIdentifier.getSequenceNumber();
  }
  LOG.info("Store RMDT with sequence number "
           + rmDTIdentifier.getSequenceNumber());
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
public RMDelegationTokenIdentifierForTest(RMDelegationTokenIdentifier token,
    String message) {
  if (token.getOwner() != null) {
    setOwner(new Text(token.getOwner()));
  }
  if (token.getRenewer() != null) {
    setRenewer(new Text(token.getRenewer()));
  }
  if (token.getRealUser() != null) {
    setRealUser(new Text(token.getRealUser()));
  }
  setIssueDate(token.getIssueDate());
  setMaxDate(token.getMaxDate());
  setSequenceNumber(token.getSequenceNumber());
  setMasterKeyId(token.getMasterKeyId());
  builder.setMessage(message);
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private void assertTokenCancelled(String encodedToken) throws Exception {
  Token<RMDelegationTokenIdentifier> realToken =
      new Token<RMDelegationTokenIdentifier>();
  realToken.decodeFromUrlString(encodedToken);
  RMDelegationTokenIdentifier ident = rm.getRMContext()
    .getRMDelegationTokenSecretManager().decodeTokenIdentifier(realToken);
  boolean exceptionCaught = false;
  try {
    rm.getRMContext().getRMDelegationTokenSecretManager()
      .verifyToken(ident, realToken.getPassword());
  } catch (InvalidToken it) {
    exceptionCaught = true;
  }
  assertTrue("InvalidToken exception not thrown", exceptionCaught);
  assertFalse(rm.getRMContext().getRMDelegationTokenSecretManager()
    .getAllTokens().containsKey(ident));
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private void checkTokenRenewal(UserGroupInformation owner,
    UserGroupInformation renewer) throws IOException, YarnException {
  RMDelegationTokenIdentifier tokenIdentifier =
      new RMDelegationTokenIdentifier(
          new Text(owner.getUserName()), new Text(renewer.getUserName()), null);
  Token<?> token =
      new Token<RMDelegationTokenIdentifier>(tokenIdentifier, dtsm);
  org.apache.hadoop.yarn.api.records.Token dToken = BuilderUtils.newDelegationToken(
      token.getIdentifier(), token.getKind().toString(),
      token.getPassword(), token.getService().toString());
  RenewDelegationTokenRequest request =
      Records.newRecord(RenewDelegationTokenRequest.class);
  request.setDelegationToken(dToken);

  RMContext rmContext = mock(RMContext.class);
  ClientRMService rmService = new ClientRMService(
      rmContext, null, null, null, null, dtsm);
  rmService.renewDelegationToken(request);
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private void checkTokenCancellation(ClientRMService rmService,
    UserGroupInformation owner, UserGroupInformation renewer)
    throws IOException, YarnException {
  RMDelegationTokenIdentifier tokenIdentifier =
      new RMDelegationTokenIdentifier(new Text(owner.getUserName()),
        new Text(renewer.getUserName()), null);
  Token<?> token =
      new Token<RMDelegationTokenIdentifier>(tokenIdentifier, dtsm);
  org.apache.hadoop.yarn.api.records.Token dToken =
      BuilderUtils.newDelegationToken(token.getIdentifier(), token.getKind()
        .toString(), token.getPassword(), token.getService().toString());
  CancelDelegationTokenRequest request =
      Records.newRecord(CancelDelegationTokenRequest.class);
  request.setDelegationToken(dToken);
  rmService.cancelDelegationToken(request);
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected synchronized void updateRMDelegationTokenState(
    RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
    throws Exception {
  SafeTransaction trx = new SafeTransaction();
  String nodeRemovePath =
      getNodePath(delegationTokensRootPath, DELEGATION_TOKEN_PREFIX
          + rmDTIdentifier.getSequenceNumber());
  if (exists(nodeRemovePath)) {
    // in case znode exists
    addStoreOrUpdateOps(trx, rmDTIdentifier, renewDate, true);
  } else {
    // in case znode doesn't exist
    addStoreOrUpdateOps(trx, rmDTIdentifier, renewDate, false);
    LOG.debug("Attempted to update a non-existing znode " + nodeRemovePath);
  }
  trx.commit();
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private void checkTokenCancellation(ClientRMService rmService,
    UserGroupInformation owner, UserGroupInformation renewer)
    throws IOException, YarnException {
  RMDelegationTokenIdentifier tokenIdentifier =
      new RMDelegationTokenIdentifier(new Text(owner.getUserName()),
          new Text(renewer.getUserName()), null);
  Token<?> token =
      new Token<RMDelegationTokenIdentifier>(tokenIdentifier, dtsm);
  org.apache.hadoop.yarn.api.records.Token dToken =
      BuilderUtils.newDelegationToken(token.getIdentifier(), token.getKind()
          .toString(), token.getPassword(), token.getService().toString());
  CancelDelegationTokenRequest request =
      Records.newRecord(CancelDelegationTokenRequest.class);
  request.setDelegationToken(dToken);
  rmService.cancelDelegationToken(request);
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected void storeNewToken(RMDelegationTokenIdentifier identifier,
    long renewDate) {
  try {
    LOG.info("storing RMDelegation token with sequence number: "
        + identifier.getSequenceNumber());
    rmContext.getStateStore().storeRMDelegationToken(identifier, renewDate);
  } catch (Exception e) {
    LOG.error("Error in storing RMDelegationToken with sequence number: "
        + identifier.getSequenceNumber());
    ExitUtil.terminate(1, e);
  }
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected void updateStoredToken(RMDelegationTokenIdentifier id,
    long renewDate) {
  try {
    LOG.info("updating RMDelegation token with sequence number: "
        + id.getSequenceNumber());
    rmContext.getStateStore().updateRMDelegationToken(id, renewDate);
  } catch (Exception e) {
    LOG.error("Error in updating persisted RMDelegationToken" +
              " with sequence number: " + id.getSequenceNumber());
    ExitUtil.terminate(1, e);
  }
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Override
protected void removeStoredToken(RMDelegationTokenIdentifier ident)
    throws IOException {
  try {
    LOG.info("removing RMDelegation token with sequence number: "
        + ident.getSequenceNumber());
    rmContext.getStateStore().removeRMDelegationToken(ident);
  } catch (Exception e) {
    LOG.error("Error in removing RMDelegationToken with sequence number: "
        + ident.getSequenceNumber());
    ExitUtil.terminate(1, e);
  }
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
@Private
@VisibleForTesting
public synchronized Map<RMDelegationTokenIdentifier, Long> getAllTokens() {
  Map<RMDelegationTokenIdentifier, Long> allTokens =
      new HashMap<RMDelegationTokenIdentifier, Long>();

  for (Map.Entry<RMDelegationTokenIdentifier,
      DelegationTokenInformation> entry : currentTokens.entrySet()) {
    allTokens.put(entry.getKey(), entry.getValue().getRenewDate());
  }
  return allTokens;
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
public long getRenewDate(RMDelegationTokenIdentifier ident)
    throws InvalidToken {
  DelegationTokenInformation info = currentTokens.get(ident);
  if (info == null) {
    throw new InvalidToken("token (" + ident.toString()
        + ") can't be found in cache");
  }
  return info.getRenewDate();
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private Response createDelegationToken(DelegationToken tokenData,
    HttpServletRequest hsr, UserGroupInformation callerUGI)
    throws AuthorizationException, IOException, InterruptedException,
    Exception {

  final String renewer = tokenData.getRenewer();
  GetDelegationTokenResponse resp;
  try {
    resp =
        callerUGI
          .doAs(new PrivilegedExceptionAction<GetDelegationTokenResponse>() {
            @Override
            public GetDelegationTokenResponse run() throws IOException,
                YarnException {
              GetDelegationTokenRequest createReq =
                  GetDelegationTokenRequest.newInstance(renewer);
              return rm.getClientRMService().getDelegationToken(createReq);
            }
          });
  } catch (Exception e) {
    LOG.info("Create delegation token request failed", e);
    throw e;
  }

  Token<RMDelegationTokenIdentifier> tk =
      new Token<RMDelegationTokenIdentifier>(resp.getRMDelegationToken()
        .getIdentifier().array(), resp.getRMDelegationToken().getPassword()
        .array(), new Text(resp.getRMDelegationToken().getKind()), new Text(
        resp.getRMDelegationToken().getService()));
  RMDelegationTokenIdentifier identifier = tk.decodeIdentifier();
  long currentExpiration =
      rm.getRMContext().getRMDelegationTokenSecretManager()
        .getRenewDate(identifier);
  DelegationToken respToken =
      new DelegationToken(tk.encodeToUrlString(), renewer, identifier
        .getOwner().toString(), tk.getKind().toString(), currentExpiration,
        identifier.getMaxDate());
  return Response.status(Status.OK).entity(respToken).build();
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private Token<RMDelegationTokenIdentifier> extractToken(
    HttpServletRequest request) {
  String encodedToken = request.getHeader(DELEGATION_TOKEN_HEADER);
  if (encodedToken == null) {
    String msg =
        "Header '" + DELEGATION_TOKEN_HEADER
            + "' containing encoded token not found";
    throw new BadRequestException(msg);
  }
  return extractToken(encodedToken);
}
 

import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; //导入依赖的package包/类
private String getRenewerForToken(Token<RMDelegationTokenIdentifier> token)
    throws IOException {
  UserGroupInformation user = UserGroupInformation.getCurrentUser();
  UserGroupInformation loginUser = UserGroupInformation.getLoginUser();
  // we can always renew our own tokens
  return loginUser.getUserName().equals(user.getUserName())
      ? token.decodeIdentifier().getRenewer().toString()
      : user.getShortUserName();
}