FileSystemTimelineReaderImpl extends AbstractService implements TimelineReader { private Set<TimelineEntity> getEntities(Path dir, String entityType, TimelineEntityFilters filters, TimelineDataToRetrieve dataToRetrieve) throws IOException { Map<Long, Set<TimelineEntity>> sortedEntities = new TreeMap<>( new Comparator<Long>() { @Override public int compare(Long l1, Long l2) { return l2.compareTo(l1); } } ); if (dir != null) { RemoteIterator<LocatedFileStatus> fileStatuses = fs.listFiles(dir, false); if (fileStatuses != null) { while (fileStatuses.hasNext()) { LocatedFileStatus locatedFileStatus = fileStatuses.next(); Path entityFile = locatedFileStatus.getPath(); if (!entityFile.getName() .contains(TIMELINE_SERVICE_STORAGE_EXTENSION)) { continue; } try (BufferedReader reader = new BufferedReader( new InputStreamReader(fs.open(entityFile), Charset.forName("UTF-8")))) { TimelineEntity entity = readEntityFromFile(reader); if (!entity.getType().equals(entityType)) { continue; } if (!isTimeInRange(entity.getCreatedTime(), filters.getCreatedTimeBegin(), filters.getCreatedTimeEnd())) { continue; } if (filters.getRelatesTo() != null && !filters.getRelatesTo().getFilterList().isEmpty() && !TimelineStorageUtils.matchRelatesTo(entity, filters.getRelatesTo())) { continue; } if (filters.getIsRelatedTo() != null && !filters.getIsRelatedTo().getFilterList().isEmpty() && !TimelineStorageUtils.matchIsRelatedTo(entity, filters.getIsRelatedTo())) { continue; } if (filters.getInfoFilters() != null && !filters.getInfoFilters().getFilterList().isEmpty() && !TimelineStorageUtils.matchInfoFilters(entity, filters.getInfoFilters())) { continue; } if (filters.getConfigFilters() != null && !filters.getConfigFilters().getFilterList().isEmpty() && !TimelineStorageUtils.matchConfigFilters(entity, filters.getConfigFilters())) { continue; } if (filters.getMetricFilters() != null && !filters.getMetricFilters().getFilterList().isEmpty() && !TimelineStorageUtils.matchMetricFilters(entity, filters.getMetricFilters())) { continue; } if (filters.getEventFilters() != null && !filters.getEventFilters().getFilterList().isEmpty() && !TimelineStorageUtils.matchEventFilters(entity, filters.getEventFilters())) { continue; } TimelineEntity entityToBeReturned = createEntityToBeReturned( entity, dataToRetrieve.getFieldsToRetrieve()); Set<TimelineEntity> entitiesCreatedAtSameTime = sortedEntities.get(entityToBeReturned.getCreatedTime()); if (entitiesCreatedAtSameTime == null) { entitiesCreatedAtSameTime = new HashSet<TimelineEntity>(); } entitiesCreatedAtSameTime.add(entityToBeReturned); sortedEntities.put(entityToBeReturned.getCreatedTime(), entitiesCreatedAtSameTime); } } } } Set<TimelineEntity> entities = new HashSet<TimelineEntity>(); long entitiesAdded = 0; for (Set<TimelineEntity> entitySet : sortedEntities.values()) { for (TimelineEntity entity : entitySet) { entities.add(entity); ++entitiesAdded; if (entitiesAdded >= filters.getLimit()) { return entities; } } } return entities; } FileSystemTimelineReaderImpl(); static T getTimelineRecordFromJSON(
      String jsonString, Class<T> clazz); @Override void serviceInit(Configuration conf); @Override TimelineEntity getEntity(TimelineReaderContext context,
      TimelineDataToRetrieve dataToRetrieve); @Override Set<TimelineEntity> getEntities(TimelineReaderContext context,
      TimelineEntityFilters filters, TimelineDataToRetrieve dataToRetrieve); @Override Set<String> getEntityTypes(TimelineReaderContext context); @Override TimelineHealth getHealthStatus(); static final String TIMELINE_SERVICE_STORAGE_DIR_ROOT; }
@Test public void testGetAllEntities() throws Exception { Set<TimelineEntity> result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().build(), new TimelineDataToRetrieve(null, null, EnumSet.of(Field.ALL), null, null, null)); Assert.assertEquals(4, result.size()); }
@Test public void testGetEntitiesWithLimit() throws Exception { Set<TimelineEntity> result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().entityLimit(2L).build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_4")) { Assert.fail("Entity not sorted by created time"); } } result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().entityLimit(3L).build(), new TimelineDataToRetrieve()); Assert.assertEquals(3, result.size()); }
@Test public void testGetEntitiesByTimeWindows() throws Exception { Set<TimelineEntity> result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().createdTimeBegin(1425016502030L) .createTimeEnd(1425016502060L).build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_4")) { Assert.fail("Incorrect filtering based on created time range"); } } result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().createTimeEnd(1425016502010L) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(3, result.size()); for (TimelineEntity entity : result) { if (entity.getId().equals("id_4")) { Assert.fail("Incorrect filtering based on created time range"); } } result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().createdTimeBegin(1425016502010L) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_4")) { Assert.fail("Incorrect filtering based on created time range"); } } }
@Test public void testGetFilteredEntities() throws Exception { TimelineFilterList infoFilterList = new TimelineFilterList(); infoFilterList.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", 3.5)); Set<TimelineEntity> result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().infoFilters(infoFilterList).build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_3")) { Assert.fail("Incorrect filtering based on info filters"); } } TimelineFilterList confFilterList = new TimelineFilterList(); confFilterList.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_1", "123")); confFilterList.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_3", "abc")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().configFilters(confFilterList) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_3")) { Assert.fail("Incorrect filtering based on config filters"); } } TimelineFilterList eventFilters = new TimelineFilterList(); eventFilters.addFilter( new TimelineExistsFilter(TimelineCompareOp.EQUAL, "event_2")); eventFilters.addFilter( new TimelineExistsFilter(TimelineCompareOp.EQUAL, "event_4")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().eventFilters(eventFilters).build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_3")) { Assert.fail("Incorrect filtering based on event filters"); } } TimelineFilterList metricFilterList = new TimelineFilterList(); metricFilterList.addFilter(new TimelineCompareFilter( TimelineCompareOp.GREATER_OR_EQUAL, "metric3", 0L)); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().metricFilters(metricFilterList) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) { Assert.fail("Incorrect filtering based on metric filters"); } } TimelineFilterList list1 = new TimelineFilterList(); list1.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_1", "129")); list1.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_3", "def")); TimelineFilterList list2 = new TimelineFilterList(); list2.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_2", "23")); list2.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_3", "abc")); TimelineFilterList confFilterList1 = new TimelineFilterList(Operator.OR, list1, list2); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().configFilters(confFilterList1) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) { Assert.fail("Incorrect filtering based on config filters"); } } TimelineFilterList list3 = new TimelineFilterList(); list3.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.NOT_EQUAL, "config_1", "123")); list3.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.NOT_EQUAL, "config_3", "abc")); TimelineFilterList list4 = new TimelineFilterList(); list4.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "config_2", "23")); TimelineFilterList confFilterList2 = new TimelineFilterList(Operator.OR, list3, list4); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().configFilters(confFilterList2) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) { Assert.fail("Incorrect filtering based on config filters"); } } TimelineFilterList confFilterList3 = new TimelineFilterList(); confFilterList3.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.NOT_EQUAL, "config_1", "127")); confFilterList3.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.NOT_EQUAL, "config_3", "abc")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().configFilters(confFilterList3) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for(TimelineEntity entity : result) { if (!entity.getId().equals("id_2")) { Assert.fail("Incorrect filtering based on config filters"); } } TimelineFilterList confFilterList4 = new TimelineFilterList(); confFilterList4.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.EQUAL, "config_dummy", "dummy")); confFilterList4.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.EQUAL, "config_3", "def")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().configFilters(confFilterList4) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(0, result.size()); TimelineFilterList confFilterList5 = new TimelineFilterList(Operator.OR); confFilterList5.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.EQUAL, "config_dummy", "dummy")); confFilterList5.addFilter(new TimelineKeyValueFilter( TimelineCompareOp.EQUAL, "config_3", "def")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().configFilters(confFilterList5) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_2")) { Assert.fail("Incorrect filtering based on config filters"); } } TimelineFilterList list6 = new TimelineFilterList(); list6.addFilter(new TimelineCompareFilter( TimelineCompareOp.GREATER_THAN, "metric1", 200)); list6.addFilter(new TimelineCompareFilter( TimelineCompareOp.EQUAL, "metric3", 23)); TimelineFilterList list7 = new TimelineFilterList(); list7.addFilter(new TimelineCompareFilter( TimelineCompareOp.GREATER_OR_EQUAL, "metric2", 74)); TimelineFilterList metricFilterList1 = new TimelineFilterList(Operator.OR, list6, list7); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().metricFilters(metricFilterList1) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_2") && !entity.getId().equals("id_3")) { Assert.fail("Incorrect filtering based on metric filters"); } } TimelineFilterList metricFilterList2 = new TimelineFilterList(); metricFilterList2.addFilter(new TimelineCompareFilter( TimelineCompareOp.LESS_THAN, "metric2", 70)); metricFilterList2.addFilter(new TimelineCompareFilter( TimelineCompareOp.LESS_OR_EQUAL, "metric3", 23)); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().metricFilters(metricFilterList2) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1")) { Assert.fail("Incorrect filtering based on metric filters"); } } TimelineFilterList metricFilterList3 = new TimelineFilterList(); metricFilterList3.addFilter(new TimelineCompareFilter( TimelineCompareOp.LESS_THAN, "dummy_metric", 30)); metricFilterList3.addFilter(new TimelineCompareFilter( TimelineCompareOp.LESS_OR_EQUAL, "metric3", 23)); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().metricFilters(metricFilterList3) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(0, result.size()); TimelineFilterList metricFilterList4 = new TimelineFilterList(Operator.OR); metricFilterList4.addFilter(new TimelineCompareFilter( TimelineCompareOp.LESS_THAN, "dummy_metric", 30)); metricFilterList4.addFilter(new TimelineCompareFilter( TimelineCompareOp.LESS_OR_EQUAL, "metric3", 23)); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().metricFilters(metricFilterList4) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) { Assert.fail("Incorrect filtering based on metric filters"); } } TimelineFilterList metricFilterList5 = new TimelineFilterList(new TimelineCompareFilter( TimelineCompareOp.NOT_EQUAL, "metric2", 74)); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().metricFilters(metricFilterList5) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_2")) { Assert.fail("Incorrect filtering based on metric filters"); } } TimelineFilterList infoFilterList1 = new TimelineFilterList(); infoFilterList1.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", 3.5)); infoFilterList1.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.NOT_EQUAL, "info4", 20)); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().infoFilters(infoFilterList1) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(0, result.size()); TimelineFilterList infoFilterList2 = new TimelineFilterList(Operator.OR); infoFilterList2.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", 3.5)); infoFilterList2.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info1", "val1")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().infoFilters(infoFilterList2) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_3")) { Assert.fail("Incorrect filtering based on info filters"); } } TimelineFilterList infoFilterList3 = new TimelineFilterList(); infoFilterList3.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "dummy_info", 1)); infoFilterList3.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", "val5")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().infoFilters(infoFilterList3) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(0, result.size()); TimelineFilterList infoFilterList4 = new TimelineFilterList(Operator.OR); infoFilterList4.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "dummy_info", 1)); infoFilterList4.addFilter( new TimelineKeyValueFilter(TimelineCompareOp.EQUAL, "info2", "val5")); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().infoFilters(infoFilterList4) .build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1")) { Assert.fail("Incorrect filtering based on info filters"); } } }
@Test public void testGetEntitiesByRelations() throws Exception { TimelineFilterList relatesTo = new TimelineFilterList(Operator.OR); Set<Object> relatesToIds = new HashSet<Object>(Arrays.asList((Object)"flow1")); relatesTo.addFilter(new TimelineKeyValuesFilter( TimelineCompareOp.EQUAL, "flow", relatesToIds)); Set<TimelineEntity> result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().relatesTo(relatesTo).build(), new TimelineDataToRetrieve()); Assert.assertEquals(1, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1")) { Assert.fail("Incorrect filtering based on relatesTo"); } } TimelineFilterList isRelatedTo = new TimelineFilterList(Operator.OR); Set<Object> isRelatedToIds = new HashSet<Object>(Arrays.asList((Object)"tid1_2")); isRelatedTo.addFilter(new TimelineKeyValuesFilter( TimelineCompareOp.EQUAL, "type1", isRelatedToIds)); result = reader.getEntities( new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1", "app", null), new TimelineEntityFilters.Builder().isRelatedTo(isRelatedTo).build(), new TimelineDataToRetrieve()); Assert.assertEquals(2, result.size()); for (TimelineEntity entity : result) { if (!entity.getId().equals("id_1") && !entity.getId().equals("id_3")) { Assert.fail("Incorrect filtering based on isRelatedTo"); } } }