BasicMetricsCollector implements IMetricsCollector { void writeHeader(PrintWriter writer) { header.add(TIME); if (collectSupervisorStats) { header.add(TOTAL_SLOTS); header.add(USED_SLOTS); } if (collectTopologyStats) { header.add(WORKERS); header.add(TASKS); header.add(EXECUTORS); } if (collectThroughput) { header.add(TRANSFERRED); header.add(THROUGHPUT); } if (collectThroughputMB) { header.add(THROUGHPUT_MB); header.add(SPOUT_THROUGHPUT_MB); } if (collectSpoutThroughput) { header.add(SPOUT_EXECUTORS); header.add(SPOUT_TRANSFERRED); header.add(SPOUT_ACKED); header.add(SPOUT_THROUGHPUT); } if (collectSpoutLatency) { header.add(SPOUT_AVG_COMPLETE_LATENCY); header.add(SPOUT_MAX_COMPLETE_LATENCY); } String str = Utils.join(header, ","); LOG.info("writing out metrics headers [" + str + "] into .csv file"); writer.println(str); writer.flush(); } BasicMetricsCollector(Config stormConfig, StormTopology topology, Set<MetricsItem> items); @Override void run(); Nimbus.Client getNimbusClient(Config stormConfig); static final String TIME; static final String TIME_FORMAT; static final String TOTAL_SLOTS; static final String USED_SLOTS; static final String WORKERS; static final String TASKS; static final String EXECUTORS; static final String TRANSFERRED; static final String THROUGHPUT; static final String THROUGHPUT_MB; static final String THROUGHPUT_MB_FORMAT; static final String SPOUT_EXECUTORS; static final String SPOUT_TRANSFERRED; static final String SPOUT_ACKED; static final String SPOUT_THROUGHPUT; static final String SPOUT_THROUGHPUT_MB; static final String SPOUT_THROUGHPUT_MB_FORMAT; static final String SPOUT_AVG_COMPLETE_LATENCY; static final String SPOUT_AVG_LATENCY_FORMAT; static final String SPOUT_MAX_COMPLETE_LATENCY; static final String SPOUT_MAX_LATENCY_FORMAT; static final String ALL_TIME; static final String LAST_TEN_MINS; static final String LAST_THREE_HOURS; static final String LAST_DAY; }
@Test(dataProvider = "getMetricsItems") public void testWriteHeaders(Set<MetricsItem> items, Set<String> expected) { BasicMetricsCollector collector = new BasicMetricsCollector(config, topology, items); PrintWriter writer = mock(PrintWriter.class); collector.writeHeader(writer); verify(writer, times(1)).println(anyString()); verify(writer, times(1)).flush(); for (String h : expected) { collector.header.contains(h); } }