diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index 96e4410e41..0e5e806f81 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -65,6 +65,7 @@ * Fix the previous analysis result missing in the ALS `k8s-mesh` analyzer. * Fix `findEndpoint` query require `keyword` when using BanyanDB. * Support to analysis the ztunnel mapped IP address in eBPF Access Log Receiver. +* Adapt BanyanDB Java Client 0.7.0-rc3. #### UI diff --git a/oap-server-bom/pom.xml b/oap-server-bom/pom.xml index f5385435d8..158f66f033 100644 --- a/oap-server-bom/pom.xml +++ b/oap-server-bom/pom.xml @@ -73,7 +73,7 @@ 4.4.13 4.1.5 1.21 - 0.7.0-rc2 + 0.7-rc3 3.4.0 2.4.6.RELEASE 1.5.3 diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java index b95737843b..9c98a99d2b 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java @@ -22,8 +22,9 @@ import io.grpc.Status; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.banyandb.v1.client.BanyanDBClient; import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException; -import org.apache.skywalking.banyandb.v1.client.metadata.Measure; -import org.apache.skywalking.banyandb.v1.client.metadata.Stream; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Measure; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Stream; +import org.apache.skywalking.banyandb.v1.client.metadata.MetadataCache; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.config.ConfigService; import org.apache.skywalking.oap.server.core.storage.StorageException; @@ -31,6 +32,7 @@ import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller; import org.apache.skywalking.oap.server.library.client.Client; import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.library.util.CollectionUtils; @Slf4j public class BanyanDBIndexInstaller extends ModelInstaller { @@ -58,20 +60,20 @@ public class BanyanDBIndexInstaller extends ModelInstaller { final boolean resourceExist = metadata.checkResourceExistence(c); if (!resourceExist) { return false; - } - - // then check entity schema - if (metadata.findRemoteSchema(c).isPresent()) { - // register models only locally but not remotely + } else { + // register models only locally(Schema cache) but not remotely if (model.isRecord()) { // stream MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService); } else { // measure MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService); } + // pre-load remote schema for java client + MetadataCache.EntityMetadata remoteMeta = metadata.updateRemoteSchema(c); + if (remoteMeta == null) { + throw new IllegalStateException("inconsistent state: metadata:" + metadata + ", remoteMeta: null"); + } return true; } - - throw new IllegalStateException("inconsistent state:" + metadata); } catch (BanyanDBException ex) { throw new StorageException("fail to check existence", ex); } @@ -84,11 +86,17 @@ public class BanyanDBIndexInstaller extends ModelInstaller { .provider() .getService(ConfigService.class); if (model.isRecord()) { // stream - Stream stream = MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService); + StreamModel streamModel = MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService); + Stream stream = streamModel.getStream(); if (stream != null) { log.info("install stream schema {}", model.getName()); + final BanyanDBClient client = ((BanyanDBStorageClient) this.client).client; try { - ((BanyanDBStorageClient) client).define(stream); + if (CollectionUtils.isNotEmpty(streamModel.getIndexRules())) { + client.define(stream, streamModel.getIndexRules()); + } else { + client.define(stream); + } } catch (BanyanDBException ex) { if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) { log.info( @@ -102,12 +110,17 @@ public class BanyanDBIndexInstaller extends ModelInstaller { } } } else { // measure - Measure measure = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService); + MeasureModel measureModel = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService); + Measure measure = measureModel.getMeasure(); if (measure != null) { - log.info("install measure schema {}", measure.name()); - final BanyanDBClient c = ((BanyanDBStorageClient) this.client).client; + log.info("install measure schema {}", model.getName()); + final BanyanDBClient client = ((BanyanDBStorageClient) this.client).client; try { - c.define(measure); + if (CollectionUtils.isNotEmpty(measureModel.getIndexRules())) { + client.define(measure, measureModel.getIndexRules()); + } else { + client.define(measure); + } } catch (BanyanDBException ex) { if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) { log.info("Measure schema {}_{} already created by another OAP node", @@ -119,7 +132,7 @@ public class BanyanDBIndexInstaller extends ModelInstaller { } final MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model); try { - schema.installTopNAggregation(c); + schema.installTopNAggregation(client); } catch (BanyanDBException ex) { if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) { log.info("Measure schema {}_{} TopN({}) already created by another OAP node", diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java index 1dc2c2b7d2..51ce5381ac 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java @@ -44,7 +44,7 @@ public class BanyanDBNoneStreamDAO extends AbstractDAO im if (schema == null) { throw new IOException(model.getName() + " is not registered"); } - StreamWrite streamWrite = getClient().client.createStreamWrite( + StreamWrite streamWrite = getClient().createStreamWrite( schema.getMetadata().getGroup(), // group name schema.getMetadata().name(), // stream-name noneStream.id().build() // identity diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java index 26069cdc0d..35ce37194e 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java @@ -19,6 +19,7 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb; import io.grpc.Status; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase; import org.apache.skywalking.banyandb.v1.client.BanyanDBClient; import org.apache.skywalking.banyandb.v1.client.MeasureBulkWriteProcessor; import org.apache.skywalking.banyandb.v1.client.MeasureQuery; @@ -30,14 +31,15 @@ import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse; 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.banyandb.v1.client.metadata.Group; -import org.apache.skywalking.banyandb.v1.client.metadata.Measure; -import org.apache.skywalking.banyandb.v1.client.metadata.Property; -import org.apache.skywalking.banyandb.v1.client.metadata.PropertyStore; -import org.apache.skywalking.banyandb.v1.client.metadata.Stream; -import org.apache.skywalking.banyandb.v1.client.metadata.TopNAggregation; 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; @@ -103,9 +105,9 @@ public class BanyanDBStorageClient implements Client, HealthCheckable { } } - public PropertyStore.DeleteResult deleteProperty(String group, String name, String id, String... tags) throws IOException { + public DeleteResponse deleteProperty(String group, String name, String id, String... tags) throws IOException { try { - PropertyStore.DeleteResult result = this.client.deleteProperty(group, name, id, tags); + DeleteResponse result = this.client.deleteProperty(group, name, id, tags); this.healthChecker.health(); return result; } catch (BanyanDBException ex) { @@ -158,7 +160,7 @@ public class BanyanDBStorageClient implements Client, HealthCheckable { } /** - * PropertyStore.Strategy is default to {@link PropertyStore.Strategy#MERGE} + * PropertyStore.Strategy is default to {@link Strategy#STRATEGY_MERGE} */ public void define(Property property) throws IOException { try { @@ -170,7 +172,7 @@ public class BanyanDBStorageClient implements Client, HealthCheckable { } } - public void define(Property property, PropertyStore.Strategy strategy) throws IOException { + public void define(Property property, Strategy strategy) throws IOException { try { this.client.apply(property, strategy); this.healthChecker.health(); @@ -190,6 +192,16 @@ public class BanyanDBStorageClient implements Client, HealthCheckable { } } + public void define(Stream stream, List indexRules) throws BanyanDBException { + try { + this.client.define(stream, indexRules); + this.healthChecker.health(); + } catch (BanyanDBException ex) { + healthChecker.unHealth(ex); + throw ex; + } + } + public void define(Measure measure) throws BanyanDBException { try { this.client.define(measure); @@ -200,6 +212,16 @@ public class BanyanDBStorageClient implements Client, HealthCheckable { } } + public void define(Measure measure, List indexRules) throws BanyanDBException { + try { + this.client.define(measure, indexRules); + this.healthChecker.health(); + } catch (BanyanDBException ex) { + healthChecker.unHealth(ex); + throw ex; + } + } + public void defineIfEmpty(Group group) throws IOException { try { try { @@ -223,12 +245,20 @@ public class BanyanDBStorageClient implements Client, HealthCheckable { } } - public StreamWrite createStreamWrite(String group, String name, String elementId) { - return this.client.createStreamWrite(group, name, elementId); + public StreamWrite createStreamWrite(String group, String name, String elementId) throws IOException { + try { + return this.client.createStreamWrite(group, name, elementId); + } catch (BanyanDBException e) { + throw new IOException("fail to create stream write", e); + } } - public MeasureWrite createMeasureWrite(String group, String name, long timestamp) { - return this.client.createMeasureWrite(group, name, timestamp); + public MeasureWrite createMeasureWrite(String group, String name, long timestamp) throws IOException { + try { + return this.client.createMeasureWrite(group, name, timestamp); + } catch (BanyanDBException e) { + throw new IOException("fail to create measure write", e); + } } public void write(StreamWrite streamWrite) { diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java index 5a6c443d7a..e7660921b3 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java @@ -18,7 +18,7 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb; -import org.apache.skywalking.banyandb.v1.client.metadata.Group; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon; 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; @@ -175,7 +175,13 @@ public class BanyanDBStorageProvider extends ModuleProvider { this.client.registerChecker(healthChecker); try { this.client.connect(); - this.client.defineIfEmpty(Group.create(BanyanDBUITemplateManagementDAO.GROUP)); + this.client.defineIfEmpty(BanyandbCommon.Group.newBuilder() + .setMetadata( + BanyandbCommon.Metadata.newBuilder() + .setName( + BanyanDBUITemplateManagementDAO.GROUP)) + .setCatalog(BanyandbCommon.Catalog.CATALOG_UNSPECIFIED) + .build()); this.modelInstaller.start(); getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(modelInstaller); diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUIMenuManagementDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUIMenuManagementDAO.java index e6b923c904..74e7cc75dd 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUIMenuManagementDAO.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUIMenuManagementDAO.java @@ -19,8 +19,11 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb; import lombok.extern.slf4j.Slf4j; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon; +import org.apache.skywalking.banyandb.model.v1.BanyandbModel; +import org.apache.skywalking.banyandb.property.v1.BanyandbProperty; import org.apache.skywalking.banyandb.v1.client.TagAndValue; -import org.apache.skywalking.banyandb.v1.client.metadata.Property; +import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.Property; import org.apache.skywalking.oap.server.core.management.ui.menu.UIMenu; import org.apache.skywalking.oap.server.core.storage.management.UIMenuManagementDAO; import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO; @@ -46,17 +49,27 @@ public class BanyanDBUIMenuManagementDAO extends AbstractBanyanDBDAO implements @Override public void saveMenu(UIMenu menu) throws IOException { - this.getClient().define(Property.create(GROUP, UIMenu.INDEX_NAME, menu.id().build()) - .addTag(TagAndValue.newStringTag(UIMenu.CONFIGURATION, menu.getConfigurationJson())) - .addTag(TagAndValue.newLongTag(UIMenu.UPDATE_TIME, menu.getUpdateTime())) - .build()); + Property property = Property.newBuilder() + .setMetadata(BanyandbProperty.Metadata.newBuilder().setId(menu.getMenuId()) + .setContainer( + BanyandbCommon.Metadata.newBuilder() + .setGroup(GROUP) + .setName( + UIMenu.INDEX_NAME))) + + .addTags(TagAndValue.newStringTag(UIMenu.CONFIGURATION, menu.getConfigurationJson()) + .build()) + .addTags(TagAndValue.newLongTag(UIMenu.UPDATE_TIME, menu.getUpdateTime()).build()) + .build(); + this.getClient().define(property); } public UIMenu parse(Property property) { UIMenu menu = new UIMenu(); - menu.setMenuId(property.id()); + menu.setMenuId(property.getMetadata().getId()); - for (TagAndValue tagAndValue : property.tags()) { + for (BanyandbModel.Tag tag : property.getTagsList()) { + TagAndValue tagAndValue = TagAndValue.fromProtobuf(tag); if (tagAndValue.getTagName().equals(UIMenu.CONFIGURATION)) { menu.setConfigurationJson((String) tagAndValue.getValue()); } else if (tagAndValue.getTagName().equals(UIMenu.UPDATE_TIME)) { diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java index f6f6414828..9bd4ef26bf 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java @@ -19,8 +19,11 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb; import lombok.extern.slf4j.Slf4j; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon; +import org.apache.skywalking.banyandb.model.v1.BanyandbModel; +import org.apache.skywalking.banyandb.property.v1.BanyandbProperty; import org.apache.skywalking.banyandb.v1.client.TagAndValue; -import org.apache.skywalking.banyandb.v1.client.metadata.Property; +import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.Property; import org.apache.skywalking.oap.server.core.management.ui.template.UITemplate; import org.apache.skywalking.oap.server.core.query.input.DashboardSetting; import org.apache.skywalking.oap.server.core.query.type.DashboardConfiguration; @@ -64,7 +67,7 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme this.getClient().define(newTemplate); return TemplateChangeStatus.builder() .status(true) - .id(newTemplate.id()) + .id(newTemplate.getMetadata().getId()) .build(); } catch (IOException ioEx) { log.error("fail to add new template", ioEx); @@ -80,7 +83,7 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme this.getClient().define(newTemplate); return TemplateChangeStatus.builder() .status(true) - .id(newTemplate.id()) + .id(newTemplate.getMetadata().getId()) .build(); } catch (IOException ioEx) { log.error("fail to modify the template", ioEx); @@ -118,9 +121,10 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme public UITemplate parse(Property property) { UITemplate uiTemplate = new UITemplate(); - uiTemplate.setTemplateId(property.id()); + uiTemplate.setTemplateId(property.getMetadata().getId()); - for (TagAndValue tagAndValue : property.tags()) { + for (BanyandbModel.Tag tag : property.getTagsList()) { + TagAndValue tagAndValue = TagAndValue.fromProtobuf(tag); if (tagAndValue.getTagName().equals(UITemplate.CONFIGURATION)) { uiTemplate.setConfiguration((String) tagAndValue.getValue()); } else if (tagAndValue.getTagName().equals(UITemplate.DISABLED)) { @@ -133,11 +137,16 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme } public Property applyAll(UITemplate uiTemplate) { - return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id().build()) - .addTag(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration())) - .addTag(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled())) - .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime())) - .build(); + return Property.newBuilder() + .setMetadata(BanyandbProperty.Metadata.newBuilder() + .setId(uiTemplate.id().build()) + .setContainer(BanyandbCommon.Metadata.newBuilder() + .setGroup(GROUP) + .setName(UITemplate.INDEX_NAME))) + .addTags(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration()).build()) + .addTags(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled()).build()) + .addTags(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()).build()) + .build(); } /** @@ -147,10 +156,15 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme * @return new property (patch) to be applied */ public Property applyStatus(UITemplate uiTemplate) { - return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id().build()) - .addTag(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled())) - .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime())) - .build(); + return Property.newBuilder() + .setMetadata(BanyandbProperty.Metadata.newBuilder() + .setId(uiTemplate.id().build()) + .setContainer(BanyandbCommon.Metadata.newBuilder() + .setGroup(GROUP) + .setName(UITemplate.INDEX_NAME))) + .addTags(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled()).build()) + .addTags(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()).build()) + .build(); } /** @@ -160,9 +174,14 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme * @return new property (patch) to be applied */ public Property applyConfiguration(UITemplate uiTemplate) { - return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id().build()) - .addTag(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration())) - .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime())) - .build(); + return Property.newBuilder() + .setMetadata(BanyandbProperty.Metadata.newBuilder() + .setId(uiTemplate.id().build()) + .setContainer(BanyandbCommon.Metadata.newBuilder() + .setGroup(GROUP) + .setName(UITemplate.INDEX_NAME))) + .addTags(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration()).build()) + .addTags(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()).build()) + .build(); } } diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MeasureModel.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MeasureModel.java new file mode 100644 index 0000000000..09f3904839 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MeasureModel.java @@ -0,0 +1,32 @@ +/* + * 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.banyandb; + +import java.util.List; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.IndexRule; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Measure; + +@RequiredArgsConstructor +@Getter +public class MeasureModel { + private final Measure measure; + private final List indexRules; +} diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java index 86795cf496..bdf53a6761 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java @@ -48,20 +48,30 @@ import lombok.Setter; import lombok.Singular; import lombok.ToString; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.banyandb.v1.client.AbstractQuery; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Group; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Metadata; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.IntervalRule; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Catalog; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.ResourceOpts; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.IndexRule; +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.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.database.v1.BanyandbDatabase.FieldSpec; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.FieldType; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.CompressionMethod; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.EncodingMethod; +import org.apache.skywalking.banyandb.model.v1.BanyandbModel; import org.apache.skywalking.banyandb.v1.client.BanyanDBClient; import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException; -import org.apache.skywalking.banyandb.v1.client.metadata.Catalog; import org.apache.skywalking.banyandb.v1.client.metadata.Duration; -import org.apache.skywalking.banyandb.v1.client.metadata.Group; -import org.apache.skywalking.banyandb.v1.client.metadata.IndexRule; -import org.apache.skywalking.banyandb.v1.client.metadata.IntervalRule; -import org.apache.skywalking.banyandb.v1.client.metadata.Measure; -import org.apache.skywalking.banyandb.v1.client.metadata.NamedSchema; +import org.apache.skywalking.banyandb.v1.client.metadata.MetadataCache; import org.apache.skywalking.banyandb.v1.client.metadata.ResourceExist; -import org.apache.skywalking.banyandb.v1.client.metadata.Stream; -import org.apache.skywalking.banyandb.v1.client.metadata.TagFamilySpec; -import org.apache.skywalking.banyandb.v1.client.metadata.TopNAggregation; 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; @@ -86,7 +96,7 @@ public enum MetadataRegistry { private Map specificGroupSettings = new HashMap<>(); - public Stream registerStreamModel(Model model, BanyanDBStorageConfig config, ConfigService configService) { + public StreamModel registerStreamModel(Model model, BanyanDBStorageConfig config, ConfigService configService) { final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService); Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata); Map modelColumnMap = model.getColumns().stream() @@ -100,12 +110,12 @@ public enum MetadataRegistry { // this can be used to build both // 1) a list of TagFamilySpec, // 2) a list of IndexRule, - List tags = parseTagMetadata(model, schemaBuilder, shardingColumns); + List tags = parseTagMetadata(model, schemaBuilder, shardingColumns, schemaMetadata.group); List tagFamilySpecs = schemaMetadata.extractTagFamilySpec(tags, false); // iterate over tagFamilySpecs to save tag names for (final TagFamilySpec tagFamilySpec : tagFamilySpecs) { - for (final TagFamilySpec.TagSpec tagSpec : tagFamilySpec.tagSpecs()) { - schemaBuilder.tag(tagSpec.getTagName()); + for (final TagSpec tagSpec : tagFamilySpec.getTagsList()) { + schemaBuilder.tag(tagSpec.getName()); } } String timestampColumn4Stream = model.getBanyanDBModelExtension().getTimestampColumn(); @@ -119,15 +129,18 @@ public enum MetadataRegistry { .filter(Objects::nonNull) .collect(Collectors.toList()); - final Stream.Builder builder = Stream.create(schemaMetadata.getGroup(), schemaMetadata.name()); - builder.setEntityRelativeTags(shardingColumns); - builder.addTagFamilies(tagFamilySpecs); - builder.addIndexes(indexRules); + final Stream.Builder builder = Stream.newBuilder(); + builder.setMetadata(BanyandbCommon.Metadata.newBuilder().setGroup(schemaMetadata.getGroup()) + .setName(schemaMetadata.name())); + builder.setEntity(BanyandbDatabase.Entity.newBuilder().addAllTagNames(shardingColumns)); + builder.addAllTagFamilies(tagFamilySpecs); + + //builder.addIndexes(indexRules); registry.put(schemaMetadata.name(), schemaBuilder.build()); - return builder.build(); + return new StreamModel(builder.build(), indexRules); } - public Measure registerMeasureModel(Model model, BanyanDBStorageConfig config, ConfigService configService) throws StorageException { + public MeasureModel registerMeasureModel(Model model, BanyanDBStorageConfig config, ConfigService configService) throws StorageException { final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService); Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata); Map modelColumnMap = model.getColumns().stream() @@ -141,12 +154,12 @@ public enum MetadataRegistry { // this can be used to build both // 1) a list of TagFamilySpec, // 2) a list of IndexRule, - MeasureMetadata tagsAndFields = parseTagAndFieldMetadata(model, schemaBuilder, shardingColumns); + MeasureMetadata tagsAndFields = parseTagAndFieldMetadata(model, schemaBuilder, shardingColumns, schemaMetadata.group); List tagFamilySpecs = schemaMetadata.extractTagFamilySpec(tagsAndFields.tags, model.getBanyanDBModelExtension().isStoreIDTag()); // iterate over tagFamilySpecs to save tag names for (final TagFamilySpec tagFamilySpec : tagFamilySpecs) { - for (final TagFamilySpec.TagSpec tagSpec : tagFamilySpec.tagSpecs()) { - schemaBuilder.tag(tagSpec.getTagName()); + for (final TagSpec tagSpec : tagFamilySpec.getTagsList()) { + schemaBuilder.tag(tagSpec.getName()); } } List indexRules = tagsAndFields.tags.stream() @@ -155,26 +168,29 @@ public enum MetadataRegistry { .collect(Collectors.toList()); if (model.getBanyanDBModelExtension().isStoreIDTag()) { - indexRules.add(IndexRule.create(BanyanDBConverter.ID, IndexRule.IndexType.INVERTED)); + indexRules.add(indexRule(schemaMetadata.group, BanyanDBConverter.ID)); + // indexRules.add(IndexRule.create(BanyanDBConverter.ID, IndexRule.IndexType.INVERTED)); } - final Measure.Builder builder = Measure.create(schemaMetadata.getGroup(), schemaMetadata.name(), - downSamplingDuration(model.getDownsampling())); - builder.setEntityRelativeTags(shardingColumns); - builder.addTagFamilies(tagFamilySpecs); - if (!indexRules.isEmpty()) { - builder.addIndexes(indexRules); - } + final Measure.Builder builder = Measure.newBuilder(); + builder.setMetadata(BanyandbCommon.Metadata.newBuilder().setGroup(schemaMetadata.getGroup()) + .setName(schemaMetadata.name())); + builder.setInterval(downSamplingDuration(model.getDownsampling()).format()); + builder.setEntity(BanyandbDatabase.Entity.newBuilder().addAllTagNames(shardingColumns)); + builder.addAllTagFamilies(tagFamilySpecs); +// if (!indexRules.isEmpty()) { +// builder.addIndexes(indexRules); +// } // parse and set field - for (Measure.FieldSpec field : tagsAndFields.fields) { - builder.addField(field); + for (BanyandbDatabase.FieldSpec field : tagsAndFields.fields) { + builder.addFields(field); schemaBuilder.field(field.getName()); } // parse TopN schemaBuilder.topNSpec(parseTopNSpec(model, schemaMetadata.name())); registry.put(schemaMetadata.name(), schemaBuilder.build()); - return builder.build(); + return new MeasureModel(builder.build(), indexRules); } private TopNSpec parseTopNSpec(final Model model, final String measureName) @@ -198,7 +214,7 @@ public enum MetadataRegistry { .countersNumber(model.getBanyanDBModelExtension().getTopN().getCountersNumber()) .fieldName(valueColumnOpt.get().getValueCName()) .groupByTagNames(model.getBanyanDBModelExtension().getTopN().getGroupByTagNames()) - .sort(AbstractQuery.Sort.UNSPECIFIED) // include both TopN and BottomN + .sort(BanyandbModel.Sort.SORT_UNSPECIFIED) // include both TopN and BottomN .build(); } @@ -245,27 +261,31 @@ public enum MetadataRegistry { return this.registry.get(SchemaMetadata.formatName(modelName, downSampling)); } - private Measure.FieldSpec parseFieldSpec(ModelColumn modelColumn) { + private FieldSpec parseFieldSpec(ModelColumn modelColumn) { String colName = modelColumn.getColumnName().getStorageName(); if (String.class.equals(modelColumn.getType())) { - return Measure.FieldSpec.newIntField(colName) - .compressWithZSTD() - .build(); + return FieldSpec.newBuilder().setName(colName) + .setFieldType(FieldType.FIELD_TYPE_STRING) + .setCompressionMethod(CompressionMethod.COMPRESSION_METHOD_ZSTD) + .build(); } else if (long.class.equals(modelColumn.getType()) || int.class.equals(modelColumn.getType())) { - return Measure.FieldSpec.newIntField(colName) - .compressWithZSTD() - .encodeWithGorilla() - .build(); + return FieldSpec.newBuilder().setName(colName) + .setFieldType(FieldType.FIELD_TYPE_INT) + .setCompressionMethod(CompressionMethod.COMPRESSION_METHOD_ZSTD) + .setEncodingMethod(EncodingMethod.ENCODING_METHOD_GORILLA) + .build(); } else if (StorageDataComplexObject.class.isAssignableFrom(modelColumn.getType()) || JsonObject.class.equals(modelColumn.getType())) { - return Measure.FieldSpec.newStringField(colName) - .compressWithZSTD() - .build(); + return FieldSpec.newBuilder().setName(colName) + .setFieldType(FieldType.FIELD_TYPE_STRING) + .setCompressionMethod(CompressionMethod.COMPRESSION_METHOD_ZSTD) + .build(); } else if (double.class.equals(modelColumn.getType())) { // TODO: natively support double/float in BanyanDB log.warn("Double is stored as binary"); - return Measure.FieldSpec.newBinaryField(colName) - .compressWithZSTD() - .build(); + return FieldSpec.newBuilder().setName(colName) + .setFieldType(FieldType.FIELD_TYPE_DATA_BINARY) + .setCompressionMethod(CompressionMethod.COMPRESSION_METHOD_ZSTD) + .build(); } else { throw new UnsupportedOperationException(modelColumn.getType().getSimpleName() + " is not supported for field"); } @@ -284,8 +304,11 @@ public enum MetadataRegistry { } } - IndexRule indexRule(String tagName) { - return IndexRule.create(tagName, IndexRule.IndexType.INVERTED); + IndexRule indexRule(String group, String tagName) { + return IndexRule.newBuilder() + .setMetadata(Metadata.newBuilder().setName(tagName).setGroup(group)) + .setType(IndexRule.Type.TYPE_INVERTED).addTags(tagName).build(); + //return IndexRule.create(tagName, IndexRule.IndexType.INVERTED); } /** @@ -314,18 +337,18 @@ public enum MetadataRegistry { * * @since 9.4.0 Skip {@link Record#TIME_BUCKET} */ - List parseTagMetadata(Model model, Schema.SchemaBuilder builder, List shardingColumns) { + List parseTagMetadata(Model model, Schema.SchemaBuilder builder, List shardingColumns, String group) { List tagMetadataList = new ArrayList<>(); for (final ModelColumn col : model.getColumns()) { final String columnStorageName = col.getColumnName().getStorageName(); if (columnStorageName.equals(Record.TIME_BUCKET)) { continue; } - final TagFamilySpec.TagSpec tagSpec = parseTagSpec(col); + final TagSpec tagSpec = parseTagSpec(col); builder.spec(columnStorageName, new ColumnSpec(ColumnType.TAG, col.getType())); String colName = col.getColumnName().getStorageName(); if (!shardingColumns.contains(colName) && col.getBanyanDBExtension().shouldIndex()) { - tagMetadataList.add(new TagMetadata(indexRule(tagSpec.getTagName()), tagSpec)); + tagMetadataList.add(new TagMetadata(indexRule(group, tagSpec.getName()), tagSpec)); } else { tagMetadataList.add(new TagMetadata(null, tagSpec)); } @@ -339,7 +362,7 @@ public enum MetadataRegistry { @Singular private final List tags; @Singular - private final List fields; + private final List fields; } /** @@ -349,7 +372,7 @@ public enum MetadataRegistry { * * @since 9.4.0 Skip {@link Metrics#TIME_BUCKET} */ - MeasureMetadata parseTagAndFieldMetadata(Model model, Schema.SchemaBuilder builder, List shardingColumns) { + MeasureMetadata parseTagAndFieldMetadata(Model model, Schema.SchemaBuilder builder, List shardingColumns, String group) { // skip metric MeasureMetadata.MeasureMetadataBuilder result = MeasureMetadata.builder(); for (final ModelColumn col : model.getColumns()) { @@ -362,10 +385,10 @@ public enum MetadataRegistry { result.field(parseFieldSpec(col)); continue; } - final TagFamilySpec.TagSpec tagSpec = parseTagSpec(col); + final TagSpec tagSpec = parseTagSpec(col); builder.spec(columnStorageName, new ColumnSpec(ColumnType.TAG, col.getType())); String colName = col.getColumnName().getStorageName(); - result.tag(new TagMetadata(!shardingColumns.contains(colName) && col.getBanyanDBExtension().shouldIndex() ? indexRule(tagSpec.getTagName()) : null, tagSpec)); + result.tag(new TagMetadata(!shardingColumns.contains(colName) && col.getBanyanDBExtension().shouldIndex() ? indexRule(group, tagSpec.getName()) : null, tagSpec)); } return result.build(); @@ -378,36 +401,36 @@ public enum MetadataRegistry { * @return a typed tag spec */ @Nonnull - private TagFamilySpec.TagSpec parseTagSpec(ModelColumn modelColumn) { + private TagSpec parseTagSpec(ModelColumn modelColumn) { final Class clazz = modelColumn.getType(); final String colName = modelColumn.getColumnName().getStorageName(); - TagFamilySpec.TagSpec tagSpec = null; + TagSpec.Builder tagSpec = TagSpec.newBuilder().setName(colName); if (String.class.equals(clazz) || StorageDataComplexObject.class.isAssignableFrom(clazz) || JsonObject.class.equals(clazz)) { - tagSpec = TagFamilySpec.TagSpec.newStringTag(colName); + tagSpec = tagSpec.setType(TagType.TAG_TYPE_STRING); } else if (int.class.equals(clazz) || long.class.equals(clazz)) { - tagSpec = TagFamilySpec.TagSpec.newIntTag(colName); + tagSpec = tagSpec.setType(TagType.TAG_TYPE_INT); } else if (byte[].class.equals(clazz)) { - tagSpec = TagFamilySpec.TagSpec.newBinaryTag(colName); + tagSpec = tagSpec.setType(TagType.TAG_TYPE_DATA_BINARY); } else if (clazz.isEnum()) { - tagSpec = TagFamilySpec.TagSpec.newIntTag(colName); + tagSpec = tagSpec.setType(TagType.TAG_TYPE_INT); } else if (double.class.equals(clazz) || Double.class.equals(clazz)) { // serialize double as binary - tagSpec = TagFamilySpec.TagSpec.newBinaryTag(colName); + tagSpec = tagSpec.setType(TagType.TAG_TYPE_DATA_BINARY); } else if (IntList.class.isAssignableFrom(clazz)) { - tagSpec = TagFamilySpec.TagSpec.newIntArrayTag(colName); + tagSpec = tagSpec.setType(TagType.TAG_TYPE_INT_ARRAY); } else if (List.class.isAssignableFrom(clazz)) { // handle exceptions ParameterizedType t = (ParameterizedType) modelColumn.getGenericType(); if (String.class.equals(t.getActualTypeArguments()[0])) { - tagSpec = TagFamilySpec.TagSpec.newStringArrayTag(colName); + tagSpec = tagSpec.setType(TagType.TAG_TYPE_STRING_ARRAY); } - } - if (tagSpec == null) { + } else { throw new IllegalStateException("type " + modelColumn.getType().toString() + " is not supported"); } + if (modelColumn.isIndexOnly()) { - tagSpec.indexedOnly(); + tagSpec.setIndexedOnly(true); } - return tagSpec; + return tagSpec.build(); } public void initializeIntervals(String specificGroupSettingsStr) { @@ -507,7 +530,7 @@ public enum MetadataRegistry { return modelName + "_" + downSampling.getName(); } - public Optional> findRemoteSchema(BanyanDBClient client) throws BanyanDBException { + public Optional findRemoteSchema(BanyanDBClient client) throws BanyanDBException { try { switch (kind) { case STREAM: @@ -526,6 +549,17 @@ public enum MetadataRegistry { } } + public MetadataCache.EntityMetadata updateRemoteSchema(BanyanDBClient client) throws BanyanDBException { + switch (kind) { + case STREAM: + return client.updateStreamMetadataCacheFromSever(this.group, this.name()); + case MEASURE: + return client.updateMeasureMetadataCacheFromSever(this.group, this.name()); + default: + throw new IllegalStateException("should not reach here"); + } + } + private List extractTagFamilySpec(List tagMetadataList, boolean shouldAddID) { final String indexFamily = SchemaMetadata.this.indexFamily(); final String nonIndexFamily = SchemaMetadata.this.nonIndexFamily(); @@ -534,10 +568,11 @@ public enum MetadataRegistry { final List tagFamilySpecs = new ArrayList<>(tagMetadataMap.size()); for (final Map.Entry> entry : tagMetadataMap.entrySet()) { - final TagFamilySpec.Builder b = TagFamilySpec.create(entry.getKey()) - .addTagSpecs(entry.getValue().stream().map(TagMetadata::getTagSpec).collect(Collectors.toList())); + final TagFamilySpec.Builder b = TagFamilySpec.newBuilder(); + b.setName(entry.getKey()); + b.addAllTags(entry.getValue().stream().map(TagMetadata::getTagSpec).collect(Collectors.toList())); if (shouldAddID && indexFamily.equals(entry.getKey())) { - b.addTagSpec(TagFamilySpec.TagSpec.newStringTag(BanyanDBConverter.ID)); + b.addTags(TagSpec.newBuilder().setType(TagType.TAG_TYPE_STRING).setName(BanyanDBConverter.ID)); } tagFamilySpecs.add(b.build()); } @@ -547,19 +582,31 @@ public enum MetadataRegistry { public boolean checkResourceExistence(BanyanDBClient client) throws BanyanDBException { ResourceExist resourceExist; + Group.Builder gBuilder + = Group.newBuilder() + .setMetadata(Metadata.newBuilder().setName(this.group)) + .setResourceOpts(ResourceOpts.newBuilder() + .setShardNum(this.shard) + .setSegmentInterval( + IntervalRule.newBuilder() + .setUnit( + IntervalRule.Unit.UNIT_DAY) + .setNum( + this.segmentIntervalDays)) + .setTtl( + IntervalRule.newBuilder() + .setUnit( + IntervalRule.Unit.UNIT_DAY) + .setNum( + this.ttlDays))); switch (kind) { case STREAM: resourceExist = client.existStream(this.group, this.name()); if (!resourceExist.hasGroup()) { try { - Group g = client.define(Group.create(this.group, Catalog.STREAM, this.shard, - IntervalRule.create( - IntervalRule.Unit.DAY, this.segmentIntervalDays), - IntervalRule.create( - IntervalRule.Unit.DAY, this.ttlDays) - )); + Group g = client.define(gBuilder.setCatalog(Catalog.CATALOG_STREAM).build()); if (g != null) { - log.info("group {} created", g.name()); + log.info("group {} created", g.getMetadata().getName()); } } catch (BanyanDBException ex) { if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) { @@ -574,14 +621,15 @@ public enum MetadataRegistry { resourceExist = client.existMeasure(this.group, this.name()); try { if (!resourceExist.hasGroup()) { - Group g = client.define(Group.create(this.group, Catalog.MEASURE, this.shard, - IntervalRule.create( - IntervalRule.Unit.DAY, this.segmentIntervalDays), - IntervalRule.create( - IntervalRule.Unit.DAY, this.ttlDays) - )); + Group g = client.define(gBuilder.setCatalog(Catalog.CATALOG_MEASURE).build()); +// Group.create(this.group, Catalog.MEASURE, this.shard, +// IntervalRule.create( +// IntervalRule.Unit.DAY, this.segmentIntervalDays), +// IntervalRule.create( +// IntervalRule.Unit.DAY, this.ttlDays) +// )); if (g != null) { - log.info("group {} created", g.name()); + log.info("group {} created", g.getMetadata().getName()); } } } catch (BanyanDBException ex) { @@ -637,7 +685,7 @@ public enum MetadataRegistry { @Getter private static class TagMetadata { private final IndexRule indexRule; - private final TagFamilySpec.TagSpec tagSpec; + private final TagSpec tagSpec; boolean isIndex() { return this.indexRule != null; @@ -679,14 +727,29 @@ public enum MetadataRegistry { } return; } - client.define(TopNAggregation.create(getMetadata().getGroup(), this.getTopNSpec().getName()) - .setSourceMeasureName(getMetadata().name()) - .setFieldValueSort(this.getTopNSpec().getSort()) - .setFieldName(this.getTopNSpec().getFieldName()) - .setGroupByTagNames(this.getTopNSpec().getGroupByTagNames()) - .setCountersNumber(this.getTopNSpec().getCountersNumber()) - .setLruSize(this.getTopNSpec().getLruSize()) - .build()); + TopNAggregation.Builder builder + = TopNAggregation.newBuilder() + .setMetadata(Metadata.newBuilder() + .setGroup(getMetadata().getGroup()) + .setName(this.getTopNSpec().getName())) + + .setSourceMeasure(Metadata.newBuilder() + .setGroup(getMetadata().getGroup()) + .setName(getMetadata().name())) + .setFieldValueSort(this.getTopNSpec().getSort()) + .setFieldName(this.getTopNSpec().getFieldName()) + .addAllGroupByTagNames(this.getTopNSpec().getGroupByTagNames()) + .setCountersNumber(this.getTopNSpec().getCountersNumber()) + .setLruSize(this.getTopNSpec().getLruSize()); + client.define(builder.build()); +// client.define(TopNAggregation.create(getMetadata().getGroup(), this.getTopNSpec().getName()) +// .setSourceMeasureName(getMetadata().name()) +// .setFieldValueSort(this.getTopNSpec().getSort()) +// .setFieldName(this.getTopNSpec().getFieldName()) +// .setGroupByTagNames(this.getTopNSpec().getGroupByTagNames()) +// .setCountersNumber(this.getTopNSpec().getCountersNumber()) +// .setLruSize(this.getTopNSpec().getLruSize()) +// .build()); log.info("installed TopN schema for measure {}", getMetadata().name()); } } @@ -700,7 +763,7 @@ public enum MetadataRegistry { @Singular private final List groupByTagNames; private final String fieldName; - private final AbstractQuery.Sort sort; + private final BanyandbModel.Sort sort; private final int lruSize; private final int countersNumber; } diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/StreamModel.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/StreamModel.java new file mode 100644 index 0000000000..ce163115af --- /dev/null +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/StreamModel.java @@ -0,0 +1,32 @@ +/* + * 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.banyandb; + +import java.util.List; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Stream; +import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.IndexRule; + +@RequiredArgsConstructor +@Getter +public class StreamModel { + private final Stream stream; + private final List indexRules; +} diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBContinuousProfilingPolicyDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBContinuousProfilingPolicyDAO.java index c37160261d..52e69a4574 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBContinuousProfilingPolicyDAO.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBContinuousProfilingPolicyDAO.java @@ -19,8 +19,11 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream; import lombok.extern.slf4j.Slf4j; +import org.apache.skywalking.banyandb.common.v1.BanyandbCommon; +import org.apache.skywalking.banyandb.model.v1.BanyandbModel; +import org.apache.skywalking.banyandb.property.v1.BanyandbProperty; import org.apache.skywalking.banyandb.v1.client.TagAndValue; -import org.apache.skywalking.banyandb.v1.client.metadata.Property; +import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.Property; import org.apache.skywalking.oap.server.core.profiling.continuous.storage.ContinuousProfilingPolicy; import org.apache.skywalking.oap.server.core.storage.profiling.continuous.IContinuousProfilingPolicyDAO; import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient; @@ -48,10 +51,15 @@ public class BanyanDBContinuousProfilingPolicyDAO extends AbstractBanyanDBDAO im } public Property applyAll(ContinuousProfilingPolicy policy) { - return Property.create(GROUP, ContinuousProfilingPolicy.INDEX_NAME, policy.id().build()) - .addTag(TagAndValue.newStringTag(ContinuousProfilingPolicy.UUID, policy.getUuid())) - .addTag(TagAndValue.newStringTag(ContinuousProfilingPolicy.CONFIGURATION_JSON, policy.getConfigurationJson())) - .build(); + return Property.newBuilder() + .setMetadata(BanyandbProperty.Metadata.newBuilder() + .setId(policy.id().build()) + .setContainer(BanyandbCommon.Metadata.newBuilder() + .setGroup(GROUP) + .setName(ContinuousProfilingPolicy.INDEX_NAME))) + .addTags(TagAndValue.newStringTag(ContinuousProfilingPolicy.UUID, policy.getUuid()).build()) + .addTags(TagAndValue.newStringTag(ContinuousProfilingPolicy.CONFIGURATION_JSON, policy.getConfigurationJson()).build()) + .build(); } @Override @@ -65,16 +73,16 @@ public class BanyanDBContinuousProfilingPolicyDAO extends AbstractBanyanDBDAO im } }).filter(Objects::nonNull).map(properties -> { final ContinuousProfilingPolicy policy = new ContinuousProfilingPolicy(); - policy.setServiceId(properties.id()); - for (TagAndValue tag : properties.tags()) { - if (tag.getTagName().equals(ContinuousProfilingPolicy.CONFIGURATION_JSON)) { - policy.setConfigurationJson((String) tag.getValue()); - } else if (tag.getTagName().equals(ContinuousProfilingPolicy.UUID)) { - policy.setUuid((String) tag.getValue()); + policy.setServiceId(properties.getMetadata().getId()); + for (BanyandbModel.Tag tag : properties.getTagsList()) { + TagAndValue tagAndValue = TagAndValue.fromProtobuf(tag); + if (tagAndValue.getTagName().equals(ContinuousProfilingPolicy.CONFIGURATION_JSON)) { + policy.setConfigurationJson((String) tagAndValue.getValue()); + } else if (tagAndValue.getTagName().equals(ContinuousProfilingPolicy.UUID)) { + policy.setUuid((String) tagAndValue.getValue()); } } return policy; }).collect(Collectors.toList()); } - -} \ No newline at end of file +} diff --git a/test/e2e-v2/script/env b/test/e2e-v2/script/env index 4dcf1af099..8631e0da5f 100644 --- a/test/e2e-v2/script/env +++ b/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=1335f15bf821a40a7cd71448fa805f0be265afcc SW_ROVER_COMMIT=6bbd39aa701984482330d9dfb4dbaaff0527d55c -SW_BANYANDB_COMMIT=59c396870ac2d81ec81113802d54277fe070d91b +SW_BANYANDB_COMMIT=0e734c462571dcf55dbb7761211c07d8b156521e SW_AGENT_PHP_COMMIT=3192c553002707d344bd6774cfab5bc61f67a1d3 SW_CTL_COMMIT=d5f3597733aa5217373986d776a3ee5ee8b3c468