YARN-6113. Re-direct NM Web Service to get container logs for finished applications. Contributed by Xuan Gong.

(cherry picked from commit 464ff479ce)

Conflicts:

	hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServices.java

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -1022,6 +1022,9 @@ public class YarnConfiguration extends Configuration {
   public static final String YARN_LOG_SERVER_URL =
     YARN_PREFIX + "log.server.url";
 
+  public static final String YARN_LOG_SERVER_WEBSERVICE_URL =
+      YARN_PREFIX + "log.server.web-service.url";
+
   public static final String YARN_TRACKING_URL_GENERATOR = 
       YARN_PREFIX + "tracking.url.generator";
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/WebAppUtils.java

@@ -26,6 +26,7 @@ import java.net.UnknownHostException;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Iterator;
 import java.util.List;
 
 import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -424,7 +425,8 @@ public class WebAppUtils {
     return Arrays.asList("text", "octet-stream");
   }
 
-  private static String getURLEncodedQueryString(HttpServletRequest request) {
+  private static String getURLEncodedQueryString(HttpServletRequest request,
+      String parameterToRemove) {
     String queryString = request.getQueryString();
     if (queryString != null && !queryString.isEmpty()) {
       String reqEncoding = request.getCharacterEncoding();
@@ -432,12 +434,33 @@ public class WebAppUtils {
         reqEncoding = "ISO-8859-1";
       }
       Charset encoding = Charset.forName(reqEncoding);
-      List<NameValuePair> params = URLEncodedUtils.parse(queryString, encoding);
+      List<NameValuePair> params = URLEncodedUtils.parse(queryString,
+          encoding);
+      if (parameterToRemove != null && !parameterToRemove.isEmpty()) {
+        Iterator<NameValuePair> paramIterator = params.iterator();
+        while(paramIterator.hasNext()) {
+          NameValuePair current = paramIterator.next();
+          if (current.getName().equals(parameterToRemove)) {
+            paramIterator.remove();
+          }
+        }
+      }
       return URLEncodedUtils.format(params, encoding);
     }
     return null;
   }
 
+  /**
+   * Get a query string which removes the passed parameter.
+   * @param httpRequest HttpServletRequest with the request details
+   * @param parameterName the query parameters must be removed
+   * @return the query parameter string
+   */
+  public static String removeQueryParams(HttpServletRequest httpRequest,
+      String parameterName) {
+    return getURLEncodedQueryString(httpRequest, parameterName);
+  }
+
   /**
    * Get a HTML escaped uri with the query parameters of the request.
    * @param request HttpServletRequest with the request details
@@ -445,7 +468,7 @@ public class WebAppUtils {
    */
   public static String getHtmlEscapedURIWithQueryString(
       HttpServletRequest request) {
-    String urlEncodedQueryString = getURLEncodedQueryString(request);
+    String urlEncodedQueryString = getURLEncodedQueryString(request, null);
     if (urlEncodedQueryString != null) {
       return HtmlQuoting.quoteHtmlChars(
           request.getRequestURI() + "?" + urlEncodedQueryString);
@@ -462,7 +485,7 @@ public class WebAppUtils {
   public static String appendQueryParams(HttpServletRequest request,
       String targetUri) {
     String ret = targetUri;
-    String urlEncodedQueryString = getURLEncodedQueryString(request);
+    String urlEncodedQueryString = getURLEncodedQueryString(request, null);
     if (urlEncodedQueryString != null) {
       ret += "?" + urlEncodedQueryString;
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml

@@ -2590,6 +2590,14 @@
     <value></value>
   </property>
 
+  <property>
+    <description>
+    URL for log aggregation server web service
+    </description>
+    <name>yarn.log.server.web-service.url</name>
+    <value></value>
+  </property>
+
   <property>
     <description>
     RM Application Tracking URL

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebServices.java

@@ -49,6 +49,7 @@ import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
@@ -86,6 +87,7 @@ public class NMWebServices {
   private WebApp webapp;
   private static RecordFactory recordFactory = RecordFactoryProvider
       .getRecordFactory(null);
+  private final String redirectWSUrl;
 
   private @javax.ws.rs.core.Context 
     HttpServletRequest request;
@@ -102,6 +104,8 @@ public class NMWebServices {
     this.nmContext = nm;
     this.rview = view;
     this.webapp = webapp;
+    this.redirectWSUrl = this.nmContext.getConf().get(
+        YarnConfiguration.YARN_LOG_SERVER_WEBSERVICE_URL);
   }
 
   private void init() {
@@ -261,6 +265,9 @@ public class NMWebServices {
       } catch (IOException ex) {
         // Something wrong with we tries to access the remote fs for the logs.
         // Skip it and do nothing
+        if (LOG.isDebugEnabled()) {
+          LOG.debug(ex.getMessage());
+        }
       }
       GenericEntity<List<ContainerLogsInfo>> meta = new GenericEntity<List<
           ContainerLogsInfo>>(containersLogsInfo){};
@@ -271,7 +278,13 @@ public class NMWebServices {
       resp.header("X-Content-Type-Options", "nosniff");
       return resp.build();
     } catch (Exception ex) {
-      throw new WebApplicationException(ex);
+      if (redirectWSUrl == null || redirectWSUrl.isEmpty()) {
+        throw new WebApplicationException(ex);
+      }
+      // redirect the request to the configured log server
+      String redirectURI = "/containers/" + containerIdStr
+          + "/logs";
+      return createRedirectResponse(hsr, redirectWSUrl, redirectURI);
     }
   }
 
@@ -369,7 +382,14 @@ public class NMWebServices {
       logFile = ContainerLogsUtils.getContainerLogFile(
           containerId, filename, request.getRemoteUser(), nmContext);
     } catch (NotFoundException ex) {
-      return Response.status(Status.NOT_FOUND).entity(ex.getMessage()).build();
+      if (redirectWSUrl == null || redirectWSUrl.isEmpty()) {
+        return Response.status(Status.NOT_FOUND).entity(ex.getMessage())
+            .build();
+      }
+      // redirect the request to the configured log server
+      String redirectURI = "/containers/" + containerIdStr
+          + "/logs/" + filename;
+      return createRedirectResponse(request, redirectWSUrl, redirectURI);
     } catch (YarnException ex) {
       return Response.serverError().entity(ex.getMessage()).build();
     }
@@ -456,4 +476,25 @@ public class NMWebServices {
     }
     return Long.parseLong(bytes);
   }
+
+  private Response createRedirectResponse(HttpServletRequest httpRequest,
+      String redirectWSUrlPrefix, String uri) {
+    // redirect the request to the configured log server
+    StringBuilder redirectPath = new StringBuilder();
+    if (redirectWSUrlPrefix.endsWith("/")) {
+      redirectWSUrlPrefix = redirectWSUrlPrefix.substring(0,
+          redirectWSUrlPrefix.length() - 1);
+    }
+    redirectPath.append(redirectWSUrlPrefix + uri);
+    // append all the request query parameters except nodeId parameter
+    String requestParams = WebAppUtils.removeQueryParams(httpRequest,
+        YarnWebServiceParams.NM_ID);
+    if (requestParams != null && !requestParams.isEmpty()) {
+      redirectPath.append("?" + requestParams);
+    }
+    ResponseBuilder res = Response.status(
+        HttpServletResponse.SC_TEMPORARY_REDIRECT);
+    res.header("Location", redirectPath.toString());
+    return res.build();
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServices.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.yarn.server.nodemanager.webapp;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -26,7 +27,11 @@ import java.io.File;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringReader;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.net.URL;
 import java.util.List;
+import javax.servlet.http.HttpServletResponse;
 import javax.ws.rs.core.MediaType;
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -57,6 +62,7 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.Conta
 import org.apache.hadoop.yarn.server.nodemanager.webapp.WebServer.NMWebApp;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
+import org.apache.hadoop.yarn.server.webapp.YarnWebServiceParams;
 import org.apache.hadoop.yarn.server.webapp.dao.ContainerLogsInfo;
 import org.apache.hadoop.yarn.util.YarnVersionInfo;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
@@ -96,6 +102,7 @@ public class TestNMWebServices extends JerseyTestBase {
   private static ApplicationACLsManager aclsManager;
   private static LocalDirsHandlerService dirsHandler;
   private static WebApp nmWebApp;
+  private static final String LOGSERVICEWSADDR = "test:1234";
 
   private static final File testRootDir = new File("target",
       TestNMWebServices.class.getSimpleName());
@@ -114,6 +121,8 @@ public class TestNMWebServices extends JerseyTestBase {
       conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
       conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
           testRemoteLogDir.getAbsolutePath());
+      conf.set(YarnConfiguration.YARN_LOG_SERVER_WEBSERVICE_URL,
+          LOGSERVICEWSADDR);
       dirsHandler = new LocalDirsHandlerService();
       NodeHealthCheckerService healthChecker = new NodeHealthCheckerService(
           NodeManager.getNodeHealthScriptRunner(conf), dirsHandler);
@@ -344,6 +353,58 @@ public class TestNMWebServices extends JerseyTestBase {
     testContainerLogs(r, containerId);
   }
 
+  @Test (timeout = 10000)
+  public void testNMRedirect() {
+    ApplicationId noExistAppId = ApplicationId.newInstance(
+        System.currentTimeMillis(), 2000);
+    ApplicationAttemptId noExistAttemptId = ApplicationAttemptId.newInstance(
+        noExistAppId, 150);
+    ContainerId noExistContainerId = ContainerId.newContainerId(
+        noExistAttemptId, 250);
+    String fileName = "syslog";
+    WebResource r = resource();
+
+    // check the old api
+    URI requestURI = r.path("ws").path("v1").path("node")
+        .path("containerlogs").path(noExistContainerId.toString())
+        .path(fileName).queryParam("user.name", "user")
+        .queryParam(YarnWebServiceParams.NM_ID, "localhost:1111")
+        .getURI();
+    String redirectURL = getRedirectURL(requestURI.toString());
+    assertTrue(redirectURL != null);
+    assertTrue(redirectURL.contains(LOGSERVICEWSADDR));
+    assertTrue(redirectURL.contains(noExistContainerId.toString()));
+    assertTrue(redirectURL.contains("/logs/" + fileName));
+    assertTrue(redirectURL.contains("user.name=" + "user"));
+    assertFalse(redirectURL.contains(YarnWebServiceParams.NM_ID));
+
+    // check the new api
+    requestURI = r.path("ws").path("v1").path("node")
+        .path("containers").path(noExistContainerId.toString())
+        .path("logs").path(fileName).queryParam("user.name", "user")
+        .queryParam(YarnWebServiceParams.NM_ID, "localhost:1111")
+        .getURI();
+    redirectURL = getRedirectURL(requestURI.toString());
+    assertTrue(redirectURL != null);
+    assertTrue(redirectURL.contains(LOGSERVICEWSADDR));
+    assertTrue(redirectURL.contains(noExistContainerId.toString()));
+    assertTrue(redirectURL.contains("/logs/" + fileName));
+    assertTrue(redirectURL.contains("user.name=" + "user"));
+    assertFalse(redirectURL.contains(YarnWebServiceParams.NM_ID));
+
+    requestURI = r.path("ws").path("v1").path("node")
+        .path("containers").path(noExistContainerId.toString())
+        .path("logs").queryParam("user.name", "user")
+        .queryParam(YarnWebServiceParams.NM_ID, "localhost:1111")
+        .getURI();
+    redirectURL = getRedirectURL(requestURI.toString());
+    assertTrue(redirectURL != null);
+    assertTrue(redirectURL.contains(LOGSERVICEWSADDR));
+    assertTrue(redirectURL.contains(noExistContainerId.toString()));
+    assertTrue(redirectURL.contains("user.name=" + "user"));
+    assertFalse(redirectURL.contains(YarnWebServiceParams.NM_ID));
+  }
+
   private void testContainerLogs(WebResource r, ContainerId containerId)
       throws IOException {
     final String containerIdStr = containerId.toString();
@@ -443,12 +504,12 @@ public class TestNMWebServices extends JerseyTestBase {
         + WebAppUtils.listSupportedLogContentType(), responseText);
     assertEquals(400, response.getStatus());
 
-    // ask for file that doesn't exist
+    // ask for file that doesn't exist and it will re-direct to
-    response = r.path("uhhh")
+    // the log server
-        .accept(MediaType.TEXT_PLAIN).get(ClientResponse.class);
+    URI requestURI = r.path("uhhh").getURI();
-    Assert.assertEquals(Status.NOT_FOUND.getStatusCode(), response.getStatus());
+    String redirectURL = getRedirectURL(requestURI.toString());
-    responseText = response.getEntity(String.class);
+    assertTrue(redirectURL != null);
-    assertTrue(responseText.contains("Cannot find this log on the local disk."));
+    assertTrue(redirectURL.contains(LOGSERVICEWSADDR));
 
     // Get container log files' name
     WebResource r1 = resource();
@@ -620,4 +681,21 @@ public class TestNMWebServices extends JerseyTestBase {
     int postfixIndex = fullMessage.indexOf(postfix);
     return fullMessage.substring(prefixIndex, postfixIndex);
   }
+
+  private static String getRedirectURL(String url) {
+    String redirectUrl = null;
+    try {
+      HttpURLConnection conn = (HttpURLConnection) new URL(url)
+          .openConnection();
+      // do not automatically follow the redirection
+      // otherwise we get too many redirections exception
+      conn.setInstanceFollowRedirects(false);
+      if(conn.getResponseCode() == HttpServletResponse.SC_TEMPORARY_REDIRECT) {
+        redirectUrl = conn.getHeaderField("Location");
+      }
+    } catch (Exception e) {
+      // throw new RuntimeException(e);
+    }
+    return redirectUrl;
+  }
 }