Supporting basic authentication for Elasticsearch connector

presto-docs/src/main/sphinx/connector/elasticsearch.rst

@@ -213,6 +213,19 @@ as part of the table name, separated by a colon. For example:
 .. _full text query: https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html#query-string-syntax
 
 
+X-Pack Authentication
+---------------------
+
+To enable X-Pack authentication, the ``elasticsearch.security`` option needs to be set to ``X_PACK``.
+Additionally, the following options need to be configured appropriately:
+
+================================================ ==================================================================
+Property Name                                    Description
+================================================ ==================================================================
+``elasticsearch.username``                       X-Pack username for connecting to the Elasticsearch. This option is required.
+``elasticsearch.password``                       X-Pack password for connecting to the Elasticsearch. This option is required
+================================================ ==================================================================
+
 AWS Authorization
 -----------------
 
@@ -227,4 +240,3 @@ Property Name                                    Description
 ``elasticsearch.aws.secret-key``                 AWS secret key to use to connect to the Elasticsearch domain.
 ``elasticsearch.aws.use-instance-credentials``   Use the EC2 metadata service to retrieve API credentials.
 ================================================ ==================================================================
-

presto-elasticsearch/src/main/java/io/prestosql/elasticsearch/ElasticsearchConfig.java

@@ -48,11 +48,14 @@ public class ElasticsearchConfig
 {
     public enum Security
     {
-        AWS
+        AWS,
+        X_PACK
     }
 
     private String host;
     private int port = 9200;
+    private String username;
+    private String password;
     private String defaultSchema = "default";
     private int scrollSize = 1_000;
     private Duration scrollTimeout = new Duration(1, MINUTES);
@@ -95,6 +98,33 @@ public ElasticsearchConfig setPort(int port)
         return this;
     }
 
+    public String getUsername()
+    {
+        return username;
+    }
+
+    @Config("elasticsearch.username")
+    @ConfigDescription("Username for connecting to X-Pack secured Elasticsearch")
+    public ElasticsearchConfig setUsername(String username)
+    {
+        this.username = username;
+        return this;
+    }
+
+    public String getPassword()
+    {
+        return password;
+    }
+
+    @Config("elasticsearch.password")
+    @ConfigSecuritySensitive
+    @ConfigDescription("Password for connecting to X-Pack secured Elasticsearch")
+    public ElasticsearchConfig setPassword(String password)
+    {
+        this.password = password;
+        return this;
+    }
+
     @NotNull
     public String getDefaultSchema()
     {

presto-elasticsearch/src/main/java/io/prestosql/elasticsearch/client/ElasticsearchClient.java

@@ -32,7 +32,11 @@
 import io.prestosql.spi.PrestoException;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
 import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.action.search.ClearScrollRequest;
@@ -86,6 +90,7 @@
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
 import static io.airlift.json.JsonCodec.jsonCodec;
+import static io.prestosql.elasticsearch.ElasticsearchConfig.Security.X_PACK;
 import static io.prestosql.elasticsearch.ElasticsearchErrorCode.ELASTICSEARCH_CONNECTION_ERROR;
 import static io.prestosql.elasticsearch.ElasticsearchErrorCode.ELASTICSEARCH_INVALID_RESPONSE;
 import static io.prestosql.elasticsearch.ElasticsearchErrorCode.ELASTICSEARCH_QUERY_FAILURE;
@@ -184,6 +189,15 @@ private static RestHighLevelClient createClient(ElasticsearchConfig config, Opti
                 .setMaxRetryTimeoutMillis((int) config.getMaxRetryTime().toMillis());
 
         builder.setHttpClientConfigCallback(clientBuilder -> {
+            if (config.getSecurity().isPresent()) {
+                if (config.getSecurity().get().equals(X_PACK)) {
+                    CredentialsProvider xPackCredentialsProvider = getXPackCredentialsProvider(config);
+                    if (xPackCredentialsProvider != null) {
+                        clientBuilder.setDefaultCredentialsProvider(xPackCredentialsProvider);
+                    }
+                }
+            }
+
             if (config.isTlsEnabled()) {
                 buildSslContext(config.getKeystorePath(), config.getKeystorePassword(), config.getTrustStorePath(), config.getTruststorePassword())
                         .ifPresent(clientBuilder::setSSLContext);
@@ -203,6 +217,17 @@ private static RestHighLevelClient createClient(ElasticsearchConfig config, Opti
         return new RestHighLevelClient(builder);
     }
 
+    private static CredentialsProvider getXPackCredentialsProvider(ElasticsearchConfig config)
+    {
+        if (config.getUsername() != null && config.getPassword() != null) {
+            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+            credentialsProvider.setCredentials(AuthScope.ANY,
+                    new UsernamePasswordCredentials(config.getUsername(), config.getPassword()));
+            return credentialsProvider;
+        }
+        return null;
+    }
+
     private static AWSCredentialsProvider getAwsCredentialsProvider(AwsSecurityConfig config)
     {
         if (config.getAccessKey().isPresent() && config.getSecretKey().isPresent()) {

presto-elasticsearch/src/test/java/io/prestosql/elasticsearch/TestElasticsearchConfig.java

@@ -35,6 +35,8 @@ public void testDefaults()
         assertRecordedDefaults(recordDefaults(ElasticsearchConfig.class)
                 .setHost(null)
                 .setPort(9200)
+                .setUsername(null)
+                .setPassword(null)
                 .setDefaultSchema("default")
                 .setScrollSize(1000)
                 .setScrollTimeout(new Duration(1, MINUTES))
@@ -71,6 +73,8 @@ public void testExplicitPropertyMappings()
                 .put("elasticsearch.tls.truststore-password", "truststore-password")
                 .put("elasticsearch.tls.verify-hostnames", "false")
                 .put("elasticsearch.security", "AWS")
+                .put("elasticsearch.username", "username")
+                .put("elasticsearch.password", "password")
                 .build();
 
         ElasticsearchConfig expected = new ElasticsearchConfig()
@@ -89,7 +93,9 @@ public void testExplicitPropertyMappings()
                 .setTrustStorePath(new File("/tmp/truststore"))
                 .setTruststorePassword("truststore-password")
                 .setVerifyHostnames(false)
-                .setSecurity(AWS);
+                .setSecurity(AWS)
+                .setUsername("username")
+                .setPassword("password");
 
         assertFullMapping(properties, expected);
     }