[Feature] Support ElasticSearch 7 as backend storage

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

@@ -111,14 +111,8 @@ private List<TopNEntity> aggregation(String indexName, String valueCName, Search
             asc = true;
         }
 
-        TermsAggregationBuilder aggregationBuilder = AggregationBuilders
-            .terms(Metrics.ENTITY_ID)
-            .field(Metrics.ENTITY_ID)
-            .order(BucketOrder.aggregation(valueCName, asc))
-            .size(topN)
-            .subAggregation(
-                AggregationBuilders.avg(valueCName).field(valueCName)
-            );
+        TermsAggregationBuilder aggregationBuilder = aggregationBuilder(valueCName, topN, asc);
+
         sourceBuilder.aggregation(aggregationBuilder);
 
         SearchResponse response = getClient().search(indexName, sourceBuilder);
@@ -135,4 +129,15 @@ private List<TopNEntity> aggregation(String indexName, String valueCName, Search
 
         return topNEntities;
     }
+
+    protected TermsAggregationBuilder aggregationBuilder(final String valueCName, final int topN, final boolean asc) {
+        return AggregationBuilders
+                .terms(Metrics.ENTITY_ID)
+                .field(Metrics.ENTITY_ID)
+                .order(BucketOrder.aggregation(valueCName, asc))
+                .size(topN)
+                .subAggregation(
+                    AggregationBuilders.avg(valueCName).field(valueCName)
+                );
+    }
 }

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

@@ -47,7 +47,6 @@
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.BatchProcessEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.HistoryDeleteEsDAO;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.AggregationQueryEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TopNRecordsQueryEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TopologyQueryEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.ttl.ElasticsearchStorageTTL;
@@ -60,6 +59,7 @@
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.dao.StorageEsInstaller;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.lock.RegisterLockDAOImpl;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.lock.RegisterLockInstaller;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.AggregationQueryEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.AlarmQueryEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.LogQueryEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.MetadataQueryEsDAO;

oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/query/AggregationQueryEsDAO.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query;
+
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.search.aggregations.BucketOrder;
+import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
+
+/**
+ * @author peng-yongsheng
+ * @author kezhenxu94
+ */
+public class AggregationQueryEsDAO extends org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.AggregationQueryEsDAO {
+
+    public AggregationQueryEsDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    protected TermsAggregationBuilder aggregationBuilder(final String valueCName, final int topN, final boolean asc) {
+        return AggregationBuilders
+            .terms(Metrics.ENTITY_ID)
+            .field(Metrics.ENTITY_ID)
+            .order(BucketOrder.aggregation(valueCName, asc))
+            .size(topN)
+            .subAggregation(
+                AggregationBuilders.avg(valueCName).field(valueCName)
+            );
+    }
+}

oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/query/MetricsQueryEsDAO.java

@@ -38,15 +38,24 @@
 
 /**
  * @author peng-yongsheng
+ * @author kezhenxu94
  */
 public class MetricsQueryEsDAO extends org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.MetricsQueryEsDAO {
 
     public MetricsQueryEsDAO(ElasticSearchClient client) {
         super(client);
     }
 
-    @Override public IntValues getValues(String indName, Downsampling downsampling, long startTB, long endTB, Where where, String valueCName,
+    @Override
+    public IntValues getValues(
+        String indName,
+        Downsampling downsampling,
+        long startTB,
+        long endTB,
+        Where where,
+        String valueCName,
         Function function) throws IOException {
+
         String indexName = ModelName.build(downsampling, indName);
 
         SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
@@ -66,15 +75,15 @@ public MetricsQueryEsDAO(ElasticSearchClient client) {
             switch (function) {
                 case Sum:
                     Sum sum = idBucket.getAggregations().get(valueCName);
-                    value = (long)sum.getValue();
+                    value = (long) sum.getValue();
                     break;
                 case Avg:
                     Avg avg = idBucket.getAggregations().get(valueCName);
-                    value = (long)avg.getValue();
+                    value = (long) avg.getValue();
                     break;
                 default:
                     avg = idBucket.getAggregations().get(valueCName);
-                    value = (long)avg.getValue();
+                    value = (long) avg.getValue();
                     break;
             }
 
@@ -85,4 +94,18 @@ public MetricsQueryEsDAO(ElasticSearchClient client) {
         }
         return intValues;
     }
+
+    protected void functionAggregation(Function function, TermsAggregationBuilder parentAggBuilder, String valueCName) {
+        switch (function) {
+            case Avg:
+                parentAggBuilder.subAggregation(AggregationBuilders.avg(valueCName).field(valueCName));
+                break;
+            case Sum:
+                parentAggBuilder.subAggregation(AggregationBuilders.sum(valueCName).field(valueCName));
+                break;
+            default:
+                parentAggBuilder.subAggregation(AggregationBuilders.avg(valueCName).field(valueCName));
+                break;
+        }
+    }
 }