ElasticSearchConsumerOffsetMetricsService extends DefaultService implements MetricStorageService { @Override public void putMetrics(List<Metric> metrics) { requireArgument(metrics.size() > 0, "Cannot push empty list of metrics to ES."); try { final long start = System.currentTimeMillis(); ConsumerOffsetRecordList indexRecordList = new ConsumerOffsetRecordList(ConsumerOffsetMetric.convertToConsumerOffsetMetrics(metrics), idgenHashAlgo); String requestBody = metricMapper.writeValueAsString(indexRecordList); Set<ConsumerOffsetMetric> failedRecords = new HashSet<>(); ElasticSearchUtils.PutResponse putResponse = ElasticSearchUtils.performESRequest(esRestClient, requestUrl, requestBody); final long latency = System.currentTimeMillis() - start; monitorService.modifyCounter(MonitorService.Counter.CONSUMER_OFFSET_RECORDS_WRITE_LATENCY, latency, null); if(putResponse.isErrors()) { for(ElasticSearchUtils.PutResponse.Item item : putResponse.getItems()) { if (item.getIndex() != null && item.getIndex().getStatus() != HttpStatus.SC_CREATED) { logger.warn("Failed to add record {} to index. Reason: {}", indexRecordList.getRecord(item.getIndex().get_id()), metricMapper.writeValueAsString(item.getIndex().getError())); failedRecords.add(indexRecordList.getRecord(item.getIndex().get_id())); } } } if (failedRecords.size() > 0) { logger.warn("{} records were not written to Consumer Offset ES", failedRecords.size()); monitorService.modifyCounter(MonitorService.Counter.CONSUMER_OFFSET_RECORDS_WRITE_FAILURES, failedRecords.size(), null); } } catch (IOException e) { throw new SystemException("Failed to index consumer offset metric to ES.", e); } } @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 testPutMetricsUsingOffsetIndex() throws IOException { Long currentTime = System.currentTimeMillis(); ElasticSearchConsumerOffsetMetricsService spyService = _initializeSpyService(esConsumerOffsetMetricsService, successReply, true); List<Metric> metrics = new ArrayList<>(); Metric record1 = new Metric("scope", "metric"); record1.setTag("topic", "topicV"); record1.addDatapoint(currentTime, 0.0); metrics.add(record1); Metric record2 = new Metric("scope2", "metric2"); record2.setTag("topic", "topicV"); record2.setTag("groupId", "groupId"); record2.addDatapoint(currentTime, 1.0); metrics.add(record2); spyService.putMetrics(metrics); ArgumentCaptor<String> requestCaptorUrl = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> requestCaptorBody = ArgumentCaptor.forClass(String.class); verifyStatic(ElasticSearchUtils.class, times(1)); ElasticSearchUtils.performESRequest(eq(restClient), requestCaptorUrl.capture(), requestCaptorBody.capture()); String expectedURL = "_bulk"; assertEquals(expectedURL, requestCaptorUrl.getValue()); String[] lines = requestCaptorBody.getValue().split("\\r?\\n"); String expectedIndexName = getExpectedIndexName(); JsonNode root = mapper.readTree(lines[0]); String actualIndexName = root.get("index").get("_index").asText(); assertEquals(expectedIndexName, actualIndexName); String expectedSerializedMetric1 = "{\"metric\":\"metric\",\"topic\":\"topicV\",\"value\":\"0.0\",\"ts\":\"" + currentTime + "\",\"tags\":\"{\\\"service\\\":\\\"scope\\\"}\"}"; String expectedSerializedMetric2 = "{\"metric\":\"metric2\",\"topic\":\"topicV\",\"value\":\"1.0\",\"ts\":\"" + currentTime + "\",\"tags\":\"{\\\"groupId\\\":\\\"groupId\\\",\\\"service\\\":\\\"scope2\\\"}\"}"; List<String> expectedOutput = new ArrayList<>(Arrays.asList(expectedSerializedMetric1, expectedSerializedMetric2)); Collections.sort(expectedOutput); List<String> actualOutput = new ArrayList<>(Arrays.asList(lines[1], lines[3])); Collections.sort(actualOutput); assertEquals(Arrays.asList(expectedOutput), Arrays.asList(actualOutput)); }