ElasticSearchConsumerOffsetMetricsService extends DefaultService implements MetricStorageService { @Override public Map<MetricQuery, List<Metric>> getMetrics(List<MetricQuery> queries) { requireNotDisposed(); requireArgument(queries != null, "ConsumerOffset queries cannot be null."); int from = 0, scrollSize = INDEX_MAX_RESULT_WINDOW, numOfMetrics = 0; Map<MetricQuery, List<Metric> > result = new HashMap<>(); String requestUrl = String.format("/%s-*/_search", INDEX_TEMPLATE_PATTERN_START); try { for (MetricQuery query : queries) { List<ConsumerOffsetMetric> consumerOffsetMetrics = new ArrayList<>(); String queryJson = constructQuery(new MetricQuery(query), from, scrollSize); final long start = System.currentTimeMillis(); Request request = new Request(ElasticSearchUtils.HttpMethod.POST.getName(), requestUrl); request.setEntity(new StringEntity(queryJson, ContentType.APPLICATION_JSON)); Response response = esRestClient.performRequest(request); final long latency = System.currentTimeMillis() - start; monitorService.modifyCounter(MonitorService.Counter.CONSUMER_OFFSET_RECORDS_READ_LATENCY, latency, null); logger.debug("ES get request for consumer offset completed in {} ms", latency); String extractResponse = ElasticSearchUtils.extractResponse(response); ConsumerOffsetRecordList recordList = ElasticSearchUtils.toEntity(extractResponse, new TypeReference<ConsumerOffsetRecordList>() {}, metricMapper); consumerOffsetMetrics.addAll(recordList.getRecords()); result.put(query, ConsumerOffsetMetric.convertToMetrics(consumerOffsetMetrics)); numOfMetrics += consumerOffsetMetrics.size(); if(numOfMetrics >= scrollSize) { logger.error("Maximum metrics limit execeeded for query- " + query.toString()); throw new RuntimeException(MessageFormat.format(EXCEPTION_MESSAGE, scrollSize)); } } } catch(IOException ex) { throw new SystemException(ex); } return result; } @Inject protected  ElasticSearchConsumerOffsetMetricsService(SystemConfiguration config, MonitorService monitorService, ElasticSearchUtils esUtils); @Override void dispose(); @Override void putMetrics(List<Metric> metrics); @Override Map<MetricQuery, List<Metric>> getMetrics(List<MetricQuery> queries); static final int INDEX_MAX_RESULT_WINDOW; static final String DATE_FORMAT; static final String INDEX_FORMAT; static String INDEX_TEMPLATE_PATTERN_START; static final String SCOPE_NAME; static final String METRIC_NAME; static final String requestUrl; }
@Test public void testGetMetrics() throws IOException { MetricQuery metricQuery = new MetricQuery("scope1", "metric1", null, 1557809359L, 1557809599L); metricQuery.setTag("topic", "topic"); metricQuery.setDownsampler(MetricQuery.Aggregator.MAX); metricQuery.setDownsamplingPeriod(5 * 60 * 1000L); List<MetricQuery> queries = new ArrayList<>(); queries.add(metricQuery); ElasticSearchConsumerOffsetMetricsService spyService = _initializeSpyService(esConsumerOffsetMetricsService, getReply, false); Map<MetricQuery, List<Metric>> metricsResult = spyService.getMetrics(queries); String expectedMetric = "[namespace=>null, scope=>ajna.consumer, metric=>metric.consumer.lag, tags=>{topic=mytopic1.name}, datapoints=>{1560902400000=21455.0, 1561075200000=21981.0, 1561248000000=22461.0}, " + "namespace=>null, scope=>ajna.consumer, metric=>metric.consumer.lag, tags=>{topic=mytopic2.name}, datapoints=>{1560902400000=19676.0, 1561075200000=21635.0, 1561248000000=22659.0}]"; assertEquals(expectedMetric, metricsResult.get(metricQuery).toString()); ArgumentCaptor<Request> requestCaptor = ArgumentCaptor.forClass(Request.class); verify(restClient, times(1)).performRequest(requestCaptor.capture()); Request capturedRequest = requestCaptor.getValue(); String queryJson = convertToPrettyJson(EntityUtils.toString(capturedRequest.getEntity())); assertEquals(queryMustTermRange, queryJson); String expectedURL = "/argus-akc-consumer-offset-*/_search"; assertEquals(expectedURL, capturedRequest.getEndpoint()); }
@Test (expected = RuntimeException.class) public void testGetMetricsExceedingLimit(){ MetricQuery metricQuery = new MetricQuery("scope1", "metric1", null, 1557809359L, 1557809599L); metricQuery.setTag("topic", "topic"); metricQuery.setDownsampler(MetricQuery.Aggregator.MAX); metricQuery.setDownsamplingPeriod(5 * 60 * 1000L); List<MetricQuery> queries = new ArrayList<>(); queries.add(metricQuery); ElasticSearchConsumerOffsetMetricsService spyService = null; try { spyService = _initializeSpyService(esConsumerOffsetMetricsService, getReply, false); } catch (IOException e) { fail(); } Whitebox.setInternalState(spyService, "INDEX_MAX_RESULT_WINDOW", 1); spyService.getMetrics(queries); }