com.codahale.metrics.MetricRegistry#getCounters ( )源码实例Demo

@Test
public void testInit() throws StageException {

  MetricRegistry metrics = metricAggregationProcessor.getMetrics();
  SortedMap<String, Timer> timers = metrics.getTimers();

  Assert.assertEquals(5, timers.size()); // 1 each for 4 stages, 1 for pipeline

  SortedMap<String, Counter> counters = metrics.getCounters();
  Assert.assertEquals(24, counters.size()); // 4 each for 4 stages, 5 for pipeline, one each for 3 output lanes

  SortedMap<String, Meter> meters = metrics.getMeters();
  Assert.assertEquals(24, meters.size()); // 4 each for 4 stages, 5 for pipeline, one each for 3 output lanes

  SortedMap<String, Histogram> histograms = metrics.getHistograms();
  Assert.assertEquals(20, histograms.size()); // 4 each for 4 stages, 4 for pipeline
}
 

@Test
public void testEvaluateDataRuleRecord() throws StageException, InterruptedException {
  Record dataRuleChangeRecord = AggregatorUtil.createDataRuleChangeRecord(
    TestHelper.createTestDataRuleDefinition("a < b", true, System.currentTimeMillis())
  );
  Thread.sleep(2); // Sleep for 2 seconds that the data rule records get a later timestamp
  List<Record> testDataRuleRecords = TestHelper.createTestDataRuleRecords();
  List<Record> batch = new ArrayList<>();
  batch.add(dataRuleChangeRecord);
  batch.addAll(testDataRuleRecords);
  runner.runProcess(batch);
  MetricRegistry metrics = metricAggregationProcessor.getMetrics();

  SortedMap<String, Counter> counters = metrics.getCounters();
  Assert.assertTrue(counters.containsKey("user.x.matched.counter"));
  Assert.assertEquals(400, counters.get("user.x.matched.counter").getCount());

  Assert.assertTrue(counters.containsKey("user.x.evaluated.counter"));
  Assert.assertEquals(2000, counters.get("user.x.evaluated.counter").getCount());

  // Alert expected as threshold type is count and threshold value is 100
  SortedMap<String, Gauge> gauges = metrics.getGauges();
  Assert.assertTrue(gauges.containsKey("alert.x.gauge"));
  Map<String, Object> gaugeValue = (Map<String, Object>) gauges.get("alert.x.gauge").getValue();
  Assert.assertEquals(400L, gaugeValue.get("currentValue"));

  List<String> alertTexts = (List<String>) gaugeValue.get("alertTexts");
  Assert.assertEquals(1, alertTexts.size());
  Assert.assertEquals("Alert!!", alertTexts.get(0));
}
 

@SuppressWarnings("rawtypes")
public static String parse(Object result){
    MetricRegistry registry = (MetricRegistry) result;
    StringBuilder sb = new StringBuilder();

    Map<String, Counter> counters = registry.getCounters();
    for(Map.Entry<String, Counter> c : counters.entrySet()){
        sb.append(c.getKey().replace(".","_")+" "+c.getValue().getCount()+"\n");
    }

    Map<String, Timer> timers = registry.getTimers();
    for(Map.Entry<String, Timer> t : timers.entrySet()){
        sb.append(t.getKey().replace(".","_")+" "+t.getValue().getCount()+"\n");
    }

    Map<String, Gauge> gauges = registry.getGauges();
    for(Map.Entry<String, Gauge> g : gauges.entrySet()){
        sb.append(g.getKey().replace(".","_")+" "+g.getValue().getValue()+"\n");
    }

    Map<String, Histogram> histograms = registry.getHistograms();
    for(Map.Entry<String, Histogram> h : histograms.entrySet()){
        sb.append(h.getKey().replace(".","_")+" "+h.getValue().getCount()+"\n");
    }

    return sb.toString();
}
 

@Test
public void testRouteMetrics() throws Exception {
    SimpleRegistry registry = new SimpleRegistry();
    MetricRegistry metricRegistry = new MetricRegistry();
    registry.bind("metricRegistry", metricRegistry);

    CamelContext camelctx = new DefaultCamelContext(registry);
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
            .to("metrics:counter:simple.counter?increment=5");
        }
    });

    try {
        camelctx.start();

        ProducerTemplate template = camelctx.createProducerTemplate();
        template.requestBody("direct:start", "fake body");

        SortedMap<String, Counter> counters = metricRegistry.getCounters();
        Counter counter = null;
        for (String counterName : counters.keySet()) {
            if (counterName.equals("simple.counter")) {
                counter = counters.get(counterName);
                break;
            }
        }

        Assert.assertNotNull("Counter simple.counter was null", counter);
        Assert.assertEquals(5, counter.getCount());
    } finally {
        camelctx.close();
    }
}
 

@Test
public void testCountersForMetricRecord() throws StageException {
  MetricsRuleDefinition metricsRuleDefinition1 = TestHelper.createTestMetricRuleDefinition("a < b", true, System.currentTimeMillis());
  Record metricRuleChangeRecord = AggregatorUtil.createMetricRuleChangeRecord(metricsRuleDefinition1);
  Record metricRecord = createTestMetricRecord();
  runner.runProcess(Arrays.asList(metricRuleChangeRecord, metricRecord));

  MetricRegistry metrics = metricAggregationProcessor.getMetrics();

  SortedMap<String, Counter> counters = metrics.getCounters();
  Assert.assertTrue(
      counters.containsKey(
          "stage.com_streamsets_pipeline_stage_origin_spooldir_SpoolDirDSource_1:com_streamsets_pipeline_stage_origin_spooldir_SpoolDirDSource_1OutputLane14578524215930.outputRecords.counter"
      )
  );
  Assert.assertEquals(
      1000,
      counters.get(
        "stage.com_streamsets_pipeline_stage_origin_spooldir_SpoolDirDSource_1:com_streamsets_pipeline_stage_origin_spooldir_SpoolDirDSource_1OutputLane14578524215930.outputRecords.counter"
      ).getCount()
  );
  Assert.assertTrue(
    counters.containsKey(
      "stage.com_streamsets_pipeline_stage_processor_fieldhasher_FieldHasherDProcessor_1:com_streamsets_pipeline_stage_processor_fieldhasher_FieldHasherDProcessor_1OutputLane14578524269500.outputRecords.counter"
    )
  );
  Assert.assertEquals(
    750,
    counters.get(
      "stage.com_streamsets_pipeline_stage_processor_fieldhasher_FieldHasherDProcessor_1:com_streamsets_pipeline_stage_processor_fieldhasher_FieldHasherDProcessor_1OutputLane14578524269500.outputRecords.counter"
    ).getCount()
  );
  Assert.assertTrue(
    counters.containsKey(
      "stage.com_streamsets_pipeline_stage_processor_expression_ExpressionDProcessor_1:com_streamsets_pipeline_stage_processor_expression_ExpressionDProcessor_1OutputLane14578524390500.outputRecords.counter"
    )
  );
  Assert.assertEquals(
    500,
    counters.get(
      "stage.com_streamsets_pipeline_stage_processor_expression_ExpressionDProcessor_1:com_streamsets_pipeline_stage_processor_expression_ExpressionDProcessor_1OutputLane14578524390500.outputRecords.counter"
    ).getCount()
  );
}