Repository: tez Updated Branches: refs/heads/master a06cd76d8 -> 6930e4baf
TEZ-2097. TEZ-UI Add dag logs backend support (jeagles) Project: http://git-wip-us.apache.org/repos/asf/tez/repo Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/6930e4ba Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/6930e4ba Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/6930e4ba Branch: refs/heads/master Commit: 6930e4baf04660ef52130169b02c2b1ef34229d6 Parents: a06cd76 Author: Jonathan Eagles <[email protected]> Authored: Fri Sep 18 17:01:41 2015 -0500 Committer: Jonathan Eagles <[email protected]> Committed: Fri Sep 18 17:01:41 2015 -0500 ---------------------------------------------------------------------- CHANGES.txt | 3 +++ .../org/apache/tez/dag/app/DAGAppMaster.java | 28 +++++++++++++++++--- .../dag/history/events/DAGRecoveredEvent.java | 12 ++++++--- .../dag/history/events/DAGSubmittedEvent.java | 9 +++++-- .../apache/tez/dag/app/TestRecoveryParser.java | 8 +++--- .../TestHistoryEventsProtoConversion.java | 4 +-- .../impl/TestHistoryEventJsonConversion.java | 4 +-- .../ats/acls/TestATSHistoryWithACLs.java | 4 +-- .../ats/HistoryEventTimelineConversion.java | 4 +++ .../ats/TestHistoryEventTimelineConversion.java | 24 ++++++++++++----- 10 files changed, 74 insertions(+), 26 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index b4a5db4..f62b64e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -7,6 +7,7 @@ Release 0.8.1: Unreleased INCOMPATIBLE CHANGES ALL CHANGES: + TEZ-2097. TEZ-UI Add dag logs backend support TEZ-2812. Preemption sometimes does not respect heartbeats between preemptions TEZ-814. Improve heuristic for determining a task has failed outputs TEZ-2832. Support tests for both SimpleHistory logging and ATS logging @@ -185,6 +186,7 @@ Release 0.7.1: Unreleased INCOMPATIBLE CHANGES ALL CHANGES: + TEZ-2097. TEZ-UI Add dag logs backend support TEZ-2812. Preemption sometimes does not respect heartbeats between preemptions TEZ-814. Improve heuristic for determining a task has failed outputs TEZ-2840. MRInputLegacy.init should set splitInfoViaEvents. @@ -443,6 +445,7 @@ Release 0.6.3: Unreleased INCOMPATIBLE CHANGES ALL CHANGES: + TEZ-2097. TEZ-UI Add dag logs backend support TEZ-2812. Preemption sometimes does not respect heartbeats between preemptions TEZ-814. Improve heuristic for determining a task has failed outputs TEZ-2809. Minimal distribution compiled on 2.6 fails to run on 2.7 http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java b/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java index c713435..e41d59c 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java @@ -180,6 +180,7 @@ import org.slf4j.LoggerFactory; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Function; +import com.google.common.base.Joiner; import com.google.common.base.Preconditions; import com.google.common.collect.Maps; import com.google.common.util.concurrent.ListeningExecutorService; @@ -213,6 +214,7 @@ public class DAGAppMaster extends AbstractService { * Priority of the DAGAppMaster shutdown hook. */ public static final int SHUTDOWN_HOOK_PRIORITY = 30; + private static final Joiner PATH_JOINER = Joiner.on('/'); private static Pattern sanitizeLabelPattern = Pattern.compile("[:\\-\\W]+"); @@ -253,6 +255,7 @@ public class DAGAppMaster extends AbstractService { private final Map<String, LocalResource> cumulativeAdditionalResources = new HashMap<String, LocalResource>(); private final int maxAppAttempts; private final List<String> diagnostics = new ArrayList<String>(); + private String containerLogs; private boolean isLocal = false; //Local mode flag @@ -347,9 +350,24 @@ public class DAGAppMaster extends AbstractService { .createRemoteUser(jobUserName); this.appMasterUgi.addCredentials(amCredentials); + this.containerLogs = getRunningLogURL(this.nmHost + ":" + this.nmHttpPort, + this.containerID.toString(), this.appMasterUgi.getShortUserName()); + LOG.info("Created DAGAppMaster for application " + applicationAttemptId + ", versionInfo=" + dagVersionInfo.toString()); } + + // Pull this WebAppUtils function into Tez until YARN-4186 + public static String getRunningLogURL(String nodeHttpAddress, + String containerId, String user) { + if (nodeHttpAddress == null || nodeHttpAddress.isEmpty() + || containerId == null || containerId.isEmpty() || user == null + || user.isEmpty()) { + return null; + } + return PATH_JOINER.join(nodeHttpAddress, "node", "containerlogs", + containerId, user); + } private void initResourceCalculatorPlugins() { Class<? extends ResourceCalculatorProcessTree> clazz = amConf.getClass( @@ -1871,7 +1889,8 @@ public class DAGAppMaster extends AbstractService { DAGRecoveredEvent dagRecoveredEvent = new DAGRecoveredEvent(this.appAttemptID, recoveredDAGData.recoveredDAG.getID(), recoveredDAGData.recoveredDAG.getName(), recoveredDAGData.recoveredDAG.getUserName(), - this.clock.getTime(), DAGState.FAILED, recoveredDAGData.reason); + this.clock.getTime(), DAGState.FAILED, recoveredDAGData.reason, + this.containerLogs); dagRecoveredEvent.setHistoryLoggingEnabled( recoveredDAGData.recoveredDAG.getConf().getBoolean( TezConfiguration.TEZ_DAG_HISTORY_LOGGING_ENABLED, @@ -1887,7 +1906,7 @@ public class DAGAppMaster extends AbstractService { DAGRecoveredEvent dagRecoveredEvent = new DAGRecoveredEvent(this.appAttemptID, recoveredDAGData.recoveredDAG.getID(), recoveredDAGData.recoveredDAG.getName(), recoveredDAGData.recoveredDAG.getUserName(), this.clock.getTime(), - recoveredDAGData.dagState, null); + recoveredDAGData.dagState, null, this.containerLogs); this.historyEventHandler.handle(new DAGHistoryEvent(recoveredDAGData.recoveredDAG.getID(), dagRecoveredEvent)); dagEventDispatcher.handle(recoverDAGEvent); @@ -1898,7 +1917,7 @@ public class DAGAppMaster extends AbstractService { _updateLoggers(recoveredDAGData.recoveredDAG, ""); DAGRecoveredEvent dagRecoveredEvent = new DAGRecoveredEvent(this.appAttemptID, recoveredDAGData.recoveredDAG.getID(), recoveredDAGData.recoveredDAG.getName(), - recoveredDAGData.recoveredDAG.getUserName(), this.clock.getTime()); + recoveredDAGData.recoveredDAG.getUserName(), this.clock.getTime(), this.containerLogs); this.historyEventHandler.handle(new DAGHistoryEvent(recoveredDAGData.recoveredDAG.getID(), dagRecoveredEvent)); DAGEventRecoverEvent recoverDAGEvent = new DAGEventRecoverEvent( @@ -2288,11 +2307,12 @@ public class DAGAppMaster extends AbstractService { String timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Calendar.getInstance().getTime()); System.err.println(timeStamp + " Running Dag: " + newDAG.getID()); System.out.println(timeStamp + " Running Dag: "+ newDAG.getID()); + // Job name is the same as the app name until we support multiple dags // for an app later DAGSubmittedEvent submittedEvent = new DAGSubmittedEvent(newDAG.getID(), submitTime, dagPlan, this.appAttemptID, cumulativeAdditionalResources, - newDAG.getUserName(), newDAG.getConf()); + newDAG.getUserName(), newDAG.getConf(), containerLogs); boolean dagLoggingEnabled = newDAG.getConf().getBoolean( TezConfiguration.TEZ_DAG_HISTORY_LOGGING_ENABLED, TezConfiguration.TEZ_DAG_HISTORY_LOGGING_ENABLED_DEFAULT); http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGRecoveredEvent.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGRecoveredEvent.java b/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGRecoveredEvent.java index 7109756..2bfa43b 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGRecoveredEvent.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGRecoveredEvent.java @@ -39,11 +39,12 @@ public class DAGRecoveredEvent implements HistoryEvent { private final String user; private boolean historyLoggingEnabled = true; + private String containerLogs; public DAGRecoveredEvent(ApplicationAttemptId applicationAttemptId, TezDAGID dagId, String dagName, String user, long recoveredTime, DAGState recoveredState, - String recoveryFailureReason) { + String recoveryFailureReason, String containerLogs) { this.applicationAttemptId = applicationAttemptId; this.dagID = dagId; this.dagName = dagName; @@ -51,11 +52,12 @@ public class DAGRecoveredEvent implements HistoryEvent { this.recoveredTime = recoveredTime; this.recoveredDagState = recoveredState; this.recoveryFailureReason = recoveryFailureReason; + this.containerLogs = containerLogs; } public DAGRecoveredEvent(ApplicationAttemptId applicationAttemptId, - TezDAGID dagId, String dagName, String user, long recoveredTime) { - this(applicationAttemptId, dagId, dagName, user, recoveredTime, null, null); + TezDAGID dagId, String dagName, String user, long recoveredTime, String containerLogs) { + this(applicationAttemptId, dagId, dagName, user, recoveredTime, null, null, containerLogs); } @Override @@ -121,6 +123,10 @@ public class DAGRecoveredEvent implements HistoryEvent { this.historyLoggingEnabled = historyLoggingEnabled; } + public String getContainerLogs() { + return containerLogs; + } + @Override public String toString() { return "applicationAttemptId=" http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGSubmittedEvent.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGSubmittedEvent.java b/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGSubmittedEvent.java index 978fd0c..07d7c07 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGSubmittedEvent.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/history/events/DAGSubmittedEvent.java @@ -55,8 +55,8 @@ public class DAGSubmittedEvent implements HistoryEvent, SummaryEvent { private String user; private Map<String, LocalResource> cumulativeAdditionalLocalResources; private boolean historyLoggingEnabled = true; - private Configuration conf; + private String containerLogs; public DAGSubmittedEvent() { } @@ -64,7 +64,7 @@ public class DAGSubmittedEvent implements HistoryEvent, SummaryEvent { public DAGSubmittedEvent(TezDAGID dagID, long submitTime, DAGProtos.DAGPlan dagPlan, ApplicationAttemptId applicationAttemptId, Map<String, LocalResource> cumulativeAdditionalLocalResources, - String user, Configuration conf) { + String user, Configuration conf, String containerLogs) { this.dagID = dagID; this.dagName = dagPlan.getName(); this.submitTime = submitTime; @@ -73,6 +73,7 @@ public class DAGSubmittedEvent implements HistoryEvent, SummaryEvent { this.cumulativeAdditionalLocalResources = cumulativeAdditionalLocalResources; this.user = user; this.conf = conf; + this.containerLogs = containerLogs; } @Override @@ -198,4 +199,8 @@ public class DAGSubmittedEvent implements HistoryEvent, SummaryEvent { public boolean isHistoryLoggingEnabled() { return historyLoggingEnabled; } + + public String getContainerLogs() { + return containerLogs; + } } http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-dag/src/test/java/org/apache/tez/dag/app/TestRecoveryParser.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/TestRecoveryParser.java b/tez-dag/src/test/java/org/apache/tez/dag/app/TestRecoveryParser.java index 4bb0615..e261df6 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/app/TestRecoveryParser.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/app/TestRecoveryParser.java @@ -139,7 +139,7 @@ public class TestRecoveryParser { rService.start(); rService.handle(new DAGHistoryEvent(dagID, new DAGSubmittedEvent(dagID, 1L, dagPlan, ApplicationAttemptId.newInstance(appId, 1), - null, "user", new Configuration()))); + null, "user", new Configuration(), null))); rService.handle(new DAGHistoryEvent(dagID, new DAGInitializedEvent(dagID, 1L, "user", dagPlan.getName(), null))); // only for testing, DAGCommitStartedEvent is not supposed to happen at this time. @@ -185,7 +185,7 @@ public class TestRecoveryParser { rService.start(); rService.handle(new DAGHistoryEvent(dagID, new DAGSubmittedEvent(dagID, 1L, dagPlan, ApplicationAttemptId.newInstance(appId, 1), - null, "user", new Configuration()))); + null, "user", new Configuration(), null))); rService.handle(new DAGHistoryEvent(dagID, new DAGInitializedEvent(dagID, 1L, "user", dagPlan.getName(), null))); rService.handle(new DAGHistoryEvent(dagID, @@ -233,7 +233,7 @@ public class TestRecoveryParser { rService.start(); rService.handle(new DAGHistoryEvent(dagID, new DAGSubmittedEvent(dagID, 1L, dagPlan, ApplicationAttemptId.newInstance(appId, 1), - null, "user", new Configuration()))); + null, "user", new Configuration(), null))); // wait until DAGSubmittedEvent is handled in the RecoveryEventHandling thread rService.await(); rService.outputStreamMap.get(dagID).writeUTF("INVALID_DATA"); @@ -279,7 +279,7 @@ public class TestRecoveryParser { // write a DAGSubmittedEvent first to initialize summaryStream rService.handle(new DAGHistoryEvent(dagID, new DAGSubmittedEvent(dagID, 1L, dagPlan, ApplicationAttemptId.newInstance(appId, 1), - null, "user", new Configuration()))); + null, "user", new Configuration(), null))); // write an corrupted SummaryEvent rService.summaryStream.writeChars("INVALID_DATA"); rService.stop(); http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java b/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java index 5a5cc53..b0b76d4 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java @@ -175,7 +175,7 @@ public class TestHistoryEventsProtoConversion { ApplicationId.newInstance(0, 1), 1), 1001l, DAGPlan.newBuilder().setName("foo").build(), ApplicationAttemptId.newInstance( - ApplicationId.newInstance(0, 1), 1), null, "", null); + ApplicationId.newInstance(0, 1), 1), null, "", null, null); DAGSubmittedEvent deserializedEvent = (DAGSubmittedEvent) testProtoConversion(event); Assert.assertEquals(event.getApplicationAttemptId(), @@ -768,7 +768,7 @@ public class TestHistoryEventsProtoConversion { DAGRecoveredEvent dagRecoveredEvent = new DAGRecoveredEvent( ApplicationAttemptId.newInstance(ApplicationId.newInstance(0, 1), 1), TezDAGID.getInstance(ApplicationId.newInstance(0, 1), 1), - "mockDagname", "mockuser", 100334l); + "mockDagname", "mockuser", 100334l, null); try { testProtoConversion(dagRecoveredEvent); Assert.fail("Proto conversion should have failed"); http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java b/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java index 003a04f..cfe9618 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java @@ -124,7 +124,7 @@ public class TestHistoryEventJsonConversion { break; case DAG_SUBMITTED: event = new DAGSubmittedEvent(tezDAGID, random.nextInt(), dagPlan, applicationAttemptId, - null, user, null); + null, user, null, null); break; case DAG_INITIALIZED: event = new DAGInitializedEvent(tezDAGID, random.nextInt(), user, dagPlan.getName(), null); @@ -190,7 +190,7 @@ public class TestHistoryEventJsonConversion { break; case DAG_RECOVERED: event = new DAGRecoveredEvent(applicationAttemptId, tezDAGID, dagPlan.getName(), user, - 1l); + 1l, null); break; case DAG_KILL_REQUEST: event = new DAGKillRequestEvent(); http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-plugins/tez-yarn-timeline-history-with-acls/src/test/java/org/apache/tez/dag/history/ats/acls/TestATSHistoryWithACLs.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-yarn-timeline-history-with-acls/src/test/java/org/apache/tez/dag/history/ats/acls/TestATSHistoryWithACLs.java b/tez-plugins/tez-yarn-timeline-history-with-acls/src/test/java/org/apache/tez/dag/history/ats/acls/TestATSHistoryWithACLs.java index fc35971..eaf24d3 100644 --- a/tez-plugins/tez-yarn-timeline-history-with-acls/src/test/java/org/apache/tez/dag/history/ats/acls/TestATSHistoryWithACLs.java +++ b/tez-plugins/tez-yarn-timeline-history-with-acls/src/test/java/org/apache/tez/dag/history/ats/acls/TestATSHistoryWithACLs.java @@ -574,7 +574,7 @@ public class TestATSHistoryWithACLs { DAGPlan dagPlan = DAGPlan.newBuilder().setName("DAGPlanMock").build(); DAGSubmittedEvent submittedEvent = new DAGSubmittedEvent(tezDAGID, 1, dagPlan, appAttemptId, null, - "usr", tezConf); + "usr", tezConf, null); submittedEvent.setHistoryLoggingEnabled(false); DAGHistoryEvent event = new DAGHistoryEvent(tezDAGID, submittedEvent); historyLoggingService.handle(new DAGHistoryEvent(tezDAGID, submittedEvent)); @@ -616,7 +616,7 @@ public class TestATSHistoryWithACLs { DAGPlan dagPlan = DAGPlan.newBuilder().setName("DAGPlanMock").build(); DAGSubmittedEvent submittedEvent = new DAGSubmittedEvent(tezDAGID, 1, dagPlan, appAttemptId, null, - "usr", tezConf); + "usr", tezConf, null); submittedEvent.setHistoryLoggingEnabled(true); DAGHistoryEvent event = new DAGHistoryEvent(tezDAGID, submittedEvent); historyLoggingService.handle(new DAGHistoryEvent(tezDAGID, submittedEvent)); http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-plugins/tez-yarn-timeline-history/src/main/java/org/apache/tez/dag/history/logging/ats/HistoryEventTimelineConversion.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-yarn-timeline-history/src/main/java/org/apache/tez/dag/history/logging/ats/HistoryEventTimelineConversion.java b/tez-plugins/tez-yarn-timeline-history/src/main/java/org/apache/tez/dag/history/logging/ats/HistoryEventTimelineConversion.java index c5ad816..77b00c4 100644 --- a/tez-plugins/tez-yarn-timeline-history/src/main/java/org/apache/tez/dag/history/logging/ats/HistoryEventTimelineConversion.java +++ b/tez-plugins/tez-yarn-timeline-history/src/main/java/org/apache/tez/dag/history/logging/ats/HistoryEventTimelineConversion.java @@ -159,6 +159,8 @@ public class HistoryEventTimelineConversion { atsEntity.addPrimaryFilter(ATSConstants.APPLICATION_ID, event.getApplicationAttemptId().getApplicationId().toString()); atsEntity.addPrimaryFilter(ATSConstants.DAG_NAME, event.getDagName()); + atsEntity.addOtherInfo(ATSConstants.IN_PROGRESS_LOGS_URL + "_" + + event.getApplicationAttemptId().getAttemptId(), event.getContainerLogs()); return atsEntity; } @@ -401,6 +403,8 @@ public class HistoryEventTimelineConversion { event.getApplicationAttemptId().toString()); atsEntity.addOtherInfo(ATSConstants.USER, event.getUser()); atsEntity.addOtherInfo(ATSConstants.DAG_AM_WEB_SERVICE_VERSION, AMWebController.VERSION); + atsEntity.addOtherInfo(ATSConstants.IN_PROGRESS_LOGS_URL + "_" + + event.getApplicationAttemptId().getAttemptId(), event.getContainerLogs()); return atsEntity; } http://git-wip-us.apache.org/repos/asf/tez/blob/6930e4ba/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java b/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java index 6e9f737..4245be3 100644 --- a/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java +++ b/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java @@ -102,6 +102,7 @@ public class TestHistoryEventTimelineConversion { private DAGPlan dagPlan; private ContainerId containerId; private NodeId nodeId; + private String containerLogs = "containerLogs"; @Before public void setup() { @@ -134,7 +135,7 @@ public class TestHistoryEventTimelineConversion { break; case DAG_SUBMITTED: event = new DAGSubmittedEvent(tezDAGID, random.nextInt(), dagPlan, applicationAttemptId, - null, user, null); + null, user, null, containerLogs); break; case DAG_INITIALIZED: event = new DAGInitializedEvent(tezDAGID, random.nextInt(), user, dagPlan.getName(), null); @@ -200,7 +201,7 @@ public class TestHistoryEventTimelineConversion { break; case DAG_RECOVERED: event = new DAGRecoveredEvent(applicationAttemptId, tezDAGID, dagPlan.getName(), - user, random.nextLong()); + user, random.nextLong(), containerLogs); break; case DAG_KILL_REQUEST: event = new DAGKillRequestEvent(); @@ -403,7 +404,7 @@ public class TestHistoryEventTimelineConversion { long submitTime = random.nextLong(); DAGSubmittedEvent event = new DAGSubmittedEvent(tezDAGID, submitTime, dagPlan, - applicationAttemptId, null, user, null); + applicationAttemptId, null, user, null, containerLogs); TimelineEntity timelineEntity = HistoryEventTimelineConversion.convertToTimelineEntity(event); Assert.assertEquals(EntityTypes.TEZ_DAG_ID.name(), timelineEntity.getEntityType()); @@ -435,7 +436,7 @@ public class TestHistoryEventTimelineConversion { Assert.assertTrue( timelineEntity.getPrimaryFilters().get(ATSConstants.USER).contains(user)); - Assert.assertEquals(5, timelineEntity.getOtherInfo().size()); + Assert.assertEquals(6, timelineEntity.getOtherInfo().size()); Assert.assertTrue(timelineEntity.getOtherInfo().containsKey(ATSConstants.DAG_PLAN)); Assert.assertEquals(applicationId.toString(), timelineEntity.getOtherInfo().get(ATSConstants.APPLICATION_ID)); @@ -447,6 +448,9 @@ public class TestHistoryEventTimelineConversion { timelineEntity.getOtherInfo().get(ATSConstants.DAG_AM_WEB_SERVICE_VERSION)); Assert.assertEquals(user, timelineEntity.getOtherInfo().get(ATSConstants.USER)); + Assert.assertEquals(containerLogs, + timelineEntity.getOtherInfo().get(ATSConstants.IN_PROGRESS_LOGS_URL + "_" + + applicationAttemptId.getAttemptId())); } @SuppressWarnings("unchecked") @@ -893,7 +897,7 @@ public class TestHistoryEventTimelineConversion { long recoverTime = random.nextLong(); DAGRecoveredEvent event = new DAGRecoveredEvent(applicationAttemptId, tezDAGID, - dagPlan.getName(), user, recoverTime); + dagPlan.getName(), user, recoverTime, containerLogs); TimelineEntity timelineEntity = HistoryEventTimelineConversion.convertToTimelineEntity(event); Assert.assertEquals(EntityTypes.TEZ_DAG_ID.name(), timelineEntity.getEntityType()); @@ -918,6 +922,9 @@ public class TestHistoryEventTimelineConversion { timelineEntity.getPrimaryFilters().get(ATSConstants.DAG_NAME).contains("DAGPlanMock")); Assert.assertTrue( timelineEntity.getPrimaryFilters().get(ATSConstants.USER).contains(user)); + Assert.assertEquals(containerLogs, + timelineEntity.getOtherInfo().get(ATSConstants.IN_PROGRESS_LOGS_URL + "_" + + applicationAttemptId.getAttemptId())); } @Test(timeout = 5000) @@ -925,7 +932,7 @@ public class TestHistoryEventTimelineConversion { long recoverTime = random.nextLong(); DAGRecoveredEvent event = new DAGRecoveredEvent(applicationAttemptId, tezDAGID, - dagPlan.getName(), user, recoverTime, DAGState.ERROR, "mock reason"); + dagPlan.getName(), user, recoverTime, DAGState.ERROR, "mock reason", containerLogs); TimelineEntity timelineEntity = HistoryEventTimelineConversion.convertToTimelineEntity(event); @@ -955,7 +962,10 @@ public class TestHistoryEventTimelineConversion { timelineEntity.getPrimaryFilters().get(ATSConstants.DAG_NAME).contains("DAGPlanMock")); Assert.assertTrue( timelineEntity.getPrimaryFilters().get(ATSConstants.USER).contains(user)); + Assert.assertEquals(containerLogs, + timelineEntity.getOtherInfo().get(ATSConstants.IN_PROGRESS_LOGS_URL + "_" + + applicationAttemptId.getAttemptId())); } -} \ No newline at end of file +}
