Enhance TopN config to support excludes filter (#13361)

* Enhance TopN config to support excludes filter

* banyandb version

Author: wankai123

docs/en/setup/backend/storages/banyandb.md

@@ -255,6 +255,7 @@ You can define the TopN rules for different metrics. The configuration is define
 # - lruSizeHourDay: Optional, default `2`. Defines how many time_buckets are held in the memory for hour and day-level metrics.
 
 # - sort: Optional, default `all`. The sorting order for the metrics, asc, des or all(include both asc and des).
+# - excludes: Optional, default `[]`. The tag values to be excluded from the candidates. If specified, the candidates will not include the entries with the specified tag values.
 
 TopN-Rules:
    # endpoint metrics
@@ -314,6 +315,16 @@ TopN-Rules:
     groupByTagNames:
       - service_id
     sort: des
+# The following rule can be used to filter out the mesh endpoints.
+# You MUST add `attr0!= MESH` to the MQE topN query to hit this rule.
+#  - name: endpoint_cpm-service
+#    metricName: endpoint_cpm
+#    groupByTagNames:
+#      - service_id
+#    sort: des
+#    excludes:
+#      - tag: attr0
+#        value: MESH
 ```
 
 ### Installation Modes

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/input/AttrCondition.java

@@ -27,4 +27,13 @@ public class AttrCondition {
     private final String key;
     private final String value;
     private final boolean isEquals;
+
+    @Override
+    public String toString() {
+        if (isEquals) {
+            return key + "==" + value;
+        } else  {
+            return key + "!=" + value;
+        }
+    }
 }

oap-server/server-starter/src/main/resources/bydb-topn.yml

@@ -30,6 +30,7 @@
 # - lruSizeHourDay: Optional, default `2`. Defines how many time_buckets are held in the memory for hour and day-level metrics.
 
 # - sort: Optional, default `all`. The sorting order for the metrics, asc, des or all(include both asc and des).
+# - excludes: Optional, default `[]`. The tag values to be excluded from the candidates. If specified, the candidates will not include the entries with the specified tag values.
 
 TopN-Rules:
    # endpoint metrics
@@ -79,4 +80,14 @@ TopN-Rules:
     metricName: browser_app_page_error_rate
     groupByTagNames:
       - service_id
-    sort: des
+    sort: des
+# The following rule can be used to filter out the mesh endpoints.
+# You MUST add `attr0!= MESH` to the MQE topN query to hit this rule.
+#  - name: endpoint_cpm-service
+#    metricName: endpoint_cpm
+#    groupByTagNames:
+#      - service_id
+#    sort: des
+#    excludes:
+#      - tag: attr0
+#        value: MESH

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java

@@ -70,21 +70,18 @@ public List<SelectedRecord> sortMetrics(TopNCondition condition, String valueCol
         // The query tags are the additional conditions and attributes defined in the TopN condition.
         // The query tags is the key to find the TopN aggregation in the schema.
         // If the TopN aggregation is defined in the schema, it will be used to perform the query.
-        // The server-side TopN only support when attribute condition `isEquals == true`.
+        // The server-side TopN only support the rules config in bydb-topn.yaml.
         ImmutableSet.Builder<String> queryTags = ImmutableSet.builder();
-        boolean equalsQuery = true;
         if (condition.getAttributes() != null) {
             for (AttrCondition attr : condition.getAttributes()) {
                 if (!attr.isEquals()) {
-                    equalsQuery = false;
-                    break;
+                    // Not equal attr condition should add full String key!=value
+                    queryTags.add(attr.toString());
+                } else {
+                    queryTags.add(attr.getKey());
                 }
-                queryTags.add(attr.getKey());
             }
         }
-        if (!equalsQuery) {
-            return directMetricsTopN(isColdStage, condition, schema, valueColumnName, spec, getTimestampRange(duration), additionalConditions);
-        }
         if (additionalConditions != null) {
             additionalConditions.forEach(additionalCondition -> queryTags.add(additionalCondition.getKey()));
         }

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConfigLoader.java

@@ -25,6 +25,7 @@
 import java.util.Map;
 import java.util.Properties;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.query.type.KeyValue;
 import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
 import org.apache.skywalking.oap.server.library.module.ModuleProvider;
 import org.apache.skywalking.oap.server.library.module.ModuleStartException;
@@ -236,6 +237,19 @@ private void loadTopNConfig() throws ModuleStartException {
             if (sort != null) {
                 topN.setSort(TopN.Sort.valueOf(sort.toString()));
             }
+            var excludes = rule.get("excludes");
+            if (excludes != null) {
+                for (Map<String, String> tag : (List<Map<String, String>>) excludes) {
+                    var tagName = tag.get("tag");
+                    var tagValue = tag.get("value");
+                    if (tagName == null || tagValue == null) {
+                        throw new ModuleStartException(
+                            "TopN rule name: " + name + ", [tag] or [value] is missing in [excludes] item in file [bydb-topn.yml].");
+                    }
+                    topN.getExcludes().add(new KeyValue(tag.get("tag"), tag.get("value")));
+                }
+            }
+
             Map<String, TopN> map = config.getTopNConfigs().computeIfAbsent(metricName.toString(), k -> new HashMap<>());
             if (map.put(name.toString(), topN) != null) {
                 throw new ModuleStartException("Duplicate TopN rule name: " + name + " in file [bydb-topn.yml].");

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java

@@ -475,6 +475,7 @@ private void checkMeasure(Measure measure, BanyanDBClient client) throws BanyanD
                                        .build()
                                        .equals(measure.toBuilder().clearMetadata().build());
             if (!equals) {
+                // banyanDB server can not delete or update Tags.
                 client.update(measure);
                 log.info("update Measure: {} from: {} to: {}", hisMeasure.getMetadata().getName(), hisMeasure, measure);
             }

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java

@@ -22,12 +22,15 @@
 import com.google.common.collect.Iterables;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import lombok.Data;
 import lombok.Getter;
 import lombok.Setter;
 import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
+import org.apache.skywalking.oap.server.core.query.type.KeyValue;
 import org.apache.skywalking.oap.server.library.module.ModuleConfig;
 
 @Getter
@@ -259,6 +262,8 @@ public static class TopN {
          */
         private Sort sort = Sort.all;
 
+        private Set<KeyValue> excludes = new HashSet<>();
+
         public enum Sort {
             all(BanyandbModel.Sort.SORT_UNSPECIFIED),
             des(BanyandbModel.Sort.SORT_DESC),

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java

@@ -26,6 +26,7 @@
 import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
 import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
 import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.RunningMode;
 import org.apache.skywalking.oap.server.core.storage.IBatchDAO;
 import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO;
 import org.apache.skywalking.oap.server.core.storage.StorageBuilderFactory;
@@ -224,49 +225,56 @@ public void start() throws ServiceNotProvidedException, ModuleStartException {
 
     @Override
     public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException {
-        // Cleanup TopN rules in BanyanDB server that are not configured in the current config.
+        if (!RunningMode.isNoInitMode()) {
+            try {
+                List<BanyandbCommon.Group> groups = this.client.client.findGroups();
+                cleanupUnusedTopNRules(groups);
+                //todo: can not delete indexRules now, because banyanDB server can not delete or update Tags.
+            } catch (BanyanDBException e) {
+                throw new ModuleStartException(e.getMessage(), e);
+            }
+        }
+    }
+
+    @Override
+    public String[] requiredModules() {
+        return new String[] {CoreModule.NAME};
+    }
+
+    // Cleanup TopN rules in BanyanDB server that are not configured in the current config.
+    private void cleanupUnusedTopNRules(List<BanyandbCommon.Group> groups) throws BanyanDBException {
         Set<String> topNNames = new HashSet<>();
         this.config.getTopNConfigs().values().forEach(topNConfig -> {
             topNNames.addAll(topNConfig.keySet());
         });
-
-        try {
-            List<BanyandbCommon.Group> groups = this.client.client.findGroups();
-            for (BanyandbCommon.Group group : groups) {
-                if (BanyandbCommon.Catalog.CATALOG_MEASURE.equals(group.getCatalog())) {
-                    String groupName = group.getMetadata().getName();
-                    List<BanyandbDatabase.TopNAggregation> topNAggregations = this.client.client.findTopNAggregations(groupName);
-                    if (CollectionUtils.isNotEmpty(topNAggregations)) {
-                        for (BanyandbDatabase.TopNAggregation topNAggregation : topNAggregations) {
-                            String topNName = topNAggregation.getMetadata().getName();
-                            if (!topNNames.contains(topNName)) {
-                                if (this.config.getGlobal().isCleanupUnusedTopNRules()) {
-                                    this.client.client.deleteTopNAggregation(groupName, topNName);
-                                    log.info(
-                                        "Deleted unused topN rule from BanyanDB server: {}, group: {}. Please check bydb-topn.yml. " +
-                                            "If you don't want to cleanup unused rules from server, please set cleanupUnusedTopNRules=false in bydb.yml",
-                                        topNName, groupName
-                                    );
-                                } else {
-                                    // Log the unused TopN aggregation.
-                                    log.warn(
-                                        "Unused topN rule in BanyanDB server: {}, group: {}. Please check bydb-topn.yml. " +
-                                            "If you want to cleanup unused rules from server, please set cleanupUnusedTopNRules=true in bydb.yml",
-                                        topNName, groupName
-                                    );
-                                }
+        for (BanyandbCommon.Group group : groups) {
+            if (BanyandbCommon.Catalog.CATALOG_MEASURE.equals(group.getCatalog())) {
+                String groupName = group.getMetadata().getName();
+                List<BanyandbDatabase.TopNAggregation> topNAggregations = this.client.client.findTopNAggregations(
+                    groupName);
+                if (CollectionUtils.isNotEmpty(topNAggregations)) {
+                    for (BanyandbDatabase.TopNAggregation topNAggregation : topNAggregations) {
+                        String topNName = topNAggregation.getMetadata().getName();
+                        if (!topNNames.contains(topNName)) {
+                            if (this.config.getGlobal().isCleanupUnusedTopNRules()) {
+                                this.client.client.deleteTopNAggregation(groupName, topNName);
+                                log.info(
+                                    "Deleted unused topN rule from BanyanDB server: {}, group: {}. Please check bydb-topn.yml. " +
+                                        "If you don't want to cleanup unused rules from server, please set cleanupUnusedTopNRules=false in bydb.yml",
+                                    topNName, groupName
+                                );
+                            } else {
+                                // Log the unused TopN aggregation.
+                                log.warn(
+                                    "Unused topN rule in BanyanDB server: {}, group: {}. Please check bydb-topn.yml. " +
+                                        "If you want to cleanup unused rules from server, please set cleanupUnusedTopNRules=true in bydb.yml",
+                                    topNName, groupName
+                                );
                             }
                         }
                     }
                 }
             }
-        } catch (BanyanDBException e) {
-            throw new ModuleStartException(e.getMessage(), e);
         }
     }
-
-    @Override
-    public String[] requiredModules() {
-        return new String[] {CoreModule.NAME};
-    }
 }

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java

@@ -20,6 +20,8 @@
 
 import com.google.common.collect.ImmutableSet;
 import com.google.gson.JsonObject;
+import java.util.HashSet;
+import java.util.function.BiFunction;
 import lombok.Builder;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -43,13 +45,17 @@
 import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagSpec;
 import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagType;
 import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TopNAggregation;
+import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
+import org.apache.skywalking.banyandb.v1.client.And;
+import org.apache.skywalking.banyandb.v1.client.PairQueryCondition;
 import org.apache.skywalking.banyandb.v1.client.metadata.Duration;
 import org.apache.skywalking.oap.server.core.analysis.DownSampling;
 import org.apache.skywalking.oap.server.core.analysis.metrics.IntList;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.record.Record;
 import org.apache.skywalking.oap.server.core.config.DownSamplingConfigService;
 import org.apache.skywalking.oap.server.core.query.enumeration.Step;
+import org.apache.skywalking.oap.server.core.query.type.KeyValue;
 import org.apache.skywalking.oap.server.core.storage.StorageException;
 import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
 import org.apache.skywalking.oap.server.core.storage.annotation.Column;
@@ -217,6 +223,7 @@ private Map<ImmutableSet<String>, TopNAggregation> parseTopNSpecs(final Model mo
         Map<ImmutableSet<String>, TopNAggregation> topNAggregations = new HashMap<>();
         topNConfig.forEach((name, topN) -> {
             ImmutableSet<String> key = ImmutableSet.of();
+            Set<String> queryConditions = new HashSet<>();
             TopNAggregation.Builder topNAggregation = TopNAggregation.newBuilder()
                                                                      .setMetadata(
                                                                          Metadata.newBuilder().setGroup(group).setName(name))
@@ -225,7 +232,7 @@ private Map<ImmutableSet<String>, TopNAggregation> parseTopNSpecs(final Model mo
                                                                      .setFieldName(valueColumnOpt.get().getValueCName())
                                                                      .setCountersNumber(topN.getCountersNumber());
             if (topN.getGroupByTagNames() != null) {
-                key = ImmutableSet.copyOf(topN.getGroupByTagNames());
+                queryConditions.addAll(topN.getGroupByTagNames());
                 //check tags
                 topN.getGroupByTagNames().forEach(tag -> {
                     if (!tags.contains(tag)) {
@@ -247,8 +254,28 @@ private Map<ImmutableSet<String>, TopNAggregation> parseTopNSpecs(final Model mo
                 default:
                     throw new UnsupportedOperationException("unsupported downsampling: " + model.getDownsampling());
             }
+
+            if (CollectionUtils.isNotEmpty(topN.getExcludes())) {
+                AbstractCriteria criteria;
+                List<AbstractCriteria> conditions = new ArrayList<>(topN.getExcludes().size());
+                for (KeyValue keyValue : topN.getExcludes()) {
+                    conditions.add(PairQueryCondition.StringQueryCondition.ne(keyValue.getKey(), keyValue.getValue()));
+                    queryConditions.remove(keyValue.getKey());
+                    queryConditions.add(keyValue.getKey() + "!=" + keyValue.getValue());
+                }
+                if (conditions.size() == 1) {
+                    criteria = conditions.get(0);
+                } else {
+                    criteria = conditions.subList(2, conditions.size()).stream().reduce(
+                        And.create(conditions.get(0), conditions.get(1)),
+                        (BiFunction<AbstractCriteria, AbstractCriteria, AbstractCriteria>) And::create,
+                        And::create);
+                }
+                topNAggregation.setCriteria(criteria.build());
+            }
+            key = ImmutableSet.copyOf(queryConditions);
             if (topNAggregations.containsKey(key)) {
-                throw new IllegalArgumentException("In file [bydb-topn.yml], TopN rule " + topN.getName() + "'s groupByTagNames " + key + " already exist in the same metric " + model.getName());
+                throw new IllegalArgumentException("In file [bydb-topn.yml], TopN rule " + topN.getName() + "'s groupByTagNames and excludes " + key + " already exist in the same metric " + model.getName());
             }
             topNAggregations.put(key, topNAggregation.build());
         });

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java

@@ -207,9 +207,9 @@ private TopNQueryResponse topNQuery(boolean isColdStage,
             attributes.forEach(attr -> {
                 if (attr.isEquals()) {
                     conditions.add(PairQueryCondition.StringQueryCondition.eq(attr.getKey(), attr.getValue()));
-                } else {
-                    conditions.add(PairQueryCondition.StringQueryCondition.ne(attr.getKey(), attr.getValue()));
                 }
+                //server side topn query does not support not equals condition
+                //the not equals condition should be handled by a specific topN rule by adding exclude condition.
             });
         }
         q.setConditions(conditions);