You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2025/03/18 13:41:48 UTC
(skywalking) branch banyandb-property updated: Bump up banyandb client
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch banyandb-property
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/banyandb-property by this push:
new ca528d9845 Bump up banyandb client
new 8e5f23c2ea Merge branch 'banyandb-property' of github.com:apache/skywalking into banyandb-property
ca528d9845 is described below
commit ca528d984506faee1276232f4ebd31eaba9db40c
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Mar 18 19:20:30 2025 +0800
Bump up banyandb client
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
oap-server-bom/pom.xml | 2 +-
.../plugin/banyandb/BanyanDBIndexInstaller.java | 252 +++++++++++++--------
.../plugin/banyandb/BanyanDBStorageClient.java | 8 +-
.../plugin/banyandb/BanyanDBStorageProvider.java | 10 -
.../banyandb/BanyanDBUIMenuManagementDAO.java | 15 +-
.../banyandb/BanyanDBUITemplateManagementDAO.java | 38 ++--
.../storage/plugin/banyandb/MetadataRegistry.java | 22 +-
.../storage/plugin/banyandb/PropertyModel.java | 11 +
.../BanyanDBContinuousProfilingPolicyDAO.java | 14 +-
9 files changed, 225 insertions(+), 147 deletions(-)
diff --git a/oap-server-bom/pom.xml b/oap-server-bom/pom.xml
index 71b927ba1c..97ecaf5416 100644
--- a/oap-server-bom/pom.xml
+++ b/oap-server-bom/pom.xml
@@ -72,7 +72,7 @@
<httpcore.version>4.4.16</httpcore.version>
<httpasyncclient.version>4.1.5</httpasyncclient.version>
<commons-compress.version>1.21</commons-compress.version>
- <banyandb-java-client.version>0.8.0-rc2</banyandb-java-client.version>
+ <banyandb-java-client.version>0.8.0-SNAPSHOT</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>
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 806903b99f..8a8cebd144 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
@@ -32,6 +32,7 @@ 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.IntervalRule;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
+import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Property;
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.IndexRule;
@@ -68,11 +69,6 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
public InstallInfo isExists(Model model) throws StorageException {
InstallInfoBanyanDB installInfo = new InstallInfoBanyanDB(model);
installInfo.setDownSampling(model.getDownsampling());
- if (!model.isTimeSeries()) {
- installInfo.setTableName(model.getName());
- installInfo.setAllExist(true);
- return installInfo;
- }
final DownSamplingConfigService downSamplingConfigService = moduleManager.find(CoreModule.NAME)
.provider()
.getService(DownSamplingConfigService.class);
@@ -91,36 +87,43 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
installInfo.setAllExist(false);
return installInfo;
} else {
- // register models only locally(Schema cache) but not remotely
- if (model.isRecord()) { // stream
- StreamModel streamModel = MetadataRegistry.INSTANCE.registerStreamModel(
- model, config, downSamplingConfigService);
- if (!RunningMode.isNoInitMode()) {
- checkStream(streamModel.getStream(), c);
- checkIndexRules(model.getName(), streamModel.getIndexRules(), c);
- checkIndexRuleBinding(
- streamModel.getIndexRules(), metadata.getGroup(), metadata.name(),
- BanyandbCommon.Catalog.CATALOG_STREAM, c
- );
- // Stream not support server side TopN pre-aggregation
+ if (model.isTimeSeries()) {
+ // register models only locally(Schema cache) but not remotely
+ if (model.isRecord()) { // stream
+ StreamModel streamModel = MetadataRegistry.INSTANCE.registerStreamModel(
+ model, config, downSamplingConfigService);
+ if (!RunningMode.isNoInitMode()) {
+ checkStream(streamModel.getStream(), c);
+ checkIndexRules(model.getName(), streamModel.getIndexRules(), c);
+ checkIndexRuleBinding(
+ streamModel.getIndexRules(), metadata.getGroup(), metadata.name(),
+ BanyandbCommon.Catalog.CATALOG_STREAM, c
+ );
+ // Stream not support server side TopN pre-aggregation
+ }
+ } else { // measure
+ MeasureModel measureModel = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, downSamplingConfigService);
+ if (!RunningMode.isNoInitMode()) {
+ checkMeasure(measureModel.getMeasure(), c);
+ checkIndexRules(model.getName(), measureModel.getIndexRules(), c);
+ checkIndexRuleBinding(
+ measureModel.getIndexRules(), metadata.getGroup(), metadata.name(),
+ BanyandbCommon.Catalog.CATALOG_MEASURE, c
+ );
+ checkTopNAggregation(model, c);
+ }
}
- } else { // measure
- MeasureModel measureModel = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, downSamplingConfigService);
+ // pre-load remote schema for java client
+ MetadataCache.EntityMetadata remoteMeta = updateSchemaFromServer(metadata, c);
+ if (remoteMeta == null) {
+ throw new IllegalStateException("inconsistent state: metadata:" + metadata + ", remoteMeta: null");
+ }
+ } else {
+ PropertyModel propertyModel = MetadataRegistry.INSTANCE.registerPropertyModel(model);
if (!RunningMode.isNoInitMode()) {
- checkMeasure(measureModel.getMeasure(), c);
- checkIndexRules(model.getName(), measureModel.getIndexRules(), c);
- checkIndexRuleBinding(
- measureModel.getIndexRules(), metadata.getGroup(), metadata.name(),
- BanyandbCommon.Catalog.CATALOG_MEASURE, c
- );
- checkTopNAggregation(model, c);
+ checkProperty(propertyModel.getProperty(), c);
}
}
- // pre-load remote schema for java client
- MetadataCache.EntityMetadata remoteMeta = updateSchemaFromServer(metadata, c);
- if (remoteMeta == null) {
- throw new IllegalStateException("inconsistent state: metadata:" + metadata + ", remoteMeta: null");
- }
installInfo.setAllExist(true);
return installInfo;
}
@@ -135,73 +138,89 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
DownSamplingConfigService configService = moduleManager.find(CoreModule.NAME)
.provider()
.getService(DownSamplingConfigService.class);
- if (model.isRecord()) { // stream
- 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 {
- client.define(stream);
- if (CollectionUtils.isNotEmpty(streamModel.getIndexRules())) {
- for (IndexRule indexRule : streamModel.getIndexRules()) {
- defineIndexRule(model.getName(), indexRule, client);
+ if (model.isTimeSeries()) {
+ if (model.isRecord()) { // stream
+ 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 {
+ client.define(stream);
+ if (CollectionUtils.isNotEmpty(streamModel.getIndexRules())) {
+ for (IndexRule indexRule : streamModel.getIndexRules()) {
+ defineIndexRule(model.getName(), indexRule, client);
+ }
+ defineIndexRuleBinding(
+ streamModel.getIndexRules(), stream.getMetadata().getGroup(), stream.getMetadata().getName(),
+ BanyandbCommon.Catalog.CATALOG_STREAM, client
+ );
+ }
+ } catch (BanyanDBException ex) {
+ if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
+ log.info(
+ "Stream schema {}_{} already created by another OAP node",
+ model.getName(),
+ model.getDownsampling()
+ );
+ } else {
+ throw ex;
}
- defineIndexRuleBinding(
- streamModel.getIndexRules(), stream.getMetadata().getGroup(), stream.getMetadata().getName(),
- BanyandbCommon.Catalog.CATALOG_STREAM, client
- );
- }
- } catch (BanyanDBException ex) {
- if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
- log.info(
- "Stream schema {}_{} already created by another OAP node",
- model.getName(),
- model.getDownsampling()
- );
- } else {
- throw ex;
}
}
- }
- } else { // measure
- MeasureModel measureModel = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
- Measure measure = measureModel.getMeasure();
- if (measure != null) {
- log.info("install measure schema {}", model.getName());
- final BanyanDBClient client = ((BanyanDBStorageClient) this.client).client;
- try {
- client.define(measure);
- if (CollectionUtils.isNotEmpty(measureModel.getIndexRules())) {
- for (IndexRule indexRule : measureModel.getIndexRules()) {
- defineIndexRule(model.getName(), indexRule, client);
+ } else { // measure
+ MeasureModel measureModel = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
+ Measure measure = measureModel.getMeasure();
+ if (measure != null) {
+ log.info("install measure schema {}", model.getName());
+ final BanyanDBClient client = ((BanyanDBStorageClient) this.client).client;
+ try {
+ client.define(measure);
+ if (CollectionUtils.isNotEmpty(measureModel.getIndexRules())) {
+ for (IndexRule indexRule : measureModel.getIndexRules()) {
+ defineIndexRule(model.getName(), indexRule, client);
+ }
+ defineIndexRuleBinding(
+ measureModel.getIndexRules(), measure.getMetadata().getGroup(), measure.getMetadata().getName(),
+ BanyandbCommon.Catalog.CATALOG_MEASURE, client
+ );
+ }
+ } catch (BanyanDBException ex) {
+ if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
+ log.info("Measure schema {}_{} already created by another OAP node",
+ model.getName(),
+ model.getDownsampling());
+ } else {
+ throw ex;
}
- defineIndexRuleBinding(
- measureModel.getIndexRules(), measure.getMetadata().getGroup(), measure.getMetadata().getName(),
- BanyandbCommon.Catalog.CATALOG_MEASURE, client
- );
}
- } catch (BanyanDBException ex) {
- if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
- log.info("Measure schema {}_{} already created by another OAP node",
- model.getName(),
- model.getDownsampling());
- } else {
- throw ex;
+ final MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
+ try {
+ defineTopNAggregation(schema, client);
+ } catch (BanyanDBException ex) {
+ if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
+ log.info("Measure schema {}_{} TopN({}) already created by another OAP node",
+ model.getName(),
+ model.getDownsampling(),
+ schema.getTopNSpec());
+ } else {
+ throw ex;
+ }
}
}
- final MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
- try {
- defineTopNAggregation(schema, client);
- } catch (BanyanDBException ex) {
- if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
- log.info("Measure schema {}_{} TopN({}) already created by another OAP node",
- model.getName(),
- model.getDownsampling(),
- schema.getTopNSpec());
- } else {
- throw ex;
- }
+ }
+ } else {
+ PropertyModel propertyModel = MetadataRegistry.INSTANCE.registerPropertyModel(model);
+ Property property = propertyModel.getProperty();
+ log.info("install property schema {}", model.getName());
+ final BanyanDBClient client = ((BanyanDBStorageClient) this.client).client;
+ try {
+ client.define(property);
+ } catch (BanyanDBException ex) {
+ if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
+ log.info("Property schema {} already created by another OAP node", model.getName());
+ } else {
+ throw ex;
}
}
}
@@ -240,18 +259,49 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
IntervalRule.Unit.UNIT_DAY)
.setNum(
metadata.getTtlDays())));
+ BanyandbCommon.ResourceOpts.Builder optsBuilder = BanyandbCommon.ResourceOpts.newBuilder().setShardNum(metadata.getShard());
+
switch (metadata.getKind()) {
case STREAM:
+ optsBuilder.setSegmentInterval(
+ IntervalRule.newBuilder()
+ .setUnit(
+ IntervalRule.Unit.UNIT_DAY)
+ .setNum(
+ metadata.getSegmentIntervalDays()))
+ .setTtl(
+ IntervalRule.newBuilder()
+ .setUnit(
+ IntervalRule.Unit.UNIT_DAY)
+ .setNum(
+ metadata.getTtlDays()));
resourceExist = client.existStream(metadata.getGroup(), metadata.name());
gBuilder.setCatalog(BanyandbCommon.Catalog.CATALOG_STREAM).build();
break;
case MEASURE:
+ optsBuilder.setSegmentInterval(
+ IntervalRule.newBuilder()
+ .setUnit(
+ IntervalRule.Unit.UNIT_DAY)
+ .setNum(
+ metadata.getSegmentIntervalDays()))
+ .setTtl(
+ IntervalRule.newBuilder()
+ .setUnit(
+ IntervalRule.Unit.UNIT_DAY)
+ .setNum(
+ metadata.getTtlDays()));
resourceExist = client.existMeasure(metadata.getGroup(), metadata.name());
gBuilder.setCatalog(BanyandbCommon.Catalog.CATALOG_MEASURE).build();
break;
+ case Property:
+ resourceExist = client.existProperty(metadata.getGroup(), metadata.name());
+ gBuilder.setCatalog(BanyandbCommon.Catalog.CATALOG_PROPERTY).build();
+ break;
default:
throw new IllegalStateException("unknown metadata kind: " + metadata.getKind());
}
+ gBuilder.setResourceOpts(optsBuilder.build());
if (!RunningMode.isNoInitMode()) {
if (!groupAligned.contains(metadata.getGroup())) {
// create the group if not exist
@@ -431,6 +481,26 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
}
}
+ /**
+ * Check if the property exists and update it if necessary
+ */
+ private void checkProperty(Property property, BanyanDBClient client) throws BanyanDBException {
+ Property hisProperty = client.findPropertyDefinition(property.getMetadata().getGroup(), property.getMetadata().getName());
+ if (hisProperty == null) {
+ throw new IllegalStateException("Property: " + property.getMetadata().getName() + " exist but can't find it from BanyanDB server");
+ } else {
+ boolean equals = hisProperty.toBuilder()
+ .clearUpdatedAt()
+ .clearMetadata()
+ .build()
+ .equals(property.toBuilder().clearUpdatedAt().clearMetadata().build());
+ if (!equals) {
+ client.update(property);
+ log.info("update Property: {} from: {} to: {}", hisProperty.getMetadata().getName(), hisProperty, property);
+ }
+ }
+ }
+
/**
* Check if the index rules exist and update them if necessary
*/
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 b2fb795691..9bca7b35a1 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
@@ -128,7 +128,7 @@ public class BanyanDBStorageClient implements Client, HealthCheckable {
BanyandbProperty.QueryResponse resp
= this.client.query(BanyandbProperty.QueryRequest.newBuilder()
.addGroups(group)
- .setContainer(name)
+ .setName(name)
.setLimit(Integer.MAX_VALUE)
.build());
this.healthChecker.health();
@@ -148,7 +148,7 @@ public class BanyanDBStorageClient implements Client, HealthCheckable {
try {
BanyandbProperty.QueryResponse resp = this.client.query(BanyandbProperty.QueryRequest.newBuilder()
.addGroups(group)
- .setContainer(name)
+ .setName(name)
.addIds(id)
.build());
this.healthChecker.health();
@@ -225,7 +225,7 @@ public class BanyanDBStorageClient implements Client, HealthCheckable {
/**
* PropertyStore.Strategy is default to {@link Strategy#STRATEGY_MERGE}
*/
- public void define(Property property) throws IOException {
+ public void apply(Property property) throws IOException {
try {
this.client.apply(property);
this.healthChecker.health();
@@ -235,7 +235,7 @@ public class BanyanDBStorageClient implements Client, HealthCheckable {
}
}
- public void define(Property property, Strategy strategy) throws IOException {
+ public void apply(Property property, Strategy strategy) throws IOException {
try {
this.client.apply(property, strategy);
this.healthChecker.health();
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 5b6392e581..4a775f0153 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
@@ -206,16 +206,6 @@ public class BanyanDBStorageProvider extends ModuleProvider {
this.client.registerChecker(healthChecker);
try {
this.client.connect();
- this.client.defineIfEmpty(BanyandbCommon.Group.newBuilder()
- .setMetadata(
- BanyandbCommon.Metadata.newBuilder()
- .setName(
- BanyanDBUITemplateManagementDAO.GROUP))
- .setCatalog(BanyandbCommon.Catalog.CATALOG_PROPERTY)
- .setResourceOpts(BanyandbCommon.ResourceOpts.newBuilder()
- .setShardNum(1)
- .build())
- .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 74e7cc75dd..d610779b90 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
@@ -21,7 +21,6 @@ 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.property.v1.BanyandbProperty.Property;
import org.apache.skywalking.oap.server.core.management.ui.menu.UIMenu;
@@ -50,23 +49,19 @@ public class BanyanDBUIMenuManagementDAO extends AbstractBanyanDBDAO implements
@Override
public void saveMenu(UIMenu menu) throws IOException {
Property property = Property.newBuilder()
- .setMetadata(BanyandbProperty.Metadata.newBuilder().setId(menu.getMenuId())
- .setContainer(
- BanyandbCommon.Metadata.newBuilder()
- .setGroup(GROUP)
- .setName(
- UIMenu.INDEX_NAME)))
-
+ .setMetadata(
+ BanyandbCommon.Metadata.newBuilder().setGroup(GROUP).setName(UIMenu.INDEX_NAME))
+ .setId(menu.getMenuId())
.addTags(TagAndValue.newStringTag(UIMenu.CONFIGURATION, menu.getConfigurationJson())
.build())
.addTags(TagAndValue.newLongTag(UIMenu.UPDATE_TIME, menu.getUpdateTime()).build())
.build();
- this.getClient().define(property);
+ this.getClient().apply(property);
}
public UIMenu parse(Property property) {
UIMenu menu = new UIMenu();
- menu.setMenuId(property.getMetadata().getId());
+ menu.setMenuId(property.getId());
for (BanyandbModel.Tag tag : property.getTagsList()) {
TagAndValue<?> tagAndValue = TagAndValue.fromProtobuf(tag);
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 9bd4ef26bf..8b19231713 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
@@ -21,7 +21,6 @@ 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.property.v1.BanyandbProperty.Property;
import org.apache.skywalking.oap.server.core.management.ui.template.UITemplate;
@@ -64,10 +63,10 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
public TemplateChangeStatus addTemplate(DashboardSetting setting) {
Property newTemplate = applyAll(setting.toEntity());
try {
- this.getClient().define(newTemplate);
+ this.getClient().apply(newTemplate);
return TemplateChangeStatus.builder()
.status(true)
- .id(newTemplate.getMetadata().getId())
+ .id(newTemplate.getId())
.build();
} catch (IOException ioEx) {
log.error("fail to add new template", ioEx);
@@ -80,10 +79,10 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
public TemplateChangeStatus changeTemplate(DashboardSetting setting) {
Property newTemplate = applyConfiguration(setting.toEntity());
try {
- this.getClient().define(newTemplate);
+ this.getClient().apply(newTemplate);
return TemplateChangeStatus.builder()
.status(true)
- .id(newTemplate.getMetadata().getId())
+ .id(newTemplate.getId())
.build();
} catch (IOException ioEx) {
log.error("fail to modify the template", ioEx);
@@ -101,7 +100,7 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
}
UITemplate uiTemplate = parse(oldProperty);
try {
- this.getClient().define(applyStatus(uiTemplate));
+ this.getClient().apply(applyStatus(uiTemplate));
return TemplateChangeStatus.builder()
.status(true)
.id(uiTemplate.id().build())
@@ -121,7 +120,7 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
public UITemplate parse(Property property) {
UITemplate uiTemplate = new UITemplate();
- uiTemplate.setTemplateId(property.getMetadata().getId());
+ uiTemplate.setTemplateId(property.getId());
for (BanyandbModel.Tag tag : property.getTagsList()) {
TagAndValue<?> tagAndValue = TagAndValue.fromProtobuf(tag);
@@ -138,11 +137,10 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
public Property applyAll(UITemplate uiTemplate) {
return Property.newBuilder()
- .setMetadata(BanyandbProperty.Metadata.newBuilder()
- .setId(uiTemplate.id().build())
- .setContainer(BanyandbCommon.Metadata.newBuilder()
+ .setMetadata(BanyandbCommon.Metadata.newBuilder()
.setGroup(GROUP)
- .setName(UITemplate.INDEX_NAME)))
+ .setName(UITemplate.INDEX_NAME))
+ .setId(uiTemplate.id().build())
.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())
@@ -157,11 +155,10 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
*/
public Property applyStatus(UITemplate uiTemplate) {
return Property.newBuilder()
- .setMetadata(BanyandbProperty.Metadata.newBuilder()
- .setId(uiTemplate.id().build())
- .setContainer(BanyandbCommon.Metadata.newBuilder()
- .setGroup(GROUP)
- .setName(UITemplate.INDEX_NAME)))
+ .setMetadata(BanyandbCommon.Metadata.newBuilder()
+ .setGroup(GROUP)
+ .setName(UITemplate.INDEX_NAME))
+ .setId(uiTemplate.id().build())
.addTags(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled()).build())
.addTags(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()).build())
.build();
@@ -175,11 +172,10 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
*/
public Property applyConfiguration(UITemplate uiTemplate) {
return Property.newBuilder()
- .setMetadata(BanyandbProperty.Metadata.newBuilder()
- .setId(uiTemplate.id().build())
- .setContainer(BanyandbCommon.Metadata.newBuilder()
- .setGroup(GROUP)
- .setName(UITemplate.INDEX_NAME)))
+ .setMetadata(BanyandbCommon.Metadata.newBuilder()
+ .setGroup(GROUP)
+ .setName(UITemplate.INDEX_NAME))
+ .setId(uiTemplate.id().build())
.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/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 f88c7619ea..2637e19240 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
@@ -30,6 +30,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Metadata;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
+import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Property;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.CompressionMethod;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.EncodingMethod;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.FieldSpec;
@@ -63,6 +64,7 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -177,6 +179,19 @@ public enum MetadataRegistry {
return new MeasureModel(builder.build(), indexRules);
}
+ public PropertyModel registerPropertyModel(Model model) {
+ final SchemaMetadata schemaMetadata = parseMetadata(model, null, null);
+ Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
+ List<TagMetadata> tags = parseTagMetadata(model, schemaBuilder, Collections.emptyList(), schemaMetadata.group);
+ final Property.Builder builder = Property.newBuilder();
+ builder.setMetadata(BanyandbCommon.Metadata.newBuilder().setGroup(schemaMetadata.getGroup())
+ .setName(schemaMetadata.name()));
+ for (TagMetadata tag : tags) {
+ builder.addTags(tag.getTagSpec());
+ }
+ return new PropertyModel(builder.build());
+ }
+
private TopNAggregation parseTopNSpec(final Model model, final String group, final String measureName)
throws StorageException {
if (model.getBanyanDBModelExtension().getTopN() == null) {
@@ -434,7 +449,7 @@ public enum MetadataRegistry {
TagSpec.Builder tagSpec = TagSpec.newBuilder().setName(colName);
if (String.class.equals(clazz) || StorageDataComplexObject.class.isAssignableFrom(clazz) || JsonObject.class.equals(clazz)) {
tagSpec = tagSpec.setType(TagType.TAG_TYPE_STRING);
- } else if (int.class.equals(clazz) || long.class.equals(clazz)) {
+ } else if (int.class.equals(clazz) || long.class.equals(clazz) || Integer.class.equals(clazz) || Long.class.equals(clazz)) {
tagSpec = tagSpec.setType(TagType.TAG_TYPE_INT);
} else if (byte[].class.equals(clazz)) {
tagSpec = tagSpec.setType(TagType.TAG_TYPE_DATA_BINARY);
@@ -461,6 +476,9 @@ public enum MetadataRegistry {
}
public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, DownSamplingConfigService configService) {
+ if (!model.isTimeSeries()) {
+ return new SchemaMetadata("sw", model.getName(), Kind.Property, DownSampling.None, 1, 0, 0);
+ }
if (model.isRecord()) { // stream
return new SchemaMetadata(model.isSuperDataset() ? model.getName() : "normal",
model.getName(),
@@ -600,7 +618,7 @@ public enum MetadataRegistry {
}
public enum Kind {
- MEASURE, STREAM;
+ MEASURE, STREAM, Property;
}
@RequiredArgsConstructor
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/PropertyModel.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/PropertyModel.java
new file mode 100644
index 0000000000..5332056a2d
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/PropertyModel.java
@@ -0,0 +1,11 @@
+package org.apache.skywalking.oap.server.storage.plugin.banyandb;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Property;
+
+@RequiredArgsConstructor
+@Getter
+public class PropertyModel {
+ private final Property property;
+}
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 52e69a4574..626567b1fd 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
@@ -21,7 +21,6 @@ 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.property.v1.BanyandbProperty.Property;
import org.apache.skywalking.oap.server.core.profiling.continuous.storage.ContinuousProfilingPolicy;
@@ -44,7 +43,7 @@ public class BanyanDBContinuousProfilingPolicyDAO extends AbstractBanyanDBDAO im
@Override
public void savePolicy(ContinuousProfilingPolicy policy) throws IOException {
try {
- this.getClient().define(applyAll(policy));
+ this.getClient().apply(applyAll(policy));
} catch (IOException e) {
log.error("fail to save policy", e);
}
@@ -52,11 +51,10 @@ public class BanyanDBContinuousProfilingPolicyDAO extends AbstractBanyanDBDAO im
public Property applyAll(ContinuousProfilingPolicy policy) {
return Property.newBuilder()
- .setMetadata(BanyandbProperty.Metadata.newBuilder()
- .setId(policy.id().build())
- .setContainer(BanyandbCommon.Metadata.newBuilder()
- .setGroup(GROUP)
- .setName(ContinuousProfilingPolicy.INDEX_NAME)))
+ .setMetadata(BanyandbCommon.Metadata.newBuilder()
+ .setGroup(GROUP)
+ .setName(ContinuousProfilingPolicy.INDEX_NAME))
+ .setId(policy.id().build())
.addTags(TagAndValue.newStringTag(ContinuousProfilingPolicy.UUID, policy.getUuid()).build())
.addTags(TagAndValue.newStringTag(ContinuousProfilingPolicy.CONFIGURATION_JSON, policy.getConfigurationJson()).build())
.build();
@@ -73,7 +71,7 @@ public class BanyanDBContinuousProfilingPolicyDAO extends AbstractBanyanDBDAO im
}
}).filter(Objects::nonNull).map(properties -> {
final ContinuousProfilingPolicy policy = new ContinuousProfilingPolicy();
- policy.setServiceId(properties.getMetadata().getId());
+ policy.setServiceId(properties.getId());
for (BanyandbModel.Tag tag : properties.getTagsList()) {
TagAndValue<?> tagAndValue = TagAndValue.fromProtobuf(tag);
if (tagAndValue.getTagName().equals(ContinuousProfilingPolicy.CONFIGURATION_JSON)) {