SchedulerApplicationAttempt implements SchedulableEntity { public ApplicationResourceUsageReport getResourceUsageReport() { try { writeLock.lock(); AggregateAppResourceUsage runningResourceUsage = getRunningAggregateAppResourceUsage(); Resource usedResourceClone = Resources.clone( attemptResourceUsage.getAllUsed()); Resource reservedResourceClone = Resources.clone( attemptResourceUsage.getReserved()); Resource cluster = rmContext.getScheduler().getClusterResource(); ResourceCalculator calc = rmContext.getScheduler().getResourceCalculator(); Map<String, Long> preemptedResourceSecondsMaps = new HashMap<>(); preemptedResourceSecondsMaps .put(ResourceInformation.MEMORY_MB.getName(), 0L); preemptedResourceSecondsMaps .put(ResourceInformation.VCORES.getName(), 0L); float queueUsagePerc = 0.0f; float clusterUsagePerc = 0.0f; if (!calc.isInvalidDivisor(cluster)) { float queueCapacityPerc = queue.getQueueInfo(false, false) .getCapacity(); if (queueCapacityPerc != 0) { queueUsagePerc = calc.divide(cluster, usedResourceClone, Resources.multiply(cluster, queueCapacityPerc)) * 100; } clusterUsagePerc = calc.divide(cluster, usedResourceClone, cluster) * 100; } return ApplicationResourceUsageReport .newInstance(liveContainers.size(), reservedContainers.size(), usedResourceClone, reservedResourceClone, Resources.add(usedResourceClone, reservedResourceClone), runningResourceUsage.getResourceUsageSecondsMap(), queueUsagePerc, clusterUsagePerc, preemptedResourceSecondsMaps); } finally { writeLock.unlock(); } } SchedulerApplicationAttempt(ApplicationAttemptId applicationAttemptId, 
      String user, Queue queue, AbstractUsersManager abstractUsersManager,
      RMContext rmContext); void setOpportunisticContainerContext(
      OpportunisticContainerContext oppContext); OpportunisticContainerContext getOpportunisticContainerContext(); Collection<RMContainer> getLiveContainers(); AppSchedulingInfo getAppSchedulingInfo(); ContainerUpdateContext getUpdateContext(); boolean isPending(); ApplicationAttemptId getApplicationAttemptId(); ApplicationId getApplicationId(); String getUser(); Set<ContainerId> getPendingRelease(); long getNewContainerId(); Collection<SchedulerRequestKey> getSchedulerKeys(); PendingAsk getPendingAsk(
      SchedulerRequestKey schedulerKey, String resourceName); int getOutstandingAsksCount(SchedulerRequestKey schedulerKey); int getOutstandingAsksCount(SchedulerRequestKey schedulerKey,
      String resourceName); String getQueueName(); Resource getAMResource(); Resource getAMResource(String label); void setAMResource(Resource amResource); void setAMResource(String label, Resource amResource); boolean isAmRunning(); void setAmRunning(boolean bool); boolean getUnmanagedAM(); RMContainer getRMContainer(ContainerId id); void addRMContainer(
      ContainerId id, RMContainer rmContainer); void removeRMContainer(ContainerId containerId); int getReReservations(SchedulerRequestKey schedulerKey); @Stable @Private Resource getCurrentReservation(); Queue getQueue(); boolean updateResourceRequests(
      List<ResourceRequest> requests); boolean updateSchedulingRequests(
      List<SchedulingRequest> requests); void recoverResourceRequestsForContainer(
      ContainerRequest containerRequest); void stop(RMAppAttemptState rmAppAttemptFinalState); boolean isStopped(); List<RMContainer> getReservedContainers(); boolean reserveIncreasedContainer(SchedulerNode node,
      SchedulerRequestKey schedulerKey, RMContainer rmContainer,
      Resource reservedResource); RMContainer reserve(SchedulerNode node,
      SchedulerRequestKey schedulerKey, RMContainer rmContainer,
      Container container); void setHeadroom(Resource globalLimit); Resource getHeadroom(); int getNumReservedContainers(
      SchedulerRequestKey schedulerKey); @SuppressWarnings("unchecked") void containerLaunchedOnNode(ContainerId containerId,
      NodeId nodeId); void showRequests(); Resource getCurrentConsumption(); void updateNMTokens(Collection<Container> containers); List<Container> pullPreviousAttemptContainers(); List<Container> pullNewlyAllocatedContainers(); synchronized void addToNewlyDemotedContainers(ContainerId containerId,
      RMContainer rmContainer); synchronized void addToNewlyDecreasedContainers(
      ContainerId containerId, RMContainer rmContainer); List<Container> pullNewlyPromotedContainers(); List<Container> pullNewlyDemotedContainers(); List<Container> pullNewlyIncreasedContainers(); List<Container> pullNewlyDecreasedContainers(); List<UpdateContainerError> pullUpdateContainerErrors(); List<NMToken> pullUpdatedNMTokens(); boolean isWaitingForAMContainer(); void updateBlacklist(List<String> blacklistAdditions,
      List<String> blacklistRemovals); boolean isPlaceBlacklisted(String resourceName); int addMissedNonPartitionedRequestSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); void resetMissedNonPartitionedRequestSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); void addSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); void subtractSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); int getSchedulingOpportunities(
      SchedulerRequestKey schedulerKey); void resetSchedulingOpportunities(
      SchedulerRequestKey schedulerKey); void resetSchedulingOpportunities(SchedulerRequestKey schedulerKey,
      long currentTimeMs); ApplicationResourceUsageReport getResourceUsageReport(); @VisibleForTesting Map<ContainerId, RMContainer> getLiveContainersMap(); Map<SchedulerRequestKey, Long> getLastScheduledContainer(); void transferStateFromPreviousAttempt(
      SchedulerApplicationAttempt appAttempt); void move(Queue newQueue); void recoverContainer(SchedulerNode node,
      RMContainer rmContainer); void incNumAllocatedContainers(NodeType containerType,
      NodeType requestType); void setApplicationHeadroomForMetrics(Resource headroom); void recordContainerRequestTime(long value); void recordContainerAllocationTime(long value); Set<String> getBlacklistedNodes(); @Private boolean hasPendingResourceRequest(String nodePartition,
      SchedulingMode schedulingMode); @VisibleForTesting ResourceUsage getAppAttemptResourceUsage(); @Override Priority getPriority(); void setPriority(Priority appPriority); @Override String getId(); @Override int compareInputOrderTo(SchedulableEntity other); @Override ResourceUsage getSchedulingResourceUsage(); void setAppAMNodePartitionName(String partitionName); String getAppAMNodePartitionName(); void updateAMContainerDiagnostics(AMState state,
      String diagnosticMessage); ReentrantReadWriteLock.WriteLock getWriteLock(); @Override boolean isRecovering(); AppPlacementAllocator<N> getAppPlacementAllocator(
      SchedulerRequestKey schedulerRequestKey); void incUnconfirmedRes(Resource res); void decUnconfirmedRes(Resource res); @Override int hashCode(); @Override boolean equals(Object o); Map<String, String> getApplicationSchedulingEnvs();  }
@Test public void testAppPercentages() throws Exception { FifoScheduler scheduler = mock(FifoScheduler.class); when(scheduler.getClusterResource()) .thenReturn(Resource.newInstance(10 * 1024, 10)); when(scheduler.getResourceCalculator()) .thenReturn(new DefaultResourceCalculator()); ApplicationAttemptId appAttId = createAppAttemptId(0, 0); RMContext rmContext = mock(RMContext.class); when(rmContext.getEpoch()).thenReturn(3L); when(rmContext.getScheduler()).thenReturn(scheduler); final String user = "user1"; Queue queue = createQueue("test", null); SchedulerApplicationAttempt app = new SchedulerApplicationAttempt(appAttId, user, queue, queue.getAbstractUsersManager(), rmContext); Resource requestedResource = Resource.newInstance(1536, 2); app.attemptResourceUsage.incUsed(requestedResource); assertEquals(15.0f, app.getResourceUsageReport().getQueueUsagePercentage(), 0.01f); assertEquals(15.0f, app.getResourceUsageReport().getClusterUsagePercentage(), 0.01f); queue = createQueue("test2", null, 0.5f); app = new SchedulerApplicationAttempt(appAttId, user, queue, queue.getAbstractUsersManager(), rmContext); app.attemptResourceUsage.incUsed(requestedResource); assertEquals(30.0f, app.getResourceUsageReport().getQueueUsagePercentage(), 0.01f); assertEquals(15.0f, app.getResourceUsageReport().getClusterUsagePercentage(), 0.01f); app.attemptResourceUsage.incUsed(requestedResource); app.attemptResourceUsage.incUsed(requestedResource); app.attemptResourceUsage.incUsed(requestedResource); assertEquals(120.0f, app.getResourceUsageReport().getQueueUsagePercentage(), 0.01f); assertEquals(60.0f, app.getResourceUsageReport().getClusterUsagePercentage(), 0.01f); queue = createQueue("test3", null, 0.0f); app = new SchedulerApplicationAttempt(appAttId, user, queue, queue.getAbstractUsersManager(), rmContext); app.attemptResourceUsage.incUsed(requestedResource); assertEquals(0.0f, app.getResourceUsageReport().getQueueUsagePercentage(), 0.01f); assertEquals(15.0f, app.getResourceUsageReport().getClusterUsagePercentage(), 0.01f); }
@Test public void testAppPercentagesOnswitch() throws Exception { FifoScheduler scheduler = mock(FifoScheduler.class); when(scheduler.getClusterResource()).thenReturn(Resource.newInstance(0, 0)); when(scheduler.getResourceCalculator()) .thenReturn(new DefaultResourceCalculator()); ApplicationAttemptId appAttId = createAppAttemptId(0, 0); RMContext rmContext = mock(RMContext.class); when(rmContext.getEpoch()).thenReturn(3L); when(rmContext.getScheduler()).thenReturn(scheduler); final String user = "user1"; Queue queue = createQueue("test", null); SchedulerApplicationAttempt app = new SchedulerApplicationAttempt(appAttId, user, queue, queue.getAbstractUsersManager(), rmContext); Resource requestedResource = Resource.newInstance(1536, 2); app.attemptResourceUsage.incUsed(requestedResource); assertEquals(0.0f, app.getResourceUsageReport().getQueueUsagePercentage(), 0.0f); assertEquals(0.0f, app.getResourceUsageReport().getClusterUsagePercentage(), 0.0f); }