StormTopologyTimeSeriesMetricsImpl implements TopologyTimeSeriesMetrics { @Override public TimeSeriesComponentMetric getComponentStats(TopologyLayout topology, Component component, long from, long to, String asUser) { String stormTopologyName = StormTopologyUtil.findOrGenerateTopologyName(client, topology.getId(), topology.getName(), asUser); String stormComponentName = getComponentName(component); StormMappedMetric[] stats; if (component instanceof Source) { stats = STATS_METRICS_SOURCE; } else { stats = STATS_METRICS; } Map<String, Map<Long, Double>> componentStats = ParallelStreamUtil.execute(() -> Arrays.asList(stats) .parallelStream() .collect(toMap(m -> m.name(), m -> queryComponentMetrics(stormTopologyName, stormComponentName, m, from, to))), FORK_JOIN_POOL); return buildTimeSeriesComponentMetric(component.getName(), componentStats); } StormTopologyTimeSeriesMetricsImpl(StormRestAPIClient client); @Override void setTimeSeriesQuerier(TimeSeriesQuerier timeSeriesQuerier); @Override TimeSeriesQuerier getTimeSeriesQuerier(); @Override Map<Long, Double> getCompleteLatency(TopologyLayout topology, Component component, long from, long to, String asUser); @Override Map<String, Map<Long, Double>> getkafkaTopicOffsets(TopologyLayout topology, Component component, long from, long to, String asUser); @Override TimeSeriesComponentMetric getTopologyStats(TopologyLayout topology, long from, long to, String asUser); @Override TimeSeriesComponentMetric getComponentStats(TopologyLayout topology, Component component, long from, long to, String asUser);  }
@Test public void getSourceComponentStats() throws Exception { final TopologyLayout topology = getTopologyLayoutForTest(); final long from = 1L; final long to = 3L; final TopologyTimeSeriesMetrics.TimeSeriesComponentMetric expectedMetric = setupExpectionToSpoutComponent(source, from, to); TopologyTimeSeriesMetrics.TimeSeriesComponentMetric actual = stormTopologyTimeSeriesMetrics.getComponentStats(topology, source, from, to, null); assertEquals(expectedMetric, actual); }
@Test public void getProcessorComponentStats() throws Exception { final TopologyLayout topology = getTopologyLayoutForTest(); final long from = 1L; final long to = 3L; final TopologyTimeSeriesMetrics.TimeSeriesComponentMetric expectedMetric = setupExpectionToBoltComponent(processor, from, to); TopologyTimeSeriesMetrics.TimeSeriesComponentMetric actual = stormTopologyTimeSeriesMetrics.getComponentStats(topology, processor, from, to, null); assertEquals(expectedMetric, actual); }
@Test public void getSinkComponentStats() throws Exception { final TopologyLayout topology = getTopologyLayoutForTest(); final long from = 1L; final long to = 3L; final TopologyTimeSeriesMetrics.TimeSeriesComponentMetric expectedMetric = setupExpectionToBoltComponent(sink, from, to); TopologyTimeSeriesMetrics.TimeSeriesComponentMetric actual = stormTopologyTimeSeriesMetrics.getComponentStats(topology, sink, from, to, null); assertEquals(expectedMetric, actual); }