AmbariInfraWithStormLogSearch implements TopologyLogSearch { @Override public LogSearchResult search(LogSearchCriteria logSearchCriteria) { SolrQuery query = new SolrQuery(); query.setQuery(buildColumnAndValue(COLUMN_NAME_LOG_MESSAGE, buildValue(logSearchCriteria.getSearchString()))); query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_TYPE, COLUMN_VALUE_TYPE_WORKER_LOG)); query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID, buildValue(logSearchCriteria.getAppId()))); query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_LOG_TIME, buildDateRangeValue(logSearchCriteria.getFrom(), logSearchCriteria.getTo()))); List<String> componentNames = logSearchCriteria.getComponentNames(); if (componentNames != null && !componentNames.isEmpty()) { query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME, buildORValues(componentNames))); } List<String> logLevels = logSearchCriteria.getLogLevels(); if (logLevels == null || logLevels.isEmpty()) { logLevels = DEFAULT_LOG_LEVELS; } query.addFilterQuery(buildColumnAndValue(COLUMN_NAME_LOG_LEVEL, buildORValues(logLevels))); if (logSearchCriteria.getAscending() == null || logSearchCriteria.getAscending()) { query.addSort(COLUMN_NAME_LOG_TIME, SolrQuery.ORDER.asc); } else { query.addSort(COLUMN_NAME_LOG_TIME, SolrQuery.ORDER.desc); } if (logSearchCriteria.getStart() != null) { query.setStart(logSearchCriteria.getStart()); } if (logSearchCriteria.getLimit() != null) { query.setRows(logSearchCriteria.getLimit()); } LOG.debug("Querying to Solr: query => {}", query); long numFound; List<LogSearchResult.LogDocument> 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 logLevel = (String) document.getFieldValue(COLUMN_NAME_LOG_LEVEL); String logMessage = (String) document.getFieldValue(COLUMN_NAME_LOG_MESSAGE); String host = (String) document.getFieldValue(COLUMN_NAME_HOST); String port = (String) document.getFieldValue(COLUMN_NAME_STORM_WORKER_PORT); Date logDate = (Date) document.getFieldValue(COLUMN_NAME_LOG_TIME); long timestamp = logDate.toInstant().toEpochMilli(); LogSearchResult.LogDocument logDocument = new LogSearchResult.LogDocument(appId, componentName, logLevel, logMessage, host, port != null ? Integer.parseInt(port) : null, timestamp); results.add(logDocument); } } catch (SolrServerException | IOException e) { throw new RuntimeException(e); } return new LogSearchResult(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 testLogSearchWithMinimumParameters() throws Exception { stubSolrUrl(); LogSearchCriteria logSearchCriteria = new LogSearchCriteria.Builder(TEST_APP_ID, TEST_FROM, TEST_TO).build(); LogSearchResult result = logSearch.search(logSearchCriteria); verifyLogSearchResults(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()); String expectedLogLevels = "(" + String.join("+OR+", AmbariInfraWithStormLogSearch.DEFAULT_LOG_LEVELS) + ")"; 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_LOG_MESSAGE + ":*")); QueryParameter fqParam = request.queryParameter("fq"); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID + ":" + TEST_APP_ID)); assertTrue(fqParam.containsValue(COLUMN_NAME_TYPE + ":" + COLUMN_VALUE_TYPE_WORKER_LOG)); assertTrue(fqParam.containsValue(dateRangeValue)); assertTrue(fqParam.containsValue(COLUMN_NAME_LOG_LEVEL + ":" + expectedLogLevels)); assertFalse(fqParam.hasValueMatching(ValuePattern.containing(COLUMN_NAME_STREAMLINE_COMPONENT_NAME))); QueryParameter sortParam = request.queryParameter("sort"); assertTrue(sortParam.containsValue(COLUMN_NAME_LOG_TIME + "+asc")); }
@Test public void testLogSearchWithFullParameters() throws Exception { stubSolrUrl(); int testStart = 100; int testLimit = 2000; List<String> testLogLevels = Lists.newArrayList("INFO", "DEBUG"); String testSearchString = "helloworld"; List<String> testComponentNames = Lists.newArrayList("testComponent", "testComponent2"); LogSearchCriteria logSearchCriteria = new LogSearchCriteria.Builder(TEST_APP_ID, TEST_FROM, TEST_TO) .setLogLevels(testLogLevels) .setSearchString(testSearchString) .setComponentNames(testComponentNames) .setStart(testStart) .setLimit(testLimit) .setAscending(false) .build(); LogSearchResult result = logSearch.search(logSearchCriteria); verifyLogSearchResults(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()); String expectedComponentNames = "(" + String.join("+OR+", testComponentNames) + ")"; String expectedLogLevels = "(" + String.join("+OR+", testLogLevels) + ")"; 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_LOG_MESSAGE + ":" + testSearchString)); QueryParameter fqParam = request.queryParameter("fq"); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_TOPOLOGY_ID + ":" + TEST_APP_ID)); assertTrue(fqParam.containsValue(COLUMN_NAME_TYPE + ":" + COLUMN_VALUE_TYPE_WORKER_LOG)); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME + ":" + expectedComponentNames)); assertTrue(fqParam.containsValue(COLUMN_NAME_LOG_LEVEL + ":" + expectedLogLevels)); assertTrue(fqParam.containsValue(dateRangeValue)); 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 testLogSearchWithSingleComponentNameAndLogLevelParameters() throws Exception { stubSolrUrl(); int testStart = 100; int testLimit = 2000; List<String> testLogLevels = Collections.singletonList("INFO"); String testSearchString = "helloworld"; List<String> testComponentNames = Collections.singletonList("testComponent"); LogSearchCriteria logSearchCriteria = new LogSearchCriteria.Builder(TEST_APP_ID, TEST_FROM, TEST_TO) .setLogLevels(testLogLevels) .setSearchString(testSearchString) .setComponentNames(testComponentNames) .setStart(testStart) .setLimit(testLimit) .build(); LogSearchResult result = logSearch.search(logSearchCriteria); verifyLogSearchResults(result); List<LoggedRequest> requests = wireMockRule.findAll(getRequestedFor(urlPathEqualTo(STUB_REQUEST_API_PATH))); assertEquals(1, requests.size()); LoggedRequest request = requests.get(0); QueryParameter fqParam = request.queryParameter("fq"); assertTrue(fqParam.containsValue(COLUMN_NAME_STREAMLINE_COMPONENT_NAME + ":" + testComponentNames.get(0))); assertTrue(fqParam.containsValue(COLUMN_NAME_TYPE + ":" + COLUMN_VALUE_TYPE_WORKER_LOG)); assertTrue(fqParam.containsValue(COLUMN_NAME_LOG_LEVEL + ":" + testLogLevels.get(0))); }