Added elasticsearch warehouse certification support (#2384)

* Added elasticsearch warehouse certification support

* Add the elasticsearch service for Http Basic instructions

* Update backend-storage.md

Change document.

docker/config/application.yml

@@ -57,6 +57,8 @@ core:
 storage:
   elasticsearch:
   # set the namespace in elasticsearch
+    user: ${SW_ES_USER:""}
+    password: ${SW_ES_PASSWORD:""}
     clusterNodes: ${SW_STORAGE_ES_CLUSTER_NODES:elasticsearch:9200}
     indexShardsNumber: ${SW_STORAGE_ES_INDEX_SHARDS_NUMBER:2}
     indexReplicasNumber: ${SW_STORAGE_ES_INDEX_REPLICAS_NUMBER:0}

docs/en/setup/backend/backend-storage.md

@@ -37,6 +37,8 @@ Setting fragment example
 storage:
   elasticsearch:
     # nameSpace: ${SW_NAMESPACE:""}
+    # user: ${SW_ES_USER:""} # User needs to be set when Http Basic authentication is enabled
+    # password: ${SW_ES_PASSWORD:""} # Password to be set when Http Basic authentication is enabled
     clusterNodes: ${SW_STORAGE_ES_CLUSTER_NODES:localhost:9200}
     indexShardsNumber: ${SW_STORAGE_ES_INDEX_SHARDS_NUMBER:2}
     indexReplicasNumber: ${SW_STORAGE_ES_INDEX_REPLICAS_NUMBER:0}
@@ -46,9 +48,15 @@ storage:
     flushInterval: ${SW_STORAGE_ES_FLUSH_INTERVAL:10} # flush the bulk every 10 seconds whatever the number of requests
     concurrentRequests: ${SW_STORAGE_ES_CONCURRENT_REQUESTS:2} # the number of concurrent requests
 ```
+
 ### About Namespace
 When namespace is set, names of all indexes in ElasticSearch will use it as prefix.
 
+### About Authentication
+We only support [basic authentication](https://www.elastic.co/guide/en/elasticsearch/client/java-rest/6.6/_basic_authentication.html). If you need that, you could set `user` and `password`.
+For how to enable http basic authentication, you could read this https://brudtkuhl.com/blog/securing-elasticsearch/
+
+
 ## MySQL
 Active MySQL as storage, set storage provider to **mysql**. 
 

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java

@@ -21,7 +21,12 @@ package org.apache.skywalking.oap.server.library.client.elasticsearch;
 import org.apache.commons.lang3.StringUtils;
 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.entity.ContentType;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
 import org.apache.http.nio.entity.NStringEntity;
 import org.apache.skywalking.oap.server.library.client.Client;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
@@ -44,6 +49,7 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
@@ -70,18 +76,35 @@ public class ElasticSearchClient implements Client {
     private static final String TYPE = "type";
     private final String clusterNodes;
     private final String namespace;
+    private final String user;
+    private final String password;
     private RestHighLevelClient client;
 
-    public ElasticSearchClient(String clusterNodes, String namespace) {
+    public ElasticSearchClient(String clusterNodes, String namespace, String user, String password) {
         this.clusterNodes = clusterNodes;
         this.namespace = namespace;
+        this.user = user;
+        this.password = password;
     }
 
     @Override public void connect() {
         List<HttpHost> pairsList = parseClusterNodes(clusterNodes);
-
-        client = new RestHighLevelClient(
-            RestClient.builder(pairsList.toArray(new HttpHost[0])));
+        RestClientBuilder builder;
+        if (StringUtils.isNotBlank(user) && StringUtils.isNotBlank(password)) {
+            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
+            builder = RestClient.builder(pairsList.toArray(new HttpHost[0]))
+                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
+                        @Override
+                        public HttpAsyncClientBuilder customizeHttpClient(
+                                HttpAsyncClientBuilder httpClientBuilder) {
+                            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+                        }
+                    });
+        } else {
+            builder = RestClient.builder(pairsList.toArray(new HttpHost[0]));
+        }
+        client = new RestHighLevelClient(builder);
     }
 
     @Override public void shutdown() {

oap-server/server-library/library-client/src/test/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClientTestCase.java

@@ -47,7 +47,7 @@ public class ElasticSearchClientTestCase {
             .endObject();
         builder.endObject();
 
-        ElasticSearchClient client = new ElasticSearchClient("localhost:9200", null);
+        ElasticSearchClient client = new ElasticSearchClient("localhost:9200", null, null, null);
         client.connect();
 
         String indexName = "test";

oap-server/server-starter/src/main/assembly/application.yml

@@ -61,6 +61,8 @@ storage:
     user: ${SW_STORAGE_H2_USER:sa}
 #  elasticsearch:
 #    # nameSpace: ${SW_NAMESPACE:""}
+#    # user: ${SW_ES_USER:""}
+#    # password: ${SW_ES_PASSWORD:""}
 #    clusterNodes: ${SW_STORAGE_ES_CLUSTER_NODES:localhost:9200}
 #    indexShardsNumber: ${SW_STORAGE_ES_INDEX_SHARDS_NUMBER:2}
 #    indexReplicasNumber: ${SW_STORAGE_ES_INDEX_REPLICAS_NUMBER:0}

oap-server/server-starter/src/main/resources/application.yml

@@ -58,6 +58,8 @@ storage:
   elasticsearch:
     nameSpace: ${SW_NAMESPACE:""}
     clusterNodes: ${SW_STORAGE_ES_CLUSTER_NODES:localhost:9200}
+    user: ${SW_ES_USER:""}
+    password: ${SW_ES_PASSWORD:""}
     indexShardsNumber: ${SW_STORAGE_ES_INDEX_SHARDS_NUMBER:2}
     indexReplicasNumber: ${SW_STORAGE_ES_INDEX_REPLICAS_NUMBER:0}
     # Batch process setting, refer to https://www.elastic.co/guide/en/elasticsearch/client/java-api/5.5/java-docs-bulk-processor.html

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java

@@ -40,6 +40,24 @@ public class StorageModuleElasticsearchConfig extends ModuleConfig {
     private int bulkSize = 20;
     private int flushInterval = 10;
     private int concurrentRequests = 2;
+    private String user;
+    private String password;
+
+    public String getUser() {
+        return user;
+    }
+
+    public void setUser(String user) {
+        this.user = user;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
 
     int getIndexShardsNumber() {
         return indexShardsNumber;

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java

@@ -98,7 +98,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
         if (!StringUtil.isEmpty(config.getNameSpace())) {
             config.setNameSpace(config.getNameSpace().toLowerCase());
         }
-        elasticSearchClient = new ElasticSearchClient(config.getClusterNodes(), config.getNameSpace());
+        elasticSearchClient = new ElasticSearchClient(config.getClusterNodes(), config.getNameSpace(), config.getUser(), config.getPassword());
 
         this.registerServiceImplementation(IBatchDAO.class, new BatchProcessEsDAO(elasticSearchClient, config.getBulkActions(), config.getBulkSize(), config.getFlushInterval(), config.getConcurrentRequests()));
         this.registerServiceImplementation(StorageDAO.class, new StorageEsDAO(elasticSearchClient));