YARN-11122. Support getClusterNodes API in FederationClientInterceptor

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterMetrics.java

@@ -55,6 +55,8 @@ public final class RouterMetrics {
   private MutableGaugeInt numAppAttemptsFailedRetrieved;
   @Metric("# of getClusterMetrics failed to be retrieved")
   private MutableGaugeInt numGetClusterMetricsFailedRetrieved;
+  @Metric("# of getClusterNodes failed to be retrieved")
+  private MutableGaugeInt numGetClusterNodesFailedRetrieved;
 
   // Aggregate metrics are shared, and don't have to be looked up per call
   @Metric("Total number of successful Submitted apps and latency(ms)")
@@ -74,7 +76,9 @@ public final class RouterMetrics {
   @Metric("Total number of successful Retrieved getClusterMetrics and "
       + "latency(ms)")
   private MutableRate totalSucceededGetClusterMetricsRetrieved;
-
+  @Metric("Total number of successful Retrieved getClusterNodes and "
+          + "latency(ms)")
+  private MutableRate totalSucceededGetClusterNodesRetrieved;
 
   /**
    * Provide quantile counters for all latencies.
@@ -86,6 +90,7 @@ public final class RouterMetrics {
   private MutableQuantiles getApplicationsReportLatency;
   private MutableQuantiles getApplicationAttemptReportLatency;
   private MutableQuantiles getClusterMetricsLatency;
+  private MutableQuantiles getClusterNodesLatency;
 
   private static volatile RouterMetrics INSTANCE = null;
   private static MetricsRegistry registry;
@@ -112,6 +117,10 @@ private RouterMetrics() {
     getClusterMetricsLatency =
         registry.newQuantiles("getClusterMetricsLatency",
             "latency of get cluster metrics", "ops", "latency", 10);
+
+    getClusterNodesLatency =
+            registry.newQuantiles("getClusterNodesLatency",
+                    "latency of get cluster nodes", "ops", "latency", 10);
   }
 
   public static RouterMetrics getMetrics() {
@@ -168,6 +177,11 @@ public long getNumSucceededGetClusterMetricsRetrieved(){
     return totalSucceededGetClusterMetricsRetrieved.lastStat().numSamples();
   }
 
+  @VisibleForTesting
+  public long getNumSucceededGetClusterNodesRetrieved(){
+    return totalSucceededGetClusterNodesRetrieved.lastStat().numSamples();
+  }
+
   @VisibleForTesting
   public double getLatencySucceededAppsCreated() {
     return totalSucceededAppsCreated.lastStat().mean();
@@ -203,6 +217,11 @@ public double getLatencySucceededGetClusterMetricsRetrieved() {
     return totalSucceededGetClusterMetricsRetrieved.lastStat().mean();
   }
 
+  @VisibleForTesting
+  public double getLatencySucceededGetClusterNodesRetrieved() {
+    return totalSucceededGetClusterNodesRetrieved.lastStat().mean();
+  }
+
   @VisibleForTesting
   public int getAppsFailedCreated() {
     return numAppsFailedCreated.value();
@@ -273,6 +292,11 @@ public void succeededGetClusterMetricsRetrieved(long duration) {
     getClusterMetricsLatency.add(duration);
   }
 
+  public void succeededGetClusterNodesRetrieved(long duration) {
+    totalSucceededGetClusterNodesRetrieved.add(duration);
+    getClusterNodesLatency.add(duration);
+  }
+
   public void incrAppsFailedCreated() {
     numAppsFailedCreated.incr();
   }
@@ -301,4 +325,7 @@ public void incrGetClusterMetricsFailedRetrieved() {
     numGetClusterMetricsFailedRetrieved.incr();
   }
 
+  public void incrClusterNodesFailedRetrieved() {
+    numGetClusterNodesFailedRetrieved.incr();
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/FederationClientInterceptor.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.yarn.server.router.clientrm;
 
+import org.apache.hadoop.thirdparty.com.google.common.collect.Maps;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.IOException;
 import java.lang.reflect.Method;
@@ -790,8 +791,32 @@ <R> Map<SubClusterId, R> invokeConcurrent(Collection<SubClusterId> clusterIds,
 
   @Override
   public GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request)
-      throws YarnException, IOException {
-    throw new NotImplementedException("Code is not implemented");
+          throws YarnException, IOException {
+    if (request == null) {
+      routerMetrics.incrClusterNodesFailedRetrieved();
+      RouterServerUtil.logAndThrowException(
+              "Missing getClusterNodes request.", null);
+    }
+    long startTime = clock.getTime();
+    Map<SubClusterId, SubClusterInfo> subclusters =
+            federationFacade.getSubClusters(true);
+    Map<SubClusterId, GetClusterNodesResponse> clusterNodes = Maps.newHashMap();
+    for (Map.Entry<SubClusterId, SubClusterInfo> entry : subclusters.entrySet()) {
+      SubClusterId subClusterId = entry.getKey();
+      ApplicationClientProtocol client = null;
+      try {
+        client = getClientRMProxyForSubCluster(subClusterId);
+        GetClusterNodesResponse response = client.getClusterNodes(request);
+        clusterNodes.put(subClusterId, response);
+      } catch (Exception ex) {
+        routerMetrics.incrClusterNodesFailedRetrieved();
+        LOG.error("Unable to get cluster nodes due to exception.", ex);
+      }
+    }
+    long stopTime = clock.getTime();
+    routerMetrics.succeededGetClusterNodesRetrieved(stopTime - startTime);
+    // Merge the NodesResponse
+    return RouterYarnClientUtils.mergeClusterNodesResponse(clusterNodes.values());
   }
 
   @Override

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterYarnClientUtils.java

@@ -17,17 +17,14 @@
  */
 package org.apache.hadoop.yarn.server.router.clientrm;
 
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
 
 import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsResponse;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.ApplicationReport;
-import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
-import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
+import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesResponse;
+import org.apache.hadoop.yarn.api.records.*;
 import org.apache.hadoop.yarn.server.uam.UnmanagedApplicationManager;
+import org.apache.hadoop.yarn.util.Records;
 import org.apache.hadoop.yarn.util.resource.Resources;
 
 /**
@@ -194,4 +191,24 @@ private static boolean mergeUamToReport(String appName,
     return !(appName.startsWith(UnmanagedApplicationManager.APP_NAME) ||
         appName.startsWith(PARTIAL_REPORT));
   }
+
+  /**
+   * Merges a list of GetClusterNodesResponse.
+   *
+   * @param responses a list of GetClusterNodesResponse to merge
+   * @return the merged GetClusterNodesResponse
+   */
+  public static GetClusterNodesResponse mergeClusterNodesResponse(
+          Collection<GetClusterNodesResponse> responses) {
+    GetClusterNodesResponse clusterNodesResponse = Records.newRecord(
+            GetClusterNodesResponse.class);
+    List<NodeReport> nodeReports = new ArrayList<>();
+    for (GetClusterNodesResponse response : responses) {
+      if (response != null && response.getNodeReports() != null) {
+        nodeReports.addAll(response.getNodeReports());
+      }
+    }
+    clusterNodesResponse.setNodeReports(nodeReports);
+    return clusterNodesResponse;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptor.java

@@ -44,6 +44,8 @@
 import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesRequest;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
@@ -641,4 +643,18 @@ public void testGetApplicationsApplicationStateNotExists() throws Exception{
     Assert.assertNotNull(responseGet);
     Assert.assertTrue(responseGet.getApplicationList().isEmpty());
   }
+
+  @Test
+  public void testGetClusterNodesRequest() throws Exception {
+    LOG.info("Test FederationClientInterceptor : Get Cluster Nodeds request");
+    // null request
+    LambdaTestUtils.intercept(YarnException.class,
+            "Missing getClusterNodes request.", () -> interceptor.getClusterNodes(null));
+
+    // normal request.
+    GetClusterNodesResponse response =
+            interceptor.getClusterNodes(GetClusterNodesRequest.newInstance());
+    Assert.assertEquals(subClusters.size(),
+            response.getNodeReports().size());
+  }
 }