AmbariMetricsServiceWithStormQuerier extends AbstractTimeSeriesQuerier { @Override public Map<Long, Double> getMetrics(String topologyName, String componentId, String metricName, AggregateFunction aggrFunction, long from, long to) { Optional<String> weightMetric = findWeightMetric(metricName); if (weightMetric.isPresent()) { Map<Long, List<Pair<String, Double>>> keyMetrics = getMetricsStreamToValueMap(topologyName, componentId, metricName, from, to); Map<Long, List<Pair<String, Double>>> weightMetrics = getMetricsStreamToValueMap(topologyName, componentId, weightMetric.get(), from, to); return aggregateWithApplyingWeightedAverage(keyMetrics, weightMetrics); } else { Map<Long, List<Pair<String, Double>>> ret = getMetricsStreamToValueMap(topologyName, componentId, metricName, from, to); return aggregateStreamsForMetricsValues(ret, aggrFunction); } } AmbariMetricsServiceWithStormQuerier(); @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 METRIC_NAME_PREFIX_KAFKA_OFFSET; }
@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 = 1234L; long to = 5678L; Map<Long, Double> metrics = querier.getMetrics(topologyName, componentId, metricName, aggrFunction, from, to); assertResult(metrics, aggrFunction); verify(getRequestedFor(urlPathEqualTo(TEST_COLLECTOR_API_PATH)) .withQueryParam("appId", equalTo(DEFAULT_APP_ID)) .withQueryParam("hostname", equalTo("")) .withQueryParam("metricNames", equalTo("topology.testTopology.testComponent.%.--test.metric.name")) .withQueryParam("startTime", equalTo("1234")) .withQueryParam("endTime", equalTo("5678"))); }
@Test public void getMetricsWithStreamAggregation() throws Exception { stubMetricUrl(); String topologyName = "testTopology"; String componentId = "testComponent"; String metricName = "__emit-count"; TimeSeriesQuerier.AggregateFunction aggrFunction = TimeSeriesQuerier.AggregateFunction.SUM; long from = 1234L; long to = 5678L; Map<Long, Double> metrics = querier.getMetrics(topologyName, componentId, metricName, aggrFunction, from, to); assertResult(metrics, aggrFunction); verify(getRequestedFor(urlPathEqualTo(TEST_COLLECTOR_API_PATH)) .withQueryParam("appId", equalTo(DEFAULT_APP_ID)) .withQueryParam("hostname", equalTo("")) .withQueryParam("metricNames", equalTo("topology.testTopology.testComponent.%.--emit-count.%")) .withQueryParam("startTime", equalTo("1234")) .withQueryParam("endTime", equalTo("5678"))); }