GraphiteWithStormQuerier extends AbstractTimeSeriesQuerier { @Override public Map<Long, Double> getMetrics(String topologyName, String componentId, String metricName, AggregateFunction aggrFunction, long from, long to) { URI targetUri = composeQueryParameters(topologyName, componentId, metricName, aggrFunction, from, to); log.debug("Calling {} for querying metric", targetUri.toString()); List<Map<String, ?>> responseList = JsonClientUtil.getEntity(client.target(targetUri), List.class); if (responseList.isEmpty()) { return Collections.emptyMap(); } Map<String, ?> metrics = responseList.get(0); List<List<Number>> dataPoints = (List<List<Number>>) metrics.get("datapoints"); return formatDataPointsFromGraphiteToMap(dataPoints); } GraphiteWithStormQuerier(); @Override void init(Map<String, String> conf); @Override Map<Long, Double> getTopologyLevelMetrics(String topologyName, String metricName,
                                                     AggregateFunction aggrFunction, long from, long to); @Override Map<Long, Double> getMetrics(String topologyName, String componentId, String metricName, AggregateFunction aggrFunction,
                                        long from, long to); @Override Map<String, Map<Long, Double>> getRawMetrics(String metricName, String parameters, long from, long to); static final String WILDCARD_ALL_COMPONENTS; }
@Test public void getMetrics() throws Exception { stubMetricUrl(); String topologyName = "testTopology"; String componentId = "testComponent"; String metricName = "__test.metric.name"; TimeSeriesQuerier.AggregateFunction aggrFunction = TimeSeriesQuerier.AggregateFunction.SUM; long from = 1234000L; long to = 5678000L; Map<Long, Double> metrics = querier.getMetrics(topologyName, componentId, metricName, aggrFunction, from, to); assertResult(metrics); verify(getRequestedFor(urlPathEqualTo(TEST_RENDER_API_PATH)) .withQueryParam("target", equalTo("sumSeries(storm.testTopology.testComponent.*.*.*.*.*.*.__test.metric.name)")) .withQueryParam("format", equalTo("json")) .withQueryParam("from", equalTo("1234")) .withQueryParam("until", equalTo("5678"))); }
@Test public void getMetricsWithStreamAggregation() throws Exception { stubMetricUrl(); String topologyName = "testTopology"; String componentId = "testComponent"; String metricName = "__complete-latency"; TimeSeriesQuerier.AggregateFunction aggrFunction = TimeSeriesQuerier.AggregateFunction.AVG; long from = 1234000L; long to = 5678000L; Map<Long, Double> metrics = querier.getMetrics(topologyName, componentId, metricName, aggrFunction, from, to); assertResult(metrics); verify(getRequestedFor(urlPathEqualTo(TEST_RENDER_API_PATH)) .withQueryParam("target", equalTo("averageSeries(storm.testTopology.testComponent.*.*.*.*.*.*.__complete-latency.*)")) .withQueryParam("format", equalTo("json")) .withQueryParam("from", equalTo("1234")) .withQueryParam("until", equalTo("5678"))); }