#243 一个参考CodaHale的全新的Metrics Api,基础计算部分完工,还剩报表部分

This commit is contained in:
Calvin
2013-12-02 18:37:31 +08:00
parent 33c65c9592
commit e92ebddd0b
14 changed files with 393 additions and 276 deletions
@@ -7,16 +7,12 @@ import org.springside.modules.metrics.utils.Clock;
public class Counter {
private AtomicLong counter = new AtomicLong(0);
private long lastReportCounter = 0L;
private long lastReportCount = 0L;
private Clock clock;
private long startTime;
private long lastReportTime;
public Counter() {
this(Clock.DEFAULT);
}
public Counter(Clock clock) {
this.clock = clock;
startTime = clock.getCurrentTime();
@@ -40,32 +36,25 @@ public class Counter {
}
public long reset() {
return counter.getAndSet(0);
long currentValue = counter.getAndSet(0);
lastReportCount = 0L;
startTime = clock.getCurrentTime();
lastReportTime = startTime;
return currentValue;
}
public long getCount() {
return counter.get();
}
public CounterMetric getMetric() {
long currentCount = counter.get();
long currentTime = clock.getCurrentTime();
public double getLastRate() {
long currentCounter = counter.get();
long reportTime = clock.getCurrentTime();
CounterMetric metric = new CounterMetric();
metric.count = currentCount;
metric.lastRate = ((currentCount - lastReportCount) * 1000) / (currentTime - lastReportTime);
metric.meanRate = ((currentCount * 1000) / (currentTime - startTime));
double rate = 0.0;
if (currentCounter != 0) {
rate = ((currentCounter - lastReportCounter) / (reportTime - lastReportTime)) * 1000;
}
lastReportCounter = currentCounter;
lastReportTime = reportTime;
lastReportCount = currentCount;
lastReportTime = currentTime;
return rate;
}
public double getMeanRate() {
if (counter.get() == 0) {
return 0.0;
} else {
return (counter.get() / (clock.getCurrentTime() - startTime)) * 1000;
}
return metric;
}
}
@@ -1,7 +1,7 @@
package org.springside.modules.metrics.report.metrics;
package org.springside.modules.metrics;
public class CounterMetric {
public long coute;
public long count;
public double lastRate;
public double meanRate;
}
@@ -3,6 +3,31 @@ package org.springside.modules.metrics;
import org.springside.modules.metrics.utils.Clock;
public class Execution {
private Counter counter;
private Histogram histogram;
private Clock clock;
public Execution(Clock clock, Double[] pcts) {
this.clock = clock;
counter = new Counter(clock);
histogram = new Histogram(pcts);
}
public ExecutiomTimer start() {
return new ExecutiomTimer(this, clock);
}
private void update(long duration) {
histogram.update(duration);
counter.inc();
}
public ExecutionMetric getMetric() {
ExecutionMetric metric = new ExecutionMetric();
metric.counter = counter.getMetric();
metric.histogram = histogram.getMetric();
return metric;
}
public static class ExecutiomTimer {
private final Execution execution;
@@ -15,47 +40,9 @@ public class Execution {
this.startTime = clock.getCurrentTime();
}
public long stop() {
public void stop() {
final long elapsed = clock.getCurrentTime() - startTime;
execution.update(elapsed);
return elapsed;
}
}
private Counter counter;
private Histogram histogram;
private Clock clock;
public Execution() {
this(Clock.DEFAULT);
}
public Execution(Clock clock) {
this.clock = clock;
}
public ExecutiomTimer start() {
return new ExecutiomTimer(this, clock);
}
private void update(long duration) {
histogram.update(duration);
counter.inc();
}
public HistogramSnapshot getSnapshot() {
return histogram.getSnapshot();
}
public long getCount() {
return counter.getCount();
}
public double getLastRate() {
return counter.getLastRate();
}
public double getMeanRate() {
return counter.getMeanRate();
}
}
@@ -0,0 +1,6 @@
package org.springside.modules.metrics;
public class ExecutionMetric {
public CounterMetric counter;
public HistogramMetric histogram;
}
@@ -1,65 +1,79 @@
package org.springside.modules.metrics;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.springside.modules.metrics.utils.Clock;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
public class Histogram {
// allow for this many duplicate ticks before overwriting measurements
private static final int COLLISION_BUFFER = 256;
// only trim on updating once every N
private static final int TRIM_THRESHOLD = 256;
private LinkedList<Long> measurements = new LinkedList<Long>();
private Double[] pcts;
private final Clock clock;
private final ConcurrentSkipListMap<Long, Long> measurements;
private final long window;
private final AtomicLong lastTick;
private final AtomicLong count;
public Histogram(long window, TimeUnit windowUnit) {
this(window, windowUnit, Clock.DEFAULT);
}
public Histogram(long window, TimeUnit windowUnit, Clock clock) {
this.clock = clock;
this.measurements = new ConcurrentSkipListMap<Long, Long>();
this.window = windowUnit.toMillis(window) * COLLISION_BUFFER;
this.lastTick = new AtomicLong();
this.count = new AtomicLong();
public Histogram(Double[] pcts) {
this.pcts = pcts;
}
public void update(long value) {
if ((count.incrementAndGet() % TRIM_THRESHOLD) == 0) {
trim();
}
measurements.put(getTick(), value);
}
public HistogramSnapshot getSnapshot() {
trim();
return new HistogramSnapshot(measurements.values());
}
/**
* TODO:??
*/
private long getTick() {
for (;;) {
final long oldTick = lastTick.get();
final long tick = clock.getCurrentTime() * COLLISION_BUFFER;
final long newTick = tick > oldTick ? tick : oldTick + 1;
if (lastTick.compareAndSet(oldTick, newTick)) {
return newTick;
}
synchronized (measurements) {
measurements.add(value);
}
}
private void trim() {
measurements.headMap(getTick() - window).clear();
public HistogramMetric getMetric() {
LinkedList<Long> snapshotList = measurements;
measurements = new LinkedList();
if (snapshotList.isEmpty()) {
return createEmptyMetric();
}
Collections.sort(snapshotList);
int count = snapshotList.size();
HistogramMetric metric = new HistogramMetric();
metric.min = snapshotList.get(0);
metric.max = snapshotList.get(count - 1);
double sum = 0;
for (long value : snapshotList) {
sum += value;
}
metric.mean = sum / count;
for (Double pct : pcts) {
metric.pcts.put(pct, getPercent(snapshotList, count, pct));
}
return metric;
}
public Long getPercent(List<Long> snapshotList, int count, double pct) {
final double pos = pct * (count + 1);
if (pos < 1) {
return snapshotList.get(0);
}
if (pos >= count) {
return snapshotList.get(count - 1);
}
return snapshotList.get((int) pos - 1);
}
private HistogramMetric createEmptyMetric() {
HistogramMetric metric = new HistogramMetric();
metric.min = 0;
metric.max = 0;
metric.mean = 0;
for (Double pct : pcts) {
metric.pcts.put(pct, 0L);
}
return metric;
}
}
@@ -0,0 +1,10 @@
package org.springside.modules.metrics;
import java.util.Map;
public class HistogramMetric {
public long min;
public long max;
public double mean;
public Map<Double, Long> pcts;
}
@@ -1,115 +0,0 @@
package org.springside.modules.metrics;
import static java.lang.Math.*;
import java.util.Arrays;
import java.util.Collection;
public class HistogramSnapshot {
private final long[] values;
/**
* Create a new {@link HistogramSnapshot} with the given values.
*
* @param values an unordered set of values in the reservoir
*/
public HistogramSnapshot(Collection<Long> values) {
final Object[] copy = values.toArray();
this.values = new long[copy.length];
for (int i = 0; i < copy.length; i++) {
this.values[i] = (Long) copy[i];
}
Arrays.sort(this.values);
}
/**
* Create a new {@link HistogramSnapshot} with the given values.
*
* @param values an unordered set of values in the reservoir
*/
public HistogramSnapshot(long[] values) {
this.values = Arrays.copyOf(values, values.length);
Arrays.sort(this.values);
}
/**
* Returns the value at the given quantile.
*
* @param quantile a given quantile, in {@code [0..1]}
* @return the value in the distribution at {@code quantile}
*/
public double getValue(double quantile) {
if ((quantile < 0.0) || (quantile > 1.0)) {
throw new IllegalArgumentException(quantile + " is not in [0..1]");
}
if (values.length == 0) {
return 0.0;
}
final double pos = quantile * (values.length + 1);
if (pos < 1) {
return values[0];
}
if (pos >= values.length) {
return values[values.length - 1];
}
final double lower = values[(int) pos - 1];
final double upper = values[(int) pos];
return lower + ((pos - floor(pos)) * (upper - lower));
}
/**
* Returns the value at the 99th percentile in the distribution.
*
* @return the value at the 99th percentile
*/
public double get90th() {
return getValue(0.9);
}
/**
* Returns the highest value in the snapshot.
*
* @return the highest value
*/
public long getMax() {
if (values.length == 0) {
return 0;
}
return values[values.length - 1];
}
/**
* Returns the lowest value in the snapshot.
*
* @return the lowest value
*/
public long getMin() {
if (values.length == 0) {
return 0;
}
return values[0];
}
/**
* Returns the arithmetic mean of the values in the snapshot.
*
* @return the arithmetic mean
*/
public double getMean() {
if (values.length == 0) {
return 0;
}
double sum = 0;
for (long value : values) {
sum += value;
}
return sum / values.length;
}
}
@@ -1,5 +1,82 @@
package org.springside.modules.metrics;
import java.util.Collections;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springside.modules.metrics.utils.Clock;
public class MetricRegistry {
public static final MetricRegistry INSTANCE = new MetricRegistry();
private Clock defaultClock = Clock.DEFAULT;
private Double[] defaultPcts = new Double[] { 0.9 };
private ConcurrentMap<String, Object> metrics = new ConcurrentHashMap<String, Object>();
public static String name(String name, String... subNames) {
final StringBuilder builder = new StringBuilder(name);
if (subNames != null) {
for (String s : subNames) {
if ((s != null) && !s.isEmpty()) {
builder.append('.').append(s);
}
}
}
return builder.toString();
}
public Counter counter(String name) {
Counter counter = new Counter(defaultClock);
return (Counter) register(name, counter);
}
public Histogram histogram(String name) {
Histogram histogram = new Histogram(defaultPcts);
return (Histogram) register(name, histogram);
}
public Execution execution(String name) {
Execution execution = new Execution(defaultClock, defaultPcts);
return (Execution) register(name, execution);
}
public Object register(String name, Object metric) {
return metrics.putIfAbsent(name, metric);
}
public SortedMap<String, Counter> getCounters() {
return getMetrics(Counter.class);
}
public SortedMap<String, Histogram> getHistograms() {
return getMetrics(Histogram.class);
}
public SortedMap<String, Execution> getExecutions() {
return getMetrics(Execution.class);
}
private <T> SortedMap<String, T> getMetrics(Class<T> klass) {
final TreeMap<String, T> result = new TreeMap<String, T>();
for (Map.Entry<String, Object> entry : metrics.entrySet()) {
if (klass.isInstance(entry.getValue())) {
result.put(entry.getKey(), (T) entry.getValue());
}
}
return Collections.unmodifiableSortedMap(result);
}
public void setDefaultClock(Clock clock) {
this.defaultClock = clock;
}
public void setDefaultPcts(Double[] defaultPcts) {
this.defaultPcts = defaultPcts;
}
}
@@ -1,11 +1,13 @@
package org.springside.modules.metrics.report;
import java.io.PrintStream;
import java.util.Date;
import java.util.Map;
import java.util.SortedMap;
import java.util.Map.Entry;
import org.springside.modules.metrics.report.metrics.CounterMetric;
import org.springside.modules.metrics.report.metrics.HistogramMetric;
import org.springside.modules.metrics.CounterMetric;
import org.springside.modules.metrics.ExecutionMetric;
import org.springside.modules.metrics.HistogramMetric;
public class ConsoleReporter implements Reporter {
private static final int CONSOLE_WIDTH = 80;
@@ -13,7 +15,12 @@ public class ConsoleReporter implements Reporter {
private PrintStream output = System.out;
@Override
public void report(SortedMap<String, CounterMetric> counters, SortedMap<String, HistogramMetric> histograms) {
public void report(Map<String, CounterMetric> counters, Map<String, HistogramMetric> histograms,
Map<String, ExecutionMetric> executions) {
printWithBanner(new Date().toString(), '=');
output.println();
if (!counters.isEmpty()) {
printWithBanner("-- Counters", '-');
for (Map.Entry<String, CounterMetric> entry : counters.entrySet()) {
@@ -22,6 +29,24 @@ public class ConsoleReporter implements Reporter {
}
output.println();
}
if (!histograms.isEmpty()) {
printWithBanner("-- Histograms", '-');
for (Map.Entry<String, HistogramMetric> entry : histograms.entrySet()) {
output.println(entry.getKey());
printHistogram(entry.getValue());
}
output.println();
}
if (!executions.isEmpty()) {
printWithBanner("-- Executions", '-');
for (Map.Entry<String, ExecutionMetric> entry : executions.entrySet()) {
output.println(entry.getKey());
printExecution(entry.getValue());
}
output.println();
}
}
private void printWithBanner(String s, char c) {
@@ -34,8 +59,23 @@ public class ConsoleReporter implements Reporter {
}
private void printCounter(CounterMetric counter) {
output.printf(" count = %d%n", counter.coute);
output.printf(" last rate = %2.2f events/%s%n", counter.lastRate);
output.printf(" mean rate = %2.2f events/%s%n", counter.meanRate);
output.printf(" count = %d%n", counter.count);
output.printf(" last rate = %2.2f /%s%n", counter.lastRate);
output.printf(" mean rate = %2.2f /%s%n", counter.meanRate);
}
private void printHistogram(HistogramMetric histogram) {
output.printf(" min = %d%n", histogram.min);
output.printf(" max = %d%n", histogram.max);
output.printf(" mean = %2.2f%n", histogram.mean);
for (Entry<Double, Long> pct : histogram.pcts.entrySet()) {
output.printf(" %d%% <= %2.2f%n", pct.getKey(), pct.getValue());
}
}
private void printExecution(ExecutionMetric execution) {
printCounter(execution.counter);
printHistogram(execution.histogram);
}
}
@@ -1,20 +1,107 @@
package org.springside.modules.metrics.report;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.SortedMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.springside.modules.metrics.Counter;
import org.springside.modules.metrics.CounterMetric;
import org.springside.modules.metrics.Execution;
import org.springside.modules.metrics.ExecutionMetric;
import org.springside.modules.metrics.Histogram;
import org.springside.modules.metrics.HistogramMetric;
import org.springside.modules.metrics.MetricRegistry;
public class ReportScheduler {
private static final String SCHEDULER_NAME = "metrics-reporter";
private List<Reporter> reporters;
private MetricRegistry metricRegistry;
private ScheduledExecutorService executor;
public ReportScheduler(MetricRegistry metricRegistry, List<Reporter> repoters) {
this.reporters = repoters;
public ReportScheduler(MetricRegistry metricRegistry, Reporter... reporters) {
this(metricRegistry, Arrays.asList(reporters));
}
public ReportScheduler(MetricRegistry metricRegistry, List<Reporter> reporters) {
this.metricRegistry = metricRegistry;
this.reporters = reporters;
this.executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory(SCHEDULER_NAME));
}
public void report() {
SortedMap<String, Counter> counterMap = metricRegistry.getCounters();
SortedMap<String, Histogram> histogramMap = metricRegistry.getHistograms();
SortedMap<String, Execution> executionMap = metricRegistry.getExecutions();
Map<String, CounterMetric> counterMetricMap = new LinkedHashMap<String, CounterMetric>();
for (Entry<String, Counter> entry : counterMap.entrySet()) {
counterMetricMap.put(entry.getKey(), entry.getValue().getMetric());
}
Map<String, HistogramMetric> histogramMetricMap = new LinkedHashMap<String, HistogramMetric>();
for (Entry<String, Histogram> entry : histogramMap.entrySet()) {
histogramMetricMap.put(entry.getKey(), entry.getValue().getMetric());
}
Map<String, ExecutionMetric> executionMetricMap = new LinkedHashMap<String, ExecutionMetric>();
for (Entry<String, Execution> entry : executionMap.entrySet()) {
executionMetricMap.put(entry.getKey(), entry.getValue().getMetric());
}
for (Reporter reporter : reporters) {
reporter.report(counterMetricMap, histogramMetricMap, executionMetricMap);
}
}
public void start(long period, TimeUnit unit) {
executor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
report();
} catch (Exception e) {
e.printStackTrace();
}
}
}, period, period, unit);
}
public void stop() {
executor.shutdownNow();
try {
executor.awaitTermination(1, TimeUnit.SECONDS);
} catch (InterruptedException ignored) {
// do nothing
}
}
private static class NamedThreadFactory implements ThreadFactory {
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
private NamedThreadFactory(String name) {
final SecurityManager s = System.getSecurityManager();
this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
this.namePrefix = "metrics-" + name + "-thread-";
}
@Override
public Thread newThread(Runnable r) {
final Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
t.setDaemon(true);
if (t.getPriority() != Thread.NORM_PRIORITY) {
t.setPriority(Thread.NORM_PRIORITY);
}
return t;
}
}
}
@@ -1,10 +1,12 @@
package org.springside.modules.metrics.report;
import java.util.SortedMap;
import java.util.Map;
import org.springside.modules.metrics.report.metrics.CounterMetric;
import org.springside.modules.metrics.report.metrics.HistogramMetric;
import org.springside.modules.metrics.CounterMetric;
import org.springside.modules.metrics.ExecutionMetric;
import org.springside.modules.metrics.HistogramMetric;
public interface Reporter {
void report(SortedMap<String, CounterMetric> counters, SortedMap<String, HistogramMetric> histograms);
void report(Map<String, CounterMetric> counters, Map<String, HistogramMetric> histograms,
Map<String, ExecutionMetric> executions);
}
@@ -1,5 +0,0 @@
package org.springside.modules.metrics.report.metrics;
public class HistogramMetric {
}
@@ -1,6 +1,5 @@
package org.springside.modules.metrics.utils;
import java.util.Date;
/**
* 日期提供者,使用它而不是直接取得系统时间,方便测试。
@@ -9,8 +8,6 @@ import java.util.Date;
*/
public interface Clock {
Date getCurrentDate();
long getCurrentTime();
static final Clock DEFAULT = new DefaultDateProvider();
@@ -20,11 +17,6 @@ public interface Clock {
*/
public static class DefaultDateProvider implements Clock {
@Override
public Date getCurrentDate() {
return new Date();
}
@Override
public long getCurrentTime() {
return System.currentTimeMillis();
@@ -39,31 +31,15 @@ public interface Clock {
private long time;
public MockedDateProvider(Date date) {
this.time = date.getTime();
}
public MockedDateProvider(long time) {
this.time = time;
}
@Override
public Date getCurrentDate() {
return new Date(time);
}
@Override
public long getCurrentTime() {
return time;
}
/**
* 重新设置时间。
*/
public void update(Date newDate) {
time = newDate.getTime();
}
/**
* 重新设置时间。
*/
@@ -0,0 +1,49 @@
package org.springside.modules.metrics;
import static org.junit.Assert.*;
import org.junit.Test;
import org.springside.modules.metrics.utils.Clock.MockedDateProvider;
public class CounterTest {
@Test
public void normal() {
MockedDateProvider clock = new MockedDateProvider(0);
Counter counter = new Counter(clock);
counter.inc(10);
counter.inc(20);
counter.inc(30);
clock.incrementTime(1000);
CounterMetric metric = counter.getMetric();
assertEquals(60, metric.count);
assertEquals(60d, metric.lastRate, 0);
assertEquals(60d, metric.meanRate, 0);
counter.inc(20);
clock.incrementTime(1000);
metric = counter.getMetric();
assertEquals(80, metric.count);
assertEquals(20d, metric.lastRate, 0);
assertEquals(40d, metric.meanRate, 0);
}
@Test
public void incAndDec() {
MockedDateProvider clock = new MockedDateProvider(0);
Counter counter = new Counter(clock);
counter.inc(20);
counter.inc();
counter.inc();
counter.dec(10);
counter.dec();
clock.incrementTime(1000);
CounterMetric metric = counter.getMetric();
assertEquals(11, metric.count);
}
}