kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support BanyanDB internal metrics query execution tracing. (#12400) 

* Support BanyanDB internal metrics query execution tracing.
* BanyanDB client config: rise the default `maxBulkSize` to 10000, add `flushTimeout` and set default to 10s.
This commit is contained in:
Wan Kai 2024-07-02 18:58:47 +08:00 committed by GitHub
parent 2e8b4f9917
commit 8b678bebe6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 75 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/en/changes/changes.md
View File

@ -24,6 +24,8 @@
* Fix BanyanDB metrics query: used the wrong `Downsampling` type to find the schema.
* Support fetch cilium flow to monitoring network traffic between cilium services.
* Support `labelCount` function in the OAL engine.
* Support BanyanDB internal metrics query execution tracing.
* BanyanDB client config: rise the default `maxBulkSize` to 10000, add `flushTimeout` and set default to 10s.
#### UI
* Highlight search log keywords.

3
docs/en/setup/backend/configuration-vocabulary.md
View File

@ -140,8 +140,9 @@ The Configuration Vocabulary lists all available configurations provided by `app
| - | - | asyncBatchPersistentPoolSize | async flush data into database thread size | SW_STORAGE_ASYNC_BATCH_PERSISTENT_POOL_SIZE | 4 |
| - | banyandb | - | BanyanDB storage. | - | - |
| - | - | targets | Hosts with ports of the BanyanDB. | SW_STORAGE_BANYANDB_TARGETS | 127.0.0.1:17912 |
| - | - | maxBulkSize | The maximum size of write entities in a single batch write call. | SW_STORAGE_BANYANDB_MAX_BULK_SIZE | 5000 |
| - | - | maxBulkSize | The maximum size of write entities in a single batch write call. | SW_STORAGE_BANYANDB_MAX_BULK_SIZE | 10000 |
| - | - | flushInterval | Period of flush interval. In the timeunit of seconds. | SW_STORAGE_BANYANDB_FLUSH_INTERVAL | 15 |
| - | - | flushTimeout | The timeout seconds of a bulk flush. | SW_STORAGE_BANYANDB_FLUSH_TIMEOUT | 10 |
| - | - | metricsShardsNumber | Shards Number for measure/metrics. | SW_STORAGE_BANYANDB_METRICS_SHARDS_NUMBER | 1 |
| - | - | recordShardsNumber | Shards Number for a normal record. | SW_STORAGE_BANYANDB_RECORD_SHARDS_NUMBER | 1 |
| - | - | superDatasetShardsFactor | Shards Factor for a super dataset record, i.e. Shard number of a super dataset is recordShardsNumber*superDatasetShardsFactor. | SW_STORAGE_BANYANDB_SUPERDATASET_SHARDS_FACTOR | 2 |

6
docs/en/setup/backend/storages/banyandb.md
View File

@ -3,7 +3,7 @@
[BanyanDB](https://github.com/apache/skywalking-banyandb) is a dedicated storage implementation developed by the SkyWalking Team and the community.
Activate BanyanDB as the storage, and set storage provider to **banyandb**.
The OAP requires BanyanDB 0.6 server. From this version, BanyanDB provides general compatibility.
The OAP requires BanyanDB 0.7 server. From this version, BanyanDB provides general compatibility.
```yaml
storage:
@ -14,12 +14,14 @@ storage:
targets: ${SW_STORAGE_BANYANDB_TARGETS:127.0.0.1:17912}
# The max number of records in a bulk write request.
# Bigger value can improve the write performance, but also increase the OAP and BanyanDB Server memory usage.
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:5000}
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:10000}
# The minimum seconds between two bulk flushes.
# If the data in a bulk is less than maxBulkSize, the data will be flushed after this period.
# If the data in a bulk is more than maxBulkSize, the data will be flushed immediately.
# Bigger value can reduce the write pressure on BanyanDB Server, but also increase the latency of the data.
flushInterval: ${SW_STORAGE_BANYANDB_FLUSH_INTERVAL:15}
# The timeout seconds of a bulk flush.
flushTimeout: ${SW_STORAGE_BANYANDB_FLUSH_TIMEOUT:10}
# The shard number of `measure` groups that store the metrics data.
metricsShardsNumber: ${SW_STORAGE_BANYANDB_METRICS_SHARDS_NUMBER:1}
# The shard number of `stream` groups that store the trace, log and profile data.

2
oap-server-bom/pom.xml
View File

@ -73,7 +73,7 @@
<httpcore.version>4.4.13</httpcore.version>
<httpasyncclient.version>4.1.5</httpasyncclient.version>
<commons-compress.version>1.21</commons-compress.version>
<banyandb-java-client.version>0.6.0</banyandb-java-client.version>
<banyandb-java-client.version>0.7.0-rc1</banyandb-java-client.version>
<kafka-clients.version>3.4.0</kafka-clients.version>
<spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
<consul.client.version>1.5.3</consul.client.version>

8
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricsQueryService.java
View File

@ -92,8 +92,8 @@ public class MetricsQueryService implements Service {
DebuggingSpan span = null;
try {
if (traceContext != null) {
span = traceContext.createSpan("Query Service");
span.setMsg("readMetricsValues, MetricsCondition: " + condition + ", Duration: " + duration);
span = traceContext.createSpan("Query Service: readMetricsValues");
span.setMsg("MetricsCondition: " + condition + ", Duration: " + duration);
}
return invokeReadMetricsValues(condition, duration);
} finally {
@ -125,8 +125,8 @@ public class MetricsQueryService implements Service {
DebuggingSpan span = null;
try {
if (traceContext != null) {
span = traceContext.createSpan("Query Service");
span.setMsg("readLabeledMetricsValues, MetricsCondition: " + condition + ", Labels: " + labels + ", Duration: " + duration);
span = traceContext.createSpan("Query Service: readLabeledMetricsValues");
span.setMsg("MetricsCondition: " + condition + ", Labels: " + labels + ", Duration: " + duration);
}
return invokeReadLabeledMetricsValues(condition, labels, duration);
} finally {

4
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/RecordQueryService.java
View File

@ -65,8 +65,8 @@ public class RecordQueryService implements Service {
DebuggingSpan span = null;
try {
if (traceContext != null) {
span = traceContext.createSpan("Query Service");
span.setMsg("readRecords, RecordCondition: " + condition + ", Duration: " + duration);
span = traceContext.createSpan("Query Service: readRecords");
span.setMsg("RecordCondition: " + condition + ", Duration: " + duration);
}
return invokeReadRecords(condition, duration);
} finally {

11
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/debugging/DebuggingSpan.java
View File

@ -28,10 +28,13 @@ public class DebuggingSpan {
private int parentSpanId;
private final String operation;
//nano seconds
private final long startTime;
@Setter
private long startTime;
//nano seconds
@Setter
private long endTime;
//nano seconds
@Setter
private long duration;
@Setter
private String msg;
@ -41,11 +44,5 @@ public class DebuggingSpan {
public DebuggingSpan(int spanId, String operation) {
this.spanId = spanId;
this.operation = operation;
this.startTime = System.nanoTime();
}
public void stopSpan() {
this.endTime = System.nanoTime();
this.duration = endTime - startTime;
}
}

13
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/debugging/DebuggingTraceContext.java
View File

@ -39,8 +39,11 @@ public class DebuggingTraceContext {
public DebuggingSpan createSpan(String operation) {
DebuggingSpan span = new DebuggingSpan(spanIdGenerator++, operation);
if (debug) {
//default start time, could be overwritten by setStartTime (BanyanDB Trace)
span.setStartTime(System.nanoTime());
DebuggingSpan parentSpan = spanStack.isEmpty() ? null : spanStack.peek();
if (parentSpan != null) {
//default parent span id, could be overwritten by setParentSpanId (BanyanDB Trace)
span.setParentSpanId(parentSpan.getSpanId());
} else {
span.setParentSpanId(-1);
@ -51,9 +54,17 @@ public class DebuggingTraceContext {
return span;
}
public DebuggingSpan getParentSpan() {
if (spanStack.isEmpty()) {
return null;
}
return spanStack.peek();
}
public void stopSpan(DebuggingSpan span) {
if (debug) {
span.stopSpan();
span.setEndTime(System.nanoTime());
span.setDuration(span.getEndTime() - span.getStartTime());
if (spanStack.isEmpty()) {
return;
}

4
oap-server/server-starter/src/main/resources/application.yml
View File

@ -222,12 +222,14 @@ storage:
targets: ${SW_STORAGE_BANYANDB_TARGETS:127.0.0.1:17912}
# The max number of records in a bulk write request.
# Bigger value can improve the write performance, but also increase the OAP and BanyanDB Server memory usage.
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:5000}
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:10000}
# The minimum seconds between two bulk flushes.
# If the data in a bulk is less than maxBulkSize, the data will be flushed after this period.
# If the data in a bulk is more than maxBulkSize, the data will be flushed immediately.
# Bigger value can reduce the write pressure on BanyanDB Server, but also increase the latency of the data.
flushInterval: ${SW_STORAGE_BANYANDB_FLUSH_INTERVAL:15}
# The timeout seconds of a bulk flush.
flushTimeout: ${SW_STORAGE_BANYANDB_FLUSH_TIMEOUT:10}
# The shard number of `measure` groups that store the metrics data.
metricsShardsNumber: ${SW_STORAGE_BANYANDB_METRICS_SHARDS_NUMBER:1}
# The shard number of `stream` groups that store the trace, log and profile data.

8
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
View File

@ -54,9 +54,11 @@ import java.util.List;
public class BanyanDBStorageClient implements Client, HealthCheckable {
final BanyanDBClient client;
private final DelegatedHealthChecker healthChecker = new DelegatedHealthChecker();
private final int flushTimeout;
public BanyanDBStorageClient(String... targets) {
public BanyanDBStorageClient(int flushTimeout, String... targets) {
this.client = new BanyanDBClient(targets);
this.flushTimeout = flushTimeout;
}
@Override
@ -234,11 +236,11 @@ public class BanyanDBStorageClient implements Client, HealthCheckable {
}
public StreamBulkWriteProcessor createStreamBulkProcessor(int maxBulkSize, int flushInterval, int concurrency) {
return this.client.buildStreamWriteProcessor(maxBulkSize, flushInterval, concurrency);
return this.client.buildStreamWriteProcessor(maxBulkSize, flushInterval, concurrency, flushTimeout);
}
public MeasureBulkWriteProcessor createMeasureBulkProcessor(int maxBulkSize, int flushInterval, int concurrency) {
return this.client.buildMeasureWriteProcessor(maxBulkSize, flushInterval, concurrency);
return this.client.buildMeasureWriteProcessor(maxBulkSize, flushInterval, concurrency, flushTimeout);
}
@Override

4
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
View File

@ -41,6 +41,10 @@ public class BanyanDBStorageConfig extends ModuleConfig {
* Period of flush interval. In the timeunit of seconds.
*/
private int flushInterval = 15;
/**
* Timeout of flush. In the timeunit of seconds.
*/
private int flushTimeout = 10;
/**
* Concurrent consumer threads for batch writing.
*/

2
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
View File

@ -117,7 +117,7 @@ public class BanyanDBStorageProvider extends ModuleProvider {
public void prepare() throws ServiceNotProvidedException, ModuleStartException {
this.registerServiceImplementation(StorageBuilderFactory.class, new StorageBuilderFactory.Default());
this.client = new BanyanDBStorageClient(config.getTargetArray());
this.client = new BanyanDBStorageClient(config.getFlushTimeout(), config.getTargetArray());
this.modelInstaller = new BanyanDBIndexInstaller(client, getManager(), this.config);
// Stream

28
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
View File

@ -26,11 +26,13 @@ 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.Span;
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.banyandb.v1.client.TopNQuery;
import org.apache.skywalking.banyandb.v1.client.TopNQueryResponse;
import org.apache.skywalking.banyandb.v1.client.Trace;
import org.apache.skywalking.oap.server.core.query.type.KeyValue;
import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingSpan;
import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingTraceContext;
@ -196,6 +198,7 @@ public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageCli
builder.append("\n").append(" Response: ").append(new Gson().toJson(response.getDataPoints()));
span.setMsg(builder.toString());
}
addDBTrace2DebuggingTrace(response.getTrace(), traceContext, span);
return response;
} finally {
if (traceContext != null && span != null) {
@ -224,10 +227,33 @@ public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageCli
}
builder.apply(query);
DebuggingTraceContext traceContext = DebuggingTraceContext.TRACE_CONTEXT.get();
if (traceContext != null && traceContext.isDebug()) {
query.enableTrace();
}
return getClient().query(query);
}
private void addDBTrace2DebuggingTrace(Trace trace, DebuggingTraceContext traceContext, DebuggingSpan parentSpan) {
if (traceContext == null || parentSpan == null || trace == null) {
return;
}
trace.getSpans().forEach(span -> addDBSpan2DebuggingTrace(span, traceContext, parentSpan));
}
private void addDBSpan2DebuggingTrace(Span span, DebuggingTraceContext traceContext, DebuggingSpan parentSpan) {
DebuggingSpan debuggingSpan = traceContext.createSpan("BanyanDB: " + span.getMessage());
debuggingSpan.setStartTime(span.getStartTime().getSeconds() * 1000_000_000 + span.getStartTime().getNanos());
debuggingSpan.setEndTime(span.getEndTime().getSeconds() * 1000_000_000 + span.getEndTime().getNanos());
debuggingSpan.setDuration(span.getDuration());
debuggingSpan.setParentSpanId(parentSpan.getSpanId());
debuggingSpan.setMsg(span.getTags().toString());
if (span.isError()) {
debuggingSpan.setError("BanyanDB occurs error.");
}
span.getChildren().forEach(child -> addDBSpan2DebuggingTrace(child, traceContext, debuggingSpan));
}
protected static QueryBuilder<MeasureQuery> emptyMeasureQuery() {
return new QueryBuilder<MeasureQuery>() {
@Override

4
oap-server/server-tools/data-generator/src/main/resources/application.yml
View File

@ -170,12 +170,14 @@ storage:
targets: ${SW_STORAGE_BANYANDB_TARGETS:127.0.0.1:17912}
# The max number of records in a bulk write request.
# Bigger value can improve the write performance, but also increase the OAP and BanyanDB Server memory usage.
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:5000}
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:10000}
# The minimum seconds between two bulk flushes.
# If the data in a bulk is less than maxBulkSize, the data will be flushed after this period.
# If the data in a bulk is more than maxBulkSize, the data will be flushed immediately.
# Bigger value can reduce the write pressure on BanyanDB Server, but also increase the latency of the data.
flushInterval: ${SW_STORAGE_BANYANDB_FLUSH_INTERVAL:15}
# The timeout seconds of a bulk flush.
flushTimeout: ${SW_STORAGE_BANYANDB_FLUSH_TIMEOUT:10}
# The shard number of `measure` groups that store the metrics data.
metricsShardsNumber: ${SW_STORAGE_BANYANDB_METRICS_SHARDS_NUMBER:1}
# The shard number of `stream` groups that store the trace, log and profile data.

2
test/e2e-v2/script/env
View File

@ -23,7 +23,7 @@ SW_AGENT_CLIENT_JS_COMMIT=af0565a67d382b683c1dbd94c379b7080db61449
SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
SW_KUBERNETES_COMMIT_SHA=1741f5a0959b85faaf0a2fc6bbd7b550ec330280
SW_ROVER_COMMIT=6bbd39aa701984482330d9dfb4dbaaff0527d55c
SW_BANYANDB_COMMIT=7a12fd03339fcab8e4d65286d2f65319e5576ba2
SW_BANYANDB_COMMIT=e7210733022566cd0cee7ea2dc12cfb2f30fa8f3
SW_AGENT_PHP_COMMIT=3192c553002707d344bd6774cfab5bc61f67a1d3
SW_CTL_COMMIT=d5f3597733aa5217373986d776a3ee5ee8b3c468