make sync metrics concurrency (#6642)
* make sync metrics concurrency * add changelog * add changelog * polish codes * change default value * remove unnecessary codes Co-authored-by: Evan <evanljp@outlook.com> Co-authored-by: Zhenxu Ke <kezhenxu94@apache.org> Co-authored-by: 吴晟 Wu Sheng <wu.sheng@foxmail.com>
This commit is contained in:
Evan
GitHub
parent
fc7d8dcb5e
commit
2af77d4d03
|
|
@ -5,7 +5,7 @@ Release Notes.
|
|||
8.5.0
|
||||
------------------
|
||||
#### Project
|
||||
* **Incompatible Change**. Indices and templates of ElasticSearch(6/7, including zipkin-elasticsearch7) storage option have been changed.
|
||||
* **Incompatible Change**. Indices and templates of ElasticSearch(6/7, including zipkin-elasticsearch7) storage option have been changed.
|
||||
* Update frontend-maven-plugin to 1.11.0, for Download node x64 binary on Apple Silicon.
|
||||
* Add E2E test for VM monitoring that metrics from Prometheus node-exporter.
|
||||
* Upgrade lombok to 1.18.16.
|
||||
|
|
@ -50,13 +50,13 @@ Release Notes.
|
|||
* Upgrade the Zipkin Elasticsearch storage from 6 to 7.
|
||||
* Require Zipkin receiver must work with `zipkin-elasticsearch7` storage option.
|
||||
* Fix `DatabaseSlowStatementBuilder` statement maybe null.
|
||||
* Remove fields of parent entity in the relation sources.
|
||||
* Remove fields of parent entity in the relation sources.
|
||||
* Save Envoy http access logs when error occurs.
|
||||
* Fix wrong `service_instance_sla` setting in the `topology-instance.yml`.
|
||||
* Fix wrong metrics name setting in the `self-observability.yml`.
|
||||
* Add telemetry data about metrics in, metrics scraping, mesh error and trace in metrics to zipkin receiver.
|
||||
* Fix tags store of log and trace on h2/mysql/pg storage.
|
||||
* Merge indices by Metrics Function and Meter Function in Elasticsearch Storage.
|
||||
* Merge indices by Metrics Function and Meter Function in Elasticsearch Storage.
|
||||
* Fix receiver don't need to get itself when healthCheck
|
||||
* Remove group concept from AvgHistogramFunction. Heatmap(function result) doesn't support labels.
|
||||
* Support metrics grouped by scope labelValue in MAL, no need global same labelValue as before.
|
||||
|
|
@ -64,6 +64,7 @@ Release Notes.
|
|||
* Optimize the self monitoring grafana dashboard.
|
||||
* Enhance the export service.
|
||||
* Add function `retagByK8sMeta` and opt type `K8sRetagType.Pod2Service` in MAL for k8s to relate pods and services.
|
||||
* Make the flushing metrics operation concurrent.
|
||||
* Fix ALS K8SServiceRegistry didn't remove the correct entry.
|
||||
* Using "service.istio.io/canonical-name" to replace "app" label to resolve Envoy ALS service name
|
||||
* Append the root slash(/) to getIndex and getTemplate requests in ES client
|
||||
|
|
|
|||
|
|
@ -62,7 +62,6 @@ storage:
|
|||
superDatasetIndexShardsFactor: ${SW_STORAGE_ES_SUPER_DATASET_INDEX_SHARDS_FACTOR:5} # This factor provides more shards for the super data set, shards number = indexShardsNumber * superDatasetIndexShardsFactor. Also, this factor effects Zipkin and Jaeger traces.
|
||||
superDatasetIndexReplicasNumber: ${SW_STORAGE_ES_SUPER_DATASET_INDEX_REPLICAS_NUMBER:0} # Represent the replicas number in the super size dataset record index, the default value is 0.
|
||||
bulkActions: ${SW_STORAGE_ES_BULK_ACTIONS:1000} # Execute the async bulk record data every ${SW_STORAGE_ES_BULK_ACTIONS} requests
|
||||
syncBulkActions: ${SW_STORAGE_ES_SYNC_BULK_ACTIONS:50000} # Execute the sync bulk metrics data every ${SW_STORAGE_ES_SYNC_BULK_ACTIONS} requests
|
||||
flushInterval: ${SW_STORAGE_ES_FLUSH_INTERVAL:10} # flush the bulk every 10 seconds whatever the number of requests
|
||||
concurrentRequests: ${SW_STORAGE_ES_CONCURRENT_REQUESTS:2} # the number of concurrent requests
|
||||
resultWindowMaxSize: ${SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE:10000}
|
||||
|
|
|
|||
|
|
@ -39,6 +39,8 @@ core|default|role|Option values, `Mixed/Receiver/Aggregator`. **Receiver** mode
|
|||
| - | - | maxSizeOfNetworkAddressAlias|Max size of network address detected in the be monitored system.| - | 1_000_000|
|
||||
| - | - | maxPageSizeOfQueryProfileSnapshot|The max size in every OAP query for snapshot analysis| - | 500 |
|
||||
| - | - | maxSizeOfAnalyzeProfileSnapshot|The max number of snapshots analyzed by OAP| - | 12000 |
|
||||
| - | - | syncThreads|The number of threads used to synchronously refresh the metrics data to the storage.| SW_CORE_SYNC_THREADS | 2 |
|
||||
| - | - | maxSyncOperationNum|The maximum number of processes supported for each synchronous storage operation. When the number of the flush data is greater than this value, it will be assigned to multiple cores for execution.| SW_CORE_MAX_SYNC_OPERATION_NUM | 50000 |
|
||||
|cluster|standalone| - | standalone is not suitable for one node running, no available configuration.| - | - |
|
||||
| - | zookeeper|nameSpace|The namespace, represented by root path, isolates the configurations in the zookeeper.|SW_NAMESPACE| `/`, root path|
|
||||
| - | - | hostPort|hosts and ports of Zookeeper Cluster|SW_CLUSTER_ZK_HOST_PORT| localhost:2181|
|
||||
|
|
@ -87,7 +89,6 @@ core|default|role|Option values, `Mixed/Receiver/Aggregator`. **Receiver** mode
|
|||
| - | - | superDatasetIndexShardsFactor | Super data set has been defined in the codes, such as trace segments. This factor provides more shards for the super data set, shards number = indexShardsNumber * superDatasetIndexShardsFactor. Also, this factor effects Zipkin and Jaeger traces.|SW_STORAGE_ES_SUPER_DATASET_INDEX_SHARDS_FACTOR|5 |
|
||||
| - | - | superDatasetIndexReplicasNumber | Represent the replicas number in the super size dataset record index.|SW_STORAGE_ES_SUPER_DATASET_INDEX_REPLICAS_NUMBER|0 |
|
||||
| - | - | bulkActions| Async bulk size of the record data batch execution. | SW_STORAGE_ES_BULK_ACTIONS| 1000|
|
||||
| - | - | syncBulkActions| Sync bulk size of the metrics data batch execution. | SW_STORAGE_ES_SYNC_BULK_ACTIONS| 50000|
|
||||
| - | - | flushInterval| Period of flush, no matter `bulkActions` reached or not. Unit is second.| SW_STORAGE_ES_FLUSH_INTERVAL | 10|
|
||||
| - | - | concurrentRequests| The number of concurrent requests allowed to be executed. | SW_STORAGE_ES_CONCURRENT_REQUESTS| 2 |
|
||||
| - | - | resultWindowMaxSize | The max size of dataset when OAP loading cache, such as network alias. | SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE | 10000|
|
||||
|
|
|
|||
|
|
@ -102,6 +102,10 @@ core:
|
|||
searchableTracesTags: ${SW_SEARCHABLE_TAG_KEYS:http.method,status_code,db.type,db.instance,mq.queue,mq.topic,mq.broker}
|
||||
# Define the set of log tag keys, which should be searchable through the GraphQL.
|
||||
searchableLogsTags: ${SW_SEARCHABLE_LOGS_TAG_KEYS:level}
|
||||
# The number of threads used to synchronously refresh the metrics data to the storage.
|
||||
syncThreads: ${SW_CORE_SYNC_THREADS:2}
|
||||
# The maximum number of processes supported for each synchronous storage operation. When the number of the flush data is greater than this value, it will be assigned to multiple cores for execution.
|
||||
maxSyncOperationNum: ${SW_CORE_MAX_SYNC_OPERATION_NUM:50000}
|
||||
storage:
|
||||
selector: ${SW_STORAGE:h2}
|
||||
elasticsearch:
|
||||
|
|
@ -121,7 +125,6 @@ storage:
|
|||
superDatasetIndexShardsFactor: ${SW_STORAGE_ES_SUPER_DATASET_INDEX_SHARDS_FACTOR:5} # This factor provides more shards for the super data set, shards number = indexShardsNumber * superDatasetIndexShardsFactor. Also, this factor effects Zipkin and Jaeger traces.
|
||||
superDatasetIndexReplicasNumber: ${SW_STORAGE_ES_SUPER_DATASET_INDEX_REPLICAS_NUMBER:0} # Represent the replicas number in the super size dataset record index, the default value is 0.
|
||||
bulkActions: ${SW_STORAGE_ES_BULK_ACTIONS:1000} # Execute the async bulk record data every ${SW_STORAGE_ES_BULK_ACTIONS} requests
|
||||
syncBulkActions: ${SW_STORAGE_ES_SYNC_BULK_ACTIONS:50000} # Execute the sync bulk metrics data every ${SW_STORAGE_ES_SYNC_BULK_ACTIONS} requests
|
||||
flushInterval: ${SW_STORAGE_ES_FLUSH_INTERVAL:10} # flush the bulk every 10 seconds whatever the number of requests
|
||||
concurrentRequests: ${SW_STORAGE_ES_CONCURRENT_REQUESTS:2} # the number of concurrent requests
|
||||
resultWindowMaxSize: ${SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE:10000}
|
||||
|
|
@ -148,7 +151,6 @@ storage:
|
|||
password: ${SW_ES_PASSWORD:""}
|
||||
secretsManagementFile: ${SW_ES_SECRETS_MANAGEMENT_FILE:""} # Secrets management file in the properties format includes the username, password, which are managed by 3rd party tool.
|
||||
bulkActions: ${SW_STORAGE_ES_BULK_ACTIONS:1000} # Execute the async bulk record data every ${SW_STORAGE_ES_BULK_ACTIONS} requests
|
||||
syncBulkActions: ${SW_STORAGE_ES_SYNC_BULK_ACTIONS:50000} # Execute the sync bulk metrics data every ${SW_STORAGE_ES_SYNC_BULK_ACTIONS} requests
|
||||
flushInterval: ${SW_STORAGE_ES_FLUSH_INTERVAL:10} # flush the bulk every 10 seconds whatever the number of requests
|
||||
concurrentRequests: ${SW_STORAGE_ES_CONCURRENT_REQUESTS:2} # the number of concurrent requests
|
||||
resultWindowMaxSize: ${SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE:10000}
|
||||
|
|
|
|||
|
|
@ -134,6 +134,23 @@ public class CoreModuleConfig extends ModuleConfig {
|
|||
@Getter
|
||||
private String searchableLogsTags = "";
|
||||
|
||||
/**
|
||||
* The number of threads used to synchronously refresh the metrics data to the storage.
|
||||
*
|
||||
* @since 8.5.0
|
||||
*/
|
||||
@Setter
|
||||
@Getter
|
||||
private int syncThreads = 2;
|
||||
|
||||
/**
|
||||
* The maximum number of processes supported for each synchronous storage operation. When the number of the flush
|
||||
* data is greater than this value, it will be assigned to multiple cores for execution.
|
||||
*/
|
||||
@Getter
|
||||
@Setter
|
||||
private int maxSyncOperationNum = 50000;
|
||||
|
||||
public CoreModuleConfig() {
|
||||
this.downsampling = new ArrayList<>();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,8 +18,11 @@
|
|||
|
||||
package org.apache.skywalking.oap.server.core.storage;
|
||||
|
||||
import com.google.common.collect.Lists;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
|
@ -47,6 +50,9 @@ public enum PersistenceTimer {
|
|||
private HistogramMetrics executeLatency;
|
||||
private long lastTime = System.currentTimeMillis();
|
||||
private final List<PrepareRequest> prepareRequests = new ArrayList<>(50000);
|
||||
private int syncOperationThreadsNum;
|
||||
private int maxSyncoperationNum;
|
||||
private ExecutorService executorService;
|
||||
|
||||
PersistenceTimer() {
|
||||
this.debug = System.getProperty("debug") != null;
|
||||
|
|
@ -71,7 +77,9 @@ public enum PersistenceTimer {
|
|||
"persistence_timer_bulk_execute_latency", "Latency of the execute stage in persistence timer",
|
||||
MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
|
||||
);
|
||||
|
||||
syncOperationThreadsNum = moduleConfig.getSyncThreads();
|
||||
maxSyncoperationNum = moduleConfig.getMaxSyncOperationNum();
|
||||
executorService = Executors.newFixedThreadPool(syncOperationThreadsNum);
|
||||
if (!isStarted) {
|
||||
Executors.newSingleThreadScheduledExecutor()
|
||||
.scheduleWithFixedDelay(
|
||||
|
|
@ -118,9 +126,22 @@ public enum PersistenceTimer {
|
|||
|
||||
HistogramMetrics.Timer executeLatencyTimer = executeLatency.createTimer();
|
||||
try {
|
||||
if (CollectionUtils.isNotEmpty(prepareRequests)) {
|
||||
batchDAO.synchronous(prepareRequests);
|
||||
List<List<PrepareRequest>> partitions = Lists.partition(prepareRequests, maxSyncoperationNum);
|
||||
CountDownLatch countDownLatch = new CountDownLatch(partitions.size());
|
||||
for (final List<PrepareRequest> partition : partitions) {
|
||||
executorService.submit(() -> {
|
||||
try {
|
||||
if (CollectionUtils.isNotEmpty(partition)) {
|
||||
batchDAO.synchronous(partition);
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
log.error(e.getMessage(), e);
|
||||
} finally {
|
||||
countDownLatch.countDown();
|
||||
}
|
||||
});
|
||||
}
|
||||
countDownLatch.await();
|
||||
} finally {
|
||||
executeLatencyTimer.finish();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -53,7 +53,6 @@ public class StorageModuleElasticsearchConfig extends ModuleConfig {
|
|||
private int superDatasetIndexShardsFactor = 5;
|
||||
private int indexRefreshInterval = 2;
|
||||
private int bulkActions = 2000;
|
||||
private int syncBulkActions = 50000;
|
||||
private int flushInterval = 10;
|
||||
private int concurrentRequests = 2;
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -167,7 +167,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
|
|||
);
|
||||
this.registerServiceImplementation(
|
||||
IBatchDAO.class,
|
||||
new BatchProcessEsDAO(elasticSearchClient, config.getBulkActions(), config.getSyncBulkActions(), config
|
||||
new BatchProcessEsDAO(elasticSearchClient, config.getBulkActions(), config
|
||||
.getFlushInterval(), config.getConcurrentRequests())
|
||||
);
|
||||
this.registerServiceImplementation(StorageDAO.class, new StorageEsDAO(elasticSearchClient));
|
||||
|
|
|
|||
|
|
@ -19,8 +19,6 @@
|
|||
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import com.google.common.collect.Lists;
|
||||
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;
|
||||
|
|
@ -39,14 +37,15 @@ public class BatchProcessEsDAO extends EsDAO implements IBatchDAO {
|
|||
|
||||
private BulkProcessor bulkProcessor;
|
||||
private final int bulkActions;
|
||||
private final int syncBulkActions;
|
||||
private final int flushInterval;
|
||||
private final int concurrentRequests;
|
||||
|
||||
public BatchProcessEsDAO(ElasticSearchClient client, int bulkActions, int syncBulkActions, int flushInterval, int concurrentRequests) {
|
||||
public BatchProcessEsDAO(ElasticSearchClient client,
|
||||
int bulkActions,
|
||||
int flushInterval,
|
||||
int concurrentRequests) {
|
||||
super(client);
|
||||
this.bulkActions = bulkActions;
|
||||
this.syncBulkActions = syncBulkActions;
|
||||
this.flushInterval = flushInterval;
|
||||
this.concurrentRequests = concurrentRequests;
|
||||
}
|
||||
|
|
@ -63,20 +62,15 @@ public class BatchProcessEsDAO extends EsDAO implements IBatchDAO {
|
|||
@Override
|
||||
public void synchronous(List<PrepareRequest> prepareRequests) {
|
||||
if (CollectionUtils.isNotEmpty(prepareRequests)) {
|
||||
List<List<PrepareRequest>> partitions = Lists.partition(prepareRequests, syncBulkActions);
|
||||
|
||||
for (List<PrepareRequest> partition : partitions) {
|
||||
BulkRequest request = new BulkRequest();
|
||||
|
||||
for (PrepareRequest prepareRequest : partition) {
|
||||
if (prepareRequest instanceof InsertRequest) {
|
||||
request.add((IndexRequest) prepareRequest);
|
||||
} else {
|
||||
request.add((UpdateRequest) prepareRequest);
|
||||
}
|
||||
BulkRequest request = new BulkRequest();
|
||||
for (PrepareRequest prepareRequest : prepareRequests) {
|
||||
if (prepareRequest instanceof InsertRequest) {
|
||||
request.add((IndexRequest) prepareRequest);
|
||||
} else {
|
||||
request.add((UpdateRequest) prepareRequest);
|
||||
}
|
||||
getClient().synchronousBulk(request);
|
||||
}
|
||||
getClient().synchronousBulk(request);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -165,8 +165,8 @@ public class StorageModuleElasticsearch7Provider extends ModuleProvider {
|
|||
);
|
||||
this.registerServiceImplementation(
|
||||
IBatchDAO.class,
|
||||
new BatchProcessEsDAO(elasticSearch7Client, config.getBulkActions(), config.getSyncBulkActions(),
|
||||
config.getFlushInterval(), config.getConcurrentRequests()
|
||||
new BatchProcessEsDAO(
|
||||
elasticSearch7Client, config.getBulkActions(), config.getFlushInterval(), config.getConcurrentRequests()
|
||||
)
|
||||
);
|
||||
this.registerServiceImplementation(StorageDAO.class, new StorageEs7DAO(elasticSearch7Client));
|
||||
|
|
|
|||
Loading…
Reference in New Issue