YARN-9664. Improve response of scheduler/app activities for better understanding. 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

@@ -30,10 +30,11 @@
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
 import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
+import java.util.function.Supplier;
+
 /**
  * Utility for logging scheduler activities
  */
-// FIXME: make sure CandidateNodeSet works with this class
 public class ActivitiesLogger {
   private static final Logger LOG =
       LoggerFactory.getLogger(ActivitiesLogger.class);
@@ -52,9 +53,9 @@ public static void recordSkippedAppActivityWithoutAllocation(
         ActivitiesManager activitiesManager, SchedulerNode node,
         SchedulerApplicationAttempt application,
         SchedulerRequestKey requestKey,
-        String diagnostic) {
+        String diagnostic, ActivityLevel level) {
       recordAppActivityWithoutAllocation(activitiesManager, node, application,
-          requestKey, diagnostic, ActivityState.SKIPPED);
+          requestKey, diagnostic, ActivityState.SKIPPED, level);
     }
 
     /*
@@ -72,7 +73,7 @@ public static void recordRejectedAppActivityFromLeafQueue(
       if (activitiesManager.shouldRecordThisNode(nodeId)) {
         recordActivity(activitiesManager, nodeId, application.getQueueName(),
             application.getApplicationId().toString(), priority,
-            ActivityState.REJECTED, diagnostic, "app");
+            ActivityState.REJECTED, diagnostic, ActivityLevel.APP);
       }
       finishSkippedAppAllocationRecording(activitiesManager,
           application.getApplicationId(), ActivityState.REJECTED, diagnostic);
@@ -87,50 +88,55 @@ public static void recordAppActivityWithoutAllocation(
         ActivitiesManager activitiesManager, SchedulerNode node,
         SchedulerApplicationAttempt application,
         SchedulerRequestKey schedulerKey,
-        String diagnostic, ActivityState appState) {
+        String diagnostic, ActivityState appState, ActivityLevel level) {
       if (activitiesManager == null) {
         return;
       }
       NodeId nodeId = getRecordingNodeId(activitiesManager, node);
       if (activitiesManager.shouldRecordThisNode(nodeId)) {
-        if (schedulerKey != null) {
-          String allocationRequestId =
-              String.valueOf(schedulerKey.getAllocationRequestId());
-          String priorityStr = getPriorityStr(schedulerKey);
-          String requestName = getRequestName(priorityStr, allocationRequestId);
-          String type = "container";
-          // Add application-container activity into specific node allocation.
-          activitiesManager.addSchedulingActivityForNode(nodeId,
-              requestName, null,
-              priorityStr, appState, diagnostic, type,
-              null);
-          type = "request";
-          // Add application-container activity into specific node allocation.
-          activitiesManager.addSchedulingActivityForNode(nodeId,
-              application.getApplicationId().toString(), requestName,
-              priorityStr, appState,
-              ActivityDiagnosticConstant.EMPTY, type, allocationRequestId);
+        String requestName = null;
+        Integer priority = null;
+        Long allocationRequestId = null;
+        if (level == ActivityLevel.NODE || level == ActivityLevel.REQUEST) {
+          if (schedulerKey == null) {
+            LOG.warn("Request key should not be null at " + level + " level.");
+            return;
+          }
+          priority = getPriority(schedulerKey);
+          allocationRequestId = schedulerKey.getAllocationRequestId();
+          requestName = getRequestName(priority, allocationRequestId);
+        }
+        switch (level) {
+        case NODE:
+          recordSchedulerActivityAtNodeLevel(activitiesManager, application,
+              requestName, priority, allocationRequestId, null, nodeId,
+              appState, diagnostic);
+          break;
+        case REQUEST:
+          recordSchedulerActivityAtRequestLevel(activitiesManager, application,
+              requestName, priority, allocationRequestId, nodeId, appState,
+              diagnostic);
+          break;
+        case APP:
+          recordSchedulerActivityAtAppLevel(activitiesManager, application,
+              nodeId, appState, diagnostic);
+          break;
+        default:
+          LOG.warn("Doesn't handle app activities at " + level + " level.");
+          break;
         }
-        // Add queue-application activity into specific node allocation.
-        activitiesManager.addSchedulingActivityForNode(nodeId,
-            application.getQueueName(),
-            application.getApplicationId().toString(),
-            application.getPriority().toString(), appState,
-            schedulerKey != null ? ActivityDiagnosticConstant.EMPTY :
-                diagnostic, "app", null);
       }
       // Add application-container activity into specific application allocation
       // Under this condition, it fails to allocate a container to this
       // application, so containerId is null.
       if (activitiesManager.shouldRecordThisApp(
           application.getApplicationId())) {
-        String type = "container";
         activitiesManager.addSchedulingActivityForApp(
             application.getApplicationId(), null,
-            getPriorityStr(schedulerKey), appState,
-            diagnostic, type, nodeId,
+            getPriority(schedulerKey), appState,
+            diagnostic, level, nodeId,
             schedulerKey == null ?
-                null : String.valueOf(schedulerKey.getAllocationRequestId()));
+                null : schedulerKey.getAllocationRequestId());
       }
     }
 
@@ -150,49 +156,68 @@ public static void recordAppActivityWithAllocation(
         nodeId = updatedContainer.getNodeId();
       }
       if (activitiesManager.shouldRecordThisNode(nodeId)) {
-        String containerPriorityStr =
-            updatedContainer.getContainer().getPriority().toString();
-        String allocationRequestId = String
-            .valueOf(updatedContainer.getContainer().getAllocationRequestId());
+        Integer containerPriority =
+            updatedContainer.getContainer().getPriority().getPriority();
+        Long allocationRequestId =
+            updatedContainer.getContainer().getAllocationRequestId();
         String requestName =
-            getRequestName(containerPriorityStr, allocationRequestId);
-        String type = "container";
-
-        // Add application-container activity into specific node allocation.
-        activitiesManager.addSchedulingActivityForNode(nodeId,
-            requestName,
-            updatedContainer.getContainer().toString(),
-            containerPriorityStr,
-            activityState, ActivityDiagnosticConstant.EMPTY, type, null);
-        type = "request";
-        // Add application-container activity into specific node allocation.
-        activitiesManager.addSchedulingActivityForNode(nodeId,
-            application.getApplicationId().toString(),
-            requestName, containerPriorityStr,
-            activityState, ActivityDiagnosticConstant.EMPTY, type,
-            allocationRequestId);
-        type = "app";
-        // Add queue-application activity into specific node allocation.
-        activitiesManager.addSchedulingActivityForNode(nodeId,
-            application.getQueueName(),
-            application.getApplicationId().toString(),
-            application.getPriority().toString(), ActivityState.ACCEPTED,
-            ActivityDiagnosticConstant.EMPTY, type, null);
+            getRequestName(containerPriority, allocationRequestId);
+        // Add node,request,app level activities into scheduler activities.
+        recordSchedulerActivityAtNodeLevel(activitiesManager, application,
+            requestName, containerPriority, allocationRequestId,
+            updatedContainer.getContainer().toString(), nodeId, activityState,
+            ActivityDiagnosticConstant.EMPTY);
       }
       // Add application-container activity into specific application allocation
       if (activitiesManager.shouldRecordThisApp(
           application.getApplicationId())) {
-        String type = "container";
         activitiesManager.addSchedulingActivityForApp(
             application.getApplicationId(),
             updatedContainer.getContainerId(),
-            updatedContainer.getContainer().getPriority().toString(),
-            activityState, ActivityDiagnosticConstant.EMPTY, type, nodeId,
-            String.valueOf(
-                updatedContainer.getContainer().getAllocationRequestId()));
+            updatedContainer.getContainer().getPriority().getPriority(),
+            activityState, ActivityDiagnosticConstant.EMPTY,
+            ActivityLevel.NODE, nodeId,
+            updatedContainer.getContainer().getAllocationRequestId());
       }
     }
 
+    @SuppressWarnings("parameternumber")
+    private static void recordSchedulerActivityAtNodeLevel(
+        ActivitiesManager activitiesManager, SchedulerApplicationAttempt app,
+        String requestName, Integer priority, Long allocationRequestId,
+        String containerId, NodeId nodeId, ActivityState state,
+        String diagnostic) {
+      activitiesManager
+          .addSchedulingActivityForNode(nodeId, requestName, containerId, null,
+              state, diagnostic, ActivityLevel.NODE, null);
+      // Record request level activity additionally.
+      recordSchedulerActivityAtRequestLevel(activitiesManager, app, requestName,
+          priority, allocationRequestId, nodeId, state,
+          ActivityDiagnosticConstant.EMPTY);
+    }
+
+    @SuppressWarnings("parameternumber")
+    private static void recordSchedulerActivityAtRequestLevel(
+        ActivitiesManager activitiesManager, SchedulerApplicationAttempt app,
+        String requestName, Integer priority, Long allocationRequestId,
+        NodeId nodeId, ActivityState state, String diagnostic) {
+      activitiesManager.addSchedulingActivityForNode(nodeId,
+          app.getApplicationId().toString(), requestName, priority,
+          state, diagnostic, ActivityLevel.REQUEST,
+          allocationRequestId);
+      // Record app level activity additionally.
+      recordSchedulerActivityAtAppLevel(activitiesManager, app, nodeId, state,
+          ActivityDiagnosticConstant.EMPTY);
+    }
+
+    private static void recordSchedulerActivityAtAppLevel(
+        ActivitiesManager activitiesManager, SchedulerApplicationAttempt app,
+        NodeId nodeId, ActivityState state, String diagnostic) {
+      activitiesManager.addSchedulingActivityForNode(nodeId, app.getQueueName(),
+          app.getApplicationId().toString(), app.getPriority().getPriority(),
+          state, diagnostic, ActivityLevel.APP, null);
+    }
+
     /*
      * Invoked when scheduler starts to look at this application within one node
      * update.
@@ -252,13 +277,20 @@ public static class QUEUE {
     public static void recordQueueActivity(ActivitiesManager activitiesManager,
         SchedulerNode node, String parentQueueName, String queueName,
         ActivityState state, String diagnostic) {
+      recordQueueActivity(activitiesManager, node, parentQueueName, queueName,
+          state, () -> diagnostic);
+    }
+
+    public static void recordQueueActivity(ActivitiesManager activitiesManager,
+        SchedulerNode node, String parentQueueName, String queueName,
+        ActivityState state, Supplier<String> diagnosticSupplier) {
       if (activitiesManager == null) {
         return;
       }
       NodeId nodeId = getRecordingNodeId(activitiesManager, node);
       if (activitiesManager.shouldRecordThisNode(nodeId)) {
         recordActivity(activitiesManager, nodeId, parentQueueName, queueName,
-            null, state, diagnostic, null);
+            null, state, diagnosticSupplier.get(), ActivityLevel.QUEUE);
       }
     }
   }
@@ -299,11 +331,11 @@ public static void finishAllocatedNodeAllocation(
      * Invoked when node heartbeat finishes
      */
     public static void finishNodeUpdateRecording(
-        ActivitiesManager activitiesManager, NodeId nodeID) {
+        ActivitiesManager activitiesManager, NodeId nodeID, String partition) {
       if (activitiesManager == null) {
         return;
       }
-      activitiesManager.finishNodeUpdateRecording(nodeID);
+      activitiesManager.finishNodeUpdateRecording(nodeID, partition);
     }
 
     /*
@@ -320,11 +352,11 @@ public static void startNodeUpdateRecording(
 
   // Add queue, application or container activity into specific node allocation.
   private static void recordActivity(ActivitiesManager activitiesManager,
-      NodeId nodeId, String parentName, String childName,
-      Priority priority, ActivityState state, String diagnostic, String type) {
+      NodeId nodeId, String parentName, String childName, Priority priority,
+      ActivityState state, String diagnostic, ActivityLevel level) {
     activitiesManager.addSchedulingActivityForNode(nodeId, parentName,
-        childName, priority != null ? priority.toString() : null, state,
-        diagnostic, type, null);
+        childName, priority != null ? priority.getPriority() : null, state,
+        diagnostic, level, null);
   }
 
   private static NodeId getRecordingNodeId(ActivitiesManager activitiesManager,
@@ -333,16 +365,16 @@ private static NodeId getRecordingNodeId(ActivitiesManager activitiesManager,
         activitiesManager.getRecordingNodeId(node);
   }
 
-  private static String getRequestName(String priority,
-      String allocationRequestId) {
+  private static String getRequestName(Integer priority,
+      Long allocationRequestId) {
     return "request_"
         + (priority == null ? "" : priority)
         + "_" + (allocationRequestId == null ? "" : allocationRequestId);
   }
 
-  private static String getPriorityStr(SchedulerRequestKey schedulerKey) {
+  private static Integer getPriority(SchedulerRequestKey schedulerKey) {
     Priority priority = schedulerKey == null ?
         null : schedulerKey.getPriority();
-    return priority == null ? null : priority.toString();
+    return priority == null ? null : priority.getPriority();
   }
 }

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

@@ -124,7 +124,7 @@ private void setupConfForCleanup(Configuration conf) {
   }
 
   public AppActivitiesInfo getAppActivitiesInfo(ApplicationId applicationId,
-      Set<String> requestPriorities, Set<String> allocationRequestIds,
+      Set<Integer> requestPriorities, Set<Long> allocationRequestIds,
       RMWSConsts.ActivitiesGroupBy groupBy, int limit, boolean summarize,
       double maxTimeInSeconds) {
     RMApp app = rmContext.getRMApps().get(applicationId);
@@ -186,20 +186,18 @@ private AppAllocation getSummarizedAppAllocation(
       }
       List<ActivityNode> activityNodes = appAllocation.getAllocationAttempts();
       for (ActivityNode an : activityNodes) {
-        if (an.getNodeId() != null) {
-          nodeActivities.putIfAbsent(
-              an.getRequestPriority() + "_" + an.getAllocationRequestId() + "_"
-                  + an.getNodeId(), an);
-        }
+        nodeActivities.putIfAbsent(
+            an.getRequestPriority() + "_" + an.getAllocationRequestId() + "_"
+                + an.getNodeId(), an);
       }
     }
     AppAllocation lastAppAllocation = allocations.get(allocations.size() - 1);
     AppAllocation summarizedAppAllocation =
         new AppAllocation(lastAppAllocation.getPriority(), null,
             lastAppAllocation.getQueueName());
-    summarizedAppAllocation
-        .updateAppContainerStateAndTime(null, lastAppAllocation.getAppState(),
-            lastAppAllocation.getTime(), lastAppAllocation.getDiagnostic());
+    summarizedAppAllocation.updateAppContainerStateAndTime(null,
+        lastAppAllocation.getActivityState(), lastAppAllocation.getTime(),
+        lastAppAllocation.getDiagnostic());
     summarizedAppAllocation
         .setAllocationAttempts(new ArrayList<>(nodeActivities.values()));
     return summarizedAppAllocation;
@@ -282,7 +280,7 @@ public void run() {
             Map.Entry<NodeId, List<NodeAllocation>> nodeAllocation = ite.next();
             List<NodeAllocation> allocations = nodeAllocation.getValue();
             if (allocations.size() > 0
-                && curTS - allocations.get(0).getTimeStamp()
+                && curTS - allocations.get(0).getTimestamp()
                 > schedulerActivitiesTTL) {
               ite.remove();
             }
@@ -383,26 +381,26 @@ void startAppAllocationRecording(NodeId nodeID, long currTS,
 
   // Add queue, application or container activity into specific node allocation.
   void addSchedulingActivityForNode(NodeId nodeId, String parentName,
-      String childName, String priority, ActivityState state, String diagnostic,
-      String type, String allocationRequestId) {
+      String childName, Integer priority, ActivityState state,
+      String diagnostic, ActivityLevel level, Long allocationRequestId) {
     if (shouldRecordThisNode(nodeId)) {
       NodeAllocation nodeAllocation = getCurrentNodeAllocation(nodeId);
       nodeAllocation.addAllocationActivity(parentName, childName, priority,
-          state, diagnostic, type, nodeId, allocationRequestId);
+          state, diagnostic, level, nodeId, allocationRequestId);
     }
   }
 
   // Add queue, application or container activity into specific application
   // allocation.
   void addSchedulingActivityForApp(ApplicationId applicationId,
-      ContainerId containerId, String priority, ActivityState state,
-      String diagnostic, String type, NodeId nodeId,
-      String allocationRequestId) {
+      ContainerId containerId, Integer priority, ActivityState state,
+      String diagnostic, ActivityLevel level, NodeId nodeId,
+      Long allocationRequestId) {
     if (shouldRecordThisApp(applicationId)) {
       AppAllocation appAllocation = appsAllocation.get().get(applicationId);
       appAllocation.addAppAllocationActivity(containerId == null ?
           "Container-Id-Not-Assigned" :
-          containerId.toString(), priority, state, diagnostic, type, nodeId,
+          containerId.toString(), priority, state, diagnostic, level, nodeId,
           allocationRequestId);
     }
   }
@@ -450,16 +448,17 @@ void finishAppAllocationRecording(ApplicationId applicationId,
     }
   }
 
-  void finishNodeUpdateRecording(NodeId nodeID) {
+  void finishNodeUpdateRecording(NodeId nodeID, String partition) {
     List<NodeAllocation> value = recordingNodesAllocation.get().get(nodeID);
-    long timeStamp = SystemClock.getInstance().getTime();
+    long timestamp = SystemClock.getInstance().getTime();
 
     if (value != null) {
       if (value.size() > 0) {
         lastAvailableNodeActivities = value;
         for (NodeAllocation allocation : lastAvailableNodeActivities) {
           allocation.transformToTree();
-          allocation.setTimeStamp(timeStamp);
+          allocation.setTimestamp(timestamp);
+          allocation.setPartition(partition);
         }
         if (recordNextAvailableNode) {
           recordNextAvailableNode = false;