NormalizationTransform implements MetricObserver { @Override public void update(List<Metric> metrics) { Preconditions.checkNotNull(metrics, "metrics"); final List<Metric> newMetrics = new ArrayList<>(metrics.size()); for (Metric m : metrics) { long offset = m.getTimestamp() % stepMillis; long stepBoundary = m.getTimestamp() - offset; String dsType = getDataSourceType(m); if (isGauge(dsType) || isNormalized(dsType)) { Metric atStepBoundary = new Metric(m.getConfig(), stepBoundary, m.getValue()); newMetrics.add(atStepBoundary); } else if (isRate(dsType)) { Metric normalized = normalize(m, stepBoundary); if (normalized != null) { newMetrics.add(normalized); } } else if (!isInformational(dsType)) { newMetrics.add(m); } } observer.update(newMetrics); } @Deprecated  NormalizationTransform(MetricObserver observer, long step, long heartbeat); @Deprecated  NormalizationTransform(MetricObserver observer, long step, final long heartbeat,
                                final Clock clock);  NormalizationTransform(MetricObserver observer, long step, long heartbeat,
                                TimeUnit unit);  NormalizationTransform(MetricObserver observer, long step, final long heartbeat,
                                TimeUnit unit, final Clock clock); @Override void update(List<Metric> metrics); @Override String getName();  }
@Test public void testUpdate() throws Exception { BasicCounter basicCounter = new BasicCounter(MonitorConfig.builder("basicCounter").build()); ManualClock manualClock = new ManualClock(0); StepCounter stepCounter = new StepCounter(MonitorConfig.builder("stepCounter").build(), manualClock); LongGauge gauge = new LongGauge(MonitorConfig.builder("longGauge").build()); List<? extends AbstractMonitor<? extends Number>> monitors = UnmodifiableList.of(basicCounter, stepCounter, gauge); MemoryMetricObserver observer = new MemoryMetricObserver("normalization-test", 1); NormalizationTransform normalizationTransform = new NormalizationTransform(observer, 60, 120, TimeUnit.SECONDS, manualClock); CounterToRateMetricTransform toRateMetricTransform = new CounterToRateMetricTransform(normalizationTransform, 60, 120, TimeUnit.SECONDS, manualClock); double[] rates = {0.5 / 60.0, 2 / 60.0, 3 / 60.0, 4 / 60.0}; double[] expectedNormalized = { rates[0] * (2.0 / 3.0), rates[0] * (1.0 / 3.0) + rates[1] * (2.0 / 3.0), rates[1] * (1.0 / 3.0) + rates[2] * (2.0 / 3.0), rates[2] * (1.0 / 3.0) + rates[3] * (2.0 / 3.0)}; for (int i = 1; i < 5; ++i) { long now = 20000L + i * 60000L; long stepBoundary = i * 60000L; manualClock.set(now); basicCounter.increment(i); stepCounter.increment(i); gauge.set((long) i); List<Metric> metrics = getValue(monitors, manualClock); toRateMetricTransform.update(metrics); List<Metric> o = observer.getObservations().get(0); assertEquals(o.size(), 3); double basicCounterVal = o.get(0).getNumberValue().doubleValue(); double stepCounterVal = o.get(1).getNumberValue().doubleValue(); double gaugeVal = o.get(2).getNumberValue().doubleValue(); assertEquals(gaugeVal, (double) i, DELTA); assertEquals(stepCounterVal, (i - 1) / 60.0, DELTA); assertEquals(basicCounterVal, expectedNormalized[i - 1], DELTA); for (Metric m : o) { assertEquals(m.getTimestamp(), stepBoundary); } } int i = 5; manualClock.set(i * 60000L + 20000L); List<Metric> metrics = getValue(monitors, manualClock); toRateMetricTransform.update(metrics); List<Metric> o = observer.getObservations().get(0); assertEquals(o.size(), 3); double basicCounterVal = o.get(0).getNumberValue().doubleValue(); double stepCounterVal = o.get(1).getNumberValue().doubleValue(); double gaugeVal = o.get(2).getNumberValue().doubleValue(); assertEquals(gaugeVal, (double) 4, DELTA); assertEquals(stepCounterVal, 4 / 60.0, DELTA); assertEquals(basicCounterVal, (1 / 3.0) * rates[3]); }
@Test public void testExpiration() { BasicCounter c1 = new BasicCounter(MonitorConfig.builder("c1").build()); BasicCounter c2 = new BasicCounter(MonitorConfig.builder("c2").build()); BasicCounter c3 = new BasicCounter(MonitorConfig.builder("c3").build()); ManualClock manualClock = new ManualClock(0); MemoryMetricObserver observer = new MemoryMetricObserver("normalization-test", 1); NormalizationTransform normalizationTransform = new NormalizationTransform(observer, 60, 120, TimeUnit.SECONDS, manualClock); CounterToRateMetricTransform toRateMetricTransform = new CounterToRateMetricTransform(normalizationTransform, 60, 120, TimeUnit.SECONDS, manualClock); manualClock.set(30000L); c1.increment(); Metric m1 = new Metric(c1.getConfig(), manualClock.now(), c1.getValue(0)); toRateMetricTransform.update(UnmodifiableList.of(m1)); assertEquals(NormalizationTransform.HEARTBEAT_EXCEEDED.getValue(0).longValue(), 0); List<Metric> o = observer.getObservations().get(0); assertEquals(o.size(), 1); manualClock.set(100000L); Metric m2 = new Metric(c2.getConfig(), manualClock.now(), c2.getValue()); toRateMetricTransform.update(UnmodifiableList.of(m2)); assertEquals(NormalizationTransform.HEARTBEAT_EXCEEDED.getValue(0).longValue(), 0); manualClock.set(160000L); Metric m3 = new Metric(c3.getConfig(), manualClock.now(), c3.getValue()); toRateMetricTransform.update(UnmodifiableList.of(m3)); assertEquals(NormalizationTransform.HEARTBEAT_EXCEEDED.getValue(0).longValue(), 1); }