fixup! Add an Elasticsearch low level client extension

quarkusio · Jul 21, 2020 · 3484146 · 3484146
1 parent fe025ad
commit 3484146
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 9 deletions.
diff --git a/...c/main/java/io/quarkus/elasticsearch/restclient/lowlevel/runtime/ElasticsearchConfig.java b/...c/main/java/io/quarkus/elasticsearch/restclient/lowlevel/runtime/ElasticsearchConfig.java
@@ -49,6 +49,18 @@ public class ElasticsearchConfig {
     @ConfigItem(defaultValue = "30S")
     public Duration socketTimeout;
 
+    /**
+     * The maximum number of connections to all the Elasticsearch servers.
+     */
+    @ConfigItem(defaultValue = "20")
+    public int maxConnections;
+
+    /**
+     * The maximum number of connections per Elasticsearch server.
+     */
+    @ConfigItem(defaultValue = "10")
+    public int maxConnectionsPerRoute;
+
     /**
      * The number of IO thread.
      * By default, this is the number of locally detected processors.

diff --git a/...in/java/io/quarkus/elasticsearch/restclient/lowlevel/runtime/RestClientBuilderHelper.java b/...in/java/io/quarkus/elasticsearch/restclient/lowlevel/runtime/RestClientBuilderHelper.java
@@ -11,12 +11,19 @@
 import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
 import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.sniff.ElasticsearchNodesSniffer;
+import org.elasticsearch.client.sniff.NodesSniffer;
 import org.elasticsearch.client.sniff.Sniffer;
+import org.elasticsearch.client.sniff.SnifferBuilder;
+import org.jboss.logging.Logger;
 
 public final class RestClientBuilderHelper {
 
+    private static final Logger LOG = Logger.getLogger("io.quarkus");
+
     private RestClientBuilderHelper() {
         // avoid instantiation
     }
@@ -31,17 +38,22 @@ static RestClientBuilder createRestClientBuilder(ElasticsearchConfig config) {
             public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                 return requestConfigBuilder
                         .setConnectTimeout((int) config.connectionTimeout.toMillis())
-                        .setSocketTimeout((int) config.socketTimeout.toMillis());
+                        .setSocketTimeout((int) config.socketTimeout.toMillis())
+                        .setConnectionRequestTimeout(0); // Avoid requests being flagged as timed out even when they didn't time out.
             }
         });
 
         builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
             @Override
             public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                 if (config.username.isPresent()) {
+                    if ("https".equalsIgnoreCase(config.protocol)) {
+                        LOG.warn("Using Basic authentication in HTTP implies sending plain text passwords over the wire, " +
+                                "use HTTPS protocol instead");
+                    }
                     CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                     credentialsProvider.setCredentials(AuthScope.ANY,
-                            new UsernamePasswordCredentials(config.username.get(), config.password.get()));
+                            new UsernamePasswordCredentials(config.username.get(), config.password.orElse(null)));
                     httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                 }
 
@@ -52,6 +64,15 @@ public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpCli
                     httpClientBuilder.setDefaultIOReactorConfig(ioReactorConfig);
                 }
 
+                httpClientBuilder.setMaxConnTotal(config.maxConnections);
+                httpClientBuilder.setMaxConnPerRoute(config.maxConnectionsPerRoute);
+
+                if ("http".equalsIgnoreCase(config.protocol)) {
+                    // In this case disable the SSL capability as it might have an impact on
+                    // bootstrap time, for example consuming entropy for no reason
+                    httpClientBuilder.setSSLStrategy(NoopIOSessionStrategy.INSTANCE);
+                }
+
                 return httpClientBuilder;
             }
         });
@@ -60,8 +81,18 @@ public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpCli
     }
 
     static Sniffer createSniffer(RestClient client, ElasticsearchConfig config) {
-        return Sniffer.builder(client)
-                .setSniffIntervalMillis((int) config.discovery.refreshInterval.toMillis())
-                .build();
+        SnifferBuilder builder = Sniffer.builder(client)
+                .setSniffIntervalMillis((int) config.discovery.refreshInterval.toMillis());
+
+        // https discovery support
+        if ("https".equalsIgnoreCase(config.protocol)) {
+            NodesSniffer hostsSniffer = new ElasticsearchNodesSniffer(
+                    client,
+                    ElasticsearchNodesSniffer.DEFAULT_SNIFF_REQUEST_TIMEOUT, // 1sec
+                    ElasticsearchNodesSniffer.Scheme.HTTPS);
+            builder.setNodesSniffer(hostsSniffer);
+        }
+
+        return builder.build();
     }
 }
diff --git a/...sts/elasticsearch-rest-client/src/main/java/io/quarkus/it/elasticsearch/FruitService.java b/...sts/elasticsearch-rest-client/src/main/java/io/quarkus/it/elasticsearch/FruitService.java
@@ -50,8 +50,11 @@ private List<Fruit> search(String term, String match) throws IOException {
         Request request = new Request(
                 "GET",
                 "/fruits/_search");
-        String query = String.format("{\"query\": { \"match\": { \"%s\":\"%s\"}}}", term, match);
-        request.setJsonEntity(query);
+        //construct a JSON query like {"query": {"match": {"<term>": "<match"}}
+        JsonObject termJson = new JsonObject().put(term, match);
+        JsonObject matchJson = new JsonObject().put("match", termJson);
+        JsonObject queryJson = new JsonObject().put("query", matchJson);
+        request.setJsonEntity(queryJson.encode());
         Response response = restClient.performRequest(request);
         String responseBody = EntityUtils.toString(response.getEntity());
 

diff --git a/integration-tests/elasticsearch-rest-client/src/main/resources/application.properties b/integration-tests/elasticsearch-rest-client/src/main/resources/application.properties
@@ -1,2 +1,4 @@
-# Configuration file
-# key = value
+quarkus.elasticsearch.hosts=localhost:9200
+quarkus.elasticsearch.socket-timeout=10S
+quarkus.elasticsearch.discovery.enabled=true
+quarkus.elasticsearch.discovery.refresh-interval=1M