AmbariInfraWithStormLogSearch implements TopologyLogSearch { @Override public EventSearchResult searchEvent(EventSearchCriteria criteria) { SolrQuery query = new SolrQuery(); String searchString = criteria.getSearchString(); List<String> queryStrings = new ArrayList<>(); addQueryStringToListOnlyIfAvailable(queryStrings, searchString, COLUMN_NAME_STREAMLINE_EVENT_KEYVALUES); addQueryStringToListOnlyIfAvailable(queryStrings, searchString, COLUMN_NAME_STREAMLINE_EVENT_HEADERS); addQueryStringToListOnlyIfAvailable(queryStrings, searchString, COLUMN_NAME_STREAMLINE_EVENT_AUX_KEYVALUES); String queryString = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_ID, buildValue(null)); if (!queryStrings.isEmpty()) { queryString += " AND (" + String.join(" OR ", queryStrings) + ")"; } query.setQuery(queryString); query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_TYPE, COLUMN_VALUE_TYPE_EVENT)); query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID, buildValue(criteria.getAppId()))); query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_LOG_TIME, buildDateRangeValue(criteria.getFrom(), criteria.getTo()))); List<String> componentNames = criteria.getComponentNames(); if (componentNames != null && !componentNames.isEmpty()) { query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME, buildORValues(componentNames))); } String searchEventId = criteria.getSearchEventId(); if (searchEventId != null) { String queryToEventId = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_ID, buildValue(searchEventId)); String queryToRootIds = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET, buildValue("*" + searchEventId + "*")); String queryToParentIds = buildColumnAndValue(COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET, buildValue("*" + searchEventId + "*")); query.addFilterQuery(queryToEventId + " OR " + queryToRootIds + " OR " + queryToParentIds); } if (criteria.getAscending() == null || criteria.getAscending()) { query.addSort(COLUMN_NAME_LOG_TIME, SolrQuery.ORDER.asc); } else { query.addSort(COLUMN_NAME_LOG_TIME, SolrQuery.ORDER.desc); } if (criteria.getStart() != null) { query.setStart(criteria.getStart()); } if (criteria.getLimit() != null) { query.setRows(criteria.getLimit()); } LOG.debug("Querying to Solr: query => {}", query); long numFound; List<EventSearchResult.Event> results = new ArrayList<>(); try { QueryResponse response = solr.query(query); SolrDocumentList docList = response.getResults(); numFound = docList.getNumFound(); for (SolrDocument document : docList) { String appId = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID); String componentName = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME); String eventId = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_ID); String rootIds = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET); String parentIds = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET); String keyValues = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_KEYVALUES); String headers = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_HEADERS); String auxKeyValues = (String) document.getFieldValue(COLUMN_NAME_STREAMLINE_EVENT_AUX_KEYVALUES); Date logDate = (Date) document.getFieldValue(COLUMN_NAME_LOG_TIME); long timestamp = logDate.toInstant().toEpochMilli(); EventSearchResult.Event event = new EventSearchResult.Event(appId, componentName, eventId, rootIds, parentIds, keyValues, headers, auxKeyValues, timestamp); results.add(event); } } catch (SolrServerException | IOException e) { throw new RuntimeException(e); } return new EventSearchResult(numFound, results); } AmbariInfraWithStormLogSearch(); @Override void init(Map<String, Object> conf); @Override LogSearchResult search(LogSearchCriteria logSearchCriteria); @Override EventSearchResult searchEvent(EventSearchCriteria criteria); static final String COLUMN_NAME_TYPE; static final String COLUMN_VALUE_TYPE_WORKER_LOG; static final String COLUMN_VALUE_TYPE_EVENT; static final String COLUMN_NAME_STREAMLINE_TOPOLOGY_ID; static final String COLUMN_NAME_STREAMLINE_COMPONENT_NAME; static final String COLUMN_NAME_STORM_WORKER_PORT; static final String COLUMN_NAME_HOST; static final String COLUMN_NAME_LOG_TIME; static final String COLUMN_NAME_LOG_LEVEL; static final String COLUMN_NAME_LOG_MESSAGE; static final String COLUMN_NAME_STREAMLINE_EVENT_ID; static final String COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET; static final String COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET; static final String COLUMN_NAME_STREAMLINE_EVENT_KEYVALUES; static final String COLUMN_NAME_STREAMLINE_EVENT_HEADERS; static final String COLUMN_NAME_STREAMLINE_EVENT_AUX_KEYVALUES; static final String DEFAULT_COLLECTION_NAME; }
@Test public void testEventSearchWithMinimumParameters() throws Exception { stubSolrUrl(); EventSearchCriteria eventSearchCriteria = new EventSearchCriteria.Builder(TEST_APP_ID, TEST_FROM, TEST_TO).build(); EventSearchResult result = logSearch.searchEvent(eventSearchCriteria); verifyEventSearchResults(result); String dateRangeValue = "%s:[%s+TO+%s]"; Instant fromInstant = Instant.ofEpochMilli(TEST_FROM); Instant toInstant = Instant.ofEpochMilli(TEST_TO); dateRangeValue = String.format(dateRangeValue, COLUMN_NAME_LOG_TIME, fromInstant.toString(), toInstant.toString()); List<LoggedRequest> requests = wireMockRule.findAll(getRequestedFor(urlPathEqualTo(STUB_REQUEST_API_PATH))); assertEquals(1, requests.size()); LoggedRequest request = requests.get(0); QueryParameter qParam = request.queryParameter("q"); assertTrue(qParam.containsValue(COLUMN_NAME_STREAMLINE_EVENT_ID + ":*")); QueryParameter fqParam = request.queryParameter("fq"); assertTrue(fqParam.containsValue(COLUMN_NAME_TYPE + ":" + COLUMN_VALUE_TYPE_EVENT)); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID + ":" + TEST_APP_ID)); assertTrue(fqParam.containsValue(dateRangeValue)); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_COMPONENT_NAME))); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_EVENT_ID))); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET))); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET))); QueryParameter sortParam = request.queryParameter("sort"); assertTrue(sortParam.containsValue(COLUMN_NAME_LOG_TIME + "+asc")); }
@Test public void testEventSearchWithComponentNamesAndStartAndLimitAndDescendingParameters() throws Exception { stubSolrUrl(); int testStart = 100; int testLimit = 2000; List<String> testComponentNames = Collections.singletonList("SOURCE"); EventSearchCriteria eventSearchCriteria = new EventSearchCriteria.Builder(TEST_APP_ID, TEST_FROM, TEST_TO) .setComponentNames(testComponentNames).setAscending(false).setStart(testStart).setLimit(testLimit).build(); EventSearchResult result = logSearch.searchEvent(eventSearchCriteria); verifyEventSearchResults(result); String dateRangeValue = "%s:[%s+TO+%s]"; Instant fromInstant = Instant.ofEpochMilli(TEST_FROM); Instant toInstant = Instant.ofEpochMilli(TEST_TO); dateRangeValue = String.format(dateRangeValue, COLUMN_NAME_LOG_TIME, fromInstant.toString(), toInstant.toString()); List<LoggedRequest> requests = wireMockRule.findAll(getRequestedFor(urlPathEqualTo(STUB_REQUEST_API_PATH))); assertEquals(1, requests.size()); LoggedRequest request = requests.get(0); QueryParameter qParam = request.queryParameter("q"); assertTrue(qParam.containsValue(COLUMN_NAME_STREAMLINE_EVENT_ID + ":*")); QueryParameter fqParam = request.queryParameter("fq"); assertTrue(fqParam.containsValue(COLUMN_NAME_TYPE + ":" + COLUMN_VALUE_TYPE_EVENT)); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID + ":" + TEST_APP_ID)); assertTrue(fqParam.containsValue(dateRangeValue)); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME + ":" + testComponentNames.get(0))); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_EVENT_ID))); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET))); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET))); QueryParameter sortParam = request.queryParameter("sort"); assertTrue(sortParam.containsValue(COLUMN_NAME_LOG_TIME + "+desc")); QueryParameter startParam = request.queryParameter("start"); assertTrue(startParam.containsValue(String.valueOf(testStart))); QueryParameter rowsParam = request.queryParameter("rows"); assertTrue(rowsParam.containsValue(String.valueOf(testLimit))); }
@Test public void testEventSearchWithEventId() throws Exception { stubSolrUrl(); String testEventId = "b7715c60-74ad-43dd-814a-8a40403a31bc"; EventSearchCriteria eventSearchCriteria = new EventSearchCriteria.Builder(TEST_APP_ID, TEST_FROM, TEST_TO) .setSearchEventId(testEventId).build(); EventSearchResult result = logSearch.searchEvent(eventSearchCriteria); verifyEventSearchResults(result); String dateRangeValue = "%s:[%s+TO+%s]"; Instant fromInstant = Instant.ofEpochMilli(TEST_FROM); Instant toInstant = Instant.ofEpochMilli(TEST_TO); dateRangeValue = String.format(dateRangeValue, COLUMN_NAME_LOG_TIME, fromInstant.toString(), toInstant.toString()); List<LoggedRequest> requests = wireMockRule.findAll(getRequestedFor(urlPathEqualTo(STUB_REQUEST_API_PATH))); assertEquals(1, requests.size()); LoggedRequest request = requests.get(0); QueryParameter qParam = request.queryParameter("q"); assertTrue(qParam.containsValue(COLUMN_NAME_STREAMLINE_EVENT_ID + ":*")); QueryParameter fqParam = request.queryParameter("fq"); assertTrue(fqParam.containsValue(COLUMN_NAME_TYPE + ":" + COLUMN_VALUE_TYPE_EVENT)); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID + ":" + TEST_APP_ID)); assertTrue(fqParam.containsValue(dateRangeValue)); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_COMPONENT_NAME))); String expectedEventIdQuery = COLUMN_NAME_STREAMLINE_EVENT_ID + ":" + testEventId; expectedEventIdQuery += "+OR+" + COLUMN_NAME_STREAMLINE_EVENT_ROOT_ID_SET + ":*" + testEventId + "*"; expectedEventIdQuery += "+OR+" + COLUMN_NAME_STREAMLINE_EVENT_PARENT_ID_SET + ":*" + testEventId + "*"; assertTrue(fqParam.containsValue(expectedEventIdQuery)); QueryParameter sortParam = request.queryParameter("sort"); assertTrue(sortParam.containsValue(COLUMN_NAME_LOG_TIME + "+asc")); }
@Test public void testEventSearchWithKeyValuesQueryAndHeadersQuery() throws Exception { stubSolrUrl(); String searchQuery = "hello=world"; EventSearchCriteria eventSearchCriteria = new EventSearchCriteria.Builder(TEST_APP_ID, TEST_FROM, TEST_TO) .setSearchString(searchQuery).build(); EventSearchResult result = logSearch.searchEvent(eventSearchCriteria); verifyEventSearchResults(result); String dateRangeValue = "%s:[%s+TO+%s]"; Instant fromInstant = Instant.ofEpochMilli(TEST_FROM); Instant toInstant = Instant.ofEpochMilli(TEST_TO); dateRangeValue = String.format(dateRangeValue, COLUMN_NAME_LOG_TIME, fromInstant.toString(), toInstant.toString()); List<LoggedRequest> requests = wireMockRule.findAll(getRequestedFor(urlPathEqualTo(STUB_REQUEST_API_PATH))); assertEquals(1, requests.size()); LoggedRequest request = requests.get(0); QueryParameter qParam = request.queryParameter("q"); String expectedQuery = COLUMN_NAME_STREAMLINE_EVENT_ID + ":*"; expectedQuery += "+AND+("; expectedQuery += COLUMN_NAME_STREAMLINE_EVENT_KEYVALUES + ":" + searchQuery; expectedQuery += "+OR+" + COLUMN_NAME_STREAMLINE_EVENT_HEADERS + ":" + searchQuery; expectedQuery += "+OR+" + COLUMN_NAME_STREAMLINE_EVENT_AUX_KEYVALUES + ":" + searchQuery; expectedQuery += ")"; assertTrue(qParam.containsValue(expectedQuery)); QueryParameter fqParam = request.queryParameter("fq"); assertTrue(fqParam.containsValue(COLUMN_NAME_TYPE + ":" + COLUMN_VALUE_TYPE_EVENT)); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID + ":" + TEST_APP_ID)); assertTrue(fqParam.containsValue(dateRangeValue)); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_COMPONENT_NAME))); QueryParameter sortParam = request.queryParameter("sort"); assertTrue(sortParam.containsValue(COLUMN_NAME_LOG_TIME + "+asc")); }