Introduce index mode and refactor banyandb group settings (#12790)
This commit is contained in:
Gao Hongtao
GitHub
parent
e1698a583f
commit
10e239f85b
|
|
@ -22,6 +22,7 @@
|
|||
* Adapt the new metadata standardization in Istio 1.24.
|
||||
* Bump up netty to 4.1.115, grpc to 1.68.1, boringssl to 2.0.69.
|
||||
* BanyanDB: Support update the Group settings when OAP starting.
|
||||
* BanyanDB: Introduce index mode and refactor banyandb group settings.
|
||||
|
||||
#### UI
|
||||
|
||||
|
|
|
|||
|
|
@ -1,70 +1,115 @@
|
|||
|
||||
## 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**.
|
||||
|
||||
The OAP requires BanyanDB 0.7 server. From this version, BanyanDB provides general compatibility.
|
||||
[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 by setting the storage provider to **banyandb**.
|
||||
|
||||
The OAP requires BanyanDB version **0.8** or later. From this version onwards, BanyanDB provides general compatibility.
|
||||
|
||||
### Configuration
|
||||
|
||||
```yaml
|
||||
storage:
|
||||
banyandb:
|
||||
# Targets is the list of BanyanDB servers, separated by commas.
|
||||
# Each target is a BanyanDB server in the format of `host:port`
|
||||
# If the BanyanDB is deployed as a standalone server, the target should be the IP address or domain name and port of the BanyanDB server.
|
||||
# If the BanyanDB is deployed in a cluster, the targets should be the IP address or domain name and port of the `liaison` nodes, separated by commas.
|
||||
# Each target is a BanyanDB server in the format of `host:port`.
|
||||
# If BanyanDB is deployed as a standalone server, the target should be the IP address or domain name and port of the BanyanDB server.
|
||||
# If BanyanDB is deployed in a cluster, the targets should be the IP address or domain name and port of the `liaison` nodes, separated by commas.
|
||||
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.
|
||||
|
||||
# The maximum number of records in a bulk write request.
|
||||
# A larger value can improve write performance but also increases OAP and BanyanDB Server memory usage.
|
||||
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.
|
||||
# If the data in a bulk exceeds maxBulkSize, the data will be flushed immediately.
|
||||
# A larger value can reduce write pressure on BanyanDB Server but increase data latency.
|
||||
flushInterval: ${SW_STORAGE_BANYANDB_FLUSH_INTERVAL:15}
|
||||
# The timeout seconds of a bulk flush.
|
||||
|
||||
# The timeout in seconds for 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.
|
||||
recordShardsNumber: ${SW_STORAGE_BANYANDB_RECORD_SHARDS_NUMBER:1}
|
||||
# The multiplier of the number of shards of the super dataset.
|
||||
# Super dataset is a special dataset that stores the trace or log data that is too large to be stored in the normal dataset.
|
||||
# If the normal dataset has `n` shards, the super dataset will have `n * superDatasetShardsFactor` shards.
|
||||
# For example, supposing `recordShardsNumber` is 3, and `superDatasetShardsFactor` is 2,
|
||||
# `segment-default` is a normal dataset that has 3 shards, and `segment-minute` is a super dataset that has 6 shards.
|
||||
superDatasetShardsFactor: ${SW_STORAGE_BANYANDB_SUPERDATASET_SHARDS_FACTOR:2}
|
||||
|
||||
# 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.
|
||||
# A higher value can improve write performance but also increases CPU usage on both OAP and BanyanDB Server.
|
||||
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
|
||||
# The maximum size of dataset when the OAP loads cache, such as network aliases.
|
||||
|
||||
# The maximum size of the dataset when the OAP loads cache, such as network aliases.
|
||||
resultWindowMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_WINDOW_SIZE:10000}
|
||||
|
||||
# The maximum size of metadata per query.
|
||||
metadataQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_SIZE:10000}
|
||||
# The maximum size of trace segments per query.
|
||||
|
||||
# The maximum number of trace segments per query.
|
||||
segmentQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_SEGMENT_SIZE:200}
|
||||
# The max number of profile task query in a request.
|
||||
|
||||
# The maximum number of profile task queries in a request.
|
||||
profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_TASK_SIZE:200}
|
||||
# The batch size of query profiling data.
|
||||
|
||||
# The batch size for querying profile data.
|
||||
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.
|
||||
segmentIntervalDays: ${SW_STORAGE_BANYANDB_SEGMENT_INTERVAL_DAYS:1}
|
||||
# The super dataset segment interval is the time range of a segment in the super dataset.
|
||||
superDatasetSegmentIntervalDays: ${SW_STORAGE_BANYANDB_SUPER_DATASET_SEGMENT_INTERVAL_DAYS:1}
|
||||
# Specific groups settings.
|
||||
# For example, {"group1": {"blockIntervalHours": 4, "segmentIntervalDays": 1}}
|
||||
# Please refer to https://github.com/apache/skywalking-banyandb/blob/${BANYANDB_RELEASE}/docs/crud/group.md#create-operation
|
||||
# for group setting details.
|
||||
specificGroupSettings: ${SW_STORAGE_BANYANDB_SPECIFIC_GROUP_SETTINGS:""}
|
||||
# If the BanyanDB server is configured with TLS, config the TLS cert file path and open tls connection.
|
||||
|
||||
# If the BanyanDB server is configured with TLS, configure the TLS cert file path and enable TLS connection.
|
||||
sslTrustCAPath: ${SW_STORAGE_BANYANDB_SSL_TRUST_CA_PATH:""}
|
||||
|
||||
# The group settings of record.
|
||||
# `gr` is the short name of the group settings of record.
|
||||
#
|
||||
# The "normal" section defines settings for datasets not specified in "super".
|
||||
# Each dataset will be grouped under a single group named "normal".
|
||||
grNormalShardNum: ${SW_STORAGE_BANYANDB_GR_NORMAL_SHARD_NUM:1}
|
||||
grNormalSIDays: ${SW_STORAGE_BANYANDB_GR_NORMAL_SI_DAYS:1}
|
||||
grNormalTTLDays: ${SW_STORAGE_BANYANDB_GR_NORMAL_TTL_DAYS:3}
|
||||
# "super" is a special dataset designed to store trace or log data that is too large for normal datasets.
|
||||
# Each super dataset will be a separate group in BanyanDB, following the settings defined in the "super" section.
|
||||
grSuperShardNum: ${SW_STORAGE_BANYANDB_GR_SUPER_SHARD_NUM:2}
|
||||
grSuperSIDays: ${SW_STORAGE_BANYANDB_GR_SUPER_SI_DAYS:1}
|
||||
grSuperTTLDays: ${SW_STORAGE_BANYANDB_GR_SUPER_TTL_DAYS:3}
|
||||
|
||||
# The group settings of metrics.
|
||||
# `gm` is the short name of the group settings of metrics.
|
||||
#
|
||||
# OAP stores metrics based its granularity.
|
||||
# Valid values are "day", "hour", and "minute". That means metrics will be stored in the three separate groups.
|
||||
# Non-"minute" are governed by the "core.downsampling" setting.
|
||||
# For example, if "core.downsampling" is set to "hour", the "hour" will be used, while "day" are ignored.
|
||||
gmMinuteShardNum: ${SW_STORAGE_BANYANDB_GM_MINUTE_SHARD_NUM:2}
|
||||
gmMinuteSIDays: ${SW_STORAGE_BANYANDB_GM_MINUTE_SI_DAYS:1}
|
||||
gmMinuteTTLDays: ${SW_STORAGE_BANYANDB_GM_MINUTE_TTL_DAYS:7}
|
||||
gmHourShardNum: ${SW_STORAGE_BANYANDB_GM_HOUR_SHARD_NUM:1}
|
||||
gmHourSIDays: ${SW_STORAGE_BANYANDB_GM_HOUR_SI_DAYS:1}
|
||||
gmHourTTLDays: ${SW_STORAGE_BANYANDB_GM_HOUR_TTL_DAYS:15}
|
||||
gmDayShardNum: ${SW_STORAGE_BANYANDB_GM_DAY_SHARD_NUM:1}
|
||||
gmDaySIDays: ${SW_STORAGE_BANYANDB_GM_DAY_SI_DAYS:1}
|
||||
gmDayTTLDays: ${SW_STORAGE_BANYANDB_GM_DAY_TTL_DAYS:30}
|
||||
# If the metrics is marked as "index_mode", the metrics will be stored in the "index" group.
|
||||
# The "index" group is designed to store metrics that are used for indexing without value columns.
|
||||
# Such as `service_traffic`, `network_address_alias`, etc.
|
||||
# "index_mode" requires BanyanDB *0.8.0* or later.
|
||||
gmIndexShardNum: ${SW_STORAGE_BANYANDB_GM_INDEX_SHARD_NUM:1}
|
||||
gmIndexSIDays: ${SW_STORAGE_BANYANDB_GM_INDEX_SI_DAYS:1}
|
||||
gmIndexTTLDays: ${SW_STORAGE_BANYANDB_GM_INDEX_TTL_DAYS:30}
|
||||
|
||||
```
|
||||
|
||||
BanyanDB Server supports two installation modes: standalone and cluster. The standalone mode is suitable for small-scale deployments, while the cluster mode is suitable for large-scale deployments.
|
||||
### Installation Modes
|
||||
|
||||
* Standalone mode: `targets` is the IP address/host name and port of the BanyanDB server.
|
||||
* Cluster mode: `targets` is the IP address/host name and port of the `liaison` nodes, separated by commas. `liaison` nodes are the entry points of the BanyanDB cluster.
|
||||
BanyanDB Server supports two installation modes:
|
||||
|
||||
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.
|
||||
- **Standalone Mode**: Suitable for small-scale deployments.
|
||||
- **Configuration**: `targets` is the IP address/hostname and port of the BanyanDB server.
|
||||
|
||||
- **Cluster Mode**: Suitable for large-scale deployments.
|
||||
- **Configuration**: `targets` is the IP address/hostname and port of the `liaison` nodes, separated by commas. `Liaison` nodes are the entry points of the BanyanDB cluster.
|
||||
|
||||
### Group Settings
|
||||
|
||||
BanyanDB supports **group settings** to configure storage groups, shards, segment intervals, and TTL (Time-To-Live). The group settings file is a YAML file required when using BanyanDB as the storage.
|
||||
|
||||
#### Basic Group Settings
|
||||
|
||||
- `ShardNum`: Number of shards in the group. Shards are the basic units of data storage in BanyanDB. Data is distributed across shards based on the hash value of the series ID. Refer to the [BanyanDB Shard](https://skywalking.apache.org/docs/skywalking-banyandb/latest/concept/clustering/#52-data-sharding) documentation for more details.
|
||||
- `SIDays`: Interval in days for creating a new segment. Segments are time-based, allowing efficient data retention and querying. `SI` stands for Segment Interval.
|
||||
- `TTLDays`: Time-to-live for the data in the group, in days. Data exceeding the TTL will be deleted.
|
||||
|
||||
For more details on setting `segmentIntervalDays` and `ttlDays`, refer to the [BanyanDB Rotation](https://skywalking.apache.org/docs/skywalking-banyandb/latest/concept/rotation/) documentation.
|
||||
|
||||
For more details, refer to the documentation of [BanyanDB](https://skywalking.apache.org/docs/skywalking-banyandb/latest/readme/) and the [BanyanDB Java Client](https://github.com/apache/skywalking-banyandb-java-client) subprojects.
|
||||
|
|
|
|||
|
|
@ -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.7.0</banyandb-java-client.version>
|
||||
<banyandb-java-client.version>0.8.0-rc0</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>
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ import java.util.Objects;
|
|||
*/
|
||||
@Stream(name = TopNCacheReadCommand.INDEX_NAME, scopeId = DefaultScopeDefine.CACHE_SLOW_ACCESS, builder = TopNCacheReadCommand.Builder.class, processor = TopNStreamProcessor.class)
|
||||
@BanyanDB.TimestampColumn(TopN.TIMESTAMP)
|
||||
@BanyanDB.IndexMode
|
||||
public class TopNCacheReadCommand extends TopN {
|
||||
public static final String INDEX_NAME = "top_n_cache_read_command";
|
||||
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ import java.util.Objects;
|
|||
*/
|
||||
@Stream(name = TopNCacheWriteCommand.INDEX_NAME, scopeId = DefaultScopeDefine.CACHE_SLOW_ACCESS, builder = TopNCacheWriteCommand.Builder.class, processor = TopNStreamProcessor.class)
|
||||
@BanyanDB.TimestampColumn(TopN.TIMESTAMP)
|
||||
@BanyanDB.IndexMode
|
||||
public class TopNCacheWriteCommand extends TopN {
|
||||
public static final String INDEX_NAME = "top_n_cache_write_command";
|
||||
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ import java.util.Objects;
|
|||
*/
|
||||
@Stream(name = TopNDatabaseStatement.INDEX_NAME, scopeId = DefaultScopeDefine.DATABASE_SLOW_STATEMENT, builder = TopNDatabaseStatement.Builder.class, processor = TopNStreamProcessor.class)
|
||||
@BanyanDB.TimestampColumn(TopN.TIMESTAMP)
|
||||
@BanyanDB.IndexMode
|
||||
public class TopNDatabaseStatement extends TopN {
|
||||
public static final String INDEX_NAME = "top_n_database_statement";
|
||||
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
"serviceId",
|
||||
"name"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class EndpointTraffic extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "endpoint_traffic";
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SE
|
|||
"serviceId",
|
||||
"name"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class InstanceTraffic extends Metrics {
|
||||
public static final String INDEX_NAME = "instance_traffic";
|
||||
public static final String SERVICE_ID = "service_id";
|
||||
|
|
|
|||
|
|
@ -44,6 +44,7 @@ import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.NE
|
|||
@EqualsAndHashCode(of = {
|
||||
"address"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class NetworkAddressAlias extends Metrics {
|
||||
public static final String INDEX_NAME = "network_address_alias";
|
||||
public static final String ADDRESS = "address";
|
||||
|
|
|
|||
|
|
@ -51,6 +51,7 @@ import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.PR
|
|||
"name",
|
||||
})
|
||||
@BanyanDB.StoreIDAsTag
|
||||
@BanyanDB.IndexMode
|
||||
public class ProcessTraffic extends Metrics {
|
||||
public static final String INDEX_NAME = "process_traffic";
|
||||
public static final String SERVICE_ID = "service_id";
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SE
|
|||
"serviceId",
|
||||
"label"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class ServiceLabelRecord extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "service_label";
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
@EqualsAndHashCode(of = {
|
||||
"entityId"
|
||||
}, callSuper = true)
|
||||
@BanyanDB.IndexMode
|
||||
public class EndpointRelationServerSideMetrics extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "endpoint_relation_server_side";
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
@EqualsAndHashCode(of = {
|
||||
"entityId"
|
||||
}, callSuper = true)
|
||||
@BanyanDB.IndexMode
|
||||
public class ServiceInstanceRelationClientSideMetrics extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "service_instance_relation_client_side";
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
@EqualsAndHashCode(of = {
|
||||
"entityId"
|
||||
}, callSuper = true)
|
||||
@BanyanDB.IndexMode
|
||||
public class ServiceInstanceRelationServerSideMetrics extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "service_instance_relation_server_side";
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
"entityId",
|
||||
"component_id"
|
||||
}, callSuper = true)
|
||||
@BanyanDB.IndexMode
|
||||
public class ProcessRelationClientSideMetrics extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "process_relation_client_side";
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
@EqualsAndHashCode(of = {
|
||||
"entityId"
|
||||
}, callSuper = true)
|
||||
@BanyanDB.IndexMode
|
||||
public class ProcessRelationServerSideMetrics extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "process_relation_server_side";
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
@EqualsAndHashCode(of = {
|
||||
"entityId"
|
||||
}, callSuper = true)
|
||||
@BanyanDB.IndexMode
|
||||
public class ServiceRelationClientSideMetrics extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "service_relation_client_side";
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
@EqualsAndHashCode(of = {
|
||||
"entityId"
|
||||
}, callSuper = true)
|
||||
@BanyanDB.IndexMode
|
||||
public class ServiceRelationServerSideMetrics extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "service_relation_server_side";
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
"tagValue",
|
||||
"tagType"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class TagAutocompleteData extends Metrics {
|
||||
public static final String INDEX_NAME = "tag_autocomplete";
|
||||
public static final String TAG_KEY = "tag_key";
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ import static org.apache.skywalking.oap.server.core.Const.DOUBLE_COLONS_SPLIT;
|
|||
"name",
|
||||
"layer"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class ServiceTraffic extends Metrics {
|
||||
public static final String INDEX_NAME = "service_traffic";
|
||||
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ import static org.apache.skywalking.oap.server.library.util.StringUtil.isNotBlan
|
|||
callSuper = false,
|
||||
of = "uuid"
|
||||
)
|
||||
@BanyanDB.IndexMode
|
||||
public class Event extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "events";
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ public abstract class PercentMetrics extends Metrics implements IntValueHolder {
|
|||
@Getter
|
||||
@Setter
|
||||
@Column(name = MATCH, storageOnly = true)
|
||||
@BanyanDB.MeasureField
|
||||
private long match;
|
||||
|
||||
@Entrance
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
builder = InstanceHierarchyRelationTraffic.Builder.class, processor = MetricsStreamProcessor.class)
|
||||
@MetricsExtension(supportDownSampling = false, supportUpdate = false)
|
||||
@EqualsAndHashCode(callSuper = false)
|
||||
@BanyanDB.IndexMode
|
||||
public class InstanceHierarchyRelationTraffic extends Metrics {
|
||||
public static final String INDEX_NAME = "instance_hierarchy_relation";
|
||||
public static final String SERVICE_LAYER = "service_layer";
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
builder = ServiceHierarchyRelationTraffic.Builder.class, processor = MetricsStreamProcessor.class)
|
||||
@MetricsExtension(supportDownSampling = false, supportUpdate = false)
|
||||
@EqualsAndHashCode(callSuper = false)
|
||||
@BanyanDB.IndexMode
|
||||
public class ServiceHierarchyRelationTraffic extends Metrics {
|
||||
public static final String INDEX_NAME = "service_hierarchy_relation";
|
||||
public static final String SERVICE_ID = "service_id";
|
||||
|
|
|
|||
|
|
@ -50,6 +50,7 @@ import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.EB
|
|||
"processId",
|
||||
"startTime",
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class EBPFProfilingScheduleRecord extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "ebpf_profiling_schedule";
|
||||
|
|
|
|||
|
|
@ -206,4 +206,15 @@ public @interface BanyanDB {
|
|||
URL
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* IndexMode is used to indicate the index mode of the metric.
|
||||
* All columns in the metric will be stored in the index exclusively.
|
||||
*
|
||||
* @since 10.2.0
|
||||
*/
|
||||
@Target({ElementType.TYPE})
|
||||
@Retention(RetentionPolicy.RUNTIME)
|
||||
@interface IndexMode {
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -52,6 +52,15 @@ public class BanyanDBModelExtension {
|
|||
@Setter
|
||||
private TopN topN;
|
||||
|
||||
/**
|
||||
* indexMode indicates whether a metric is in the index mode.
|
||||
*
|
||||
* @since 10.2.0
|
||||
*/
|
||||
@Getter
|
||||
@Setter
|
||||
private boolean indexMode;
|
||||
|
||||
public static class TopN {
|
||||
/**
|
||||
* lru_size defines how many time_buckets are held in the memory.
|
||||
|
|
|
|||
|
|
@ -99,6 +99,10 @@ public class StorageModels implements IModelManager, ModelCreator, ModelManipula
|
|||
banyanDBModelExtension.setStoreIDTag(true);
|
||||
}
|
||||
|
||||
if (aClass.isAnnotationPresent(BanyanDB.IndexMode.class)) {
|
||||
banyanDBModelExtension.setIndexMode(true);
|
||||
}
|
||||
|
||||
// Set routing rules for ElasticSearch
|
||||
elasticSearchModelExtension.setRouting(storage.getModelName(), modelColumns);
|
||||
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
"serviceName",
|
||||
"remoteServiceName"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class ZipkinServiceRelationTraffic extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "zipkin_service_relation_traffic";
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
"serviceName",
|
||||
"spanName"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class ZipkinServiceSpanTraffic extends Metrics {
|
||||
|
||||
public static final String INDEX_NAME = "zipkin_service_span_traffic";
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
|
|||
@EqualsAndHashCode(of = {
|
||||
"serviceName"
|
||||
})
|
||||
@BanyanDB.IndexMode
|
||||
public class ZipkinServiceTraffic extends Metrics {
|
||||
public static final String INDEX_NAME = "zipkin_service_traffic";
|
||||
|
||||
|
|
|
|||
|
|
@ -219,57 +219,84 @@ storage:
|
|||
asyncBatchPersistentPoolSize: ${SW_STORAGE_ASYNC_BATCH_PERSISTENT_POOL_SIZE:4}
|
||||
banyandb:
|
||||
# Targets is the list of BanyanDB servers, separated by commas.
|
||||
# Each target is a BanyanDB server in the format of `host:port`
|
||||
# The host is the IP address or domain name of the BanyanDB server, and the port is the port number of the BanyanDB server.
|
||||
targets: ${SW_STORAGE_BANYANDB_TARGETS:localhost: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.
|
||||
# Each target is a BanyanDB server in the format of `host:port`.
|
||||
# If BanyanDB is deployed as a standalone server, the target should be the IP address or domain name and port of the BanyanDB server.
|
||||
# If BanyanDB is deployed in a cluster, the targets should be the IP address or domain name and port of the `liaison` nodes, separated by commas.
|
||||
targets: ${SW_STORAGE_BANYANDB_TARGETS:127.0.0.1:17912}
|
||||
|
||||
# The maximum number of records in a bulk write request.
|
||||
# A larger value can improve write performance but also increases OAP and BanyanDB Server memory usage.
|
||||
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.
|
||||
# If the data in a bulk exceeds maxBulkSize, the data will be flushed immediately.
|
||||
# A larger value can reduce write pressure on BanyanDB Server but increase data latency.
|
||||
flushInterval: ${SW_STORAGE_BANYANDB_FLUSH_INTERVAL:15}
|
||||
# The timeout seconds of a bulk flush.
|
||||
|
||||
# The timeout in seconds for 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.
|
||||
recordShardsNumber: ${SW_STORAGE_BANYANDB_RECORD_SHARDS_NUMBER:1}
|
||||
# The multiplier of the number of shards of the super dataset.
|
||||
# Super dataset is a special dataset that stores the trace or log data that is too large to be stored in the normal dataset.
|
||||
# If the normal dataset has `n` shards, the super dataset will have `n * superDatasetShardsFactor` shards.
|
||||
# For example, supposing `recordShardsNumber` is 3, and `superDatasetShardsFactor` is 2,
|
||||
# `segment-default` is a normal dataset that has 3 shards, and `segment-minute` is a super dataset that has 6 shards.
|
||||
superDatasetShardsFactor: ${SW_STORAGE_BANYANDB_SUPERDATASET_SHARDS_FACTOR:2}
|
||||
|
||||
# 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.
|
||||
# A higher value can improve write performance but also increases CPU usage on both OAP and BanyanDB Server.
|
||||
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
|
||||
# The maximum size of dataset when the OAP loads cache, such as network aliases.
|
||||
|
||||
# The maximum size of the dataset when the OAP loads cache, such as network aliases.
|
||||
resultWindowMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_WINDOW_SIZE:10000}
|
||||
|
||||
# The maximum size of metadata per query.
|
||||
metadataQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_SIZE:10000}
|
||||
# The maximum size of trace segments per query.
|
||||
|
||||
# The maximum number of trace segments per query.
|
||||
segmentQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_SEGMENT_SIZE:200}
|
||||
# The max number of profile task query in a request.
|
||||
|
||||
# The maximum number of profile task queries in a request.
|
||||
profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_TASK_SIZE:200}
|
||||
# The batch size of query profiling data.
|
||||
|
||||
# The batch size for querying profile data.
|
||||
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.
|
||||
# The value should be less or equal to data TTL relevant settings.
|
||||
segmentIntervalDays: ${SW_STORAGE_BANYANDB_SEGMENT_INTERVAL_DAYS:1}
|
||||
# The super dataset segment interval is the time range of a segment in the super dataset.
|
||||
superDatasetSegmentIntervalDays: ${SW_STORAGE_BANYANDB_SUPER_DATASET_SEGMENT_INTERVAL_DAYS:1}
|
||||
# Specific groups settings.
|
||||
# For example, {"group1": {"blockIntervalHours": 4, "segmentIntervalDays": 1}}
|
||||
# Please refer to https://github.com/apache/skywalking-banyandb/blob/${BANYANDB_RELEASE}/docs/interacting/bydbctl/schema/group.md#create-operation
|
||||
# for group setting details.
|
||||
specificGroupSettings: ${SW_STORAGE_BANYANDB_SPECIFIC_GROUP_SETTINGS:""}
|
||||
# If the BanyanDB server is configured with TLS, config the TLS cert file path and open tls connection.
|
||||
|
||||
# If the BanyanDB server is configured with TLS, configure the TLS cert file path and enable TLS connection.
|
||||
sslTrustCAPath: ${SW_STORAGE_BANYANDB_SSL_TRUST_CA_PATH:""}
|
||||
|
||||
# The group settings of record.
|
||||
# `gr` is the short name of the group settings of record.
|
||||
#
|
||||
# The "normal" section defines settings for datasets not specified in "super".
|
||||
# Each dataset will be grouped under a single group named "normal".
|
||||
grNormalShardNum: ${SW_STORAGE_BANYANDB_GR_NORMAL_SHARD_NUM:1}
|
||||
grNormalSIDays: ${SW_STORAGE_BANYANDB_GR_NORMAL_SI_DAYS:1}
|
||||
grNormalTTLDays: ${SW_STORAGE_BANYANDB_GR_NORMAL_TTL_DAYS:3}
|
||||
# "super" is a special dataset designed to store trace or log data that is too large for normal datasets.
|
||||
# Each super dataset will be a separate group in BanyanDB, following the settings defined in the "super" section.
|
||||
grSuperShardNum: ${SW_STORAGE_BANYANDB_GR_SUPER_SHARD_NUM:2}
|
||||
grSuperSIDays: ${SW_STORAGE_BANYANDB_GR_SUPER_SI_DAYS:1}
|
||||
grSuperTTLDays: ${SW_STORAGE_BANYANDB_GR_SUPER_TTL_DAYS:3}
|
||||
|
||||
# The group settings of metrics.
|
||||
# `gm` is the short name of the group settings of metrics.
|
||||
#
|
||||
# OAP stores metrics based its granularity.
|
||||
# Valid values are "day", "hour", and "minute". That means metrics will be stored in the three separate groups.
|
||||
# Non-"minute" are governed by the "core.downsampling" setting.
|
||||
# For example, if "core.downsampling" is set to "hour", the "hour" will be used, while "day" are ignored.
|
||||
gmMinuteShardNum: ${SW_STORAGE_BANYANDB_GM_MINUTE_SHARD_NUM:2}
|
||||
gmMinuteSIDays: ${SW_STORAGE_BANYANDB_GM_MINUTE_SI_DAYS:7}
|
||||
gmMinuteTTLDays: ${SW_STORAGE_BANYANDB_GM_MINUTE_TTL_DAYS:7}
|
||||
gmHourShardNum: ${SW_STORAGE_BANYANDB_GM_HOUR_SHARD_NUM:1}
|
||||
gmHourSIDays: ${SW_STORAGE_BANYANDB_GM_HOUR_SI_DAYS:1}
|
||||
gmHourTTLDays: ${SW_STORAGE_BANYANDB_GM_HOUR_TTL_DAYS:7}
|
||||
gmDayShardNum: ${SW_STORAGE_BANYANDB_GM_DAY_SHARD_NUM:1}
|
||||
gmDaySIDays: ${SW_STORAGE_BANYANDB_GM_DAY_SI_DAYS:1}
|
||||
gmDayTTLDays: ${SW_STORAGE_BANYANDB_GM_DAY_TTL_DAYS:30}
|
||||
# If the metrics is marked as "index_mode", the metrics will be stored in the "index" group.
|
||||
# The "index" group is designed to store metrics that are used for indexing without value columns.
|
||||
# Such as `service_traffic`, `network_address_alias`, etc.
|
||||
# "index_mode" requires BanyanDB *0.8.0* or later.
|
||||
gmIndexShardNum: ${SW_STORAGE_BANYANDB_GM_INDEX_SHARD_NUM:1}
|
||||
gmIndexSIDays: ${SW_STORAGE_BANYANDB_GM_INDEX_SI_DAYS:1}
|
||||
gmIndexTTLDays: ${SW_STORAGE_BANYANDB_GM_INDEX_TTL_DAYS:30}
|
||||
|
||||
agent-analyzer:
|
||||
selector: ${SW_AGENT_ANALYZER:default}
|
||||
default:
|
||||
|
|
|
|||
|
|
@ -20,13 +20,13 @@ package org.apache.skywalking.oap.server.storage.plugin.banyandb;
|
|||
|
||||
import io.grpc.Status;
|
||||
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.database.v1.BanyandbDatabase.Measure;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Stream;
|
||||
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.MetadataCache;
|
||||
import org.apache.skywalking.oap.server.core.CoreModule;
|
||||
import org.apache.skywalking.oap.server.core.config.ConfigService;
|
||||
import org.apache.skywalking.oap.server.core.config.DownSamplingConfigService;
|
||||
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;
|
||||
|
|
@ -41,7 +41,6 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
|
|||
public BanyanDBIndexInstaller(Client client, ModuleManager moduleManager, BanyanDBStorageConfig config) {
|
||||
super(client, moduleManager);
|
||||
this.config = config;
|
||||
MetadataRegistry.INSTANCE.initializeIntervals(config.getSpecificGroupSettings());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -49,11 +48,11 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
|
|||
if (!model.isTimeSeries()) {
|
||||
return true;
|
||||
}
|
||||
final ConfigService configService = moduleManager.find(CoreModule.NAME)
|
||||
final DownSamplingConfigService downSamplingConfigService = moduleManager.find(CoreModule.NAME)
|
||||
.provider()
|
||||
.getService(ConfigService.class);
|
||||
.getService(DownSamplingConfigService.class);
|
||||
final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(
|
||||
model, config, configService);
|
||||
model, config, downSamplingConfigService);
|
||||
try {
|
||||
final BanyanDBClient c = ((BanyanDBStorageClient) this.client).client;
|
||||
// first check resource existence and create group if necessary
|
||||
|
|
@ -63,9 +62,9 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
|
|||
} else {
|
||||
// register models only locally(Schema cache) but not remotely
|
||||
if (model.isRecord()) { // stream
|
||||
MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
|
||||
MetadataRegistry.INSTANCE.registerStreamModel(model, config, downSamplingConfigService);
|
||||
} else { // measure
|
||||
MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
|
||||
MetadataRegistry.INSTANCE.registerMeasureModel(model, config, downSamplingConfigService);
|
||||
}
|
||||
// pre-load remote schema for java client
|
||||
MetadataCache.EntityMetadata remoteMeta = metadata.updateRemoteSchema(c);
|
||||
|
|
@ -82,9 +81,9 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
|
|||
@Override
|
||||
public void createTable(Model model) throws StorageException {
|
||||
try {
|
||||
ConfigService configService = moduleManager.find(CoreModule.NAME)
|
||||
DownSamplingConfigService configService = moduleManager.find(CoreModule.NAME)
|
||||
.provider()
|
||||
.getService(ConfigService.class);
|
||||
.getService(DownSamplingConfigService.class);
|
||||
if (model.isRecord()) { // stream
|
||||
StreamModel streamModel = MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
|
||||
Stream stream = streamModel.getStream();
|
||||
|
|
|
|||
|
|
@ -54,44 +54,6 @@ public class BanyanDBStorageConfig extends ModuleConfig {
|
|||
* in a single request.
|
||||
*/
|
||||
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;
|
||||
/**
|
||||
* Default global segment interval for non-super-dataset models.
|
||||
* Unit is day.
|
||||
*
|
||||
* @since 9.4.0
|
||||
*/
|
||||
private int segmentIntervalDays;
|
||||
/**
|
||||
* Default global segment interval for super-dataset models.
|
||||
* Unit is day.
|
||||
*
|
||||
* @since 9.4.0
|
||||
*/
|
||||
private int superDatasetSegmentIntervalDays;
|
||||
/**
|
||||
* Specify the settings for each group individually. All groups created in BanyanDB can
|
||||
* be found with <a href="https://skywalking.apache.org/docs/skywalking-banyandb/next/crud/group/#list-operation">bydbctl</a>.
|
||||
* <p>
|
||||
* NOTE: setting intervals works for all groups except `measure-default`.
|
||||
* <p>
|
||||
* NOTE: available groups: `measure-default`, `measure-sampled`, `stream-default`
|
||||
* and `stream-*` with names of the super dataset as the suffix.
|
||||
*
|
||||
* @since 9.4.0
|
||||
*/
|
||||
private String specificGroupSettings;
|
||||
|
||||
/**
|
||||
* If the BanyanDB server is configured with TLS, config the TLS cert file path and open tls connection.
|
||||
|
|
@ -108,6 +70,29 @@ public class BanyanDBStorageConfig extends ModuleConfig {
|
|||
private int segmentQueryMaxSize = 200;
|
||||
private int profileDataQueryBatchSize = 100;
|
||||
|
||||
/**
|
||||
* The configuration of the groups.
|
||||
* @since 10.2.0
|
||||
*/
|
||||
private int grNormalShardNum = 1;
|
||||
private int grNormalSIDays = 1;
|
||||
private int grNormalTTLDays = 3;
|
||||
private int grSuperShardNum = 2;
|
||||
private int grSuperSIDays = 1;
|
||||
private int grSuperTTLDays = 3;
|
||||
private int gmMinuteShardNum = 2;
|
||||
private int gmMinuteSIDays = 1;
|
||||
private int gmMinuteTTLDays = 7;
|
||||
private int gmHourShardNum = 1;
|
||||
private int gmHourSIDays = 1;
|
||||
private int gmHourTTLDays = 15;
|
||||
private int gmDayShardNum = 1;
|
||||
private int gmDaySIDays = 1;
|
||||
private int gmDayTTLDays = 30;
|
||||
private int gmIndexShardNum = 1;
|
||||
private int gmIndexSIDays = 1;
|
||||
private int gmIndexTTLDays = 30;
|
||||
|
||||
public String[] getTargetArray() {
|
||||
return Iterables.toArray(Splitter.on(",").omitEmptyStrings().trimResults().split(this.targets), String.class);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,12 +18,61 @@
|
|||
|
||||
package org.apache.skywalking.oap.server.storage.plugin.banyandb;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.google.gson.JsonObject;
|
||||
import io.grpc.Status;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.Setter;
|
||||
import lombok.Singular;
|
||||
import lombok.ToString;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Catalog;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Group;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.IntervalRule;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Metadata;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.ResourceOpts;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.CompressionMethod;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.EncodingMethod;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.FieldSpec;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.FieldType;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.IndexRule;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Measure;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Stream;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagFamilySpec;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagSpec;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagType;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TopNAggregation;
|
||||
import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
|
||||
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.Duration;
|
||||
import org.apache.skywalking.banyandb.v1.client.metadata.MetadataCache;
|
||||
import org.apache.skywalking.banyandb.v1.client.metadata.ResourceExist;
|
||||
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.analysis.metrics.Metrics;
|
||||
import org.apache.skywalking.oap.server.core.analysis.record.Record;
|
||||
import org.apache.skywalking.oap.server.core.config.DownSamplingConfigService;
|
||||
import org.apache.skywalking.oap.server.core.query.enumeration.Step;
|
||||
import org.apache.skywalking.oap.server.core.storage.StorageException;
|
||||
import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
|
||||
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
|
||||
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 org.apache.skywalking.oap.server.library.util.CollectionUtils;
|
||||
import org.apache.skywalking.oap.server.library.util.StringUtil;
|
||||
|
||||
import java.io.IOException;
|
||||
import javax.annotation.Nonnull;
|
||||
import javax.annotation.Nullable;
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
|
|
@ -36,58 +85,6 @@ import java.util.Optional;
|
|||
import java.util.Set;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
import javax.annotation.Nonnull;
|
||||
import javax.annotation.Nullable;
|
||||
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.Setter;
|
||||
import lombok.Singular;
|
||||
import lombok.ToString;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Group;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Metadata;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.IntervalRule;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Catalog;
|
||||
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.ResourceOpts;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.IndexRule;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Measure;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.Stream;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagFamilySpec;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagSpec;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TagType;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.TopNAggregation;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.FieldSpec;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.FieldType;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.CompressionMethod;
|
||||
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase.EncodingMethod;
|
||||
import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
|
||||
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.Duration;
|
||||
import org.apache.skywalking.banyandb.v1.client.metadata.MetadataCache;
|
||||
import org.apache.skywalking.banyandb.v1.client.metadata.ResourceExist;
|
||||
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.analysis.metrics.Metrics;
|
||||
import org.apache.skywalking.oap.server.core.analysis.record.Record;
|
||||
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.StorageException;
|
||||
import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
|
||||
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
|
||||
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 org.apache.skywalking.oap.server.library.util.CollectionUtils;
|
||||
import org.apache.skywalking.oap.server.library.util.StringUtil;
|
||||
|
||||
@Slf4j
|
||||
public enum MetadataRegistry {
|
||||
|
|
@ -98,9 +95,7 @@ public enum MetadataRegistry {
|
|||
private static final Set<String> GROUP_ALIGNED = new HashSet<>();
|
||||
private final Map<String, Schema> registry = new HashMap<>();
|
||||
|
||||
private Map<String, GroupSetting> specificGroupSettings = new HashMap<>();
|
||||
|
||||
public StreamModel registerStreamModel(Model model, BanyanDBStorageConfig config, ConfigService configService) {
|
||||
public StreamModel registerStreamModel(Model model, BanyanDBStorageConfig config, DownSamplingConfigService configService) {
|
||||
final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
|
||||
Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
|
||||
Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
|
||||
|
|
@ -143,7 +138,7 @@ public enum MetadataRegistry {
|
|||
return new StreamModel(builder.build(), indexRules);
|
||||
}
|
||||
|
||||
public MeasureModel registerMeasureModel(Model model, BanyanDBStorageConfig config, ConfigService configService) throws StorageException {
|
||||
public MeasureModel registerMeasureModel(Model model, BanyanDBStorageConfig config, DownSamplingConfigService configService) throws StorageException {
|
||||
final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
|
||||
Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
|
||||
Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
|
||||
|
|
@ -180,6 +175,12 @@ public enum MetadataRegistry {
|
|||
builder.setInterval(downSamplingDuration(model.getDownsampling()).format());
|
||||
builder.setEntity(BanyandbDatabase.Entity.newBuilder().addAllTagNames(shardingColumns));
|
||||
builder.addAllTagFamilies(tagFamilySpecs);
|
||||
if (model.getBanyanDBModelExtension().isIndexMode()) {
|
||||
builder.setIndexMode(true);
|
||||
if (!tagsAndFields.fields.isEmpty()) {
|
||||
throw new StorageException("index mode is enabled, but fields are defined");
|
||||
}
|
||||
}
|
||||
// parse and set field
|
||||
for (BanyandbDatabase.FieldSpec field : tagsAndFields.fields) {
|
||||
builder.addFields(field);
|
||||
|
|
@ -465,69 +466,59 @@ public enum MetadataRegistry {
|
|||
return tagSpec.build();
|
||||
}
|
||||
|
||||
public void initializeIntervals(String specificGroupSettingsStr) {
|
||||
if (StringUtil.isBlank(specificGroupSettingsStr)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
specificGroupSettings = MAPPER.readerFor(new TypeReference<Map<String, GroupSetting>>() {
|
||||
}).readValue(specificGroupSettingsStr);
|
||||
} catch (IOException ioEx) {
|
||||
log.warn("fail to parse specificGroupSettings", ioEx);
|
||||
}
|
||||
}
|
||||
|
||||
public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, ConfigService configService) {
|
||||
int segmentIntervalDays = config.getSegmentIntervalDays();
|
||||
if (model.isSuperDataset()) {
|
||||
segmentIntervalDays = config.getSuperDatasetSegmentIntervalDays();
|
||||
}
|
||||
String group;
|
||||
int metricShardNum = config.getMetricsShardsNumber();
|
||||
public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, DownSamplingConfigService configService) {
|
||||
if (model.isRecord()) { // stream
|
||||
group = "stream-default";
|
||||
if (model.isSuperDataset()) {
|
||||
// for superDataset, we should use separate group
|
||||
group = "stream-" + model.getName();
|
||||
}
|
||||
} else if (model.getDownsampling() == DownSampling.Minute && model.isTimeRelativeID()) { // measure
|
||||
group = "measure-minute";
|
||||
// apply super dataset's settings to measure-minute
|
||||
segmentIntervalDays = config.getSuperDatasetSegmentIntervalDays();
|
||||
metricShardNum = metricShardNum * config.getSuperDatasetShardsFactor();
|
||||
} else {
|
||||
// Solution: 2 * TTL < T * (1 + 0.8)
|
||||
// e.g. if TTL=7, T=8: a new block/segment will be created at 14.4 days,
|
||||
// while the first block has been deleted at 2*TTL
|
||||
final int intervalDays = Double.valueOf(Math.ceil(configService.getMetricsDataTTL() * 2.0 / 1.8)).intValue();
|
||||
return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE,
|
||||
model.getDownsampling(),
|
||||
config.getMetricsShardsNumber(),
|
||||
intervalDays, // use 10-day/240-hour strategy
|
||||
configService.getMetricsDataTTL());
|
||||
}
|
||||
|
||||
GroupSetting groupSetting = this.specificGroupSettings.get(group);
|
||||
if (groupSetting != null) {
|
||||
segmentIntervalDays = groupSetting.getSegmentIntervalDays();
|
||||
}
|
||||
if (model.isRecord()) {
|
||||
return new SchemaMetadata(group,
|
||||
return new SchemaMetadata(model.isSuperDataset() ? model.getName() : "normal",
|
||||
model.getName(),
|
||||
Kind.STREAM,
|
||||
model.getDownsampling(),
|
||||
config.getRecordShardsNumber() *
|
||||
(model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1),
|
||||
segmentIntervalDays,
|
||||
configService.getRecordDataTTL()
|
||||
);
|
||||
model.isSuperDataset() ? config.getGrSuperShardNum() : config.getGrNormalShardNum(),
|
||||
model.isSuperDataset() ? config.getGrSuperSIDays() : config.getGrNormalSIDays(),
|
||||
model.isSuperDataset() ? config.getGrSuperTTLDays() : config.getGrNormalTTLDays());
|
||||
}
|
||||
// FIX: address issue #10104
|
||||
return new SchemaMetadata(group, model.getName(), Kind.MEASURE,
|
||||
|
||||
if (model.getBanyanDBModelExtension().isIndexMode()) {
|
||||
return new SchemaMetadata("index", model.getName(), Kind.MEASURE,
|
||||
model.getDownsampling(),
|
||||
metricShardNum,
|
||||
segmentIntervalDays,
|
||||
configService.getMetricsDataTTL());
|
||||
config.getGmIndexShardNum(),
|
||||
config.getGmIndexSIDays(),
|
||||
config.getGmIndexTTLDays());
|
||||
}
|
||||
|
||||
switch (model.getDownsampling()) {
|
||||
case Minute:
|
||||
return new SchemaMetadata(DownSampling.Minute.getName(),
|
||||
model.getName(),
|
||||
Kind.MEASURE,
|
||||
model.getDownsampling(),
|
||||
config.getGmMinuteShardNum(),
|
||||
config.getGmMinuteSIDays(),
|
||||
config.getGmMinuteTTLDays());
|
||||
case Hour:
|
||||
if (!configService.shouldToHour()) {
|
||||
throw new UnsupportedOperationException("downsampling to hour is not supported");
|
||||
}
|
||||
return new SchemaMetadata(DownSampling.Hour.getName(),
|
||||
model.getName(),
|
||||
Kind.MEASURE,
|
||||
model.getDownsampling(),
|
||||
config.getGmHourShardNum(),
|
||||
config.getGmHourSIDays(),
|
||||
config.getGmHourTTLDays());
|
||||
case Day:
|
||||
if (!configService.shouldToDay()) {
|
||||
throw new UnsupportedOperationException("downsampling to day is not supported");
|
||||
}
|
||||
return new SchemaMetadata(DownSampling.Day.getName(),
|
||||
model.getName(),
|
||||
Kind.MEASURE,
|
||||
model.getDownsampling(),
|
||||
config.getGmDayShardNum(),
|
||||
config.getGmDaySIDays(),
|
||||
config.getGmDayTTLDays());
|
||||
default:
|
||||
throw new UnsupportedOperationException("unsupported downSampling interval:" + model.getDownsampling());
|
||||
}
|
||||
}
|
||||
|
||||
@RequiredArgsConstructor
|
||||
|
|
@ -694,7 +685,6 @@ public enum MetadataRegistry {
|
|||
log.info("group {} updated", this.group);
|
||||
}
|
||||
}
|
||||
GROUP_ALIGNED.add(this.group);
|
||||
}
|
||||
return resourceExist.hasResource();
|
||||
default:
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ SW_AGENT_CLIENT_JS_COMMIT=af0565a67d382b683c1dbd94c379b7080db61449
|
|||
SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
|
||||
SW_KUBERNETES_COMMIT_SHA=1335f15bf821a40a7cd71448fa805f0be265afcc
|
||||
SW_ROVER_COMMIT=6bbd39aa701984482330d9dfb4dbaaff0527d55c
|
||||
SW_BANYANDB_COMMIT=9770bf48f9b88574683f56e2a137555cd68066d9
|
||||
SW_BANYANDB_COMMIT=05cf5e1426e0b3926819c2272497923cd705f72b
|
||||
SW_AGENT_PHP_COMMIT=3192c553002707d344bd6774cfab5bc61f67a1d3
|
||||
|
||||
SW_CTL_COMMIT=bce7faaabbd57ed1f40156af8a8eb6c3eccea4ae
|
||||
|
|
|
|||
Loading…
Reference in New Issue