kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Enhance TopN config to support excludes filter (#13361) 

* Enhance TopN config to support excludes filter

* banyandb version
This commit is contained in:
Wan Kai 2025-07-09 09:33:27 +08:00 committed by GitHub
parent 4b508b930d
commit 960d5f4e36
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 222 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
docs/en/setup/backend/storages/banyandb.md
View File

@ -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

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

@ -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;
}
}
}

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

@ -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

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

@ -70,21 +70,18 @@ public class BanyanDBAggregationQueryDAO extends AbstractBanyanDBDAO implements
// 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()));
}

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

@ -25,6 +25,7 @@ import java.util.List;
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 @@ public class BanyanDBConfigLoader {
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].");

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

@ -475,6 +475,7 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
.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);
}

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

@ -22,12 +22,15 @@ import com.google.common.base.Splitter;
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 class BanyanDBStorageConfig extends ModuleConfig {
*/
private Sort sort = Sort.all;
private Set<KeyValue> excludes = new HashSet<>();
public enum Sort {
all(BanyandbModel.Sort.SORT_UNSPECIFIED),
des(BanyandbModel.Sort.SORT_DESC),

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

@ -26,6 +26,7 @@ import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
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,44 +225,14 @@ public class BanyanDBStorageProvider extends ModuleProvider {
@Override
public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException {
// Cleanup TopN rules in BanyanDB server that are not configured in the current config.
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
);
}
}
}
}
}
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);
}
} catch (BanyanDBException e) {
throw new ModuleStartException(e.getMessage(), e);
}
}
@ -269,4 +240,41 @@ public class BanyanDBStorageProvider extends ModuleProvider {
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());
});
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
);
}
}
}
}
}
}
}
}

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

@ -20,6 +20,8 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb;
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,6 +45,9 @@ import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagFamilySpec
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;
@ -50,6 +55,7 @@ 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 @@ public enum MetadataRegistry {
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 @@ public enum MetadataRegistry {
.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 @@ public enum MetadataRegistry {
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());
});

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

@ -207,9 +207,9 @@ public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageCli
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);

86
test/e2e-v2/cases/storage/banyandb/config/bydb-topn.yml Normal file
View File

@ -0,0 +1,86 @@
# 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.
# This file is used to configure the TopN rules for BanyanDB in SkyWalking OAP server.
# The rules define how to aggregate and sort `metrics (Measure)` for services, endpoints, and instances.
#
# - name: Required. The name of the TopN rule, uniquely identifies the rule.
# - metricName: Required. The name of the metric to be aggregated.
# - groupByTagNames: Optional, default `[]`. The tag names to group the metrics by. If not specified, the metrics will sort without grouped.
# - countersNumber: Optional, default `1000`. The max size of entries in a time window for the pre-aggregation.
# The size of LRU determines the maximally tolerated time range.
# The buffers in the time range are kept in the memory so that
# the data in [T - lruSize * n, T] would be accepted in the pre-aggregation process.
# T = the current time in the current dimensionality.
# n = interval in the current dimensionality.
# - lruSizeMinute: Optional, default `10`. Defines how many time_buckets are held in the memory for minute-level metrics.
# - 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
# `attr0` is defined in the `EndpointDecorator` as the Layer.
- name: endpoint_cpm-layer
metricName: endpoint_cpm
groupByTagNames:
- attr0
sort: des
- name: endpoint_cpm-service
metricName: endpoint_cpm
groupByTagNames:
- service_id
sort: des
excludes:
- tag: attr0
value: MESH
- name: endpoint_sla-layer
metricName: endpoint_sla
groupByTagNames:
- attr0
sort: asc
- name: endpoint_sla-service
metricName: endpoint_sla
groupByTagNames:
- service_id
sort: asc
- name: endpoint_resp_time-layer
metricName: endpoint_resp_time
groupByTagNames:
- attr0
sort: des
- name: endpoint_resp_time-service
metricName: endpoint_resp_time
groupByTagNames:
- service_id
sort: des
# browser_app_page_pv metrics
- name: browser_app_page_pv-service
metricName: browser_app_page_pv
groupByTagNames:
- service_id
sort: des
- name: browser_app_page_error_sum-service
metricName: browser_app_page_error_sum
groupByTagNames:
- service_id
sort: des
- name: browser_app_page_error_rate-service
metricName: browser_app_page_error_rate
groupByTagNames:
- service_id
sort: des

2
test/e2e-v2/cases/storage/banyandb/docker-compose.yml
View File

@ -27,6 +27,8 @@ services:
extends:
file: ../../../script/docker-compose/base-compose.yml
service: oap
volumes:
- ./config/bydb-topn.yml:/skywalking/config/bydb-topn.yml
environment:
SW_STORAGE: banyandb
ports:

2
test/e2e-v2/script/env
View File

@ -23,7 +23,7 @@ SW_AGENT_CLIENT_JS_COMMIT=af0565a67d382b683c1dbd94c379b7080db61449
SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
SW_KUBERNETES_COMMIT_SHA=6fe5e6f0d3b7686c6be0457733e825ee68cb9b35
SW_ROVER_COMMIT=79292fe07f17f98f486e0c4471213e1961fb2d1d
SW_BANYANDB_COMMIT=dccf207dc9a74225fd4e5ba7c2c3ad6f36e6694a
SW_BANYANDB_COMMIT=b90a0ff31a4d8560422c04d152e7e7188d72ea89
SW_AGENT_PHP_COMMIT=d1114e7be5d89881eec76e5b56e69ff844691e35
SW_PREDICTOR_COMMIT=54a0197654a3781a6f73ce35146c712af297c994