ElasticSearchSchemaService extends AbstractSchemaService { @Override public List<MetricSchemaRecord> getUnique(MetricSchemaRecordQuery query, RecordType type) { requireNotDisposed(); SystemAssert.requireArgument(query != null, "MetricSchemaRecordQuery cannot be null."); long size = (long) query.getLimit() * query.getPage(); SystemAssert.requireArgument(size > 0 && size <= Integer.MAX_VALUE, "(limit * page) must be greater than 0 and atmost Integer.MAX_VALUE"); Map<String, String> tags = new HashMap<>(); tags.put("type", "REGEXP_WITH_AGGREGATION"); long start = System.currentTimeMillis(); String indexName = TAGS_INDEX_NAME; String typeName = TAGS_TYPE_NAME; if (query.isQueryOnlyOnScope() && RecordType.SCOPE.equals(type)) { indexName = SCOPE_INDEX_NAME; typeName = SCOPE_TYPE_NAME; } String requestUrl = new StringBuilder().append("/") .append(indexName) .append("/") .append(typeName) .append("/") .append("_search") .toString(); try { String queryJson = _constructTermAggregationQuery(query, type); _logger.debug("getUnique POST requestUrl {} queryJson {}", requestUrl, queryJson); Request request = new Request(HttpMethod.POST.getName(), requestUrl); request.setEntity(new StringEntity(queryJson, ContentType.APPLICATION_JSON)); Response response = _esRestClient.performRequest(request); String str = extractResponse(response); List<MetricSchemaRecord> records = SchemaService.constructMetricSchemaRecordsForType(toEntity(str, new TypeReference<List<String>>() {}), type); if (query.isQueryOnlyOnScope() && RecordType.SCOPE.equals(type)) { _monitorService.modifyCounter(Counter.SCOPENAMES_QUERY_COUNT, 1, tags); _monitorService.modifyCounter(Counter.SCOPENAMES_QUERY_LATENCY, (System.currentTimeMillis() - start), tags); } else { _monitorService.modifyCounter(Counter.SCHEMARECORDS_QUERY_COUNT, 1, tags); _monitorService.modifyCounter(Counter.SCHEMARECORDS_QUERY_LATENCY, (System.currentTimeMillis() - start), tags); } int fromIndex = query.getLimit() * (query.getPage() - 1); if(records.size() <= fromIndex) { return Collections.emptyList(); } if(records.size() < query.getLimit() * query.getPage()) { return records.subList(fromIndex, records.size()); } else { return records.subList(fromIndex, query.getLimit() * query.getPage()); } } catch (IOException e) { throw new SystemException(e); } } @Inject  ElasticSearchSchemaService(SystemConfiguration config, MonitorService monitorService, ElasticSearchUtils esUtils); @Override void dispose(); @Override Properties getServiceProperties(); @Override List<MetricSchemaRecord> get(MetricSchemaRecordQuery query); @Override List<String> browseUnique(MetricSchemaRecordQuery query, RecordType type, int indexLevel); @Override List<MetricSchemaRecord> getUnique(MetricSchemaRecordQuery query, RecordType type); @Override List<MetricSchemaRecord> keywordSearch(KeywordQuery kq);  }
@Test public void testGetUniqueUsingScopeSchemaIndex() throws IOException { MetricSchemaRecordQuery queryForScope = new MetricSchemaRecordQuery.MetricSchemaRecordQueryBuilder().scope("system*") .metric("*") .tagKey("*") .tagValue("*") .namespace("*") .limit(2) .build(); SchemaService.RecordType scopeType = SchemaService.RecordType.SCOPE; MonitorService mockedMonitor = mock(MonitorService.class); ElasticSearchSchemaService spyService = _initializeSpyService(_esSchemaService, getReply, getReply); spyService.getUnique(queryForScope, scopeType); ArgumentCaptor<Request> captor = ArgumentCaptor.forClass(Request.class); verify(restClient, times(1)).performRequest(captor.capture()); String requestUrl = captor.getValue().getEndpoint(); String queryJson = convertToPrettyJson(EntityUtils.toString(captor.getValue().getEntity())); assertEquals(scopeQuery, queryJson); assertEquals("/scopenames/scope_type/_search", requestUrl); assertTrue(queryForScope.isQueryOnlyOnScope()); assertTrue(queryForScope.isQueryOnlyOnScopeAndMetric()); }
@Test public void testGetUniqueUsingMetricTagvRegexSchemaIndex() throws IOException { MetricSchemaRecordQuery queryForMetric = new MetricSchemaRecordQuery.MetricSchemaRecordQueryBuilder().scope("system") .metric("argus") .tagKey("device") .tagValue("abc*") .namespace("*") .limit(2) .build(); SchemaService.RecordType scopeType = SchemaService.RecordType.TAGV; ElasticSearchSchemaService spyService = _initializeSpyService(_esSchemaService, getReply, getReply); spyService.getUnique(queryForMetric, scopeType); ArgumentCaptor<Request> captor = ArgumentCaptor.forClass(Request.class); verify(restClient, times(1)).performRequest(captor.capture()); String requestUrl = captor.getValue().getEndpoint(); String queryJson = convertToPrettyJson(EntityUtils.toString(captor.getValue().getEntity())); assertEquals(metricQueryTagvRegex, queryJson); assertEquals("/metadata_index/metadata_type/_search", requestUrl); assertFalse(queryForMetric.isQueryOnlyOnScope()); assertFalse(queryForMetric.isQueryOnlyOnScopeAndMetric()); }
@Test public void testGetUniqueUsingMetricNamespaceRegexSchemaIndex() throws IOException { MetricSchemaRecordQuery queryForMetric = new MetricSchemaRecordQuery.MetricSchemaRecordQueryBuilder().scope("system") .metric("argus") .tagKey("*") .tagValue("*") .namespace("common*") .limit(2) .build(); SchemaService.RecordType scopeType = SchemaService.RecordType.NAMESPACE; ElasticSearchSchemaService spyService = _initializeSpyService(_esSchemaService, getReply, getReply); spyService.getUnique(queryForMetric, scopeType); ArgumentCaptor<Request> captor = ArgumentCaptor.forClass(Request.class); verify(restClient, times(1)).performRequest(captor.capture()); String requestUrl = captor.getValue().getEndpoint(); String queryJson = convertToPrettyJson(EntityUtils.toString(captor.getValue().getEntity())); assertEquals(metricQueryNamespaceRegex, queryJson); assertEquals("/metadata_index/metadata_type/_search", requestUrl); assertFalse(queryForMetric.isQueryOnlyOnScope()); assertFalse(queryForMetric.isQueryOnlyOnScopeAndMetric()); }