diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 978a621531..015e71e9ac 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -36,6 +36,10 @@
* Optimize `flushInterval` of ElasticSearch BulkProcessor to avoid extra periodical flush in the continuous bulk streams.
* An unexpected dot is added when exp is a pure metric name and expPrefix != null.
* Support monitoring MariaDB.
+* Remove measure/stream specific interval settings in BanyanDB.
+* Add global-specific settings used to override global configurations (e.g `segmentIntervalDays`, `blockIntervalHours`) in BanyanDB.
+* Use TTL-driven interval settings for the `measure-default` group in BanyanDB.
+* Fix wrong group of non time-relative metadata in BanyanDB.
#### UI
diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml
index f6819beada..32da4e1c48 100644
--- a/oap-server/server-starter/src/main/resources/application.yml
+++ b/oap-server/server-starter/src/main/resources/application.yml
@@ -235,10 +235,11 @@ storage:
superDatasetShardsFactor: ${SW_STORAGE_BANYANDB_SUPERDATASET_SHARDS_FACTOR:2}
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_PROFILE_TASK_QUERY_MAX_SIZE:200} # the max number of fetch task in a request
- streamBlockInterval: ${SW_STORAGE_BANYANDB_STREAM_BLOCK_INTERVAL:4} # Unit is hour
- streamSegmentInterval: ${SW_STORAGE_BANYANDB_STREAM_SEGMENT_INTERVAL:24} # Unit is hour
- measureBlockInterval: ${SW_STORAGE_BANYANDB_MEASURE_BLOCK_INTERVAL:4} # Unit is hour
- measureSegmentInterval: ${SW_STORAGE_BANYANDB_MEASURE_SEGMENT_INTERVAL:24} # Unit is hour
+ blockIntervalHours: ${SW_STORAGE_BANYANDB_BLOCK_INTERVAL_HOURS:24} # Unit is hour
+ segmentIntervalDays: ${SW_STORAGE_BANYANDB_SEGMENT_INTERVAL_DAYS:1} # Unit is day
+ superDatasetBlockIntervalHours: ${SW_STORAGE_BANYANDB_SUPER_DATASET_BLOCK_INTERVAL_HOURS:24} # Unit is hour
+ superDatasetSegmentIntervalDays: ${SW_STORAGE_BANYANDB_SUPER_DATASET_SEGMENT_INTERVAL_DAYS:1} # Unit is day
+ specificGroupSettings: ${SW_STORAGE_BANYANDB_SPECIFIC_GROUP_SETTINGS:""} # For example, {"group1": {"blockIntervalHours": 4, "segmentIntervalDays": 1}}
agent-analyzer:
selector: ${SW_AGENT_ANALYZER:default}
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 d97c9a8699..2d40e2d6c9 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
@@ -40,6 +40,7 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
public BanyanDBIndexInstaller(Client client, ModuleManager moduleManager, BanyanDBStorageConfig config) {
super(client, moduleManager);
this.config = config;
+ MetadataRegistry.INSTANCE.initializeIntervals(config.getSpecificGroupSettings());
}
@Override
@@ -60,9 +61,9 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
// then check entity schema
if (metadata.findRemoteSchema(c).isPresent()) {
// register models only locally but not remotely
- if (model.isTimeSeries() && model.isRecord()) { // stream
+ if (model.isRecord()) { // stream
MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
- } else if (model.isTimeSeries() && !model.isRecord()) { // measure
+ } else { // measure
MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
}
return true;
@@ -78,20 +79,18 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
public void createTable(Model model) throws StorageException {
try {
ConfigService configService = moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class);
- if (model.isTimeSeries() && model.isRecord()) { // stream
+ if (model.isRecord()) { // stream
Stream stream = MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
if (stream != null) {
log.info("install stream schema {}", model.getName());
((BanyanDBStorageClient) client).define(stream);
}
- } else if (model.isTimeSeries() && !model.isRecord()) { // measure
+ } else { // measure
Measure measure = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
if (measure != null) {
log.info("install measure schema {}", model.getName());
((BanyanDBStorageClient) client).define(measure);
}
- } else if (!model.isTimeSeries()) {
- log.info("skip property index {}", model.getName());
}
} catch (IOException ex) {
throw new StorageException("fail to install schema", ex);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
index 3a6569135b..39267d1b7f 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
@@ -57,31 +57,43 @@ public class BanyanDBStorageConfig extends ModuleConfig {
*/
private int superDatasetShardsFactor;
/**
- * block interval for Stream group.
- * Unit is hours.
+ * Default global block interval for non-super-dataset models.
+ * Unit is hour.
*
- * @since 9.3.0
+ * @since 9.4.0
*/
- private int streamBlockInterval;
+ private int blockIntervalHours;
/**
- * segment interval for Stream group.
- * Unit is hours.
+ * Default global segment interval for non-super-dataset models.
+ * Unit is day.
*
- * @since 9.3.0
+ * @since 9.4.0
*/
- private int streamSegmentInterval;
+ private int segmentIntervalDays;
/**
- * block interval for Measure group.
- * Unit is hours.
+ * Default global block interval for super-dataset models.
+ * Unit is hour.
*
- * @since 9.3.0
+ * @since 9.4.0
*/
- private int measureBlockInterval;
+ private int superDatasetBlockIntervalHours;
/**
- * segment interval for Measure group.
- * Unit is hours.
+ * Default global segment interval for super-dataset models.
+ * Unit is day.
*
- * @since 9.3.0
+ * @since 9.4.0
*/
- private int measureSegmentInterval;
+ private int superDatasetSegmentIntervalDays;
+ /**
+ * Specify the settings for each group individually. All groups created in BanyanDB can
+ * be found with bydbctl.
+ *
+ * NOTE: setting intervals works for all groups except `measure-default`.
+ *
+ * NOTE: available groups: `measure-default`, `measure-sampled`, `stream-default`
+ * and `stream-*` with names of the super dataset as the suffix.
+ *
+ * @since 9.4.0
+ */
+ private String specificGroupSettings;
}
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 e32dfa0e74..d362035ff1 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
@@ -18,8 +18,12 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.JsonObject;
import io.grpc.Status;
+
+import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.Comparator;
@@ -32,11 +36,14 @@ import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
+
import lombok.Builder;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.Getter;
+import lombok.NoArgsConstructor;
import lombok.RequiredArgsConstructor;
+import lombok.Setter;
import lombok.Singular;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.banyandb.v1.client.BanyanDBClient;
@@ -66,8 +73,11 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
public enum MetadataRegistry {
INSTANCE;
+ private static final ObjectMapper MAPPER = new ObjectMapper();
private final Map registry = new HashMap<>();
+ private Map specificGroupSettings = new HashMap<>();
+
public Stream registerStreamModel(Model model, BanyanDBStorageConfig config, ConfigService configService) {
final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
@@ -90,7 +100,7 @@ public enum MetadataRegistry {
String timestampColumn4Stream = model.getBanyanDBModelExtension().getTimestampColumn();
if (StringUtil.isBlank(timestampColumn4Stream)) {
throw new IllegalStateException(
- "Model[stream." + model.getName() + "] miss defined @BanyanDB.TimestampColumn");
+ "Model[stream." + model.getName() + "] miss defined @BanyanDB.TimestampColumn");
}
schemaBuilder.timestampColumn4Stream(timestampColumn4Stream);
List indexRules = tags.stream()
@@ -336,29 +346,70 @@ public enum MetadataRegistry {
return tagSpec;
}
+ public void initializeIntervals(String specificGroupSettingsStr) {
+ if (StringUtil.isBlank(specificGroupSettingsStr)) {
+ return;
+ }
+ try {
+ specificGroupSettings = MAPPER.readerFor(new TypeReference>() {
+ }).readValue(specificGroupSettingsStr);
+ } catch (IOException ioEx) {
+ log.warn("fail to parse specificGroupSettings", ioEx);
+ }
+ }
+
public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, ConfigService configService) {
- if (model.isRecord()) {
- String group = "stream-default";
+ String group;
+ if (model.isRecord()) { // stream
+ group = "stream-default";
if (model.isSuperDataset()) {
// for superDataset, we should use separate group
group = "stream-" + model.getName();
}
+ } else if (model.getDownsampling() == DownSampling.Minute && model.isTimeRelativeID()) { // measure
+ group = "measure-sampled";
+ } else {
+ // Solution: 2 * TTL < T * (1 + 0.8)
+ // e.g. if TTL=7, T=8: a new block/segment will be created at 14.4 days,
+ // while the first block has been deleted at 2*TTL
+ final int intervalDays = Double.valueOf(Math.ceil(configService.getMetricsDataTTL() * 2.0 / 1.8)).intValue();
+ return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE,
+ model.getDownsampling(),
+ config.getMetricsShardsNumber(),
+ intervalDays * 24,
+ intervalDays * 24, // use 10-day/240-hour strategy
+ configService.getMetricsDataTTL());
+ }
+
+ int blockIntervalHrs = config.getBlockIntervalHours();
+ int segmentIntervalDays = config.getSegmentIntervalDays();
+ if (model.isSuperDataset()) {
+ blockIntervalHrs = config.getSuperDatasetBlockIntervalHours();
+ segmentIntervalDays = config.getSuperDatasetSegmentIntervalDays();
+ }
+ GroupSetting groupSetting = this.specificGroupSettings.get(group);
+ if (groupSetting != null) {
+ blockIntervalHrs = groupSetting.getBlockIntervalHours();
+ segmentIntervalDays = groupSetting.getSegmentIntervalDays();
+ }
+ if (model.isRecord()) {
return new SchemaMetadata(group,
model.getName(),
Kind.STREAM,
model.getDownsampling(),
config.getRecordShardsNumber() *
(model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1),
- config.getStreamBlockInterval(),
- config.getStreamSegmentInterval(),
+ blockIntervalHrs,
+ segmentIntervalDays * 24,
configService.getRecordDataTTL()
);
}
- return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE,
+ // FIX: address issue #10104
+ return new SchemaMetadata(group, model.getName(), Kind.MEASURE,
model.getDownsampling(),
config.getMetricsShardsNumber(),
- config.getMeasureBlockInterval(),
- config.getMeasureSegmentInterval(),
+ blockIntervalHrs,
+ segmentIntervalDays * 24,
configService.getMetricsDataTTL());
}
@@ -542,4 +593,12 @@ public enum MetadataRegistry {
public enum ColumnType {
TAG, FIELD;
}
+
+ @Getter
+ @Setter
+ @NoArgsConstructor
+ public static class GroupSetting {
+ private int blockIntervalHours;
+ private int segmentIntervalDays;
+ }
}
diff --git a/test/e2e-v2/script/env b/test/e2e-v2/script/env
index a7656d6b36..241b59264a 100644
--- a/test/e2e-v2/script/env
+++ b/test/e2e-v2/script/env
@@ -23,6 +23,6 @@ SW_AGENT_CLIENT_JS_COMMIT=af0565a67d382b683c1dbd94c379b7080db61449
SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
SW_KUBERNETES_COMMIT_SHA=b670c41d94a82ddefcf466d54bab5c492d88d772
SW_ROVER_COMMIT=d956eaede57b62108b78bca48045bd09ba88e653
-SW_BANYANDB_COMMIT=566314a3d72299879233e8b66be218df3b73b6e0
+SW_BANYANDB_COMMIT=005b02210caacee0141de8085edebed367ef5a6f
SW_CTL_COMMIT=e684fae0107045fc23799146d62f04cb68bd5a3b