@Test
public void testMetricsSystemIntegration() throws Exception {

  RocksDBStore metadataStore = getTestRocksDBStoreWithData();
  Thread.sleep(2000);

  MetricsSystem ms = DefaultMetricsSystem.instance();
  MetricsSource rdbSource =
      ms.getSource("Rocksdb_TestRocksDBStoreMBean-withstat");

  BufferedMetricsCollector metricsCollector = new BufferedMetricsCollector();
  rdbSource.getMetrics(metricsCollector, true);

  Map<String, Double> metrics = metricsCollector.getMetricsRecordBuilder()
      .getMetrics();
  assertTrue(10.0 == metrics.get("NUMBER_KEYS_WRITTEN"));
  assertTrue(metrics.get("DB_WRITE_AVERAGE") > 0);
  metadataStore.close();
}
 

@Test
public void testCollectAllMetrics() {
  String queueName = "single";
  QueueMetrics.forQueue(ms, queueName, null, false, conf);
  MetricsSource queueSource = queueSource(ms, queueName);

  checkApps(queueSource, 0, 0, 0, 0, 0, 0, true);
  try {
    // do not collect all metrics
    checkApps(queueSource, 0, 0, 0, 0, 0, 0, false);
    Assert.fail();
  } catch (AssertionError e) {
    Assert.assertTrue(e.getMessage().contains(
      "Expected exactly one metric for name "));
  }
  // collect all metrics
  checkApps(queueSource, 0, 0, 0, 0, 0, 0, true);
}
 

public static void checkResources(MetricsSource source, int allocatedMB,
    int allocatedCores, int allocatedGCores, int allocCtnrs, long aggreAllocCtnrs,
    long aggreReleasedCtnrs, int availableMB, int availableCores, int availableGCores, int pendingMB,
    int pendingCores, int pendingGCores, int pendingCtnrs, int reservedMB, int reservedCores,
    int reservedGCores, int reservedCtnrs) {
  MetricsRecordBuilder rb = getMetrics(source);
  assertGauge("AllocatedMB", allocatedMB, rb);
  assertGauge("AllocatedVCores", allocatedCores, rb);
  assertGauge("AllocatedGCores", allocatedGCores, rb);
  assertGauge("AllocatedContainers", allocCtnrs, rb);
  assertCounter("AggregateContainersAllocated", aggreAllocCtnrs, rb);
  assertCounter("AggregateContainersReleased", aggreReleasedCtnrs, rb);
  assertGauge("AvailableMB", availableMB, rb);
  assertGauge("AvailableVCores", availableCores, rb);
  assertGauge("AvailableGCores", availableGCores, rb);
  assertGauge("PendingMB", pendingMB, rb);
  assertGauge("PendingVCores", pendingCores, rb);
  assertGauge("PendingGCores", pendingGCores, rb);
  assertGauge("PendingContainers", pendingCtnrs, rb);
  assertGauge("ReservedMB", reservedMB, rb);
  assertGauge("ReservedVCores", reservedCores, rb);
  assertGauge("ReservedGCores", reservedGCores, rb);
  assertGauge("ReservedContainers", reservedCtnrs, rb);
}
 

public MetricsSource build() {
  if (source instanceof MetricsSource) {
    if (hasAtMetric && !hasRegistry) {
      throw new MetricsException("Hybrid metrics: registry required.");
    }
    return (MetricsSource) source;
  }
  else if (!hasAtMetric) {
    throw new MetricsException("No valid @Metric annotation found.");
  }
  return new MetricsSource() {
    @Override
    public void getMetrics(MetricsCollector builder, boolean all) {
      registry.snapshot(builder.addRecord(registry.info()), all);
    }
  };
}
 

@Override public synchronized <T>
T register(String name, String desc, T source) {
  MetricsSourceBuilder sb = MetricsAnnotations.newSourceBuilder(source);
  final MetricsSource s = sb.build();
  MetricsInfo si = sb.info();
  String name2 = name == null ? si.name() : name;
  final String finalDesc = desc == null ? si.description() : desc;
  final String finalName = // be friendly to non-metrics tests
      DefaultMetricsSystem.sourceName(name2, !monitoring);
  allSources.put(finalName, s);
  LOG.debug(finalName +", "+ finalDesc);
  if (monitoring) {
    registerSource(finalName, finalDesc, s);
  }
  // We want to re-register the source to pick up new config when the
  // metrics system restarts.
  register(finalName, new AbstractCallback() {
    @Override public void postStart() {
      registerSource(finalName, finalDesc, s);
    }
  });
  return source;
}
 

@Test public void testHybrid() {
  HybridMetrics metrics = new HybridMetrics();
  MetricsSource source = MetricsAnnotations.makeSource(metrics);

  assertSame(metrics, source);
  metrics.C0.incr();
  MetricsRecordBuilder rb = getMetrics(source);
  MetricsCollector collector = rb.parent();

  verify(collector).addRecord("foo");
  verify(collector).addRecord("bar");
  verify(collector).addRecord(info("HybridMetrics", "HybridMetrics"));
  verify(rb).setContext("foocontext");
  verify(rb).addCounter(info("C1", "C1 desc"), 1);
  verify(rb).setContext("barcontext");
  verify(rb).addGauge(info("G1", "G1 desc"), 1);
  verify(rb).add(tag(MsInfo.Context, "hybrid"));
  verify(rb).addCounter(info("C0", "C0 desc"), 1);
  verify(rb).addGauge(info("G0", "G0"), 0);
}
 

@Test public void testUnregisterSource() {
  MetricsSystem ms = new MetricsSystemImpl();
  TestSource ts1 = new TestSource("ts1");
  TestSource ts2 = new TestSource("ts2");
  ms.register("ts1", "", ts1);
  ms.register("ts2", "", ts2);
  MetricsSource s1 = ms.getSource("ts1");
  assertNotNull(s1);
  // should work when metrics system is not started
  ms.unregisterSource("ts1");
  s1 = ms.getSource("ts1");
  assertNull(s1);
  MetricsSource s2 = ms.getSource("ts2");
  assertNotNull(s2);
  ms.shutdown();
}
 

@Test
public void testCollectAllMetrics() {
  String queueName = "single";
  QueueMetrics.forQueue(ms, queueName, null, false, conf);
  MetricsSource queueSource = queueSource(ms, queueName);

  checkApps(queueSource, 0, 0, 0, 0, 0, 0, true);
  try {
    // do not collect all metrics
    checkApps(queueSource, 0, 0, 0, 0, 0, 0, false);
    Assert.fail();
  } catch (AssertionError e) {
    Assert.assertTrue(e.getMessage().contains(
      "Expected exactly one metric for name "));
  }
  // collect all metrics
  checkApps(queueSource, 0, 0, 0, 0, 0, 0, true);
}
 

public static void checkResources(MetricsSource source, int allocatedMB,
    int allocatedCores, int allocCtnrs, long aggreAllocCtnrs,
    long aggreReleasedCtnrs, int availableMB, int availableCores, int pendingMB,
    int pendingCores, int pendingCtnrs, int reservedMB, int reservedCores,
    int reservedCtnrs) {
  MetricsRecordBuilder rb = getMetrics(source);
  assertGauge("AllocatedMB", allocatedMB, rb);
  assertGauge("AllocatedVCores", allocatedCores, rb);
  assertGauge("AllocatedContainers", allocCtnrs, rb);
  assertCounter("AggregateContainersAllocated", aggreAllocCtnrs, rb);
  assertCounter("AggregateContainersReleased", aggreReleasedCtnrs, rb);
  assertGauge("AvailableMB", availableMB, rb);
  assertGauge("AvailableVCores", availableCores, rb);
  assertGauge("PendingMB", pendingMB, rb);
  assertGauge("PendingVCores", pendingCores, rb);
  assertGauge("PendingContainers", pendingCtnrs, rb);
  assertGauge("ReservedMB", reservedMB, rb);
  assertGauge("ReservedVCores", reservedCores, rb);
  assertGauge("ReservedContainers", reservedCtnrs, rb);
}
 

public MetricsSource build() {
  if (source instanceof MetricsSource) {
    if (hasAtMetric && !hasRegistry) {
      throw new MetricsException("Hybrid metrics: registry required.");
    }
    return (MetricsSource) source;
  }
  else if (!hasAtMetric) {
    throw new MetricsException("No valid @Metric annotation found.");
  }
  return new MetricsSource() {
    @Override
    public void getMetrics(MetricsCollector builder, boolean all) {
      registry.snapshot(builder.addRecord(registry.info()), all);
    }
  };
}
 

@Override public synchronized <T>
T register(String name, String desc, T source) {
  MetricsSourceBuilder sb = MetricsAnnotations.newSourceBuilder(source);
  final MetricsSource s = sb.build();
  MetricsInfo si = sb.info();
  String name2 = name == null ? si.name() : name;
  final String finalDesc = desc == null ? si.description() : desc;
  final String finalName = // be friendly to non-metrics tests
      DefaultMetricsSystem.sourceName(name2, !monitoring);
  allSources.put(finalName, s);
  LOG.debug(finalName +", "+ finalDesc);
  if (monitoring) {
    registerSource(finalName, finalDesc, s);
  }
  // We want to re-register the source to pick up new config when the
  // metrics system restarts.
  register(finalName, new AbstractCallback() {
    @Override public void postStart() {
      registerSource(finalName, finalDesc, s);
    }
  });
  return source;
}
 

@Test public void testHybrid() {
  HybridMetrics metrics = new HybridMetrics();
  MetricsSource source = MetricsAnnotations.makeSource(metrics);

  assertSame(metrics, source);
  metrics.C0.incr();
  MetricsRecordBuilder rb = getMetrics(source);
  MetricsCollector collector = rb.parent();

  verify(collector).addRecord("foo");
  verify(collector).addRecord("bar");
  verify(collector).addRecord(info("HybridMetrics", "HybridMetrics"));
  verify(rb).setContext("foocontext");
  verify(rb).addCounter(info("C1", "C1 desc"), 1);
  verify(rb).setContext("barcontext");
  verify(rb).addGauge(info("G1", "G1 desc"), 1);
  verify(rb).add(tag(MsInfo.Context, "hybrid"));
  verify(rb).addCounter(info("C0", "C0 desc"), 1);
  verify(rb).addGauge(info("G0", "G0"), 0);
}
 

@Test public void testUnregisterSource() {
  MetricsSystem ms = new MetricsSystemImpl();
  TestSource ts1 = new TestSource("ts1");
  TestSource ts2 = new TestSource("ts2");
  ms.register("ts1", "", ts1);
  ms.register("ts2", "", ts2);
  MetricsSource s1 = ms.getSource("ts1");
  assertNotNull(s1);
  // should work when metrics system is not started
  ms.unregisterSource("ts1");
  s1 = ms.getSource("ts1");
  assertNull(s1);
  MetricsSource s2 = ms.getSource("ts2");
  assertNotNull(s2);
  ms.shutdown();
}
 

public static RocksDBStoreMBean create(Statistics statistics,
                                       String contextName) {

  RocksDBStoreMBean rocksDBStoreMBean = new RocksDBStoreMBean(
      statistics, contextName);
  MetricsSystem ms = DefaultMetricsSystem.instance();
  MetricsSource metricsSource = ms.getSource(rocksDBStoreMBean.contextName);
  if (metricsSource != null) {
    return (RocksDBStoreMBean)metricsSource;
  } else {
    return ms.register(rocksDBStoreMBean.contextName,
        "RocksDB Metrics",
        rocksDBStoreMBean);
  }
}
 

static CleanerMetrics create() {
  MetricsSystem ms = DefaultMetricsSystem.instance();

  CleanerMetrics metricObject = new CleanerMetrics();
  MetricsSourceBuilder sb = MetricsAnnotations.newSourceBuilder(metricObject);
  final MetricsSource s = sb.build();
  ms.register("cleaner", "The cleaner service of truly shared cache", s);
  metricObject.metricSource = s;
  return metricObject;
}
 

@Test public void testDefaultSingleQueueMetrics() {
  String queueName = "single";
  String user = "alice";

  QueueMetrics metrics = QueueMetrics.forQueue(ms, queueName, null, false,
				 conf);
  MetricsSource queueSource= queueSource(ms, queueName);
  AppSchedulingInfo app = mockApp(user);

  metrics.submitApp(user);
  MetricsSource userSource = userSource(ms, queueName, user);
  checkApps(queueSource, 1, 0, 0, 0, 0, 0, true);
  metrics.submitAppAttempt(user);
  checkApps(queueSource, 1, 1, 0, 0, 0, 0, true);

  metrics.setAvailableResourcesToQueue(Resources.createResource(100*GB, 100, 100));
  metrics.incrPendingResources(user, 5, Resources.createResource(3*GB, 3, 3));
  // Available resources is set externally, as it depends on dynamic
  // configurable cluster/queue resources
  checkResources(queueSource, 0, 0, 0, 0, 0, 0, 100*GB, 100, 100, 15*GB, 15, 15, 5, 0, 0, 0, 0);

  metrics.runAppAttempt(app.getApplicationId(), user);
  checkApps(queueSource, 1, 0, 1, 0, 0, 0, true);

  metrics.allocateResources(user, 3, Resources.createResource(2*GB, 2, 2), true);
  checkResources(queueSource, 6*GB, 6, 6, 3, 3, 0, 100*GB, 100, 100, 9*GB, 9, 9, 2, 0, 0, 0, 0);

  metrics.releaseResources(user, 1, Resources.createResource(2*GB, 2, 2));
  checkResources(queueSource, 4*GB, 4, 4, 2, 3, 1, 100*GB, 100, 100, 9*GB, 9, 9, 2, 0, 0, 0, 0);

  metrics.finishAppAttempt(
      app.getApplicationId(), app.isPending(), app.getUser());
  checkApps(queueSource, 1, 0, 0, 0, 0, 0, true);
  metrics.finishApp(user, RMAppState.FINISHED);
  checkApps(queueSource, 1, 0, 0, 1, 0, 0, true);
  assertNull(userSource);
}
 

public static void checkApps(MetricsSource source, int submitted, int pending,
    int running, int completed, int failed, int killed, boolean all) {
  MetricsRecordBuilder rb = getMetrics(source, all);
  assertCounter("AppsSubmitted", submitted, rb);
  assertGauge("AppsPending", pending, rb);
  assertGauge("AppsRunning", running, rb);
  assertCounter("AppsCompleted", completed, rb);
  assertCounter("AppsFailed", failed, rb);
  assertCounter("AppsKilled", killed, rb);
}
 

static void checkShuffleMetrics(MetricsSystem ms, long bytes, int failed,
                                int succeeded, int connections) {
  MetricsSource source = ms.getSource("ShuffleMetrics");
  MetricsRecordBuilder rb = getMetrics(source);
  assertCounter("ShuffleOutputBytes", bytes, rb);
  assertCounter("ShuffleOutputsFailed", failed, rb);
  assertCounter("ShuffleOutputsOK", succeeded, rb);
  assertGauge("ShuffleConnections", connections, rb);
}
 

@After
public void tearDown() throws Exception {
  MetricsSource source = DefaultMetricsSystem.instance().getSource("UgiMetrics");
  if (source != null) {
    // Run only once since the UGI metrics is cleaned up during teardown
    MetricsRecordBuilder rb = getMetrics(source);
    assertQuantileGauges("GetGroups1s", rb);
  }
  cluster.shutdown();
}
 

MetricsSourceAdapter(String prefix, String name, String description,
                     MetricsSource source, Iterable<MetricsTag> injectedTags,
                     MetricsFilter recordFilter, MetricsFilter metricFilter,
                     int jmxCacheTTL, boolean startMBeans) {
  this.prefix = checkNotNull(prefix, "prefix");
  this.name = checkNotNull(name, "name");
  this.source = checkNotNull(source, "source");
  attrCache = Maps.newHashMap();
  infoBuilder = new MBeanInfoBuilder(name, description);
  this.injectedTags = injectedTags;
  this.recordFilter = recordFilter;
  this.metricFilter = metricFilter;
  this.jmxCacheTTL = checkArg(jmxCacheTTL, jmxCacheTTL > 0, "jmxCacheTTL");
  this.startMBeans = startMBeans;
}
 

MetricsSourceAdapter(String prefix, String name, String description,
                     MetricsSource source, Iterable<MetricsTag> injectedTags,
                     int period, MetricsConfig conf) {
  this(prefix, name, description, source, injectedTags,
       conf.getFilter(RECORD_FILTER_KEY),
       conf.getFilter(METRIC_FILTER_KEY),
       period + 1, // hack to avoid most of the "innocuous" races.
       conf.getBoolean(START_MBEANS_KEY, true));
}
 

synchronized
void registerSource(String name, String desc, MetricsSource source) {
  checkNotNull(config, "config");
  MetricsConfig conf = sourceConfigs.get(name);
  MetricsSourceAdapter sa = conf != null
      ? new MetricsSourceAdapter(prefix, name, desc, source,
                                 injectedTags, period, conf)
      : new MetricsSourceAdapter(prefix, name, desc, source,
        injectedTags, period, config.subset(SOURCE_KEY));
  sources.put(name, sa);
  sa.start();
  LOG.debug("Registered source "+ name);
}
 

@Test public void testFields() {
  MyMetrics metrics = new MyMetrics();
  MetricsSource source = MetricsAnnotations.makeSource(metrics);

  metrics.c1.incr();
  metrics.c2.incr();
  metrics.g1.incr();
  metrics.g2.incr();
  metrics.g3.incr();
  metrics.r1.add(1);
  metrics.s1.add(1);
  metrics.rs1.add("rs1", 1);

  MetricsRecordBuilder rb = getMetrics(source);

  verify(rb).addCounter(info("C1", "C1"), 1);
  verify(rb).addCounter(info("Counter2", "Counter2 desc"), 1L);
  verify(rb).addGauge(info("G1", "G1"), 1);
  verify(rb).addGauge(info("G2", "G2"), 1);
  verify(rb).addGauge(info("G3", "g3 desc"), 1L);
  verify(rb).addCounter(info("R1NumOps", "Number of ops for r1"), 1L);
  verify(rb).addGauge(info("R1AvgTime", "Average time for r1"), 1.0);
  verify(rb).addCounter(info("S1NumOps", "Number of ops for s1"), 1L);
  verify(rb).addGauge(info("S1AvgTime", "Average time for s1"), 1.0);
  verify(rb).addCounter(info("Rs1NumOps", "Number of ops for rs1"), 1L);
  verify(rb).addGauge(info("Rs1AvgTime", "Average time for rs1"), 1.0);
}
 

@Test public void testMethods() {
  MyMetrics2 metrics = new MyMetrics2();
  MetricsSource source = MetricsAnnotations.makeSource(metrics);
  MetricsRecordBuilder rb = getMetrics(source);

  verify(rb).addGauge(info("G1", "G1"), 1);
  verify(rb).addGauge(info("G2", "G2"), 2L);
  verify(rb).addGauge(info("G3", "G3"), 3.0f);
  verify(rb).addGauge(info("G4", "G4"), 4.0);
  verify(rb).addCounter(info("C1", "C1"), 1);
  verify(rb).addCounter(info("C2", "C2"), 2L);
  verify(rb).tag(info("T1", "T1"), "t1");
}
 

@Test public void testRegisterDups() {
  MetricsSystem ms = new MetricsSystemImpl();
  TestSource ts1 = new TestSource("ts1");
  TestSource ts2 = new TestSource("ts2");
  ms.register("ts1", "", ts1);
  MetricsSource s1 = ms.getSource("ts1");
  assertNotNull(s1);
  // should work when metrics system is not started
  ms.register("ts1", "", ts2);
  MetricsSource s2 = ms.getSource("ts1");
  assertNotNull(s2);
  assertNotSame(s1, s2);
  ms.shutdown();
}
 

@Test
public void testGetMetricsAndJmx() throws Exception {
  // create test source with a single metric counter of value 0
  TestSource source = new TestSource("test");
  MetricsSourceBuilder sb = MetricsAnnotations.newSourceBuilder(source);
  final MetricsSource s = sb.build();

  List<MetricsTag> injectedTags = new ArrayList<MetricsTag>();
  MetricsSourceAdapter sa = new MetricsSourceAdapter(
      "test", "test", "test desc", s, injectedTags, null, null, 1, false);

  // all metrics are initially assumed to have changed
  MetricsCollectorImpl builder = new MetricsCollectorImpl();
  Iterable<MetricsRecordImpl> metricsRecords = sa.getMetrics(builder, true);

  // Validate getMetrics and JMX initial values
  MetricsRecordImpl metricsRecord = metricsRecords.iterator().next();
  assertEquals(0L,
      metricsRecord.metrics().iterator().next().value().longValue());

  Thread.sleep(100); // skip JMX cache TTL
  assertEquals(0L, (Number)sa.getAttribute("C1"));

  // change metric value
  source.incrementCnt();

  // validate getMetrics and JMX
  builder = new MetricsCollectorImpl();
  metricsRecords = sa.getMetrics(builder, true);
  metricsRecord = metricsRecords.iterator().next();
  assertTrue(metricsRecord.metrics().iterator().hasNext());
  Thread.sleep(100); // skip JMX cache TTL
  assertEquals(1L, (Number)sa.getAttribute("C1"));
}
 

/**
 * Call getMetrics on source and get a record builder mock to verify
 * @param source  the metrics source
 * @param all     if true, return all metrics even if not changed
 * @return the record builder mock to verify
 */
public static MetricsRecordBuilder getMetrics(MetricsSource source,
                                              boolean all) {
  MetricsRecordBuilder rb = mockMetricsRecordBuilder();
  MetricsCollector mc = rb.parent();
  source.getMetrics(mc, all);
  return rb;
}
 

static CleanerMetrics create() {
  MetricsSystem ms = DefaultMetricsSystem.instance();

  CleanerMetrics metricObject = new CleanerMetrics();
  MetricsSourceBuilder sb = MetricsAnnotations.newSourceBuilder(metricObject);
  final MetricsSource s = sb.build();
  ms.register("cleaner", "The cleaner service of truly shared cache", s);
  metricObject.metricSource = s;
  return metricObject;
}
 

@Test public void testDefaultSingleQueueMetrics() {
  String queueName = "single";
  String user = "alice";

  QueueMetrics metrics = QueueMetrics.forQueue(ms, queueName, null, false,
				 conf);
  MetricsSource queueSource= queueSource(ms, queueName);
  AppSchedulingInfo app = mockApp(user);

  metrics.submitApp(user);
  MetricsSource userSource = userSource(ms, queueName, user);
  checkApps(queueSource, 1, 0, 0, 0, 0, 0, true);
  metrics.submitAppAttempt(user);
  checkApps(queueSource, 1, 1, 0, 0, 0, 0, true);

  metrics.setAvailableResourcesToQueue(Resources.createResource(100*GB, 100));
  metrics.incrPendingResources(user, 5, Resources.createResource(3*GB, 3));
  // Available resources is set externally, as it depends on dynamic
  // configurable cluster/queue resources
  checkResources(queueSource, 0, 0, 0, 0, 0, 100*GB, 100, 15*GB, 15, 5, 0, 0, 0);

  metrics.runAppAttempt(app.getApplicationId(), user);
  checkApps(queueSource, 1, 0, 1, 0, 0, 0, true);

  metrics.allocateResources(user, 3, Resources.createResource(2*GB, 2), true);
  checkResources(queueSource, 6*GB, 6, 3, 3, 0, 100*GB, 100, 9*GB, 9, 2, 0, 0, 0);

  metrics.releaseResources(user, 1, Resources.createResource(2*GB, 2));
  checkResources(queueSource, 4*GB, 4, 2, 3, 1, 100*GB, 100, 9*GB, 9, 2, 0, 0, 0);

  metrics.finishAppAttempt(
      app.getApplicationId(), app.isPending(), app.getUser());
  checkApps(queueSource, 1, 0, 0, 0, 0, 0, true);
  metrics.finishApp(user, RMAppState.FINISHED);
  checkApps(queueSource, 1, 0, 0, 1, 0, 0, true);
  assertNull(userSource);
}
 

public static void checkApps(MetricsSource source, int submitted, int pending,
    int running, int completed, int failed, int killed, boolean all) {
  MetricsRecordBuilder rb = getMetrics(source, all);
  assertCounter("AppsSubmitted", submitted, rb);
  assertGauge("AppsPending", pending, rb);
  assertGauge("AppsRunning", running, rb);
  assertCounter("AppsCompleted", completed, rb);
  assertCounter("AppsFailed", failed, rb);
  assertCounter("AppsKilled", killed, rb);
}