ElasticSearchConsumerOffsetMetricsService extends DefaultService implements MetricStorageService { @VisibleForTesting protected String constructQuery(MetricQuery query, int scrollFrom, int size) { query.setStartTimestamp(ElasticSearchUtils.convertTimestampToMillis(query.getStartTimestamp())); query.setEndTimestamp(ElasticSearchUtils.convertTimestampToMillis(query.getEndTimestamp())); ObjectNode aggsNode = constructAggregationNode(query, genericObjectMapper); ObjectNode queryNode = constructQueryNode(query, genericObjectMapper); ObjectNode rootNode = metricMapper.createObjectNode(); if (aggsNode != null && queryNode != null) { rootNode.set("aggs", aggsNode); rootNode.set("query", queryNode); rootNode.put("from", scrollFrom); rootNode.put("size", size); return rootNode.toString(); } else { logger.error("Failed to construct query"); return null; } } @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 testConstructQuery() throws IOException { Map<String, String> tags = new HashMap<>(); tags.put("service", "scope*"); tags.put("groupId", "groupId*"); tags.put("topic", "*"); esConsumerOffsetMetricsService = _initializeSpyService(esConsumerOffsetMetricsService, getReply, false); MetricQuery mQ = new MetricQuery("test", "test", tags, 0L, 1L); mQ.setDownsamplingPeriod((long)(2 * 60 * 1000)); mQ.setDownsampler(MetricQuery.Aggregator.MAX); String actualOutput = esConsumerOffsetMetricsService.constructQuery(mQ, 0, 0); String expectedOutput = "{\n" + " \"aggs\": {\n" + " \"max_topic_offset_per_unit_time_greater_than\": {\n" + " \"terms\": {\n" + " \"field\": \"topic.raw\"\n" + " },\n" + " \"aggs\": {\n" + " \"max_offset_per_unit_time_greater_than\": {\n" + " \"date_histogram\": {\n" + " \"field\": \"ts\",\n" + " \"interval\": \"2m\"\n" + " },\n" + " \"aggs\": {\n" + " \"max_offset_greater_than\": {\n" + " \"max\": {\n" + " \"field\": \"value\"\n" + " }\n" + " }\n" + " }\n" + " }\n" + " }\n" + " }\n" + " },\n" + " \"query\": {\n" + " \"bool\": {\n" + " \"must\": [\n" + " {\n" + " \"range\": {\n" + " \"ts\": {\n" + " \"gte\": \"0\",\n" + " \"lte\": \"1000\"\n" + " }\n" + " }\n" + " }\n" + " ],\n" + " \"filter\": [\n" + " {\n" + " \"regexp\": {\n" + " \"topic.raw\": \".*\"\n" + " }\n" + " },\n" + " {\n" + " \"regexp\": {\n" + " \"tags.raw\": \".*([\\\"]groupId[\\\"]:[\\\"]groupId.*[\\\"]).*\"\n" + " }\n" + " },\n" + " {\n" + " \"regexp\": {\n" + " \"tags.raw\": \".*([\\\"]service[\\\"]:[\\\"]scope.*[\\\"]).*\"\n" + " }\n" + " }\n" + " ]\n" + " }\n" + " },\n" + " \"from\": 0,\n" + " \"size\": 0\n" + "}"; assertEquals(convertToPrettyJson(expectedOutput), convertToPrettyJson(actualOutput)); }