You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ae...@apache.org on 2016/05/09 17:36:30 UTC
[21/50] [abbrv] hadoop git commit: YARN-4920. ATS/NM should support a
link to dowload/get the logs in text format. Contributed by Xuan Gong.
YARN-4920. ATS/NM should support a link to dowload/get the logs in text format. Contributed by Xuan Gong.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/e61d4312
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/e61d4312
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/e61d4312
Branch: refs/heads/HDFS-1312
Commit: e61d431275d7fe5641fe9da4903e285b10330fa0
Parents: af94258
Author: Junping Du <ju...@apache.org>
Authored: Wed May 4 09:40:13 2016 -0700
Committer: Junping Du <ju...@apache.org>
Committed: Wed May 4 10:35:49 2016 -0700
----------------------------------------------------------------------
.../webapp/AHSWebServices.java | 270 ++++++++++++++++++-
...pplicationHistoryManagerOnTimelineStore.java | 29 +-
.../webapp/TestAHSWebServices.java | 203 +++++++++++++-
.../yarn/server/webapp/dao/ContainerInfo.java | 6 +
.../nodemanager/webapp/NMWebServices.java | 22 +-
.../nodemanager/webapp/TestNMWebServices.java | 12 +-
6 files changed, 525 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e61d4312/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/AHSWebServices.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/AHSWebServices.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/AHSWebServices.java
index e7a22bd..75dce07 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/AHSWebServices.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/AHSWebServices.java
@@ -18,6 +18,11 @@
package org.apache.hadoop.yarn.server.applicationhistoryservice.webapp;
+import java.io.DataInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.Charset;
import java.util.Collections;
import java.util.Set;
@@ -28,13 +33,30 @@ import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.StreamingOutput;
+import javax.ws.rs.core.Response.ResponseBuilder;
+import javax.ws.rs.core.Response.Status;
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.api.ApplicationBaseProtocol;
import org.apache.hadoop.yarn.api.records.timeline.TimelineAbout;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat;
+import org.apache.hadoop.yarn.logaggregation.LogAggregationUtils;
+import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat.LogKey;
import org.apache.hadoop.yarn.server.webapp.WebServices;
import org.apache.hadoop.yarn.server.webapp.dao.AppAttemptInfo;
import org.apache.hadoop.yarn.server.webapp.dao.AppAttemptsInfo;
@@ -42,9 +64,10 @@ import org.apache.hadoop.yarn.server.webapp.dao.AppInfo;
import org.apache.hadoop.yarn.server.webapp.dao.AppsInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainerInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainersInfo;
+import org.apache.hadoop.yarn.util.Times;
import org.apache.hadoop.yarn.util.timeline.TimelineUtils;
import org.apache.hadoop.yarn.webapp.BadRequestException;
-
+import com.google.common.base.Joiner;
import com.google.inject.Inject;
import com.google.inject.Singleton;
@@ -52,9 +75,17 @@ import com.google.inject.Singleton;
@Path("/ws/v1/applicationhistory")
public class AHSWebServices extends WebServices {
+ private static final String NM_DOWNLOAD_URI_STR =
+ "/ws/v1/node/containerlogs";
+ private static final Joiner JOINER = Joiner.on("");
+ private static final Joiner DOT_JOINER = Joiner.on(". ");
+ private final Configuration conf;
+
@Inject
- public AHSWebServices(ApplicationBaseProtocol appBaseProt) {
+ public AHSWebServices(ApplicationBaseProtocol appBaseProt,
+ Configuration conf) {
super(appBaseProt);
+ this.conf = conf;
}
@GET
@@ -173,4 +204,239 @@ public class AHSWebServices extends WebServices {
}
}
+ @GET
+ @Path("/containerlogs/{containerid}/{filename}")
+ @Produces({ MediaType.TEXT_PLAIN })
+ @Public
+ @Unstable
+ public Response getLogs(@Context HttpServletRequest req,
+ @Context HttpServletResponse res,
+ @PathParam("containerid") String containerIdStr,
+ @PathParam("filename") String filename,
+ @QueryParam("download") String download) {
+ init(res);
+ ContainerId containerId;
+ try {
+ containerId = ContainerId.fromString(containerIdStr);
+ } catch (IllegalArgumentException ex) {
+ return createBadResponse(Status.NOT_FOUND,
+ "Invalid ContainerId: " + containerIdStr);
+ }
+
+ boolean downloadFile = parseBooleanParam(download);
+
+ ApplicationId appId = containerId.getApplicationAttemptId()
+ .getApplicationId();
+ AppInfo appInfo;
+ try {
+ appInfo = super.getApp(req, res, appId.toString());
+ } catch (Exception ex) {
+ // directly find logs from HDFS.
+ return sendStreamOutputResponse(appId, null, null, containerIdStr,
+ filename, downloadFile);
+ }
+ String appOwner = appInfo.getUser();
+
+ ContainerInfo containerInfo;
+ try {
+ containerInfo = super.getContainer(
+ req, res, appId.toString(),
+ containerId.getApplicationAttemptId().toString(),
+ containerId.toString());
+ } catch (Exception ex) {
+ if (isFinishedState(appInfo.getAppState())) {
+ // directly find logs from HDFS.
+ return sendStreamOutputResponse(appId, appOwner, null, containerIdStr,
+ filename, downloadFile);
+ }
+ return createBadResponse(Status.INTERNAL_SERVER_ERROR,
+ "Can not get ContainerInfo for the container: " + containerId);
+ }
+ String nodeId = containerInfo.getNodeId();
+ if (isRunningState(appInfo.getAppState())) {
+ String nodeHttpAddress = containerInfo.getNodeHttpAddress();
+ String uri = "/" + containerId.toString() + "/" + filename;
+ String resURI = JOINER.join(nodeHttpAddress, NM_DOWNLOAD_URI_STR, uri);
+ String query = req.getQueryString();
+ if (query != null && !query.isEmpty()) {
+ resURI += "?" + query;
+ }
+ ResponseBuilder response = Response.status(
+ HttpServletResponse.SC_TEMPORARY_REDIRECT);
+ response.header("Location", resURI);
+ return response.build();
+ } else if (isFinishedState(appInfo.getAppState())) {
+ return sendStreamOutputResponse(appId, appOwner, nodeId,
+ containerIdStr, filename, downloadFile);
+ } else {
+ return createBadResponse(Status.NOT_FOUND,
+ "The application is not at Running or Finished State.");
+ }
+ }
+
+ // TODO: YARN-5029. RM would send the update event. We could get
+ // the consistent YarnApplicationState.
+ // Will remove YarnApplicationState.ACCEPTED.
+ private boolean isRunningState(YarnApplicationState appState) {
+ return appState == YarnApplicationState.ACCEPTED
+ || appState == YarnApplicationState.RUNNING;
+ }
+
+ private boolean isFinishedState(YarnApplicationState appState) {
+ return appState == YarnApplicationState.FINISHED
+ || appState == YarnApplicationState.FAILED
+ || appState == YarnApplicationState.KILLED;
+ }
+
+ private Response createBadResponse(Status status, String errMessage) {
+ Response response = Response.status(status)
+ .entity(DOT_JOINER.join(status.toString(), errMessage)).build();
+ return response;
+ }
+
+ private boolean parseBooleanParam(String param) {
+ return ("true").equalsIgnoreCase(param);
+ }
+
+ private Response sendStreamOutputResponse(ApplicationId appId,
+ String appOwner, String nodeId, String containerIdStr,
+ String fileName, boolean downloadFile) {
+ StreamingOutput stream = null;
+ try {
+ stream = getStreamingOutput(appId, appOwner, nodeId,
+ containerIdStr, fileName);
+ } catch (Exception ex) {
+ return createBadResponse(Status.INTERNAL_SERVER_ERROR,
+ ex.getMessage());
+ }
+ if (stream == null) {
+ return createBadResponse(Status.INTERNAL_SERVER_ERROR,
+ "Can not get log for container: " + containerIdStr);
+ }
+ ResponseBuilder response = Response.ok(stream);
+ if (downloadFile) {
+ response.header("Content-Type", "application/octet-stream");
+ response.header("Content-Disposition", "attachment; filename="
+ + fileName);
+ }
+ return response.build();
+ }
+
+ private StreamingOutput getStreamingOutput(ApplicationId appId,
+ String appOwner, final String nodeId, final String containerIdStr,
+ final String logFile) throws IOException{
+ String suffix = LogAggregationUtils.getRemoteNodeLogDirSuffix(conf);
+ org.apache.hadoop.fs.Path remoteRootLogDir = new org.apache.hadoop.fs.Path(
+ conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
+ YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
+ org.apache.hadoop.fs.Path qualifiedRemoteRootLogDir =
+ FileContext.getFileContext(conf).makeQualified(remoteRootLogDir);
+ FileContext fc = FileContext.getFileContext(
+ qualifiedRemoteRootLogDir.toUri(), conf);
+ org.apache.hadoop.fs.Path remoteAppDir = null;
+ if (appOwner == null) {
+ org.apache.hadoop.fs.Path toMatch = LogAggregationUtils
+ .getRemoteAppLogDir(remoteRootLogDir, appId, "*", suffix);
+ FileStatus[] matching = fc.util().globStatus(toMatch);
+ if (matching == null || matching.length != 1) {
+ return null;
+ }
+ remoteAppDir = matching[0].getPath();
+ } else {
+ remoteAppDir = LogAggregationUtils
+ .getRemoteAppLogDir(remoteRootLogDir, appId, appOwner, suffix);
+ }
+ final RemoteIterator<FileStatus> nodeFiles;
+ nodeFiles = fc.listStatus(remoteAppDir);
+ if (!nodeFiles.hasNext()) {
+ return null;
+ }
+
+ StreamingOutput stream = new StreamingOutput() {
+
+ @Override
+ public void write(OutputStream os) throws IOException,
+ WebApplicationException {
+ byte[] buf = new byte[65535];
+ boolean findLogs = false;
+ while (nodeFiles.hasNext()) {
+ final FileStatus thisNodeFile = nodeFiles.next();
+ String nodeName = thisNodeFile.getPath().getName();
+ if ((nodeId == null || nodeName.contains(LogAggregationUtils
+ .getNodeString(nodeId))) && !nodeName.endsWith(
+ LogAggregationUtils.TMP_FILE_SUFFIX)) {
+ AggregatedLogFormat.LogReader reader =
+ new AggregatedLogFormat.LogReader(conf,
+ thisNodeFile.getPath());
+ DataInputStream valueStream;
+ LogKey key = new LogKey();
+ valueStream = reader.next(key);
+ while (valueStream != null && !key.toString()
+ .equals(containerIdStr)) {
+ // Next container
+ key = new LogKey();
+ valueStream = reader.next(key);
+ }
+ if (valueStream == null) {
+ continue;
+ }
+ while (true) {
+ try {
+ String fileType = valueStream.readUTF();
+ String fileLengthStr = valueStream.readUTF();
+ long fileLength = Long.parseLong(fileLengthStr);
+ if (fileType.equalsIgnoreCase(logFile)) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("LogType:");
+ sb.append(fileType + "\n");
+ sb.append("Log Upload Time:");
+ sb.append(Times.format(System.currentTimeMillis()) + "\n");
+ sb.append("LogLength:");
+ sb.append(fileLengthStr + "\n");
+ sb.append("Log Contents:\n");
+ byte[] b = sb.toString().getBytes(Charset.forName("UTF-8"));
+ os.write(b, 0, b.length);
+
+ long curRead = 0;
+ long pendingRead = fileLength - curRead;
+ int toRead = pendingRead > buf.length ? buf.length
+ : (int) pendingRead;
+ int len = valueStream.read(buf, 0, toRead);
+ while (len != -1 && curRead < fileLength) {
+ os.write(buf, 0, len);
+ curRead += len;
+
+ pendingRead = fileLength - curRead;
+ toRead = pendingRead > buf.length ? buf.length
+ : (int) pendingRead;
+ len = valueStream.read(buf, 0, toRead);
+ }
+ sb = new StringBuilder();
+ sb.append("\nEnd of LogType:" + fileType + "\n");
+ b = sb.toString().getBytes(Charset.forName("UTF-8"));
+ os.write(b, 0, b.length);
+ findLogs = true;
+ } else {
+ long totalSkipped = 0;
+ long currSkipped = 0;
+ while (currSkipped != -1 && totalSkipped < fileLength) {
+ currSkipped = valueStream.skip(fileLength - totalSkipped);
+ totalSkipped += currSkipped;
+ }
+ }
+ } catch (EOFException eof) {
+ break;
+ }
+ }
+ }
+ }
+ os.flush();
+ if (!findLogs) {
+ throw new IOException("Can not find logs for container:"
+ + containerIdStr);
+ }
+ }
+ };
+ return stream;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e61d4312/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryManagerOnTimelineStore.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryManagerOnTimelineStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryManagerOnTimelineStore.java
index dfc5b81..3c97584 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryManagerOnTimelineStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryManagerOnTimelineStore.java
@@ -80,9 +80,11 @@ public class TestApplicationHistoryManagerOnTimelineStore {
store = createStore(SCALE);
TimelineEntities entities = new TimelineEntities();
entities.addEntity(createApplicationTimelineEntity(
- ApplicationId.newInstance(0, SCALE + 1), true, true, false, false));
+ ApplicationId.newInstance(0, SCALE + 1), true, true, false, false,
+ YarnApplicationState.FINISHED));
entities.addEntity(createApplicationTimelineEntity(
- ApplicationId.newInstance(0, SCALE + 2), true, false, true, false));
+ ApplicationId.newInstance(0, SCALE + 2), true, false, true, false,
+ YarnApplicationState.FINISHED));
store.put(entities);
}
@@ -140,10 +142,10 @@ public class TestApplicationHistoryManagerOnTimelineStore {
ApplicationId appId = ApplicationId.newInstance(0, i);
if (i == 2) {
entities.addEntity(createApplicationTimelineEntity(
- appId, true, false, false, true));
+ appId, true, false, false, true, YarnApplicationState.FINISHED));
} else {
entities.addEntity(createApplicationTimelineEntity(
- appId, false, false, false, false));
+ appId, false, false, false, false, YarnApplicationState.FINISHED));
}
store.put(entities);
for (int j = 1; j <= scale; ++j) {
@@ -160,6 +162,16 @@ public class TestApplicationHistoryManagerOnTimelineStore {
}
}
}
+ TimelineEntities entities = new TimelineEntities();
+ ApplicationId appId = ApplicationId.newInstance(1234, 1);
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, 1);
+ ContainerId containerId = ContainerId.newContainerId(appAttemptId, 1);
+ entities.addEntity(createApplicationTimelineEntity(
+ appId, true, false, false, false, YarnApplicationState.RUNNING));
+ entities.addEntity(createAppAttemptTimelineEntity(appAttemptId));
+ entities.addEntity(createContainerEntity(containerId));
+ store.put(entities);
}
@Test
@@ -355,7 +367,7 @@ public class TestApplicationHistoryManagerOnTimelineStore {
historyManager.getApplications(Long.MAX_VALUE, 0L, Long.MAX_VALUE)
.values();
Assert.assertNotNull(apps);
- Assert.assertEquals(SCALE + 1, apps.size());
+ Assert.assertEquals(SCALE + 2, apps.size());
ApplicationId ignoredAppId = ApplicationId.newInstance(0, SCALE + 2);
for (ApplicationReport app : apps) {
Assert.assertNotEquals(ignoredAppId, app.getApplicationId());
@@ -467,7 +479,8 @@ public class TestApplicationHistoryManagerOnTimelineStore {
private static TimelineEntity createApplicationTimelineEntity(
ApplicationId appId, boolean emptyACLs, boolean noAttemptId,
- boolean wrongAppId, boolean enableUpdateEvent) {
+ boolean wrongAppId, boolean enableUpdateEvent,
+ YarnApplicationState state) {
TimelineEntity entity = new TimelineEntity();
entity.setEntityType(ApplicationMetricsConstants.ENTITY_TYPE);
if (wrongAppId) {
@@ -517,7 +530,7 @@ public class TestApplicationHistoryManagerOnTimelineStore {
eventInfo.put(ApplicationMetricsConstants.FINAL_STATUS_EVENT_INFO,
FinalApplicationStatus.UNDEFINED.toString());
eventInfo.put(ApplicationMetricsConstants.STATE_EVENT_INFO,
- YarnApplicationState.FINISHED.toString());
+ state.toString());
if (!noAttemptId) {
eventInfo.put(ApplicationMetricsConstants.LATEST_APP_ATTEMPT_EVENT_INFO,
ApplicationAttemptId.newInstance(appId, 1));
@@ -610,6 +623,8 @@ public class TestApplicationHistoryManagerOnTimelineStore {
entityInfo.put(ContainerMetricsConstants.ALLOCATED_PORT_ENTITY_INFO, 100);
entityInfo
.put(ContainerMetricsConstants.ALLOCATED_PRIORITY_ENTITY_INFO, -1);
+ entityInfo.put(ContainerMetricsConstants
+ .ALLOCATED_HOST_HTTP_ADDRESS_ENTITY_INFO, "http://test:1234");
entity.setOtherInfo(entityInfo);
TimelineEvent tEvent = new TimelineEvent();
tEvent.setEventType(ContainerMetricsConstants.CREATED_EVENT_TYPE);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e61d4312/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java
index 20dfe45..f985fe4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java
@@ -19,17 +19,30 @@
package org.apache.hadoop.yarn.server.applicationhistoryservice.webapp;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.Writer;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.net.URL;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.List;
import java.util.Properties;
import javax.servlet.FilterConfig;
import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.core.MediaType;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authentication.server.AuthenticationFilter;
import org.apache.hadoop.security.authentication.server.PseudoAuthenticationHandler;
import org.apache.hadoop.yarn.api.ApplicationBaseProtocol;
@@ -42,6 +55,8 @@ import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.YarnApplicationAttemptState;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat;
+import org.apache.hadoop.yarn.logaggregation.LogAggregationUtils;
import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryClientService;
import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryManagerOnTimelineStore;
import org.apache.hadoop.yarn.server.applicationhistoryservice.TestApplicationHistoryManagerOnTimelineStore;
@@ -81,12 +96,17 @@ import com.sun.jersey.test.framework.WebAppDescriptor;
public class TestAHSWebServices extends JerseyTestBase {
private static ApplicationHistoryClientService historyClientService;
+ private static AHSWebServices ahsWebservice;
private static final String[] USERS = new String[] { "foo" , "bar" };
private static final int MAX_APPS = 5;
+ private static Configuration conf;
+ private static FileSystem fs;
+ private static final String remoteLogRootDir = "target/logs/";
+ private static final String rootLogDir = "target/LocalLogs";
@BeforeClass
public static void setupClass() throws Exception {
- Configuration conf = new YarnConfiguration();
+ conf = new YarnConfiguration();
TimelineStore store =
TestApplicationHistoryManagerOnTimelineStore.createStore(MAX_APPS);
TimelineACLsManager aclsManager = new TimelineACLsManager(conf);
@@ -95,6 +115,8 @@ public class TestAHSWebServices extends JerseyTestBase {
new TimelineDataManager(store, aclsManager);
conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
conf.set(YarnConfiguration.YARN_ADMIN_ACL, "foo");
+ conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
+ conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteLogRootDir);
dataManager.init(conf);
ApplicationACLsManager appAclsManager = new ApplicationACLsManager(conf);
ApplicationHistoryManagerOnTimelineStore historyManager =
@@ -108,6 +130,8 @@ public class TestAHSWebServices extends JerseyTestBase {
};
historyClientService.init(conf);
historyClientService.start();
+ ahsWebservice = new AHSWebServices(historyClientService, conf);
+ fs = FileSystem.get(conf);
}
@AfterClass
@@ -115,6 +139,8 @@ public class TestAHSWebServices extends JerseyTestBase {
if (historyClientService != null) {
historyClientService.stop();
}
+ fs.delete(new Path(remoteLogRootDir), true);
+ fs.delete(new Path(rootLogDir), true);
}
@Parameterized.Parameters
@@ -127,7 +153,7 @@ public class TestAHSWebServices extends JerseyTestBase {
@Override
protected void configureServlets() {
bind(JAXBContextResolver.class);
- bind(AHSWebServices.class);
+ bind(AHSWebServices.class).toInstance(ahsWebservice);;
bind(GenericExceptionHandler.class);
bind(ApplicationBaseProtocol.class).toInstance(historyClientService);
serve("/*").with(GuiceContainer.class);
@@ -471,4 +497,177 @@ public class TestAHSWebServices extends JerseyTestBase {
assertEquals(ContainerState.COMPLETE.toString(),
container.getString("containerState"));
}
+
+ @Test(timeout = 10000)
+ public void testContainerLogsForFinishedApps() throws Exception {
+ String fileName = "syslog";
+ String user = "user1";
+ UserGroupInformation ugi = UserGroupInformation.createRemoteUser("user1");
+ NodeId nodeId = NodeId.newInstance("test host", 100);
+ NodeId nodeId2 = NodeId.newInstance("host2", 1234);
+ //prepare the logs for remote directory
+ ApplicationId appId = ApplicationId.newInstance(0, 1);
+ // create local logs
+ List<String> rootLogDirList = new ArrayList<String>();
+ rootLogDirList.add(rootLogDir);
+ Path rootLogDirPath = new Path(rootLogDir);
+ if (fs.exists(rootLogDirPath)) {
+ fs.delete(rootLogDirPath, true);
+ }
+ assertTrue(fs.mkdirs(rootLogDirPath));
+
+ Path appLogsDir = new Path(rootLogDirPath, appId.toString());
+ if (fs.exists(appLogsDir)) {
+ fs.delete(appLogsDir, true);
+ }
+ assertTrue(fs.mkdirs(appLogsDir));
+
+ // create container logs in local log file dir
+ // create two container log files. We can get containerInfo
+ // for container1 from AHS, but can not get such info for
+ // container100
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, 1);
+ ContainerId containerId1 = ContainerId.newContainerId(appAttemptId, 1);
+ ContainerId containerId100 = ContainerId.newContainerId(appAttemptId, 100);
+ createContainerLogInLocalDir(appLogsDir, containerId1, fs, fileName,
+ ("Hello." + containerId1));
+ createContainerLogInLocalDir(appLogsDir, containerId100, fs, fileName,
+ ("Hello." + containerId100));
+
+ // upload container logs to remote log dir
+ Path path = new Path(conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR) +
+ user + "/logs/" + appId.toString());
+ if (fs.exists(path)) {
+ fs.delete(path, true);
+ }
+ assertTrue(fs.mkdirs(path));
+ uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirList, nodeId,
+ containerId1, path, fs);
+ uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirList, nodeId2,
+ containerId100, path, fs);
+
+ // test whether we can find container log from remote diretory if
+ // the containerInfo for this container could be fetched from AHS.
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1")
+ .path("applicationhistory").path("containerlogs")
+ .path(containerId1.toString()).path(fileName)
+ .queryParam("user.name", user)
+ .accept(MediaType.TEXT_PLAIN)
+ .get(ClientResponse.class);
+ String responseText = response.getEntity(String.class);
+ assertTrue(responseText.contains("Hello." + containerId1));
+
+ // test whether we can find container log from remote diretory if
+ // the containerInfo for this container could not be fetched from AHS.
+ r = resource();
+ response = r.path("ws").path("v1")
+ .path("applicationhistory").path("containerlogs")
+ .path(containerId100.toString()).path(fileName)
+ .queryParam("user.name", user)
+ .accept(MediaType.TEXT_PLAIN)
+ .get(ClientResponse.class);
+ responseText = response.getEntity(String.class);
+ assertTrue(responseText.contains("Hello." + containerId100));
+
+ // create an application which can not be found from AHS
+ ApplicationId appId100 = ApplicationId.newInstance(0, 100);
+ appLogsDir = new Path(rootLogDirPath, appId100.toString());
+ if (fs.exists(appLogsDir)) {
+ fs.delete(appLogsDir, true);
+ }
+ assertTrue(fs.mkdirs(appLogsDir));
+ ApplicationAttemptId appAttemptId100 =
+ ApplicationAttemptId.newInstance(appId100, 1);
+ ContainerId containerId1ForApp100 = ContainerId
+ .newContainerId(appAttemptId100, 1);
+ createContainerLogInLocalDir(appLogsDir, containerId1ForApp100, fs,
+ fileName, ("Hello." + containerId1ForApp100));
+ path = new Path(conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR) +
+ user + "/logs/" + appId100.toString());
+ if (fs.exists(path)) {
+ fs.delete(path, true);
+ }
+ assertTrue(fs.mkdirs(path));
+ uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirList, nodeId2,
+ containerId1ForApp100, path, fs);
+ r = resource();
+ response = r.path("ws").path("v1")
+ .path("applicationhistory").path("containerlogs")
+ .path(containerId1ForApp100.toString()).path(fileName)
+ .queryParam("user.name", user)
+ .accept(MediaType.TEXT_PLAIN)
+ .get(ClientResponse.class);
+ responseText = response.getEntity(String.class);
+ assertTrue(responseText.contains("Hello." + containerId1ForApp100));
+ }
+
+ private static void createContainerLogInLocalDir(Path appLogsDir,
+ ContainerId containerId, FileSystem fs, String fileName, String content)
+ throws Exception {
+ Path containerLogsDir = new Path(appLogsDir, containerId.toString());
+ if (fs.exists(containerLogsDir)) {
+ fs.delete(containerLogsDir, true);
+ }
+ assertTrue(fs.mkdirs(containerLogsDir));
+ Writer writer =
+ new FileWriter(new File(containerLogsDir.toString(), fileName));
+ writer.write(content);
+ writer.close();
+ }
+
+ private static void uploadContainerLogIntoRemoteDir(UserGroupInformation ugi,
+ Configuration configuration, List<String> rootLogDirs, NodeId nodeId,
+ ContainerId containerId, Path appDir, FileSystem fs) throws Exception {
+ Path path =
+ new Path(appDir, LogAggregationUtils.getNodeString(nodeId));
+ AggregatedLogFormat.LogWriter writer =
+ new AggregatedLogFormat.LogWriter(configuration, path, ugi);
+ writer.writeApplicationOwner(ugi.getUserName());
+
+ writer.append(new AggregatedLogFormat.LogKey(containerId),
+ new AggregatedLogFormat.LogValue(rootLogDirs, containerId,
+ ugi.getShortUserName()));
+ writer.close();
+ }
+
+ @Test(timeout = 10000)
+ public void testContainerLogsForRunningApps() throws Exception {
+ String fileName = "syslog";
+ String user = "user1";
+ ApplicationId appId = ApplicationId.newInstance(
+ 1234, 1);
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, 1);
+ ContainerId containerId1 = ContainerId.newContainerId(appAttemptId, 1);
+ WebResource r = resource();
+ URI requestURI = r.path("ws").path("v1")
+ .path("applicationhistory").path("containerlogs")
+ .path(containerId1.toString()).path(fileName)
+ .queryParam("user.name", user).getURI();
+ String redirectURL = getRedirectURL(requestURI.toString());
+ assertTrue(redirectURL != null);
+ assertTrue(redirectURL.contains("test:1234"));
+ assertTrue(redirectURL.contains("ws/v1/node/containerlogs"));
+ assertTrue(redirectURL.contains(containerId1.toString()));
+ assertTrue(redirectURL.contains("user.name=" + user));
+ }
+
+ 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;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e61d4312/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/webapp/dao/ContainerInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/webapp/dao/ContainerInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/webapp/dao/ContainerInfo.java
index d0d4df6..f127f9c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/webapp/dao/ContainerInfo.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/webapp/dao/ContainerInfo.java
@@ -48,6 +48,7 @@ public class ContainerInfo {
protected int containerExitStatus;
protected ContainerState containerState;
protected String nodeHttpAddress;
+ protected String nodeId;
public ContainerInfo() {
// JAXB needs this
@@ -71,6 +72,7 @@ public class ContainerInfo {
containerExitStatus = container.getContainerExitStatus();
containerState = container.getContainerState();
nodeHttpAddress = container.getNodeHttpAddress();
+ nodeId = container.getAssignedNode().toString();
}
public String getContainerId() {
@@ -124,4 +126,8 @@ public class ContainerInfo {
public String getNodeHttpAddress() {
return nodeHttpAddress;
}
+
+ public String getNodeId() {
+ return nodeId;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e61d4312/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebServices.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebServices.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebServices.java
index fddeb04..57e729c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebServices.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebServices.java
@@ -33,6 +33,7 @@ import javax.ws.rs.QueryParam;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.ResponseBuilder;
import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.StreamingOutput;
import javax.ws.rs.core.UriInfo;
@@ -215,7 +216,8 @@ public class NMWebServices {
@Public
@Unstable
public Response getLogs(@PathParam("containerid") String containerIdStr,
- @PathParam("filename") String filename) {
+ @PathParam("filename") String filename,
+ @QueryParam("download") String download) {
ContainerId containerId;
try {
containerId = ConverterUtils.toContainerId(containerIdStr);
@@ -232,7 +234,7 @@ public class NMWebServices {
} catch (YarnException ex) {
return Response.serverError().entity(ex.getMessage()).build();
}
-
+ boolean downloadFile = parseBooleanParam(download);
try {
final FileInputStream fis = ContainerLogsUtils.openLogFileForRead(
containerIdStr, logFile, nmContext);
@@ -250,10 +252,22 @@ public class NMWebServices {
os.flush();
}
};
-
- return Response.ok(stream).build();
+ ResponseBuilder resp = Response.ok(stream);
+ if (downloadFile) {
+ resp.header("Content-Type", "application/octet-stream");
+ resp.header("Content-Disposition", "attachment; filename="
+ + logFile.getName());
+ }
+ return resp.build();
} catch (IOException ex) {
return Response.serverError().entity(ex.getMessage()).build();
}
}
+
+ private boolean parseBooleanParam(String param) {
+ if (param != null) {
+ return ("true").equalsIgnoreCase(param);
+ }
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e61d4312/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServices.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServices.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServices.java
index 2ac0956..ce1b309 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServices.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServices.java
@@ -50,7 +50,6 @@ import org.apache.hadoop.yarn.server.nodemanager.NodeHealthCheckerService;
import org.apache.hadoop.yarn.server.nodemanager.NodeManager;
import org.apache.hadoop.yarn.server.nodemanager.ResourceView;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationImpl;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
import org.apache.hadoop.yarn.server.nodemanager.webapp.WebServer.NMWebApp;
@@ -352,7 +351,16 @@ public class TestNMWebServices extends JerseyTestBase {
.accept(MediaType.TEXT_PLAIN).get(ClientResponse.class);
String responseText = response.getEntity(String.class);
assertEquals(logMessage, responseText);
-
+
+ // ask and download it
+ response = r.path("ws").path("v1").path("node").path("containerlogs")
+ .path(containerIdStr).path(filename).queryParam("download", "true")
+ .accept(MediaType.TEXT_PLAIN).get(ClientResponse.class);
+ responseText = response.getEntity(String.class);
+ assertEquals(logMessage, responseText);
+ assertEquals(200, response.getStatus());
+ assertEquals("application/octet-stream", response.getType().toString());
+
// ask for file that doesn't exist
response = r.path("ws").path("v1").path("node")
.path("containerlogs").path(containerIdStr).path("uhhh")
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org