Java 类org.apache.hadoop.yarn.api.records.timeline.TimelineEntity 实例源码

private static TimelineEntity generateEntity() {
  TimelineEntity entity = new TimelineEntity();
  entity.setEntityId("entity id");
  entity.setEntityType("entity type");
  entity.setStartTime(System.currentTimeMillis());
  for (int i = 0; i < 2; ++i) {
    TimelineEvent event = new TimelineEvent();
    event.setTimestamp(System.currentTimeMillis());
    event.setEventType("test event type " + i);
    event.addEventInfo("key1", "val1");
    event.addEventInfo("key2", "val2");
    entity.addEvent(event);
  }
  entity.addRelatedEntity("test ref type 1", "test ref id 1");
  entity.addRelatedEntity("test ref type 2", "test ref id 2");
  entity.addPrimaryFilter("pkey1", "pval1");
  entity.addPrimaryFilter("pkey2", "pval2");
  entity.addOtherInfo("okey1", "oval1");
  entity.addOtherInfo("okey2", "oval2");
  entity.setDomainId("domain id 1");
  return entity;
}

private void publishApplicationCreatedEvent(ApplicationCreatedEvent event) {
  TimelineEntity entity =
      createApplicationEntity(event.getApplicationId());
  Map<String, Object> entityInfo = new HashMap<String, Object>();
  entityInfo.put(ApplicationMetricsConstants.NAME_ENTITY_INFO,
      event.getApplicationName());
  entityInfo.put(ApplicationMetricsConstants.TYPE_ENTITY_INFO,
      event.getApplicationType());
  entityInfo.put(ApplicationMetricsConstants.USER_ENTITY_INFO,
      event.getUser());
  entityInfo.put(ApplicationMetricsConstants.QUEUE_ENTITY_INFO,
      event.getQueue());
  entityInfo.put(ApplicationMetricsConstants.SUBMITTED_TIME_ENTITY_INFO,
      event.getSubmittedTime());
  entity.setOtherInfo(entityInfo);
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(
      ApplicationMetricsConstants.CREATED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  entity.addEvent(tEvent);
  putEntity(entity);
}

private void
    publishAppAttemptRegisteredEvent(AppAttemptRegisteredEvent event) {
  TimelineEntity entity =
      createAppAttemptEntity(event.getApplicationAttemptId());
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(
      AppAttemptMetricsConstants.REGISTERED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  Map<String, Object> eventInfo = new HashMap<String, Object>();
  eventInfo.put(
      AppAttemptMetricsConstants.TRACKING_URL_EVENT_INFO,
      event.getTrackingUrl());
  eventInfo.put(
      AppAttemptMetricsConstants.ORIGINAL_TRACKING_URL_EVENT_INFO,
      event.getOriginalTrackingURL());
  eventInfo.put(AppAttemptMetricsConstants.HOST_EVENT_INFO,
      event.getHost());
  eventInfo.put(AppAttemptMetricsConstants.RPC_PORT_EVENT_INFO,
      event.getRpcPort());
  eventInfo.put(
      AppAttemptMetricsConstants.MASTER_CONTAINER_EVENT_INFO,
      event.getMasterContainerId().toString());
  tEvent.setEventInfo(eventInfo);
  entity.addEvent(tEvent);
  putEntity(entity);
}

private void publishAppAttemptFinishedEvent(AppAttemptFinishedEvent event) {
  TimelineEntity entity =
      createAppAttemptEntity(event.getApplicationAttemptId());
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(AppAttemptMetricsConstants.FINISHED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  Map<String, Object> eventInfo = new HashMap<String, Object>();
  eventInfo.put(
      AppAttemptMetricsConstants.TRACKING_URL_EVENT_INFO,
      event.getTrackingUrl());
  eventInfo.put(
      AppAttemptMetricsConstants.ORIGINAL_TRACKING_URL_EVENT_INFO,
      event.getOriginalTrackingURL());
  eventInfo.put(AppAttemptMetricsConstants.DIAGNOSTICS_INFO_EVENT_INFO,
      event.getDiagnosticsInfo());
  eventInfo.put(AppAttemptMetricsConstants.FINAL_STATUS_EVENT_INFO,
      event.getFinalApplicationStatus().toString());
  eventInfo.put(AppAttemptMetricsConstants.STATE_EVENT_INFO,
      event.getYarnApplicationAttemptState().toString());
  tEvent.setEventInfo(eventInfo);
  entity.addEvent(tEvent);
  putEntity(entity);
}

private void publishContainerCreatedEvent(ContainerCreatedEvent event) {
  TimelineEntity entity = createContainerEntity(event.getContainerId());
  Map<String, Object> entityInfo = new HashMap<String, Object>();
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_MEMORY_ENTITY_INFO,
      event.getAllocatedResource().getMemory());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_VCORE_ENTITY_INFO,
      event.getAllocatedResource().getVirtualCores());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_GCORE_ENTITY_INFO,
      event.getAllocatedResource().getGpuCores());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_HOST_ENTITY_INFO,
      event.getAllocatedNode().getHost());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_PORT_ENTITY_INFO,
      event.getAllocatedNode().getPort());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_PRIORITY_ENTITY_INFO,
      event.getAllocatedPriority().getPriority());
  entityInfo.put(
    ContainerMetricsConstants.ALLOCATED_HOST_HTTP_ADDRESS_ENTITY_INFO,
    event.getNodeHttpAddress());
  entity.setOtherInfo(entityInfo);
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(ContainerMetricsConstants.CREATED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  entity.addEvent(tEvent);
  putEntity(entity);
}

private void publishContainerFinishedEvent(ContainerFinishedEvent event) {
  TimelineEntity entity = createContainerEntity(event.getContainerId());
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(ContainerMetricsConstants.FINISHED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  Map<String, Object> eventInfo = new HashMap<String, Object>();
  eventInfo.put(ContainerMetricsConstants.DIAGNOSTICS_INFO_EVENT_INFO,
      event.getDiagnosticsInfo());
  eventInfo.put(ContainerMetricsConstants.EXIT_STATUS_EVENT_INFO,
      event.getContainerExitStatus());
  eventInfo.put(ContainerMetricsConstants.STATE_EVENT_INFO,
      event.getContainerState().toString());
  tEvent.setEventInfo(eventInfo);
  entity.addEvent(tEvent);
  putEntity(entity);
}

@Override
public Map<ApplicationId, ApplicationReport> getAllApplications()
    throws YarnException, IOException {
  TimelineEntities entities = timelineDataManager.getEntities(
      ApplicationMetricsConstants.ENTITY_TYPE, null, null, null, null,
      null, null, Long.MAX_VALUE, EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  Map<ApplicationId, ApplicationReport> apps =
      new LinkedHashMap<ApplicationId, ApplicationReport>();
  if (entities != null && entities.getEntities() != null) {
    for (TimelineEntity entity : entities.getEntities()) {
      try {
        ApplicationReportExt app =
            generateApplicationReport(entity, ApplicationReportField.ALL);
        apps.put(app.appReport.getApplicationId(), app.appReport);
      } catch (Exception e) {
        LOG.error("Error on generating application report for " +
            entity.getEntityId(), e);
      }
    }
  }
  return apps;
}

@Override
public Map<ApplicationAttemptId, ApplicationAttemptReport>
    getApplicationAttempts(ApplicationId appId)
        throws YarnException, IOException {
  ApplicationReportExt app = getApplication(
      appId, ApplicationReportField.USER_AND_ACLS);
  checkAccess(app);
  TimelineEntities entities = timelineDataManager.getEntities(
      AppAttemptMetricsConstants.ENTITY_TYPE,
      new NameValuePair(
          AppAttemptMetricsConstants.PARENT_PRIMARY_FILTER, appId
              .toString()), null, null, null, null, null,
      Long.MAX_VALUE, EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  Map<ApplicationAttemptId, ApplicationAttemptReport> appAttempts =
      new LinkedHashMap<ApplicationAttemptId, ApplicationAttemptReport>();
  for (TimelineEntity entity : entities.getEntities()) {
    ApplicationAttemptReport appAttempt =
        convertToApplicationAttemptReport(entity);
    appAttempts.put(appAttempt.getApplicationAttemptId(), appAttempt);
  }
  return appAttempts;
}

private ApplicationAttemptReport getApplicationAttempt(
    ApplicationAttemptId appAttemptId, boolean checkACLs)
    throws YarnException, IOException {
  if (checkACLs) {
    ApplicationReportExt app = getApplication(
        appAttemptId.getApplicationId(),
        ApplicationReportField.USER_AND_ACLS);
    checkAccess(app);
  }
  TimelineEntity entity = timelineDataManager.getEntity(
      AppAttemptMetricsConstants.ENTITY_TYPE,
      appAttemptId.toString(), EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  if (entity == null) {
    throw new ApplicationAttemptNotFoundException(
        "The entity for application attempt " + appAttemptId +
        " doesn't exist in the timeline store");
  } else {
    return convertToApplicationAttemptReport(entity);
  }
}

@Override
public ContainerReport getContainer(ContainerId containerId)
    throws YarnException, IOException {
  ApplicationReportExt app = getApplication(
      containerId.getApplicationAttemptId().getApplicationId(),
      ApplicationReportField.USER_AND_ACLS);
  checkAccess(app);
  TimelineEntity entity = timelineDataManager.getEntity(
      ContainerMetricsConstants.ENTITY_TYPE,
      containerId.toString(), EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  if (entity == null) {
    throw new ContainerNotFoundException(
        "The entity for container " + containerId +
        " doesn't exist in the timeline store");
  } else {
    return convertToContainerReport(
        entity, serverHttpAddress, app.appReport.getUser());
  }
}

@Override
public Map<ContainerId, ContainerReport> getContainers(
    ApplicationAttemptId appAttemptId) throws YarnException, IOException {
  ApplicationReportExt app = getApplication(
      appAttemptId.getApplicationId(), ApplicationReportField.USER_AND_ACLS);
  checkAccess(app);
  TimelineEntities entities = timelineDataManager.getEntities(
      ContainerMetricsConstants.ENTITY_TYPE,
      new NameValuePair(
          ContainerMetricsConstants.PARENT_PRIMARIY_FILTER,
          appAttemptId.toString()), null, null, null,
      null, null, Long.MAX_VALUE, EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  Map<ContainerId, ContainerReport> containers =
      new LinkedHashMap<ContainerId, ContainerReport>();
  if (entities != null && entities.getEntities() != null) {
    for (TimelineEntity entity : entities.getEntities()) {
      ContainerReport container = convertToContainerReport(
          entity, serverHttpAddress, app.appReport.getUser());
      containers.put(container.getContainerId(), container);
    }
  }
  return containers;
}

/**
 * Get the single timeline entity that the given user has access to. The
 * meaning of each argument has been documented with
 * {@link TimelineReader#getEntity}.
 * 
 * @see TimelineReader#getEntity
 */
public TimelineEntity getEntity(
    String entityType,
    String entityId,
    EnumSet<Field> fields,
    UserGroupInformation callerUGI) throws YarnException, IOException {
  TimelineEntity entity = null;
  entity =
      store.getEntity(entityId, entityType, fields);
  if (entity != null) {
    addDefaultDomainIdIfAbsent(entity);
    // check ACLs
    if (!timelineACLsManager.checkAccess(
        callerUGI, ApplicationAccessType.VIEW_APP, entity)) {
      entity = null;
    }
  }
  return entity;
}

@Test
public void testPutTimelineEntities() throws Exception {
  KerberosTestUtils.doAs(HTTP_USER + "/localhost", new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      TimelineClient client = createTimelineClientForUGI();
      TimelineEntity entityToStore = new TimelineEntity();
      entityToStore.setEntityType(
          TestTimelineAuthenticationFilter.class.getName());
      entityToStore.setEntityId("entity1");
      entityToStore.setStartTime(0L);
      TimelinePutResponse putResponse = client.putEntities(entityToStore);
      Assert.assertEquals(0, putResponse.getErrors().size());
      TimelineEntity entityToRead =
          testTimelineServer.getTimelineStore().getEntity(
              "entity1", TestTimelineAuthenticationFilter.class.getName(), null);
      Assert.assertNotNull(entityToRead);
      return null;
    }
  });
}

@Test
public void testYarnACLsNotEnabledForEntity() throws Exception {
  Configuration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, false);
  TimelineACLsManager timelineACLsManager =
      new TimelineACLsManager(conf);
  timelineACLsManager.setTimelineStore(new TestTimelineStore());
  TimelineEntity entity = new TimelineEntity();
  entity.addPrimaryFilter(
      TimelineStore.SystemFilter.ENTITY_OWNER
          .toString(), "owner");
  entity.setDomainId("domain_id_1");
  Assert.assertTrue(
      "Always true when ACLs are not enabled",
      timelineACLsManager.checkAccess(
          UserGroupInformation.createRemoteUser("user"),
          ApplicationAccessType.VIEW_APP, entity));
  Assert.assertTrue(
      "Always true when ACLs are not enabled",
      timelineACLsManager.checkAccess(
          UserGroupInformation.createRemoteUser("user"),
          ApplicationAccessType.MODIFY_APP, entity));
}

@Test
public void testCorruptedOwnerInfoForEntity() throws Exception {
  Configuration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
  conf.set(YarnConfiguration.YARN_ADMIN_ACL, "owner");
  TimelineACLsManager timelineACLsManager =
      new TimelineACLsManager(conf);
  timelineACLsManager.setTimelineStore(new TestTimelineStore());
  TimelineEntity entity = new TimelineEntity();
  try {
    timelineACLsManager.checkAccess(
        UserGroupInformation.createRemoteUser("owner"),
        ApplicationAccessType.VIEW_APP, entity);
    Assert.fail("Exception is expected");
  } catch (YarnException e) {
    Assert.assertTrue("It's not the exact expected exception", e.getMessage()
        .contains("doesn't exist."));
  }
}

@Test
public void testGetEntity() throws Exception {
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("timeline")
      .path("type_1").path("id_1")
      .accept(MediaType.APPLICATION_JSON)
      .get(ClientResponse.class);
  assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
  TimelineEntity entity = response.getEntity(TimelineEntity.class);
  Assert.assertNotNull(entity);
  Assert.assertEquals("id_1", entity.getEntityId());
  Assert.assertEquals("type_1", entity.getEntityType());
  Assert.assertEquals(123l, entity.getStartTime().longValue());
  Assert.assertEquals(2, entity.getEvents().size());
  Assert.assertEquals(4, entity.getPrimaryFilters().size());
  Assert.assertEquals(4, entity.getOtherInfo().size());
}

@Test
public void testGetEntityFields1() throws Exception {
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("timeline")
      .path("type_1").path("id_1").queryParam("fields", "events,otherinfo")
      .accept(MediaType.APPLICATION_JSON)
      .get(ClientResponse.class);
  assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
  TimelineEntity entity = response.getEntity(TimelineEntity.class);
  Assert.assertNotNull(entity);
  Assert.assertEquals("id_1", entity.getEntityId());
  Assert.assertEquals("type_1", entity.getEntityType());
  Assert.assertEquals(123l, entity.getStartTime().longValue());
  Assert.assertEquals(2, entity.getEvents().size());
  Assert.assertEquals(0, entity.getPrimaryFilters().size());
  Assert.assertEquals(4, entity.getOtherInfo().size());
}

@Test
public void testGetEntityFields2() throws Exception {
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("timeline")
      .path("type_1").path("id_1").queryParam("fields", "lasteventonly," +
          "primaryfilters,relatedentities")
      .accept(MediaType.APPLICATION_JSON)
      .get(ClientResponse.class);
  assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
  TimelineEntity entity = response.getEntity(TimelineEntity.class);
  Assert.assertNotNull(entity);
  Assert.assertEquals("id_1", entity.getEntityId());
  Assert.assertEquals("type_1", entity.getEntityType());
  Assert.assertEquals(123l, entity.getStartTime().longValue());
  Assert.assertEquals(1, entity.getEvents().size());
  Assert.assertEquals(4, entity.getPrimaryFilters().size());
  Assert.assertEquals(0, entity.getOtherInfo().size());
}

@Test
public void testPostEntitiesWithPrimaryFilter() throws Exception {
  TimelineEntities entities = new TimelineEntities();
  TimelineEntity entity = new TimelineEntity();
  Map<String, Set<Object>> filters = new HashMap<String, Set<Object>>();
  filters.put(TimelineStore.SystemFilter.ENTITY_OWNER.toString(),
      new HashSet<Object>());
  entity.setPrimaryFilters(filters);
  entity.setEntityId("test id 6");
  entity.setEntityType("test type 6");
  entity.setStartTime(System.currentTimeMillis());
  entities.addEntity(entity);
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("timeline")
      .queryParam("user.name", "tester")
      .accept(MediaType.APPLICATION_JSON)
      .type(MediaType.APPLICATION_JSON)
      .post(ClientResponse.class, entities);
  TimelinePutResponse putResposne =
      response.getEntity(TimelinePutResponse.class);
  Assert.assertEquals(0, putResposne.getErrors().size());
}

@Test
public void testRelatingToNonExistingEntity() throws IOException {
  TimelineEntity entityToStore = new TimelineEntity();
  entityToStore.setEntityType("TEST_ENTITY_TYPE_1");
  entityToStore.setEntityId("TEST_ENTITY_ID_1");
  entityToStore.setDomainId(TimelineDataManager.DEFAULT_DOMAIN_ID);
  entityToStore.addRelatedEntity("TEST_ENTITY_TYPE_2", "TEST_ENTITY_ID_2");
  TimelineEntities entities = new TimelineEntities();
  entities.addEntity(entityToStore);
  store.put(entities);
  TimelineEntity entityToGet =
      store.getEntity("TEST_ENTITY_ID_2", "TEST_ENTITY_TYPE_2", null);
  Assert.assertNotNull(entityToGet);
  Assert.assertEquals("DEFAULT", entityToGet.getDomainId());
  Assert.assertEquals("TEST_ENTITY_TYPE_1",
      entityToGet.getRelatedEntities().keySet().iterator().next());
  Assert.assertEquals("TEST_ENTITY_ID_1",
      entityToGet.getRelatedEntities().values().iterator().next()
          .iterator().next());
}

/**
 * Create a test entity
 */
protected static TimelineEntity createEntity(String entityId, String entityType,
    Long startTime, List<TimelineEvent> events,
    Map<String, Set<String>> relatedEntities,
    Map<String, Set<Object>> primaryFilters,
    Map<String, Object> otherInfo, String domainId) {
  TimelineEntity entity = new TimelineEntity();
  entity.setEntityId(entityId);
  entity.setEntityType(entityType);
  entity.setStartTime(startTime);
  entity.setEvents(events);
  if (relatedEntities != null) {
    for (Entry<String, Set<String>> e : relatedEntities.entrySet()) {
      for (String v : e.getValue()) {
        entity.addRelatedEntity(e.getKey(), v);
      }
    }
  } else {
    entity.setRelatedEntities(null);
  }
  entity.setPrimaryFilters(primaryFilters);
  entity.setOtherInfo(otherInfo);
  entity.setDomainId(domainId);
  return entity;
}

private static void publishContainerEndEvent(
    final TimelineClient timelineClient, ContainerStatus container,
    String domainId, UserGroupInformation ugi) {
  final TimelineEntity entity = new TimelineEntity();
  entity.setEntityId(container.getContainerId().toString());
  entity.setEntityType(DSEntity.DS_CONTAINER.toString());
  entity.setDomainId(domainId);
  entity.addPrimaryFilter("user", ugi.getShortUserName());
  TimelineEvent event = new TimelineEvent();
  event.setTimestamp(System.currentTimeMillis());
  event.setEventType(DSEvent.DS_CONTAINER_END.toString());
  event.addEventInfo("State", container.getState().name());
  event.addEventInfo("Exit Status", container.getExitStatus());
  entity.addEvent(event);
  try {
    timelineClient.putEntities(entity);
  } catch (YarnException | IOException e) {
    LOG.error("Container end event could not be published for "
        + container.getContainerId().toString(), e);
  }
}

private static void publishApplicationAttemptEvent(
    final TimelineClient timelineClient, String appAttemptId,
    DSEvent appEvent, String domainId, UserGroupInformation ugi) {
  final TimelineEntity entity = new TimelineEntity();
  entity.setEntityId(appAttemptId);
  entity.setEntityType(DSEntity.DS_APP_ATTEMPT.toString());
  entity.setDomainId(domainId);
  entity.addPrimaryFilter("user", ugi.getShortUserName());
  TimelineEvent event = new TimelineEvent();
  event.setEventType(appEvent.toString());
  event.setTimestamp(System.currentTimeMillis());
  entity.addEvent(event);
  try {
    timelineClient.putEntities(entity);
  } catch (YarnException | IOException e) {
    LOG.error("App Attempt "
        + (appEvent.equals(DSEvent.DS_APP_ATTEMPT_START) ? "start" : "end")
        + " event could not be published for "
        + appAttemptId.toString(), e);
  }
}

@Test
public void testPostEntitiesWithPrimaryFilter() throws Exception {
  TimelineEntities entities = new TimelineEntities();
  TimelineEntity entity = new TimelineEntity();
  Map<String, Set<Object>> filters = new HashMap<String, Set<Object>>();
  filters.put(TimelineStore.SystemFilter.ENTITY_OWNER.toString(),
      new HashSet<Object>());
  entity.setPrimaryFilters(filters);
  entity.setEntityId("test id 6");
  entity.setEntityType("test type 6");
  entity.setStartTime(System.currentTimeMillis());
  entities.addEntity(entity);
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("timeline")
      .queryParam("user.name", "tester")
      .accept(MediaType.APPLICATION_JSON)
      .type(MediaType.APPLICATION_JSON)
      .post(ClientResponse.class, entities);
  TimelinePutResponse putResposne =
      response.getEntity(TimelinePutResponse.class);
  Assert.assertEquals(0, putResposne.getErrors().size());
}

private void writeSummmaryEntityLogs(FileSystem fs, Path logPath,
    ObjectMapper objMapper, ApplicationAttemptId attemptId,
    List<TimelineEntity> entities, boolean isAppendSupported,
    Map<ApplicationAttemptId, EntityLogFD> logFDs) throws IOException {
  EntityLogFD logFD = null;
  logFD = logFDs.get(attemptId);
  if (logFD != null) {
    try {
      logFD.lock();
      if (serviceStopped) {
        return;
      }
      logFD.writeEntities(entities);
    } finally {
      logFD.unlock();
    }
  } else {
    createSummaryFDAndWrite(fs, logPath, objMapper, attemptId, entities,
        isAppendSupported, logFDs);
  }
}

private static TimelineEntity generateEntity() {
  TimelineEntity entity = new TimelineEntity();
  entity.setEntityId("entity id");
  entity.setEntityType("entity type");
  entity.setStartTime(System.currentTimeMillis());
  for (int i = 0; i < 2; ++i) {
    TimelineEvent event = new TimelineEvent();
    event.setTimestamp(System.currentTimeMillis());
    event.setEventType("test event type " + i);
    event.addEventInfo("key1", "val1");
    event.addEventInfo("key2", "val2");
    entity.addEvent(event);
  }
  entity.addRelatedEntity("test ref type 1", "test ref id 1");
  entity.addRelatedEntity("test ref type 2", "test ref id 2");
  entity.addPrimaryFilter("pkey1", "pval1");
  entity.addPrimaryFilter("pkey2", "pval2");
  entity.addOtherInfo("okey1", "oval1");
  entity.addOtherInfo("okey2", "oval2");
  entity.setDomainId("domain id 1");
  return entity;
}

private void publishContainerCreatedEvent(ContainerCreatedEvent event) {
  TimelineEntity entity = createContainerEntity(event.getContainerId());
  Map<String, Object> entityInfo = new HashMap<String, Object>();
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_MEMORY_ENTITY_INFO,
      event.getAllocatedResource().getMemory());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_VCORE_ENTITY_INFO,
      event.getAllocatedResource().getVirtualCores());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_HOST_ENTITY_INFO,
      event.getAllocatedNode().getHost());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_PORT_ENTITY_INFO,
      event.getAllocatedNode().getPort());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_PRIORITY_ENTITY_INFO,
      event.getAllocatedPriority().getPriority());
  entityInfo.put(
    ContainerMetricsConstants.ALLOCATED_HOST_HTTP_ADDRESS_ENTITY_INFO,
    event.getNodeHttpAddress());
  entity.setOtherInfo(entityInfo);
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(ContainerMetricsConstants.CREATED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  entity.addEvent(tEvent);
  putEntity(entity);
}

private void publishContainerFinishedEvent(ContainerFinishedEvent event) {
  TimelineEntity entity = createContainerEntity(event.getContainerId());
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(ContainerMetricsConstants.FINISHED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  Map<String, Object> eventInfo = new HashMap<String, Object>();
  eventInfo.put(ContainerMetricsConstants.DIAGNOSTICS_INFO_EVENT_INFO,
      event.getDiagnosticsInfo());
  eventInfo.put(ContainerMetricsConstants.EXIT_STATUS_EVENT_INFO,
      event.getContainerExitStatus());
  eventInfo.put(ContainerMetricsConstants.STATE_EVENT_INFO,
      event.getContainerState().toString());
  tEvent.setEventInfo(eventInfo);
  entity.addEvent(tEvent);
  putEntity(entity);
}

@Test(timeout = 10000)
public void testPublishAppAttemptMetricsForUnmanagedAM() throws Exception {
  ApplicationAttemptId appAttemptId =
      ApplicationAttemptId.newInstance(ApplicationId.newInstance(0, 1), 1);
  RMAppAttempt appAttempt = createRMAppAttempt(appAttemptId,true);
  metricsPublisher.appAttemptRegistered(appAttempt, Integer.MAX_VALUE + 1L);
  RMApp app = mock(RMApp.class);
  when(app.getFinalApplicationStatus()).thenReturn(FinalApplicationStatus.UNDEFINED);
  metricsPublisher.appAttemptFinished(appAttempt, RMAppAttemptState.FINISHED, app,
      Integer.MAX_VALUE + 2L);
  TimelineEntity entity = null;
  do {
    entity =
        store.getEntity(appAttemptId.toString(),
            AppAttemptMetricsConstants.ENTITY_TYPE,
            EnumSet.allOf(Field.class));
    // ensure two events are both published before leaving the loop
  } while (entity == null || entity.getEvents().size() < 2);
}

private void publishContainerCreatedEvent(ContainerCreatedEvent event) {
  TimelineEntity entity = createContainerEntity(event.getContainerId());
  Map<String, Object> entityInfo = new HashMap<String, Object>();
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_MEMORY_ENTITY_INFO,
      event.getAllocatedResource().getMemory());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_VCORE_ENTITY_INFO,
      event.getAllocatedResource().getVirtualCores());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_HOST_ENTITY_INFO,
      event.getAllocatedNode().getHost());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_PORT_ENTITY_INFO,
      event.getAllocatedNode().getPort());
  entityInfo.put(ContainerMetricsConstants.ALLOCATED_PRIORITY_ENTITY_INFO,
      event.getAllocatedPriority().getPriority());
  entityInfo.put(
    ContainerMetricsConstants.ALLOCATED_HOST_HTTP_ADDRESS_ENTITY_INFO,
    event.getNodeHttpAddress());
  entity.setOtherInfo(entityInfo);
  TimelineEvent tEvent = new TimelineEvent();
  tEvent.setEventType(ContainerMetricsConstants.CREATED_EVENT_TYPE);
  tEvent.setTimestamp(event.getTimestamp());
  entity.addEvent(tEvent);
  putEntity(entity);
}

private ApplicationAttemptReport getApplicationAttempt(
    ApplicationAttemptId appAttemptId, boolean checkACLs)
    throws YarnException, IOException {
  if (checkACLs) {
    ApplicationReportExt app = getApplication(
        appAttemptId.getApplicationId(),
        ApplicationReportField.USER_AND_ACLS);
    checkAccess(app);
  }
  TimelineEntity entity = timelineDataManager.getEntity(
      AppAttemptMetricsConstants.ENTITY_TYPE,
      appAttemptId.toString(), EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  if (entity == null) {
    throw new ApplicationAttemptNotFoundException(
        "The entity for application attempt " + appAttemptId +
        " doesn't exist in the timeline store");
  } else {
    return convertToApplicationAttemptReport(entity);
  }
}

@Override
public ContainerReport getContainer(ContainerId containerId)
    throws YarnException, IOException {
  ApplicationReportExt app = getApplication(
      containerId.getApplicationAttemptId().getApplicationId(),
      ApplicationReportField.USER_AND_ACLS);
  checkAccess(app);
  TimelineEntity entity = timelineDataManager.getEntity(
      ContainerMetricsConstants.ENTITY_TYPE,
      containerId.toString(), EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  if (entity == null) {
    throw new ContainerNotFoundException(
        "The entity for container " + containerId +
        " doesn't exist in the timeline store");
  } else {
    return convertToContainerReport(
        entity, serverHttpAddress, app.appReport.getUser());
  }
}

@Override
public Map<ContainerId, ContainerReport> getContainers(
    ApplicationAttemptId appAttemptId) throws YarnException, IOException {
  ApplicationReportExt app = getApplication(
      appAttemptId.getApplicationId(), ApplicationReportField.USER_AND_ACLS);
  checkAccess(app);
  TimelineEntities entities = timelineDataManager.getEntities(
      ContainerMetricsConstants.ENTITY_TYPE,
      new NameValuePair(
          ContainerMetricsConstants.PARENT_PRIMARIY_FILTER,
          appAttemptId.toString()), null, null, null,
      null, null, Long.MAX_VALUE, EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  Map<ContainerId, ContainerReport> containers =
      new LinkedHashMap<ContainerId, ContainerReport>();
  if (entities != null && entities.getEntities() != null) {
    for (TimelineEntity entity : entities.getEntities()) {
      ContainerReport container = convertToContainerReport(
          entity, serverHttpAddress, app.appReport.getUser());
      containers.put(container.getContainerId(), container);
    }
  }
  return containers;
}

@Test
public void testCorruptedOwnerInfoForEntity() throws Exception {
  Configuration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
  conf.set(YarnConfiguration.YARN_ADMIN_ACL, "owner");
  TimelineACLsManager timelineACLsManager =
      new TimelineACLsManager(conf);
  timelineACLsManager.setTimelineStore(new TestTimelineStore());
  TimelineEntity entity = new TimelineEntity();
  try {
    timelineACLsManager.checkAccess(
        UserGroupInformation.createRemoteUser("owner"),
        ApplicationAccessType.VIEW_APP, entity);
    Assert.fail("Exception is expected");
  } catch (YarnException e) {
    Assert.assertTrue("It's not the exact expected exception", e.getMessage()
        .contains("doesn't exist."));
  }
}

private TimelineEntity createTaskAttemptEntity(TaskAttemptInfo taskAttemptInfo) {
  TimelineEntity taskAttempt = new TimelineEntity();
  taskAttempt.setEntityType(TASK_ATTEMPT);
  taskAttempt.setEntityId(taskAttemptInfo.getAttemptId().toString());
  taskAttempt.setStartTime(taskAttemptInfo.getStartTime());

  taskAttempt.addOtherInfo("START_TIME", taskAttemptInfo.getStartTime());
  taskAttempt.addOtherInfo("FINISH_TIME", taskAttemptInfo.getFinishTime());
  taskAttempt.addOtherInfo("MAP_FINISH_TIME",
      taskAttemptInfo.getMapFinishTime());
  taskAttempt.addOtherInfo("SHUFFLE_FINISH_TIME",
      taskAttemptInfo.getShuffleFinishTime());
  taskAttempt.addOtherInfo("SORT_FINISH_TIME",
      taskAttemptInfo.getSortFinishTime());
  taskAttempt.addOtherInfo("TASK_STATUS", taskAttemptInfo.getTaskStatus());
  taskAttempt.addOtherInfo("STATE", taskAttemptInfo.getState());
  taskAttempt.addOtherInfo("ERROR", taskAttemptInfo.getError());
  taskAttempt.addOtherInfo("CONTAINER_ID",
      taskAttemptInfo.getContainerId().toString());

  LOG.info("converted task attempt " + taskAttemptInfo.getAttemptId() +
      " to a timeline entity");
  return taskAttempt;
}

private TimelineEntity createJobEntity(JobInfo jobInfo, Configuration conf) {
  TimelineEntity job = new TimelineEntity();
  job.setEntityType(JOB);
  job.setEntityId(jobInfo.getJobId().toString());
  job.setStartTime(jobInfo.getSubmitTime());

  job.addPrimaryFilter("JOBNAME", jobInfo.getJobname());
  job.addPrimaryFilter("USERNAME", jobInfo.getUsername());
  job.addOtherInfo("JOB_QUEUE_NAME", jobInfo.getJobQueueName());
  job.addOtherInfo("SUBMIT_TIME", jobInfo.getSubmitTime());
  job.addOtherInfo("LAUNCH_TIME", jobInfo.getLaunchTime());
  job.addOtherInfo("FINISH_TIME", jobInfo.getFinishTime());
  job.addOtherInfo("JOB_STATUS", jobInfo.getJobStatus());
  job.addOtherInfo("PRIORITY", jobInfo.getPriority());
  job.addOtherInfo("TOTAL_MAPS", jobInfo.getTotalMaps());
  job.addOtherInfo("TOTAL_REDUCES", jobInfo.getTotalReduces());
  job.addOtherInfo("UBERIZED", jobInfo.getUberized());
  job.addOtherInfo("ERROR_INFO", jobInfo.getErrorInfo());

  LOG.info("converted job " + jobInfo.getJobId() + " to a timeline entity");
  return job;
}

@Override
public ContainerReport getContainer(ContainerId containerId)
    throws YarnException, IOException {
  ApplicationReportExt app = getApplication(
      containerId.getApplicationAttemptId().getApplicationId(),
      ApplicationReportField.USER_AND_ACLS);
  checkAccess(app);
  TimelineEntity entity = timelineDataManager.getEntity(
      ContainerMetricsConstants.ENTITY_TYPE,
      containerId.toString(), EnumSet.allOf(Field.class),
      UserGroupInformation.getLoginUser());
  if (entity == null) {
    throw new ContainerNotFoundException(
        "The entity for container " + containerId +
        " doesn't exist in the timeline store");
  } else {
    return convertToContainerReport(
        entity, serverHttpAddress, app.appReport.getUser());
  }
}

@Test
public void testPutTimelineEntities() throws Exception {
  KerberosTestUtils.doAs(HTTP_USER + "/localhost", new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      TimelineClient client = createTimelineClientForUGI();
      TimelineEntity entityToStore = new TimelineEntity();
      entityToStore.setEntityType(
          TestTimelineAuthenticationFilter.class.getName());
      entityToStore.setEntityId("entity1");
      entityToStore.setStartTime(0L);
      TimelinePutResponse putResponse = client.putEntities(entityToStore);
      Assert.assertEquals(0, putResponse.getErrors().size());
      TimelineEntity entityToRead =
          testTimelineServer.getTimelineStore().getEntity(
              "entity1", TestTimelineAuthenticationFilter.class.getName(), null);
      Assert.assertNotNull(entityToRead);
      return null;
    }
  });
}

@Test
public void testPutTimelineEntities() throws Exception {
  KerberosTestUtils.doAs(HTTP_USER + "/localhost", new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      TimelineClient client = createTimelineClientForUGI();
      TimelineEntity entityToStore = new TimelineEntity();
      entityToStore.setEntityType(
          TestTimelineAuthenticationFilter.class.getName());
      entityToStore.setEntityId("entity1");
      entityToStore.setStartTime(0L);
      TimelinePutResponse putResponse = client.putEntities(entityToStore);
      Assert.assertEquals(0, putResponse.getErrors().size());
      TimelineEntity entityToRead =
          testTimelineServer.getTimelineStore().getEntity(
              "entity1", TestTimelineAuthenticationFilter.class.getName(), null);
      Assert.assertNotNull(entityToRead);
      return null;
    }
  });
}

@Test
public void testGetEntity() throws Exception {
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1").path("timeline")
      .path("type_1").path("id_1")
      .accept(MediaType.APPLICATION_JSON)
      .get(ClientResponse.class);
  assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
  TimelineEntity entity = response.getEntity(TimelineEntity.class);
  Assert.assertNotNull(entity);
  Assert.assertEquals("id_1", entity.getEntityId());
  Assert.assertEquals("type_1", entity.getEntityType());
  Assert.assertEquals(123l, entity.getStartTime().longValue());
  Assert.assertEquals(2, entity.getEvents().size());
  Assert.assertEquals(4, entity.getPrimaryFilters().size());
  Assert.assertEquals(4, entity.getOtherInfo().size());
}