YARN-9590. Correct incompatible, incomplete and redundant activities. Contributed by Tao Yang.

Author: yangwwei

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/activities/ActivitiesLogger.java

@@ -102,20 +102,20 @@ public static void recordAppActivityWithoutAllocation(
           // Add application-container activity into specific node allocation.
           activitiesManager.addSchedulingActivityForNode(nodeId,
               requestName, null,
-              priorityStr, ActivityState.SKIPPED, diagnostic, type,
+              priorityStr, appState, diagnostic, type,
               null);
           type = "request";
           // Add application-container activity into specific node allocation.
           activitiesManager.addSchedulingActivityForNode(nodeId,
               application.getApplicationId().toString(), requestName,
-              priorityStr, ActivityState.SKIPPED,
+              priorityStr, appState,
               ActivityDiagnosticConstant.EMPTY, type, allocationRequestId);
         }
         // Add queue-application activity into specific node allocation.
         activitiesManager.addSchedulingActivityForNode(nodeId,
             application.getQueueName(),
             application.getApplicationId().toString(),
-            application.getPriority().toString(), ActivityState.SKIPPED,
+            application.getPriority().toString(), appState,
             schedulerKey != null ? ActivityDiagnosticConstant.EMPTY :
                 diagnostic, "app", null);
       }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/activities/ActivitiesManager.java

@@ -339,8 +339,10 @@ void finishAppAllocationRecording(ApplicationId applicationId,
           appAllocations = curAppAllocations;
         }
       }
-      if (appAllocations.size() == appActivitiesMaxQueueLength) {
+      int curQueueLength = appAllocations.size();
+      while (curQueueLength >= appActivitiesMaxQueueLength) {
         appAllocations.poll();
+        --curQueueLength;
       }
       appAllocations.add(appAllocation);
       Long stopTime =

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java

@@ -1597,7 +1597,7 @@ private void allocateFromReservedContainer(FiCaSchedulerNode node,
     } else{
       ActivitiesLogger.QUEUE.recordQueueActivity(activitiesManager, node,
           queue.getParent().getQueueName(), queue.getQueueName(),
-          ActivityState.ACCEPTED, ActivityDiagnosticConstant.EMPTY);
+          ActivityState.SKIPPED, ActivityDiagnosticConstant.EMPTY);
       ActivitiesLogger.NODE.finishAllocatedNodeAllocation(activitiesManager,
           node, reservedContainer.getContainerId(), AllocationState.SKIPPED);
     }
@@ -1687,6 +1687,10 @@ private CSAssignment allocateContainersOnMultiNodes(
       }
       LOG.debug("This node or this node partition doesn't have available or "
           + "killable resource");
+      ActivitiesLogger.QUEUE.recordQueueActivity(activitiesManager, null,
+          "", getRootQueue().getQueueName(), ActivityState.REJECTED,
+          ActivityDiagnosticConstant.NOT_ABLE_TO_ACCESS_PARTITION + " "
+              + candidates.getPartition());
       return null;
     }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java

@@ -1188,6 +1188,9 @@ public CSAssignment assignContainers(Resource clusterResource,
         application.updateNodeInfoForAMDiagnostics(node);
       } else if (assignment.getSkippedType()
           == CSAssignment.SkippedType.QUEUE_LIMIT) {
+        ActivitiesLogger.QUEUE.recordQueueActivity(activitiesManager, node,
+            getParent().getQueueName(), getQueueName(), ActivityState.SKIPPED,
+            ActivityDiagnosticConstant.QUEUE_SKIPPED_HEADROOM);
         return assignment;
       } else{
         // If we don't allocate anything, and it is not skipped by application,

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/AbstractContainerAllocator.java

@@ -109,16 +109,10 @@ protected CSAssignment getCSAssignmentFromAllocateResult(
             allocatedResource);
 
         if (rmContainer != null) {
-          ActivitiesLogger.APP.recordAppActivityWithAllocation(
-              activitiesManager, node, application, updatedContainer,
-              ActivityState.RE_RESERVED);
           ActivitiesLogger.APP.finishSkippedAppAllocationRecording(
               activitiesManager, application.getApplicationId(),
               ActivityState.SKIPPED, ActivityDiagnosticConstant.EMPTY);
         } else {
-          ActivitiesLogger.APP.recordAppActivityWithAllocation(
-              activitiesManager, node, application, updatedContainer,
-              ActivityState.RESERVED);
           ActivitiesLogger.APP.finishAllocatedAppAllocationRecording(
               activitiesManager, application.getApplicationId(),
               updatedContainer.getContainerId(), ActivityState.RESERVED,
@@ -149,7 +143,7 @@ protected CSAssignment getCSAssignmentFromAllocateResult(
             node, application, updatedContainer, ActivityState.ALLOCATED);
         ActivitiesLogger.APP.finishAllocatedAppAllocationRecording(
             activitiesManager, application.getApplicationId(),
-            updatedContainer.getContainerId(), ActivityState.ACCEPTED,
+            updatedContainer.getContainerId(), ActivityState.ALLOCATED,
             ActivityDiagnosticConstant.EMPTY);
 
         // Update unformed resource
@@ -162,6 +156,9 @@ protected CSAssignment getCSAssignmentFromAllocateResult(
         assignment.setSkippedType(
             CSAssignment.SkippedType.QUEUE_LIMIT);
       }
+      ActivitiesLogger.APP.finishSkippedAppAllocationRecording(
+          activitiesManager, application.getApplicationId(),
+          ActivityState.SKIPPED, ActivityDiagnosticConstant.EMPTY);
     }
 
     return assignment;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java

@@ -628,6 +628,12 @@ private ContainerAllocation assignContainer(Resource clusterResource,
           }
         }
 
+        ActivitiesLogger.APP.recordAppActivityWithoutAllocation(
+            activitiesManager, node, application, schedulerKey,
+            ActivityDiagnosticConstant.NOT_SUFFICIENT_RESOURCE
+                + getResourceDiagnostics(capability, availableForDC),
+            rmContainer == null ?
+                ActivityState.RESERVED : ActivityState.RE_RESERVED);
         ContainerAllocation result = new ContainerAllocation(null,
             pendingAsk.getPerAllocationResource(), AllocationState.RESERVED);
         result.containerNodeType = type;
@@ -824,7 +830,7 @@ private ContainerAllocation allocate(Resource clusterResource,
           ActivityDiagnosticConstant.
               APPLICATION_PRIORITY_DO_NOT_NEED_RESOURCE);
       return new ContainerAllocation(reservedContainer, null,
-          AllocationState.QUEUE_SKIPPED);
+          AllocationState.PRIORITY_SKIPPED);
     }
 
     result = ContainerAllocation.PRIORITY_SKIPPED;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/AppAllocationInfo.java

@@ -43,6 +43,7 @@ public class AppAllocationInfo {
   private long timestamp;
   private String dateTime;
   private String allocationState;
+  private String diagnostic;
   private List<AppRequestAllocationInfo> requestAllocation;
 
   AppAllocationInfo() {
@@ -57,6 +58,7 @@ public class AppAllocationInfo {
     this.timestamp = allocation.getTime();
     this.dateTime = new Date(allocation.getTime()).toString();
     this.allocationState = allocation.getAppState().name();
+    this.diagnostic = allocation.getDiagnostic();
     Map<String, List<ActivityNode>> requestToActivityNodes =
         allocation.getAllocationAttempts().stream().collect(Collectors
             .groupingBy((e) -> e.getRequestPriority() + "_" + e
@@ -96,4 +98,8 @@ public String getAllocationState() {
   public List<AppRequestAllocationInfo> getRequestAllocation() {
     return requestAllocation;
   }
+
+  public String getDiagnostic() {
+    return diagnostic;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesSchedulerActivities.java

@@ -460,7 +460,7 @@ public void testAppActivityJSON() throws Exception {
       //Check app activities
       verifyNumberOfAllocations(json, 1);
       JSONObject allocations = json.getJSONObject("allocations");
-      verifyStateOfAllocations(allocations, "allocationState", "ACCEPTED");
+      verifyStateOfAllocations(allocations, "allocationState", "ALLOCATED");
       //Check request allocation
       JSONObject requestAllocationObj =
           allocations.getJSONObject("requestAllocation");
@@ -527,7 +527,7 @@ public void testAppAssignMultipleContainersPerNodeHeartbeat()
       JSONArray allocations = json.getJSONArray("allocations");
       for (int i = 0; i < allocations.length(); i++) {
         verifyStateOfAllocations(allocations.getJSONObject(i),
-            "allocationState", "ACCEPTED");
+            "allocationState", "ALLOCATED");
       }
     } finally {
       rm.stop();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesSchedulerActivitiesWithMultiNodesEnabled.java

@@ -277,7 +277,7 @@ public void testAppAssignContainer() throws Exception {
       verifyNumberOfAllocations(json, 1);
 
       JSONObject allocationObj = json.getJSONObject("allocations");
-      verifyStateOfAllocations(allocationObj, "allocationState", "ACCEPTED");
+      verifyStateOfAllocations(allocationObj, "allocationState", "ALLOCATED");
       JSONObject requestAllocationObj =
           allocationObj.getJSONObject("requestAllocation");
       verifyNumberOfAllocationAttempts(requestAllocationObj, 2);
@@ -437,7 +437,7 @@ public void testAppInsufficientResourceDiagnostic() throws Exception {
       }
       // check second activity is for first allocation with ALLOCATED state
       allocationObj = allocationArray.getJSONObject(1);
-      verifyStateOfAllocations(allocationObj, "allocationState", "ACCEPTED");
+      verifyStateOfAllocations(allocationObj, "allocationState", "ALLOCATED");
       requestAllocationObj = allocationObj.getJSONObject("requestAllocation");
       verifyNumberOfAllocationAttempts(requestAllocationObj, 1);
       verifyStateOfAllocations(requestAllocationObj, "allocationState",
@@ -610,7 +610,7 @@ public void testAppGroupByDiagnostics() throws Exception {
       }
       // check second activity is for first allocation with ALLOCATED state
       allocationObj = allocationArray.getJSONObject(1);
-      verifyStateOfAllocations(allocationObj, "allocationState", "ACCEPTED");
+      verifyStateOfAllocations(allocationObj, "allocationState", "ALLOCATED");
       requestAllocationObj = allocationObj.getJSONObject("requestAllocation");
       verifyNumberOfAllocationAttempts(requestAllocationObj, 1);
       verifyStateOfAllocations(requestAllocationObj, "allocationState",