diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index bffa9084e3..4b1efc3c67 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -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 diff --git a/docs/en/setup/backend/backend-storage.md b/docs/en/setup/backend/backend-storage.md index 0920b8efc3..28c0ffcb01 100644 --- a/docs/en/setup/backend/backend-storage.md +++ b/docs/en/setup/backend/backend-storage.md @@ -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) diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md index ca37749a88..1d1a1ed84f 100644 --- a/docs/en/setup/backend/configuration-vocabulary.md +++ b/docs/en/setup/backend/configuration-vocabulary.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 | diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml index c2f934d2ef..341e504041 100755 --- a/oap-server/server-starter/src/main/resources/application.yml +++ b/oap-server/server-starter/src/main/resources/application.yml @@ -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 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 8341d9f244..e2f281a0b8 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 @@ -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); 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 b146134f05..e31874e7f5 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 @@ -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; } 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 db7f1cddbf..d9da6a5b19 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 @@ -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); 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 b598e0ccfd..9287d5c710 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 @@ -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 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 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 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 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> 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"); }