AmbariMetricsServiceWithStormQuerier extends AbstractTimeSeriesQuerier { @Override public Map<String, Map<Long, Double>> getRawMetrics(String metricName, String parameters, long from, long to) { Map<String, String> queryParams = parseParameters(parameters); URI targetUri = composeRawQueryParameters(metricName, queryParams, from, to); log.debug("Calling {} for querying metric", targetUri.toString()); Map<String, ?> responseMap = JsonClientUtil.getEntity(client.target(targetUri), Map.class); List<Map<String, ?>> metrics = (List<Map<String, ?>>) responseMap.get("metrics"); if (metrics.size() > 0) { Map<String, Map<Long, Double>> ret = new HashMap<>(metrics.size()); for (Map<String, ?> metric : metrics) { String retrievedMetricName = (String) metric.get("metricname"); Map<String, Number> retrievedPoints = (Map<String, Number>) metric.get("metrics"); Map<Long, Double> pointsForOutput; if (retrievedPoints == null || retrievedPoints.isEmpty()) { pointsForOutput = Collections.emptyMap(); } else { pointsForOutput = new HashMap<>(retrievedPoints.size()); for (Map.Entry<String, Number> timestampToValue : retrievedPoints.entrySet()) { pointsForOutput.put(Long.valueOf(timestampToValue.getKey()), timestampToValue.getValue().doubleValue()); } } ret.put(retrievedMetricName, pointsForOutput); } return ret; } else { return Collections.emptyMap(); } } 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 getRawMetrics() throws Exception { stubMetricUrlForRawMetric(); String metricName = "metric"; String parameters = "precision=seconds,appId=appId"; long from = 1234L; long to = 5678L; Map<String, Map<Long, Double>> metrics = querier.getRawMetrics(metricName, parameters, from, to); assertRawMetricResult(metrics.get("metric")); verify(getRequestedFor(urlPathEqualTo(TEST_COLLECTOR_API_PATH)) .withQueryParam("appId", equalTo("appId")) .withQueryParam("metricNames", equalTo("metric")) .withQueryParam("startTime", equalTo("1234")) .withQueryParam("endTime", equalTo("5678"))); }