Support custom elasticsearch settings in index level. (#3913)

docker/oap/docker-entrypoint.sh

@@ -305,6 +305,7 @@ core:
     # Cache metric data for 1 minute to reduce database queries, and if the OAP cluster changes within that minute,
     # the metrics may not be accurate within that minute.
     enableDatabaseSession: \${SW_CORE_ENABLE_DATABASE_SESSION:true}
+    topNReportPeriod: \${SW_CORE_TOPN_REPORT_PERIOD:10}
 EOT
 
     # generate storage

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

@@ -63,6 +63,7 @@ storage:
     resultWindowMaxSize: ${SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE:10000}
     metadataQueryMaxSize: ${SW_STORAGE_ES_QUERY_MAX_SIZE:5000}
     segmentQueryMaxSize: ${SW_STORAGE_ES_QUERY_SEGMENT_SIZE:200}
+    advanced: ${SW_STORAGE_ES_ADVANCED:""}
 ```
 
 and there're also some configurations that are ES7 specific, as follows:
@@ -103,12 +104,25 @@ storage:
     bulkSize: ${SW_STORAGE_ES_BULK_SIZE:20} # flush the bulk every 20mb
     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
+    advanced: ${SW_STORAGE_ES_ADVANCED:""}
 ```
 
 
 ### Data TTL
 TTL in ElasticSearch overrides the settings of core, read [ElasticSearch section in TTL document](ttl.md#elasticsearch-6-storage-ttl)
 
+### Advanced Configurations For Elasticsearch Index
+You can add advanced configurations in `JSON` format to set `ElasticSearch index settings` by following [ElasticSearch doc](https://www.elastic.co/guide/en/elasticsearch/reference/current/index-modules.html)
+
+For example, set [translog](https://www.elastic.co/guide/en/elasticsearch/reference/master/index-modules-translog.html) settings:
+
+```yaml
+storage:
+  elasticsearch:
+    # ......
+    advanced: ${SW_STORAGE_ES_ADVANCED:"{\"index.translog.durability\":\"request\",\"index.translog.sync_interval\":\"5s\"}"}
+```
+
 ### Recommended ElasticSearch server-side configurations
 You could add following config to `elasticsearch.yml`, set the value based on your env.
 

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

@@ -94,6 +94,7 @@ storage:
 #    resultWindowMaxSize: ${SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE:10000}
 #    metadataQueryMaxSize: ${SW_STORAGE_ES_QUERY_MAX_SIZE:5000}
 #    segmentQueryMaxSize: ${SW_STORAGE_ES_QUERY_SEGMENT_SIZE:200}
+#    advanced: ${SW_STORAGE_ES_ADVANCED:""}
   elasticsearch7:
     nameSpace: ${SW_NAMESPACE:""}
     clusterNodes: ${SW_STORAGE_ES_CLUSTER_NODES:localhost:9200}
@@ -119,6 +120,7 @@ storage:
     # instead, give more query criteria (e.g. service id or time range), to narrow the query results.
     # see https://www.elastic.co/guide/en/elasticsearch/guide/current/pagination.html for more information
     indexMaxResultWindow: ${SW_STORAGE_ES_INDEX_MAX_RESULT_WINDOW:5000}
+    advanced: ${SW_STORAGE_ES_ADVANCED:""}
 #  h2:
 #    driver: ${SW_STORAGE_H2_DRIVER:org.h2.jdbcx.JdbcDataSource}
 #    url: ${SW_STORAGE_H2_URL:jdbc:h2:mem:skywalking-oap-db}

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

@@ -50,6 +50,7 @@ public class StorageModuleElasticsearchConfig extends ModuleConfig {
     @Setter private int dayMetricsDataTTL = 2;
     private int otherMetricsDataTTL = 0;
     @Setter private int monthMetricsDataTTL = 18;
+    @Setter private String advanced;
 
     public int getMinuteMetricsDataTTL() {
         if (otherMetricsDataTTL > 0) {

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

@@ -132,7 +132,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
         try {
             elasticSearchClient.connect();
 
-            StorageEsInstaller installer = new StorageEsInstaller(getManager(), config.getIndexShardsNumber(), config.getIndexReplicasNumber(), config.getIndexRefreshInterval());
+            StorageEsInstaller installer = new StorageEsInstaller(getManager(), config);
             installer.install(elasticSearchClient);
 
             RegisterLockInstaller lockInstaller = new RegisterLockInstaller(elasticSearchClient);

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

@@ -18,6 +18,11 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
 
+import com.google.gson.Gson;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.skywalking.apm.util.StringUtil;
 import org.apache.skywalking.oap.server.core.storage.StorageException;
 import org.apache.skywalking.oap.server.core.storage.model.Model;
 import org.apache.skywalking.oap.server.core.storage.model.ModelColumn;
@@ -25,32 +30,26 @@ import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller;
 import org.apache.skywalking.oap.server.library.client.Client;
 import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
 import org.elasticsearch.common.unit.TimeValue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
 /**
- * @author peng-yongsheng
+ * @author peng-yongsheng, jian.tan
  */
 public class StorageEsInstaller extends ModelInstaller {
 
     private static final Logger logger = LoggerFactory.getLogger(StorageEsInstaller.class);
+    private final Gson gson = new Gson();
 
-    protected final int indexShardsNumber;
-    protected final int indexReplicasNumber;
-    protected final int indexRefreshInterval;
+    private final StorageModuleElasticsearchConfig config;
     protected final ColumnTypeEsMapping columnTypeEsMapping;
 
-    public StorageEsInstaller(ModuleManager moduleManager, int indexShardsNumber, int indexReplicasNumber, int indexRefreshInterval) {
+    public StorageEsInstaller(ModuleManager moduleManager, final StorageModuleElasticsearchConfig config) {
         super(moduleManager);
-        this.indexShardsNumber = indexShardsNumber;
-        this.indexReplicasNumber = indexReplicasNumber;
-        this.indexRefreshInterval = indexRefreshInterval;
         this.columnTypeEsMapping = new ColumnTypeEsMapping();
+        this.config = config;
     }
 
     @Override protected boolean isExists(Client client, Model model) throws StorageException {
@@ -104,10 +103,14 @@ public class StorageEsInstaller extends ModelInstaller {
 
     protected Map<String, Object> createSetting(boolean record) {
         Map<String, Object> setting = new HashMap<>();
-        setting.put("index.number_of_shards", indexShardsNumber);
-        setting.put("index.number_of_replicas", indexReplicasNumber);
-        setting.put("index.refresh_interval", record ? TimeValue.timeValueSeconds(10).toString() : TimeValue.timeValueSeconds(indexRefreshInterval).toString());
+        setting.put("index.number_of_shards", config.getIndexShardsNumber());
+        setting.put("index.number_of_replicas", config.getIndexReplicasNumber());
+        setting.put("index.refresh_interval", record ? TimeValue.timeValueSeconds(10).toString() : TimeValue.timeValueSeconds(config.getFlushInterval()).toString());
         setting.put("analysis.analyzer.oap_analyzer.type", "stop");
+        if (!StringUtil.isEmpty(config.getAdvanced())) {
+            Map<String, Object> advancedSettings = gson.fromJson(config.getAdvanced(), Map.class);
+            advancedSettings.forEach(setting::put);
+        }
         return setting;
     }
 

oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/base/StorageEs7Installer.java

@@ -28,7 +28,7 @@ import org.slf4j.LoggerFactory;
 import java.util.Map;
 
 /**
- * @author kezhenxu94
+ * @author kezhenxu94, jian.tan
  */
 public class StorageEs7Installer extends StorageEsInstaller {
 
@@ -38,8 +38,7 @@ public class StorageEs7Installer extends StorageEsInstaller {
 
     public StorageEs7Installer(final ModuleManager moduleManager,
                                final StorageModuleElasticsearch7Config config) {
-        super(moduleManager, config.getIndexShardsNumber(), config.getIndexReplicasNumber(), config.getIndexRefreshInterval());
-
+        super(moduleManager, config);
         this.config = config;
     }