LogInfo { public long parseForStore(TimelineDataManager tdm, Path appDirPath, boolean appCompleted, JsonFactory jsonFactory, ObjectMapper objMapper, FileSystem fs) throws IOException { LOG.debug("Parsing for log dir {} on attempt {}", appDirPath, attemptDirName); Path logPath = getPath(appDirPath); FileStatus status = fs.getFileStatus(logPath); long numParsed = 0; if (status != null) { long startTime = Time.monotonicNow(); try { LOG.debug("Parsing {} at offset {}", logPath, offset); long count = parsePath(tdm, logPath, appCompleted, jsonFactory, objMapper, fs); LOG.info("Parsed {} entities from {} in {} msec", count, logPath, Time.monotonicNow() - startTime); numParsed += count; } catch (RuntimeException e) { if (e.getCause() instanceof JsonParseException && (status.getLen() > 0 || offset > 0)) { LOG.info("Log {} appears to be corrupted. Skip. ", logPath); } } } else { LOG.warn("{} no longer exists. Skip for scanning. ", logPath); } return numParsed; } LogInfo(String attemptDirName, String file, String owner); String getAttemptDirName(); long getOffset(); void setOffset(long newOffset); Path getPath(Path rootPath); String getFilename(); boolean matchesGroupId(TimelineEntityGroupId groupId); long parseForStore(TimelineDataManager tdm, Path appDirPath,
      boolean appCompleted, JsonFactory jsonFactory, ObjectMapper objMapper,
      FileSystem fs); static final String ENTITY_FILE_NAME_DELIMITERS; }
@Test public void testParseEntity() throws Exception { TimelineDataManager tdm = PluginStoreTestUtils.getTdmWithMemStore(config); EntityLogInfo testLogInfo = new EntityLogInfo(TEST_ATTEMPT_DIR_NAME, TEST_ENTITY_FILE_NAME, UserGroupInformation.getLoginUser().getUserName()); testLogInfo.parseForStore(tdm, getTestRootPath(), true, jsonFactory, objMapper, fs); PluginStoreTestUtils.verifyTestEntities(tdm); TimelineEntity entityNew = PluginStoreTestUtils .createEntity("id_3", "type_3", 789l, null, null, null, null, "domain_id_1"); TimelineEntities entityList = new TimelineEntities(); entityList.addEntity(entityNew); writeEntitiesLeaveOpen(entityList, new Path(getTestRootPath(TEST_ATTEMPT_DIR_NAME), TEST_ENTITY_FILE_NAME)); testLogInfo.parseForStore(tdm, getTestRootPath(), true, jsonFactory, objMapper, fs); TimelineEntity entity3 = tdm.getEntity(entityNew.getEntityType(), entityNew.getEntityId(), EnumSet.allOf(TimelineReader.Field.class), UserGroupInformation.getLoginUser()); assertNotNull(entity3); assertEquals("Failed to read out entity new", entityNew.getStartTime(), entity3.getStartTime()); tdm.close(); }
@Test public void testParseBrokenEntity() throws Exception { TimelineDataManager tdm = PluginStoreTestUtils.getTdmWithMemStore(config); EntityLogInfo testLogInfo = new EntityLogInfo(TEST_ATTEMPT_DIR_NAME, TEST_BROKEN_FILE_NAME, UserGroupInformation.getLoginUser().getUserName()); DomainLogInfo domainLogInfo = new DomainLogInfo(TEST_ATTEMPT_DIR_NAME, TEST_BROKEN_FILE_NAME, UserGroupInformation.getLoginUser().getUserName()); testLogInfo.parseForStore(tdm, getTestRootPath(), true, jsonFactory, objMapper, fs); domainLogInfo.parseForStore(tdm, getTestRootPath(), true, jsonFactory, objMapper, fs); tdm.close(); }
@Test public void testParseDomain() throws Exception { TimelineDataManager tdm = PluginStoreTestUtils.getTdmWithMemStore(config); DomainLogInfo domainLogInfo = new DomainLogInfo(TEST_ATTEMPT_DIR_NAME, TEST_DOMAIN_FILE_NAME, UserGroupInformation.getLoginUser().getUserName()); domainLogInfo.parseForStore(tdm, getTestRootPath(), true, jsonFactory, objMapper, fs); TimelineDomain resultDomain = tdm.getDomain("domain_1", UserGroupInformation.getLoginUser()); assertNotNull(resultDomain); assertEquals(testDomain.getReaders(), resultDomain.getReaders()); assertEquals(testDomain.getOwner(), resultDomain.getOwner()); assertEquals(testDomain.getDescription(), resultDomain.getDescription()); }