Add shards factor for BanyanDB (#9116)
This commit is contained in:
Jiajing LU
GitHub
parent
e8b24e7fb2
commit
dc39ce9bb4
|
|
@ -9,6 +9,8 @@
|
|||
* Upgrade OAP dependencies zipkin to 2.23.16, H2 to 2.1.212, Apache Freemarker to 2.3.31, gRPC-java 1.46.0, netty to
|
||||
4.1.76.
|
||||
* Upgrade Webapp dependencies, spring-cloud-dependencies to 2021.0.2, logback-classic to 1.2.11
|
||||
* [**IMPORTANT**] Add BanyanDB storage implementation. Notice BanyanDB is currently under active development
|
||||
and **SHOULD NOT** be used in production cluster.
|
||||
|
||||
#### OAP Server
|
||||
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ Natively supported storage:
|
|||
- MySQL
|
||||
- TiDB
|
||||
- PostgreSQL
|
||||
- BanyanDB
|
||||
|
||||
|
||||
## H2
|
||||
|
|
@ -262,6 +263,26 @@ storage:
|
|||
All connection-related settings, including URL link, username, and password, are found in `application.yml`.
|
||||
Only part of the settings is listed here. Please follow [HikariCP](https://github.com/brettwooldridge/HikariCP) connection pool document for full settings.
|
||||
|
||||
## BanyanDB
|
||||
[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**.
|
||||
|
||||
```yaml
|
||||
storage:
|
||||
banyandb:
|
||||
host: ${SW_STORAGE_BANYANDB_HOST:127.0.0.1}
|
||||
port: ${SW_STORAGE_BANYANDB_PORT:17912}
|
||||
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:5000}
|
||||
flushInterval: ${SW_STORAGE_BANYANDB_FLUSH_INTERVAL:15}
|
||||
metricsShardsNumber: ${SW_STORAGE_BANYANDB_METRICS_SHARDS_NUMBER:1}
|
||||
recordShardsNumber: ${SW_STORAGE_BANYANDB_RECORD_SHARDS_NUMBER:1}
|
||||
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
|
||||
```
|
||||
|
||||
For more details, please refer to the documents of [BanyanDB](https://skywalking.apache.org/docs/skywalking-banyandb/latest/readme/)
|
||||
and [BanyanDB Java Client](https://github.com/apache/skywalking-banyandb-java-client) subprojects.
|
||||
|
||||
## More storage extension solutions
|
||||
Follow the [Storage extension development guide](../../guides/storage-extention.md)
|
||||
|
|
|
|||
|
|
@ -133,6 +133,16 @@ The Configuration Vocabulary lists all available configurations provided by `app
|
|||
| - | - | numOfSearchableValuesPerTag | In a trace segment, this includes multiple spans with multiple tags. Different spans may have same tag key, e.g. multiple HTTP exit spans all have their own `http.method` tags. This configuration sets the limit on the maximum number of values for the same tag key. | SW_STORAGE_NUM_OF_SEARCHABLE_VALUES_PER_TAG | 2 |
|
||||
| - | - | maxSizeOfBatchSql | The maximum size of batch size of SQL execution | SW_STORAGE_MAX_SIZE_OF_BATCH_SQL | 2000 |
|
||||
| - | - | asyncBatchPersistentPoolSize | async flush data into database thread size | SW_STORAGE_ASYNC_BATCH_PERSISTENT_POOL_SIZE | 4 |
|
||||
| - | banyandb | - | BanyanDB storage. | - | - |
|
||||
| - | - | host | Host of the BanyanDB. | SW_STORAGE_BANYANDB_HOST | 127.0.0.1 |
|
||||
| - | - | port | Port of the BanyanDB. | SW_STORAGE_BANYANDB_PORT | 17912 |
|
||||
| - | - | maxBulkSize | The maximum size of write entities in a single batch write call. | SW_STORAGE_BANYANDB_MAX_BULK_SIZE | 5000 |
|
||||
| - | - | flushInterval | Period of flush interval. In the timeunit of seconds. | SW_STORAGE_BANYANDB_FLUSH_INTERVAL | 15 |
|
||||
| - | - | 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 |
|
||||
| - | - | concurrentWriteThreads | Concurrent consumer threads for batch writing. | SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS | 15 |
|
||||
| - | - | profileTaskQueryMaxSize | Max size of ProfileTask to be fetched. | SW_STORAGE_BANYANDB_PROFILE_TASK_QUERY_MAX_SIZE | 200 |
|
||||
| agent-analyzer | default | Agent Analyzer. | SW_AGENT_ANALYZER | default | |
|
||||
| - | - | traceSamplingPolicySettingsFile | The sampling policy including `sampling rate` and `the threshold of trace segment latency` can be configured by the `traceSamplingPolicySettingsFile` file. | SW_TRACE_SAMPLING_POLICY_SETTINGS_FILE | `trace-sampling-policy-settings.yml` |
|
||||
| - | - | slowDBAccessThreshold | The slow database access threshold (in milliseconds). | SW_SLOW_DB_THRESHOLD | default:200,mongodb:100 |
|
||||
|
|
@ -184,7 +194,7 @@ The Configuration Vocabulary lists all available configurations provided by `app
|
|||
| - | - | restPort | Binding port of RESTful services. | SW_RECEIVER_ZIPKIN_REST_PORT | 9411 |
|
||||
| - | - | restContextPath | Web context path of RESTful services. | SW_RECEIVER_ZIPKIN_REST_CONTEXT_PATH | / |
|
||||
| - | - | restMaxThreads | Maximum thread number of RESTful services. | SW_RECEIVER_ZIPKIN_REST_MAX_THREADS | 200 |
|
||||
| - | - | restIdleTimeOut | Connector idle timeout of RESTful services (in milliseconds). | SW_RECEIVER_ZIPKIN_REST_IDLE_TIMEOUT | 30000 | |
|
||||
| - | - | restIdleTimeOut | Connector idle timeout of RESTful services (in milliseconds). | SW_RECEIVER_ZIPKIN_REST_IDLE_TIMEOUT | 30000 |
|
||||
| - | - | restAcceptQueueSize | Maximum request header size accepted. | SW_RECEIVER_ZIPKIN_REST_QUEUE_SIZE | 0 |
|
||||
| - | - | instanceNameRule | Get the instance name from these tags. | SW_RECEIVER_ZIPKIN_INSTANCE_NAME_RULE | [spring.instance_id,node_id] |
|
||||
| - | - | searchableTracesTags | Defines a set of span tag keys which are searchable. Multiple values are separated by commas. | SW_ZIPKIN_SEARCHABLE_TAG_KEYS | http.method |
|
||||
|
|
|
|||
|
|
@ -236,9 +236,11 @@ storage:
|
|||
banyandb:
|
||||
host: ${SW_STORAGE_BANYANDB_HOST:127.0.0.1}
|
||||
port: ${SW_STORAGE_BANYANDB_PORT:17912}
|
||||
group: ${SW_STORAGE_BANYANDB_GROUP:default}
|
||||
maxBulkSize: ${SW_STORAGE_BANYANDB_MAX_BULK_SIZE:5000}
|
||||
flushInterval: ${SW_STORAGE_BANYANDB_FLUSH_INTERVAL:15}
|
||||
metricsShardsNumber: ${SW_STORAGE_BANYANDB_METRICS_SHARDS_NUMBER:1}
|
||||
recordShardsNumber: ${SW_STORAGE_BANYANDB_RECORD_SHARDS_NUMBER:1}
|
||||
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
|
||||
|
||||
|
|
|
|||
|
|
@ -34,13 +34,16 @@ import java.io.IOException;
|
|||
|
||||
@Slf4j
|
||||
public class BanyanDBIndexInstaller extends ModelInstaller {
|
||||
public BanyanDBIndexInstaller(Client client, ModuleManager moduleManager) {
|
||||
private final BanyanDBStorageConfig config;
|
||||
|
||||
public BanyanDBIndexInstaller(Client client, ModuleManager moduleManager, BanyanDBStorageConfig config) {
|
||||
super(client, moduleManager);
|
||||
this.config = config;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean isExists(Model model) throws StorageException {
|
||||
final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(model);
|
||||
final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(model, config);
|
||||
try {
|
||||
final BanyanDBClient c = ((BanyanDBStorageClient) this.client).client;
|
||||
// first check group
|
||||
|
|
@ -51,7 +54,7 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
|
|||
log.info("group {} created", g.name());
|
||||
// then check entity schema
|
||||
if (metadata.findRemoteSchema(c).isPresent()) {
|
||||
MetadataRegistry.INSTANCE.registerModel(model);
|
||||
MetadataRegistry.INSTANCE.registerModel(model, config);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
|
|
@ -64,13 +67,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);
|
||||
Stream stream = (Stream) MetadataRegistry.INSTANCE.registerModel(model, config);
|
||||
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);
|
||||
Measure measure = (Measure) MetadataRegistry.INSTANCE.registerModel(model, config);
|
||||
if (measure != null) {
|
||||
log.info("install measure schema {}", model.getName());
|
||||
((BanyanDBStorageClient) client).define(measure);
|
||||
|
|
|
|||
|
|
@ -27,10 +27,6 @@ import org.apache.skywalking.oap.server.library.module.ModuleConfig;
|
|||
public class BanyanDBStorageConfig extends ModuleConfig {
|
||||
private String host = "127.0.0.1";
|
||||
private int port = 17912;
|
||||
/**
|
||||
* Group of the schema in BanyanDB
|
||||
*/
|
||||
private String group = "default";
|
||||
/**
|
||||
* The maximum size of write entities in a single batch write call.
|
||||
*/
|
||||
|
|
@ -47,5 +43,17 @@ public class BanyanDBStorageConfig extends ModuleConfig {
|
|||
* Max size of {@link org.apache.skywalking.oap.server.core.query.type.ProfileTask} to be fetched
|
||||
* in a single request.
|
||||
*/
|
||||
private int profileTaskQueryMaxSize = 200;
|
||||
private int profileTaskQueryMaxSize;
|
||||
/**
|
||||
* Shards Number for measure/metrics.
|
||||
*/
|
||||
private int metricsShardsNumber;
|
||||
/**
|
||||
* Shards Number for a normal record.
|
||||
*/
|
||||
private int recordShardsNumber;
|
||||
/**
|
||||
* Shards Factor for a super dataset
|
||||
*/
|
||||
private int superDatasetShardsFactor;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -145,7 +145,7 @@ public class BanyanDBStorageProvider extends ModuleProvider {
|
|||
this.client.registerChecker(healthChecker);
|
||||
try {
|
||||
this.client.connect();
|
||||
BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, getManager());
|
||||
BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, getManager(), this.config);
|
||||
getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
|
||||
} catch (Exception e) {
|
||||
throw new ModuleStartException(e.getMessage(), e);
|
||||
|
|
|
|||
|
|
@ -37,21 +37,16 @@ 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.Stream;
|
||||
import org.apache.skywalking.banyandb.v1.client.metadata.TagFamilySpec;
|
||||
import org.apache.skywalking.oap.server.core.alarm.AlarmRecord;
|
||||
import org.apache.skywalking.oap.server.core.analysis.DownSampling;
|
||||
import org.apache.skywalking.oap.server.core.analysis.manual.log.LogRecord;
|
||||
import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
|
||||
import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
|
||||
import org.apache.skywalking.oap.server.core.analysis.metrics.IntList;
|
||||
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.library.util.StringUtil;
|
||||
import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject;
|
||||
|
||||
import javax.annotation.Nullable;
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
|
@ -68,8 +63,8 @@ public enum MetadataRegistry {
|
|||
|
||||
private final Map<String, Schema> registry = new HashMap<>();
|
||||
|
||||
public NamedSchema<?> registerModel(Model model) {
|
||||
final SchemaMetadata schemaMetadata = parseMetadata(model);
|
||||
public NamedSchema<?> registerModel(Model model, BanyanDBStorageConfig config) {
|
||||
final SchemaMetadata schemaMetadata = parseMetadata(model, config);
|
||||
Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
|
||||
Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
|
||||
.collect(Collectors.toMap(modelColumn -> modelColumn.getColumnName().getStorageName(), Function.identity()));
|
||||
|
|
@ -136,8 +131,8 @@ public enum MetadataRegistry {
|
|||
.compressWithZSTD()
|
||||
.encodeWithGorilla()
|
||||
.build();
|
||||
} else if (DataTable.class.equals(modelColumn.getType())) {
|
||||
return Measure.FieldSpec.newBinaryField(valueColumn.getValueCName())
|
||||
} else if (StorageDataComplexObject.class.isAssignableFrom(modelColumn.getType())) {
|
||||
return Measure.FieldSpec.newStringField(valueColumn.getValueCName())
|
||||
.compressWithZSTD()
|
||||
.build();
|
||||
} else if (double.class.equals(modelColumn.getType())) {
|
||||
|
|
@ -224,30 +219,6 @@ public enum MetadataRegistry {
|
|||
return tagMetadataList;
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract extra tags from Configuration.
|
||||
* They are for tags defined for {@link SegmentRecord}, {@link LogRecord} and {@link AlarmRecord}.
|
||||
*
|
||||
* @param tags a series of tags joint by comma
|
||||
* @return a list of {@link org.apache.skywalking.banyandb.v1.client.metadata.TagFamilySpec.TagSpec} generated from input
|
||||
*/
|
||||
private List<TagMetadata> parseExtraTagSpecs(String tags, Schema.SchemaBuilder builder) {
|
||||
if (StringUtil.isEmpty(tags)) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
String[] tagsArray = tags.split(",");
|
||||
if (tagsArray.length == 0) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<TagMetadata> extraTagMetadataList = new ArrayList<>();
|
||||
for (final String tagName : tagsArray) {
|
||||
builder.spec(tagName, new ColumnSpec(ColumnType.TAG, String.class));
|
||||
extraTagMetadataList.add(new TagMetadata(parseIndexRule(tagName, null),
|
||||
TagFamilySpec.TagSpec.newStringTag(tagName)));
|
||||
}
|
||||
return extraTagMetadataList;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse TagSpec from {@link ModelColumn}
|
||||
*
|
||||
|
|
@ -258,7 +229,7 @@ public enum MetadataRegistry {
|
|||
private TagFamilySpec.TagSpec parseTagSpec(ModelColumn modelColumn) {
|
||||
final Class<?> clazz = modelColumn.getType();
|
||||
final String colName = modelColumn.getColumnName().getStorageName();
|
||||
if (String.class.equals(clazz) || DataTable.class.equals(clazz) || JsonObject.class.equals(clazz)) {
|
||||
if (String.class.equals(clazz) || StorageDataComplexObject.class.isAssignableFrom(clazz) || JsonObject.class.equals(clazz)) {
|
||||
return TagFamilySpec.TagSpec.newStringTag(colName);
|
||||
} else if (int.class.equals(clazz) || long.class.equals(clazz)) {
|
||||
return TagFamilySpec.TagSpec.newIntTag(colName);
|
||||
|
|
@ -280,16 +251,21 @@ public enum MetadataRegistry {
|
|||
throw new IllegalStateException("type " + modelColumn.getType().toString() + " is not supported");
|
||||
}
|
||||
|
||||
public SchemaMetadata parseMetadata(Model model) {
|
||||
public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config) {
|
||||
if (model.isRecord()) {
|
||||
String group = "stream-default";
|
||||
if (model.isSuperDataset()) {
|
||||
// for superDataset, we should use separate group
|
||||
group = "stream-" + model.getName();
|
||||
}
|
||||
return new SchemaMetadata(group, model.getName(), Kind.STREAM);
|
||||
return new SchemaMetadata(group,
|
||||
model.getName(),
|
||||
Kind.STREAM,
|
||||
config.getRecordShardsNumber() *
|
||||
(model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1)
|
||||
);
|
||||
}
|
||||
return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE);
|
||||
return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE, config.getMetricsShardsNumber());
|
||||
}
|
||||
|
||||
@RequiredArgsConstructor
|
||||
|
|
@ -299,6 +275,8 @@ public enum MetadataRegistry {
|
|||
private final String name;
|
||||
private final Kind kind;
|
||||
|
||||
private final int shard;
|
||||
|
||||
public Optional<NamedSchema<?>> findRemoteSchema(BanyanDBClient client) throws BanyanDBException {
|
||||
try {
|
||||
switch (kind) {
|
||||
|
|
@ -344,9 +322,9 @@ public enum MetadataRegistry {
|
|||
}
|
||||
switch (kind) {
|
||||
case STREAM:
|
||||
return client.define(Group.create(this.group, Catalog.STREAM, 2, 0, Duration.ofDays(7)));
|
||||
return client.define(Group.create(this.group, Catalog.STREAM, this.shard, 0, Duration.ofDays(7)));
|
||||
case MEASURE:
|
||||
return client.define(Group.create(this.group, Catalog.MEASURE, 2, 12, Duration.ofDays(7)));
|
||||
return client.define(Group.create(this.group, Catalog.MEASURE, this.shard, 12, Duration.ofDays(7)));
|
||||
default:
|
||||
throw new IllegalStateException("should not reach here");
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue