From c49a82a6f5fed3bcab8f25e3a0490c5c9657ddba Mon Sep 17 00:00:00 2001 From: kezhenxu94 Date: Thu, 10 Feb 2022 14:46:40 +0800 Subject: [PATCH] Expose responseTimeout of ES client (#8531) --- CHANGES.md | 1 + docs/en/setup/backend/configuration-vocabulary.md | 1 + .../client/elasticsearch/ElasticSearchClient.java | 5 +++++ .../library/elasticsearch/bulk/ITElasticSearch.java | 2 +- .../library/elasticsearch/ElasticSearch.java | 5 ++++- .../library/elasticsearch/ElasticSearchBuilder.java | 11 ++++++++++- .../server-starter/src/main/resources/application.yml | 1 + .../StorageModuleElasticsearchConfig.java | 5 +++++ .../StorageModuleElasticsearchProvider.java | 3 ++- 9 files changed, 30 insertions(+), 4 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index e72d10c120..595bd51ea9 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -54,6 +54,7 @@ Release Notes. * Bump up PostgreSQL driver to fix CVE. * Add Guava EventBus component ID(123) of Java agent. * Add OpenFunction component ID(5013). +* Expose configuration `responseTimeout` of ES client. #### UI diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md index 518ffe5dfe..185ec66d56 100644 --- a/docs/en/setup/backend/configuration-vocabulary.md +++ b/docs/en/setup/backend/configuration-vocabulary.md @@ -88,6 +88,7 @@ core|default|role|Option values: `Mixed/Receiver/Aggregator`. **Receiver** mode | - | - | protocol | HTTP or HTTPs. | SW_STORAGE_ES_HTTP_PROTOCOL | HTTP| | - | - | connectTimeout | Connect timeout of ElasticSearch client (in milliseconds). | SW_STORAGE_ES_CONNECT_TIMEOUT | 500| | - | - | socketTimeout | Socket timeout of ElasticSearch client (in milliseconds). | SW_STORAGE_ES_SOCKET_TIMEOUT | 30000| +| - | - | responseTimeout | Response timeout of ElasticSearch client (in milliseconds), `0` disables the timeout. | SW_STORAGE_ES_RESPONSE_TIMEOUT | 1500 | | - | - | numHttpClientThread | The number of threads for the underlying HTTP client to perform socket I/O. If the value is <= 0, the number of available processors will be used. | SW_STORAGE_ES_NUM_HTTP_CLIENT_THREAD | 0 | | - | - | user| Username of ElasticSearch cluster. | SW_ES_USER | - | | - | - | password | Password of ElasticSearch cluster. | SW_ES_PASSWORD | - | 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 2a457f4172..05b29dac8f 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 @@ -81,6 +81,8 @@ public class ElasticSearchClient implements Client, HealthCheckable { private final int socketTimeout; + private final int responseTimeout; + private final int numHttpClientThread; private final AtomicReference es = new AtomicReference<>(); @@ -94,6 +96,7 @@ public class ElasticSearchClient implements Client, HealthCheckable { Function indexNameConverter, int connectTimeout, int socketTimeout, + int responseTimeout, int numHttpClientThread) { this.clusterNodes = clusterNodes; this.protocol = protocol; @@ -104,6 +107,7 @@ public class ElasticSearchClient implements Client, HealthCheckable { this.indexNameConverter = indexNameConverter; this.connectTimeout = connectTimeout; this.socketTimeout = socketTimeout; + this.responseTimeout = responseTimeout; this.numHttpClientThread = numHttpClientThread; } @@ -117,6 +121,7 @@ public class ElasticSearchClient implements Client, HealthCheckable { .endpoints(clusterNodes.split(",")) .protocol(protocol) .connectTimeout(connectTimeout) + .responseTimeout(responseTimeout) .socketTimeout(socketTimeout) .numHttpClientThread(numHttpClientThread) .healthyListener(healthy -> { diff --git a/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ITElasticSearch.java b/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ITElasticSearch.java index b7b103d3cd..6a6980a626 100644 --- a/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ITElasticSearch.java +++ b/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ITElasticSearch.java @@ -80,7 +80,7 @@ public class ITElasticSearch { server.getHttpHostAddress(), "http", "", "", "test", "test", indexNameConverter(namespace), 500, 6000, - 0 + 0, 15 ); client.connect(); } diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java index 9136a09163..e39e53afb8 100644 --- a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java +++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java @@ -36,6 +36,7 @@ import com.linecorp.armeria.common.util.Exceptions; import java.io.Closeable; import java.io.IOException; import java.io.InputStream; +import java.time.Duration; import java.util.Collections; import java.util.List; import java.util.Map; @@ -81,7 +82,8 @@ public final class ElasticSearch implements Closeable { String username, String password, EndpointGroup endpointGroup, ClientFactory clientFactory, - Consumer healthyListener) { + Consumer healthyListener, + Duration responseTimeout) { this.endpointGroup = endpointGroup; this.clientFactory = clientFactory; if (healthyListener != null) { @@ -94,6 +96,7 @@ public final class ElasticSearch implements Closeable { final WebClientBuilder builder = WebClient.builder(protocol, endpointGroup) .factory(clientFactory) + .responseTimeout(responseTimeout) .decorator(LoggingClient.builder() .logger(log) .newDecorator()) diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearchBuilder.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearchBuilder.java index 13df9275fd..39bbc66b79 100644 --- a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearchBuilder.java +++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearchBuilder.java @@ -60,6 +60,8 @@ public final class ElasticSearchBuilder { private String trustStorePass; + private Duration responseTimeout = Duration.ofSeconds(15); + private Duration connectTimeout = Duration.ofMillis(500); private Duration socketTimeout = Duration.ofSeconds(30); @@ -118,6 +120,12 @@ public final class ElasticSearchBuilder { return this; } + public ElasticSearchBuilder responseTimeout(int responseTimeout) { + checkArgument(responseTimeout >= 0, "responseTimeout must be 0 or positive"); + this.responseTimeout = Duration.ofMillis(responseTimeout); + return this; + } + public ElasticSearchBuilder socketTimeout(int socketTimeout) { checkArgument(socketTimeout > 0, "socketTimeout must be positive"); this.socketTimeout = Duration.ofMillis(socketTimeout); @@ -188,7 +196,8 @@ public final class ElasticSearchBuilder { password, endpointGroup, clientFactory, - healthyListener + healthyListener, + responseTimeout ); } } diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml index dcbcd35d27..ff3a62fd4f 100755 --- a/oap-server/server-starter/src/main/resources/application.yml +++ b/oap-server/server-starter/src/main/resources/application.yml @@ -135,6 +135,7 @@ storage: protocol: ${SW_STORAGE_ES_HTTP_PROTOCOL:"http"} connectTimeout: ${SW_STORAGE_ES_CONNECT_TIMEOUT:500} socketTimeout: ${SW_STORAGE_ES_SOCKET_TIMEOUT:30000} + responseTimeout: ${SW_STORAGE_ES_RESPONSE_TIMEOUT:15000} numHttpClientThread: ${SW_STORAGE_ES_NUM_HTTP_CLIENT_THREAD:0} user: ${SW_ES_USER:""} password: ${SW_ES_PASSWORD:""} 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 aecf642f44..67e6af1ec2 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 @@ -41,6 +41,11 @@ public class StorageModuleElasticsearchConfig extends ModuleConfig { * @since 8.7.0 */ private int socketTimeout = 30000; + /** + * @since 9.0.0 the response timeout of ElasticSearch client (Armeria under the hood), set to 0 to disable response + * timeout. + */ + private int responseTimeout = 15000; /** * @since 6.4.0, the index of metrics and traces data in minute/hour/month precision are organized in days. ES * storage creates new indexes in every day. 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 04fe917623..d956245373 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 @@ -156,7 +156,8 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider { config.getClusterNodes(), config.getProtocol(), config.getTrustStorePath(), config .getTrustStorePass(), config.getUser(), config.getPassword(), indexNameConverter(config.getNamespace()), config.getConnectTimeout(), - config.getSocketTimeout(), config.getNumHttpClientThread() + config.getSocketTimeout(), config.getNumHttpClientThread(), + config.getResponseTimeout() ); this.registerServiceImplementation( IBatchDAO.class,