MetricRegistryImpl implements MetricRegistry { public void shutdown() { synchronized (lock) { Future<Boolean> stopFuture = null; FiniteDuration stopTimeout = null; if (queryService != null) { stopTimeout = new FiniteDuration(1L, TimeUnit.SECONDS); try { stopFuture = Patterns.gracefulStop(queryService, stopTimeout); } catch (IllegalStateException ignored) { LOG.debug("The metric query service actor has already been stopped because the " + "underlying ActorSystem has already been shut down."); } } if (reporters != null) { for (MetricReporter reporter : reporters) { try { reporter.close(); } catch (Throwable t) { LOG.warn("Metrics reporter did not shut down cleanly", t); } } reporters = null; } shutdownExecutor(); if (stopFuture != null) { boolean stopped = false; try { stopped = Await.result(stopFuture, stopTimeout); } catch (Exception e) { LOG.warn("Query actor did not properly stop.", e); } if (!stopped) { queryService.tell(Kill.getInstance(), ActorRef.noSender()); } } } } MetricRegistryImpl(MetricRegistryConfiguration config); void startQueryService(ActorSystem actorSystem, ResourceID resourceID); @Override @Nullable String getMetricQueryServicePath(); @Override char getDelimiter(); @Override char getDelimiter(int reporterIndex); @Override int getNumberReporters(); @VisibleForTesting List<MetricReporter> getReporters(); boolean isShutdown(); void shutdown(); @Override ScopeFormats getScopeFormats(); @Override void register(Metric metric, String metricName, AbstractMetricGroup group); @Override void unregister(Metric metric, String metricName, AbstractMetricGroup group); @VisibleForTesting @Nullable ActorRef getQueryService();  }
@Test public void testReporterArgumentForwarding() { Configuration config = new Configuration(); config.setString(MetricOptions.REPORTERS_LIST, "test"); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, TestReporter2.class.getName()); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test.arg1", "hello"); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test.arg2", "world"); new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(config)).shutdown(); Assert.assertEquals("hello", TestReporter2.mc.getString("arg1", null)); Assert.assertEquals("world", TestReporter2.mc.getString("arg2", null)); }
@Test public void testReporterScheduling() throws InterruptedException { Configuration config = new Configuration(); config.setString(MetricOptions.REPORTERS_LIST, "test"); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, TestReporter3.class.getName()); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test.arg1", "hello"); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test." + ConfigConstants.METRICS_REPORTER_INTERVAL_SUFFIX, "50 MILLISECONDS"); MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(config)); long start = System.currentTimeMillis(); TestReporter3.reportCount = 0; for (int x = 0; x < 10; x++) { Thread.sleep(100); int reportCount = TestReporter3.reportCount; long curT = System.currentTimeMillis(); long maxAllowedReports = (curT - start) / 50 + 2; Assert.assertTrue("Too many reports were triggered.", maxAllowedReports >= reportCount); } Assert.assertTrue("No report was triggered.", TestReporter3.reportCount > 0); registry.shutdown(); }
@Test public void testReporterNotifications() { Configuration config = new Configuration(); config.setString(MetricOptions.REPORTERS_LIST, "test1,test2"); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test1." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, TestReporter6.class.getName()); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test2." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, TestReporter7.class.getName()); MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(config)); TaskManagerMetricGroup root = new TaskManagerMetricGroup(registry, "host", "id"); root.counter("rootCounter"); assertTrue(TestReporter6.addedMetric instanceof Counter); assertEquals("rootCounter", TestReporter6.addedMetricName); assertTrue(TestReporter7.addedMetric instanceof Counter); assertEquals("rootCounter", TestReporter7.addedMetricName); root.close(); assertTrue(TestReporter6.removedMetric instanceof Counter); assertEquals("rootCounter", TestReporter6.removedMetricName); assertTrue(TestReporter7.removedMetric instanceof Counter); assertEquals("rootCounter", TestReporter7.removedMetricName); registry.shutdown(); }
@Test public void testConfigurableDelimiter() { Configuration config = new Configuration(); config.setString(MetricOptions.SCOPE_DELIMITER, "_"); config.setString(MetricOptions.SCOPE_NAMING_TM, "A.B.C.D.E"); MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(config)); TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "host", "id"); assertEquals("A_B_C_D_E_name", tmGroup.getMetricIdentifier("name")); registry.shutdown(); }