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

Optimize L1 aggregation, enhance gRPC client between L1 and L2 aggregation. (#7206) 

1.0Performance: Add L1 aggregation flush period, which reduces the CPU load and helps young GC.

2. Replace do not direct send after the first aggregation to reduce the network #6400.

3. Enhance the DataCarrier to notify the consumer in no enqueue event in short term.

4. L1 aggregation flush period still works even no further metrics generated, powered by <3>

5. Fix gRPC remote client OOM. The concurrency control mechanism failed.
This commit is contained in:
吴晟 Wu Sheng 2021-06-30 19:14:20 +08:00 committed by GitHub
parent 4374fac4cb
commit 50a0bcbf73
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 112 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGES.md
View File

@ -9,6 +9,8 @@ Release Notes.
* Extract dependency management to a bom.
* Add JDK 16 to test matrix.
* DataCarrier consumer add a new event notification, call `nothingToConsume` method if the queue has no element to
consume.
#### Java Agent
@ -56,7 +58,8 @@ Release Notes.
* Performance: cache regex pattern and result, optimize string concatenation in Envy ALS analyzer.
* Performance: cache metrics id and entity id in `Metrics` and `ISource`.
* Performance: enhance persistent session mechanism, about differentiating cache timeout for different dimensionality
metrics. The timeout of the cache for minute and hour level metrics has been prolonged to ~5 min.
metrics. The timeout of the cache for minute and hour level metrics has been prolonged to ~5 min.
* Performance: Add L1 aggregation flush period, which reduce the CPU load and help young GC.
#### UI

1
apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java
View File

@ -80,6 +80,7 @@ public class ConsumerThread<T> extends Thread {
}
return true;
}
consumer.nothingToConsume();
return false;
}

8
apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/IConsumer.java
View File

@ -28,4 +28,12 @@ public interface IConsumer<T> {
void onError(List<T> data, Throwable t);
void onExit();
/**
* Notify the implementation, if there is nothing fetched from the queue. This could be used as a timer to trigger
* reaction if the queue has no element.
*/
default void nothingToConsume() {
return;
}
}

3
apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/MultipleChannelsConsumer.java
View File

@ -85,6 +85,7 @@ public class MultipleChannelsConsumer extends Thread {
}
return true;
}
target.consumer.nothingToConsume();
return false;
}
@ -112,7 +113,7 @@ public class MultipleChannelsConsumer extends Thread {
}
private static class Group {
private Channels channels;
private Channels channels;
private IConsumer consumer;
public Group(Channels channels, IConsumer consumer) {

1
docs/en/setup/backend/configuration-vocabulary.md
View File

@ -23,6 +23,7 @@ core|default|role|Option values, `Mixed/Receiver/Aggregator`. **Receiver** mode
| - | - | dataKeeperExecutePeriod|The execution period of TTL scheduler, unit is minute. Execution doesn't mean deleting data. The storage provider could override this, such as ElasticSearch storage.|SW_CORE_DATA_KEEPER_EXECUTE_PERIOD|5|
| - | - | recordDataTTL|The lifecycle of record data. Record data includes traces, top n sampled records, and logs. Unit is day. Minimal value is 2.|SW_CORE_RECORD_DATA_TTL|3|
| - | - | metricsDataTTL|The lifecycle of metrics data, including the metadata. Unit is day. Recommend metricsDataTTL >= recordDataTTL. Minimal value is 2.| SW_CORE_METRICS_DATA_TTL|7|
| - | - | l1FlushPeriod| The period of L1 aggregation flush to L2 aggregation. Unit is ms. | SW_CORE_L1_AGGREGATION_FLUSH_PERIOD | 500 |
| - | - | enableDatabaseSession|Cache metrics data for 1 minute to reduce database queries, and if the OAP cluster changes within that minute.|SW_CORE_ENABLE_DATABASE_SESSION|true|
| - | - | topNReportPeriod|The execution period of top N sampler, which saves sampled data into the storage. Unit is minute|SW_CORE_TOPN_REPORT_PERIOD|10|
| - | - | activeExtraModelColumns|Append the names of entity, such as service name, into the metrics storage entities.|SW_CORE_ACTIVE_EXTRA_MODEL_COLUMNS|false|

2
oap-server/server-bootstrap/src/main/resources/application.yml
View File

@ -85,6 +85,8 @@ core:
dataKeeperExecutePeriod: ${SW_CORE_DATA_KEEPER_EXECUTE_PERIOD:5} # How often the data keeper executor runs periodically, unit is minute
recordDataTTL: ${SW_CORE_RECORD_DATA_TTL:3} # Unit is day
metricsDataTTL: ${SW_CORE_METRICS_DATA_TTL:7} # Unit is day
# The period of L1 aggregation flush to L2 aggregation. Unit is ms.
l1FlushPeriod: ${SW_CORE_L1_AGGREGATION_FLUSH_PERIOD:500}
# Cache metrics data for 1 minute to reduce database queries, and if the OAP cluster changes within that minute,
# the metrics may not be accurate within that minute.
enableDatabaseSession: ${SW_CORE_ENABLE_DATABASE_SESSION:true}

9
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java
View File

@ -46,8 +46,15 @@ public class CoreModuleConfig extends ModuleConfig {
private String gRPCSslTrustedCAPath;
private int maxConcurrentCallsPerConnection;
private int maxMessageSize;
private boolean enableDatabaseSession;
private int topNReportPeriod;
/**
* The period of L1 aggregation flush. Unit is ms.
*/
private long l1FlushPeriod = 500;
/**
* Enable database flush session.
*/
private boolean enableDatabaseSession;
private final List<String> downsampling;
/**
* The period of doing data persistence. Unit is second.

3
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
View File

@ -46,9 +46,9 @@ import org.apache.skywalking.oap.server.core.config.ConfigService;
import org.apache.skywalking.oap.server.core.config.DownSamplingConfigService;
import org.apache.skywalking.oap.server.core.config.IComponentLibraryCatalogService;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.config.group.openapi.EndpointGroupingRuleReader4Openapi;
import org.apache.skywalking.oap.server.core.config.group.EndpointNameGrouping;
import org.apache.skywalking.oap.server.core.config.group.EndpointNameGroupingRuleWatcher;
import org.apache.skywalking.oap.server.core.config.group.openapi.EndpointGroupingRuleReader4Openapi;
import org.apache.skywalking.oap.server.core.management.ui.template.UITemplateInitializer;
import org.apache.skywalking.oap.server.core.management.ui.template.UITemplateManagementService;
import org.apache.skywalking.oap.server.core.oal.rt.DisableOALDefine;
@ -289,6 +289,7 @@ public class CoreModuleProvider extends ModuleProvider {
UITemplateManagementService.class, new UITemplateManagementService(getManager()));
MetricsStreamProcessor.getInstance().setEnableDatabaseSession(moduleConfig.isEnableDatabaseSession());
MetricsStreamProcessor.getInstance().setL1FlushPeriod(moduleConfig.getL1FlushPeriod());
TopNStreamProcessor.getInstance().setTopNWorkerReportCycle(moduleConfig.getTopNReportPeriod());
apdexThresholdConfig = new ApdexThresholdConfig(this);
ApdexMetrics.setDICT(apdexThresholdConfig);

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

@ -42,13 +42,15 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
*/
@Slf4j
public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
public final long l1FlushPeriod;
private AbstractWorker<Metrics> nextWorker;
private final DataCarrier<Metrics> dataCarrier;
private final MergableBufferedData<Metrics> mergeDataCache;
private CounterMetrics aggregationCounter;
private long lastSendTime = 0;
MetricsAggregateWorker(ModuleDefineHolder moduleDefineHolder, AbstractWorker<Metrics> nextWorker,
String modelName) {
String modelName, long l1FlushPeriod) {
super(moduleDefineHolder);
this.nextWorker = nextWorker;
this.mergeDataCache = new MergableBufferedData();
@ -69,8 +71,10 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
.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", "minute")
new MetricsTag.Keys("metricName", "level", "dimensionality"),
new MetricsTag.Values(modelName, "1", "minute")
);
this.l1FlushPeriod = l1FlushPeriod;
}
/**
@ -93,14 +97,22 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
mergeDataCache.accept(metrics);
});
mergeDataCache.read().forEach(
data -> {
if (log.isDebugEnabled()) {
log.debug(data.toString());
flush();
}
private void flush() {
long currentTime = System.currentTimeMillis();
if (currentTime - lastSendTime > l1FlushPeriod) {
mergeDataCache.read().forEach(
data -> {
if (log.isDebugEnabled()) {
log.debug(data.toString());
}
nextWorker.in(data);
}
nextWorker.in(data);
}
);
);
lastSendTime = currentTime;
}
}
private class AggregatorConsumer implements IConsumer<Metrics> {
@ -121,5 +133,10 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
@Override
public void onExit() {
}
@Override
public void nothingToConsume() {
flush();
}
}
}
}

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

@ -70,6 +70,12 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
@Getter
private List<MetricsPersistentWorker> persistentWorkers = new ArrayList<>();
/**
* The period of L1 aggregation flush. Unit is ms.
*/
@Setter
@Getter
private long l1FlushPeriod = 500;
/**
* Hold and forward CoreModuleConfig#enableDatabaseSession to the persistent worker.
*/
@ -97,7 +103,9 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
* @param metricsClass data type of the streaming calculation.
*/
@Override
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends Metrics> metricsClass) throws StorageException {
public void create(ModuleDefineHolder moduleDefineHolder,
Stream stream,
Class<? extends Metrics> metricsClass) throws StorageException {
this.create(moduleDefineHolder, StreamDefinition.from(stream), metricsClass);
}
@ -108,7 +116,8 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
final StorageBuilderFactory storageBuilderFactory = moduleDefineHolder.find(StorageModule.NAME)
.provider()
.getService(StorageBuilderFactory.class);
final Class<? extends StorageBuilder> builder = storageBuilderFactory.builderOf(metricsClass, stream.getBuilder());
final Class<? extends StorageBuilder> builder = storageBuilderFactory.builderOf(
metricsClass, stream.getBuilder());
StorageDAO storageDAO = moduleDefineHolder.find(StorageModule.NAME).provider().getService(StorageDAO.class);
IMetricsDAO metricsDAO;
@ -167,7 +176,7 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
MetricsRemoteWorker remoteWorker = new MetricsRemoteWorker(moduleDefineHolder, remoteReceiverWorkerName);
MetricsAggregateWorker aggregateWorker = new MetricsAggregateWorker(
moduleDefineHolder, remoteWorker, stream.getName());
moduleDefineHolder, remoteWorker, stream.getName(), l1FlushPeriod);
entryWorkers.put(metricsClass, aggregateWorker);
}

74
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java
View File

@ -25,6 +25,7 @@ import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.core.remote.data.StreamData;
@ -37,17 +38,13 @@ import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This is a wrapper of the gRPC client for sending message to each other OAP server. It contains a block queue to
* buffering the message and sending the message by batch.
*/
@Slf4j
public class GRPCRemoteClient implements RemoteClient {
private static final Logger LOGGER = LoggerFactory.getLogger(GRPCRemoteClient.class);
private final int channelSize;
private final int bufferSize;
private final Address address;
@ -76,13 +73,23 @@ public class GRPCRemoteClient implements RemoteClient {
remoteOutCounter = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class)
.createCounter("remote_out_count", "The number(client side) of inside remote inside aggregate rpc.", new MetricsTag.Keys("dest", "self"), new MetricsTag.Values(address
.toString(), "N"));
.createCounter(
"remote_out_count",
"The number(client side) of inside remote inside aggregate rpc.",
new MetricsTag.Keys("dest", "self"), new MetricsTag.Values(
address
.toString(), "N")
);
remoteOutErrorCounter = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class)
.createCounter("remote_out_error_count", "The error number(client side) of inside remote inside aggregate rpc.", new MetricsTag.Keys("dest", "self"), new MetricsTag.Values(address
.toString(), "N"));
.createCounter(
"remote_out_error_count",
"The error number(client side) of inside remote inside aggregate rpc.",
new MetricsTag.Keys("dest", "self"), new MetricsTag.Values(
address
.toString(), "N")
);
}
@Override
@ -160,13 +167,13 @@ public class GRPCRemoteClient implements RemoteClient {
streamObserver.onCompleted();
} catch (Throwable t) {
remoteOutErrorCounter.inc();
LOGGER.error(t.getMessage(), t);
log.error(t.getMessage(), t);
}
}
@Override
public void onError(List<RemoteMessage> remoteMessages, Throwable t) {
LOGGER.error(t.getMessage(), t);
log.error(t.getMessage(), t);
}
@Override
@ -183,38 +190,47 @@ public class GRPCRemoteClient implements RemoteClient {
private StreamObserver<RemoteMessage> createStreamObserver() {
int sleepTotalMillis = 0;
int sleepMillis = 10;
while (concurrentStreamObserverNumber.incrementAndGet() > 10) {
concurrentStreamObserverNumber.addAndGet(-1);
// Control the concurrency of gRPC streaming stub.
// If over 10 created and not finished/error, this blocks the method.
while (concurrentStreamObserverNumber.get() > 10) {
try {
Thread.sleep(sleepMillis);
} catch (InterruptedException e) {
LOGGER.error(e.getMessage(), e);
log.error(e.getMessage(), e);
}
sleepTotalMillis += sleepMillis;
if (sleepTotalMillis > 60000) {
LOGGER.warn("Remote client block times over 60 seconds.");
log.warn("Remote client [{}] block times over 60 seconds. Current streaming number {}",
address, concurrentStreamObserverNumber.get()
);
// Reset sleepTotalMillis to avoid too many warn logs.
sleepTotalMillis = 0;
}
}
return getStub().withDeadlineAfter(remoteTimeout, TimeUnit.SECONDS).call(new StreamObserver<Empty>() {
@Override
public void onNext(Empty empty) {
}
final StreamObserver<RemoteMessage> remoteMessageStreamObserver
= getStub().withDeadlineAfter(remoteTimeout, TimeUnit.SECONDS)
.call(new StreamObserver<Empty>() {
@Override
public void onNext(Empty empty) {
}
@Override
public void onError(Throwable throwable) {
concurrentStreamObserverNumber.addAndGet(-1);
LOGGER.error(throwable.getMessage(), throwable);
}
@Override
public void onError(Throwable throwable) {
concurrentStreamObserverNumber.addAndGet(-1);
log.error(throwable.getMessage(), throwable);
}
@Override
public void onCompleted() {
concurrentStreamObserverNumber.addAndGet(-1);
}
});
@Override
public void onCompleted() {
concurrentStreamObserverNumber.addAndGet(-1);
}
});
concurrentStreamObserverNumber.incrementAndGet();
return remoteMessageStreamObserver;
}
@Override