Add support for user/password auth - Elasticsearch

presto-elasticsearch/pom.xml

@@ -94,12 +94,6 @@
             </exclusions>
         </dependency>
 
-        <dependency>
-            <groupId>org.elasticsearch</groupId>
-            <artifactId>elasticsearch</artifactId>
-            <version>${dep.elasticsearch.version}</version>
-        </dependency>
-
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-high-level-client</artifactId>
@@ -242,6 +236,53 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>net.java.dev.jna</groupId>
+            <artifactId>jna</artifactId>
+            <version>5.5.0</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>1.15.2</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-annotations</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.commons</groupId>
+                    <artifactId>commons-compress</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>com.facebook.presto</groupId>
+            <artifactId>presto-main</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>com.facebook.presto</groupId>
             <artifactId>presto-client</artifactId>
@@ -320,6 +361,18 @@
                  </exclusion>
             </exclusions>
         </dependency>
+
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>${dep.elasticsearch.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.elasticsearch</groupId>
+                    <artifactId>jna</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 
 </project>

presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchConfig.java

@@ -32,7 +32,8 @@ public class ElasticsearchConfig
 {
     public enum Security
     {
-        AWS
+        AWS,
+        PASSWORD,
     }
 
     private String host;

presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchConnectorModule.java

@@ -30,6 +30,7 @@
 import static com.facebook.airlift.json.JsonBinder.jsonBinder;
 import static com.facebook.presto.common.type.TypeSignature.parseTypeSignature;
 import static com.facebook.presto.elasticsearch.ElasticsearchConfig.Security.AWS;
+import static com.facebook.presto.elasticsearch.ElasticsearchConfig.Security.PASSWORD;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
 import static java.util.Objects.requireNonNull;
@@ -55,13 +56,21 @@ protected void setup(Binder binder)
         binder.install(new DecoderModule());
 
         newOptionalBinder(binder, AwsSecurityConfig.class);
+        newOptionalBinder(binder, PasswordConfig.class);
 
         install(installModuleIf(
                 ElasticsearchConfig.class,
                 config -> config.getSecurity()
                         .filter(isEqual(AWS))
                         .isPresent(),
                 conditionalBinder -> configBinder(conditionalBinder).bindConfig(AwsSecurityConfig.class)));
+
+        install(installModuleIf(
+                ElasticsearchConfig.class,
+                config -> config.getSecurity()
+                        .filter(isEqual(PASSWORD))
+                        .isPresent(),
+                conditionalBinder -> configBinder(conditionalBinder).bindConfig(PasswordConfig.class)));
     }
 
     private static final class TypeDeserializer

presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchMetadata.java

@@ -57,7 +57,6 @@
 import static com.facebook.presto.common.type.DoubleType.DOUBLE;
 import static com.facebook.presto.common.type.IntegerType.INTEGER;
 import static com.facebook.presto.common.type.RealType.REAL;
-import static com.facebook.presto.common.type.RowType.Field;
 import static com.facebook.presto.common.type.SmallintType.SMALLINT;
 import static com.facebook.presto.common.type.TimestampType.TIMESTAMP;
 import static com.facebook.presto.common.type.TinyintType.TINYINT;
@@ -66,7 +65,6 @@
 import static com.facebook.presto.elasticsearch.ElasticsearchTableHandle.Type.QUERY;
 import static com.facebook.presto.elasticsearch.ElasticsearchTableHandle.Type.SCAN;
 import static com.facebook.presto.spi.StandardErrorCode.INVALID_ARGUMENTS;
-import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static java.lang.String.format;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -320,11 +318,21 @@ else if (type instanceof DateTimeType) {
         else if (type instanceof ObjectType) {
             ObjectType objectType = (ObjectType) type;
 
-            List<Field> fields = objectType.getFields().stream()
-                    .map(field -> RowType.field(field.getName(), toPrestoType(field)))
-                    .collect(toImmutableList());
+            ImmutableList.Builder<RowType.Field> builder = ImmutableList.builder();
+            for (IndexMetadata.Field field : objectType.getFields()) {
+                Type prestoType = toPrestoType(field);
+                if (prestoType != null) {
+                    builder.add(RowType.field(field.getName(), prestoType));
+                }
+            }
+
+            List<RowType.Field> fields = builder.build();
+
+            if (!fields.isEmpty()) {
+                return RowType.from(fields);
+            }
 
-            return RowType.from(fields);
+            // otherwise, skip -- row types must have at least 1 field
         }
 
         return null;

presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/PasswordConfig.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.elasticsearch;
+
+import com.facebook.airlift.configuration.Config;
+import com.facebook.airlift.configuration.ConfigSecuritySensitive;
+
+import javax.validation.constraints.NotNull;
+
+public class PasswordConfig
+{
+    private String user;
+    private String password;
+
+    @NotNull
+    public String getUser()
+    {
+        return user;
+    }
+
+    @Config("elasticsearch.auth.user")
+    public PasswordConfig setUser(String user)
+    {
+        this.user = user;
+        return this;
+    }
+
+    @NotNull
+    public String getPassword()
+    {
+        return password;
+    }
+
+    @Config("elasticsearch.auth.password")
+    @ConfigSecuritySensitive
+    public PasswordConfig setPassword(String password)
+    {
+        this.password = password;
+        return this;
+    }
+}

presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/client/ElasticsearchClient.java

@@ -24,6 +24,7 @@
 import com.facebook.airlift.security.pem.PemReader;
 import com.facebook.presto.elasticsearch.AwsSecurityConfig;
 import com.facebook.presto.elasticsearch.ElasticsearchConfig;
+import com.facebook.presto.elasticsearch.PasswordConfig;
 import com.facebook.presto.spi.PrestoException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -35,10 +36,14 @@
 import io.airlift.units.Duration;
 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.client.config.RequestConfig;
 import org.apache.http.conn.ssl.NoopHostnameVerifier;
 import org.apache.http.entity.ByteArrayEntity;
 import org.apache.http.entity.StringEntity;
+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.message.BasicHeader;
@@ -129,9 +134,12 @@ public class ElasticsearchClient
     private final boolean ignorePublishAddress;
 
     @Inject
-    public ElasticsearchClient(ElasticsearchConfig config, Optional<AwsSecurityConfig> awsSecurityConfig)
+    public ElasticsearchClient(
+            ElasticsearchConfig config,
+            Optional<AwsSecurityConfig> awsSecurityConfig,
+            Optional<PasswordConfig> passwordConfig)
     {
-        client = createClient(config, awsSecurityConfig);
+        client = createClient(config, awsSecurityConfig, passwordConfig);
 
         this.ignorePublishAddress = config.isIgnorePublishAddress();
         this.scrollSize = config.getScrollSize();
@@ -183,7 +191,10 @@ private void refreshNodes()
         }
     }
 
-    private static RestHighLevelClient createClient(ElasticsearchConfig config, Optional<AwsSecurityConfig> awsSecurityConfig)
+    private static RestHighLevelClient createClient(
+            ElasticsearchConfig config,
+            Optional<AwsSecurityConfig> awsSecurityConfig,
+            Optional<PasswordConfig> passwordConfig)
     {
         RestClientBuilder builder = RestClient.builder(
                 new HttpHost(config.getHost(), config.getPort(), config.isTlsEnabled() ? "https" : "http"))
@@ -216,6 +227,12 @@ private static RestHighLevelClient createClient(ElasticsearchConfig config, Opti
                 }
             }
 
+            passwordConfig.ifPresent(securityConfig -> {
+                CredentialsProvider credentials = new BasicCredentialsProvider();
+                credentials.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(securityConfig.getUser(), securityConfig.getPassword()));
+                clientBuilder.setDefaultCredentialsProvider(credentials);
+            });
+
             awsSecurityConfig.ifPresent(securityConfig -> clientBuilder.addInterceptorLast(new AwsRequestSigner(
                     securityConfig.getRegion(),
                     getAwsCredentialsProvider(securityConfig))));

presto-elasticsearch/src/test/java/com/facebook/presto/elasticsearch/ElasticsearchLoader.java

@@ -24,7 +24,8 @@
 import com.facebook.presto.tests.ResultsSession;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.client.Client;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 
 import java.io.IOException;
@@ -42,25 +43,24 @@
 import static com.facebook.presto.common.type.Varchars.isVarcharType;
 import static com.google.common.base.Preconditions.checkState;
 import static java.util.Objects.requireNonNull;
-import static org.elasticsearch.client.Requests.flushRequest;
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 
 public class ElasticsearchLoader
         extends AbstractTestingPrestoClient<Void>
 {
     private final String tableName;
-    private final Client client;
+    private final RestHighLevelClient restClient;
 
     public ElasticsearchLoader(
-            Client client,
+            RestHighLevelClient client,
             String tableName,
             TestingPrestoServer prestoServer,
             Session defaultSession)
     {
         super(prestoServer, defaultSession);
 
         this.tableName = requireNonNull(tableName, "tableName is null");
-        this.client = requireNonNull(client, "client is null");
+        this.restClient = requireNonNull(client, "client is null");
     }
 
     @Override
@@ -108,8 +108,13 @@ public void addResults(QueryStatusInfo statusInfo, QueryData data)
                     throw new UncheckedIOException("Error loading data into Elasticsearch index: " + tableName, e);
                 }
             }
-            client.bulk(request).actionGet();
-            client.admin().indices().flush(flushRequest(tableName)).actionGet();
+            request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+            try {
+                restClient.bulk(request);
+            }
+            catch (IOException e) {
+                throw new RuntimeException(e);
+            }
         }
 
         @Override