Expose responseTimeout of ES client (#8531)

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
 

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 | - |

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<ElasticSearch> es = new AtomicReference<>();
@@ -94,6 +96,7 @@ public class ElasticSearchClient implements Client, HealthCheckable {
                                Function<String, String> 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 -> {

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();
     }

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<Boolean> healthyListener) {
+                  Consumer<Boolean> 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())

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
         );
     }
 }

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:""}

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.

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,