| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * <p> |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * <p> |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.hadoop.yarn.server.applicationhistoryservice; |
| |
| import java.security.PrivilegedExceptionAction; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import org.junit.jupiter.api.AfterEach; |
| import org.junit.jupiter.api.BeforeAll; |
| import org.junit.jupiter.params.ParameterizedTest; |
| import org.junit.jupiter.params.provider.MethodSource; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.security.SaslRpcServer.AuthMethod; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.hadoop.security.authorize.AuthorizationException; |
| import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; |
| import org.apache.hadoop.yarn.api.records.ApplicationAttemptReport; |
| 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.ContainerId; |
| import org.apache.hadoop.yarn.api.records.ContainerReport; |
| import org.apache.hadoop.yarn.api.records.ContainerState; |
| import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; |
| import org.apache.hadoop.yarn.api.records.NodeId; |
| import org.apache.hadoop.yarn.api.records.Priority; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.YarnApplicationAttemptState; |
| import org.apache.hadoop.yarn.api.records.YarnApplicationState; |
| import org.apache.hadoop.yarn.api.records.timeline.TimelineEntities; |
| import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity; |
| import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.server.metrics.AppAttemptMetricsConstants; |
| import org.apache.hadoop.yarn.server.metrics.ApplicationMetricsConstants; |
| import org.apache.hadoop.yarn.server.metrics.ContainerMetricsConstants; |
| import org.apache.hadoop.yarn.server.security.ApplicationACLsManager; |
| import org.apache.hadoop.yarn.server.timeline.MemoryTimelineStore; |
| import org.apache.hadoop.yarn.server.timeline.TimelineDataManager; |
| import org.apache.hadoop.yarn.server.timeline.TimelineStore; |
| import org.apache.hadoop.yarn.server.timeline.security.TimelineACLsManager; |
| |
| import static org.junit.jupiter.api.Assertions.assertEquals; |
| import static org.junit.jupiter.api.Assertions.assertNotEquals; |
| import static org.junit.jupiter.api.Assertions.assertNotNull; |
| import static org.junit.jupiter.api.Assertions.assertTrue; |
| import static org.junit.jupiter.api.Assertions.fail; |
| |
| public class TestApplicationHistoryManagerOnTimelineStore { |
| |
| private static final int SCALE = 5; |
| private static TimelineStore store; |
| |
| private ApplicationHistoryManagerOnTimelineStore historyManager; |
| private UserGroupInformation callerUGI; |
| private Configuration conf; |
| |
| @BeforeAll |
| public static void prepareStore() throws Exception { |
| store = createStore(SCALE); |
| TimelineEntities entities = new TimelineEntities(); |
| entities.addEntity( |
| createApplicationTimelineEntity(ApplicationId.newInstance(0, SCALE + 1), true, true, false, |
| false, YarnApplicationState.FINISHED)); |
| entities.addEntity( |
| createApplicationTimelineEntity(ApplicationId.newInstance(0, SCALE + 2), true, false, true, |
| false, YarnApplicationState.FINISHED)); |
| store.put(entities); |
| } |
| |
| public static TimelineStore createStore(int scale) throws Exception { |
| store = new MemoryTimelineStore(); |
| prepareTimelineStore(scale); |
| return store; |
| } |
| |
| @AfterEach |
| public void tearDown() { |
| if (historyManager != null) { |
| historyManager.stop(); |
| } |
| } |
| |
| public static Collection<Object[]> callers() { |
| // user1 is the owner |
| // user2 is the authorized user |
| // user3 is the unauthorized user |
| // admin is the admin acl |
| return Arrays.asList(new Object[][] {{""}, {"user1"}, {"user2"}, {"user3"}, {"admin"}}); |
| } |
| |
| public void initTestApplicationHistoryManagerOnTimelineStore(String caller) { |
| conf = new YarnConfiguration(); |
| if (!caller.equals("")) { |
| callerUGI = UserGroupInformation.createRemoteUser(caller, AuthMethod.SIMPLE); |
| conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true); |
| conf.set(YarnConfiguration.YARN_ADMIN_ACL, "admin"); |
| } |
| TimelineACLsManager aclsManager = new TimelineACLsManager(new YarnConfiguration()); |
| aclsManager.setTimelineStore(store); |
| TimelineDataManager dataManager = new TimelineDataManager(store, aclsManager); |
| dataManager.init(conf); |
| ApplicationACLsManager appAclsManager = new ApplicationACLsManager(conf); |
| historyManager = new ApplicationHistoryManagerOnTimelineStore(dataManager, appAclsManager); |
| historyManager.init(conf); |
| historyManager.start(); |
| } |
| |
| private static void prepareTimelineStore(int scale) throws Exception { |
| for (int i = 1; i <= scale; ++i) { |
| TimelineEntities entities = new TimelineEntities(); |
| ApplicationId appId = ApplicationId.newInstance(0, i); |
| if (i == 2) { |
| entities.addEntity(createApplicationTimelineEntity(appId, true, false, false, true, |
| YarnApplicationState.FINISHED)); |
| } else if (i == 3) { |
| entities.addEntity(createApplicationTimelineEntity(appId, false, false, false, false, |
| YarnApplicationState.FINISHED, true, false)); |
| } else if (i == SCALE + 1) { |
| entities.addEntity(createApplicationTimelineEntity(appId, false, false, false, false, |
| YarnApplicationState.FINISHED, false, true)); |
| } else { |
| entities.addEntity(createApplicationTimelineEntity(appId, false, false, false, false, |
| YarnApplicationState.FINISHED)); |
| } |
| store.put(entities); |
| for (int j = 1; j <= scale; ++j) { |
| entities = new TimelineEntities(); |
| ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId, j); |
| entities.addEntity(createAppAttemptTimelineEntity(appAttemptId)); |
| store.put(entities); |
| for (int k = 1; k <= scale; ++k) { |
| entities = new TimelineEntities(); |
| ContainerId containerId = ContainerId.newContainerId(appAttemptId, k); |
| entities.addEntity(createContainerEntity(containerId)); |
| store.put(entities); |
| } |
| } |
| } |
| 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); |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetApplicationReport(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| for (int i = 1; i <= 3; ++i) { |
| final ApplicationId appId = ApplicationId.newInstance(0, i); |
| ApplicationReport app; |
| if (callerUGI == null) { |
| app = historyManager.getApplication(appId); |
| } else { |
| app = callerUGI.doAs(new PrivilegedExceptionAction<ApplicationReport>() { |
| @Override |
| public ApplicationReport run() throws Exception { |
| return historyManager.getApplication(appId); |
| } |
| }); |
| } |
| assertNotNull(app); |
| assertEquals(appId, app.getApplicationId()); |
| assertEquals("test app", app.getName()); |
| assertEquals("test app type", app.getApplicationType()); |
| assertEquals("user1", app.getUser()); |
| if (i == 2) { |
| // Change event is fired only in case of app with ID 2, hence verify |
| // with updated changes. And make sure last updated change is accepted. |
| assertEquals("changed queue1", app.getQueue()); |
| assertEquals(Priority.newInstance(6), app.getPriority()); |
| } else { |
| assertEquals("test queue", app.getQueue()); |
| assertEquals(Priority.newInstance(0), app.getPriority()); |
| } |
| assertEquals(Integer.MAX_VALUE + 2L + app.getApplicationId().getId(), app.getStartTime()); |
| assertEquals(Integer.MAX_VALUE + 1L, app.getSubmitTime()); |
| assertEquals(Integer.MAX_VALUE + 3L + +app.getApplicationId().getId(), app.getFinishTime()); |
| assertTrue(Math.abs(app.getProgress() - 1.0F) < 0.0001); |
| assertEquals(2, app.getApplicationTags().size()); |
| assertTrue(app.getApplicationTags().contains("Test_APP_TAGS_1")); |
| assertTrue(app.getApplicationTags().contains("Test_APP_TAGS_2")); |
| // App 2 doesn't have the ACLs, such that the default ACLs " " will be used. |
| // Nobody except admin and owner has access to the details of the app. |
| if ((i != 2 && callerUGI != null && callerUGI.getShortUserName().equals("user3")) || (i == 2 |
| && callerUGI != null && (callerUGI.getShortUserName().equals("user2") |
| || callerUGI.getShortUserName().equals("user3")))) { |
| assertEquals(ApplicationAttemptId.newInstance(appId, -1), |
| app.getCurrentApplicationAttemptId()); |
| assertEquals(ApplicationHistoryManagerOnTimelineStore.UNAVAILABLE, app.getHost()); |
| assertEquals(-1, app.getRpcPort()); |
| assertEquals(ApplicationHistoryManagerOnTimelineStore.UNAVAILABLE, app.getTrackingUrl()); |
| assertEquals(ApplicationHistoryManagerOnTimelineStore.UNAVAILABLE, |
| app.getOriginalTrackingUrl()); |
| assertEquals("", app.getDiagnostics()); |
| } else { |
| assertEquals(ApplicationAttemptId.newInstance(appId, 1), |
| app.getCurrentApplicationAttemptId()); |
| assertEquals("test host", app.getHost()); |
| assertEquals(100, app.getRpcPort()); |
| assertEquals("test tracking url", app.getTrackingUrl()); |
| assertEquals("test original tracking url", app.getOriginalTrackingUrl()); |
| assertEquals("test diagnostics info", app.getDiagnostics()); |
| } |
| ApplicationResourceUsageReport applicationResourceUsageReport = |
| app.getApplicationResourceUsageReport(); |
| assertEquals(123, applicationResourceUsageReport.getMemorySeconds()); |
| assertEquals(345, applicationResourceUsageReport.getVcoreSeconds()); |
| long expectedPreemptMemSecs = 456; |
| long expectedPreemptVcoreSecs = 789; |
| if (i == 3) { |
| expectedPreemptMemSecs = 0; |
| expectedPreemptVcoreSecs = 0; |
| } |
| assertEquals(expectedPreemptMemSecs, |
| applicationResourceUsageReport.getPreemptedMemorySeconds()); |
| assertEquals(expectedPreemptVcoreSecs, |
| applicationResourceUsageReport.getPreemptedVcoreSeconds()); |
| assertEquals(FinalApplicationStatus.UNDEFINED, app.getFinalApplicationStatus()); |
| assertEquals(YarnApplicationState.FINISHED, app.getYarnApplicationState()); |
| } |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetApplicationReportWithNotAttempt(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| final ApplicationId appId = ApplicationId.newInstance(0, SCALE + 1); |
| ApplicationReport app; |
| if (callerUGI == null) { |
| app = historyManager.getApplication(appId); |
| } else { |
| app = callerUGI.doAs(new PrivilegedExceptionAction<ApplicationReport>() { |
| @Override |
| public ApplicationReport run() throws Exception { |
| return historyManager.getApplication(appId); |
| } |
| }); |
| } |
| assertNotNull(app); |
| assertEquals(appId, app.getApplicationId()); |
| assertEquals(ApplicationAttemptId.newInstance(appId, -1), app.getCurrentApplicationAttemptId()); |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetApplicationAttemptReport(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| final ApplicationAttemptId appAttemptId = |
| ApplicationAttemptId.newInstance(ApplicationId.newInstance(0, 1), 1); |
| ApplicationAttemptReport appAttempt; |
| if (callerUGI == null) { |
| appAttempt = historyManager.getApplicationAttempt(appAttemptId); |
| } else { |
| try { |
| appAttempt = callerUGI.doAs(new PrivilegedExceptionAction<ApplicationAttemptReport>() { |
| @Override |
| public ApplicationAttemptReport run() throws Exception { |
| return historyManager.getApplicationAttempt(appAttemptId); |
| } |
| }); |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| fail(); |
| } |
| } catch (AuthorizationException e) { |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| return; |
| } |
| throw e; |
| } |
| } |
| assertNotNull(appAttempt); |
| assertEquals(appAttemptId, appAttempt.getApplicationAttemptId()); |
| assertEquals(ContainerId.newContainerId(appAttemptId, 1), appAttempt.getAMContainerId()); |
| assertEquals("test host", appAttempt.getHost()); |
| assertEquals(100, appAttempt.getRpcPort()); |
| assertEquals("test tracking url", appAttempt.getTrackingUrl()); |
| assertEquals("test original tracking url", appAttempt.getOriginalTrackingUrl()); |
| assertEquals("test diagnostics info", appAttempt.getDiagnostics()); |
| assertEquals(YarnApplicationAttemptState.FINISHED, appAttempt.getYarnApplicationAttemptState()); |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetContainerReport(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| final ContainerId containerId = ContainerId.newContainerId( |
| ApplicationAttemptId.newInstance(ApplicationId.newInstance(0, 1), 1), 1); |
| ContainerReport container; |
| if (callerUGI == null) { |
| container = historyManager.getContainer(containerId); |
| } else { |
| try { |
| container = callerUGI.doAs(new PrivilegedExceptionAction<ContainerReport>() { |
| @Override |
| public ContainerReport run() throws Exception { |
| return historyManager.getContainer(containerId); |
| } |
| }); |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| fail(); |
| } |
| } catch (AuthorizationException e) { |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| return; |
| } |
| throw e; |
| } |
| } |
| assertNotNull(container); |
| assertEquals(Integer.MAX_VALUE + 1L, container.getCreationTime()); |
| assertEquals(Integer.MAX_VALUE + 2L, container.getFinishTime()); |
| assertEquals(Resource.newInstance(-1, -1), container.getAllocatedResource()); |
| assertEquals(NodeId.newInstance("test host", 100), container.getAssignedNode()); |
| assertEquals(Priority.UNDEFINED, container.getPriority()); |
| assertEquals("test diagnostics info", container.getDiagnosticsInfo()); |
| assertEquals(ContainerState.COMPLETE, container.getContainerState()); |
| assertEquals(-1, container.getContainerExitStatus()); |
| assertEquals( |
| "http://0.0.0.0:8188/applicationhistory/logs/" + "test host:100/container_0_0001_01_000001/" |
| + "container_0_0001_01_000001/user1", container.getLogUrl()); |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetApplications(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| Collection<ApplicationReport> apps = |
| historyManager.getApplications(Long.MAX_VALUE, 0L, Long.MAX_VALUE).values(); |
| assertNotNull(apps); |
| assertEquals(SCALE + 2, apps.size()); |
| ApplicationId ignoredAppId = ApplicationId.newInstance(0, SCALE + 2); |
| for (ApplicationReport app : apps) { |
| assertNotEquals(ignoredAppId, app.getApplicationId()); |
| } |
| |
| // Get apps by given appStartedTime period |
| apps = historyManager.getApplications(Long.MAX_VALUE, 2147483653L, Long.MAX_VALUE).values(); |
| assertNotNull(apps); |
| assertEquals(2, apps.size()); |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetApplicationAttempts(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| final ApplicationId appId = ApplicationId.newInstance(0, 1); |
| Collection<ApplicationAttemptReport> appAttempts; |
| if (callerUGI == null) { |
| appAttempts = historyManager.getApplicationAttempts(appId).values(); |
| } else { |
| try { |
| appAttempts = |
| callerUGI.doAs(new PrivilegedExceptionAction<Collection<ApplicationAttemptReport>>() { |
| @Override |
| public Collection<ApplicationAttemptReport> run() throws Exception { |
| return historyManager.getApplicationAttempts(appId).values(); |
| } |
| }); |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| fail(); |
| } |
| } catch (AuthorizationException e) { |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| return; |
| } |
| throw e; |
| } |
| } |
| assertNotNull(appAttempts); |
| assertEquals(SCALE, appAttempts.size()); |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetContainers(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| final ApplicationAttemptId appAttemptId = |
| ApplicationAttemptId.newInstance(ApplicationId.newInstance(0, 1), 1); |
| Collection<ContainerReport> containers; |
| if (callerUGI == null) { |
| containers = historyManager.getContainers(appAttemptId).values(); |
| } else { |
| try { |
| containers = callerUGI.doAs(new PrivilegedExceptionAction<Collection<ContainerReport>>() { |
| @Override |
| public Collection<ContainerReport> run() throws Exception { |
| return historyManager.getContainers(appAttemptId).values(); |
| } |
| }); |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| fail(); |
| } |
| } catch (AuthorizationException e) { |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| return; |
| } |
| throw e; |
| } |
| } |
| assertNotNull(containers); |
| assertEquals(SCALE, containers.size()); |
| } |
| |
| @MethodSource("callers") |
| @ParameterizedTest |
| void testGetAMContainer(String caller) throws Exception { |
| initTestApplicationHistoryManagerOnTimelineStore(caller); |
| final ApplicationAttemptId appAttemptId = |
| ApplicationAttemptId.newInstance(ApplicationId.newInstance(0, 1), 1); |
| ContainerReport container; |
| if (callerUGI == null) { |
| container = historyManager.getAMContainer(appAttemptId); |
| } else { |
| try { |
| container = callerUGI.doAs(new PrivilegedExceptionAction<ContainerReport>() { |
| @Override |
| public ContainerReport run() throws Exception { |
| return historyManager.getAMContainer(appAttemptId); |
| } |
| }); |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| fail(); |
| } |
| } catch (AuthorizationException e) { |
| if (callerUGI != null && callerUGI.getShortUserName().equals("user3")) { |
| // The exception is expected |
| return; |
| } |
| throw e; |
| } |
| } |
| assertNotNull(container); |
| assertEquals(appAttemptId, container.getContainerId().getApplicationAttemptId()); |
| } |
| |
| private static TimelineEntity createApplicationTimelineEntity(ApplicationId appId, |
| boolean emptyACLs, boolean noAttemptId, boolean wrongAppId, boolean enableUpdateEvent, |
| YarnApplicationState state) { |
| return createApplicationTimelineEntity(appId, emptyACLs, noAttemptId, wrongAppId, |
| enableUpdateEvent, state, false, false); |
| } |
| |
| private static TimelineEntity createApplicationTimelineEntity(ApplicationId appId, |
| boolean emptyACLs, boolean noAttemptId, boolean wrongAppId, boolean enableUpdateEvent, |
| YarnApplicationState state, boolean missingPreemptMetrics, boolean missingQueue) { |
| TimelineEntity entity = new TimelineEntity(); |
| entity.setEntityType(ApplicationMetricsConstants.ENTITY_TYPE); |
| if (wrongAppId) { |
| entity.setEntityId("wrong_app_id"); |
| } else { |
| entity.setEntityId(appId.toString()); |
| } |
| entity.setDomainId(TimelineDataManager.DEFAULT_DOMAIN_ID); |
| entity.addPrimaryFilter(TimelineStore.SystemFilter.ENTITY_OWNER.toString(), "yarn"); |
| Map<String, Object> entityInfo = new HashMap<String, Object>(); |
| entityInfo.put(ApplicationMetricsConstants.NAME_ENTITY_INFO, "test app"); |
| entityInfo.put(ApplicationMetricsConstants.TYPE_ENTITY_INFO, "test app type"); |
| entityInfo.put(ApplicationMetricsConstants.USER_ENTITY_INFO, "user1"); |
| if (!missingQueue) { |
| entityInfo.put(ApplicationMetricsConstants.QUEUE_ENTITY_INFO, "test queue"); |
| } |
| entityInfo.put(ApplicationMetricsConstants.UNMANAGED_APPLICATION_ENTITY_INFO, "false"); |
| entityInfo.put(ApplicationMetricsConstants.APPLICATION_PRIORITY_INFO, Priority.newInstance(0)); |
| entityInfo.put(ApplicationMetricsConstants.SUBMITTED_TIME_ENTITY_INFO, Integer.MAX_VALUE + 1L); |
| entityInfo.put(ApplicationMetricsConstants.APP_MEM_METRICS, 123); |
| entityInfo.put(ApplicationMetricsConstants.APP_CPU_METRICS, 345); |
| if (!missingPreemptMetrics) { |
| entityInfo.put(ApplicationMetricsConstants.APP_MEM_PREEMPT_METRICS, 456); |
| entityInfo.put(ApplicationMetricsConstants.APP_CPU_PREEMPT_METRICS, 789); |
| } |
| if (emptyACLs) { |
| entityInfo.put(ApplicationMetricsConstants.APP_VIEW_ACLS_ENTITY_INFO, ""); |
| } else { |
| entityInfo.put(ApplicationMetricsConstants.APP_VIEW_ACLS_ENTITY_INFO, "user2"); |
| } |
| Set<String> appTags = new HashSet<String>(); |
| appTags.add("Test_APP_TAGS_1"); |
| appTags.add("Test_APP_TAGS_2"); |
| entityInfo.put(ApplicationMetricsConstants.APP_TAGS_INFO, appTags); |
| entity.setOtherInfo(entityInfo); |
| TimelineEvent tEvent = new TimelineEvent(); |
| tEvent.setEventType(ApplicationMetricsConstants.CREATED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + 2L + appId.getId()); |
| entity.addEvent(tEvent); |
| tEvent = new TimelineEvent(); |
| tEvent.setEventType(ApplicationMetricsConstants.FINISHED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + 3L + appId.getId()); |
| Map<String, Object> eventInfo = new HashMap<String, Object>(); |
| eventInfo.put(ApplicationMetricsConstants.DIAGNOSTICS_INFO_EVENT_INFO, "test diagnostics info"); |
| eventInfo.put(ApplicationMetricsConstants.FINAL_STATUS_EVENT_INFO, |
| FinalApplicationStatus.UNDEFINED.toString()); |
| eventInfo.put(ApplicationMetricsConstants.STATE_EVENT_INFO, state.toString()); |
| if (!noAttemptId) { |
| eventInfo.put(ApplicationMetricsConstants.LATEST_APP_ATTEMPT_EVENT_INFO, |
| ApplicationAttemptId.newInstance(appId, 1)); |
| } |
| tEvent.setEventInfo(eventInfo); |
| entity.addEvent(tEvent); |
| // send a YARN_APPLICATION_STATE_UPDATED event |
| // after YARN_APPLICATION_FINISHED |
| // The final YarnApplicationState should not be changed |
| tEvent = new TimelineEvent(); |
| tEvent.setEventType(ApplicationMetricsConstants.STATE_UPDATED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + 4L + appId.getId()); |
| eventInfo = new HashMap<String, Object>(); |
| eventInfo.put(ApplicationMetricsConstants.STATE_EVENT_INFO, YarnApplicationState.KILLED); |
| tEvent.setEventInfo(eventInfo); |
| entity.addEvent(tEvent); |
| if (enableUpdateEvent) { |
| tEvent = new TimelineEvent(); |
| long updatedTimeIndex = 4L; |
| createAppModifiedEvent(appId, tEvent, updatedTimeIndex++, "changed queue", 5); |
| entity.addEvent(tEvent); |
| // Change priority alone |
| tEvent = new TimelineEvent(); |
| createAppModifiedEvent(appId, tEvent, updatedTimeIndex++, "changed queue", 6); |
| // Now change queue |
| tEvent = new TimelineEvent(); |
| createAppModifiedEvent(appId, tEvent, updatedTimeIndex++, "changed queue1", 6); |
| entity.addEvent(tEvent); |
| } |
| return entity; |
| } |
| |
| private static void createAppModifiedEvent(ApplicationId appId, TimelineEvent tEvent, |
| long updatedTimeIndex, String queue, int priority) { |
| tEvent.setEventType(ApplicationMetricsConstants.UPDATED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + updatedTimeIndex + appId.getId()); |
| Map<String, Object> eventInfo = new HashMap<String, Object>(); |
| eventInfo.put(ApplicationMetricsConstants.QUEUE_ENTITY_INFO, queue); |
| eventInfo.put(ApplicationMetricsConstants.APPLICATION_PRIORITY_INFO, priority); |
| tEvent.setEventInfo(eventInfo); |
| } |
| |
| private static TimelineEntity createAppAttemptTimelineEntity(ApplicationAttemptId appAttemptId) { |
| TimelineEntity entity = new TimelineEntity(); |
| entity.setEntityType(AppAttemptMetricsConstants.ENTITY_TYPE); |
| entity.setEntityId(appAttemptId.toString()); |
| entity.setDomainId(TimelineDataManager.DEFAULT_DOMAIN_ID); |
| entity.addPrimaryFilter(AppAttemptMetricsConstants.PARENT_PRIMARY_FILTER, |
| appAttemptId.getApplicationId().toString()); |
| entity.addPrimaryFilter(TimelineStore.SystemFilter.ENTITY_OWNER.toString(), "yarn"); |
| TimelineEvent tEvent = new TimelineEvent(); |
| tEvent.setEventType(AppAttemptMetricsConstants.REGISTERED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + 1L); |
| Map<String, Object> eventInfo = new HashMap<String, Object>(); |
| eventInfo.put(AppAttemptMetricsConstants.TRACKING_URL_INFO, "test tracking url"); |
| eventInfo.put(AppAttemptMetricsConstants.ORIGINAL_TRACKING_URL_INFO, |
| "test original tracking url"); |
| eventInfo.put(AppAttemptMetricsConstants.HOST_INFO, "test host"); |
| eventInfo.put(AppAttemptMetricsConstants.RPC_PORT_INFO, 100); |
| eventInfo.put(AppAttemptMetricsConstants.MASTER_CONTAINER_INFO, |
| ContainerId.newContainerId(appAttemptId, 1)); |
| tEvent.setEventInfo(eventInfo); |
| entity.addEvent(tEvent); |
| tEvent = new TimelineEvent(); |
| tEvent.setEventType(AppAttemptMetricsConstants.FINISHED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + 2L); |
| eventInfo = new HashMap<String, Object>(); |
| eventInfo.put(AppAttemptMetricsConstants.TRACKING_URL_INFO, "test tracking url"); |
| eventInfo.put(AppAttemptMetricsConstants.ORIGINAL_TRACKING_URL_INFO, |
| "test original tracking url"); |
| eventInfo.put(AppAttemptMetricsConstants.DIAGNOSTICS_INFO, "test diagnostics info"); |
| eventInfo.put(AppAttemptMetricsConstants.FINAL_STATUS_INFO, |
| FinalApplicationStatus.UNDEFINED.toString()); |
| eventInfo.put(AppAttemptMetricsConstants.STATE_INFO, |
| YarnApplicationAttemptState.FINISHED.toString()); |
| tEvent.setEventInfo(eventInfo); |
| entity.addEvent(tEvent); |
| return entity; |
| } |
| |
| private static TimelineEntity createContainerEntity(ContainerId containerId) { |
| TimelineEntity entity = new TimelineEntity(); |
| entity.setEntityType(ContainerMetricsConstants.ENTITY_TYPE); |
| entity.setEntityId(containerId.toString()); |
| entity.setDomainId(TimelineDataManager.DEFAULT_DOMAIN_ID); |
| entity.addPrimaryFilter(ContainerMetricsConstants.PARENT_PRIMARIY_FILTER, |
| containerId.getApplicationAttemptId().toString()); |
| entity.addPrimaryFilter(TimelineStore.SystemFilter.ENTITY_OWNER.toString(), "yarn"); |
| Map<String, Object> entityInfo = new HashMap<String, Object>(); |
| entityInfo.put(ContainerMetricsConstants.ALLOCATED_MEMORY_INFO, -1); |
| entityInfo.put(ContainerMetricsConstants.ALLOCATED_VCORE_INFO, -1); |
| entityInfo.put(ContainerMetricsConstants.ALLOCATED_HOST_INFO, "test host"); |
| entityInfo.put(ContainerMetricsConstants.ALLOCATED_PORT_INFO, 100); |
| entityInfo.put(ContainerMetricsConstants.ALLOCATED_PRIORITY_INFO, -1); |
| entityInfo.put(ContainerMetricsConstants.ALLOCATED_HOST_HTTP_ADDRESS_INFO, "http://test:1234"); |
| entity.setOtherInfo(entityInfo); |
| TimelineEvent tEvent = new TimelineEvent(); |
| tEvent.setEventType(ContainerMetricsConstants.CREATED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + 1L); |
| entity.addEvent(tEvent); |
| tEvent = new TimelineEvent(); |
| tEvent.setEventType(ContainerMetricsConstants.FINISHED_EVENT_TYPE); |
| tEvent.setTimestamp(Integer.MAX_VALUE + 2L); |
| Map<String, Object> eventInfo = new HashMap<String, Object>(); |
| eventInfo.put(ContainerMetricsConstants.DIAGNOSTICS_INFO, "test diagnostics info"); |
| eventInfo.put(ContainerMetricsConstants.EXIT_STATUS_INFO, -1); |
| eventInfo.put(ContainerMetricsConstants.STATE_INFO, ContainerState.COMPLETE.toString()); |
| tEvent.setEventInfo(eventInfo); |
| entity.addEvent(tEvent); |
| return entity; |
| } |
| } |
