kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Aggregate worker will hold the metrics while there is no new metrics in. (#3100)

This commit is contained in:
彭勇升 pengys 2019-07-17 13:59:54 +08:00 committed by 吴晟 Wu Sheng
parent 8faa356403
commit 54b01223de
1 changed files with 3 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java
View File

@ -19,7 +19,7 @@
package org.apache.skywalking.oap.server.core.analysis.worker;
import java.util.*;
import org.apache.skywalking.apm.commons.datacarrier.*;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
import org.apache.skywalking.apm.commons.datacarrier.consumer.*;
import org.apache.skywalking.oap.server.core.UnexpectedException;
import org.apache.skywalking.oap.server.core.analysis.data.*;
@ -40,15 +40,10 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
private AbstractWorker<Metrics> nextWorker;
private final DataCarrier<Metrics> dataCarrier;
private final MergeDataCache<Metrics> mergeDataCache;
private final String modelName;
private CounterMetrics aggregationCounter;
private final long l2AggregationSendCycle;
private long lastSendTimestamp;
MetricsAggregateWorker(ModuleDefineHolder moduleDefineHolder, AbstractWorker<Metrics> nextWorker,
String modelName) {
MetricsAggregateWorker(ModuleDefineHolder moduleDefineHolder, AbstractWorker<Metrics> nextWorker, String modelName) {
super(moduleDefineHolder);
this.modelName = modelName;
this.nextWorker = nextWorker;
this.mergeDataCache = new MergeDataCache<>();
String name = "METRICS_L1_AGGREGATION";
@ -65,9 +60,6 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
MetricsCreator metricsCreator = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class);
aggregationCounter = metricsCreator.createCounter("metrics_aggregation", "The number of rows in aggregation",
new MetricsTag.Keys("metricName", "level", "dimensionality"), new MetricsTag.Values(modelName, "1", "min"));
lastSendTimestamp = System.currentTimeMillis();
l2AggregationSendCycle = EnvUtil.getLong("METRICS_L1_AGGREGATION_SEND_CYCLE", 1000);
}
@Override public final void in(Metrics metrics) {
@ -80,22 +72,10 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
aggregate(metrics);
if (metrics.getEndOfBatchContext().isEndOfBatch()) {
if (shouldSend()) {
sendToNext();
}
sendToNext();
}
}
private boolean shouldSend() {
long now = System.currentTimeMillis();
// Continue L2 aggregation in certain cycle.
if (now - lastSendTimestamp > l2AggregationSendCycle) {
lastSendTimestamp = now;
return true;
}
return false;
}
private void sendToNext() {
mergeDataCache.switchPointer();
while (mergeDataCache.getLast().isWriting()) {