You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2024/11/05 03:08:26 UTC
(skywalking) 01/01: Add batch settings and fix trace query bug
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch banyandb-settings
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 5568695f6b0aef06ccd8c65327b5bf8ac0e7776a
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Nov 5 11:07:27 2024 +0800
Add batch settings and fix trace query bug
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
docs/en/setup/backend/configuration-vocabulary.md | 4 ++++
docs/en/setup/backend/storages/banyandb.md | 6 +++++-
.../server-starter/src/main/resources/application.yml | 6 +++++-
.../storage/plugin/banyandb/BanyanDBStorageConfig.java | 5 +++++
.../storage/plugin/banyandb/BanyanDBStorageProvider.java | 14 +++++++-------
.../plugin/banyandb/measure/BanyanDBHierarchyQueryDAO.java | 6 +++++-
.../plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java | 12 +++++++++++-
.../banyandb/measure/BanyanDBNetworkAddressAliasDAO.java | 6 +++++-
.../plugin/banyandb/measure/BanyanDBServiceLabelDAO.java | 6 +++++-
.../banyandb/stream/BanyanDBEBPFProfilingDataDAO.java | 5 ++++-
.../banyandb/stream/BanyanDBSpanAttachedEventQueryDAO.java | 5 ++++-
.../plugin/banyandb/stream/BanyanDBTraceQueryDAO.java | 12 +++++++++++-
.../data-generator/src/main/resources/application.yml | 6 +++++-
13 files changed, 76 insertions(+), 17 deletions(-)
diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md
index 483753ed66..bf79d2a889 100644
--- a/docs/en/setup/backend/configuration-vocabulary.md
+++ b/docs/en/setup/backend/configuration-vocabulary.md
@@ -150,7 +150,11 @@ The Configuration Vocabulary lists all available configurations provided by `app
| - | - | superDatasetShardsFactor | Shards Factor for a super dataset record, i.e. Shard number of a super dataset is recordShardsNumber*superDatasetShardsFactor. [...]
| - | - | concurrentWriteThreads | Concurrent consumer threads for batch writing. [...]
| - | - | profileTaskQueryMaxSize | Max size of ProfileTask to be fetched. [...]
+| - | - | resultWindowMaxSize | The maximum size of dataset when the OAP loads cache, such as network aliases. [...]
+| - | - | metadataQueryMaxSize | The maximum size of metadata per query. [...]
+| - | - | segmentQueryMaxSize | The maximum size of trace segments per query. [...]
| - | - | asyncProfilerTaskQueryMaxSize | Max size of AsyncProfilerTask to be fetched. [...]
+| - | - | profileDataQueryScrollBatchSize | The batch size of query profiling data. [...]
| - | - | segmentIntervalDays | Data is stored in BanyanDB in segments. A segment is a time range of data.The segment interval is the time range of a segment.The value should be less or equal to data TTL relevant settings. [...]
| - | - | superDatasetSegmentIntervalDays | The super dataset segment interval is the time range of a segment in the super dataset. [...]
| - | - | specificGroupSettings | Specific groups settings. [...]
diff --git a/docs/en/setup/backend/storages/banyandb.md b/docs/en/setup/backend/storages/banyandb.md
index 29b98b2fa4..2a8e161c1e 100644
--- a/docs/en/setup/backend/storages/banyandb.md
+++ b/docs/en/setup/backend/storages/banyandb.md
@@ -36,8 +36,12 @@ storage:
# The number of threads that write data to BanyanDB concurrently.
# Bigger value can improve the write performance, but also increase the OAP and BanyanDB Server CPU usage.
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
+ resultWindowMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_WINDOW_SIZE:10000}
+ metadataQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_SIZE:10000}
+ segmentQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_SEGMENT_SIZE:200}
# The max number of profile task query in a request.
- profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_PROFILE_TASK_QUERY_MAX_SIZE:200}
+ profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_TASK_SIZE:200}
+ profileDataQueryBatchSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_DATA_BATCH_SIZE:100}
# Data is stored in BanyanDB in segments. A segment is a time range of data.
# The segment interval is the time range of a segment.
# The value should be less or equal to data TTL relevant settings.
diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml
index 5abb402c91..800de4d989 100644
--- a/oap-server/server-starter/src/main/resources/application.yml
+++ b/oap-server/server-starter/src/main/resources/application.yml
@@ -245,8 +245,12 @@ storage:
# The number of threads that write data to BanyanDB concurrently.
# Bigger value can improve the write performance, but also increase the OAP and BanyanDB Server CPU usage.
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
+ resultWindowMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_WINDOW_SIZE:10000}
+ metadataQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_SIZE:10000}
+ segmentQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_SEGMENT_SIZE:200}
# The max number of profile task query in a request.
- profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_PROFILE_TASK_QUERY_MAX_SIZE:200}
+ profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_TASK_SIZE:200}
+ profileDataQueryBatchSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_DATA_BATCH_SIZE:100}
asyncProfilerTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_ASYNC_PROFILER_TASK_QUERY_MAX_SIZE:200}
# Data is stored in BanyanDB in segments. A segment is a time range of data.
# The segment interval is the time range of a segment.
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 c61e5e2a83..3410234895 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
@@ -103,6 +103,11 @@ public class BanyanDBStorageConfig extends ModuleConfig {
*/
private int asyncProfilerTaskQueryMaxSize;
+ private int resultWindowMaxSize = 10000;
+ private int metadataQueryMaxSize = 5000;
+ private int segmentQueryMaxSize = 200;
+ private int profileDataQueryBatchSize = 100;
+
public String[] getTargetArray() {
return Iterables.toArray(Splitter.on(",").omitEmptyStrings().trimResults().split(this.targets), String.class);
}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
index 0c917cf7ff..807ac5c616 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
@@ -132,10 +132,10 @@ public class BanyanDBStorageProvider extends ModuleProvider {
config.getConcurrentWriteThreads()
));
this.registerServiceImplementation(StorageDAO.class, new BanyanDBStorageDAO(client));
- this.registerServiceImplementation(INetworkAddressAliasDAO.class, new BanyanDBNetworkAddressAliasDAO(client));
- this.registerServiceImplementation(ITraceQueryDAO.class, new BanyanDBTraceQueryDAO(client));
+ this.registerServiceImplementation(INetworkAddressAliasDAO.class, new BanyanDBNetworkAddressAliasDAO(client, this.config));
+ this.registerServiceImplementation(ITraceQueryDAO.class, new BanyanDBTraceQueryDAO(client, this.config.getSegmentQueryMaxSize()));
this.registerServiceImplementation(IBrowserLogQueryDAO.class, new BanyanDBBrowserLogQueryDAO(client));
- this.registerServiceImplementation(IMetadataQueryDAO.class, new BanyanDBMetadataQueryDAO(client));
+ this.registerServiceImplementation(IMetadataQueryDAO.class, new BanyanDBMetadataQueryDAO(client, this.config));
this.registerServiceImplementation(IAlarmQueryDAO.class, new BanyanDBAlarmQueryDAO(client));
this.registerServiceImplementation(ILogQueryDAO.class, new BanyanDBLogQueryDAO(client));
this.registerServiceImplementation(
@@ -155,20 +155,20 @@ public class BanyanDBStorageProvider extends ModuleProvider {
this.registerServiceImplementation(IEventQueryDAO.class, new BanyanDBEventQueryDAO(client));
this.registerServiceImplementation(ITopologyQueryDAO.class, new BanyanDBTopologyQueryDAO(client));
this.registerServiceImplementation(IEBPFProfilingTaskDAO.class, new BanyanDBEBPFProfilingTaskDAO(client));
- this.registerServiceImplementation(IEBPFProfilingDataDAO.class, new BanyanDBEBPFProfilingDataDAO(client));
+ this.registerServiceImplementation(IEBPFProfilingDataDAO.class, new BanyanDBEBPFProfilingDataDAO(client, this.config.getProfileDataQueryBatchSize()));
this.registerServiceImplementation(
IEBPFProfilingScheduleDAO.class, new BanyanDBEBPFProfilingScheduleQueryDAO(client));
this.registerServiceImplementation(IContinuousProfilingPolicyDAO.class, new BanyanDBContinuousProfilingPolicyDAO(client));
- this.registerServiceImplementation(IServiceLabelDAO.class, new BanyanDBServiceLabelDAO(client));
+ this.registerServiceImplementation(IServiceLabelDAO.class, new BanyanDBServiceLabelDAO(client, this.config));
this.registerServiceImplementation(ITagAutoCompleteQueryDAO.class, new BanyanDBTagAutocompleteQueryDAO(client));
this.registerServiceImplementation(IHistoryDeleteDAO.class, new BanyanDBHistoryDeleteDAO());
this.registerServiceImplementation(IMetricsQueryDAO.class, new BanyanDBMetricsQueryDAO(client));
this.registerServiceImplementation(IAggregationQueryDAO.class, new BanyanDBAggregationQueryDAO(client));
this.registerServiceImplementation(IRecordsQueryDAO.class, new BanyanDBRecordsQueryDAO(client));
this.registerServiceImplementation(IZipkinQueryDAO.class, new BanyanDBZipkinQueryDAO(client));
- this.registerServiceImplementation(ISpanAttachedEventQueryDAO.class, new BanyanDBSpanAttachedEventQueryDAO(client));
- this.registerServiceImplementation(IHierarchyQueryDAO.class, new BanyanDBHierarchyQueryDAO(client));
+ this.registerServiceImplementation(ISpanAttachedEventQueryDAO.class, new BanyanDBSpanAttachedEventQueryDAO(client, this.config.getProfileDataQueryBatchSize()));
+ this.registerServiceImplementation(IHierarchyQueryDAO.class, new BanyanDBHierarchyQueryDAO(client, this.config));
this.registerServiceImplementation(
IAsyncProfilerTaskQueryDAO.class, new BanyanDBAsyncProfilerTaskQueryDAO(client,
this.config.getAsyncProfilerTaskQueryMaxSize()
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBHierarchyQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBHierarchyQueryDAO.java
index c2be98910c..a733fbedd1 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBHierarchyQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBHierarchyQueryDAO.java
@@ -35,6 +35,7 @@ import org.apache.skywalking.oap.server.core.hierarchy.service.ServiceHierarchyR
import org.apache.skywalking.oap.server.core.storage.query.IHierarchyQueryDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBConverter;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageConfig;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
@@ -52,9 +53,11 @@ public class BanyanDBHierarchyQueryDAO extends AbstractBanyanDBDAO implements IH
InstanceHierarchyRelationTraffic.RELATED_INSTANCE_ID,
InstanceHierarchyRelationTraffic.RELATED_SERVICE_LAYER
);
+ private final int limit;
- public BanyanDBHierarchyQueryDAO(final BanyanDBStorageClient client) {
+ public BanyanDBHierarchyQueryDAO(final BanyanDBStorageClient client, BanyanDBStorageConfig config) {
super(client);
+ this.limit = config.getMetadataQueryMaxSize();
}
@Override
@@ -65,6 +68,7 @@ public class BanyanDBHierarchyQueryDAO extends AbstractBanyanDBDAO implements IH
Collections.emptySet(), new QueryBuilder<>() {
@Override
protected void apply(MeasureQuery query) {
+ query.limit(limit);
}
}
);
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 6b6eb28807..74d8dfd210 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
@@ -51,6 +51,7 @@ import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBConverter;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageConfig;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
@@ -81,9 +82,11 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
ProcessTraffic.PROFILING_SUPPORT_STATUS);
private static final Gson GSON = new Gson();
+ private final int limit;
- public BanyanDBMetadataQueryDAO(BanyanDBStorageClient client) {
+ public BanyanDBMetadataQueryDAO(BanyanDBStorageClient client, BanyanDBStorageConfig config) {
super(client);
+ this.limit = config.getMetadataQueryMaxSize();
}
@Override
@@ -95,6 +98,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
@Override
protected void apply(MeasureQuery query) {
+ query.limit(limit);
}
});
@@ -125,6 +129,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
}
final var minuteTimeBucket = TimeBucket.getMinuteTimeBucket(duration.getStartTimestamp());
query.and(gte(InstanceTraffic.LAST_PING_TIME_BUCKET, minuteTimeBucket));
+ query.limit(limit);
}
});
@@ -170,6 +175,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
);
}
query.criteria(or(instanceRelationsQueryConditions));
+ query.limit(instanceIds.size());
}
});
return resp.getDataPoints().stream().map(e -> buildInstance(e, schema)).collect(Collectors.toList());
@@ -200,6 +206,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
query.and(lte(EndpointTraffic.LAST_PING_TIME_BUCKET, endTimeBucket));
}
query.setOrderBy(new AbstractQuery.OrderBy(AbstractQuery.Sort.DESC));
+ query.limit(limit);
}
});
@@ -230,6 +237,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, supportStatus.value()));
}
query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ query.limit(limit);
}
});
@@ -256,6 +264,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
if (!includeVirtual) {
query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
}
+ query.limit(limit);
}
});
@@ -280,6 +289,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, startPingTimeBucket));
query.and(lte(ProcessTraffic.LAST_PING_TIME_BUCKET, endPingTimeBucket));
query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+ query.limit(limit);
}
});
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 44d4482227..44f38d7106 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
@@ -34,12 +34,14 @@ import org.apache.skywalking.oap.server.core.analysis.manual.networkalias.Networ
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.BanyanDBStorageConfig;
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();
+ protected final int limit;
private MetadataRegistry.Schema schema;
@@ -47,8 +49,9 @@ public class BanyanDBNetworkAddressAliasDAO extends AbstractBanyanDBDAO implemen
NetworkAddressAlias.LAST_UPDATE_TIME_BUCKET,
NetworkAddressAlias.REPRESENT_SERVICE_ID, NetworkAddressAlias.REPRESENT_SERVICE_INSTANCE_ID);
- public BanyanDBNetworkAddressAliasDAO(final BanyanDBStorageClient client) {
+ public BanyanDBNetworkAddressAliasDAO(final BanyanDBStorageClient client, BanyanDBStorageConfig config) {
super(client);
+ this.limit = config.getResultWindowMaxSize();
}
private MetadataRegistry.Schema getSchema() {
@@ -69,6 +72,7 @@ public class BanyanDBNetworkAddressAliasDAO extends AbstractBanyanDBDAO implemen
@Override
protected void apply(final MeasureQuery query) {
query.and(gte(NetworkAddressAlias.LAST_UPDATE_TIME_BUCKET, timeBucket));
+ query.limit(limit);
}
}
);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBServiceLabelDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBServiceLabelDAO.java
index 734edb7ef7..630c041546 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBServiceLabelDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBServiceLabelDAO.java
@@ -31,14 +31,17 @@ import org.apache.skywalking.oap.server.core.analysis.DownSampling;
import org.apache.skywalking.oap.server.core.analysis.manual.process.ServiceLabelRecord;
import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IServiceLabelDAO;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageConfig;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
public class BanyanDBServiceLabelDAO extends AbstractBanyanDBDAO implements IServiceLabelDAO {
private static final Set<String> TAGS = ImmutableSet.of(ServiceLabelRecord.LABEL, ServiceLabelRecord.SERVICE_ID);
+ private final int limit;
- public BanyanDBServiceLabelDAO(final BanyanDBStorageClient client) {
+ public BanyanDBServiceLabelDAO(final BanyanDBStorageClient client, BanyanDBStorageConfig config) {
super(client);
+ this.limit = config.getMetadataQueryMaxSize();
}
@Override
@@ -49,6 +52,7 @@ public class BanyanDBServiceLabelDAO extends AbstractBanyanDBDAO implements ISer
@Override
protected void apply(final MeasureQuery query) {
query.and(eq(ServiceLabelRecord.SERVICE_ID, serviceId));
+ query.limit(limit);
}
}).getDataPoints()
.stream()
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBEBPFProfilingDataDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBEBPFProfilingDataDAO.java
index c03a4ff3be..c6110c414a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBEBPFProfilingDataDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBEBPFProfilingDataDAO.java
@@ -40,9 +40,11 @@ public class BanyanDBEBPFProfilingDataDAO extends AbstractBanyanDBDAO implements
EBPFProfilingDataRecord.TARGET_TYPE,
EBPFProfilingDataRecord.DATA_BINARY,
EBPFProfilingDataRecord.TASK_ID);
+ private final int profileDataQueryBatchSize;
- public BanyanDBEBPFProfilingDataDAO(BanyanDBStorageClient client) {
+ public BanyanDBEBPFProfilingDataDAO(BanyanDBStorageClient client, int profileDataQueryBatchSize) {
super(client);
+ this.profileDataQueryBatchSize = profileDataQueryBatchSize;
}
@Override
@@ -57,6 +59,7 @@ public class BanyanDBEBPFProfilingDataDAO extends AbstractBanyanDBDAO implements
query.and(eq(EBPFProfilingDataRecord.SCHEDULE_ID, scheduleId));
query.and(gte(EBPFProfilingDataRecord.UPLOAD_TIME, beginTime));
query.and(lte(EBPFProfilingDataRecord.UPLOAD_TIME, endTime));
+ query.setLimit(profileDataQueryBatchSize);
}
}
);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBSpanAttachedEventQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBSpanAttachedEventQueryDAO.java
index 984bc43801..e745c114fb 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBSpanAttachedEventQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBSpanAttachedEventQueryDAO.java
@@ -46,9 +46,11 @@ public class BanyanDBSpanAttachedEventQueryDAO extends AbstractBanyanDBDAO imple
SpanAttachedEventRecord.TRACE_SPAN_ID,
SpanAttachedEventRecord.DATA_BINARY,
SpanAttachedEventRecord.TIMESTAMP);
+ private final int batchSize;
- public BanyanDBSpanAttachedEventQueryDAO(BanyanDBStorageClient client) {
+ public BanyanDBSpanAttachedEventQueryDAO(BanyanDBStorageClient client, int profileDataQueryBatchSize) {
super(client);
+ this.batchSize = profileDataQueryBatchSize;
}
@Override
@@ -59,6 +61,7 @@ public class BanyanDBSpanAttachedEventQueryDAO extends AbstractBanyanDBDAO imple
query.and(in(SpanAttachedEventRecord.RELATED_TRACE_ID, traceIds));
query.and(eq(SpanAttachedEventRecord.TRACE_REF_TYPE, type.value()));
query.setOrderBy(new StreamQuery.OrderBy(AbstractQuery.Sort.ASC));
+ query.setLimit(batchSize);
}
});
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java
index b63baad81b..bb8c3b4271 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java
@@ -18,6 +18,7 @@
package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
+import com.google.common.base.Strings;
import com.google.common.collect.ImmutableSet;
import org.apache.skywalking.banyandb.v1.client.AbstractQuery;
import org.apache.skywalking.banyandb.v1.client.Element;
@@ -70,9 +71,11 @@ public class BanyanDBTraceQueryDAO extends AbstractBanyanDBDAO implements ITrace
SegmentRecord.START_TIME,
SegmentRecord.SEGMENT_ID,
SegmentRecord.DATA_BINARY);
+ private final int segmentQueryMaxSize;
- public BanyanDBTraceQueryDAO(BanyanDBStorageClient client) {
+ public BanyanDBTraceQueryDAO(BanyanDBStorageClient client, int segmentQueryMaxSize) {
super(client);
+ this.segmentQueryMaxSize = segmentQueryMaxSize;
}
@Override
@@ -107,6 +110,10 @@ public class BanyanDBTraceQueryDAO extends AbstractBanyanDBDAO implements ITrace
query.and(eq(SegmentRecord.ENDPOINT_ID, endpointId));
}
+ if (!Strings.isNullOrEmpty(traceId)) {
+ query.and(eq(SegmentRecord.TRACE_ID, traceId));
+ }
+
switch (traceState) {
case ERROR:
query.and(eq(SegmentRecord.IS_ERROR, BooleanUtils.TRUE));
@@ -181,6 +188,7 @@ public class BanyanDBTraceQueryDAO extends AbstractBanyanDBDAO implements ITrace
@Override
public void apply(StreamQuery query) {
query.and(eq(SegmentRecord.TRACE_ID, traceId));
+ query.setLimit(segmentQueryMaxSize);
}
});
return buildRecords(resp);
@@ -193,6 +201,7 @@ public class BanyanDBTraceQueryDAO extends AbstractBanyanDBDAO implements ITrace
@Override
public void apply(StreamQuery query) {
query.and(in(SegmentRecord.SEGMENT_ID, segmentIdList));
+ query.setLimit(segmentQueryMaxSize);
}
});
return buildRecords(resp);
@@ -206,6 +215,7 @@ public class BanyanDBTraceQueryDAO extends AbstractBanyanDBDAO implements ITrace
public void apply(StreamQuery query) {
query.and(in(SegmentRecord.TRACE_ID, traceIdList));
query.and(in(SegmentRecord.SERVICE_INSTANCE_ID, instanceIdList));
+ query.setLimit(segmentQueryMaxSize);
}
});
return buildRecords(resp);
diff --git a/oap-server/server-tools/data-generator/src/main/resources/application.yml b/oap-server/server-tools/data-generator/src/main/resources/application.yml
index 6aa06b8dbd..3e0b565167 100755
--- a/oap-server/server-tools/data-generator/src/main/resources/application.yml
+++ b/oap-server/server-tools/data-generator/src/main/resources/application.yml
@@ -192,8 +192,12 @@ storage:
# The number of threads that write data to BanyanDB concurrently.
# Bigger value can improve the write performance, but also increase the OAP and BanyanDB Server CPU usage.
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
+ resultWindowMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_WINDOW_SIZE:10000}
+ metadataQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_SIZE:10000}
+ segmentQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_SEGMENT_SIZE:200}
# The max number of profile task query in a request.
- profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_PROFILE_TASK_QUERY_MAX_SIZE:200}
+ profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_TASK_SIZE:200}
+ profileDataQueryBatchSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_DATA_BATCH_SIZE:100}
# Data is stored in BanyanDB in segments. A segment is a time range of data.
# The segment interval is the time range of a segment.
# The value should be less or equal to data TTL relevant settings.