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 2024/11/22 03:31:22 UTC
(skywalking) 01/01: Merge remote-tracking branch 'origin/master' into banyandb
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch banyandb
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit cf9373c05f5ec0fbc39804336a589adaa08c0a8d
Merge: c167a709cf e1698a583f
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Nov 22 11:25:59 2024 +0800
Merge remote-tracking branch 'origin/master' into banyandb
docs/en/changes/changes.md | 1 +
oap-server-bom/pom.xml | 2 +-
.../src/main/resources/banyandb-min.yml | 10 ---
.../server-starter/src/main/resources/banyandb.yml | 25 --------
.../storage/plugin/banyandb/MetadataRegistry.java | 73 ++++++++++++++++------
5 files changed, 55 insertions(+), 56 deletions(-)
diff --cc oap-server-bom/pom.xml
index 2e4646bcb9,0a6a8e1ce1..64799507c3
--- a/oap-server-bom/pom.xml
+++ b/oap-server-bom/pom.xml
@@@ -73,7 -73,7 +73,7 @@@
<httpcore.version>4.4.13</httpcore.version>
<httpasyncclient.version>4.1.5</httpasyncclient.version>
<commons-compress.version>1.21</commons-compress.version>
- <banyandb-java-client.version>0.8.0-SNAPSHOT</banyandb-java-client.version>
- <banyandb-java-client.version>0.7.0</banyandb-java-client.version>
++ <banyandb-java-client.version>0.8.0-rc0</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 --cc oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java
index ee0ccdcd34,f921aa847d..8813773386
--- 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
@@@ -71,28 -89,18 +71,31 @@@ import org.apache.skywalking.oap.server
import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.apache.skywalking.oap.server.library.util.StringUtil;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.lang.reflect.ParameterizedType;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
++import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
@Slf4j
public enum MetadataRegistry {
INSTANCE;
private static final ObjectMapper MAPPER = new ObjectMapper();
+ // BanyanDB group setting aligned with the OAP settings
+ private static final Set<String> GROUP_ALIGNED = new HashSet<>();
private final Map<String, Schema> registry = new HashMap<>();
- private Map<String, GroupSetting> specificGroupSettings = new HashMap<>();
-
- public StreamModel registerStreamModel(Model model, BanyanDBStorageConfig config, ConfigService configService) {
+ public StreamModel registerStreamModel(Model model, BanyanDBStorageConfig config, DownSamplingConfigService configService) {
final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
@@@ -463,59 -465,69 +466,59 @@@
return tagSpec.build();
}
- public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, DownSamplingConfigService configService) {
- public void initializeIntervals(String specificGroupSettingsStr) {
- if (StringUtil.isBlank(specificGroupSettingsStr)) {
- return;
- }
- try {
- specificGroupSettings = MAPPER.readerFor(new TypeReference<Map<String, GroupSetting>>() {
- }).readValue(specificGroupSettingsStr);
- } catch (IOException ioEx) {
- log.warn("fail to parse specificGroupSettings", ioEx);
++ public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, DownSamplingConfigService configService) {
+ if (model.isRecord()) { // stream
+ return new SchemaMetadata(model.isSuperDataset() ? model.getName() : "normal",
+ model.getName(),
+ Kind.STREAM,
+ model.getDownsampling(),
+ model.isSuperDataset() ? config.getGrSuperShardNum() : config.getGrNormalShardNum(),
+ model.isSuperDataset() ? config.getGrSuperSIDays() : config.getGrNormalSIDays(),
+ model.isSuperDataset() ? config.getGrSuperTTLDays() : config.getGrNormalTTLDays());
}
- }
- public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, ConfigService configService) {
- int segmentIntervalDays = config.getSegmentIntervalDays();
- if (model.isSuperDataset()) {
- segmentIntervalDays = config.getSuperDatasetSegmentIntervalDays();
- }
- String group;
- int metricShardNum = config.getMetricsShardsNumber();
- 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-minute";
- // apply super dataset's settings to measure-minute
- segmentIntervalDays = config.getSuperDatasetSegmentIntervalDays();
- metricShardNum = metricShardNum * config.getSuperDatasetShardsFactor();
- } 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,
+ if (model.getBanyanDBModelExtension().isIndexMode()) {
+ return new SchemaMetadata("index", model.getName(), Kind.MEASURE,
model.getDownsampling(),
- config.getMetricsShardsNumber(),
- intervalDays, // use 10-day/240-hour strategy
- configService.getMetricsDataTTL());
+ config.getGmIndexShardNum(),
+ config.getGmIndexSIDays(),
+ config.getGmIndexTTLDays());
}
- GroupSetting groupSetting = this.specificGroupSettings.get(group);
- if (groupSetting != null) {
- segmentIntervalDays = groupSetting.getSegmentIntervalDays();
+ switch (model.getDownsampling()) {
+ case Minute:
+ return new SchemaMetadata(DownSampling.Minute.getName(),
+ model.getName(),
+ Kind.MEASURE,
+ model.getDownsampling(),
+ config.getGmMinuteShardNum(),
+ config.getGmMinuteSIDays(),
+ config.getGmMinuteTTLDays());
+ case Hour:
+ if (!configService.shouldToHour()) {
+ throw new UnsupportedOperationException("downsampling to hour is not supported");
+ }
+ return new SchemaMetadata(DownSampling.Hour.getName(),
+ model.getName(),
+ Kind.MEASURE,
+ model.getDownsampling(),
+ config.getGmHourShardNum(),
+ config.getGmHourSIDays(),
+ config.getGmHourTTLDays());
+ case Day:
+ if (!configService.shouldToDay()) {
+ throw new UnsupportedOperationException("downsampling to day is not supported");
+ }
+ return new SchemaMetadata(DownSampling.Day.getName(),
+ model.getName(),
+ Kind.MEASURE,
+ model.getDownsampling(),
+ config.getGmDayShardNum(),
+ config.getGmDaySIDays(),
+ config.getGmDayTTLDays());
+ default:
+ throw new UnsupportedOperationException("unsupported downSampling interval:" + model.getDownsampling());
}
- if (model.isRecord()) {
- return new SchemaMetadata(group,
- model.getName(),
- Kind.STREAM,
- model.getDownsampling(),
- config.getRecordShardsNumber() *
- (model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1),
- segmentIntervalDays,
- configService.getRecordDataTTL()
- );
- }
- // FIX: address issue #10104
- return new SchemaMetadata(group, model.getName(), Kind.MEASURE,
- model.getDownsampling(),
- metricShardNum,
- segmentIntervalDays,
- configService.getMetricsDataTTL());
}
@RequiredArgsConstructor
@@@ -600,6 -612,16 +603,16 @@@
return tagFamilySpecs;
}
+ /**
+ * Check if the group settings need to be updated
+ */
+ private boolean checkGroupUpdate(BanyanDBClient client) throws BanyanDBException {
+ Group g = client.findGroup(this.group);
+ return g.getResourceOpts().getShardNum() != this.shard
- || g.getResourceOpts().getSegmentInterval().getNum() != this.segmentIntervalDays
- || g.getResourceOpts().getTtl().getNum() != this.ttlDays;
++ || g.getResourceOpts().getSegmentInterval().getNum() != this.segmentIntervalDays
++ || g.getResourceOpts().getTtl().getNum() != this.ttlDays;
+ }
+
public boolean checkResourceExistence(BanyanDBClient client) throws BanyanDBException {
ResourceExist resourceExist;
Group.Builder gBuilder
@@@ -639,19 -673,28 +664,27 @@@
return resourceExist.hasResource();
case MEASURE:
resourceExist = client.existMeasure(this.group, this.name());
- try {
+ gBuilder.setCatalog(Catalog.CATALOG_MEASURE).build();
+ if (!GROUP_ALIGNED.contains(this.group)) {
if (!resourceExist.hasGroup()) {
- Group g = client.define(gBuilder.setCatalog(Catalog.CATALOG_MEASURE).build());
- if (g != null) {
- log.info("group {} created", g.getMetadata().getName());
+ try {
+ Group g = client.define(gBuilder.build());
+ if (g != null) {
+ log.info("group {} created", g.getMetadata().getName());
+ }
+ } catch (BanyanDBException ex) {
+ if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
+ log.info("group {} already created by another OAP node", this.group);
+ } else {
+ throw ex;
+ }
}
- }
- } catch (BanyanDBException ex) {
- if (ex.getStatus().equals(Status.Code.ALREADY_EXISTS)) {
- log.info("group {} already created by another OAP node", this.group);
} else {
- throw ex;
+ if (this.checkGroupUpdate(client)) {
+ client.update(gBuilder.build());
+ log.info("group {} updated", this.group);
+ }
}
- GROUP_ALIGNED.add(this.group);
}
return resourceExist.hasResource();
default: