diff --git a/CHANGES.md b/CHANGES.md index 154104f77..ce5160ba7 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -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 diff --git a/docs/en/setup/backend/backend-storage.md b/docs/en/setup/backend/backend-storage.md index e3f83a773..0d0d28227 100644 --- a/docs/en/setup/backend/backend-storage.md +++ b/docs/en/setup/backend/backend-storage.md @@ -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} diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md index 40955f667..7a91147b2 100644 --- a/docs/en/setup/backend/configuration-vocabulary.md +++ b/docs/en/setup/backend/configuration-vocabulary.md @@ -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| diff --git a/oap-server/server-bootstrap/src/main/resources/application.yml b/oap-server/server-bootstrap/src/main/resources/application.yml index a5d6f27ec..29c6754d4 100755 --- a/oap-server/server-bootstrap/src/main/resources/application.yml +++ b/oap-server/server-bootstrap/src/main/resources/application.yml @@ -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} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java index 22f546711..2a55fa458 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java @@ -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<>(); } 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 f35459cb1..441061668 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 @@ -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 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> partitions = Lists.partition(prepareRequests, maxSyncoperationNum); + CountDownLatch countDownLatch = new CountDownLatch(partitions.size()); + for (final List 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(); } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java index 286a592ee..126efc2fc 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java @@ -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; /** diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java index c8875a0cb..5b8019f45 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java @@ -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)); 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 2722b60c6..bcbe21b82 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,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 prepareRequests) { if (CollectionUtils.isNotEmpty(prepareRequests)) { - List> partitions = Lists.partition(prepareRequests, syncBulkActions); - - for (List 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); } } } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java index acc632772..b0eebef90 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java @@ -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));