BanyanDB: Add support for compatibility checks based on API version (#12959)

Author: wu-sheng

docs/en/changes/changes.md

@@ -13,6 +13,8 @@
     source tar from the website and publish them to your private maven repository.
 * [Breaking Change] Remove H2 as storage option permanently. BanyanDB 0.8(OAP 10.2 required) is easy, stable and 
   production-ready. Don't need H2 as default storage anymore.
+* [Breaking Change] Bump up BanyanDB server version to 0.8.0. This version is not compatible with the previous 
+  versions. Please upgrade the BanyanDB server to 0.8.0 before upgrading OAP to 10.2.0.
 
 #### OAP Server
 
@@ -58,6 +60,7 @@
 * Added `maxLabelCount` parameter in the `labelCount` function of OAL to limit the number of labels can be counted.
 * Adapt the new Browser API(`/browser/perfData/webVitals`, `/browser/perfData/resources`) protocol.
 * Add Circuit Breaking mechanism.
+* BanyanDB: Add support for compatibility checks based on the BanyanDB server's API version.
 
 #### UI
 

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

@@ -14,6 +14,12 @@ bydb.version=x.y
 bydb.api.version=x.y
 ```
 
+If the BanyanDB server API version is not compatible with the OAP server, the OAP server will not start, and the following error message will be displayed:
+```shell
+... ERROR [] - ... Incompatible BanyanDB server API version: 0.x. But accepted versions: 0.y
+org.apache.skywalking.oap.server.library.module.ModuleStartException: Incompatible BanyanDB server API version...
+```
+
 ### Configuration
 
 ```yaml

oap-server-bom/pom.xml

@@ -72,7 +72,7 @@
         <httpcore.version>4.4.13</httpcore.version>
         <httpasyncclient.version>4.1.5</httpasyncclient.version>
         <commons-compress.version>1.21</commons-compress.version>
-        <banyandb-java-client.version>0.8.0-rc1</banyandb-java-client.version>
+        <banyandb-java-client.version>0.8.0-rc2</banyandb-java-client.version>
         <kafka-clients.version>3.4.0</kafka-clients.version>
         <spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
         <consul.client.version>1.5.3</consul.client.version>

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

@@ -19,8 +19,22 @@
 package org.apache.skywalking.oap.server.storage.plugin.banyandb;
 
 import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Group;
 import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
+import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Measure;
+import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Stream;
+import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TopNAggregation;
 import org.apache.skywalking.banyandb.property.v1.BanyandbProperty;
+import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.ApplyRequest.Strategy;
+import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.DeleteResponse;
+import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.Property;
 import org.apache.skywalking.banyandb.v1.client.BanyanDBClient;
 import org.apache.skywalking.banyandb.v1.client.MeasureBulkWriteProcessor;
 import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
@@ -33,29 +47,20 @@
 import org.apache.skywalking.banyandb.v1.client.StreamWrite;
 import org.apache.skywalking.banyandb.v1.client.TopNQuery;
 import org.apache.skywalking.banyandb.v1.client.TopNQueryResponse;
-import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Group;
-import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TopNAggregation;
-import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Measure;
-import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Stream;
-import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.Property;
-import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.ApplyRequest.Strategy;
-import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.DeleteResponse;
 import org.apache.skywalking.banyandb.v1.client.grpc.exception.AlreadyExistsException;
 import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
 import org.apache.skywalking.oap.server.library.client.Client;
 import org.apache.skywalking.oap.server.library.client.healthcheck.DelegatedHealthChecker;
 import org.apache.skywalking.oap.server.library.client.healthcheck.HealthCheckable;
 import org.apache.skywalking.oap.server.library.util.HealthChecker;
 
-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-
 /**
  * BanyanDBStorageClient is a simple wrapper for the underlying {@link BanyanDBClient},
  * which implement {@link Client} and {@link HealthCheckable}.
  */
+@Slf4j
 public class BanyanDBStorageClient implements Client, HealthCheckable {
+    private static final String[] COMPATIBLE_SERVER_API_VERSIONS = {"0.8"};
     final BanyanDBClient client;
     private final DelegatedHealthChecker healthChecker = new DelegatedHealthChecker();
     private final int flushTimeout;
@@ -70,6 +75,29 @@ public BanyanDBStorageClient(BanyanDBStorageConfig config) {
     @Override
     public void connect() throws Exception {
         this.client.connect();
+        BanyandbCommon.APIVersion apiVersion;
+        try {
+            apiVersion = this.client.getAPIVersion();
+        } catch (BanyanDBException e) {
+            final Throwable cause = e.getCause();
+            if (cause instanceof StatusRuntimeException) {
+                final Status status = ((StatusRuntimeException) cause).getStatus();
+                if (Status.Code.UNIMPLEMENTED.equals(status.getCode())) {
+                    log.error("fail to get BanyanDB API version, server version < 0.8 is not supported.");
+                }
+            }
+            throw e;
+        }
+        final boolean isCompatible = Arrays.stream(COMPATIBLE_SERVER_API_VERSIONS)
+                                           .anyMatch(v -> v.equals(apiVersion.getVersion()));
+        final String revision = apiVersion.getRevision();
+        log.info("BanyanDB server API version: {}, revision: {}", apiVersion.getVersion(), revision);
+        if (!isCompatible) {
+            throw new IllegalStateException(
+                "Incompatible BanyanDB server API version: " + apiVersion.getVersion() + ". But accepted versions: "
+                    + String.join(", ", COMPATIBLE_SERVER_API_VERSIONS));
+        }
+
     }
 
     @Override
@@ -79,10 +107,10 @@ public void shutdown() throws IOException {
 
     public List<Property> listProperties(String group, String name) throws IOException {
         try {
-             BanyandbProperty.QueryResponse resp = this.client.query(BanyandbProperty.QueryRequest.newBuilder()
-                            .addGroups(group)
-                            .setContainer(name)
-                    .build());
+            BanyandbProperty.QueryResponse resp = this.client.query(BanyandbProperty.QueryRequest.newBuilder()
+                                                                                                 .addGroups(group)
+                                                                                                 .setContainer(name)
+                                                                                                 .build());
             this.healthChecker.health();
             return resp.getPropertiesList();
         } catch (BanyanDBException ex) {
@@ -99,10 +127,10 @@ public List<Property> listProperties(String group, String name) throws IOExcepti
     public Property queryProperty(String group, String name, String id) throws IOException {
         try {
             BanyandbProperty.QueryResponse resp = this.client.query(BanyandbProperty.QueryRequest.newBuilder()
-                    .addGroups(group)
-                    .setContainer(name)
-                    .addIds(id)
-                    .build());
+                                                                                                 .addGroups(group)
+                                                                                                 .setContainer(name)
+                                                                                                 .addIds(id)
+                                                                                                 .build());
             this.healthChecker.health();
             if (resp.getPropertiesCount() == 0) {
                 return null;

test/e2e-v2/script/env

@@ -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=0ae8f12d6eb6cc9fa125c603ee57d0b21fc8c6d0
-SW_BANYANDB_COMMIT=8dfe616f332848f041b0f184b023d9a256eac39e
+SW_BANYANDB_COMMIT=362d68ed79c532e6d61dd6674cce38090caa0da7
 SW_AGENT_PHP_COMMIT=3192c553002707d344bd6774cfab5bc61f67a1d3
 
 SW_CTL_COMMIT=67cbc89dd7b214d5791321a7ca992f940cb586ba