import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@GET
@Path("/apps/{appid}/state")
@Produces({ MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML })
public AppState getAppState(@Context HttpServletRequest hsr,
    @PathParam("appid") String appId) throws AuthorizationException {
  init();
  UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
  String userName = "";
  if (callerUGI != null) {
    userName = callerUGI.getUserName();
  }
  RMApp app = null;
  try {
    app = getRMAppForAppId(appId);
  } catch (NotFoundException e) {
    RMAuditLogger.logFailure(userName, AuditConstants.KILL_APP_REQUEST,
      "UNKNOWN", "RMWebService",
      "Trying to get state of an absent application " + appId);
    throw e;
  }

  AppState ret = new AppState();
  ret.setState(app.getState().toString());

  return ret;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@GET
@Path("/apps/{appid}/queue")
@Produces({ MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML })
public AppQueue getAppQueue(@Context HttpServletRequest hsr,
    @PathParam("appid") String appId) throws AuthorizationException {
  init();
  UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
  String userName = "UNKNOWN-USER";
  if (callerUGI != null) {
    userName = callerUGI.getUserName();
  }
  RMApp app = null;
  try {
    app = getRMAppForAppId(appId);
  } catch (NotFoundException e) {
    RMAuditLogger.logFailure(userName, AuditConstants.KILL_APP_REQUEST,
      "UNKNOWN", "RMWebService",
      "Trying to get state of an absent application " + appId);
    throw e;
  }

  AppQueue ret = new AppQueue();
  ret.setQueue(app.getQueue());

  return ret;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
private String checkReservationACLs(String queueName, String auditConstant)
    throws YarnException {
  UserGroupInformation callerUGI;
  try {
    callerUGI = UserGroupInformation.getCurrentUser();
  } catch (IOException ie) {
    RMAuditLogger.logFailure("UNKNOWN", auditConstant, queueName,
        "ClientRMService", "Error getting UGI");
    throw RPCUtil.getRemoteException(ie);
  }
  // Check if user has access on the managed queue
  if (!queueACLsManager.checkAccess(callerUGI, QueueACL.SUBMIT_APPLICATIONS,
      queueName)) {
    RMAuditLogger.logFailure(
        callerUGI.getShortUserName(),
        auditConstant,
        "User doesn't have permissions to "
            + QueueACL.SUBMIT_APPLICATIONS.toString(), "ClientRMService",
        AuditConstants.UNAUTHORIZED_USER);
    throw RPCUtil.getRemoteException(new AccessControlException("User "
        + callerUGI.getShortUserName() + " cannot perform operation "
        + QueueACL.SUBMIT_APPLICATIONS.name() + " on queue" + queueName));
  }
  return callerUGI.getShortUserName();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
protected void createReleaseCache() {
  // Cleanup the cache after nm expire interval.
  new Timer().schedule(new TimerTask() {
    @Override
    public void run() {
      for (SchedulerApplication<T> app : applications.values()) {

        T attempt = app.getCurrentAppAttempt();
        synchronized (attempt) {
          for (ContainerId containerId : attempt.getPendingRelease()) {
            RMAuditLogger.logFailure(
              app.getUser(),
              AuditConstants.RELEASE_CONTAINER,
              "Unauthorized access or invalid container",
              "Scheduler",
              "Trying to release container not owned by app or with invalid id.",
              attempt.getApplicationId(), containerId);
          }
          attempt.getPendingRelease().clear();
        }
      }
      LOG.info("Release request cache is cleaned up");
    }
  }, nmExpireInterval);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
protected void releaseContainers(List<ContainerId> containers,
    SchedulerApplicationAttempt attempt) {
  for (ContainerId containerId : containers) {
    RMContainer rmContainer = getRMContainer(containerId);
    if (rmContainer == null) {
      if (System.currentTimeMillis() - ResourceManager.getClusterTimeStamp()
          < nmExpireInterval) {
        LOG.info(containerId + " doesn't exist. Add the container"
            + " to the release request cache as it maybe on recovery.");
        synchronized (attempt) {
          attempt.getPendingRelease().add(containerId);
        }
      } else {
        RMAuditLogger.logFailure(attempt.getUser(),
          AuditConstants.RELEASE_CONTAINER,
          "Unauthorized access or invalid container", "Scheduler",
          "Trying to release container not owned by app or with invalid id.",
          attempt.getApplicationId(), containerId);
      }
    }
    completedContainer(rmContainer,
      SchedulerUtils.createAbnormalContainerStatus(containerId,
        SchedulerUtils.RELEASED_CONTAINER), RMContainerEventType.RELEASED);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@VisibleForTesting
public void clearPendingContainerCache() {
  for (SchedulerApplication<T> app : applications.values()) {
    T attempt = app.getCurrentAppAttempt();
    if (attempt != null) {
      synchronized (attempt) {
        for (ContainerId containerId : attempt.getPendingRelease()) {
          RMAuditLogger.logFailure(app.getUser(),
              AuditConstants.RELEASE_CONTAINER,
              "Unauthorized access or invalid container", "Scheduler",
              "Trying to release container not owned by app "
                  + "or with invalid id.", attempt.getApplicationId(),
              containerId);
        }
        attempt.getPendingRelease().clear();
      }
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@GET
@Path("/apps/{appid}/state")
@Produces({ MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML })
public AppState getAppState(@Context HttpServletRequest hsr,
    @PathParam("appid") String appId) throws AuthorizationException {
  init();
  UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
  String userName = "";
  if (callerUGI != null) {
    userName = callerUGI.getUserName();
  }
  RMApp app = null;
  try {
    app = getRMAppForAppId(appId);
  } catch (NotFoundException e) {
    RMAuditLogger.logFailure(userName, AuditConstants.GET_APP_STATE,
      "UNKNOWN", "RMWebService",
      "Trying to get state of an absent application " + appId);
    throw e;
  }

  AppState ret = new AppState();
  ret.setState(app.getState().toString());

  return ret;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@GET
@Path("/apps/{appid}/queue")
@Produces({ MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML })
public AppQueue getAppQueue(@Context HttpServletRequest hsr,
    @PathParam("appid") String appId) throws AuthorizationException {
  init();
  UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
  String userName = "UNKNOWN-USER";
  if (callerUGI != null) {
    userName = callerUGI.getUserName();
  }
  RMApp app = null;
  try {
    app = getRMAppForAppId(appId);
  } catch (NotFoundException e) {
    RMAuditLogger.logFailure(userName, AuditConstants.GET_APP_QUEUE,
      "UNKNOWN", "RMWebService",
      "Trying to get queue of an absent application " + appId);
    throw e;
  }

  AppQueue ret = new AppQueue();
  ret.setQueue(app.getQueue());

  return ret;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
/**
 * Quick validation on the input to check some obvious fail conditions (fail
 * fast) the input and returns the appropriate {@link Plan} associated with
 * the specified {@link Queue} or throws an exception message illustrating the
 * details of any validation check failures
 * 
 * @param reservationSystem the {@link ReservationSystem} to validate against
 * @param request the {@link ReservationSubmissionRequest} defining the
 *          resources required over time for the request
 * @param reservationId the {@link ReservationId} associated with the current
 *          request
 * @return the {@link Plan} to submit the request to
 * @throws YarnException
 */
public Plan validateReservationSubmissionRequest(
    ReservationSystem reservationSystem,
    ReservationSubmissionRequest request, ReservationId reservationId)
    throws YarnException {
  String message;
  if (reservationId == null) {
    message = "Reservation id cannot be null. Please try again " +
      "specifying a valid reservation id by creating a new reservation id.";
    throw RPCUtil.getRemoteException(message);
  }
  // Check if it is a managed queue
  String queue = request.getQueue();
  Plan plan = getPlanFromQueue(reservationSystem, queue,
          AuditConstants.SUBMIT_RESERVATION_REQUEST);

  validateReservationDefinition(reservationId,
      request.getReservationDefinition(), plan,
      AuditConstants.SUBMIT_RESERVATION_REQUEST);
  return plan;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
/**
 * Quick validation on the input to check some obvious fail conditions (fail
 * fast) the input and returns the appropriate {@link Plan} associated with
 * the specified {@link Queue} or throws an exception message illustrating the
 * details of any validation check failures.
 *
 * @param reservationSystem the {@link ReservationSystem} to validate against
 * @param request the {@link ReservationListRequest} defining search
 *                parameters for reservations in the {@link ReservationSystem}
 *                that is being validated against.
 * @return the {@link Plan} to list reservations of.
 * @throws YarnException
 */
public Plan validateReservationListRequest(
    ReservationSystem reservationSystem,
    ReservationListRequest request)
    throws YarnException {
  String queue = request.getQueue();
  if (request.getEndTime() < request.getStartTime()) {
    String errorMessage = "The specified end time must be greater than " +
            "the specified start time.";
    RMAuditLogger.logFailure("UNKNOWN",
            AuditConstants.LIST_RESERVATION_REQUEST,
            "validate list reservation input", "ClientRMService",
            errorMessage);
    throw RPCUtil.getRemoteException(errorMessage);
  }
  // Check if it is a managed queue
  return getPlanFromQueue(reservationSystem, queue,
          AuditConstants.LIST_RESERVATION_REQUEST);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
/**
 * Quick validation on the input to check some obvious fail conditions (fail
 * fast) the input and returns the appropriate {@link Plan} associated with
 * the specified {@link Queue} or throws an exception message illustrating the
 * details of any validation check failures
 * 
 * @param reservationSystem the {@link ReservationSystem} to validate against
 * @param request the {@link ReservationUpdateRequest} defining the resources
 *          required over time for the request
 * @return the {@link Plan} to submit the request to
 * @throws YarnException
 */
public Plan validateReservationUpdateRequest(
    ReservationSystem reservationSystem, ReservationUpdateRequest request)
    throws YarnException {
  ReservationId reservationId = request.getReservationId();
  Plan plan =
      validateReservation(reservationSystem, reservationId,
          AuditConstants.UPDATE_RESERVATION_REQUEST);
  validateReservationDefinition(reservationId,
      request.getReservationDefinition(), plan,
      AuditConstants.UPDATE_RESERVATION_REQUEST);
  return plan;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@Override
public ReservationUpdateResponse updateReservation(
    ReservationUpdateRequest request) throws YarnException, IOException {
  // Check if reservation system is enabled
  checkReservationSytem(AuditConstants.UPDATE_RESERVATION_REQUEST);
  ReservationUpdateResponse response =
      recordFactory.newRecordInstance(ReservationUpdateResponse.class);
  // Validate the input
  Plan plan =
      rValidator.validateReservationUpdateRequest(reservationSystem, request);
  ReservationId reservationId = request.getReservationId();
  String queueName = reservationSystem.getQueueForReservation(reservationId);
  // Check ACLs
  String user =
      checkReservationACLs(queueName,
          AuditConstants.UPDATE_RESERVATION_REQUEST);
  // Try to update the reservation using default agent
  try {
    boolean result =
        plan.getReservationAgent().updateReservation(reservationId, user,
            plan, request.getReservationDefinition());
    if (!result) {
      String errMsg = "Unable to update reservation: " + reservationId;
      RMAuditLogger.logFailure(user,
          AuditConstants.UPDATE_RESERVATION_REQUEST, errMsg,
          "ClientRMService", errMsg);
      throw RPCUtil.getRemoteException(errMsg);
    }
  } catch (PlanningException e) {
    RMAuditLogger.logFailure(user, AuditConstants.UPDATE_RESERVATION_REQUEST,
        e.getMessage(), "ClientRMService",
        "Unable to update the reservation: " + reservationId);
    throw RPCUtil.getRemoteException(e);
  }
  RMAuditLogger.logSuccess(user, AuditConstants.UPDATE_RESERVATION_REQUEST,
      "ClientRMService: " + reservationId);
  return response;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@Override
public ReservationDeleteResponse deleteReservation(
    ReservationDeleteRequest request) throws YarnException, IOException {
  // Check if reservation system is enabled
  checkReservationSytem(AuditConstants.DELETE_RESERVATION_REQUEST);
  ReservationDeleteResponse response =
      recordFactory.newRecordInstance(ReservationDeleteResponse.class);
  // Validate the input
  Plan plan =
      rValidator.validateReservationDeleteRequest(reservationSystem, request);
  ReservationId reservationId = request.getReservationId();
  String queueName = reservationSystem.getQueueForReservation(reservationId);
  // Check ACLs
  String user =
      checkReservationACLs(queueName,
          AuditConstants.DELETE_RESERVATION_REQUEST);
  // Try to update the reservation using default agent
  try {
    boolean result =
        plan.getReservationAgent().deleteReservation(reservationId, user,
            plan);
    if (!result) {
      String errMsg = "Could not delete reservation: " + reservationId;
      RMAuditLogger.logFailure(user,
          AuditConstants.DELETE_RESERVATION_REQUEST, errMsg,
          "ClientRMService", errMsg);
      throw RPCUtil.getRemoteException(errMsg);
    }
  } catch (PlanningException e) {
    RMAuditLogger.logFailure(user, AuditConstants.DELETE_RESERVATION_REQUEST,
        e.getMessage(), "ClientRMService",
        "Unable to delete the reservation: " + reservationId);
    throw RPCUtil.getRemoteException(e);
  }
  RMAuditLogger.logSuccess(user, AuditConstants.DELETE_RESERVATION_REQUEST,
      "ClientRMService: " + reservationId);
  return response;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
synchronized public void containerCompleted(RMContainer rmContainer,
    ContainerStatus containerStatus, RMContainerEventType event) {
  
  Container container = rmContainer.getContainer();
  ContainerId containerId = container.getId();
  
  // Remove from the list of newly allocated containers if found
  newlyAllocatedContainers.remove(rmContainer);
  
  // Inform the container
  rmContainer.handle(
      new RMContainerFinishedEvent(
          containerId,
          containerStatus, 
          event)
      );
  LOG.info("Completed container: " + rmContainer.getContainerId() + 
      " in state: " + rmContainer.getState() + " event:" + event);
  
  // Remove from the list of containers
  liveContainers.remove(rmContainer.getContainerId());

  RMAuditLogger.logSuccess(getUser(), 
      AuditConstants.RELEASE_CONTAINER, "SchedulerApp", 
      getApplicationId(), containerId);
  
  // Update usage metrics 
  Resource containerResource = rmContainer.getContainer().getResource();
  queue.getMetrics().releaseResources(getUser(), 1, containerResource);
  this.attemptResourceUsage.decUsed(containerResource);

  // remove from preemption map if it is completed
  preemptionMap.remove(rmContainer);

  // Clear resource utilization metrics cache.
  lastMemoryAggregateAllocationUpdateTime = -1;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
protected void writeAuditLog(ApplicationId appId) {
  RMApp app = rmContext.getRMApps().get(appId);
  String operation = "UNKONWN";
  boolean success = false;
  switch (app.getState()) {
    case FAILED: 
      operation = AuditConstants.FINISH_FAILED_APP;
      break;
    case FINISHED:
      operation = AuditConstants.FINISH_SUCCESS_APP;
      success = true;
      break;
    case KILLED: 
      operation = AuditConstants.FINISH_KILLED_APP;
      success = true;
      break;
    default:
      break;
  }
  
  if (success) {
    RMAuditLogger.logSuccess(app.getUser(), operation,
        "RMAppManager", app.getApplicationId());
  } else {
    StringBuilder diag = app.getDiagnostics(); 
    String msg = diag == null ? null : diag.toString();
    RMAuditLogger.logFailure(app.getUser(), operation, msg, "RMAppManager",
        "App failed with state: " + app.getState(), appId);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
@GET
@Path("/apps/{appid}/priority")
@Produces({ MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML })
public AppPriority getAppPriority(@Context HttpServletRequest hsr,
    @PathParam("appid") String appId) throws AuthorizationException {
  init();
  UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
  String userName = "UNKNOWN-USER";
  if (callerUGI != null) {
    userName = callerUGI.getUserName();
  }
  RMApp app = null;
  try {
    app = getRMAppForAppId(appId);
  } catch (NotFoundException e) {
    RMAuditLogger.logFailure(userName, AuditConstants.KILL_APP_REQUEST,
        "UNKNOWN", "RMWebService",
        "Trying to get state of an absent application " + appId);
    throw e;
  }

  AppPriority ret = new AppPriority();
  ret.setPriority(
      app.getApplicationSubmissionContext().getPriority().getPriority());

  return ret;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
synchronized public boolean containerCompleted(RMContainer rmContainer,
    ContainerStatus containerStatus, RMContainerEventType event,
    String partition) {

  // Remove from the list of containers
  if (null == liveContainers.remove(rmContainer.getContainerId())) {
    return false;
  }
  
  // Remove from the list of newly allocated containers if found
  newlyAllocatedContainers.remove(rmContainer);

  Container container = rmContainer.getContainer();
  ContainerId containerId = container.getId();

  // Inform the container
  rmContainer.handle(
      new RMContainerFinishedEvent(containerId, containerStatus, event));

  containersToPreempt.remove(rmContainer.getContainerId());

  RMAuditLogger.logSuccess(getUser(),
      AuditConstants.RELEASE_CONTAINER, "SchedulerApp",
      getApplicationId(), containerId);
  
  // Update usage metrics 
  Resource containerResource = rmContainer.getContainer().getResource();
  queue.getMetrics().releaseResources(getUser(), 1, containerResource);
  attemptResourceUsage.decUsed(partition, containerResource);

  // Clear resource utilization metrics cache.
  lastMemoryAggregateAllocationUpdateTime = -1;

  return true;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; //导入依赖的package包/类
synchronized public void containerCompleted(RMContainer rmContainer,
    ContainerStatus containerStatus, RMContainerEventType event) {
  
  Container container = rmContainer.getContainer();
  ContainerId containerId = container.getId();
  
  // Remove from the list of newly allocated containers if found
  newlyAllocatedContainers.remove(rmContainer);
  
  // Inform the container
  rmContainer.handle(
      new RMContainerFinishedEvent(
          containerId,
          containerStatus, 
          event)
      );
  LOG.info("Completed container: " + rmContainer.getContainerId() + 
      " in state: " + rmContainer.getState() + " event:" + event);
  
  // Remove from the list of containers
  liveContainers.remove(rmContainer.getContainerId());

  RMAuditLogger.logSuccess(getUser(), 
      AuditConstants.RELEASE_CONTAINER, "SchedulerApp", 
      getApplicationId(), containerId);
  
  // Update usage metrics 
  Resource containerResource = rmContainer.getContainer().getResource();
  queue.getMetrics().releaseResources(getUser(), 1, containerResource);
  Resources.subtractFrom(currentConsumption, containerResource);

  // remove from preemption map if it is completed
  preemptionMap.remove(rmContainer);

  // Clear resource utilization metrics cache.
  lastMemoryAggregateAllocationUpdateTime = -1;
}