Merge pull request #46 from tokenmill/handle-auth-for-elasticsearch

Handle auth for elasticsearch

Author: zaibacu

README.md

@@ -28,6 +28,10 @@ Framework writes its configuration and stores crawled data to ElasticSearch. Bef
 
 Crawling Framework is a Java lib which will have to be extended to run Storm Crawler topology, thus Java (JDK8, Maven) infrastructure will be needed. 
 
+### Using password protected ElasticSearch
+
+Some providers hide ElasticSearch under authentification step (Which makes sense). Just set environment variables `ES_USERNAME` and `ES_PASSWORD` accordingly, everything else can remain the same. Authentification step will be done implicitly if proper credentials are there
+
 ## Configuring and Running a crawl
 
 See [Crawling Framework Example](https://github.com/tokenmill/crawling-framework-example) project's documentation.

elasticsearch/src/main/java/lt/tokenmill/crawling/es/ElasticConnection.java

@@ -1,6 +1,10 @@
 package lt.tokenmill.crawling.es;
 
 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.impl.client.BasicCredentialsProvider;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkProcessor;
@@ -38,6 +42,30 @@ private ElasticConnection(BulkProcessor processor, RestHighLevelClient restHighL
         this.restClientBuilder = restClient;
     }
 
+    private static class ESCredentials{
+        private final String username;
+        private final String password;
+        private final CredentialsProvider credentialsProvider;
+        public ESCredentials(){
+            this.username = System.getenv("ES_USERNAME");
+            this.password = System.getenv("ES_PASSWORD");
+            this.credentialsProvider = new BasicCredentialsProvider();
+        }
+
+        public boolean hasCredentials(){
+            return this.username != null && this.password != null;
+        }
+
+        public CredentialsProvider getCredentials(){
+            this.credentialsProvider.setCredentials(
+                    AuthScope.ANY,
+                    new UsernamePasswordCredentials(this.username, this.password)
+            );
+
+            return this.credentialsProvider;
+        }
+    }
+
     public static Builder builder() {
         return new Builder();
     }
@@ -109,8 +137,13 @@ private static ElasticConnection getConnection(String hostname, int restPort, St
         System.setProperty("es.set.netty.runtime.available.processors", "false");
 
         TimeValue flushInterval = TimeValue.parseTimeValue(flushIntervalString, TimeValue.timeValueSeconds(5), "flush");
-
+        ESCredentials credentials = new ESCredentials();
         RestClientBuilder restClient = RestClient.builder(new HttpHost(hostname, restPort, restScheme));
+        if(credentials.hasCredentials()) {
+            LOG.info("Found credentials. Applying");
+            restClient.setHttpClientConfigCallback(b -> b.setDefaultCredentialsProvider(credentials.getCredentials()));
+        }
+
         RestHighLevelClient restHighLevelClient = new RestHighLevelClient(restClient);
 
         BulkProcessor bulkProcessor = BulkProcessor.builder(restHighLevelClient::bulkAsync, listener)