YARN-6625. yarn application -list returns a tracking URL for AM that doesn't work in secured and HA environment. (Yufei Gu)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/amfilter/AmIpFilter.java

@@ -23,10 +23,12 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.UnknownHostException;
+import java.net.HttpURLConnection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.Collection;
 
 import javax.servlet.Filter;
 import javax.servlet.FilterChain;
@@ -38,12 +40,12 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.yarn.conf.HAUtil;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.webproxy.ProxyUtils;
 import org.apache.hadoop.yarn.server.webproxy.WebAppProxyServlet;
-import org.apache.hadoop.yarn.util.RMHAUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,7 +68,8 @@ public class AmIpFilter implements Filter {
   private String[] proxyHosts;
   private Set<String> proxyAddresses = null;
   private long lastUpdate;
-  private Map<String, String> proxyUriBases;
+  @VisibleForTesting
+  Map<String, String> proxyUriBases;
 
   @Override
   public void init(FilterConfig conf) throws ServletException {
@@ -187,24 +190,55 @@ public void doFilter(ServletRequest req, ServletResponse resp,
     }
   }
 
-  protected String findRedirectUrl() throws ServletException {
-    String addr;
-    if (proxyUriBases.size() == 1) {  // external proxy or not RM HA
+  @VisibleForTesting
+  public String findRedirectUrl() throws ServletException {
+    String addr = null;
+    if (proxyUriBases.size() == 1) {
+      // external proxy or not RM HA
       addr = proxyUriBases.values().iterator().next();
-    } else {                          // RM HA
+    } else {
+      // RM HA
       YarnConfiguration conf = new YarnConfiguration();
-      String activeRMId = RMHAUtils.findActiveRMHAId(conf);
-      String addressPropertyPrefix = YarnConfiguration.useHttps(conf)
-          ? YarnConfiguration.RM_WEBAPP_HTTPS_ADDRESS
-          : YarnConfiguration.RM_WEBAPP_ADDRESS;
-      String host = conf.get(
-          HAUtil.addSuffix(addressPropertyPrefix, activeRMId));
-      addr = proxyUriBases.get(host);
+      for (String rmId : getRmIds(conf)) {
+        String url = getUrlByRmId(conf, rmId);
+        if (isValidUrl(url)) {
+          addr = url;
+          break;
+        }
+      }
     }
+
     if (addr == null) {
       throw new ServletException(
           "Could not determine the proxy server for redirection");
     }
     return addr;
   }
+
+  @VisibleForTesting
+  Collection<String> getRmIds(YarnConfiguration conf) {
+    return conf.getStringCollection(YarnConfiguration.RM_HA_IDS);
+  }
+
+  @VisibleForTesting
+  String getUrlByRmId(YarnConfiguration conf, String rmId) {
+    String addressPropertyPrefix = YarnConfiguration.useHttps(conf) ?
+        YarnConfiguration.RM_WEBAPP_HTTPS_ADDRESS :
+        YarnConfiguration.RM_WEBAPP_ADDRESS;
+    String host = conf.get(HAUtil.addSuffix(addressPropertyPrefix, rmId));
+    return proxyUriBases.get(host);
+  }
+
+  private boolean isValidUrl(String url) {
+    boolean isValid = false;
+    try {
+      HttpURLConnection conn =
+          (HttpURLConnection) new URL(url).openConnection();
+      conn.connect();
+      isValid = conn.getResponseCode() == HttpURLConnection.HTTP_OK;
+    } catch (Exception e) {
+      LOG.debug("Failed to connect to " + url + ": " + e.toString());
+    }
+    return isValid;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/test/java/org/apache/hadoop/yarn/server/webproxy/amfilter/TestAmFilter.java

@@ -22,18 +22,41 @@
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.net.HttpURLConnection;
-import java.util.*;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.Enumeration;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import javax.servlet.*;
+import javax.servlet.FilterConfig;
+import javax.servlet.FilterChain;
+import javax.servlet.Filter;
+import javax.servlet.ServletContext;
+import javax.servlet.ServletResponse;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletException;
 import javax.servlet.http.Cookie;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 
+import org.apache.hadoop.http.TestHttpServer;
 import org.apache.hadoop.yarn.server.webproxy.ProxyUtils;
 import org.apache.hadoop.yarn.server.webproxy.WebAppProxyServlet;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.eclipse.jetty.util.thread.QueuedThreadPool;
 import org.glassfish.grizzly.servlet.HttpServletResponseImpl;
 import org.junit.Test;
 import org.mockito.Mockito;
@@ -121,6 +144,47 @@ public void doFilter(ServletRequest servletRequest,
     filter.destroy();
   }
 
+  @Test
+  public void testFindRedirectUrl() throws Exception {
+    final String rm1 = "rm1";
+    final String rm2 = "rm2";
+    // generate a valid URL
+    final String rm1Url = startHttpServer();
+    // invalid url
+    final String rm2Url = "host2:8088";
+
+    TestAmIpFilter filter = new TestAmIpFilter();
+    TestAmIpFilter spy = Mockito.spy(filter);
+    // make sure findRedirectUrl() go to HA branch
+    spy.proxyUriBases = new HashMap<>();
+    spy.proxyUriBases.put(rm1, rm1Url);
+    spy.proxyUriBases.put(rm2, rm2Url);
+
+    Collection<String> rmIds = new ArrayList<>(Arrays.asList(rm1, rm2));
+    Mockito.doReturn(rmIds).when(spy).getRmIds(Mockito.any());
+    Mockito.doReturn(rm1Url).when(spy)
+        .getUrlByRmId(Mockito.any(), Mockito.eq(rm2));
+    Mockito.doReturn(rm2Url).when(spy)
+        .getUrlByRmId(Mockito.any(), Mockito.eq(rm1));
+
+    assertEquals(spy.findRedirectUrl(), rm1Url);
+  }
+
+  private String startHttpServer() throws Exception {
+    Server server = new Server(0);
+    ((QueuedThreadPool)server.getThreadPool()).setMaxThreads(10);
+    ServletContextHandler context = new ServletContextHandler();
+    context.setContextPath("/foo");
+    server.setHandler(context);
+    String servletPath = "/bar";
+    context.addServlet(new ServletHolder(TestHttpServer.EchoServlet.class),
+        servletPath);
+    ((ServerConnector)server.getConnectors()[0]).setHost("localhost");
+    server.start();
+    System.setProperty("sun.net.http.allowRestrictedHeaders", "true");
+    return server.getURI().toString() + servletPath;
+  }
+
   /**
    * Test AmIpFilter
    */