Clean up TopN rules in BanyanDB server that are not configured in the current config. (#13336)

Author: wankai123

docs/en/setup/backend/configuration-vocabulary.md

@@ -371,6 +371,7 @@ The global settings for the whole BanyanDB:
 | asyncProfilerTaskQueryMaxSize   | Max size of AsyncProfilerTask to be fetched.                                                          | SW_STORAGE_BANYANDB_ASYNC_PROFILER_TASK_QUERY_MAX_SIZE  | 200             |
 | profileDataQueryScrollBatchSize | The batch size of query profiling data.                                                               | SW_STORAGE_BANYAND_QUERY_PROFILE_DATA_BATCH_SIZE        | 100             |
 | sslTrustCAPath                  | If the BanyanDB server is configured with TLS, config the TLS cert file path and open tls connection. | SW_STORAGE_BANYANDB_SSL_TRUST_CA_PATH                   | -               |
+| cleanupUnusedTopNRules          | Cleanup TopN rules in BanyanDB server that are not configured in the bydb-topn.yml config.            | SW_STORAGE_BANYANDB_CLEANUP_UNUSED_TOPN_RULES           | true            |
 
 ### Group Configuration
 The settings for each group:

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

@@ -45,6 +45,8 @@ global:
   asyncProfilerTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_ASYNC_PROFILER_TASK_QUERY_MAX_SIZE:200}
   # If the BanyanDB server is configured with TLS, configure the TLS cert file path and enable TLS connection.
   sslTrustCAPath: ${SW_STORAGE_BANYANDB_SSL_TRUST_CA_PATH:""}
+  # Cleanup TopN rules in BanyanDB server that are not configured in the bydb-topn.yml config.
+  cleanupUnusedTopNRules: ${SW_STORAGE_BANYANDB_CLEANUP_UNUSED_TOPN_RULES:false}
 
 groups:
   # The group settings of record.

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

@@ -619,7 +619,7 @@ private void checkIndexRuleBinding(List<IndexRule> indexRules,
 
     /**
      * Check if the TopN aggregation exists and update it if necessary.
-     * todo:// can not delete TopN here now.
+     * If the TopN rules are not used, will be checked and deleted after install, in the `BanyanDBStorageProvider.notifyAfterCompleted()`
      */
     private void checkTopNAggregation(Model model, BanyanDBClient client) throws BanyanDBException {
         MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);

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

@@ -46,7 +46,7 @@ public class BanyanDBStorageConfig extends ModuleConfig {
     private Metadata metadata = new Metadata();
     private Property property = new Property();
 
-    private Map<String/*metric name*/, Map<String, TopN>> topNConfigs = new HashMap<>();
+    private Map<String/*metric name*/, Map<String, TopN>/*ruleName, topN*/> topNConfigs = new HashMap<>();
 
     public String[] getTargetArray() {
         return Iterables.toArray(
@@ -98,6 +98,7 @@ public static class Global {
         private int metadataQueryMaxSize = 5000;
         private int segmentQueryMaxSize = 200;
         private int profileDataQueryBatchSize = 100;
+        private boolean cleanupUnusedTopNRules = true;
     }
 
     // The configuration of the groups.

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

@@ -18,6 +18,13 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.banyandb;
 
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
+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.storage.IBatchDAO;
 import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO;
@@ -59,6 +66,7 @@
 import org.apache.skywalking.oap.server.library.module.ModuleProvider;
 import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.measure.BanyanDBEBPFProfilingScheduleQueryDAO;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBEventQueryDAO;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.measure.BanyanDBHierarchyQueryDAO;
@@ -89,6 +97,7 @@
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
+@Slf4j
 public class BanyanDBStorageProvider extends ModuleProvider {
     private BanyanDBStorageConfig config;
     private BanyanDBStorageClient client;
@@ -215,7 +224,45 @@ 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.
+        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
+                                    );
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        } catch (BanyanDBException e) {
+            throw new ModuleStartException(e.getMessage(), e);
+        }
     }
 
     @Override