YARN-11028. Add metrics for container allocation latency

Author: Minni Mittal

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptMetrics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -395,7 +396,19 @@ public AllocateResponse allocate(AllocateRequest request)
 
     ApplicationAttemptId appAttemptId =
         amrmTokenIdentifier.getApplicationAttemptId();
+    RMAppAttemptMetrics rmMetrics = getAppAttemptMetrics(appAttemptId);
+    // we do this here to prevent the internal lock in allocate()
+    rmMetrics.setAllocateLatenciesTimestamps(request.getAskList());
+    AllocateResponse response = allocate(request, amrmTokenIdentifier);
+    rmMetrics.updateAllocateLatencies(response.getAllocatedContainers());
+    return response;
+  }
 
+  protected AllocateResponse allocate(AllocateRequest request,
+      AMRMTokenIdentifier amrmTokenIdentifier)
+      throws YarnException, IOException {
+    ApplicationAttemptId appAttemptId =
+        amrmTokenIdentifier.getApplicationAttemptId();
     this.amLivelinessMonitor.receivedPing(appAttemptId);
 
     /* check if its in cache */
@@ -472,6 +485,23 @@ public AllocateResponse allocate(AllocateRequest request)
     }
   }
 
+  protected RMAppAttemptMetrics getAppAttemptMetrics(
+      ApplicationAttemptId appAttemptId) {
+    if (appAttemptId == null) {
+      return null;
+    }
+    RMApp app = this.rmContext.getRMApps().get(appAttemptId.getApplicationId());
+    if (app == null) {
+      return null;
+    }
+    RMAppAttempt attempt = app.getAppAttempts().get(appAttemptId);
+    if (attempt == null) {
+      return null;
+    }
+
+    return attempt.getRMAppAttemptMetrics();
+  }
+
   public void registerAppAttempt(ApplicationAttemptId attemptId) {
     AllocateResponse response =
         recordFactory.newRecordInstance(AllocateResponse.class);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClusterMetrics.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.hadoop.metrics2.lib.MutableQuantiles;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.metrics2.MetricsInfo;
@@ -76,7 +77,10 @@ public class ClusterMetrics {
     MutableGaugeInt rmDispatcherEventQueueSize;
   @Metric("# of scheduler dispatcher event queue size")
     MutableGaugeInt schedulerDispatcherEventQueueSize;
-
+  @Metric("Allocation Latencies for Guarantee containers")
+  MutableQuantiles allocateLatencyGuarQuantiles;
+  @Metric("Allocation Latencies for Opportunistic containers")
+  MutableQuantiles allocateLatencyOppQuantiles;
   private boolean rmEventProcMonitorEnable = false;
 
   private static final MetricsInfo RECORD_INFO = info("ClusterMetrics",
@@ -117,13 +121,25 @@ public static ClusterMetrics getMetrics() {
         if(INSTANCE == null){
           INSTANCE = new ClusterMetrics();
           registerMetrics();
+          INSTANCE.initialize();
           isInitialized.set(true);
         }
       }
     }
     return INSTANCE;
   }
 
+  private void initialize() {
+    allocateLatencyGuarQuantiles = registry
+        .newQuantiles("AllocateLatencyGuaranteed",
+            "Latency to fulfill an Allocate(Guaranteed) requests", "ops",
+            "latency", 5);
+    allocateLatencyOppQuantiles = registry
+        .newQuantiles("AllocateLatencyOpportunistic",
+            "Latency to fulfill an Allocate(Opportunistic) requests", "ops",
+            "latency", 5);
+  }
+
   private static void registerMetrics() {
     registry = new MetricsRegistry(RECORD_INFO);
     registry.tag(RECORD_INFO, "ResourceManager");
@@ -357,6 +373,14 @@ public void incrNumContainerAssigned() {
     numContainersAssigned.incrementAndGet();
   }
 
+  public void addAllocateGuarLatencyEntry(long processingTime) {
+    allocateLatencyGuarQuantiles.add(processingTime);
+  }
+
+  public void addAllocateOppLatencyEntry(long processingTime) {
+    allocateLatencyOppQuantiles.add(processingTime);
+  }
+
   private ScheduledThreadPoolExecutor getAssignCounterExecutor(){
     return assignCounterExecutor;
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptMetrics.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -29,6 +30,11 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
 
 import org.apache.commons.lang3.time.DateUtils;
+import org.apache.hadoop.util.Time;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.server.resourcemanager.ClusterMetrics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
@@ -63,13 +69,18 @@ public class RMAppAttemptMetrics {
       new int[NodeType.values().length][NodeType.values().length];
   private volatile int totalAllocatedContainers;
 
+  private ConcurrentHashMap<Long, Long> allocationGuaranteedLatencies = null;
+  private ConcurrentHashMap<Long, Long> allocationOpportunisticLatencies = null;
+
   public RMAppAttemptMetrics(ApplicationAttemptId attemptId,
       RMContext rmContext) {
     this.attemptId = attemptId;
     ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
     this.readLock = lock.readLock();
     this.writeLock = lock.writeLock();
     this.rmContext = rmContext;
+    this.allocationGuaranteedLatencies = new ConcurrentHashMap<Long, Long>();
+    this.allocationOpportunisticLatencies = new ConcurrentHashMap<Long, Long>();
   }
 
   public void updatePreemptionInfo(Resource resource, RMContainer container) {
@@ -242,4 +253,149 @@ public Resource getApplicationAttemptHeadroom() {
   public void setApplicationAttemptHeadRoom(Resource headRoom) {
     this.applicationHeadroom = headRoom;
   }
+
+  /**
+   * Add allocationID latency to the application ID with a timestap =
+   * CurrentTime (guaranteed)
+   *
+   * @param allocId the allocation Id to add If the allocation ID is already
+   *                present (which shouldn't happen) it ignores the entry
+   */
+  public void addAllocationGuarLatencyIfNotExists(long allocId) {
+    addAllocationGuarLatencyIfNotExists(allocId, System.currentTimeMillis());
+  }
+
+  /**
+   * Add allocationID latency to the application ID with a specific timestamp
+   * (guaranteed)
+   *
+   * @param allocId   allocationId
+   * @param timestamp the timestamp to associate
+   */
+  public void addAllocationGuarLatencyIfNotExists(long allocId,
+      long timestamp) {
+    allocationGuaranteedLatencies.putIfAbsent(allocId, timestamp);
+  }
+
+  /**
+   * Add allocationID latency to the application ID with a timestap =
+   * CurrentTime (opportunistic)
+   *
+   * @param allocId the allocation Id to add If the allocation ID is already
+   *                present (which shouldn't happen) it ignores the entry
+   */
+  public void addAllocationOppLatencyIfNotExists(long allocId) {
+    this.addAllocationOppLatencyIfNotExists(allocId,
+        System.currentTimeMillis());
+  }
+
+  /**
+   * Add allocationID latency to the application ID with a specific timestamp
+   * (opportunistic)
+   *
+   * @param allocId   allocationId
+   * @param timestamp the timestamp to associate
+   */
+  public void addAllocationOppLatencyIfNotExists(long allocId, long timestamp) {
+    allocationOpportunisticLatencies.putIfAbsent(allocId, timestamp);
+  }
+
+  /**
+   * Returns the time associated when the allocation Id was added This method
+   * removes the allocation Id from the class (guaranteed)
+   *
+   * @param allocId the allocation ID to get the associated time
+   * @return the timestamp associated with that allocation id as well as stop
+   * tracking it
+   */
+  public long getAndRemoveGuaAllocationLatencies(long allocId) {
+    Long ret = allocationGuaranteedLatencies.remove(new Long(allocId));
+    return ret != null ? ret : 0l;
+  }
+
+  /**
+   * Returns the time associated when the allocation Id was added This method
+   * removes the allocation Id from the class (opportunistic)
+   *
+   * @param allocId the allocation ID to get the associated time
+   * @return the timestamp associated with that allocation id as well as stop
+   * tracking it
+   */
+  public long getAndRemoveOppAllocationLatencies(long allocId) {
+    Long ret = allocationOpportunisticLatencies.remove(new Long(allocId));
+    return ret != null ? ret : 0l;
+  }
+
+  /**
+   * Set timestamp for the provided ResourceRequest. It will correctly identify
+   * their ExecutionType, provided they have they have allocateId != 0 (DEFAULT)
+   * This is used in conjunction with This is used in conjunction with
+   * updatePromoteLatencies method method
+   *
+   * @param requests the ResourceRequests to add.
+   */
+  public void setAllocateLatenciesTimestamps(List<ResourceRequest> requests) {
+    long now = Time.now();
+    for (ResourceRequest req : requests) {
+      if (req.getNumContainers() > 0) {
+        // we dont support tracking with negative or zero allocationIds
+        long allocationRequestId = req.getAllocationRequestId();
+        if (allocationRequestId > 0) {
+          if (req.getExecutionTypeRequest() != null) {
+            if (ExecutionType.GUARANTEED
+                .equals(req.getExecutionTypeRequest().getExecutionType())) {
+              addAllocationGuarLatencyIfNotExists(allocationRequestId, now);
+            } else {
+              addAllocationOppLatencyIfNotExists(allocationRequestId, now);
+            }
+          }
+        } else {
+          LOG.warn(String.format(
+              "Can't register allocate latency for %s container "
+                  + "with negative or zero allocation IDs",
+              req.getExecutionTypeRequest().getExecutionType()));
+        }
+      }
+    }
+  }
+
+  /**
+   * Updated the JMX metrics class (ClusterMetrics) with the delta time when
+   * these containers where added. It will correctly identify their
+   * ExecutionType, provided they have they have allocateId != 0 (DEFAULT)
+   *
+   * @param response the list of the containers to allocate.
+   */
+  public void updateAllocateLatencies(List<Container> response) {
+
+    for (Container container : response) {
+      long allocationRequestId = container.getAllocationRequestId();
+      // we dont support tracking with negative or zero allocationIds
+      if (allocationRequestId > 0) {
+        long now = System.currentTimeMillis();
+        long allocIdTime =
+            (container.getExecutionType() == ExecutionType.GUARANTEED) ?
+                getAndRemoveGuaAllocationLatencies(allocationRequestId) :
+                getAndRemoveOppAllocationLatencies(allocationRequestId);
+        if (allocIdTime != 0) {
+          if (container.getExecutionType() == ExecutionType.GUARANTEED) {
+            ClusterMetrics.getMetrics()
+                .addAllocateGuarLatencyEntry(now - allocIdTime);
+          } else {
+            ClusterMetrics.getMetrics()
+                .addAllocateOppLatencyEntry(now - allocIdTime);
+          }
+        } else {
+          LOG.error(String.format(
+              "Can't register allocate latency for %s container %s; allotTime=%d ",
+              container.getExecutionType(), container.getId(), allocIdTime));
+        }
+      } else {
+        LOG.warn(String.format("Cant register promotion latency "
+                + "for container %s. Either allocationID is less than equal to 0 or "
+                + "lost the container ID", container.getExecutionType().name(),
+            container.getId()));
+      }
+    }
+  }
 }