You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2021/07/13 15:33:38 UTC
[skywalking] 01/01: Remove synchronous persistence mechanism in API
level and ElasticSearch implementation.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch new-metric-flush
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 5f5909549de6dee574fad570b87a4099f0da92a4
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Tue Jul 13 23:33:17 2021 +0800
Remove synchronous persistence mechanism in API level and ElasticSearch implementation.
---
CHANGES.md | 8 ++++++--
.../analysis/worker/RecordPersistentWorker.java | 2 +-
.../oap/server/core/storage/IBatchDAO.java | 16 +++++++--------
.../oap/server/core/storage/PersistenceTimer.java | 2 +-
.../server/core/storage/PersistenceTimerTest.java | 4 ++--
.../client/elasticsearch/ElasticSearchClient.java | 4 ++++
.../elasticsearch/base/BatchProcessEsDAO.java | 24 ++++++++++------------
.../elasticsearch/base/StorageEsInstaller.java | 5 +----
.../client/ElasticSearch7Client.java | 4 ++++
.../storage/plugin/influxdb/base/BatchDAO.java | 4 ++--
.../storage/plugin/jdbc/h2/dao/H2BatchDAO.java | 8 ++++----
11 files changed, 43 insertions(+), 38 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 0c536f4..7491e73 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -72,15 +72,19 @@ Release Notes.
* Support connectTimeout and socketTimeout settings for ElasticSearch6 and ElasticSearch7 storages.
* Re-implement storage session mechanism, cached metrics are removed only according to their last access timestamp,
rather than first time. This makes sure hot data never gets removed unexpectedly.
-* Support session expired threshold configurable.
+* Support session expired threshold configurable.
* Fix InfluxDB storage-plugin Metrics#multiGet issue.
* Replace zuul proxy with spring cloud gateway 2.x. in webapp module.
* Upgrade etcd cluster coordinator and dynamic configuration to v3.x.
-* Configuration: Allow to configure server maximum request header size.
+* Configuration: Allow configuring server maximum request header size.
* Add thread state metric and class loaded info metric to JVMMetric.
* Performance: compile LAL DSL statically and run with type checked.
* Add pagination to event query protocol.
* Performance: optimize Envoy error logs persistence performance.
+* Performance: remove the synchronous persistence mechanism from batch ElasticSearch DAO. Because the current enhanced
+ persistent session mechanism, don't require the data queryable immediately after the insert and update anymore.
+* Performance: share `flushInterval` setting for both metrics and record data, due
+ to `synchronous persistence mechanism` removed. Record flush interval used to be hardcoded as 10s.
#### UI
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.java
index 9c5f8be..0ef237f 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.java
@@ -49,7 +49,7 @@ public class RecordPersistentWorker extends AbstractWorker<Record> {
public void in(Record record) {
try {
InsertRequest insertRequest = recordDAO.prepareBatchInsert(model, record);
- batchDAO.asynchronous(insertRequest);
+ batchDAO.insert(insertRequest);
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java
index c6bb736..ffc73cd 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java
@@ -23,25 +23,23 @@ import org.apache.skywalking.oap.server.library.client.request.InsertRequest;
import org.apache.skywalking.oap.server.library.client.request.PrepareRequest;
/**
- * IBatchDAO provides two modes of data persistence supported by most databases, including synchronous and
- * asynchronous.
+ * IBatchDAO provides two modes of data persistence supported by most databases, including pure insert and
+ * batch hybrid insert/update.
*/
public interface IBatchDAO extends DAO {
/**
* Push data into the database in async mode. This method is driven by streaming process. This method doesn't
* request the data queryable immediately after the method finished.
*
- * All data are in the additional mode, no modification.
- *
- * @param insertRequest data to insert.
+ * @param prepareRequest data to insert.
*/
- void asynchronous(InsertRequest insertRequest);
+ void insert(InsertRequest prepareRequest);
/**
- * Make all given PrepareRequest efficient in the sync mode. All requests could be confirmed by the database. All
- * changes are required queryable after method returns.
+ * Push data collection into the database in async mode. This method is driven by streaming process. This method
+ * doesn't request the data queryable immediately after the method finished.
*
* @param prepareRequests data to insert or update. No delete happens in streaming mode.
*/
- void synchronous(List<PrepareRequest> prepareRequests);
+ void flush(List<PrepareRequest> prepareRequests);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java
index a7e1e89..6c92af0 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java
@@ -161,7 +161,7 @@ public enum PersistenceTimer {
HistogramMetrics.Timer executeLatencyTimer = executeLatency.createTimer();
try {
if (CollectionUtils.isNotEmpty(partition)) {
- batchDAO.synchronous(partition);
+ batchDAO.flush(partition);
}
} catch (Throwable e) {
log.error(e.getMessage(), e);
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimerTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimerTest.java
index 1fc7355..3258dc8 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimerTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimerTest.java
@@ -57,12 +57,12 @@ public class PersistenceTimerTest {
moduleConfig.setPersistentPeriod(Integer.MAX_VALUE);
IBatchDAO iBatchDAO = new IBatchDAO() {
@Override
- public void asynchronous(InsertRequest insertRequest) {
+ public void insert(InsertRequest prepareRequest) {
}
@Override
- public void synchronous(List<PrepareRequest> prepareRequests) {
+ public void flush(final List<PrepareRequest> prepareRequests) {
synchronized (result) {
result.addAll(prepareRequests);
}
diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java
index 3727fd8..4bc5bab 100644
--- a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java
+++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java
@@ -547,6 +547,10 @@ public class ElasticSearchClient implements Client, HealthCheckable {
return response.getStatusLine().getStatusCode();
}
+ /**
+ * @since 8.7.0 SkyWalking don't use sync bulk anymore. This method is just kept for unexpected case in the future.
+ */
+ @Deprecated
public void synchronousBulk(BulkRequest request) {
request.timeout(TimeValue.timeValueMinutes(2));
request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java
index bcbe21b..7dea42d 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java
@@ -19,22 +19,18 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
import java.util.List;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.storage.IBatchDAO;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.apache.skywalking.oap.server.library.client.request.InsertRequest;
import org.apache.skywalking.oap.server.library.client.request.PrepareRequest;
import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.elasticsearch.action.bulk.BulkProcessor;
-import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+@Slf4j
public class BatchProcessEsDAO extends EsDAO implements IBatchDAO {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(BatchProcessEsDAO.class);
-
private BulkProcessor bulkProcessor;
private final int bulkActions;
private final int flushInterval;
@@ -51,26 +47,28 @@ public class BatchProcessEsDAO extends EsDAO implements IBatchDAO {
}
@Override
- public void asynchronous(InsertRequest insertRequest) {
+ public void insert(InsertRequest prepareRequest) {
if (bulkProcessor == null) {
this.bulkProcessor = getClient().createBulkProcessor(bulkActions, flushInterval, concurrentRequests);
}
- this.bulkProcessor.add((IndexRequest) insertRequest);
+ this.bulkProcessor.add((IndexRequest) prepareRequest);
}
@Override
- public void synchronous(List<PrepareRequest> prepareRequests) {
+ public void flush(List<PrepareRequest> prepareRequests) {
+ if (bulkProcessor == null) {
+ this.bulkProcessor = getClient().createBulkProcessor(bulkActions, flushInterval, concurrentRequests);
+ }
+
if (CollectionUtils.isNotEmpty(prepareRequests)) {
- BulkRequest request = new BulkRequest();
for (PrepareRequest prepareRequest : prepareRequests) {
if (prepareRequest instanceof InsertRequest) {
- request.add((IndexRequest) prepareRequest);
+ this.bulkProcessor.add((IndexRequest) prepareRequest);
} else {
- request.add((UpdateRequest) prepareRequest);
+ this.bulkProcessor.add((UpdateRequest) prepareRequest);
}
}
- getClient().synchronousBulk(request);
}
}
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
index 7d7e6a9..073792c 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
@@ -33,7 +33,6 @@ import org.apache.skywalking.oap.server.library.client.Client;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
-import org.elasticsearch.common.unit.TimeValue;
@Slf4j
public class StorageEsInstaller extends ModelInstaller {
@@ -162,9 +161,7 @@ public class StorageEsInstaller extends ModelInstaller {
setting.put("index.number_of_shards", model.isSuperDataset()
? config.getIndexShardsNumber() * config.getSuperDatasetIndexShardsFactor()
: config.getIndexShardsNumber());
- setting.put("index.refresh_interval", model.isRecord()
- ? TimeValue.timeValueSeconds(10).toString()
- : TimeValue.timeValueSeconds(config.getFlushInterval()).toString());
+ setting.put("index.refresh_interval", config.getFlushInterval());
setting.put("analysis", getAnalyzerSetting(model.getColumns()));
if (!StringUtil.isEmpty(config.getAdvanced())) {
Map<String, Object> advancedSettings = gson.fromJson(config.getAdvanced(), Map.class);
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java
index 2d720f0..4129a88 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java
@@ -392,6 +392,10 @@ public class ElasticSearch7Client extends ElasticSearchClient {
return HttpStatus.SC_OK;
}
+ /**
+ * @since 8.7.0 SkyWalking don't use sync bulk anymore. This method is just kept for unexpected case in the future.
+ */
+ @Deprecated
@Override
public void synchronousBulk(BulkRequest request) {
request.timeout(TimeValue.timeValueMinutes(2));
diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/BatchDAO.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/BatchDAO.java
index 713a0b1..105383d 100644
--- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/BatchDAO.java
+++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/base/BatchDAO.java
@@ -36,12 +36,12 @@ public class BatchDAO implements IBatchDAO {
}
@Override
- public void asynchronous(InsertRequest insertRequest) {
+ public void insert(InsertRequest insertRequest) {
client.write(((InfluxInsertRequest) insertRequest).getPoint());
}
@Override
- public void synchronous(List<PrepareRequest> prepareRequests) {
+ public void flush(List<PrepareRequest> prepareRequests) {
if (CollectionUtils.isEmpty(prepareRequests)) {
return;
}
diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2BatchDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2BatchDAO.java
index 3539fe7..cd39060 100644
--- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2BatchDAO.java
+++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2BatchDAO.java
@@ -56,7 +56,7 @@ public class H2BatchDAO implements IBatchDAO {
}
@Override
- public void synchronous(List<PrepareRequest> prepareRequests) {
+ public void flush(List<PrepareRequest> prepareRequests) {
if (CollectionUtils.isEmpty(prepareRequests)) {
return;
}
@@ -81,8 +81,8 @@ public class H2BatchDAO implements IBatchDAO {
}
@Override
- public void asynchronous(InsertRequest insertRequest) {
- this.dataCarrier.produce(insertRequest);
+ public void insert(InsertRequest prepareRequest) {
+ this.dataCarrier.produce(prepareRequest);
}
private class H2BatchConsumer implements IConsumer<PrepareRequest> {
@@ -100,7 +100,7 @@ public class H2BatchDAO implements IBatchDAO {
@Override
public void consume(List<PrepareRequest> prepareRequests) {
- h2BatchDAO.synchronous(prepareRequests);
+ h2BatchDAO.flush(prepareRequests);
}
@Override