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 2022/11/04 03:33:29 UTC
[skywalking] branch master updated: Update BanyanDB storage module to adapt BanyanDB 0.2 (#9839)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/master by this push:
new d32a318637 Update BanyanDB storage module to adapt BanyanDB 0.2 (#9839)
d32a318637 is described below
commit d32a318637bc506364928456aff0f5b02f9bc9c9
Author: Jiajing LU <lu...@gmail.com>
AuthorDate: Fri Nov 4 11:33:21 2022 +0800
Update BanyanDB storage module to adapt BanyanDB 0.2 (#9839)
* update to banyandb client 0.2
* simplify group check and creation
* support indexOnly
* partial apply configuration
* support CompleteableFuture
* fix BanyanDBProfileThreadSnapshotQueryDAO
* optimize BanyanDBProfileTaskQueryDAO
* optimize *LogQueryDAO
* support indexType
* add configuration
* use TTL from core config
Signed-off-by: Megrez Lu <lu...@gmail.com>
Co-authored-by: 吴晟 Wu Sheng <wu...@foxmail.com>
---
.licenserc.yaml | 6 +
dist-material/release-docs/LICENSE | 11 +
docs/en/changes/changes.md | 13 ++
docs/en/setup/backend/backend-storage.md | 4 +
oap-server-bom/pom.xml | 3 +-
.../server/core/storage/annotation/BanyanDB.java | 39 +++-
.../core/storage/model/BanyanDBExtension.java | 8 +
.../server/core/storage/model/StorageModels.java | 7 +-
.../core/storage/query/IBrowserLogQueryDAO.java | 9 +-
.../server/core/storage/query/ILogQueryDAO.java | 10 +-
.../server/core/storage/model/ModelColumnTest.java | 11 +-
.../src/main/resources/application.yml | 4 +
.../banyandb/BanyanDBAggregationQueryDAO.java | 2 +-
.../storage/plugin/banyandb/BanyanDBBatchDAO.java | 13 +-
.../storage/plugin/banyandb/BanyanDBConverter.java | 8 +-
.../plugin/banyandb/BanyanDBIndexInstaller.java | 31 ++-
.../plugin/banyandb/BanyanDBNoneStreamDAO.java | 4 +-
.../plugin/banyandb/BanyanDBRecordsQueryDAO.java | 2 +-
.../plugin/banyandb/BanyanDBStorageClient.java | 2 +-
.../plugin/banyandb/BanyanDBStorageConfig.java | 28 +++
.../banyandb/BanyanDBUITemplateManagementDAO.java | 36 ++-
.../storage/plugin/banyandb/MetadataRegistry.java | 250 ++++++++++++++++-----
.../banyandb/measure/BanyanDBEventQueryDAO.java | 155 ++++++-------
.../banyandb/measure/BanyanDBMetadataQueryDAO.java | 119 +++++-----
.../banyandb/measure/BanyanDBMetricsDAO.java | 36 +--
.../banyandb/measure/BanyanDBMetricsQueryDAO.java | 99 ++++----
.../measure/BanyanDBNetworkAddressAliasDAO.java | 17 +-
.../banyandb/stream/AbstractBanyanDBDAO.java | 78 +++++--
.../stream/BanyanDBBrowserLogQueryDAO.java | 31 ---
.../banyandb/stream/BanyanDBLogQueryDAO.java | 16 --
.../stream/BanyanDBProfileTaskQueryDAO.java | 8 +-
.../BanyanDBProfileThreadSnapshotQueryDAO.java | 72 +++---
.../plugin/banyandb/stream/BanyanDBRecordDAO.java | 2 +-
test/e2e-v2/script/env | 2 +-
34 files changed, 726 insertions(+), 410 deletions(-)
diff --git a/.licenserc.yaml b/.licenserc.yaml
index e7a33ff930..ec0af00052 100644
--- a/.licenserc.yaml
+++ b/.licenserc.yaml
@@ -119,3 +119,9 @@ dependency:
- name: com.google.flatbuffers:flatbuffers-java
version: 1.12.0
license: Apache-2.0
+ - name: build.buf.protoc-gen-validate:pgv-java-stub
+ version: 0.6.13
+ license: Apache-2.0
+ - name: build.buf.protoc-gen-validate:protoc-gen-validate
+ version: 0.6.13
+ license: Apache-2.0
diff --git a/dist-material/release-docs/LICENSE b/dist-material/release-docs/LICENSE
index 2a4cf10095..6bb498a288 100755
--- a/dist-material/release-docs/LICENSE
+++ b/dist-material/release-docs/LICENSE
@@ -208,6 +208,8 @@ Apache-2.0 licenses
========================================================================
The following components are provided under the Apache-2.0 License. See project link for details.
The text of each license is the standard Apache 2.0 license.
+ https://mvnrepository.com/artifact/build.buf.protoc-gen-validate/pgv-java-stub/0.6.13 Apache-2.0
+ https://mvnrepository.com/artifact/build.buf.protoc-gen-validate/protoc-gen-validate/0.6.13 Apache-2.0
https://mvnrepository.com/artifact/com.aayushatharva.brotli4j/brotli4j/1.7.1 Apache-2.0
https://mvnrepository.com/artifact/com.alibaba.nacos/nacos-api/1.4.2 Apache-2.0
https://mvnrepository.com/artifact/com.alibaba.nacos/nacos-client/1.4.2 Apache-2.0
@@ -263,6 +265,7 @@ The text of each license is the standard Apache 2.0 license.
https://mvnrepository.com/artifact/commons-codec/commons-codec/1.15 Apache-2.0
https://mvnrepository.com/artifact/commons-io/commons-io/2.7 Apache-2.0
https://mvnrepository.com/artifact/commons-logging/commons-logging/1.2 Apache-2.0
+ https://mvnrepository.com/artifact/commons-validator/commons-validator/1.7 Apache-2.0
https://npmjs.com/package/d3-flame-graph/v/4.1.3 4.1.3 Apache-2.0
https://npmjs.com/package/echarts/v/5.2.2 5.2.2 Apache-2.0
https://mvnrepository.com/artifact/io.etcd/jetcd-common/0.5.3 Apache-2.0
@@ -618,6 +621,14 @@ The text of each license is also included in licenses/LICENSE-[project].txt.
https://npmjs.com/package/robust-predicates/v/3.0.1 3.0.1 Unlicense
+========================================================================
+https://golang.org/LICENSE licenses
+========================================================================
+The following components are provided under the https://golang.org/LICENSE License. See project link for details.
+The text of each license is also included in licenses/LICENSE-[project].txt.
+
+ https://mvnrepository.com/artifact/com.google.re2j/re2j/1.5 https://golang.org/LICENSE
+
========================================================================
https://spdx.org/licenses/MIT-0.html licenses
========================================================================
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 6c024b1da2..7705fcf3f6 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -83,6 +83,19 @@
* Support `sampledTrace` in LAL.
* Support multiple rules with different names under the same layer of LAL script.
* (Optimization) Reduce the buffer size(queue) of MAL(only) metric streams. Set L1 queue size as 1/20, L2 queue size as 1/2.
+* [**Breaking Change**] Migrate to BanyanDB v0.2.0.
+ * Adopt new OR logical operator for,
+ 1. `MeasureIDs` query
+ 2. `BanyanDBProfileThreadSnapshotQueryDAO` query
+ 3. Multiple `Event` conditions query
+ 4. Metrics query
+ * Simplify Group check and creation
+ * Partially apply `UITemplate` changes
+ * Support `index_only`
+ * Return `CompletableFuture<Void>` directly from BanyanDB client
+ * Optimize data binary parse methods in *LogQueryDAO
+ * Support different indexType
+ * Support configuration for TTL and (block|segment) intervals
#### UI
diff --git a/docs/en/setup/backend/backend-storage.md b/docs/en/setup/backend/backend-storage.md
index 8bcbcc63a5..bac970ca18 100644
--- a/docs/en/setup/backend/backend-storage.md
+++ b/docs/en/setup/backend/backend-storage.md
@@ -308,6 +308,10 @@ 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
```
For more details, please refer to the documents of [BanyanDB](https://skywalking.apache.org/docs/skywalking-banyandb/next/readme/)
diff --git a/oap-server-bom/pom.xml b/oap-server-bom/pom.xml
index bd1792d1ab..d14d8a0a94 100644
--- a/oap-server-bom/pom.xml
+++ b/oap-server-bom/pom.xml
@@ -73,7 +73,8 @@
<awaitility.version>3.0.0</awaitility.version>
<httpcore.version>4.4.13</httpcore.version>
<commons-compress.version>1.21</commons-compress.version>
- <banyandb-java-client.version>0.1.0</banyandb-java-client.version>
+ <!-- TODO: use release version of banyandb java client -->
+ <banyandb-java-client.version>0.2.0-SNAPSHOT</banyandb-java-client.version>
<kafka-clients.version>2.4.1</kafka-clients.version>
<spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
</properties>
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/BanyanDB.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/BanyanDB.java
index 87890c5f22..80acbdfc42 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/BanyanDB.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/BanyanDB.java
@@ -31,12 +31,12 @@ import java.lang.annotation.Target;
public @interface BanyanDB {
/**
* GlobalIndex declares advanced global index, which are only available in BanyanDB.
- *
+ * <p>
* Global index should only be considered if a column value has a huge value candidates, but we will need a direct
* equal
* query without timestamp.
* The typical global index is designed for huge candidate of indexed values, such as `trace ID` or `segment ID`.
- *
+ * <p>
* Only work with {@link Column}
*/
@Target({ElementType.FIELD})
@@ -51,17 +51,17 @@ public @interface BanyanDB {
* ServiceA's traffic gauge, service call per minute, includes following timestamp values, then it should be sharded
* by service ID
* [ServiceA(encoded ID): 01-28 18:30 values-1, 01-28 18:31 values-2, 01-28 18:32 values-3, 01-28 18:32 values-4]
- *
+ * <p>
* BanyanDB is the 1st storage implementation supporting this. It would make continuous time series metrics stored
* closely and compressed better.
- *
+ * <p>
* 1. One entity could have multiple sharding keys
* 2. If no column is appointed for this, {@link org.apache.skywalking.oap.server.core.storage.StorageData#id}
* would be used by the storage implementation accordingly.
- *
+ * <p>
* NOTICE, this sharding concept is NOT just for splitting data into different database instances or physical
* files.
- *
+ * <p>
* Only work with {@link Column}
*
* @return non-negative if this column be used for sharding. -1 means not as a sharding key
@@ -91,4 +91,31 @@ public @interface BanyanDB {
@interface NoIndexing {
}
+
+ /**
+ * Additional information for constructing Index in BanyanDB.
+ *
+ * @since 9.3.0
+ */
+ @Target({ElementType.FIELD})
+ @Retention(RetentionPolicy.RUNTIME)
+ @interface IndexRule {
+ /**
+ * IndexRule supports selecting two distinct kinds of index structures.
+ */
+ IndexType indexType() default IndexType.INVERTED;
+
+ enum IndexType {
+ /**
+ * The `INVERTED` index is the primary option when users set up an index rule.
+ * It's suitable for most tag indexing due to a better memory usage ratio and query performance.
+ */
+ INVERTED,
+ /**
+ * The `TREE` index could be better when there are high cardinalities, such as the `ID` tag and numeric duration tag.
+ * In these cases, it saves much memory space.
+ */
+ TREE;
+ }
+ }
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java
index fe90b85ef9..d1fdb008c8 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java
@@ -55,6 +55,14 @@ public class BanyanDBExtension {
*/
private final boolean shouldIndex;
+ /**
+ * indexType is the type of index built for a {@link ModelColumn} in BanyanDB.
+ *
+ * @since 9.3.0
+ */
+ @Getter
+ private final BanyanDB.IndexRule.IndexType indexType;
+
/**
* @return true if this column is a part of sharding key
*/
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
index 7b8ce242e1..b0aee43678 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
@@ -203,10 +203,13 @@ public class StorageModels implements IModelManager, ModelCreator, ModelManipula
BanyanDB.GlobalIndex.class);
final BanyanDB.NoIndexing banyanDBNoIndex = field.getAnnotation(
BanyanDB.NoIndexing.class);
+ final BanyanDB.IndexRule banyanDBIndexRule = field.getAnnotation(
+ BanyanDB.IndexRule.class);
BanyanDBExtension banyanDBExtension = new BanyanDBExtension(
banyanDBShardingKey == null ? -1 : banyanDBShardingKey.index(),
- banyanDBGlobalIndex == null ? false : true,
- banyanDBNoIndex != null ? false : column.storageOnly()
+ banyanDBGlobalIndex != null,
+ banyanDBNoIndex == null && column.storageOnly(),
+ banyanDBIndexRule == null ? BanyanDB.IndexRule.IndexType.INVERTED : banyanDBIndexRule.indexType()
);
final ModelColumn modelColumn = new ModelColumn(
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java
index 0f0f51e98e..593057067a 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java
@@ -36,15 +36,18 @@ public interface IBrowserLogQueryDAO extends Service {
int limit,
int from) throws IOException;
+ default BrowserErrorLog parserDataBinary(String dataBinaryBase64) {
+ return parserDataBinary(Base64.getDecoder().decode(dataBinaryBase64));
+ }
+
/**
* Parser the raw error log.
*/
- default BrowserErrorLog parserDataBinary(
- String dataBinaryBase64) {
+ default BrowserErrorLog parserDataBinary(byte[] dataBinary) {
try {
BrowserErrorLog log = new BrowserErrorLog();
org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog browserErrorLog = org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog
- .parseFrom(Base64.getDecoder().decode(dataBinaryBase64));
+ .parseFrom(dataBinary);
log.setService(browserErrorLog.getService());
log.setServiceVersion(browserErrorLog.getServiceVersion());
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java
index f4365fbbac..3e412d5178 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java
@@ -19,9 +19,11 @@
package org.apache.skywalking.oap.server.core.storage.query;
import com.google.protobuf.InvalidProtocolBufferException;
+
import java.io.IOException;
import java.util.Base64;
import java.util.List;
+
import org.apache.skywalking.apm.network.logging.v3.LogTags;
import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.Tag;
import org.apache.skywalking.oap.server.core.query.enumeration.Order;
@@ -50,11 +52,15 @@ public interface ILogQueryDAO extends Service {
final List<String> excludingKeywordsOfContent) throws IOException;
/**
- * Parser the raw tags.
+ * Parse the raw tags with base64 representation of data binary
*/
default void parserDataBinary(String dataBinaryBase64, List<KeyValue> tags) {
+ parserDataBinary(Base64.getDecoder().decode(dataBinaryBase64), tags);
+ }
+
+ default void parserDataBinary(byte[] dataBinary, List<KeyValue> tags) {
try {
- LogTags logTags = LogTags.parseFrom(Base64.getDecoder().decode(dataBinaryBase64));
+ LogTags logTags = LogTags.parseFrom(dataBinary);
logTags.getDataList().forEach(pair -> tags.add(new KeyValue(pair.getKey(), pair.getValue())));
} catch (InvalidProtocolBufferException e) {
throw new RuntimeException(e);
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
index d9fc9dd6a0..e2574a1dde 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
@@ -19,6 +19,7 @@
package org.apache.skywalking.oap.server.core.storage.model;
import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
+import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
import org.apache.skywalking.oap.server.core.storage.annotation.ElasticSearch;
import org.junit.Assert;
import org.junit.Test;
@@ -31,7 +32,7 @@ public class ModelColumnTest {
new SQLDatabaseExtension(),
new ElasticSearchExtension(
ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
- new BanyanDBExtension(-1, false, true)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
);
Assert.assertEquals(true, column.isStorageOnly());
Assert.assertEquals("abc", column.getColumnName().getName());
@@ -40,7 +41,7 @@ public class ModelColumnTest {
false, false, true, 200,
new SQLDatabaseExtension(),
new ElasticSearchExtension(ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
- new BanyanDBExtension(-1, false, true)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
);
Assert.assertEquals(true, column.isStorageOnly());
Assert.assertEquals("abc", column.getColumnName().getName());
@@ -50,7 +51,7 @@ public class ModelColumnTest {
false, false, true, 200,
new SQLDatabaseExtension(),
new ElasticSearchExtension(ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
- new BanyanDBExtension(-1, false, true)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
);
Assert.assertEquals(false, column.isStorageOnly());
Assert.assertEquals("abc", column.getColumnName().getName());
@@ -63,7 +64,7 @@ public class ModelColumnTest {
new SQLDatabaseExtension(),
new ElasticSearchExtension(
ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
- new BanyanDBExtension(-1, false, true)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
);
}
@@ -74,7 +75,7 @@ public class ModelColumnTest {
new SQLDatabaseExtension(),
new ElasticSearchExtension(
ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
- new BanyanDBExtension(-1, false, true)
+ new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
);
}
}
diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml
index 751c038045..06e7eb565e 100644
--- a/oap-server/server-starter/src/main/resources/application.yml
+++ b/oap-server/server-starter/src/main/resources/application.yml
@@ -236,6 +236,10 @@ 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
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/BanyanDBAggregationQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java
index 0928361ef5..58d61c9831 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java
@@ -77,7 +77,7 @@ public class BanyanDBAggregationQueryDAO extends AbstractBanyanDBDAO implements
return Collections.emptyList();
}
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName, duration.getStep());
if (schema == null) {
throw new IOException("schema is not registered");
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java
index 0e6d761aa9..9564de27da 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java
@@ -31,6 +31,7 @@ import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBS
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
public class BanyanDBBatchDAO extends AbstractDAO<BanyanDBStorageClient> implements IBatchDAO {
private static final Object STREAM_SYNCHRONIZER = new Object();
@@ -64,16 +65,16 @@ public class BanyanDBBatchDAO extends AbstractDAO<BanyanDBStorageClient> impleme
@Override
public CompletableFuture<Void> flush(List<PrepareRequest> prepareRequests) {
if (CollectionUtils.isNotEmpty(prepareRequests)) {
- for (final PrepareRequest r : prepareRequests) {
+ return CompletableFuture.allOf(prepareRequests.stream().map((Function<PrepareRequest, CompletableFuture<Void>>) r -> {
if (r instanceof BanyanDBStreamInsertRequest) {
- // TODO: return CompletableFuture<Void>
- getStreamBulkWriteProcessor().add(((BanyanDBStreamInsertRequest) r).getStreamWrite());
+ return getStreamBulkWriteProcessor().add(((BanyanDBStreamInsertRequest) r).getStreamWrite());
} else if (r instanceof BanyanDBMeasureInsertRequest) {
- getMeasureBulkWriteProcessor().add(((BanyanDBMeasureInsertRequest) r).getMeasureWrite());
+ return getMeasureBulkWriteProcessor().add(((BanyanDBMeasureInsertRequest) r).getMeasureWrite());
} else if (r instanceof BanyanDBMeasureUpdateRequest) {
- getMeasureBulkWriteProcessor().add(((BanyanDBMeasureUpdateRequest) r).getMeasureWrite());
+ return getMeasureBulkWriteProcessor().add(((BanyanDBMeasureUpdateRequest) r).getMeasureWrite());
}
- }
+ return CompletableFuture.completedFuture(null);
+ }).toArray(CompletableFuture[]::new));
}
return CompletableFuture.completedFuture(null);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java
index c4ddc13c8f..e40c6b7015 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java
@@ -41,8 +41,8 @@ public class BanyanDBConverter {
private final MetadataRegistry.Schema schema;
private final RowEntity rowEntity;
- public StorageToStream(String modelName, RowEntity rowEntity) {
- this.schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+ public StorageToStream(String streamModelName, RowEntity rowEntity) {
+ this.schema = MetadataRegistry.INSTANCE.findRecordMetadata(streamModelName);
this.rowEntity = rowEntity;
}
@@ -215,8 +215,8 @@ public class BanyanDBConverter {
private final MetadataRegistry.Schema schema;
private final DataPoint dataPoint;
- public StorageToMeasure(String modelName, DataPoint dataPoint) {
- this.schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+ public StorageToMeasure(MetadataRegistry.Schema schema, DataPoint dataPoint) {
+ this.schema = schema;
this.dataPoint = dataPoint;
}
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 e2f281a0b8..aabe4dbd49 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
@@ -21,9 +21,10 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb;
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.Group;
import org.apache.skywalking.banyandb.v1.client.metadata.Measure;
import org.apache.skywalking.banyandb.v1.client.metadata.Stream;
+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;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller;
@@ -35,29 +36,37 @@ import java.io.IOException;
@Slf4j
public class BanyanDBIndexInstaller extends ModelInstaller {
private final BanyanDBStorageConfig config;
+ private final ConfigService configService;
public BanyanDBIndexInstaller(Client client, ModuleManager moduleManager, BanyanDBStorageConfig config) {
super(client, moduleManager);
this.config = config;
+ this.configService = moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class);
}
@Override
protected boolean isExists(Model model) throws StorageException {
- final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(model, config);
+ final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(model, config, configService);
try {
final BanyanDBClient c = ((BanyanDBStorageClient) this.client).client;
- // first check group
- Group g = metadata.getOrCreateGroup(c);
- if (g == null) {
- throw new StorageException("fail to create group " + metadata.getGroup());
+ // first check resource existence and create group if necessary
+ final boolean resourceExist = metadata.checkResourceExistence(c);
+ if (!resourceExist) {
+ return false;
}
- log.info("group {} created", g.name());
+
// then check entity schema
if (metadata.findRemoteSchema(c).isPresent()) {
- MetadataRegistry.INSTANCE.registerModel(model, config);
+ // register models only locally but not remotely
+ if (model.isTimeSeries() && model.isRecord()) { // stream
+ MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
+ } else if (model.isTimeSeries() && !model.isRecord()) { // measure
+ MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
+ }
return true;
}
- return false;
+
+ throw new IllegalStateException("inconsistent state");
} catch (BanyanDBException ex) {
throw new StorageException("fail to check existence", ex);
}
@@ -67,13 +76,13 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
protected void createTable(Model model) throws StorageException {
try {
if (model.isTimeSeries() && model.isRecord()) { // stream
- Stream stream = (Stream) MetadataRegistry.INSTANCE.registerModel(model, config);
+ 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
- Measure measure = (Measure) MetadataRegistry.INSTANCE.registerModel(model, config);
+ Measure measure = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
if (measure != null) {
log.info("install measure schema {}", model.getName());
((BanyanDBStorageClient) client).define(measure);
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 316e0af413..12b03a94eb 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
@@ -41,12 +41,12 @@ public class BanyanDBNoneStreamDAO extends AbstractDAO<BanyanDBStorageClient> im
@Override
public void insert(Model model, NoneStream noneStream) throws IOException {
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findRecordMetadata(model.getName());
if (schema == null) {
throw new IOException(model.getName() + " is not registered");
}
StreamWrite streamWrite = new StreamWrite(schema.getMetadata().getGroup(), // group name
- model.getName(), // index-name
+ schema.getMetadata().name(), // stream-name
noneStream.id(), // identity
TimeBucket.getTimestamp(noneStream.getTimeBucket(), model.getDownsampling())); // timestamp
Convert2Storage<StreamWrite> convert2Storage = new BanyanDBConverter.StreamToStorage(schema, streamWrite);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java
index 3955fac142..d099a0a419 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java
@@ -71,7 +71,7 @@ public class BanyanDBRecordsQueryDAO extends AbstractBanyanDBDAO implements IRec
return Collections.emptyList();
}
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findRecordMetadata(modelName);
if (schema == null) {
throw new IOException("schema is not registered");
}
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 9ba378125c..f2408f8695 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
@@ -118,7 +118,7 @@ public class BanyanDBStorageClient implements Client, HealthCheckable {
public void define(Property property) throws IOException {
try {
- this.client.save(property);
+ this.client.apply(property);
this.healthChecker.health();
} catch (BanyanDBException ex) {
healthChecker.unHealth(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 e31874e7f5..3a6569135b 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
@@ -56,4 +56,32 @@ public class BanyanDBStorageConfig extends ModuleConfig {
* Shards Factor for a super dataset
*/
private int superDatasetShardsFactor;
+ /**
+ * block interval for Stream group.
+ * Unit is hours.
+ *
+ * @since 9.3.0
+ */
+ private int streamBlockInterval;
+ /**
+ * segment interval for Stream group.
+ * Unit is hours.
+ *
+ * @since 9.3.0
+ */
+ private int streamSegmentInterval;
+ /**
+ * block interval for Measure group.
+ * Unit is hours.
+ *
+ * @since 9.3.0
+ */
+ private int measureBlockInterval;
+ /**
+ * segment interval for Measure group.
+ * Unit is hours.
+ *
+ * @since 9.3.0
+ */
+ private int measureSegmentInterval;
}
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 ab0dd31f71..d616ed032e 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
@@ -26,7 +26,6 @@ import org.apache.skywalking.oap.server.core.query.input.DashboardSetting;
import org.apache.skywalking.oap.server.core.query.type.DashboardConfiguration;
import org.apache.skywalking.oap.server.core.query.type.TemplateChangeStatus;
import org.apache.skywalking.oap.server.core.storage.management.UITemplateManagementDAO;
-import org.apache.skywalking.oap.server.library.util.BooleanUtils;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
import java.io.IOException;
@@ -60,7 +59,7 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
@Override
public TemplateChangeStatus addTemplate(DashboardSetting setting) {
- Property newTemplate = convert(setting.toEntity());
+ Property newTemplate = applyAll(setting.toEntity());
try {
this.getClient().define(newTemplate);
return TemplateChangeStatus.builder()
@@ -76,7 +75,7 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
@Override
public TemplateChangeStatus changeTemplate(DashboardSetting setting) {
- Property newTemplate = convert(setting.toEntity());
+ Property newTemplate = applyConfiguration(setting.toEntity());
try {
this.getClient().define(newTemplate);
return TemplateChangeStatus.builder()
@@ -98,9 +97,8 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
.build();
}
UITemplate uiTemplate = parse(oldProperty);
- uiTemplate.setDisabled(BooleanUtils.FALSE);
try {
- this.getClient().define(convert(uiTemplate));
+ this.getClient().define(applyStatus(uiTemplate));
return TemplateChangeStatus.builder()
.status(true)
.id(uiTemplate.id())
@@ -134,11 +132,37 @@ public class BanyanDBUITemplateManagementDAO extends AbstractBanyanDBDAO impleme
return uiTemplate;
}
- public Property convert(UITemplate uiTemplate) {
+ public Property applyAll(UITemplate uiTemplate) {
return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id())
.addTag(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration()))
.addTag(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled()))
.addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()))
.build();
}
+
+ /**
+ * Partial apply status, i.e. disable tag.
+ *
+ * @param uiTemplate previous UITemplate
+ * @return new property (patch) to be applied
+ */
+ public Property applyStatus(UITemplate uiTemplate) {
+ return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id())
+ .addTag(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled()))
+ .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()))
+ .build();
+ }
+
+ /**
+ * Partial apply configuration, i.e. configuration tag.
+ *
+ * @param uiTemplate previous UITemplate
+ * @return new property (patch) to be applied
+ */
+ public Property applyConfiguration(UITemplate uiTemplate) {
+ return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id())
+ .addTag(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration()))
+ .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()))
+ .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 9287d5c710..9f6f2e4cf8 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
@@ -33,18 +33,23 @@ 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.ResourceExist;
import org.apache.skywalking.banyandb.v1.client.metadata.Stream;
import org.apache.skywalking.banyandb.v1.client.metadata.TagFamilySpec;
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.config.ConfigService;
+import org.apache.skywalking.oap.server.core.query.enumeration.Step;
+import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.core.storage.model.ModelColumn;
import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject;
-import javax.annotation.Nullable;
+import javax.annotation.Nonnull;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.Comparator;
@@ -63,8 +68,8 @@ public enum MetadataRegistry {
private final Map<String, Schema> registry = new HashMap<>();
- public NamedSchema<?> registerModel(Model model, BanyanDBStorageConfig config) {
- final SchemaMetadata schemaMetadata = parseMetadata(model, config);
+ public Stream registerStreamModel(Model model, BanyanDBStorageConfig config, ConfigService configService) {
+ final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
.collect(Collectors.toMap(modelColumn -> modelColumn.getColumnName().getStorageName(), Function.identity()));
@@ -87,38 +92,100 @@ public enum MetadataRegistry {
.filter(Objects::nonNull)
.collect(Collectors.toList());
- if (schemaMetadata.getKind() == Kind.STREAM) {
- final Stream.Builder builder = Stream.create(schemaMetadata.getGroup(), schemaMetadata.getName());
- if (entities.isEmpty()) {
- throw new IllegalStateException("sharding keys of model[stream." + model.getName() + "] must not be empty");
+ final Stream.Builder builder = Stream.create(schemaMetadata.getGroup(), schemaMetadata.name());
+ if (entities.isEmpty()) {
+ throw new IllegalStateException("sharding keys of model[stream." + model.getName() + "] must not be empty");
+ }
+ builder.setEntityRelativeTags(entities);
+ builder.addTagFamilies(tagFamilySpecs);
+ builder.addIndexes(indexRules);
+ registry.put(schemaMetadata.name(), schemaBuilder.build());
+ return builder.build();
+ }
+
+ public Measure registerMeasureModel(Model model, BanyanDBStorageConfig config, ConfigService configService) {
+ final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
+ Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
+ Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
+ .collect(Collectors.toMap(modelColumn -> modelColumn.getColumnName().getStorageName(), Function.identity()));
+ // parse and set sharding keys
+ List<String> entities = parseEntityNames(modelColumnMap);
+ // parse tag metadata
+ // this can be used to build both
+ // 1) a list of TagFamilySpec,
+ // 2) a list of IndexRule,
+ List<TagMetadata> tags = parseTagMetadata(model, schemaBuilder);
+ List<TagFamilySpec> tagFamilySpecs = schemaMetadata.extractTagFamilySpec(tags);
+ // iterate over tagFamilySpecs to save tag names
+ for (final TagFamilySpec tagFamilySpec : tagFamilySpecs) {
+ for (final TagFamilySpec.TagSpec tagSpec : tagFamilySpec.tagSpecs()) {
+ schemaBuilder.tag(tagSpec.getTagName());
}
- builder.setEntityRelativeTags(entities);
- builder.addTagFamilies(tagFamilySpecs);
- builder.addIndexes(indexRules);
- registry.put(model.getName(), schemaBuilder.build());
- return builder.build();
+ }
+ List<IndexRule> indexRules = tags.stream()
+ .map(TagMetadata::getIndexRule)
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+
+ final Measure.Builder builder = Measure.create(schemaMetadata.getGroup(), schemaMetadata.name(),
+ downSamplingDuration(model.getDownsampling()));
+ if (entities.isEmpty()) { // if shardingKeys is empty, for measure, we can use ID as a single sharding key.
+ builder.setEntityRelativeTags(Measure.ID);
} else {
- final Measure.Builder builder = Measure.create(schemaMetadata.getGroup(), schemaMetadata.getName(),
- downSamplingDuration(model.getDownsampling()));
- if (entities.isEmpty()) { // if shardingKeys is empty, for measure, we can use ID as a single sharding key.
- builder.setEntityRelativeTags(Measure.ID);
- } else {
- builder.setEntityRelativeTags(entities);
- }
- builder.addTagFamilies(tagFamilySpecs);
- builder.addIndexes(indexRules);
- // parse and set field
- Optional<ValueColumnMetadata.ValueColumn> valueColumnOpt = ValueColumnMetadata.INSTANCE
- .readValueColumnDefinition(model.getName());
- valueColumnOpt.ifPresent(valueColumn -> builder.addField(parseFieldSpec(modelColumnMap.get(valueColumn.getValueCName()), valueColumn)));
- valueColumnOpt.ifPresent(valueColumn -> schemaBuilder.field(valueColumn.getValueCName()));
- registry.put(model.getName(), schemaBuilder.build());
- return builder.build();
+ builder.setEntityRelativeTags(entities);
}
+ builder.addTagFamilies(tagFamilySpecs);
+ builder.addIndexes(indexRules);
+ // parse and set field
+ Optional<ValueColumnMetadata.ValueColumn> valueColumnOpt = ValueColumnMetadata.INSTANCE
+ .readValueColumnDefinition(model.getName());
+ valueColumnOpt.ifPresent(valueColumn -> builder.addField(parseFieldSpec(modelColumnMap.get(valueColumn.getValueCName()), valueColumn)));
+ valueColumnOpt.ifPresent(valueColumn -> schemaBuilder.field(valueColumn.getValueCName()));
+ registry.put(schemaMetadata.name(), schemaBuilder.build());
+ return builder.build();
}
- public Schema findMetadata(final String name) {
- return this.registry.get(name);
+ public Schema findMetadata(final Model model) {
+ if (model.isRecord()) {
+ return findRecordMetadata(model.getName());
+ }
+ return findMetadata(model.getName(), model.getDownsampling());
+ }
+
+ public Schema findRecordMetadata(final String recordModelName) {
+ final Schema s = this.registry.get(recordModelName);
+ if (s == null) {
+ return null;
+ }
+ // impose sanity check
+ if (s.getMetadata().getKind() != Kind.STREAM) {
+ throw new IllegalArgumentException(recordModelName + "is not a record");
+ }
+ return s;
+ }
+
+ static DownSampling deriveFromStep(Step step) {
+ switch (step) {
+ case DAY:
+ return DownSampling.Day;
+ case HOUR:
+ return DownSampling.Hour;
+ case SECOND:
+ return DownSampling.Second;
+ default:
+ return DownSampling.Minute;
+ }
+ }
+
+ public Schema findMetadata(final String modelName, Step step) {
+ return findMetadata(modelName, deriveFromStep(step));
+ }
+
+ /**
+ * Find metadata with down-sampling
+ */
+ public Schema findMetadata(final String modelName, DownSampling downSampling) {
+ return this.registry.get(SchemaMetadata.formatName(modelName, downSampling));
}
private Measure.FieldSpec parseFieldSpec(ModelColumn modelColumn, ValueColumnMetadata.ValueColumn valueColumn) {
@@ -167,10 +234,17 @@ public enum MetadataRegistry {
return IndexRule.create(tagName, IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.SERIES);
}
if (modelColumn.getBanyanDBExtension().isGlobalIndexing()) {
- return IndexRule.create(tagName, IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.GLOBAL);
+ return IndexRule.create(tagName, parseIndexType(modelColumn.getBanyanDBExtension().getIndexType()), IndexRule.IndexLocation.GLOBAL);
} else {
- return IndexRule.create(tagName, IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.SERIES);
+ return IndexRule.create(tagName, parseIndexType(modelColumn.getBanyanDBExtension().getIndexType()), IndexRule.IndexLocation.SERIES);
+ }
+ }
+
+ private static IndexRule.IndexType parseIndexType(BanyanDB.IndexRule.IndexType indexType) {
+ if (indexType == BanyanDB.IndexRule.IndexType.INVERTED) {
+ return IndexRule.IndexType.INVERTED;
}
+ return IndexRule.IndexType.TREE;
}
/**
@@ -203,9 +277,6 @@ public enum MetadataRegistry {
continue;
}
final TagFamilySpec.TagSpec tagSpec = parseTagSpec(col);
- if (tagSpec == null) {
- continue;
- }
builder.spec(col.getColumnName().getStorageName(), new ColumnSpec(ColumnType.TAG, col.getType()));
if (col.shouldIndex()) {
// build indexRule
@@ -225,33 +296,40 @@ public enum MetadataRegistry {
* @param modelColumn the column in the model to be parsed
* @return a typed tag spec
*/
- @Nullable
+ @Nonnull
private TagFamilySpec.TagSpec parseTagSpec(ModelColumn modelColumn) {
final Class<?> clazz = modelColumn.getType();
final String colName = modelColumn.getColumnName().getStorageName();
+ TagFamilySpec.TagSpec tagSpec = null;
if (String.class.equals(clazz) || StorageDataComplexObject.class.isAssignableFrom(clazz) || JsonObject.class.equals(clazz)) {
- return TagFamilySpec.TagSpec.newStringTag(colName);
+ tagSpec = TagFamilySpec.TagSpec.newStringTag(colName);
} else if (int.class.equals(clazz) || long.class.equals(clazz)) {
- return TagFamilySpec.TagSpec.newIntTag(colName);
+ tagSpec = TagFamilySpec.TagSpec.newIntTag(colName);
} else if (byte[].class.equals(clazz)) {
- return TagFamilySpec.TagSpec.newBinaryTag(colName);
+ tagSpec = TagFamilySpec.TagSpec.newBinaryTag(colName);
} else if (clazz.isEnum()) {
- return TagFamilySpec.TagSpec.newIntTag(colName);
+ tagSpec = TagFamilySpec.TagSpec.newIntTag(colName);
} else if (double.class.equals(clazz) || Double.class.equals(clazz)) {
// serialize double as binary
- return TagFamilySpec.TagSpec.newBinaryTag(colName);
+ tagSpec = TagFamilySpec.TagSpec.newBinaryTag(colName);
} else if (IntList.class.isAssignableFrom(clazz)) {
- return TagFamilySpec.TagSpec.newIntArrayTag(colName);
+ tagSpec = TagFamilySpec.TagSpec.newIntArrayTag(colName);
} else if (List.class.isAssignableFrom(clazz)) { // handle exceptions
ParameterizedType t = (ParameterizedType) modelColumn.getGenericType();
if (String.class.equals(t.getActualTypeArguments()[0])) {
- return TagFamilySpec.TagSpec.newStringArrayTag(colName);
+ tagSpec = TagFamilySpec.TagSpec.newStringArrayTag(colName);
}
}
- throw new IllegalStateException("type " + modelColumn.getType().toString() + " is not supported");
+ if (tagSpec == null) {
+ throw new IllegalStateException("type " + modelColumn.getType().toString() + " is not supported");
+ }
+ if (modelColumn.isIndexOnly()) {
+ tagSpec.indexedOnly();
+ }
+ return tagSpec;
}
- public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config) {
+ public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, ConfigService configService) {
if (model.isRecord()) {
String group = "stream-default";
if (model.isSuperDataset()) {
@@ -261,29 +339,61 @@ public enum MetadataRegistry {
return new SchemaMetadata(group,
model.getName(),
Kind.STREAM,
+ model.getDownsampling(),
config.getRecordShardsNumber() *
- (model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1)
+ (model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1),
+ config.getStreamBlockInterval(),
+ config.getStreamSegmentInterval(),
+ configService.getRecordDataTTL()
);
}
- return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE, config.getMetricsShardsNumber());
+ return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE,
+ model.getDownsampling(),
+ config.getMetricsShardsNumber(),
+ config.getMeasureBlockInterval(),
+ config.getMeasureSegmentInterval(),
+ configService.getMetricsDataTTL());
}
@RequiredArgsConstructor
@Data
public static class SchemaMetadata {
private final String group;
- private final String name;
+ /**
+ * name of the {@link Model}
+ */
+ private final String modelName;
private final Kind kind;
-
+ /**
+ * down-sampling of the {@link Model}
+ */
+ private final DownSampling downSampling;
private final int shard;
+ private final int blockIntervalHrs;
+ private final int segmentIntervalHrs;
+ private final int ttlDays;
+
+ /**
+ * Format the entity name for BanyanDB
+ *
+ * @param modelName name of the model
+ * @param downSampling not used if it is {@link DownSampling#None}
+ * @return entity (e.g. measure, stream) name
+ */
+ static String formatName(String modelName, DownSampling downSampling) {
+ if (downSampling == null || downSampling == DownSampling.None) {
+ return modelName;
+ }
+ return modelName + "_" + downSampling.getName();
+ }
public Optional<NamedSchema<?>> findRemoteSchema(BanyanDBClient client) throws BanyanDBException {
try {
switch (kind) {
case STREAM:
- return Optional.ofNullable(client.findStream(this.group, this.name));
+ return Optional.ofNullable(client.findStream(this.group, this.name()));
case MEASURE:
- return Optional.ofNullable(client.findMeasure(this.group, this.name));
+ return Optional.ofNullable(client.findMeasure(this.group, this.name()));
default:
throw new IllegalStateException("should not reach here");
}
@@ -315,21 +425,45 @@ public enum MetadataRegistry {
return tagFamilySpecs;
}
- public Group getOrCreateGroup(BanyanDBClient client) throws BanyanDBException {
- Group g = client.findGroup(this.group);
- if (g != null) {
- return g;
- }
+ public boolean checkResourceExistence(BanyanDBClient client) throws BanyanDBException {
+ ResourceExist resourceExist;
switch (kind) {
case STREAM:
- return client.define(Group.create(this.group, Catalog.STREAM, this.shard, 0, Duration.ofDays(7)));
+ resourceExist = client.existStream(this.group, this.name());
+ if (!resourceExist.hasGroup()) {
+ Group g = client.define(Group.create(this.group, Catalog.STREAM, this.shard,
+ IntervalRule.create(IntervalRule.Unit.HOUR, this.blockIntervalHrs),
+ IntervalRule.create(IntervalRule.Unit.HOUR, this.segmentIntervalHrs),
+ IntervalRule.create(IntervalRule.Unit.DAY, this.ttlDays)));
+ if (g != null) {
+ log.info("group {} created", g.name());
+ }
+ }
+ return resourceExist.hasResource();
case MEASURE:
- return client.define(Group.create(this.group, Catalog.MEASURE, this.shard, 12, Duration.ofDays(7)));
+ resourceExist = client.existMeasure(this.group, this.name());
+ if (!resourceExist.hasGroup()) {
+ Group g = client.define(Group.create(this.group, Catalog.MEASURE, this.shard,
+ IntervalRule.create(IntervalRule.Unit.HOUR, this.blockIntervalHrs),
+ IntervalRule.create(IntervalRule.Unit.HOUR, this.segmentIntervalHrs),
+ IntervalRule.create(IntervalRule.Unit.DAY, this.ttlDays)));
+ if (g != null) {
+ log.info("group {} created", g.name());
+ }
+ }
+ return resourceExist.hasResource();
default:
throw new IllegalStateException("should not reach here");
}
}
+ public String name() {
+ if (this.kind == Kind.MEASURE) {
+ return formatName(this.modelName, this.downSampling);
+ }
+ return this.modelName;
+ }
+
public String indexFamily() {
switch (kind) {
case MEASURE:
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java
index 6a522f7630..17ebe6d1ec 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java
@@ -18,16 +18,17 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.measure;
+import java.util.ArrayList;
import java.util.Collections;
-import java.util.Comparator;
import java.util.List;
import java.util.Set;
-import java.util.stream.Collectors;
import com.google.common.collect.ImmutableSet;
+import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
import org.apache.skywalking.banyandb.v1.client.DataPoint;
import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.banyandb.v1.client.PairQueryCondition;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.query.PaginationUtils;
import org.apache.skywalking.oap.server.core.query.enumeration.Order;
@@ -56,48 +57,7 @@ public class BanyanDBEventQueryDAO extends AbstractBanyanDBDAO implements IEvent
@Override
public Events queryEvents(EventQueryCondition condition) throws Exception {
MeasureQueryResponse resp = query(Event.INDEX_NAME, TAGS,
- Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- if (!isNullOrEmpty(condition.getUuid())) {
- query.and(eq(Event.UUID, condition.getUuid()));
- }
- final Source source = condition.getSource();
- if (source != null) {
- if (!isNullOrEmpty(source.getService())) {
- query.and(eq(Event.SERVICE, source.getService()));
- }
- if (!isNullOrEmpty(source.getServiceInstance())) {
- query.and(eq(Event.SERVICE_INSTANCE, source.getServiceInstance()));
- }
- if (!isNullOrEmpty(source.getEndpoint())) {
- query.and(eq(Event.ENDPOINT, source.getEndpoint()));
- }
- }
-
- if (!isNullOrEmpty(condition.getName())) {
- query.and(eq(Event.NAME, condition.getName()));
- }
-
- if (condition.getType() != null) {
- query.and(eq(Event.TYPE, condition.getType().name()));
- }
-
- final Duration startTime = condition.getTime();
- if (startTime != null) {
- if (startTime.getStartTimestamp() > 0) {
- query.and(gte(Event.START_TIME, startTime.getStartTimestamp()));
- }
- if (startTime.getEndTimestamp() > 0) {
- query.and(lte(Event.END_TIME, startTime.getEndTimestamp()));
- }
- }
-
- if (!isNullOrEmpty(condition.getLayer())) {
- query.and(eq(Event.LAYER, Layer.valueOf(condition.getLayer()).value()));
- }
- }
- });
+ Collections.emptySet(), buildQuery(Collections.singletonList(condition)));
Events events = new Events();
if (resp.size() == 0) {
return events;
@@ -105,18 +65,87 @@ public class BanyanDBEventQueryDAO extends AbstractBanyanDBDAO implements IEvent
for (final DataPoint dataPoint : resp.getDataPoints()) {
events.getEvents().add(buildEventView(dataPoint));
}
- sortEvents(events, condition);
return events;
}
@Override
public Events queryEvents(List<EventQueryCondition> conditionList) throws Exception {
- Events totalEvents = new Events();
- for (final EventQueryCondition cond : conditionList) {
- final Events singleEvents = this.queryEvents(cond);
- totalEvents.getEvents().addAll(singleEvents.getEvents());
+ MeasureQueryResponse resp = query(Event.INDEX_NAME, TAGS,
+ Collections.emptySet(), buildQuery(conditionList));
+ Events events = new Events();
+ if (resp.size() == 0) {
+ return events;
+ }
+ for (final DataPoint dataPoint : resp.getDataPoints()) {
+ events.getEvents().add(buildEventView(dataPoint));
}
- return totalEvents;
+ return events;
+ }
+
+ public QueryBuilder<MeasureQuery> buildQuery(final List<EventQueryCondition> conditionList) {
+ EventQueryCondition condition = conditionList.get(0);
+ final Order queryOrder = isNull(condition.getOrder()) ? Order.DES : condition.getOrder();
+ final PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
+
+ return new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ List<AbstractCriteria> eventsQueryConditions = new ArrayList<>(conditionList.size());
+ query.limit(page.getLimit());
+ query.offset(page.getFrom());
+ if (queryOrder == Order.ASC) {
+ query.setOrderBy(asc(Event.START_TIME));
+ } else {
+ query.setOrderBy(desc(Event.START_TIME));
+ }
+ for (final EventQueryCondition condition : conditionList) {
+ List<PairQueryCondition<?>> queryConditions = new ArrayList<>();
+ if (!isNullOrEmpty(condition.getUuid())) {
+ queryConditions.add(eq(Event.UUID, condition.getUuid()));
+ }
+ final Source source = condition.getSource();
+ if (source != null) {
+ if (!isNullOrEmpty(source.getService())) {
+ queryConditions.add(eq(Event.SERVICE, source.getService()));
+ }
+ if (!isNullOrEmpty(source.getServiceInstance())) {
+ queryConditions.add(eq(Event.SERVICE_INSTANCE, source.getServiceInstance()));
+ }
+ if (!isNullOrEmpty(source.getEndpoint())) {
+ queryConditions.add(eq(Event.ENDPOINT, source.getEndpoint()));
+ }
+ }
+
+ if (!isNullOrEmpty(condition.getName())) {
+ queryConditions.add(eq(Event.NAME, condition.getName()));
+ }
+
+ if (condition.getType() != null) {
+ queryConditions.add(eq(Event.TYPE, condition.getType().name()));
+ }
+
+ final Duration startTime = condition.getTime();
+ if (startTime != null) {
+ if (startTime.getStartTimestamp() > 0) {
+ queryConditions.add(gte(Event.START_TIME, startTime.getStartTimestamp()));
+ }
+ if (startTime.getEndTimestamp() > 0) {
+ queryConditions.add(lte(Event.END_TIME, startTime.getEndTimestamp()));
+ }
+ }
+
+ if (!isNullOrEmpty(condition.getLayer())) {
+ queryConditions.add(eq(Event.LAYER, Layer.valueOf(condition.getLayer()).value()));
+ }
+ eventsQueryConditions.add(and(queryConditions));
+ }
+ if (eventsQueryConditions.size() == 1) {
+ query.criteria(eventsQueryConditions.get(0));
+ } else if (eventsQueryConditions.size() > 1) {
+ query.criteria(or(eventsQueryConditions));
+ }
+ }
+ };
}
protected org.apache.skywalking.oap.server.core.query.type.event.Event buildEventView(
@@ -142,30 +171,4 @@ public class BanyanDBEventQueryDAO extends AbstractBanyanDBDAO implements IEvent
return event;
}
-
- private void sortEvents(Events events, EventQueryCondition condition) {
- if (events.getEvents().isEmpty()) {
- return;
- }
-
- final Comparator<org.apache.skywalking.oap.server.core.query.type.event.Event> c =
- buildComparator(isNull(condition.getOrder()) ? Order.DES : condition.getOrder());
- final PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
- events.setEvents(
- events.getEvents()
- .stream()
- .sorted(c)
- .skip(page.getFrom())
- .limit(page.getLimit())
- .collect(Collectors.toList())
- );
- }
-
- private Comparator<org.apache.skywalking.oap.server.core.query.type.event.Event> buildComparator(Order queryOrder) {
- Comparator<org.apache.skywalking.oap.server.core.query.type.event.Event> c = Comparator.comparingLong(org.apache.skywalking.oap.server.core.query.type.event.Event::getStartTime);
- if (queryOrder == Order.DES) {
- c = c.reversed();
- }
- return c;
- }
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
index e666929a76..d10b92a22f 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
@@ -167,7 +167,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
@Override
protected void apply(MeasureQuery query) {
if (StringUtil.isNotEmpty(instanceId)) {
- query.andWithID(instanceId);
+ query.and(id(instanceId));
}
}
});
@@ -204,17 +204,17 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
@Override
public List<Process> listProcesses(String serviceId, ProfilingSupportStatus supportStatus, long lastPingStartTimeBucket, long lastPingEndTimeBucket) throws IOException {
MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
- PROCESS_TRAFFIC_TAGS,
- Collections.emptySet(),
- new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
- query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
- query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, supportStatus.value()));
- query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
- }
- });
+ PROCESS_TRAFFIC_TAGS,
+ Collections.emptySet(),
+ new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
+ query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
+ query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, supportStatus.value()));
+ query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ }
+ });
final List<Process> processes = new ArrayList<>();
@@ -228,20 +228,19 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
@Override
public List<Process> listProcesses(String serviceInstanceId, Duration duration, boolean includeVirtual) throws IOException {
long lastPingStartTimeBucket = duration.getStartTimeBucket();
- long lastPingEndTimeBucket = duration.getEndTimeBucket();
MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
- PROCESS_TRAFFIC_TAGS,
- Collections.emptySet(),
- new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- query.and(eq(ProcessTraffic.INSTANCE_ID, serviceInstanceId));
- query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
- if (!includeVirtual) {
- query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ PROCESS_TRAFFIC_TAGS,
+ Collections.emptySet(),
+ new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ query.and(eq(ProcessTraffic.INSTANCE_ID, serviceInstanceId));
+ query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
+ if (!includeVirtual) {
+ query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ }
}
- }
- });
+ });
final List<Process> processes = new ArrayList<>();
@@ -255,15 +254,15 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
@Override
public List<Process> listProcesses(String agentId) throws IOException {
MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
- PROCESS_TRAFFIC_TAGS,
- Collections.emptySet(),
- new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- query.and(eq(ProcessTraffic.AGENT_ID, agentId));
- query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
- }
- });
+ PROCESS_TRAFFIC_TAGS,
+ Collections.emptySet(),
+ new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ query.and(eq(ProcessTraffic.AGENT_ID, agentId));
+ query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ }
+ });
final List<Process> processes = new ArrayList<>();
@@ -277,41 +276,41 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
@Override
public long getProcessCount(String serviceId, ProfilingSupportStatus profilingSupportStatus, long lastPingStartTimeBucket, long lastPingEndTimeBucket) throws IOException {
MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
- PROCESS_TRAFFIC_TAGS,
- Collections.emptySet(),
- new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
- query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
- query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, profilingSupportStatus.value()));
- query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
- }
- });
+ PROCESS_TRAFFIC_TAGS,
+ Collections.emptySet(),
+ new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
+ query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
+ query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, profilingSupportStatus.value()));
+ query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ }
+ });
return resp.getDataPoints()
- .stream()
- .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
- .size();
+ .stream()
+ .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
+ .size();
}
@Override
public long getProcessCount(String instanceId) throws IOException {
MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
- PROCESS_TRAFFIC_TAGS,
- Collections.emptySet(),
- new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- query.and(eq(ProcessTraffic.INSTANCE_ID, instanceId));
- query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
- }
- });
+ PROCESS_TRAFFIC_TAGS,
+ Collections.emptySet(),
+ new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ query.and(eq(ProcessTraffic.INSTANCE_ID, instanceId));
+ query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ }
+ });
return resp.getDataPoints()
- .stream()
- .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
- .size();
+ .stream()
+ .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
+ .size();
}
@Override
@@ -323,7 +322,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
@Override
protected void apply(MeasureQuery query) {
if (StringUtil.isNotEmpty(processId)) {
- query.andWithID(processId);
+ query.and(id(processId));
}
}
});
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java
index f0343e5579..0b24eb0e29 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java
@@ -37,6 +37,7 @@ import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractB
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
@Slf4j
@@ -51,25 +52,26 @@ public class BanyanDBMetricsDAO extends AbstractBanyanDBDAO implements IMetricsD
@Override
public List<Metrics> multiGet(Model model, List<Metrics> metrics) throws IOException {
log.info("multiGet {} from BanyanDB", model.getName());
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
if (schema == null) {
throw new IOException(model.getName() + " is not registered");
}
- // TODO: add time range
List<Metrics> metricsInStorage = new ArrayList<>(metrics.size());
- for (final Metrics missCachedMetric : metrics) {
- MeasureQueryResponse resp = query(model.getName(), schema.getTags(), schema.getFields(), new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- query.andWithID(missCachedMetric.id());
+ MeasureQueryResponse resp = query(model.getName(), schema.getTags(), schema.getFields(), new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ for (final Metrics missCachedMetric : metrics) {
+ query.or(id(missCachedMetric.id()));
}
- });
- if (resp.size() == 0) {
- continue;
- }
- for (final DataPoint dataPoint : resp.getDataPoints()) {
- metricsInStorage.add(storageBuilder.storage2Entity(new BanyanDBConverter.StorageToMeasure(model.getName(), dataPoint)));
}
+ });
+
+ if (resp.size() == 0) {
+ return Collections.emptyList();
+ }
+
+ for (final DataPoint dataPoint : resp.getDataPoints()) {
+ metricsInStorage.add(storageBuilder.storage2Entity(new BanyanDBConverter.StorageToMeasure(schema, dataPoint)));
}
return metricsInStorage;
}
@@ -77,12 +79,12 @@ public class BanyanDBMetricsDAO extends AbstractBanyanDBDAO implements IMetricsD
@Override
public InsertRequest prepareBatchInsert(Model model, Metrics metrics) throws IOException {
log.info("prepare to insert {}", model.getName());
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
if (schema == null) {
throw new IOException(model.getName() + " is not registered");
}
MeasureWrite measureWrite = new MeasureWrite(schema.getMetadata().getGroup(), // group name
- model.getName(), // index-name
+ schema.getMetadata().name(), // measure-name
TimeBucket.getTimestamp(metrics.getTimeBucket(), model.getDownsampling())); // timestamp
final BanyanDBConverter.MeasureToStorage toStorage = new BanyanDBConverter.MeasureToStorage(schema, measureWrite);
storageBuilder.entity2Storage(metrics, toStorage);
@@ -93,12 +95,12 @@ public class BanyanDBMetricsDAO extends AbstractBanyanDBDAO implements IMetricsD
@Override
public UpdateRequest prepareBatchUpdate(Model model, Metrics metrics) throws IOException {
log.info("prepare to update {}", model.getName());
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
if (schema == null) {
throw new IOException(model.getName() + " is not registered");
}
MeasureWrite measureWrite = new MeasureWrite(schema.getMetadata().getGroup(), // group name
- model.getName(), // index-name
+ schema.getMetadata().name(), // measure-name
TimeBucket.getTimestamp(metrics.getTimeBucket(), model.getDownsampling())); // timestamp
final BanyanDBConverter.MeasureToStorage toStorage = new BanyanDBConverter.MeasureToStorage(schema, measureWrite);
storageBuilder.entity2Storage(metrics, toStorage);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
index 74cc787625..ee306052c8 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
@@ -19,13 +19,15 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.measure;
import com.google.common.collect.ImmutableSet;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
+
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.banyandb.v1.client.DataPoint;
import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
@@ -48,6 +50,7 @@ import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.util.ByteUtil;
+@Slf4j
public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMetricsQueryDAO {
public BanyanDBMetricsQueryDAO(BanyanDBStorageClient client) {
super(client);
@@ -55,6 +58,12 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
@Override
public long readMetricsValue(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
+ String modelName = condition.getName();
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName, duration.getStep());
+ if (schema == null) {
+ throw new IOException("schema is not registered");
+ }
+
int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName());
Function function = ValueColumnMetadata.INSTANCE.getValueFunction(condition.getName());
if (function == Function.Latest) {
@@ -62,9 +71,8 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
.getValues().latestValue(defaultValue);
}
- final String modelName = condition.getName();
TimestampRange timestampRange = new TimestampRange(duration.getStartTimestamp(), duration.getEndTimestamp());
- final MeasureQueryResponse resp = query(modelName,
+ final MeasureQueryResponse resp = query(schema,
ImmutableSet.of(Metrics.ENTITY_ID),
ImmutableSet.of(valueColumnName),
timestampRange,
@@ -99,18 +107,17 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
@Override
public MetricsValues readMetricsValues(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
String modelName = condition.getName();
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName, duration.getStep());
if (schema == null) {
throw new IOException("schema is not registered");
}
- final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
- final List<String> ids = pointOfTimes.stream().map(pointOfTime -> {
- String id = pointOfTime.id(condition.getEntity().buildId());
- return id;
- }).collect(Collectors.toList());
+
+ final String entityID = condition.getEntity().buildId();
+ Map<String, DataPoint> idMap = queryByEntityID(schema, valueColumnName, duration, entityID);
+
+ List<String> ids = extractMeasureIDs(duration, entityID);
MetricsValues metricsValues = new MetricsValues();
- Map<String, DataPoint> idMap = queryIDs(modelName, valueColumnName, ids);
if (!idMap.isEmpty()) {
// Label is null, because in readMetricsValues, no label parameter.
IntValues intValues = metricsValues.getValues();
@@ -150,15 +157,10 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
@Override
public List<MetricsValues> readLabeledMetricsValues(MetricsCondition condition, String valueColumnName, List<String> labels, Duration duration) throws IOException {
- final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
- String modelName = condition.getName();
- List<String> ids = new ArrayList<>(pointOfTimes.size());
- pointOfTimes.forEach(pointOfTime -> {
- String id = pointOfTime.id(condition.getEntity().buildId());
- ids.add(id);
- });
+ Map<String, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
+
+ List<String> ids = extractMeasureIDs(duration, condition.getEntity().buildId());
- Map<String, DataPoint> idMap = queryIDs(modelName, valueColumnName, ids);
Map<String, DataTable> dataTableMap = new HashMap<>(idMap.size());
for (final Map.Entry<String, DataPoint> entry : idMap.entrySet()) {
dataTableMap.put(
@@ -168,29 +170,23 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
}
return Util.sortValues(
- Util.composeLabelValue(condition, labels, ids, dataTableMap),
- ids,
- ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName())
+ Util.composeLabelValue(condition, labels, ids, dataTableMap),
+ ids,
+ ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName())
);
}
@Override
public HeatMap readHeatMap(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
- final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
- String modelName = condition.getName();
- List<String> ids = new ArrayList<>(pointOfTimes.size());
- pointOfTimes.forEach(pointOfTime -> {
- String id = pointOfTime.id(condition.getEntity().buildId());
- ids.add(id);
- });
+ Map<String, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
HeatMap heatMap = new HeatMap();
- Map<String, DataPoint> idMap = queryIDs(modelName, valueColumnName, ids);
-
if (idMap.isEmpty()) {
return heatMap;
}
+ List<String> ids = extractMeasureIDs(duration, condition.getEntity().buildId());
+
final int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName());
for (String id : ids) {
DataPoint dataPoint = idMap.get(id);
@@ -205,19 +201,40 @@ public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMet
return heatMap;
}
- private Map<String, DataPoint> queryIDs(String modelName, String valueColumnName, List<String> measureIDs) throws IOException {
- Map<String, DataPoint> map = new HashMap<>(measureIDs.size());
- for (final String id : measureIDs) {
- MeasureQueryResponse resp = query(modelName, Collections.emptySet(), ImmutableSet.of(valueColumnName), new QueryBuilder<MeasureQuery>() {
- @Override
- protected void apply(MeasureQuery query) {
- query.andWithID(id);
- }
- });
- if (resp.size() > 0) {
- map.putIfAbsent(resp.getDataPoints().get(0).getId(), resp.getDataPoints().get(0));
+ private List<String> extractMeasureIDs(Duration duration, String entityID) {
+ final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
+ List<String> ids = new ArrayList<>(pointOfTimes.size());
+ pointOfTimes.forEach(pointOfTime -> {
+ String id = pointOfTime.id(entityID);
+ ids.add(id);
+ });
+ return ids;
+ }
+
+ private Map<String, DataPoint> queryByEntityID(final MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
+ final MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(condition.getName(), duration.getStep());
+ if (schema == null) {
+ throw new IOException("schema is not registered");
+ }
+ return queryByEntityID(schema, valueColumnName, duration, condition.getEntity().buildId());
+ }
+
+ private Map<String, DataPoint> queryByEntityID(MetadataRegistry.Schema schema, String valueColumnName, Duration duration, String entityID) throws IOException {
+ TimestampRange timestampRange = new TimestampRange(duration.getStartTimestamp(), duration.getEndTimestamp());
+
+ Map<String, DataPoint> map = new HashMap<>();
+ MeasureQueryResponse resp = query(schema, Collections.emptySet(), ImmutableSet.of(valueColumnName), timestampRange, new QueryBuilder<MeasureQuery>() {
+ @Override
+ protected void apply(MeasureQuery query) {
+ query.and(eq(Metrics.ENTITY_ID, entityID));
+ }
+ });
+ for (final DataPoint dp : resp.getDataPoints()) {
+ if (map.putIfAbsent(dp.getId(), dp) != null) {
+ log.warn("duplicated data point");
}
}
+
return map;
}
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java
index 94bf71d726..679568e24e 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java
@@ -29,16 +29,20 @@ import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
import org.apache.skywalking.oap.server.core.analysis.manual.networkalias.NetworkAddressAlias;
import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBConverter.StorageToMeasure;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
@Slf4j
public class BanyanDBNetworkAddressAliasDAO extends AbstractBanyanDBDAO implements INetworkAddressAliasDAO {
private final NetworkAddressAlias.Builder builder = new NetworkAddressAlias.Builder();
+ private MetadataRegistry.Schema schema;
+
private static final Set<String> TAGS = ImmutableSet.of(NetworkAddressAlias.ADDRESS,
NetworkAddressAlias.TIME_BUCKET, NetworkAddressAlias.LAST_UPDATE_TIME_BUCKET,
NetworkAddressAlias.REPRESENT_SERVICE_ID, NetworkAddressAlias.REPRESENT_SERVICE_INSTANCE_ID);
@@ -47,6 +51,13 @@ public class BanyanDBNetworkAddressAliasDAO extends AbstractBanyanDBDAO implemen
super(client);
}
+ private MetadataRegistry.Schema getSchema() {
+ if (schema == null) {
+ schema = MetadataRegistry.INSTANCE.findMetadata(NetworkAddressAlias.INDEX_NAME, DownSampling.Minute);
+ }
+ return schema;
+ }
+
@Override
public List<NetworkAddressAlias> loadLastUpdate(long timeBucket) {
try {
@@ -61,10 +72,14 @@ public class BanyanDBNetworkAddressAliasDAO extends AbstractBanyanDBDAO implemen
}
}
);
+ /**
+ * Currently, only used by {@link org.apache.skywalking.oap.server.storage.plugin.banyandb.measure.BanyanDBNetworkAddressAliasDAO}.
+ * The default DownSampling strategy, i.e. {@link DownSampling#Minute} is assumed in this case.
+ */
return resp.getDataPoints()
.stream()
.map(
- point -> builder.storage2Entity(new StorageToMeasure(NetworkAddressAlias.INDEX_NAME, point))
+ point -> builder.storage2Entity(new StorageToMeasure(getSchema(), point))
)
.collect(Collectors.toList());
} catch (IOException e) {
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
index bee2069770..4978b3f64a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
@@ -18,13 +18,18 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
+import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
import org.apache.skywalking.banyandb.v1.client.AbstractQuery;
+import org.apache.skywalking.banyandb.v1.client.And;
import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.banyandb.v1.client.Or;
import org.apache.skywalking.banyandb.v1.client.PairQueryCondition;
import org.apache.skywalking.banyandb.v1.client.StreamQuery;
import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
import org.apache.skywalking.banyandb.v1.client.TimestampRange;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.storage.AbstractDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
@@ -33,6 +38,7 @@ import java.io.IOException;
import java.time.Instant;
import java.util.List;
import java.util.Set;
+import java.util.function.BiFunction;
public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageClient> {
private static final Instant UPPER_BOUND = Instant.ofEpochSecond(0, Long.MAX_VALUE);
@@ -43,21 +49,21 @@ public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageCli
super(client);
}
- protected StreamQueryResponse query(String modelName, Set<String> tags, QueryBuilder<StreamQuery> builder) throws IOException {
- return this.query(modelName, tags, null, builder);
+ protected StreamQueryResponse query(String streamModelName, Set<String> tags, QueryBuilder<StreamQuery> builder) throws IOException {
+ return this.query(streamModelName, tags, null, builder);
}
- protected StreamQueryResponse query(String modelName, Set<String> tags, TimestampRange timestampRange,
+ protected StreamQueryResponse query(String streamModelName, Set<String> tags, TimestampRange timestampRange,
QueryBuilder<StreamQuery> builder) throws IOException {
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findRecordMetadata(streamModelName);
if (schema == null) {
- throw new IllegalStateException("schema is not registered");
+ throw new IllegalArgumentException("schema is not registered");
}
final StreamQuery query;
if (timestampRange == null) {
- query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), LARGEST_TIME_RANGE, tags);
+ query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), LARGEST_TIME_RANGE, tags);
} else {
- query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), timestampRange, tags);
+ query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), timestampRange, tags);
}
builder.apply(query);
@@ -65,22 +71,27 @@ public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageCli
return getClient().query(query);
}
- protected MeasureQueryResponse query(String modelName, Set<String> tags, Set<String> fields,
+ protected MeasureQueryResponse query(String measureModelName, Set<String> tags, Set<String> fields,
QueryBuilder<MeasureQuery> builder) throws IOException {
- return this.query(modelName, tags, fields, null, builder);
+ return this.query(measureModelName, tags, fields, null, builder);
}
- protected MeasureQueryResponse query(String modelName, Set<String> tags, Set<String> fields,
+ protected MeasureQueryResponse query(String measureModelName, Set<String> tags, Set<String> fields,
TimestampRange timestampRange, QueryBuilder<MeasureQuery> builder) throws IOException {
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(measureModelName, DownSampling.Minute);
if (schema == null) {
- throw new IllegalStateException("schema is not registered");
+ throw new IllegalArgumentException("measure is not registered");
}
+ return this.query(schema, tags, fields, timestampRange, builder);
+ }
+
+ protected MeasureQueryResponse query(MetadataRegistry.Schema schema, Set<String> tags, Set<String> fields,
+ TimestampRange timestampRange, QueryBuilder<MeasureQuery> builder) throws IOException {
final MeasureQuery query;
if (timestampRange == null) {
- query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), LARGEST_TIME_RANGE, tags, fields);
+ query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), LARGEST_TIME_RANGE, tags, fields);
} else {
- query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), timestampRange, tags, fields);
+ query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), timestampRange, tags, fields);
}
builder.apply(query);
@@ -126,5 +137,44 @@ public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageCli
protected PairQueryCondition<Long> ne(String name, long value) {
return PairQueryCondition.LongQueryCondition.ne(name, value);
}
+
+ protected PairQueryCondition<String> id(String value) {
+ return PairQueryCondition.IDQueryCondition.eq(Metrics.ID, value);
+ }
+
+ protected AbstractQuery.OrderBy desc(String name) {
+ return new AbstractQuery.OrderBy(name, AbstractQuery.Sort.DESC);
+ }
+
+ protected AbstractQuery.OrderBy asc(String name) {
+ return new AbstractQuery.OrderBy(name, AbstractQuery.Sort.ASC);
+ }
+
+ protected AbstractCriteria and(List<? extends AbstractCriteria> conditions) {
+ if (conditions.isEmpty()) {
+ return null;
+ }
+ if (conditions.size() == 1) {
+ return conditions.get(0);
+ }
+
+ return conditions.subList(2, conditions.size()).stream().reduce(
+ And.create(conditions.get(0), conditions.get(1)),
+ (BiFunction<AbstractCriteria, AbstractCriteria, AbstractCriteria>) And::create,
+ And::create);
+ }
+
+ protected AbstractCriteria or(List<? extends AbstractCriteria> conditions) {
+ if (conditions.isEmpty()) {
+ return null;
+ }
+ if (conditions.size() == 1) {
+ return conditions.get(0);
+ }
+ return conditions.subList(2, conditions.size()).stream().reduce(
+ Or.create(conditions.get(0), conditions.get(1)),
+ (BiFunction<AbstractCriteria, AbstractCriteria, AbstractCriteria>) Or::create,
+ Or::create);
+ }
}
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java
index 2ac1be38c7..1733953db8 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java
@@ -19,7 +19,6 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
import com.google.common.collect.ImmutableSet;
-import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.skywalking.banyandb.v1.client.RowEntity;
import org.apache.skywalking.banyandb.v1.client.StreamQuery;
import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
@@ -30,7 +29,6 @@ import org.apache.skywalking.oap.server.core.browser.source.BrowserErrorCategory
import org.apache.skywalking.oap.server.core.query.input.Duration;
import org.apache.skywalking.oap.server.core.query.type.BrowserErrorLog;
import org.apache.skywalking.oap.server.core.query.type.BrowserErrorLogs;
-import org.apache.skywalking.oap.server.core.query.type.ErrorCategory;
import org.apache.skywalking.oap.server.core.storage.query.IBrowserLogQueryDAO;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
@@ -98,33 +96,4 @@ public class BanyanDBBrowserLogQueryDAO extends AbstractBanyanDBDAO implements I
}
return logs;
}
-
- /**
- * TODO: merge the default method in the interface
- */
- private BrowserErrorLog parserDataBinary(
- byte[] dataBinary) {
- try {
- BrowserErrorLog log = new BrowserErrorLog();
- org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog browserErrorLog = org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog
- .parseFrom(dataBinary);
-
- log.setService(browserErrorLog.getService());
- log.setServiceVersion(browserErrorLog.getServiceVersion());
- log.setTime(browserErrorLog.getTime());
- log.setPagePath(browserErrorLog.getPagePath());
- log.setCategory(ErrorCategory.valueOf(browserErrorLog.getCategory().name().toUpperCase()));
- log.setGrade(browserErrorLog.getGrade());
- log.setMessage(browserErrorLog.getMessage());
- log.setLine(browserErrorLog.getLine());
- log.setCol(browserErrorLog.getCol());
- log.setStack(browserErrorLog.getStack());
- log.setErrorUrl(browserErrorLog.getErrorUrl());
- log.setFirstReportedError(browserErrorLog.getFirstReportedError());
-
- return log;
- } catch (InvalidProtocolBufferException e) {
- throw new RuntimeException(e);
- }
- }
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java
index 6c3c9d35de..12c9205e54 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java
@@ -19,8 +19,6 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
import com.google.common.collect.ImmutableSet;
-import com.google.protobuf.InvalidProtocolBufferException;
-import org.apache.skywalking.apm.network.logging.v3.LogTags;
import org.apache.skywalking.banyandb.v1.client.RowEntity;
import org.apache.skywalking.banyandb.v1.client.StreamQuery;
import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
@@ -34,7 +32,6 @@ import org.apache.skywalking.oap.server.core.query.enumeration.Order;
import org.apache.skywalking.oap.server.core.query.input.Duration;
import org.apache.skywalking.oap.server.core.query.input.TraceScopeCondition;
import org.apache.skywalking.oap.server.core.query.type.ContentType;
-import org.apache.skywalking.oap.server.core.query.type.KeyValue;
import org.apache.skywalking.oap.server.core.query.type.Log;
import org.apache.skywalking.oap.server.core.query.type.Logs;
import org.apache.skywalking.oap.server.core.storage.query.ILogQueryDAO;
@@ -149,17 +146,4 @@ public class BanyanDBLogQueryDAO extends AbstractBanyanDBDAO implements ILogQuer
}
return logs;
}
-
- /**
- * Parser the raw tags.
- * TODO: merge default method
- */
- private void parserDataBinary(byte[] dataBinary, List<KeyValue> tags) {
- try {
- LogTags logTags = LogTags.parseFrom(dataBinary);
- logTags.getDataList().forEach(pair -> tags.add(new KeyValue(pair.getKey(), pair.getValue())));
- } catch (InvalidProtocolBufferException e) {
- throw new RuntimeException(e);
- }
- }
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java
index 2def417448..29108e6e9c 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java
@@ -103,9 +103,9 @@ public class BanyanDBProfileTaskQueryDAO extends AbstractBanyanDBDAO implements
@Override
protected void apply(StreamQuery query) {
if (StringUtil.isNotEmpty(id)) {
- // TODO: support search by ID
+ query.and(eq(ProfileTaskRecord.TASK_ID, id));
}
- // query.setLimit(1);
+ query.setLimit(1);
}
});
@@ -113,9 +113,7 @@ public class BanyanDBProfileTaskQueryDAO extends AbstractBanyanDBDAO implements
return null;
}
- RowEntity first = resp.getElements().stream().filter(e -> id.equals(e.getTagValue(ProfileTaskRecord.TASK_ID)))
- .findFirst().orElse(null);
- return first == null ? null : buildProfileTask(first);
+ return buildProfileTask(resp.getElements().get(0));
}
private ProfileTask buildProfileTask(RowEntity data) {
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
index a0bff696b6..985544bbe9 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
@@ -34,12 +34,9 @@ import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageC
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
-import java.util.function.Function;
-import java.util.stream.Collectors;
/**
* {@link ProfileThreadSnapshotRecord} is a stream
@@ -92,6 +89,7 @@ public class BanyanDBProfileThreadSnapshotQueryDAO extends AbstractBanyanDBDAO i
public void apply(StreamQuery query) {
query.and(eq(ProfileThreadSnapshotRecord.TASK_ID, taskId))
.and(eq(ProfileThreadSnapshotRecord.SEQUENCE, 0L));
+ query.setOrderBy(desc(ProfileThreadSnapshotRecord.DUMP_TIME));
query.setLimit(querySegmentMaxSize);
}
});
@@ -105,43 +103,41 @@ public class BanyanDBProfileThreadSnapshotQueryDAO extends AbstractBanyanDBDAO i
segmentIds.add(rowEntity.getTagValue(ProfileThreadSnapshotRecord.SEGMENT_ID));
}
- // TODO: support `IN` or `OR` logic operation in BanyanDB
- List<BasicTrace> basicTraces = new ArrayList<>();
- for (String segmentID : segmentIds) {
- final StreamQueryResponse segmentRecordResp = query(SegmentRecord.INDEX_NAME,
- TAGS_TRACE,
- new QueryBuilder<StreamQuery>() {
- @Override
- public void apply(StreamQuery traceQuery) {
- traceQuery.and(eq(SegmentRecord.SEGMENT_ID, segmentID));
- }
- });
-
- for (final RowEntity row : segmentRecordResp.getElements()) {
- BasicTrace basicTrace = new BasicTrace();
-
- basicTrace.setSegmentId(row.getId());
- basicTrace.setStart(String.valueOf((Number) row.getTagValue(SegmentRecord.START_TIME)));
- basicTrace.getEndpointNames().add(IDManager.EndpointID.analysisId(
- row.getTagValue(SegmentRecord.ENDPOINT_ID)
- ).getEndpointName());
- basicTrace.setDuration(((Number) row.getTagValue(SegmentRecord.LATENCY)).intValue());
- basicTrace.setError(BooleanUtils.valueToBoolean(
- ((Number) row.getTagValue(SegmentRecord.IS_ERROR)).intValue()
- ));
- basicTrace.getTraceIds().add(row.getTagValue(SegmentRecord.TRACE_ID));
-
- basicTraces.add(basicTrace);
- }
+ if (segmentIds.isEmpty()) {
+ return Collections.emptyList();
}
- // TODO: Sort in DB with DESC
- basicTraces = basicTraces.stream()
- // comparing start_time
- .sorted(Comparator.comparing((Function<BasicTrace, Long>) basicTrace -> Long.parseLong(basicTrace.getStart()))
- // and sort in reverse order
- .reversed())
- .collect(Collectors.toList());
+ final StreamQueryResponse segmentRecordResp = query(SegmentRecord.INDEX_NAME,
+ TAGS_TRACE,
+ new QueryBuilder<StreamQuery>() {
+ @Override
+ public void apply(StreamQuery traceQuery) {
+ // TODO: use "in" operator
+ for (final String segmentID : segmentIds) {
+ traceQuery.or(eq(SegmentRecord.SEGMENT_ID, segmentID));
+ }
+ traceQuery.setLimit(segmentIds.size());
+ traceQuery.setOrderBy(desc(SegmentRecord.START_TIME));
+ }
+ });
+
+ List<BasicTrace> basicTraces = new ArrayList<>();
+ for (final RowEntity row : segmentRecordResp.getElements()) {
+ BasicTrace basicTrace = new BasicTrace();
+
+ basicTrace.setSegmentId(row.getId());
+ basicTrace.setStart(String.valueOf((Number) row.getTagValue(SegmentRecord.START_TIME)));
+ basicTrace.getEndpointNames().add(IDManager.EndpointID.analysisId(
+ row.getTagValue(SegmentRecord.ENDPOINT_ID)
+ ).getEndpointName());
+ basicTrace.setDuration(((Number) row.getTagValue(SegmentRecord.LATENCY)).intValue());
+ basicTrace.setError(BooleanUtils.valueToBoolean(
+ ((Number) row.getTagValue(SegmentRecord.IS_ERROR)).intValue()
+ ));
+ basicTrace.getTraceIds().add(row.getTagValue(SegmentRecord.TRACE_ID));
+
+ basicTraces.add(basicTrace);
+ }
return basicTraces;
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java
index 1598dc73d1..11b0d9abec 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java
@@ -38,7 +38,7 @@ public class BanyanDBRecordDAO implements IRecordDAO {
@Override
public InsertRequest prepareBatchInsert(Model model, Record record) throws IOException {
- MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+ MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
if (schema == null) {
throw new IOException(model.getName() + " is not registered");
}
diff --git a/test/e2e-v2/script/env b/test/e2e-v2/script/env
index 7c08ebc39c..50ce99a58e 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=0f3ec68e5a7e1608cec8688716b848ed15e971e5
SW_ROVER_COMMIT=d956eaede57b62108b78bca48045bd09ba88e653
-SW_BANYANDB_COMMIT=5a326d7e36a008c5ea10e3ae506309cb29733c53
+SW_BANYANDB_COMMIT=6b344929519dfd90203076e471fbf1136bde2262
SW_CTL_COMMIT=e684fae0107045fc23799146d62f04cb68bd5a3b