// Copyright 2017 TiKV Project Authors. Licensed under Apache-2.0. // Modifications Copyright 2022 PingCAP, Inc. //! Configuration for the entire server. //! //! TiKV is configured through the `TikvConfig` type, which is in turn //! made up of many other configuration types. use std::{ cmp, collections::{HashMap, HashSet}, convert::TryFrom, error::Error, fs, io::{Error as IoError, Write}, path::Path, str, sync::{Arc, RwLock}, }; use engine_traits::{CF_DEFAULT, CF_LOCK, CF_RAFT, CF_WRITE}; use kvengine::{KvEngineConfig, dfs::DFSConfig}; #[cfg(test)] use kvproto::kvrpcpb::ApiVersion; use lazy_static::lazy_static; use memory_profile::MemoryProfileAutoDumpConfig; use online_config::{ConfigChange, ConfigManager, ConfigValue, OnlineConfig, Result as CfgResult}; use overload_protector::OverloadConfig; use pd_client::Config as PdConfig; use raftstore::{ coprocessor::Config as CopConfig, store::{Config as RaftstoreConfig, SplitConfig}, }; use resource_control::Config as ResourceControlConfig; use resource_metering::Config as ResourceMeteringConfig; use rfengine::RfEngineConfig; use rfstore::store::Config as RfStoreConfig; use security::SecurityConfig; use serde::{ Deserialize, Deserializer, Serialize, Serializer, de::{Error as DError, Unexpected}, }; use serde_json::{Map, Value, to_value}; use tikv_util::{ config::{self, GIB, LogFormat, MIB, ReadableDuration, ReadableSize, TomlWriter}, logger::{get_level_by_string, get_string_by_level, set_log_level, set_txn_info_logging}, sys::SysQuota, task_reschedule::DEFAULT_MAX_TIME_SLICE_US, time::duration_to_sec, yatp_pool, }; use crate::{ import::Config as ImportConfig, server::{ CONFIG_ROCKSDB_GAUGE, Config as ServerConfig, gc_worker::GcConfig, lock_manager::Config as PessimisticTxnConfig, }, storage::config::{Config as StorageConfig, DEFAULT_DATA_DIR}, }; mod rocksdb_config; use rocksdb_config as rocks_config; pub use rocksdb_config::{ BlobRunMode, ChecksumType, CompactionPriority, CompressionType, DBCompactionStyle, DBCompressionType, DBRateLimiterMode, DBRecoveryMode, LogLevel as RocksLogLevel, PrepopulateBlockCache, }; pub const DEFAULT_ROCKSDB_SUB_DIR: &str = "db"; /// By default, block cache size will be set to 35% of system memory. pub const BLOCK_CACHE_RATE: f64 = 0.35; /// By default, TiKV will try to limit its overall memory usage to 75% of /// system memory. This is the primary soft limit that other components /// should respect when sizing their own memory usage. pub const MEMORY_USAGE_LIMIT_RATE: f64 = 0.75; /// By default, TiKV will enable additional throttling once memory usage /// reaches 90% of system memory. Note that this threshold is intentionally /// higher than `MEMORY_USAGE_LIMIT_RATE` and is meant to act as a last-resort /// safeguard close to the OOM boundary, not as the main memory control knob. /// Other mechanisms are expected to take effect before this level is reached, /// and operators should tune both values together with that interaction in /// mind. pub const MEMORY_USAGE_THROTTLING_RATE: f64 = 0.9; /// Min block cache shard's size. If a shard is too small, the index/filter data /// may not fit one shard pub const MIN_BLOCK_CACHE_SHARD_SIZE: usize = 128 * MIB as usize; /// Maximum of 15% of system memory can be used by Raft Engine. Normally its /// memory usage is much smaller than that. const LOCKCF_MIN_MEM: usize = 256 * MIB as usize; const LOCKCF_MAX_MEM: usize = GIB as usize; const RAFT_MIN_MEM: usize = 256 * MIB as usize; const RAFT_MAX_MEM: usize = 2 * GIB as usize; /// Configs that actually took effect in the last run pub const LAST_CONFIG_FILE: &str = "last_tikv.toml"; const TMP_CONFIG_FILE: &str = "tmp_tikv.toml"; const MAX_BLOCK_SIZE: usize = 32 * MIB as usize; const ALIBABA_CLOUD_REGION_ID: &str = "ALIBABA_CLOUD_REGION_ID"; fn memory_limit_for_cf(is_raft_db: bool, cf: &str, total_mem: u64) -> ReadableSize { let (ratio, min, max) = match (is_raft_db, cf) { (true, CF_DEFAULT) => (0.02, RAFT_MIN_MEM, RAFT_MAX_MEM), (false, CF_DEFAULT) => (0.25, 0, usize::MAX), (false, CF_LOCK) => (0.02, LOCKCF_MIN_MEM, LOCKCF_MAX_MEM), (false, CF_WRITE) => (0.15, 0, usize::MAX), _ => unreachable!(), }; let size = ((total_mem as f64 * ratio) as usize).clamp(min, max); ReadableSize::mb(size as u64 / MIB) } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct TitanCfConfig { #[online_config(skip)] pub min_blob_size: ReadableSize, #[online_config(skip)] pub blob_file_compression: CompressionType, #[online_config(skip)] pub blob_cache_size: ReadableSize, #[online_config(skip)] pub min_gc_batch_size: ReadableSize, #[online_config(skip)] pub max_gc_batch_size: ReadableSize, #[online_config(skip)] pub discardable_ratio: f64, // deprecated. #[online_config(skip)] #[doc(hidden)] #[serde(skip_serializing)] pub sample_ratio: Option, #[online_config(skip)] pub merge_small_file_threshold: ReadableSize, pub blob_run_mode: BlobRunMode, #[online_config(skip)] pub level_merge: bool, #[online_config(skip)] pub range_merge: bool, #[online_config(skip)] pub max_sorted_runs: i32, // deprecated. #[online_config(skip)] #[doc(hidden)] #[serde(skip_serializing)] pub gc_merge_rewrite: bool, } impl Default for TitanCfConfig { fn default() -> Self { Self { min_blob_size: ReadableSize::kb(1), // disable titan default blob_file_compression: CompressionType::Lz4, blob_cache_size: ReadableSize::mb(0), min_gc_batch_size: ReadableSize::mb(16), max_gc_batch_size: ReadableSize::mb(64), discardable_ratio: 0.5, sample_ratio: None, merge_small_file_threshold: ReadableSize::mb(8), blob_run_mode: BlobRunMode::Normal, level_merge: false, range_merge: true, max_sorted_runs: 20, gc_merge_rewrite: false, } } } impl TitanCfConfig { fn validate(&self) -> Result<(), Box> { if self.gc_merge_rewrite { return Err( "gc-merge-rewrite is deprecated. The data produced when this \ option is enabled cannot be read by this version. Therefore, if \ this option has been applied to an existing node, you must downgrade \ it to the previous version and fully clean up the old data. See more \ details of how to do that in the documentation for the blob-run-mode \ confuguration." .into(), ); } if self.sample_ratio.is_some() { warn!("sample-ratio is deprecated. Ignoring the value."); } Ok(()) } } #[derive(Clone, Copy, Debug, PartialEq)] struct BackgroundJobLimits { max_background_jobs: u32, max_background_flushes: u32, max_sub_compactions: u32, max_titan_background_gc: u32, } const KVDB_DEFAULT_BACKGROUND_JOB_LIMITS: BackgroundJobLimits = BackgroundJobLimits { max_background_jobs: 9, max_background_flushes: 3, max_sub_compactions: 3, max_titan_background_gc: 4, }; const RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS: BackgroundJobLimits = BackgroundJobLimits { max_background_jobs: 4, max_background_flushes: 1, max_sub_compactions: 2, max_titan_background_gc: 4, }; // `defaults` serves as an upper bound for returning limits. fn get_background_job_limits_impl( cpu_num: u32, defaults: &BackgroundJobLimits, ) -> BackgroundJobLimits { // At the minimum, we should have two background jobs: one for flush and one for // compaction. Otherwise, the number of background jobs should not exceed // cpu_num - 1. let max_background_jobs = cmp::max(2, cmp::min(defaults.max_background_jobs, cpu_num - 1)); // Scale flush threads proportionally to cpu cores. Also make sure the number of // flush threads doesn't exceed total jobs. let max_background_flushes = cmp::min( max_background_jobs.div_ceil(4), defaults.max_background_flushes, ); // Cap max_sub_compactions to allow at least two compactions. let max_compactions = max_background_jobs - max_background_flushes; let max_sub_compactions: u32 = (max_compactions - 1).clamp(1, defaults.max_sub_compactions); // Maximum background GC threads for Titan let max_titan_background_gc = cmp::min(defaults.max_titan_background_gc, cpu_num); BackgroundJobLimits { max_background_jobs, max_background_flushes, max_sub_compactions, max_titan_background_gc, } } fn get_background_job_limits(defaults: &BackgroundJobLimits) -> BackgroundJobLimits { let cpu_num = cmp::max(SysQuota::cpu_cores_quota() as u32, 1); get_background_job_limits_impl(cpu_num, defaults) } macro_rules! cf_config { ($name:ident) => { #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct $name { #[online_config(skip)] pub block_size: ReadableSize, pub block_cache_size: ReadableSize, #[online_config(skip)] pub disable_block_cache: bool, #[online_config(skip)] pub cache_index_and_filter_blocks: bool, #[online_config(skip)] pub pin_l0_filter_and_index_blocks: bool, #[online_config(skip)] pub use_bloom_filter: bool, #[online_config(skip)] pub optimize_filters_for_hits: bool, #[online_config(skip)] pub whole_key_filtering: bool, #[online_config(skip)] pub bloom_filter_bits_per_key: i32, #[online_config(skip)] pub block_based_bloom_filter: bool, #[online_config(skip)] pub read_amp_bytes_per_bit: u32, #[serde(with = "rocks_config::compression_type_level_serde")] #[online_config(skip)] pub compression_per_level: [DBCompressionType; 7], pub write_buffer_size: ReadableSize, pub max_write_buffer_number: i32, #[online_config(skip)] pub min_write_buffer_number_to_merge: i32, pub max_bytes_for_level_base: ReadableSize, pub target_file_size_base: ReadableSize, pub level0_file_num_compaction_trigger: i32, pub level0_slowdown_writes_trigger: Option, pub level0_stop_writes_trigger: Option, pub max_compaction_bytes: ReadableSize, #[serde(with = "rocks_config::compaction_pri_serde")] #[online_config(skip)] pub compaction_pri: CompactionPriority, #[online_config(skip)] pub dynamic_level_bytes: bool, #[online_config(skip)] pub num_levels: i32, pub max_bytes_for_level_multiplier: i32, #[serde(with = "rocks_config::compaction_style_serde")] #[online_config(skip)] pub compaction_style: DBCompactionStyle, pub disable_auto_compactions: bool, pub disable_write_stall: bool, pub soft_pending_compaction_bytes_limit: Option, pub hard_pending_compaction_bytes_limit: Option, #[online_config(skip)] pub force_consistency_checks: bool, #[online_config(skip)] pub enable_doubly_skiplist: bool, #[online_config(skip)] pub enable_compaction_guard: bool, #[online_config(skip)] pub compaction_guard_min_output_file_size: ReadableSize, #[online_config(skip)] pub compaction_guard_max_output_file_size: ReadableSize, #[serde(with = "rocks_config::compression_type_serde")] #[online_config(skip)] pub bottommost_level_compression: DBCompressionType, #[online_config(skip)] pub bottommost_zstd_compression_dict_size: i32, #[online_config(skip)] pub bottommost_zstd_compression_sample_size: i32, #[serde(with = "rocks_config::prepopulate_block_cache_serde")] #[online_config(skip)] pub prepopulate_block_cache: PrepopulateBlockCache, #[online_config(skip)] pub format_version: u32, #[serde(with = "rocks_config::checksum_serde")] #[online_config(skip)] pub checksum: ChecksumType, #[online_config(submodule)] pub titan: TitanCfConfig, } impl $name { fn validate(&self) -> Result<(), Box> { if self.block_size.0 as usize > MAX_BLOCK_SIZE { return Err(format!( "invalid block-size {} for {}, exceed max size {}", self.block_size.0, stringify!($name), MAX_BLOCK_SIZE ) .into()); } if self.format_version > 5 { // TODO: allow version 5 if we have another LTS capable of reading it? return Err("format-version larger than 5 is unsupported".into()); } self.titan.validate()?; Ok(()) } } }; } macro_rules! write_into_metrics { ($cf:expr, $tag:expr, $metrics:expr) => {{ $metrics .with_label_values(&[$tag, "block_size"]) .set($cf.block_size.0 as f64); $metrics .with_label_values(&[$tag, "block_cache_size"]) .set($cf.block_cache_size.0 as f64); $metrics .with_label_values(&[$tag, "disable_block_cache"]) .set(($cf.disable_block_cache as i32).into()); $metrics .with_label_values(&[$tag, "cache_index_and_filter_blocks"]) .set(($cf.cache_index_and_filter_blocks as i32).into()); $metrics .with_label_values(&[$tag, "pin_l0_filter_and_index_blocks"]) .set(($cf.pin_l0_filter_and_index_blocks as i32).into()); $metrics .with_label_values(&[$tag, "use_bloom_filter"]) .set(($cf.use_bloom_filter as i32).into()); $metrics .with_label_values(&[$tag, "optimize_filters_for_hits"]) .set(($cf.optimize_filters_for_hits as i32).into()); $metrics .with_label_values(&[$tag, "whole_key_filtering"]) .set(($cf.whole_key_filtering as i32).into()); $metrics .with_label_values(&[$tag, "bloom_filter_bits_per_key"]) .set($cf.bloom_filter_bits_per_key.into()); $metrics .with_label_values(&[$tag, "block_based_bloom_filter"]) .set(($cf.block_based_bloom_filter as i32).into()); $metrics .with_label_values(&[$tag, "read_amp_bytes_per_bit"]) .set($cf.read_amp_bytes_per_bit.into()); $metrics .with_label_values(&[$tag, "write_buffer_size"]) .set($cf.write_buffer_size.0 as f64); $metrics .with_label_values(&[$tag, "max_write_buffer_number"]) .set($cf.max_write_buffer_number.into()); $metrics .with_label_values(&[$tag, "min_write_buffer_number_to_merge"]) .set($cf.min_write_buffer_number_to_merge.into()); $metrics .with_label_values(&[$tag, "max_bytes_for_level_base"]) .set($cf.max_bytes_for_level_base.0 as f64); $metrics .with_label_values(&[$tag, "target_file_size_base"]) .set($cf.target_file_size_base.0 as f64); $metrics .with_label_values(&[$tag, "level0_file_num_compaction_trigger"]) .set($cf.level0_file_num_compaction_trigger.into()); $metrics .with_label_values(&[$tag, "level0_slowdown_writes_trigger"]) .set( $cf.level0_slowdown_writes_trigger .unwrap_or_default() .into(), ); $metrics .with_label_values(&[$tag, "level0_stop_writes_trigger"]) .set($cf.level0_stop_writes_trigger.unwrap_or_default().into()); $metrics .with_label_values(&[$tag, "max_compaction_bytes"]) .set($cf.max_compaction_bytes.0 as f64); $metrics .with_label_values(&[$tag, "dynamic_level_bytes"]) .set(($cf.dynamic_level_bytes as i32).into()); $metrics .with_label_values(&[$tag, "num_levels"]) .set($cf.num_levels.into()); $metrics .with_label_values(&[$tag, "max_bytes_for_level_multiplier"]) .set($cf.max_bytes_for_level_multiplier.into()); $metrics .with_label_values(&[$tag, "disable_auto_compactions"]) .set(($cf.disable_auto_compactions as i32).into()); $metrics .with_label_values(&[$tag, "disable_write_stall"]) .set(($cf.disable_write_stall as i32).into()); $metrics .with_label_values(&[$tag, "soft_pending_compaction_bytes_limit"]) .set( $cf.soft_pending_compaction_bytes_limit .unwrap_or_default() .0 as f64, ); $metrics .with_label_values(&[$tag, "hard_pending_compaction_bytes_limit"]) .set( $cf.hard_pending_compaction_bytes_limit .unwrap_or_default() .0 as f64, ); $metrics .with_label_values(&[$tag, "force_consistency_checks"]) .set(($cf.force_consistency_checks as i32).into()); $metrics .with_label_values(&[$tag, "enable_doubly_skiplist"]) .set(($cf.enable_doubly_skiplist as i32).into()); $metrics .with_label_values(&[$tag, "titan_min_blob_size"]) .set($cf.titan.min_blob_size.0 as f64); $metrics .with_label_values(&[$tag, "titan_blob_cache_size"]) .set($cf.titan.blob_cache_size.0 as f64); $metrics .with_label_values(&[$tag, "titan_min_gc_batch_size"]) .set($cf.titan.min_gc_batch_size.0 as f64); $metrics .with_label_values(&[$tag, "titan_max_gc_batch_size"]) .set($cf.titan.max_gc_batch_size.0 as f64); $metrics .with_label_values(&[$tag, "titan_discardable_ratio"]) .set($cf.titan.discardable_ratio); $metrics .with_label_values(&[$tag, "titan_merge_small_file_threshold"]) .set($cf.titan.merge_small_file_threshold.0 as f64); }}; } cf_config!(DefaultCfConfig); impl Default for DefaultCfConfig { fn default() -> DefaultCfConfig { let total_mem = SysQuota::memory_limit_in_bytes(); DefaultCfConfig { block_size: ReadableSize::kb(64), block_cache_size: memory_limit_for_cf(false, CF_DEFAULT, total_mem), disable_block_cache: false, cache_index_and_filter_blocks: true, pin_l0_filter_and_index_blocks: true, use_bloom_filter: true, optimize_filters_for_hits: true, whole_key_filtering: true, bloom_filter_bits_per_key: 10, block_based_bloom_filter: false, read_amp_bytes_per_bit: 0, compression_per_level: [ DBCompressionType::No, DBCompressionType::No, DBCompressionType::Lz4, DBCompressionType::Lz4, DBCompressionType::Lz4, DBCompressionType::Zstd, DBCompressionType::Zstd, ], write_buffer_size: ReadableSize::mb(128), max_write_buffer_number: 5, min_write_buffer_number_to_merge: 1, max_bytes_for_level_base: ReadableSize::mb(512), target_file_size_base: ReadableSize::mb(8), level0_file_num_compaction_trigger: 4, level0_slowdown_writes_trigger: None, level0_stop_writes_trigger: None, max_compaction_bytes: ReadableSize::gb(2), compaction_pri: CompactionPriority::MinOverlappingRatio, dynamic_level_bytes: true, num_levels: 7, max_bytes_for_level_multiplier: 10, compaction_style: DBCompactionStyle::Level, disable_auto_compactions: false, disable_write_stall: false, soft_pending_compaction_bytes_limit: None, hard_pending_compaction_bytes_limit: None, force_consistency_checks: false, enable_doubly_skiplist: true, enable_compaction_guard: true, compaction_guard_min_output_file_size: ReadableSize::mb(8), compaction_guard_max_output_file_size: ReadableSize::mb(128), bottommost_level_compression: DBCompressionType::Zstd, bottommost_zstd_compression_dict_size: 0, bottommost_zstd_compression_sample_size: 0, prepopulate_block_cache: PrepopulateBlockCache::Disabled, format_version: 2, checksum: ChecksumType::CRC32c, titan: TitanCfConfig::default(), } } } cf_config!(WriteCfConfig); impl Default for WriteCfConfig { fn default() -> WriteCfConfig { let total_mem = SysQuota::memory_limit_in_bytes(); // Setting blob_run_mode=read_only effectively disable Titan. let titan = TitanCfConfig { blob_run_mode: BlobRunMode::ReadOnly, ..Default::default() }; WriteCfConfig { block_size: ReadableSize::kb(32), block_cache_size: memory_limit_for_cf(false, CF_WRITE, total_mem), disable_block_cache: false, cache_index_and_filter_blocks: true, pin_l0_filter_and_index_blocks: true, use_bloom_filter: true, optimize_filters_for_hits: false, whole_key_filtering: false, bloom_filter_bits_per_key: 10, block_based_bloom_filter: false, read_amp_bytes_per_bit: 0, compression_per_level: [ DBCompressionType::No, DBCompressionType::No, DBCompressionType::Lz4, DBCompressionType::Lz4, DBCompressionType::Lz4, DBCompressionType::Zstd, DBCompressionType::Zstd, ], write_buffer_size: ReadableSize::mb(128), max_write_buffer_number: 5, min_write_buffer_number_to_merge: 1, max_bytes_for_level_base: ReadableSize::mb(512), target_file_size_base: ReadableSize::mb(16), level0_file_num_compaction_trigger: 4, level0_slowdown_writes_trigger: None, level0_stop_writes_trigger: None, max_compaction_bytes: ReadableSize::gb(2), compaction_pri: CompactionPriority::MinOverlappingRatio, dynamic_level_bytes: true, num_levels: 7, max_bytes_for_level_multiplier: 10, compaction_style: DBCompactionStyle::Level, disable_auto_compactions: false, disable_write_stall: false, soft_pending_compaction_bytes_limit: None, hard_pending_compaction_bytes_limit: None, force_consistency_checks: false, enable_doubly_skiplist: true, enable_compaction_guard: true, compaction_guard_min_output_file_size: ReadableSize::mb(8), compaction_guard_max_output_file_size: ReadableSize::mb(128), bottommost_level_compression: DBCompressionType::Zstd, bottommost_zstd_compression_dict_size: 0, bottommost_zstd_compression_sample_size: 0, prepopulate_block_cache: PrepopulateBlockCache::Disabled, format_version: 2, checksum: ChecksumType::CRC32c, titan, } } } cf_config!(LockCfConfig); impl Default for LockCfConfig { fn default() -> LockCfConfig { let total_mem = SysQuota::memory_limit_in_bytes(); // Setting blob_run_mode=read_only effectively disable Titan. let titan = TitanCfConfig { blob_run_mode: BlobRunMode::ReadOnly, ..Default::default() }; LockCfConfig { block_size: ReadableSize::kb(16), block_cache_size: memory_limit_for_cf(false, CF_LOCK, total_mem), disable_block_cache: false, cache_index_and_filter_blocks: true, pin_l0_filter_and_index_blocks: true, use_bloom_filter: true, optimize_filters_for_hits: false, whole_key_filtering: true, bloom_filter_bits_per_key: 10, block_based_bloom_filter: false, read_amp_bytes_per_bit: 0, compression_per_level: [DBCompressionType::No; 7], write_buffer_size: ReadableSize::mb(32), max_write_buffer_number: 5, min_write_buffer_number_to_merge: 1, max_bytes_for_level_base: ReadableSize::mb(128), target_file_size_base: ReadableSize::mb(8), level0_file_num_compaction_trigger: 1, level0_slowdown_writes_trigger: None, level0_stop_writes_trigger: None, max_compaction_bytes: ReadableSize::gb(2), compaction_pri: CompactionPriority::ByCompensatedSize, dynamic_level_bytes: true, num_levels: 7, max_bytes_for_level_multiplier: 10, compaction_style: DBCompactionStyle::Level, disable_auto_compactions: false, disable_write_stall: false, soft_pending_compaction_bytes_limit: None, hard_pending_compaction_bytes_limit: None, force_consistency_checks: false, enable_doubly_skiplist: true, enable_compaction_guard: false, compaction_guard_min_output_file_size: ReadableSize::mb(8), compaction_guard_max_output_file_size: ReadableSize::mb(128), bottommost_level_compression: DBCompressionType::Disable, bottommost_zstd_compression_dict_size: 0, bottommost_zstd_compression_sample_size: 0, prepopulate_block_cache: PrepopulateBlockCache::Disabled, format_version: 2, checksum: ChecksumType::CRC32c, titan, } } } cf_config!(RaftCfConfig); impl Default for RaftCfConfig { fn default() -> RaftCfConfig { // Setting blob_run_mode=read_only effectively disable Titan. let titan = TitanCfConfig { blob_run_mode: BlobRunMode::ReadOnly, ..Default::default() }; RaftCfConfig { block_size: ReadableSize::kb(16), block_cache_size: ReadableSize::mb(128), disable_block_cache: false, cache_index_and_filter_blocks: true, pin_l0_filter_and_index_blocks: true, use_bloom_filter: true, optimize_filters_for_hits: true, whole_key_filtering: true, bloom_filter_bits_per_key: 10, block_based_bloom_filter: false, read_amp_bytes_per_bit: 0, compression_per_level: [DBCompressionType::No; 7], write_buffer_size: ReadableSize::mb(128), max_write_buffer_number: 5, min_write_buffer_number_to_merge: 1, max_bytes_for_level_base: ReadableSize::mb(128), target_file_size_base: ReadableSize::mb(8), level0_file_num_compaction_trigger: 1, level0_slowdown_writes_trigger: None, level0_stop_writes_trigger: None, max_compaction_bytes: ReadableSize::gb(2), compaction_pri: CompactionPriority::ByCompensatedSize, dynamic_level_bytes: true, num_levels: 7, max_bytes_for_level_multiplier: 10, compaction_style: DBCompactionStyle::Level, disable_auto_compactions: false, disable_write_stall: false, soft_pending_compaction_bytes_limit: None, hard_pending_compaction_bytes_limit: None, force_consistency_checks: false, enable_doubly_skiplist: true, enable_compaction_guard: false, compaction_guard_min_output_file_size: ReadableSize::mb(8), compaction_guard_max_output_file_size: ReadableSize::mb(128), bottommost_level_compression: DBCompressionType::Disable, bottommost_zstd_compression_dict_size: 0, bottommost_zstd_compression_sample_size: 0, prepopulate_block_cache: PrepopulateBlockCache::Disabled, format_version: 2, checksum: ChecksumType::CRC32c, titan, } } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug)] #[serde(default)] #[serde(rename_all = "kebab-case")] // Note that Titan is still an experimental feature. Once enabled, it can't fall // back. Forced fallback may result in data loss. pub struct TitanDbConfig { pub enabled: bool, pub dirname: String, pub disable_gc: bool, pub max_background_gc: i32, // The value of this field will be truncated to seconds. pub purge_obsolete_files_period: ReadableDuration, } impl Default for TitanDbConfig { fn default() -> Self { Self { enabled: false, dirname: "".to_owned(), disable_gc: false, max_background_gc: 4, purge_obsolete_files_period: ReadableDuration::secs(10), } } } impl TitanDbConfig { fn validate(&self) -> Result<(), Box> { Ok(()) } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct DbConfig { #[online_config(skip)] pub info_log_level: RocksLogLevel, #[serde(with = "rocks_config::recovery_mode_serde")] #[online_config(skip)] pub wal_recovery_mode: DBRecoveryMode, #[online_config(skip)] pub wal_dir: String, #[online_config(skip)] pub wal_ttl_seconds: u64, #[online_config(skip)] pub wal_size_limit: ReadableSize, pub max_total_wal_size: ReadableSize, pub max_background_jobs: i32, pub max_background_flushes: i32, #[online_config(skip)] pub max_manifest_file_size: ReadableSize, #[online_config(skip)] pub create_if_missing: bool, pub max_open_files: i32, #[online_config(skip)] pub enable_statistics: bool, #[online_config(skip)] pub stats_dump_period: ReadableDuration, pub compaction_readahead_size: ReadableSize, #[online_config(skip)] pub info_log_max_size: ReadableSize, #[online_config(skip)] pub info_log_roll_time: ReadableDuration, #[online_config(skip)] pub info_log_keep_log_file_num: u64, #[online_config(skip)] pub info_log_dir: String, pub rate_bytes_per_sec: ReadableSize, #[online_config(skip)] pub rate_limiter_refill_period: ReadableDuration, #[serde(with = "rocks_config::rate_limiter_mode_serde")] #[online_config(skip)] pub rate_limiter_mode: DBRateLimiterMode, // deprecated. use rate_limiter_auto_tuned. #[online_config(skip)] #[doc(hidden)] #[serde(skip_serializing)] pub auto_tuned: Option, pub rate_limiter_auto_tuned: bool, pub bytes_per_sync: ReadableSize, pub wal_bytes_per_sync: ReadableSize, #[online_config(skip)] pub max_sub_compactions: u32, pub writable_file_max_buffer_size: ReadableSize, #[online_config(skip)] pub use_direct_io_for_flush_and_compaction: bool, #[online_config(skip)] pub enable_pipelined_write: bool, // deprecated. TiKV will use a new write mode when set `enable_pipelined_write` false and fall // back to write mode in 3.0 when set `enable_pipelined_write` true. The code of // multi-batch-write in RocksDB has been removed. #[online_config(skip)] #[serde(skip_serializing)] pub enable_multi_batch_write: bool, #[online_config(skip)] pub enable_unordered_write: bool, #[online_config(submodule)] pub defaultcf: DefaultCfConfig, #[online_config(submodule)] pub writecf: WriteCfConfig, #[online_config(submodule)] pub lockcf: LockCfConfig, #[online_config(submodule)] pub raftcf: RaftCfConfig, #[online_config(skip)] pub titan: TitanDbConfig, } impl Default for DbConfig { fn default() -> DbConfig { let bg_job_limits = get_background_job_limits(&KVDB_DEFAULT_BACKGROUND_JOB_LIMITS); let titan_config = TitanDbConfig { max_background_gc: bg_job_limits.max_titan_background_gc as i32, ..Default::default() }; DbConfig { wal_recovery_mode: DBRecoveryMode::PointInTime, wal_dir: "".to_owned(), wal_ttl_seconds: 0, wal_size_limit: ReadableSize::kb(0), max_total_wal_size: ReadableSize::gb(4), max_background_jobs: bg_job_limits.max_background_jobs as i32, max_background_flushes: bg_job_limits.max_background_flushes as i32, max_manifest_file_size: ReadableSize::mb(128), create_if_missing: true, max_open_files: 40960, enable_statistics: true, stats_dump_period: ReadableDuration::minutes(10), compaction_readahead_size: ReadableSize::kb(0), info_log_max_size: ReadableSize::gb(1), info_log_roll_time: ReadableDuration::secs(0), info_log_keep_log_file_num: 10, info_log_dir: "".to_owned(), info_log_level: RocksLogLevel::Info, rate_bytes_per_sec: ReadableSize::gb(10), rate_limiter_refill_period: ReadableDuration::millis(100), rate_limiter_mode: DBRateLimiterMode::WriteOnly, auto_tuned: None, // deprecated rate_limiter_auto_tuned: true, bytes_per_sync: ReadableSize::mb(1), wal_bytes_per_sync: ReadableSize::kb(512), max_sub_compactions: bg_job_limits.max_sub_compactions, writable_file_max_buffer_size: ReadableSize::mb(1), use_direct_io_for_flush_and_compaction: false, enable_pipelined_write: false, enable_multi_batch_write: true, // deprecated enable_unordered_write: false, defaultcf: DefaultCfConfig::default(), writecf: WriteCfConfig::default(), lockcf: LockCfConfig::default(), raftcf: RaftCfConfig::default(), titan: titan_config, } } } impl DbConfig { fn validate(&mut self) -> Result<(), Box> { self.defaultcf.validate()?; self.lockcf.validate()?; self.writecf.validate()?; self.raftcf.validate()?; self.titan.validate()?; if self.enable_unordered_write { if self.titan.enabled { return Err("RocksDB.unordered_write does not support Titan".into()); } self.enable_pipelined_write = false; } // Since the following configuration supports online update, in order to // prevent mistakenly inputting too large values, the max limit is made // according to the cpu quota * 10. Notice 10 is only an estimate, not an // empirical value. let limit = SysQuota::cpu_cores_quota() as i32 * 10; if self.max_background_jobs <= 0 || self.max_background_jobs > limit { return Err(format!( "max_background_jobs should be greater than 0 and less than or equal to {:?}", limit, ) .into()); } if self.max_sub_compactions == 0 || self.max_sub_compactions as i32 > self.max_background_jobs { return Err(format!( "max_sub_compactions should be greater than 0 and less than or equal to {:?}", self.max_background_jobs, ) .into()); } if self.max_background_flushes <= 0 || self.max_background_flushes > limit { return Err(format!( "max_background_flushes should be greater than 0 and less than or equal to {:?}", limit, ) .into()); } Ok(()) } fn write_into_metrics(&self) { write_into_metrics!(self.defaultcf, CF_DEFAULT, CONFIG_ROCKSDB_GAUGE); write_into_metrics!(self.lockcf, CF_LOCK, CONFIG_ROCKSDB_GAUGE); write_into_metrics!(self.writecf, CF_WRITE, CONFIG_ROCKSDB_GAUGE); write_into_metrics!(self.raftcf, CF_RAFT, CONFIG_ROCKSDB_GAUGE); } } cf_config!(RaftDefaultCfConfig); impl Default for RaftDefaultCfConfig { fn default() -> RaftDefaultCfConfig { let total_mem = SysQuota::memory_limit_in_bytes(); RaftDefaultCfConfig { block_size: ReadableSize::kb(64), block_cache_size: memory_limit_for_cf(true, CF_DEFAULT, total_mem), disable_block_cache: false, cache_index_and_filter_blocks: true, pin_l0_filter_and_index_blocks: true, use_bloom_filter: false, optimize_filters_for_hits: true, whole_key_filtering: true, bloom_filter_bits_per_key: 10, block_based_bloom_filter: false, read_amp_bytes_per_bit: 0, compression_per_level: [ DBCompressionType::No, DBCompressionType::No, DBCompressionType::Lz4, DBCompressionType::Lz4, DBCompressionType::Lz4, DBCompressionType::Zstd, DBCompressionType::Zstd, ], write_buffer_size: ReadableSize::mb(128), max_write_buffer_number: 5, min_write_buffer_number_to_merge: 1, max_bytes_for_level_base: ReadableSize::mb(512), target_file_size_base: ReadableSize::mb(8), level0_file_num_compaction_trigger: 4, level0_slowdown_writes_trigger: None, level0_stop_writes_trigger: None, max_compaction_bytes: ReadableSize::gb(2), compaction_pri: CompactionPriority::ByCompensatedSize, dynamic_level_bytes: true, num_levels: 7, max_bytes_for_level_multiplier: 10, compaction_style: DBCompactionStyle::Level, disable_auto_compactions: false, disable_write_stall: false, soft_pending_compaction_bytes_limit: None, hard_pending_compaction_bytes_limit: None, force_consistency_checks: false, enable_doubly_skiplist: true, enable_compaction_guard: false, compaction_guard_min_output_file_size: ReadableSize::mb(8), compaction_guard_max_output_file_size: ReadableSize::mb(128), bottommost_level_compression: DBCompressionType::Disable, bottommost_zstd_compression_dict_size: 0, bottommost_zstd_compression_sample_size: 0, prepopulate_block_cache: PrepopulateBlockCache::Disabled, format_version: 2, checksum: ChecksumType::CRC32c, titan: TitanCfConfig::default(), } } } // RocksDB Env associate thread pools of multiple instances from the same // process. When construct Options, options.env is set to same singleton // Env::Default() object. So total max_background_jobs = // max(rocksdb.max_background_jobs, raftdb.max_background_jobs) // But each instance will limit their background jobs according to their own // max_background_jobs #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct RaftDbConfig { #[serde(with = "rocks_config::recovery_mode_serde")] #[online_config(skip)] pub wal_recovery_mode: DBRecoveryMode, #[online_config(skip)] pub wal_dir: String, #[online_config(skip)] pub wal_ttl_seconds: u64, #[online_config(skip)] pub wal_size_limit: ReadableSize, pub max_total_wal_size: ReadableSize, pub max_background_jobs: i32, pub max_background_flushes: i32, #[online_config(skip)] pub max_manifest_file_size: ReadableSize, #[online_config(skip)] pub create_if_missing: bool, pub max_open_files: i32, #[online_config(skip)] pub enable_statistics: bool, #[online_config(skip)] pub stats_dump_period: ReadableDuration, pub compaction_readahead_size: ReadableSize, #[online_config(skip)] pub info_log_max_size: ReadableSize, #[online_config(skip)] pub info_log_roll_time: ReadableDuration, #[online_config(skip)] pub info_log_keep_log_file_num: u64, #[online_config(skip)] pub info_log_dir: String, #[online_config(skip)] pub info_log_level: RocksLogLevel, pub max_sub_compactions: u32, pub writable_file_max_buffer_size: ReadableSize, #[online_config(skip)] pub use_direct_io_for_flush_and_compaction: bool, #[online_config(skip)] pub enable_pipelined_write: bool, #[online_config(skip)] pub enable_unordered_write: bool, #[online_config(skip)] pub allow_concurrent_memtable_write: bool, pub bytes_per_sync: ReadableSize, pub wal_bytes_per_sync: ReadableSize, #[online_config(submodule)] pub defaultcf: RaftDefaultCfConfig, #[online_config(skip)] pub titan: TitanDbConfig, } impl Default for RaftDbConfig { fn default() -> RaftDbConfig { let bg_job_limits = get_background_job_limits(&RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS); let titan_config = TitanDbConfig { max_background_gc: bg_job_limits.max_titan_background_gc as i32, ..Default::default() }; RaftDbConfig { wal_recovery_mode: DBRecoveryMode::PointInTime, wal_dir: "".to_owned(), wal_ttl_seconds: 0, wal_size_limit: ReadableSize::kb(0), max_total_wal_size: ReadableSize::gb(4), max_background_jobs: bg_job_limits.max_background_jobs as i32, max_background_flushes: bg_job_limits.max_background_flushes as i32, max_manifest_file_size: ReadableSize::mb(20), create_if_missing: true, max_open_files: 40960, enable_statistics: true, stats_dump_period: ReadableDuration::minutes(10), compaction_readahead_size: ReadableSize::kb(0), info_log_max_size: ReadableSize::gb(1), info_log_roll_time: ReadableDuration::secs(0), info_log_keep_log_file_num: 10, info_log_dir: "".to_owned(), info_log_level: RocksLogLevel::Info, max_sub_compactions: bg_job_limits.max_sub_compactions, writable_file_max_buffer_size: ReadableSize::mb(1), use_direct_io_for_flush_and_compaction: false, enable_pipelined_write: true, enable_unordered_write: false, allow_concurrent_memtable_write: true, bytes_per_sync: ReadableSize::mb(1), wal_bytes_per_sync: ReadableSize::kb(512), defaultcf: RaftDefaultCfConfig::default(), titan: titan_config, } } } impl RaftDbConfig { fn validate(&mut self) -> Result<(), Box> { self.defaultcf.validate()?; if self.enable_unordered_write { if self.titan.enabled { return Err("raftdb: unordered_write is not compatible with Titan".into()); } if self.enable_pipelined_write { return Err( "raftdb: pipelined_write is not compatible with unordered_write".into(), ); } } Ok(()) } } #[derive(Clone, Copy, Debug)] pub enum DbType { Kv, Raft, } // Convert `ConfigValue` to formatted String that can pass to // `DB::set_db_options` #[cfg(test)] fn config_value_to_string(config_change: Vec<(String, ConfigValue)>) -> Vec<(String, String)> { config_change .into_iter() .filter_map(|(name, value)| { let v = match value { d @ ConfigValue::Duration(_) => { let d: ReadableDuration = d.into(); Some(d.as_secs().to_string()) } s @ ConfigValue::Size(_) => { let s: ReadableSize = s.into(); Some(s.0.to_string()) } ConfigValue::Module(_) => unreachable!(), v => Some(format!("{}", v)), }; v.map(|v| (name, v)) }) .collect() } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct MetricConfig { pub job: String, // Push is deprecated. #[doc(hidden)] #[serde(skip_serializing)] pub interval: ReadableDuration, #[doc(hidden)] #[serde(skip_serializing)] pub address: String, } impl Default for MetricConfig { fn default() -> MetricConfig { MetricConfig { interval: ReadableDuration::secs(15), address: "".to_owned(), job: "tikv".to_owned(), } } } #[derive(Clone, Copy, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct FairnessConfig { #[online_config(skip)] pub enabled: bool, /// Online override for the fair global CPU gate's read CPU ratio. /// /// `0` inherits the startup-only /// `resource-control.max-read-cpu-ratio`. A positive value affects only /// the keyspace-fair YATP read pool; the legacy admission path continues /// to use the resource-control value. It has no runtime effect unless /// keyspace fairness and its global CPU gate are active. pub max_read_cpu_ratio: f64, /// Virtual time charged before polling one task. /// /// It is recommended to keep this much smaller than the YATP keyspace fair /// queue's max virtual-time lag. Multiple read-pool workers can precharge /// the same keyspace concurrently; a small value keeps the temporary global /// max virtual-time advance bounded because later correction does not roll /// back that global watermark. pub slice_precharge_us: u64, pub use_wall_time: bool, pub max_tasks_per_keyspace: usize, /// Wall-time slice a read-pool task may poll before it cooperatively /// reschedules. /// /// A shorter slice lets a keyspace that is behind on virtual time reach the /// worker sooner, which bounds the latency a bursting keyspace can add to /// the others. It also reschedules long tasks more often, so the scheduling /// overhead a task pays grows as the slice shrinks. pub max_time_slice_us: u64, } impl Default for FairnessConfig { fn default() -> Self { Self { enabled: false, max_read_cpu_ratio: 0.0, slice_precharge_us: 0, use_wall_time: true, max_tasks_per_keyspace: DEFAULT_KEYSPACE_FAIRNESS_MAX_TASKS_PER_KEYSPACE, max_time_slice_us: DEFAULT_MAX_TIME_SLICE_US, } } } const MAX_KEYSPACE_FAIRNESS_SLICE_PRECHARGE_US: u64 = 1_000_000; const DEFAULT_KEYSPACE_FAIRNESS_MAX_TASKS_PER_KEYSPACE: usize = 2048; const MIN_KEYSPACE_FAIRNESS_MAX_READ_CPU_RATIO: f64 = 0.01; const MIN_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US: u64 = 100; const MAX_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US: u64 = 100_000; #[derive(Clone, Copy, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct UnifiedReadPoolConfig { #[online_config(skip)] pub min_thread_count: usize, pub max_thread_count: usize, #[online_config(skip)] pub stack_size: ReadableSize, #[online_config(skip)] pub max_tasks_per_worker: usize, pub auto_adjust_pool_size: bool, // FIXME: Add more configs when they are effective in yatp pub use_tokio: bool, /// Enables per-keyspace YATP metrics independently of keyspace fairness. /// Disabled by default to avoid unbounded metric cardinality. #[online_config(skip)] pub enable_yatp_keyspace_metrics: bool, #[online_config(submodule)] pub keyspace_fairness: FairnessConfig, } impl UnifiedReadPoolConfig { fn adjust_cpu_limited_worker_count(&mut self, cpu_cores_quota: f64) { let limit = cmp::max(UNIFIED_READPOOL_MIN_CONCURRENCY, cpu_cores_quota as usize); if self.max_thread_count > limit { warn!( "configured thread pool size exceeds its limit, adjusting to the limit"; "config" => "readpool.unified.max-thread-count", "configured" => self.max_thread_count, "limit" => limit, ); self.max_thread_count = limit; } } fn validate(&self) -> Result<(), Box> { if self.min_thread_count == 0 { return Err("readpool.unified.min-thread-count should be > 0".into()); } if self.max_thread_count < self.min_thread_count { return Err( "readpool.unified.max-thread-count should be >= readpool.unified.min-thread-count" .into(), ); } let limit = cmp::max( UNIFIED_READPOOL_MIN_CONCURRENCY, SysQuota::cpu_cores_quota() as usize, ); if self.max_thread_count > limit { return Err(format!( "readpool.unified.max-thread-count should be smaller than {}", limit ) .into()); } if self.stack_size.0 < ReadableSize::mb(2).0 { return Err("readpool.unified.stack-size should be >= 2mb".into()); } if self.max_tasks_per_worker <= 1 { return Err("readpool.unified.max-tasks-per-worker should be > 1".into()); } // disable auto_adjust_pool_size when tokio is used. if self.use_tokio && self.auto_adjust_pool_size { return Err("auto_adjust_pool_size is not supported when tokio is used".into()); } if self.use_tokio && self.keyspace_fairness.enabled { return Err("enable_keyspace_fairness is not supported when tokio is used".into()); } let fairness_max_read_cpu_ratio = self.keyspace_fairness.max_read_cpu_ratio; if fairness_max_read_cpu_ratio != 0.0 && !(MIN_KEYSPACE_FAIRNESS_MAX_READ_CPU_RATIO..1.0) .contains(&fairness_max_read_cpu_ratio) { return Err(format!( "readpool.unified.keyspace-fairness.max-read-cpu-ratio must be 0 (inherit) or a float in the range [{}, 1.0)", MIN_KEYSPACE_FAIRNESS_MAX_READ_CPU_RATIO ) .into()); } if self.keyspace_fairness.slice_precharge_us > MAX_KEYSPACE_FAIRNESS_SLICE_PRECHARGE_US { return Err(format!( "readpool.unified.keyspace-fairness.slice-precharge-us should be <= {}", MAX_KEYSPACE_FAIRNESS_SLICE_PRECHARGE_US ) .into()); } if self.keyspace_fairness.max_tasks_per_keyspace <= 1 { return Err( "readpool.unified.keyspace-fairness.max-tasks-per-keyspace should be > 1".into(), ); } if !(MIN_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US..=MAX_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US) .contains(&self.keyspace_fairness.max_time_slice_us) { return Err(format!( "readpool.unified.keyspace-fairness.max-time-slice-us should be between {} and {}", MIN_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US, MAX_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US ) .into()); } Ok(()) } } pub const UNIFIED_READPOOL_MIN_CONCURRENCY: usize = 4; // FIXME: Use macros to generate it if yatp is used elsewhere besides readpool. impl Default for UnifiedReadPoolConfig { fn default() -> UnifiedReadPoolConfig { let cpu_num = SysQuota::cpu_cores_quota(); let mut concurrency = cpu_num as usize; concurrency = cmp::max(UNIFIED_READPOOL_MIN_CONCURRENCY, concurrency); Self { min_thread_count: 1, max_thread_count: concurrency, stack_size: ReadableSize::mb(DEFAULT_READPOOL_STACK_SIZE_MB), max_tasks_per_worker: DEFAULT_READPOOL_MAX_TASKS_PER_WORKER, auto_adjust_pool_size: false, use_tokio: false, enable_yatp_keyspace_metrics: false, keyspace_fairness: FairnessConfig::default(), } } } #[cfg(test)] mod unified_read_pool_tests { use super::*; #[test] fn test_default_uses_yatp() { assert!(!UnifiedReadPoolConfig::default().use_tokio); } #[test] fn test_fairness_max_read_cpu_ratio_is_online_config() { let config = FairnessConfig::default(); assert_eq!(config.max_read_cpu_ratio, 0.0); let parsed: FairnessConfig = toml::from_str("max-read-cpu-ratio = 0.3").unwrap(); assert_eq!(parsed.max_read_cpu_ratio, 0.3); let mut changed = config; changed.max_read_cpu_ratio = 0.3; assert_eq!( config.diff(&changed).get("max_read_cpu_ratio"), Some(&ConfigValue::F64(0.3)) ); } #[test] fn test_validate() { let cpu_cores_quota = SysQuota::cpu_cores_quota(); let limit = cmp::max(UNIFIED_READPOOL_MIN_CONCURRENCY, cpu_cores_quota as usize); let cfg = UnifiedReadPoolConfig { min_thread_count: 1, max_thread_count: 2, stack_size: ReadableSize::mb(2), max_tasks_per_worker: 2000, auto_adjust_pool_size: false, use_tokio: true, enable_yatp_keyspace_metrics: false, keyspace_fairness: FairnessConfig::default(), }; cfg.validate().unwrap(); let cfg = UnifiedReadPoolConfig { min_thread_count: 1, max_thread_count: limit, ..cfg }; cfg.validate().unwrap(); let invalid_cfg = UnifiedReadPoolConfig { min_thread_count: 0, ..cfg }; invalid_cfg.validate().unwrap_err(); let invalid_cfg = UnifiedReadPoolConfig { min_thread_count: 2, max_thread_count: 1, ..cfg }; invalid_cfg.validate().unwrap_err(); let invalid_cfg = UnifiedReadPoolConfig { stack_size: ReadableSize::mb(1), ..cfg }; invalid_cfg.validate().unwrap_err(); let invalid_cfg = UnifiedReadPoolConfig { max_tasks_per_worker: 1, ..cfg }; invalid_cfg.validate().unwrap_err(); let mut oversized_cfg = UnifiedReadPoolConfig { min_thread_count: 1, max_thread_count: limit + 1, ..cfg }; oversized_cfg.validate().unwrap_err(); oversized_cfg.adjust_cpu_limited_worker_count(cpu_cores_quota); oversized_cfg.validate().unwrap(); assert_eq!(oversized_cfg.max_thread_count, limit); let invalid_cfg = UnifiedReadPoolConfig { min_thread_count: limit + 1, max_thread_count: limit + 2, ..cfg }; invalid_cfg.validate().unwrap_err(); let invalid_cfg = UnifiedReadPoolConfig { use_tokio: true, keyspace_fairness: FairnessConfig { enabled: true, ..Default::default() }, ..cfg }; invalid_cfg.validate().unwrap_err(); let cfg = UnifiedReadPoolConfig { use_tokio: false, keyspace_fairness: FairnessConfig { enabled: true, ..Default::default() }, ..cfg }; cfg.validate().unwrap(); let cfg = UnifiedReadPoolConfig { keyspace_fairness: FairnessConfig { slice_precharge_us: MAX_KEYSPACE_FAIRNESS_SLICE_PRECHARGE_US, ..cfg.keyspace_fairness }, ..cfg }; cfg.validate().unwrap(); let invalid_cfg = UnifiedReadPoolConfig { keyspace_fairness: FairnessConfig { slice_precharge_us: MAX_KEYSPACE_FAIRNESS_SLICE_PRECHARGE_US + 1, ..cfg.keyspace_fairness }, ..cfg }; invalid_cfg.validate().unwrap_err(); let invalid_cfg = UnifiedReadPoolConfig { keyspace_fairness: FairnessConfig { max_tasks_per_keyspace: 1, ..cfg.keyspace_fairness }, ..cfg }; invalid_cfg.validate().unwrap_err(); for max_read_cpu_ratio in [0.0, MIN_KEYSPACE_FAIRNESS_MAX_READ_CPU_RATIO, 0.999] { let valid_cfg = UnifiedReadPoolConfig { keyspace_fairness: FairnessConfig { max_read_cpu_ratio, ..cfg.keyspace_fairness }, ..cfg }; valid_cfg.validate().unwrap(); } for max_read_cpu_ratio in [-0.1, 0.001, 1.0] { let invalid_cfg = UnifiedReadPoolConfig { keyspace_fairness: FairnessConfig { max_read_cpu_ratio, ..cfg.keyspace_fairness }, ..cfg }; invalid_cfg.validate().unwrap_err(); } for max_time_slice_us in [ MIN_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US, MAX_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US, ] { let cfg = UnifiedReadPoolConfig { keyspace_fairness: FairnessConfig { max_time_slice_us, ..cfg.keyspace_fairness }, ..cfg }; cfg.validate().unwrap(); } for max_time_slice_us in [ 0, MIN_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US - 1, MAX_KEYSPACE_FAIRNESS_MAX_TIME_SLICE_US + 1, ] { let invalid_cfg = UnifiedReadPoolConfig { keyspace_fairness: FairnessConfig { max_time_slice_us, ..cfg.keyspace_fairness }, ..cfg }; invalid_cfg.validate().unwrap_err(); } } } macro_rules! readpool_config { ($struct_name:ident, $test_mod_name:ident, $display_name:expr) => { #[derive(Clone, Copy, Serialize, Deserialize, PartialEq, Debug)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct $struct_name { pub use_unified_pool: Option, pub high_concurrency: usize, pub normal_concurrency: usize, pub low_concurrency: usize, pub max_tasks_per_worker_high: usize, pub max_tasks_per_worker_normal: usize, pub max_tasks_per_worker_low: usize, pub stack_size: ReadableSize, } impl $struct_name { /// Builds configurations for low, normal and high priority pools. pub fn to_yatp_pool_configs(self) -> Vec { vec![ yatp_pool::Config { workers: self.low_concurrency, max_tasks_per_worker: self.max_tasks_per_worker_low, stack_size: self.stack_size.0 as usize, }, yatp_pool::Config { workers: self.normal_concurrency, max_tasks_per_worker: self.max_tasks_per_worker_normal, stack_size: self.stack_size.0 as usize, }, yatp_pool::Config { workers: self.high_concurrency, max_tasks_per_worker: self.max_tasks_per_worker_high, stack_size: self.stack_size.0 as usize, }, ] } pub fn default_for_test() -> Self { Self { use_unified_pool: None, high_concurrency: 2, normal_concurrency: 2, low_concurrency: 2, max_tasks_per_worker_high: 2000, max_tasks_per_worker_normal: 2000, max_tasks_per_worker_low: 2000, stack_size: ReadableSize::mb(1), } } pub fn use_unified_pool(&self) -> bool { // The unified pool is used by default unless the corresponding module has // customized configurations. self.use_unified_pool .unwrap_or_else(|| *self == Default::default()) } pub fn adjust_use_unified_pool(&mut self) { if self.use_unified_pool.is_none() { // The unified pool is used by default unless the corresponding module has customized configurations. if *self == Default::default() { info!("readpool.{}.use-unified-pool is not set, set to true by default", $display_name); self.use_unified_pool = Some(true); } else { info!("readpool.{}.use-unified-pool is not set, set to false because there are other customized configurations", $display_name); self.use_unified_pool = Some(false); } } } pub fn validate(&self) -> Result<(), Box> { if self.use_unified_pool() { return Ok(()); } if self.high_concurrency == 0 { return Err(format!( "readpool.{}.high-concurrency should be > 0", $display_name ) .into()); } if self.normal_concurrency == 0 { return Err(format!( "readpool.{}.normal-concurrency should be > 0", $display_name ) .into()); } if self.low_concurrency == 0 { return Err(format!( "readpool.{}.low-concurrency should be > 0", $display_name ) .into()); } if self.stack_size.0 < ReadableSize::mb(MIN_READPOOL_STACK_SIZE_MB).0 { return Err(format!( "readpool.{}.stack-size should be >= {}mb", $display_name, MIN_READPOOL_STACK_SIZE_MB ) .into()); } if self.max_tasks_per_worker_high <= 1 { return Err(format!( "readpool.{}.max-tasks-per-worker-high should be > 1", $display_name ) .into()); } if self.max_tasks_per_worker_normal <= 1 { return Err(format!( "readpool.{}.max-tasks-per-worker-normal should be > 1", $display_name ) .into()); } if self.max_tasks_per_worker_low <= 1 { return Err(format!( "readpool.{}.max-tasks-per-worker-low should be > 1", $display_name ) .into()); } Ok(()) } } #[cfg(test)] mod $test_mod_name { use super::*; #[test] fn test_validate() { let cfg = $struct_name::default(); cfg.validate().unwrap(); let mut invalid_cfg = cfg.clone(); invalid_cfg.high_concurrency = 0; assert!(invalid_cfg.validate().is_err()); let mut invalid_cfg = cfg.clone(); invalid_cfg.normal_concurrency = 0; assert!(invalid_cfg.validate().is_err()); let mut invalid_cfg = cfg.clone(); invalid_cfg.low_concurrency = 0; assert!(invalid_cfg.validate().is_err()); let mut invalid_cfg = cfg.clone(); invalid_cfg.stack_size = ReadableSize::mb(1); assert!(invalid_cfg.validate().is_err()); let mut invalid_cfg = cfg.clone(); invalid_cfg.max_tasks_per_worker_high = 0; assert!(invalid_cfg.validate().is_err()); invalid_cfg.max_tasks_per_worker_high = 1; assert!(invalid_cfg.validate().is_err()); invalid_cfg.max_tasks_per_worker_high = 100; invalid_cfg.validate().unwrap(); let mut invalid_cfg = cfg.clone(); invalid_cfg.max_tasks_per_worker_normal = 0; assert!(invalid_cfg.validate().is_err()); invalid_cfg.max_tasks_per_worker_normal = 1; assert!(invalid_cfg.validate().is_err()); invalid_cfg.max_tasks_per_worker_normal = 100; invalid_cfg.validate().unwrap(); let mut invalid_cfg = cfg.clone(); invalid_cfg.max_tasks_per_worker_low = 0; assert!(invalid_cfg.validate().is_err()); invalid_cfg.max_tasks_per_worker_low = 1; assert!(invalid_cfg.validate().is_err()); invalid_cfg.max_tasks_per_worker_low = 100; invalid_cfg.validate().unwrap(); let mut invalid_but_unified = cfg.clone(); invalid_but_unified.use_unified_pool = Some(true); invalid_but_unified.low_concurrency = 0; invalid_but_unified.validate().unwrap(); } } }; } const DEFAULT_STORAGE_READPOOL_MIN_CONCURRENCY: usize = 4; const DEFAULT_STORAGE_READPOOL_MAX_CONCURRENCY: usize = 8; // Assume a request can be finished in 1ms, a request at position x will wait // about 0.001 * x secs to be actual started. A server-is-busy error will // trigger 2 seconds backoff. So when it needs to wait for more than 2 seconds, // return error won't causse larger latency. // const DEFAULT_READPOOL_MAX_TASKS_PER_WORKER: usize = 2 * 1000; // Use double pool size as it's used for both read & write. // TODO: Use 4 * 1000 const DEFAULT_READPOOL_MAX_TASKS_PER_WORKER: usize = 4 * 100000; const MIN_READPOOL_STACK_SIZE_MB: u64 = 2; const DEFAULT_READPOOL_STACK_SIZE_MB: u64 = 10; readpool_config!(StorageReadPoolConfig, storage_read_pool_test, "storage"); impl Default for StorageReadPoolConfig { fn default() -> Self { let cpu_num = SysQuota::cpu_cores_quota(); let mut concurrency = (cpu_num * 0.5) as usize; concurrency = cmp::max(DEFAULT_STORAGE_READPOOL_MIN_CONCURRENCY, concurrency); concurrency = cmp::min(DEFAULT_STORAGE_READPOOL_MAX_CONCURRENCY, concurrency); Self { use_unified_pool: None, high_concurrency: concurrency, normal_concurrency: concurrency, low_concurrency: concurrency, max_tasks_per_worker_high: DEFAULT_READPOOL_MAX_TASKS_PER_WORKER, max_tasks_per_worker_normal: DEFAULT_READPOOL_MAX_TASKS_PER_WORKER, max_tasks_per_worker_low: DEFAULT_READPOOL_MAX_TASKS_PER_WORKER, stack_size: ReadableSize::mb(DEFAULT_READPOOL_STACK_SIZE_MB), } } } const DEFAULT_COPROCESSOR_READPOOL_MIN_CONCURRENCY: usize = 2; readpool_config!( CoprReadPoolConfig, coprocessor_read_pool_test, "coprocessor" ); impl Default for CoprReadPoolConfig { fn default() -> Self { let cpu_num = SysQuota::cpu_cores_quota(); let mut concurrency = (cpu_num * 0.8) as usize; concurrency = cmp::max(DEFAULT_COPROCESSOR_READPOOL_MIN_CONCURRENCY, concurrency); Self { use_unified_pool: None, high_concurrency: concurrency, normal_concurrency: concurrency, low_concurrency: concurrency, max_tasks_per_worker_high: DEFAULT_READPOOL_MAX_TASKS_PER_WORKER, max_tasks_per_worker_normal: DEFAULT_READPOOL_MAX_TASKS_PER_WORKER, max_tasks_per_worker_low: DEFAULT_READPOOL_MAX_TASKS_PER_WORKER, stack_size: ReadableSize::mb(DEFAULT_READPOOL_STACK_SIZE_MB), } } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] #[derive(Default)] pub struct ReadPoolConfig { #[online_config(submodule)] pub unified: UnifiedReadPoolConfig, #[online_config(skip)] pub storage: StorageReadPoolConfig, #[online_config(skip)] pub coprocessor: CoprReadPoolConfig, } impl ReadPoolConfig { pub fn is_unified_pool_enabled(&self) -> bool { self.storage.use_unified_pool() || self.coprocessor.use_unified_pool() } pub fn adjust_use_unified_pool(&mut self) { self.storage.adjust_use_unified_pool(); self.coprocessor.adjust_use_unified_pool(); } fn adjust_cpu_limited_worker_counts(&mut self, cpu_cores_quota: f64) { self.unified .adjust_cpu_limited_worker_count(cpu_cores_quota); } pub fn validate(&self) -> Result<(), Box> { self.unified.validate()?; self.storage.validate()?; self.coprocessor.validate()?; Ok(()) } } #[cfg(test)] mod readpool_tests { use super::*; #[test] fn test_unified_disabled() { // The cloud_server binary always builds the unified read pool, so the // unified config must be valid even when legacy gates are disabled. let unified = UnifiedReadPoolConfig { min_thread_count: 0, max_thread_count: 0, stack_size: ReadableSize::mb(0), max_tasks_per_worker: 0, auto_adjust_pool_size: false, use_tokio: true, enable_yatp_keyspace_metrics: false, keyspace_fairness: FairnessConfig::default(), }; unified.validate().unwrap_err(); let storage = StorageReadPoolConfig { use_unified_pool: Some(false), ..Default::default() }; storage.validate().unwrap(); let coprocessor = CoprReadPoolConfig { use_unified_pool: Some(false), ..Default::default() }; coprocessor.validate().unwrap(); let cfg = ReadPoolConfig { unified, storage, coprocessor, }; assert!(!cfg.is_unified_pool_enabled()); cfg.validate().unwrap_err(); // Storage and coprocessor config must be valid when yatp is not used. let unified = UnifiedReadPoolConfig::default(); unified.validate().unwrap(); let storage = StorageReadPoolConfig { use_unified_pool: Some(false), high_concurrency: 0, ..Default::default() }; storage.validate().unwrap_err(); let coprocessor = CoprReadPoolConfig { use_unified_pool: Some(false), ..Default::default() }; let invalid_cfg = ReadPoolConfig { unified, storage, coprocessor, }; assert!(!invalid_cfg.is_unified_pool_enabled()); invalid_cfg.validate().unwrap_err(); } #[test] fn test_keyspace_fairness_validated_when_unified_gate_is_disabled() { let cfg = ReadPoolConfig { unified: UnifiedReadPoolConfig { use_tokio: false, keyspace_fairness: FairnessConfig { enabled: true, slice_precharge_us: MAX_KEYSPACE_FAIRNESS_SLICE_PRECHARGE_US + 1, ..Default::default() }, ..Default::default() }, storage: StorageReadPoolConfig { use_unified_pool: Some(false), ..Default::default() }, coprocessor: CoprReadPoolConfig { use_unified_pool: Some(false), ..Default::default() }, }; assert!(!cfg.is_unified_pool_enabled()); let err = cfg.validate().unwrap_err().to_string(); assert!(err.contains("slice-precharge-us")); } #[test] fn test_unified_enabled() { // Yatp config must be valid when yatp is used. let unified = UnifiedReadPoolConfig { min_thread_count: 0, max_thread_count: 0, ..Default::default() }; unified.validate().unwrap_err(); let storage = StorageReadPoolConfig { use_unified_pool: Some(true), ..Default::default() }; storage.validate().unwrap(); let coprocessor = CoprReadPoolConfig::default(); coprocessor.validate().unwrap(); let mut cfg = ReadPoolConfig { unified, storage, coprocessor, }; cfg.adjust_use_unified_pool(); assert!(cfg.is_unified_pool_enabled()); cfg.validate().unwrap_err(); } #[test] fn test_is_unified() { let storage = StorageReadPoolConfig { use_unified_pool: Some(false), ..Default::default() }; assert!(!storage.use_unified_pool()); let coprocessor = CoprReadPoolConfig::default(); assert!(coprocessor.use_unified_pool()); let mut cfg = ReadPoolConfig { storage, coprocessor, ..Default::default() }; assert!(cfg.is_unified_pool_enabled()); cfg.storage.use_unified_pool = Some(false); cfg.coprocessor.use_unified_pool = Some(false); assert!(!cfg.is_unified_pool_enabled()); } #[test] fn test_partially_unified() { let storage = StorageReadPoolConfig { use_unified_pool: Some(false), low_concurrency: 0, ..Default::default() }; assert!(!storage.use_unified_pool()); let coprocessor = CoprReadPoolConfig { use_unified_pool: Some(true), ..Default::default() }; assert!(coprocessor.use_unified_pool()); let mut cfg = ReadPoolConfig { storage, coprocessor, ..Default::default() }; assert!(cfg.is_unified_pool_enabled()); cfg.validate().unwrap_err(); cfg.storage.low_concurrency = 1; cfg.validate().unwrap(); let storage = StorageReadPoolConfig { use_unified_pool: Some(true), ..Default::default() }; assert!(storage.use_unified_pool()); let coprocessor = CoprReadPoolConfig { use_unified_pool: Some(false), low_concurrency: 0, ..Default::default() }; assert!(!coprocessor.use_unified_pool()); let mut cfg = ReadPoolConfig { storage, coprocessor, ..Default::default() }; assert!(cfg.is_unified_pool_enabled()); cfg.validate().unwrap_err(); cfg.coprocessor.low_concurrency = 1; cfg.validate().unwrap(); } } #[derive(Clone, Default, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct HadoopConfig { pub home: String, pub linux_user: String, } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct BackupConfig { pub num_threads: usize, pub batch_size: usize, pub sst_max_size: ReadableSize, pub enable_auto_tune: bool, pub auto_tune_remain_threads: usize, pub auto_tune_refresh_interval: ReadableDuration, pub io_thread_size: usize, // Do not expose this config to user. // It used to debug s3 503 error. pub s3_multi_part_size: ReadableSize, #[online_config(submodule)] pub hadoop: HadoopConfig, } impl BackupConfig { pub fn validate(&mut self) -> Result<(), Box> { let limit = SysQuota::cpu_cores_quota() as usize; let default_cfg = BackupConfig::default(); if self.num_threads == 0 || self.num_threads > limit { warn!( "backup.num_threads cannot be 0 or larger than {}, change it to {}", limit, default_cfg.num_threads ); self.num_threads = default_cfg.num_threads; } if self.batch_size == 0 { warn!( "backup.batch_size cannot be 0, change it to {}", default_cfg.batch_size ); self.batch_size = default_cfg.batch_size; } if self.s3_multi_part_size.0 > ReadableSize::gb(5).0 { warn!( "backup.s3_multi_part_size cannot larger than 5GB, change it to {:?}", default_cfg.s3_multi_part_size ); self.s3_multi_part_size = default_cfg.s3_multi_part_size; } Ok(()) } } impl Default for BackupConfig { fn default() -> Self { let cpu_num = SysQuota::cpu_cores_quota(); Self { // use at most 50% of vCPU by default num_threads: (cpu_num * 0.5).clamp(1.0, 8.0) as usize, batch_size: 8, sst_max_size: ReadableSize::mb(16), enable_auto_tune: true, auto_tune_remain_threads: (cpu_num * 0.2).round() as usize, auto_tune_refresh_interval: ReadableDuration::secs(60), io_thread_size: 2, // 5MB is the minimum part size that S3 allowed. s3_multi_part_size: ReadableSize::mb(5), hadoop: Default::default(), } } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct BackupStreamConfig { #[online_config(skip)] pub max_flush_interval: ReadableDuration, #[online_config(skip)] pub num_threads: usize, #[online_config(skip)] pub enable: bool, #[online_config(skip)] pub temp_path: String, #[online_config(skip)] pub file_size_limit: ReadableSize, #[online_config(skip)] pub initial_scan_pending_memory_quota: ReadableSize, #[online_config(skip)] pub initial_scan_rate_limit: ReadableSize, } impl BackupStreamConfig { pub fn validate(&mut self) -> Result<(), Box> { let limit = SysQuota::cpu_cores_quota() as usize; let default_cfg = BackupStreamConfig::default(); if self.num_threads == 0 || self.num_threads > limit { warn!( "log_backup.num_threads cannot be 0 or larger than {}, change it to {}", limit, default_cfg.num_threads ); self.num_threads = default_cfg.num_threads; } Ok(()) } } impl Default for BackupStreamConfig { fn default() -> Self { let cpu_num = SysQuota::cpu_cores_quota(); let total_mem = SysQuota::memory_limit_in_bytes(); let quota_size = (total_mem as f64 * 0.1).min(ReadableSize::mb(512).0 as _); Self { max_flush_interval: ReadableDuration::minutes(3), // use at most 50% of vCPU by default num_threads: (cpu_num * 0.5).clamp(2.0, 12.0) as usize, enable: true, // TODO: may be use raft store directory temp_path: String::new(), file_size_limit: ReadableSize::mb(256), initial_scan_pending_memory_quota: ReadableSize(quota_size as _), initial_scan_rate_limit: ReadableSize::mb(60), } } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct CdcConfig { #[online_config(skip)] pub enable: bool, pub min_ts_interval: ReadableDuration, // TODO: `hibernate_regions_compatible` is not verified currently, so // we do not support change its value. #[online_config(skip)] pub hibernate_regions_compatible: bool, // TODO(hi-rustin): Consider resizing the thread pool based on `incremental_scan_threads`. #[online_config(skip)] pub incremental_scan_threads: usize, /// Number of normal slots shared by low- and high-priority incremental /// scans. pub incremental_scan_concurrency: usize, /// Base admission limit for queued and running incremental scans. /// /// High-priority scans may use the additional admission capacity provided /// by `emergency_high_slots`. pub incremental_scan_concurrency_limit: usize, /// Additional slots reserved for high-priority scans after normal slots are /// full. /// /// The scheduler gives high-priority scans strict priority for normal /// slots. Its maximum running concurrency is /// `incremental_scan_concurrency` plus `emergency_high_slots`. pub emergency_high_slots: usize, /// Limit scan speed based on disk I/O traffic. pub incremental_scan_speed_limit: ReadableSize, /// Limit scan speed based on memory accesing traffic. #[doc(hidden)] pub incremental_fetch_speed_limit: ReadableSize, /// `TsFilter` can increase speed and decrease resource usage when /// incremental content is much less than total content. However in /// other cases, `TsFilter` can make performance worse because it needs /// to re-fetch old row values if they are required. /// /// `TsFilter` will be enabled if `incremental/total <= /// incremental_scan_ts_filter_ratio`. /// Set `incremental_scan_ts_filter_ratio` to 0 will disable it. pub incremental_scan_ts_filter_ratio: f64, /// Count of threads to confirm Region leadership in TiKV instances, 1 by /// default. Please consider to increase it if count of regions on one /// TiKV instance is greater than 20k. #[online_config(skip)] pub tso_worker_threads: usize, pub sink_memory_quota: ReadableSize, pub old_value_cache_memory_quota: ReadableSize, } impl Default for CdcConfig { #[allow(deprecated)] fn default() -> Self { Self { enable: false, min_ts_interval: ReadableDuration::secs(1), // TODO: currently, we use resolve_by_raft by default, // we may switch to check_leader if it's performance is better. hibernate_regions_compatible: false, // 4 threads for incremental scan. incremental_scan_threads: 4, // At most 6 concurrent running tasks. incremental_scan_concurrency: 6, // At most 10000 tasks can exist simultaneously. incremental_scan_concurrency_limit: 10000, emergency_high_slots: 1, // TiCDC requires a SSD, the typical write speed of SSD // is more than 500MB/s, so 128MB/s is enough. incremental_scan_speed_limit: ReadableSize::mb(128), incremental_fetch_speed_limit: ReadableSize::mb(512), incremental_scan_ts_filter_ratio: 0.2, tso_worker_threads: 1, // 512MB memory for CDC sink. sink_memory_quota: ReadableSize::mb(512), // 512MB memory for old value cache. old_value_cache_memory_quota: ReadableSize::mb(512), } } } impl CdcConfig { pub fn validate(&mut self) -> Result<(), Box> { let default_cfg = CdcConfig::default(); if self.min_ts_interval.is_zero() { warn!( "cdc.min-ts-interval can't be 0, change it to {}", default_cfg.min_ts_interval ); self.min_ts_interval = default_cfg.min_ts_interval; } if self.incremental_scan_threads == 0 { warn!( "cdc.incremental-scan-threads can't be 0, change it to {}", default_cfg.incremental_scan_threads ); self.incremental_scan_threads = default_cfg.incremental_scan_threads; } if self.incremental_scan_concurrency < self.incremental_scan_threads { warn!( "cdc.incremental-scan-concurrency must be larger than cdc.incremental-scan-threads, change it to {}", self.incremental_scan_threads ); self.incremental_scan_concurrency = self.incremental_scan_threads } if self.incremental_scan_concurrency_limit < self.incremental_scan_concurrency { warn!( "cdc.incremental-scan-concurrency-limit must be larger than cdc.incremental-scan-concurrency, change it to {}", self.incremental_scan_concurrency ); self.incremental_scan_concurrency_limit = self.incremental_scan_concurrency } if self.incremental_scan_ts_filter_ratio < 0.0 || self.incremental_scan_ts_filter_ratio > 1.0 { warn!( "cdc.incremental-scan-ts-filter-ratio should be larger than 0 and less than 1, change it to {}", default_cfg.incremental_scan_ts_filter_ratio ); self.incremental_scan_ts_filter_ratio = default_cfg.incremental_scan_ts_filter_ratio; } // TODO: remove this force disable once we carefully test the CheckLeader API. if self.hibernate_regions_compatible { warn!("feature 'hibernate_regions_compatible' is not available."); self.hibernate_regions_compatible = false; } Ok(()) } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct ResolvedTsConfig { #[online_config(skip)] pub enable: bool, pub advance_ts_interval: ReadableDuration, #[online_config(skip)] pub scan_lock_pool_size: usize, pub memory_quota: ReadableSize, pub incremental_scan_concurrency: usize, } impl ResolvedTsConfig { fn validate(&self) -> Result<(), Box> { if self.advance_ts_interval.is_zero() { return Err("resolved-ts.advance-ts-interval can't be zero".into()); } if self.scan_lock_pool_size == 0 { return Err("resolved-ts.scan-lock-pool-size can't be zero".into()); } Ok(()) } } impl Default for ResolvedTsConfig { fn default() -> Self { Self { enable: true, advance_ts_interval: ReadableDuration::secs(1), scan_lock_pool_size: 2, memory_quota: ReadableSize::mb(256), incremental_scan_concurrency: 6, } } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct File { pub filename: String, // The unit is MB pub max_size: u64, // The unit is Day pub max_days: u64, pub max_backups: usize, } impl Default for File { fn default() -> Self { Self { filename: "".to_owned(), max_size: 300, max_days: 0, max_backups: 0, } } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct LogConfig { pub level: LogLevel, pub txn_info_logging: bool, #[online_config(skip)] pub format: LogFormat, #[online_config(skip)] pub enable_timestamp: bool, #[online_config(skip)] pub file: File, } /// LogLevel is a wrapper type of `slog::Level` #[derive(Copy, Clone, Debug, PartialEq)] pub struct LogLevel(slog::Level); impl From for slog::Level { fn from(l: LogLevel) -> Self { l.0 } } impl From for LogLevel { fn from(l: slog::Level) -> Self { Self(l) } } impl Serialize for LogLevel { fn serialize(&self, serializer: S) -> Result where S: Serializer, { get_string_by_level(self.0).serialize(serializer) } } impl<'de> Deserialize<'de> for LogLevel { fn deserialize(deserializer: D) -> Result where D: Deserializer<'de>, { let string = String::deserialize(deserializer)?; get_level_by_string(&string) .map(LogLevel) .ok_or_else(|| D::Error::invalid_value(Unexpected::Str(&string), &"a valid log level")) } } impl From for ConfigValue { fn from(l: LogLevel) -> Self { Self::String(get_string_by_level(l.0).into()) } } impl TryFrom for LogLevel { type Error = String; fn try_from(value: ConfigValue) -> Result { if let ConfigValue::String(s) = value { get_level_by_string(&s) .map(LogLevel) .ok_or_else(|| format!("invalid log level: '{}'", s)) } else { panic!("expect ConfigValue::String, found: {:?}", value) } } } impl TryFrom<&ConfigValue> for LogLevel { type Error = String; fn try_from(value: &ConfigValue) -> Result { Self::try_from(value.clone()) } } impl Default for LogConfig { fn default() -> Self { Self { level: LogLevel(slog::Level::Info), txn_info_logging: true, format: LogFormat::Text, enable_timestamp: true, file: File::default(), } } } impl LogConfig { fn validate(&self) -> Result<(), Box> { if self.file.max_size > 4096 { return Err("Max log file size upper limit to 4096MB".to_string().into()); } Ok(()) } } pub struct LogConfigManager; impl ConfigManager for LogConfigManager { fn dispatch(&mut self, changes: ConfigChange) -> CfgResult<()> { if let Some(v) = changes.get("level") { let log_level = LogLevel::try_from(v)?; set_log_level(log_level.0); } if let Some(ConfigValue::Bool(enable)) = changes.get("txn_info_logging") { set_txn_info_logging(*enable); } info!("update log config"; "config" => ?changes); Ok(()) } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct MemoryConfig { // Whether enables the heap profiling which may have a bit performance overhead about 2% for // the default sample rate. pub enable_heap_profiling: bool, // Average interval between allocation samples, as measured in bytes of allocation activity. // Increasing the sampling interval decreases profile fidelity, but also decreases the // computational overhead. // The default sample interval is 512 KB. It only accepts power of two, otherwise it will be // rounded up to the next power of two. pub profiling_sample_per_bytes: ReadableSize, // Whether allocates the exclusive arena for threads. // When disabled, the metric of memory usage for each thread would be unavailable. #[online_config(skip)] pub enable_thread_exclusive_arena: bool, } impl Default for MemoryConfig { fn default() -> Self { Self { enable_heap_profiling: true, profiling_sample_per_bytes: ReadableSize::kb(512), enable_thread_exclusive_arena: true, } } } impl MemoryConfig { pub fn init(&self) { if self.enable_heap_profiling { if let Err(e) = tikv_alloc::activate_prof() { error!("failed to enable heap profiling"; "err" => ?e); return; } if let Err(e) = tikv_alloc::set_prof_sample(self.profiling_sample_per_bytes.0) { error!("failed to set heap profiling sample"; "err" => ?e); return; } info!( "heap profiling is enabled"; "sample_per_bytes" => ?self.profiling_sample_per_bytes, ); } else { info!("heap profiling is not enabled"); } tikv_alloc::set_thread_exclusive_arena(self.enable_thread_exclusive_arena); } } pub struct MemoryConfigManager; impl ConfigManager for MemoryConfigManager { fn dispatch(&mut self, changes: ConfigChange) -> CfgResult<()> { if let Some(ConfigValue::Bool(enable)) = changes.get("enable_heap_profiling") { if *enable { tikv_alloc::activate_prof()?; } else { tikv_alloc::deactivate_prof()?; } } if let Some(ConfigValue::Size(sample_rate)) = changes.get("profiling_sample_per_bytes") { tikv_alloc::set_prof_sample(*sample_rate)?; } info!("update memory config"; "config" => ?changes); Ok(()) } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct QuotaConfig { pub foreground_cpu_time: usize, pub foreground_write_bandwidth: ReadableSize, pub foreground_read_bandwidth: ReadableSize, pub max_delay_duration: ReadableDuration, pub background_cpu_time: usize, pub background_write_bandwidth: ReadableSize, pub background_read_bandwidth: ReadableSize, pub enable_auto_tune: bool, } impl Default for QuotaConfig { fn default() -> Self { Self { foreground_cpu_time: 0, foreground_write_bandwidth: ReadableSize(0), foreground_read_bandwidth: ReadableSize(0), max_delay_duration: ReadableDuration::millis(500), background_cpu_time: 0, background_write_bandwidth: ReadableSize(0), background_read_bandwidth: ReadableSize(0), enable_auto_tune: false, } } } impl QuotaConfig { pub fn validate(&self) -> Result<(), Box> { const MAX_DELAY_DURATION: ReadableDuration = ReadableDuration::micros(u64::MAX / 1000); if self.max_delay_duration > MAX_DELAY_DURATION { return Err(format!("quota.max-delay-duration must <= {}", MAX_DELAY_DURATION).into()); } Ok(()) } } #[derive(Clone, Serialize, Deserialize, PartialEq, Debug, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct TikvConfig { #[doc(hidden)] #[serde(skip_serializing)] #[online_config(hidden)] pub cfg_path: String, // Deprecated! These configuration has been moved to LogConfig. // They are preserved for compatibility check. #[doc(hidden)] #[online_config(skip)] pub log_level: LogLevel, #[doc(hidden)] #[online_config(skip)] pub log_file: String, #[doc(hidden)] #[online_config(skip)] pub log_format: LogFormat, #[online_config(skip)] pub log_rotation_timespan: ReadableDuration, #[doc(hidden)] #[online_config(skip)] pub log_rotation_size: ReadableSize, #[online_config(skip)] pub slow_log_file: String, #[online_config(skip)] pub slow_log_threshold: ReadableDuration, #[online_config(hidden)] pub panic_when_unexpected_key_or_data: bool, #[doc(hidden)] #[serde(skip_serializing)] #[online_config(skip)] pub enable_io_snoop: bool, #[online_config(skip)] pub abort_on_panic: bool, #[doc(hidden)] #[online_config(skip)] pub memory_usage_limit: Option, #[doc(hidden)] #[online_config(skip)] pub memory_usage_high_water: f64, #[online_config(skip)] pub black_list_path: String, #[doc(hidden)] #[online_config(skip)] pub recovery_mode: bool, #[online_config(submodule)] pub log: LogConfig, #[online_config(submodule)] pub memory: MemoryConfig, #[online_config(submodule)] pub quota: QuotaConfig, #[online_config(submodule)] pub readpool: ReadPoolConfig, #[online_config(submodule)] pub server: ServerConfig, #[online_config(submodule)] pub storage: StorageConfig, #[online_config(skip)] pub pd: PdConfig, #[online_config(hidden)] pub metric: MetricConfig, #[online_config(submodule)] #[serde(rename = "raftstore")] pub raft_store: RfStoreConfig, #[online_config(submodule)] pub coprocessor: CopConfig, // Deprecated for nextgen. #[online_config(submodule)] pub rocksdb: DbConfig, #[online_config(submodule)] pub raftdb: RaftDbConfig, #[online_config(submodule)] pub rfengine: RfEngineConfig, #[online_config(submodule)] pub kvengine: KvEngineConfig, #[online_config(skip)] pub security: SecurityConfig, #[online_config(skip)] pub import: ImportConfig, #[online_config(submodule)] pub backup: BackupConfig, // Deprecated for next-gen #[online_config(skip)] // The term "log backup" and "backup stream" are identity. // The "log backup" should be the only product name exposed to the user. #[serde(rename = "log-backup")] pub backup_stream: BackupStreamConfig, #[online_config(submodule)] pub pessimistic_txn: PessimisticTxnConfig, // Deprecated for next-gen. // NOTE: `enable_safe_point_v2` and `disable_safe_point_fallback_v1` // has already been deprecated in kvengine. #[online_config(submodule)] pub gc: GcConfig, // TODO: load-based split is not supported in next-gen yet. #[online_config(submodule)] pub split: SplitConfig, #[online_config(submodule)] pub cdc: CdcConfig, // TODO: resolved_ts is not supported in next-gen. #[online_config(submodule)] pub resolved_ts: ResolvedTsConfig, #[online_config(submodule)] pub resource_metering: ResourceMeteringConfig, #[online_config(skip)] pub dfs: DFSConfig, #[online_config(submodule)] pub overload: OverloadConfig, #[online_config(submodule)] pub resource_control: ResourceControlConfig, #[online_config(skip)] pub memory_profile_auto_dump: MemoryProfileAutoDumpConfig, } impl Default for TikvConfig { fn default() -> TikvConfig { TikvConfig { cfg_path: "".to_owned(), log_level: slog::Level::Info.into(), log_file: "".to_owned(), log_format: LogFormat::Text, log_rotation_timespan: ReadableDuration::hours(0), log_rotation_size: ReadableSize::mb(300), slow_log_file: "".to_owned(), slow_log_threshold: ReadableDuration::secs(1), panic_when_unexpected_key_or_data: false, enable_io_snoop: true, abort_on_panic: false, memory_usage_limit: None, memory_usage_high_water: 0.9, black_list_path: "".to_owned(), recovery_mode: false, log: LogConfig::default(), memory: MemoryConfig::default(), quota: QuotaConfig::default(), readpool: ReadPoolConfig::default(), server: ServerConfig::default(), metric: MetricConfig::default(), raft_store: RfStoreConfig::default(), coprocessor: CopConfig::default(), pd: PdConfig::default(), rocksdb: DbConfig::default(), raftdb: RaftDbConfig::default(), rfengine: RfEngineConfig::default(), kvengine: KvEngineConfig::default(), storage: StorageConfig::default(), security: SecurityConfig::default(), import: ImportConfig::default(), backup: BackupConfig::default(), pessimistic_txn: PessimisticTxnConfig::default(), gc: GcConfig::default(), split: SplitConfig::default(), cdc: CdcConfig::default(), resolved_ts: ResolvedTsConfig::default(), resource_metering: ResourceMeteringConfig::default(), backup_stream: BackupStreamConfig::default(), dfs: DFSConfig::default(), overload: OverloadConfig::default(), resource_control: ResourceControlConfig::default(), memory_profile_auto_dump: MemoryProfileAutoDumpConfig::default(), } } } impl TikvConfig { pub fn infer_raft_db_path(&self, data_dir: Option<&str>) -> Result> { if self.raft_store.raftdb_path.is_empty() { let data_dir = data_dir.unwrap_or(&self.storage.data_dir); config::canonicalize_sub_path(data_dir, "raft") } else { config::canonicalize_path(&self.raft_store.raftdb_path) } } pub fn infer_kv_engine_path(&self, data_dir: Option<&str>) -> Result> { let data_dir = data_dir.unwrap_or(&self.storage.data_dir); config::canonicalize_sub_path(data_dir, DEFAULT_ROCKSDB_SUB_DIR) } /// Adjusts worker counts that are bounded by the current CPU quota. /// /// Call this after applying all config sources and compatibility /// migrations, and before [`TikvConfig::validate`]. pub fn adjust_cpu_limited_worker_counts(&mut self) { let cpu_cores_quota = SysQuota::cpu_cores_quota(); self.readpool .adjust_cpu_limited_worker_counts(cpu_cores_quota); self.storage .adjust_cpu_limited_worker_counts(cpu_cores_quota); self.raft_store .adjust_cpu_limited_worker_counts(cpu_cores_quota); } pub fn validate(&mut self) -> Result<(), Box> { self.log.validate()?; self.readpool.validate()?; self.storage.validate()?; if self.readpool.unified.keyspace_fairness.enabled && !self.storage.use_separated_scheduler_pool { return Err( "keyspace fairness is not supported when scheduler pool is merged with read pool" .into(), ); } if self.cfg_path.is_empty() { self.cfg_path = Path::new(&self.storage.data_dir) .join(LAST_CONFIG_FILE) .to_str() .unwrap() .to_owned(); } self.raft_store.raftdb_path = self.infer_raft_db_path(None)?; self.rfengine.wal_sync_dir = self .rfengine .wal_sync_dir .replace("${data-dir}", &self.storage.data_dir); self.rfengine.wal_secondary_dir = self .rfengine .wal_secondary_dir .replace("${data-dir}", &self.storage.data_dir); if !self.rfengine.wal_secondary_dir.is_empty() && self.rfengine.wal_secondary_dir == self.rfengine.wal_sync_dir { return Err("rfengine.wal_secondary_dir can't be same as rfengine.wal_sync_dir".into()); } if !self.rfengine.enable_compact_rate_limiter { // enable compact rate limiter in alibaba cloud environment if std::env::var(ALIBABA_CLOUD_REGION_ID).is_ok() { self.rfengine.enable_compact_rate_limiter = true; } } let kv_db_path = self.infer_kv_engine_path(None)?; if kv_db_path == self.raft_store.raftdb_path { return Err("raft_store.raftdb_path can't be same as storage.data_dir/db".into()); } let kv_db_wal_path = if self.rocksdb.wal_dir.is_empty() { config::canonicalize_path(&kv_db_path)? } else { config::canonicalize_path(&self.rocksdb.wal_dir)? }; let raft_db_wal_path = if self.raftdb.wal_dir.is_empty() { config::canonicalize_path(&self.raft_store.raftdb_path)? } else { config::canonicalize_path(&self.raftdb.wal_dir)? }; if kv_db_wal_path == raft_db_wal_path { return Err("raftdb.wal_dir can't be same as rocksdb.wal_dir".into()); } // Check blob file dir is empty when titan is disabled if !self.rocksdb.titan.enabled { let titandb_path = if self.rocksdb.titan.dirname.is_empty() { Path::new(&kv_db_path).join("titandb") } else { Path::new(&self.rocksdb.titan.dirname).to_path_buf() }; if let Err(e) = tikv_util::config::check_data_dir_empty(titandb_path.to_str().unwrap(), "blob") { return Err(format!( "check: titandb-data-dir-empty; err: \"{}\"; \ hint: You have disabled titan when its data directory is not empty. \ To properly shutdown titan, please enter fallback blob-run-mode and \ wait till titandb files are all safely ingested.", e ) .into()); } } let expect_keepalive = self.raft_store.raft_heartbeat_interval() * 2; if expect_keepalive > self.server.grpc_keepalive_time.0 { return Err(format!( "grpc_keepalive_time is too small, it should not less than the double of \ raft tick interval (>= {})", duration_to_sec(expect_keepalive) ) .into()); } if self.backup_stream.temp_path.is_empty() { self.backup_stream.temp_path = config::canonicalize_sub_path(&self.storage.data_dir, "log-backup-temp")?; } self.rocksdb.validate()?; self.raftdb.validate()?; self.rfengine.validate(&self.raft_store.raftdb_path)?; self.server.validate()?; self.pd.validate()?; self.coprocessor.validate()?; self.raft_store.validate( self.coprocessor.region_split_size, self.coprocessor.region_split_keys, self.coprocessor.enable_region_bucket, self.coprocessor.region_bucket_size, )?; // Keep disabled load-split settings upgrade-compatible. Cross-field // constraints are validated when the feature is enabled, including // through online config. if self.split.enable_load_based_split { self.split.validate()?; } self.security.validate()?; self.import.validate()?; self.backup.validate()?; self.backup_stream.validate()?; self.cdc.validate()?; self.pessimistic_txn.validate()?; self.gc.validate()?; self.resolved_ts.validate()?; self.resource_metering.validate()?; self.quota.validate()?; self.memory_profile_auto_dump.validate()?; self.kvengine.validate()?; if !self.kvengine.enable_crossbeam_memtable && self.raft_store.raft_entry_max_size > ReadableSize::mb(16) { return Err("raftstore.raft-entry-max-size above 16MiB requires \ kvengine.enable-crossbeam-memtable = true" .into()); } if self.storage.flow_control.enable { self.rocksdb.defaultcf.disable_write_stall = true; self.rocksdb.writecf.disable_write_stall = true; self.rocksdb.lockcf.disable_write_stall = true; self.rocksdb.raftcf.disable_write_stall = true; } // Fill in values for unspecified write stall configurations. macro_rules! fill_cf_opts { ($cf_opts:expr, $cfg:expr) => { if let Some(v) = &mut $cf_opts.level0_slowdown_writes_trigger { if $cfg.enable && *v > $cfg.l0_files_threshold as i32 { warn!( "{}.level0-slowdown-writes-trigger is too large. Setting it to \ storage.flow-control.l0-files-threshold ({})", stringify!($cf_opts), $cfg.l0_files_threshold ); *v = $cfg.l0_files_threshold as i32; } } else { $cf_opts.level0_slowdown_writes_trigger = Some($cfg.l0_files_threshold as i32); } if let Some(v) = &mut $cf_opts.level0_stop_writes_trigger { if $cfg.enable && *v > $cfg.l0_files_threshold as i32 { warn!( "{}.level0-stop-writes-trigger is too large. Setting it to \ storage.flow-control.l0-files-threshold ({})", stringify!($cf_opts), $cfg.l0_files_threshold ); *v = $cfg.l0_files_threshold as i32; } } else { $cf_opts.level0_stop_writes_trigger = Some($cfg.l0_files_threshold as i32); } if let Some(v) = &mut $cf_opts.soft_pending_compaction_bytes_limit { if $cfg.enable && v.0 > $cfg.soft_pending_compaction_bytes_limit.0 { warn!( "{}.soft-pending-compaction-bytes-limit is too large. Setting it to \ storage.flow-control.soft-pending-compaction-bytes-limit ({})", stringify!($cf_opts), $cfg.soft_pending_compaction_bytes_limit.0 ); *v = $cfg.soft_pending_compaction_bytes_limit; } } else { $cf_opts.soft_pending_compaction_bytes_limit = Some($cfg.soft_pending_compaction_bytes_limit); } if let Some(v) = &mut $cf_opts.hard_pending_compaction_bytes_limit { if $cfg.enable && v.0 > $cfg.hard_pending_compaction_bytes_limit.0 { warn!( "{}.hard-pending-compaction-bytes-limit is too large. Setting it to \ storage.flow-control.hard-pending-compaction-bytes-limit ({})", stringify!($cf_opts), $cfg.hard_pending_compaction_bytes_limit.0 ); *v = $cfg.hard_pending_compaction_bytes_limit; } } else { $cf_opts.hard_pending_compaction_bytes_limit = Some($cfg.hard_pending_compaction_bytes_limit); } }; } let flow_control_cfg = if self.storage.flow_control.enable { self.storage.flow_control.clone() } else { crate::storage::config::FlowControlConfig { enable: false, ..Default::default() } }; fill_cf_opts!(self.raftdb.defaultcf, flow_control_cfg); fill_cf_opts!(self.rocksdb.defaultcf, flow_control_cfg); fill_cf_opts!(self.rocksdb.writecf, flow_control_cfg); fill_cf_opts!(self.rocksdb.lockcf, flow_control_cfg); fill_cf_opts!(self.rocksdb.raftcf, flow_control_cfg); let total_mem_quota = SysQuota::memory_limit_in_bytes(); if let Some(memory_usage_limit) = self.memory_usage_limit { if memory_usage_limit.0 > total_mem_quota { // Explicitly exceeds system memory capacity is not allowed. return Err(format!( "memory_usage_limit is greater than system memory capacity {}", total_mem_quota ) .into()); } } else { // Adjust `memory_usage_limit` if necessary. if self.storage.block_cache.shared { if let Some(cap) = self.storage.block_cache.capacity { let limit = (cap.0 as f64 / BLOCK_CACHE_RATE * MEMORY_USAGE_LIMIT_RATE) as u64; self.memory_usage_limit = Some(ReadableSize(limit)); } else { self.memory_usage_limit = Some(Self::suggested_memory_usage_limit()); } } else { let cap = self.rocksdb.defaultcf.block_cache_size.0 + self.rocksdb.writecf.block_cache_size.0 + self.rocksdb.lockcf.block_cache_size.0 + self.raftdb.defaultcf.block_cache_size.0; let limit = (cap as f64 / BLOCK_CACHE_RATE * MEMORY_USAGE_LIMIT_RATE) as u64; self.memory_usage_limit = Some(ReadableSize(limit)); } } let mut limit = self.memory_usage_limit.unwrap(); let total = ReadableSize(SysQuota::memory_limit_in_bytes()); if limit.0 > total.0 { warn!( "memory_usage_limit:{:?} > total:{:?}, fallback to total", limit, total, ); self.memory_usage_limit = Some(total); limit = total; } let default = Self::suggested_memory_usage_limit(); if limit.0 > default.0 { warn!( "memory_usage_limit:{:?} > recommanded:{:?}, maybe page cache isn't enough", limit, default, ); } if self.rocksdb.writecf.write_buffer_size.0 > kvengine::KV_ENGINE_MEM_TABLE_MAX_SIZE { return Err(format!( "rocksdb.writecf.write_buffer_size is greater than {}", kvengine::KV_ENGINE_MEM_TABLE_MAX_SIZE ) .into()); } Ok(()) } // As the init of `logger` is very early, this adjust needs to be separated and // called immediately after parsing the command line. pub fn logger_compatible_adjust(&mut self) { let default_tikv_cfg = TikvConfig::default(); let default_log_cfg = LogConfig::default(); if self.log_level != default_tikv_cfg.log_level { eprintln!("deprecated configuration, log-level has been moved to log.level"); if self.log.level == default_log_cfg.level { eprintln!("override log.level with log-level, {:?}", self.log_level); self.log.level = self.log_level; } self.log_level = default_tikv_cfg.log_level; } if self.log_file != default_tikv_cfg.log_file { eprintln!("deprecated configuration, log-file has been moved to log.file.filename"); if self.log.file.filename == default_log_cfg.file.filename { eprintln!( "override log.file.filename with log-file, {:?}", self.log_file ); self.log.file.filename = self.log_file.clone(); } self.log_file = default_tikv_cfg.log_file; } if self.log_format != default_tikv_cfg.log_format { eprintln!("deprecated configuration, log-format has been moved to log.format"); if self.log.format == default_log_cfg.format { eprintln!("override log.format with log-format, {:?}", self.log_format); self.log.format = self.log_format; } self.log_format = default_tikv_cfg.log_format; } if self.log_rotation_timespan.as_secs() > 0 { eprintln!( "deprecated configuration, log-rotation-timespan is no longer used and ignored." ); } if self.log_rotation_size != default_tikv_cfg.log_rotation_size { eprintln!( "deprecated configuration, \ log-ratation-size has been moved to log.file.max-size" ); if self.log.file.max_size == default_log_cfg.file.max_size { eprintln!( "override log.file.max_size with log-rotation-size, {:?}", self.log_rotation_size ); self.log.file.max_size = self.log_rotation_size.as_mb(); } self.log_rotation_size = default_tikv_cfg.log_rotation_size; } } pub fn compatible_adjust(&mut self) { let default_raft_store = RaftstoreConfig::default(); let default_coprocessor = CopConfig::default(); if self.raft_store.region_max_size != default_raft_store.region_max_size { warn!( "deprecated configuration, \ raftstore.region-max-size has been moved to coprocessor" ); if self.coprocessor.region_max_size == default_coprocessor.region_max_size { warn!( "override coprocessor.region-max-size with raftstore.region-max-size, {:?}", self.raft_store.region_max_size ); self.coprocessor.region_max_size = Some(self.raft_store.region_max_size); } self.raft_store.region_max_size = default_raft_store.region_max_size; } if self.raft_store.region_split_size != default_raft_store.region_split_size { warn!( "deprecated configuration, \ raftstore.region-split-size has been moved to coprocessor", ); if self.coprocessor.region_split_size == default_coprocessor.region_split_size { warn!( "override coprocessor.region-split-size with raftstore.region-split-size, {:?}", self.raft_store.region_split_size ); self.coprocessor.region_split_size = self.raft_store.region_split_size; } self.raft_store.region_split_size = default_raft_store.region_split_size; } if self.raft_store.ia_kv_size_discount != default_raft_store.ia_kv_size_discount { warn!( "deprecated configuration, {} is no longer used and ignored.", "raftstore.ia-kv-size-discount", ); self.raft_store.ia_kv_size_discount = default_raft_store.ia_kv_size_discount; } if self.server.end_point_concurrency.is_some() { warn!( "deprecated configuration, {} has been moved to {}", "server.end-point-concurrency", "readpool.coprocessor.xxx-concurrency", ); warn!( "override {} with {}, {:?}", "readpool.coprocessor.xxx-concurrency", "server.end-point-concurrency", self.server.end_point_concurrency ); let concurrency = self.server.end_point_concurrency.take().unwrap(); self.readpool.coprocessor.high_concurrency = concurrency; self.readpool.coprocessor.normal_concurrency = concurrency; self.readpool.coprocessor.low_concurrency = concurrency; } if self.server.end_point_stack_size.is_some() { warn!( "deprecated configuration, {} has been moved to {}", "server.end-point-stack-size", "readpool.coprocessor.stack-size", ); warn!( "override {} with {}, {:?}", "readpool.coprocessor.stack-size", "server.end-point-stack-size", self.server.end_point_stack_size ); self.readpool.coprocessor.stack_size = self.server.end_point_stack_size.take().unwrap(); } if self.server.end_point_max_tasks.is_some() { warn!( "deprecated configuration, {} is no longer used and ignored, please use {}.", "server.end-point-max-tasks", "readpool.coprocessor.max-tasks-per-worker-xxx", ); // Note: // Our `end_point_max_tasks` is mostly mistakenly configured, so we don't // override new configuration using old values. self.server.end_point_max_tasks = None; } if self.raft_store.clean_stale_peer_delay.as_secs() > 0 { warn!( "deprecated configuration, {} is no longer used and ignored.", "raft_store.clean_stale_peer_delay", ); } if self.rocksdb.auto_tuned.is_some() { warn!( "deprecated configuration, {} is no longer used and ignored, please use {}.", "rocksdb.auto_tuned", "rocksdb.rate_limiter_auto_tuned", ); self.rocksdb.auto_tuned = None; } if self.backup.sst_max_size.0 < default_coprocessor.region_max_size().0 / 10 { warn!( "override backup.sst-max-size with min sst-max-size, {:?}", default_coprocessor.region_max_size() / 10 ); self.backup.sst_max_size = default_coprocessor.region_max_size() / 10; } else if self.backup.sst_max_size.0 > default_coprocessor.region_max_size().0 * 2 { warn!( "override backup.sst-max-size with max sst-max-size, {:?}", default_coprocessor.region_max_size() * 2 ); self.backup.sst_max_size = default_coprocessor.region_max_size() * 2; } self.readpool.adjust_use_unified_pool(); } pub fn check_critical_cfg_with(&self, last_cfg: &Self) -> Result<(), String> { if last_cfg.storage.data_dir != self.storage.data_dir { // In tikv 3.0 the default value of storage.data-dir changed // from "" to "./" let using_default_after_upgrade = last_cfg.storage.data_dir.is_empty() && self.storage.data_dir == DEFAULT_DATA_DIR; if !using_default_after_upgrade { return Err(format!( "storage data dir have been changed, former data dir is {}, \ current data dir is {}, please check if it is expected.", last_cfg.storage.data_dir, self.storage.data_dir )); } } if last_cfg.rocksdb.wal_dir != self.rocksdb.wal_dir { return Err(format!( "db wal dir have been changed, former is '{}', \ current db wal_dir is '{}', please guarantee all data wal logs \ have been moved to destination directory.", last_cfg.rocksdb.wal_dir, self.rocksdb.wal_dir )); } // It's possible that `last_cfg` is not fully validated. let last_raftdb_dir = last_cfg .infer_raft_db_path(None) .map_err(|e| e.to_string())?; // FIXME: We cannot reliably determine the actual value of // `last_cfg.raft_engine.enable`, because some old versions don't have // this field (so it is automatically interpreted as the current // default value). To be safe, we will check both engines regardless // of whether raft engine is enabled. if last_raftdb_dir != self.raft_store.raftdb_path { return Err(format!( "raft db dir have been changed, former is '{}', \ current is '{}', please check if it is expected.", last_raftdb_dir, self.raft_store.raftdb_path )); } if last_cfg.raftdb.wal_dir != self.raftdb.wal_dir { return Err(format!( "raft db wal dir have been changed, former is '{}', \ current is '{}', please check if it is expected.", last_cfg.raftdb.wal_dir, self.raftdb.wal_dir )); } // Check validation of api version change between API V1 and V1ttl only. // Validation between V1/V1ttl and V2 is checked in `Node::check_api_version`. if last_cfg.storage.api_version == 1 && self.storage.api_version == 1 { if last_cfg.storage.enable_ttl && !self.storage.enable_ttl { return Err("can't disable ttl on a ttl instance".to_owned()); } else if !last_cfg.storage.enable_ttl && self.storage.enable_ttl { return Err("can't enable ttl on a non-ttl instance".to_owned()); } } if last_cfg.raftdb.defaultcf.format_version > 5 || last_cfg.rocksdb.defaultcf.format_version > 5 || last_cfg.rocksdb.writecf.format_version > 5 || last_cfg.rocksdb.lockcf.format_version > 5 || last_cfg.rocksdb.raftcf.format_version > 5 { return Err("format_version larger than 5 is unsupported".into()); } if !last_cfg.rfengine.wal_sync_dir.is_empty() && last_cfg.rfengine.wal_sync_dir != self.rfengine.wal_sync_dir { return Err(format!( "rfengine wal_sync_dir have been changed, former is '{:?}', \ current rfengine wal_sync_dir is '{:?}', please check if it is expected.", last_cfg.rfengine.wal_sync_dir, self.rfengine.wal_sync_dir )); } Ok(()) } pub fn from_file( path: &Path, unrecognized_keys: Option<&mut Vec>, ) -> Result> { let s = fs::read_to_string(path)?; let mut deserializer = toml::Deserializer::new(&s); let mut cfg = if let Some(keys) = unrecognized_keys { serde_ignored::deserialize(&mut deserializer, |key| keys.push(key.to_string())) } else { ::deserialize(&mut deserializer) }?; deserializer.end()?; cfg.cfg_path = path.display().to_string(); Ok(cfg) } pub fn write_to_file>(&self, path: P) -> Result<(), IoError> { let content = ::toml::to_string(&self).unwrap(); let mut f = fs::File::create(&path)?; f.write_all(content.as_bytes())?; f.sync_all()?; Ok(()) } pub fn write_into_metrics(&self) { self.raft_store.write_into_metrics(); self.rocksdb.write_into_metrics(); } pub fn with_tmp() -> Result<(TikvConfig, tempfile::TempDir), IoError> { let tmp = tempfile::tempdir()?; let mut cfg = TikvConfig::default(); cfg.storage.data_dir = tmp.path().display().to_string(); cfg.cfg_path = tmp.path().join(LAST_CONFIG_FILE).display().to_string(); Ok((cfg, tmp)) } fn suggested_memory_usage_limit() -> ReadableSize { let total = SysQuota::memory_limit_in_bytes(); // Reserve some space for page cache. The ReadableSize((total as f64 * MEMORY_USAGE_LIMIT_RATE) as u64) } pub fn compatible_adjust_to_raftstore(rfstore: &RfStoreConfig) -> RaftstoreConfig { let mut cfg = RaftstoreConfig::default(); macro_rules! assign { ($($field:ident),*) => { $(cfg.$field = rfstore.$field;)* } } // Direct field mappings assign!( raft_base_tick_interval, raft_heartbeat_ticks, split_region_check_tick_interval, raft_election_timeout_ticks, raft_min_election_timeout_ticks, raft_max_election_timeout_ticks, raft_max_size_per_msg, raft_max_inflight_msgs, raft_entry_max_size, raft_store_max_leader_lease, renew_leader_lease_advance_duration, pd_heartbeat_tick_interval, pd_store_heartbeat_tick_interval, max_peer_down_duration, max_leader_missing_duration, abnormal_leader_missing_duration, peer_stale_state_check_interval, leader_transfer_max_log_lag, raft_log_gc_tick_interval, raft_log_gc_size_limit, region_split_size, snap_generator_pool_size, snap_apply_batch_size, max_snapshot_file_raw_size, region_split_check_diff ); // Supplement additional fields for full compatibility assign!( raft_log_compact_sync_interval, raft_log_gc_threshold, raft_log_gc_count_limit, raft_log_reserve_max_ticks, raft_engine_purge_interval, raft_entry_cache_life_time, raft_reject_transfer_leader_duration, region_compact_check_interval, region_compact_check_step, region_compact_min_tombstones, region_compact_tombstones_percent, snap_mgr_gc_tick_interval, snap_gc_timeout, lock_cf_compact_interval, lock_cf_compact_bytes_threshold, notify_capacity, messages_per_tick, region_worker_tick_interval, clean_stale_ranges_tick, consistency_check_interval, report_region_flow_interval, check_leader_lease_interval, right_derive_when_split, merge_max_log_gap, merge_check_tick_interval, use_delete_range, cleanup_import_sst_interval, local_read_batch_size, store_io_pool_size, store_io_notify_capacity, future_poll_size, hibernate_regions, dev_assert, apply_yield_duration, apply_yield_write_size, evict_cache_on_memory_ratio, cmd_batch, cmd_batch_concurrent_ready_max_count, raft_write_size_limit, waterfall_metrics, io_reschedule_concurrent_max_count, io_reschedule_hotpot_duration, raft_msg_flush_interval, region_max_size, clean_stale_peer_delay, inspect_interval, report_min_resolved_ts_interval, reactive_memory_lock_tick_interval, reactive_memory_lock_timeout_tick, report_region_buckets_tick_interval, local_file_gc_tick_interval, local_file_gc_timeout, check_long_uncommitted_interval, long_uncommitted_base_threshold, max_entry_cache_warmup_duration, unreachable_backoff, check_peers_availability_interval, raft_worker_max_batch_size ); // BatchSystemConfig submodules cfg.apply_batch_system.pool_size = rfstore.apply_batch_system.pool_size; cfg.apply_batch_system.max_batch_size = rfstore.apply_batch_system.max_batch_size; cfg.store_batch_system.pool_size = rfstore.store_batch_system.pool_size; cfg.store_batch_system.max_batch_size = rfstore.store_batch_system.max_batch_size; cfg } } /// Prevents launching with an incompatible configuration /// /// Loads the previously-loaded configuration from `last_tikv.toml`, /// compares key configuration items and fails if they are not /// identical. pub fn check_critical_config(config: &TikvConfig) -> Result<(), String> { // Check current critical configurations with last time, if there are some // changes, user must guarantee relevant works have been done. if let Some(mut cfg) = get_last_config(&config.storage.data_dir) { cfg.compatible_adjust(); cfg.adjust_cpu_limited_worker_counts(); if let Err(e) = cfg.validate() { warn!("last_tikv.toml is invalid but ignored: {:?}", e); } config.check_critical_cfg_with(&cfg)?; } Ok(()) } fn get_last_config(data_dir: &str) -> Option { let store_path = Path::new(data_dir); let last_cfg_path = store_path.join(LAST_CONFIG_FILE); if last_cfg_path.exists() { return Some( TikvConfig::from_file(&last_cfg_path, None).unwrap_or_else(|e| { panic!( "invalid auto generated configuration file {}, err {}", last_cfg_path.display(), e ); }), ); } None } /// Persists config to `last_tikv.toml` pub fn persist_config(config: &TikvConfig) -> Result<(), String> { let store_path = Path::new(&config.storage.data_dir); let last_cfg_path = store_path.join(LAST_CONFIG_FILE); let tmp_cfg_path = store_path.join(TMP_CONFIG_FILE); let same_as_last_cfg = fs::read_to_string(&last_cfg_path) .is_ok_and(|last_cfg| toml::to_string(&config).unwrap() == last_cfg); if same_as_last_cfg { return Ok(()); } // Create parent directory if missing. if let Err(e) = fs::create_dir_all(store_path) { return Err(format!( "create parent directory '{}' failed: {}", store_path.to_str().unwrap(), e )); } // Persist current configurations to temporary file. if let Err(e) = config.write_to_file(&tmp_cfg_path) { return Err(format!( "persist config to '{}' failed: {}", tmp_cfg_path.to_str().unwrap(), e )); } // Rename temporary file to last config file. if let Err(e) = fs::rename(&tmp_cfg_path, &last_cfg_path) { return Err(format!( "rename config file from '{}' to '{}' failed: {}", tmp_cfg_path.to_str().unwrap(), last_cfg_path.to_str().unwrap(), e )); } Ok(()) } pub fn write_config>(path: P, content: &[u8]) -> CfgResult<()> { let tmp_cfg_path = match path.as_ref().parent() { Some(p) => p.join(TMP_CONFIG_FILE), None => { return Err(Box::new(IoError::other(format!( "failed to get parent path of config file: {}", path.as_ref().display() )))); } }; { let mut f = fs::File::create(&tmp_cfg_path)?; f.write_all(content)?; f.sync_all()?; } fs::rename(&tmp_cfg_path, &path)?; Ok(()) } // convert tikv config to a flatten array. pub fn to_flatten_config_info(cfg: &TikvConfig) -> Vec { fn to_cfg_value(default_value: &Value, cfg_value: Option<&Value>, key: &str) -> Value { let mut res = Map::with_capacity(2); res.insert("Name".into(), Value::String(key.into())); res.insert("DefaultValue".into(), default_value.clone()); if let Some(cfg_val) = cfg_value { if default_value != cfg_val { res.insert("ValueInFile".into(), cfg_val.clone()); } } Value::Object(res) } // configs that should not be flatten because the config type is HashMap instead // of submodule. lazy_static! { static ref NO_FLATTEN_CFGS: HashSet<&'static str> = { let mut set = HashSet::new(); set.insert("server.labels"); set }; } fn flatten_value( default_obj: &Map, value_obj: &Map, key_buf: &mut String, res: &mut Vec, ) { for (k, v) in default_obj.iter() { let cfg_val = value_obj.get(k); let prev_len = key_buf.len(); if !key_buf.is_empty() { key_buf.push('.'); } key_buf.push_str(k); if v.is_object() && !NO_FLATTEN_CFGS.contains(key_buf.as_str()) { flatten_value( v.as_object().unwrap(), cfg_val.unwrap().as_object().unwrap(), key_buf, res, ); } else { res.push(to_cfg_value(v, cfg_val, key_buf)); } key_buf.truncate(prev_len); } } let cfg_value = to_value(cfg).unwrap(); let default_value = to_value(TikvConfig::default()).unwrap(); let mut key_buf = String::new(); let mut res = Vec::new(); flatten_value( default_value.as_object().unwrap(), cfg_value.as_object().unwrap(), &mut key_buf, &mut res, ); res } lazy_static! { pub static ref TIKVCONFIG_TYPED: ConfigChange = TikvConfig::default().typed(); } fn serde_to_online_config(name: String) -> String { let res = name.replace("raftstore", "raft_store").replace('-', "_"); match res.as_ref() { "raft_store.store_pool_size" | "raft_store.store_max_batch_size" => { res.replace("store_", "store_batch_system.") } "raft_store.apply_pool_size" | "raft_store.apply_max_batch_size" => { res.replace("apply_", "apply_batch_system.") } _ => res, } } fn to_config_change(change: HashMap) -> CfgResult { fn helper( mut fields: Vec, dst: &mut ConfigChange, typed: &ConfigChange, value: String, ) -> CfgResult<()> { if let Some(field) = fields.pop() { return match typed.get(&field) { None => Err(format!("unexpect fields: {}", field).into()), Some(ConfigValue::Skip) => { Err(format!("config {} can not be changed", field).into()) } Some(ConfigValue::Module(m)) => { if let ConfigValue::Module(n_dst) = dst .entry(field) .or_insert_with(|| ConfigValue::Module(HashMap::new())) { return helper(fields, n_dst, m, value); } panic!("unexpect config value"); } Some(v) => { if fields.is_empty() { return match to_change_value(&value, v) { Err(_) => Err(format!("failed to parse: {}", value).into()), Ok(v) => { dst.insert(field, v); Ok(()) } }; } let c: Vec<_> = fields.into_iter().rev().collect(); Err(format!("unexpect fields: {}", c[..].join(".")).into()) } }; } Ok(()) } let mut res = HashMap::new(); for (mut name, value) in change { name = serde_to_online_config(name); let fields: Vec<_> = name .as_str() .split('.') .map(|s| s.to_owned()) .rev() .collect(); helper(fields, &mut res, &TIKVCONFIG_TYPED, value)?; } Ok(res) } fn to_change_value(v: &str, typed: &ConfigValue) -> CfgResult { let v = v.trim_matches('\"'); let res = match typed { ConfigValue::Duration(_) => ConfigValue::from(v.parse::()?), ConfigValue::Size(_) => ConfigValue::from(v.parse::()?), ConfigValue::U64(_) => ConfigValue::from(v.parse::()?), ConfigValue::F64(_) => ConfigValue::from(v.parse::()?), ConfigValue::U32(_) => ConfigValue::from(v.parse::()?), ConfigValue::I32(_) => ConfigValue::from(v.parse::()?), ConfigValue::Usize(_) => ConfigValue::from(v.parse::()?), ConfigValue::Bool(_) => ConfigValue::from(v.parse::()?), ConfigValue::String(_) => ConfigValue::String(v.to_owned()), _ => unreachable!(), }; Ok(res) } fn to_toml_encode(change: HashMap) -> CfgResult> { fn helper(mut fields: Vec, typed: &ConfigChange) -> CfgResult { if let Some(field) = fields.pop() { match typed.get(&field) { None | Some(ConfigValue::Skip) => Err(Box::new(IoError::other(format!( "failed to get field: {}", field )))), Some(ConfigValue::Module(m)) => helper(fields, m), Some(c) => { if !fields.is_empty() { return Err(Box::new(IoError::other(format!( "unexpect fields: {:?}", fields )))); } match c { ConfigValue::Duration(_) | ConfigValue::Size(_) | ConfigValue::String(_) => Ok(true), ConfigValue::None => Err(Box::new(IoError::other(format!( "unexpect none field: {:?}", c )))), _ => Ok(false), } } } } else { Err(Box::new(IoError::other("failed to get field"))) } } let mut dst = HashMap::new(); for (name, value) in change { let online_config_name = serde_to_online_config(name.clone()); let fields: Vec<_> = online_config_name .as_str() .split('.') .map(|s| s.to_owned()) .rev() .collect(); if helper(fields, &TIKVCONFIG_TYPED)? { dst.insert(name.replace('_', "-"), format!("\"{}\"", value)); } else { dst.insert(name.replace('_', "-"), value); } } Ok(dst) } #[derive(PartialEq, Eq, Hash, Debug, Clone)] pub enum Module { Readpool, Server, Metric, Raftstore, Coprocessor, Pd, Rocksdb, Raftdb, RaftEngine, Rfengine, Storage, Security, Encryption, Import, Backup, PessimisticTxn, Gc, Split, Cdc, ResolvedTs, ResourceMetering, Overload, ResourceControl, BackupStream, Quota, Kvengine, Log, Memory, Unknown(String), } impl From<&str> for Module { fn from(m: &str) -> Module { match m { "readpool" => Module::Readpool, "server" => Module::Server, "metric" => Module::Metric, "raft_store" => Module::Raftstore, "coprocessor" => Module::Coprocessor, "pd" => Module::Pd, "split" => Module::Split, "rocksdb" => Module::Rocksdb, "raftdb" => Module::Raftdb, "raft_engine" => Module::RaftEngine, "rfengine" => Module::Rfengine, "storage" => Module::Storage, "security" => Module::Security, "import" => Module::Import, "backup" => Module::Backup, "backup_stream" => Module::BackupStream, "pessimistic_txn" => Module::PessimisticTxn, "gc" => Module::Gc, "cdc" => Module::Cdc, "resolved_ts" => Module::ResolvedTs, "resource_metering" => Module::ResourceMetering, "overload" => Module::Overload, "resource_control" => Module::ResourceControl, "quota" => Module::Quota, "kvengine" => Module::Kvengine, "log" => Module::Log, "memory" => Module::Memory, n => Module::Unknown(n.to_owned()), } } } /// ConfigController use to register each module's config manager, /// and dispatch the change of config to corresponding managers or /// return the change if the incoming change is invalid. #[derive(Default, Clone)] pub struct ConfigController { inner: Arc>, } #[derive(Default)] struct ConfigInner { current: TikvConfig, config_mgrs: HashMap>, } impl ConfigController { pub fn new(current: TikvConfig) -> Self { ConfigController { inner: Arc::new(RwLock::new(ConfigInner { current, config_mgrs: HashMap::new(), })), } } pub fn update(&self, change: HashMap) -> CfgResult<()> { let diff = to_config_change(change.clone())?; self.update_impl(diff, Some(change)) } pub fn update_from_toml_file(&self) -> CfgResult<()> { let current = self.get_current(); match TikvConfig::from_file(Path::new(¤t.cfg_path), None) { Ok(incoming) => { let diff = current.diff(&incoming); self.update_impl(diff, None) } Err(e) => Err(e), } } fn update_impl( &self, mut diff: HashMap, change: Option>, ) -> CfgResult<()> { diff = { let incoming = self.get_current(); let mut updated = incoming.clone(); updated.update(diff)?; updated.adjust_cpu_limited_worker_counts(); // Other config values might still be adjusted in `validate`. updated.validate()?; incoming.diff(&updated) }; let mut inner = self.inner.write().unwrap(); let mut to_update = HashMap::with_capacity(diff.len()); for (name, change) in diff.into_iter() { match change { ConfigValue::Module(change) => { // update a submodule's config only if changes had been successfully // dispatched to corresponding config manager, to avoid dispatch change twice if let Some(mgr) = inner.config_mgrs.get_mut(&Module::from(name.as_str())) { if let Err(e) = mgr.dispatch(change.clone()) { // we already verified the correctness at the beginning of this // function. inner.current.update(to_update).unwrap(); return Err(e); } } to_update.insert(name, ConfigValue::Module(change)); } _ => { let _ = to_update.insert(name, change); } } } debug!("all config change had been dispatched"; "change" => ?to_update); // we already verified the correctness at the beginning of this function. inner.current.update(to_update).unwrap(); // Write change to the config file if let Some(change) = change { let content = { let change = to_toml_encode(change)?; let src = if Path::new(&inner.current.cfg_path).exists() { fs::read_to_string(&inner.current.cfg_path)? } else { String::new() }; let mut t = TomlWriter::new(); t.write_change(src, change); t.finish() }; write_config(&inner.current.cfg_path, &content)?; } Ok(()) } pub fn update_config(&self, name: &str, value: &str) -> CfgResult<()> { let mut m = HashMap::new(); m.insert(name.to_owned(), value.to_owned()); self.update(m) } pub fn register(&self, module: Module, cfg_mgr: Box) { let mut inner = self.inner.write().unwrap(); if inner.config_mgrs.insert(module.clone(), cfg_mgr).is_some() { warn!("config manager for module {:?} already registered", module) } } pub fn get_current(&self) -> TikvConfig { self.inner.read().unwrap().current.clone() } } #[cfg(test)] mod tests { use std::{convert::TryInto, sync::Arc, time::Duration}; use case_macros::*; use futures::executor::block_on; use grpcio::ResourceQuota; use itertools::Itertools; use slog::Level; use tempfile::Builder; use tikv_util::{ config::VersionTrack, logger::{get_log_level, txn_info_logging_enabled}, quota_limiter::{QuotaLimitConfigManager, QuotaLimiter}, sys::SysQuota, }; use super::*; use crate::server::config::ServerConfigManager; #[test] fn test_case_macro() { let h = kebab_case!(HelloWorld); assert_eq!(h, "hello-world"); let h = kebab_case!(WelcomeToMyHouse); assert_eq!(h, "welcome-to-my-house"); let h = snake_case!(HelloWorld); assert_eq!(h, "hello_world"); let h = snake_case!(WelcomeToMyHouse); assert_eq!(h, "welcome_to_my_house"); } #[test] fn test_check_critical_cfg_with() { let mut tikv_cfg = TikvConfig::default(); let last_cfg = TikvConfig::default(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap(); let mut tikv_cfg = TikvConfig::default(); let mut last_cfg = TikvConfig::default(); tikv_cfg.rocksdb.wal_dir = "/data/wal_dir".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap_err(); last_cfg.rocksdb.wal_dir = "/data/wal_dir".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap(); let mut tikv_cfg = TikvConfig::default(); let mut last_cfg = TikvConfig::default(); tikv_cfg.storage.data_dir = "/data1".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap_err(); last_cfg.storage.data_dir = "/data1".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap(); // Disable Raft Engine and uses RocksDB. let mut tikv_cfg = TikvConfig::default(); let mut last_cfg = TikvConfig::default(); tikv_cfg.raftdb.wal_dir = "/raft/wal_dir".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap_err(); last_cfg.raftdb.wal_dir = "/raft/wal_dir".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap(); tikv_cfg.raft_store.raftdb_path = "/raft_path".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap_err(); last_cfg.raft_store.raftdb_path = "/raft_path".to_owned(); tikv_cfg.validate().unwrap(); tikv_cfg.check_critical_cfg_with(&last_cfg).unwrap(); // Check api version. { let cases = [ (ApiVersion::V1, ApiVersion::V1, true), (ApiVersion::V1, ApiVersion::V1ttl, false), (ApiVersion::V1, ApiVersion::V2, true), (ApiVersion::V1ttl, ApiVersion::V1, false), (ApiVersion::V1ttl, ApiVersion::V1ttl, true), (ApiVersion::V1ttl, ApiVersion::V2, true), (ApiVersion::V2, ApiVersion::V1, true), (ApiVersion::V2, ApiVersion::V1ttl, true), (ApiVersion::V2, ApiVersion::V2, true), ]; for (from_api, to_api, expected) in cases { last_cfg.storage.set_api_version(from_api); tikv_cfg.storage.set_api_version(to_api); tikv_cfg.validate().unwrap(); assert_eq!( tikv_cfg.check_critical_cfg_with(&last_cfg).is_ok(), expected ); } } } #[test] fn test_last_cfg_modified() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); let store_path = Path::new(&cfg.storage.data_dir); let last_cfg_path = store_path.join(LAST_CONFIG_FILE); cfg.write_to_file(&last_cfg_path).unwrap(); let mut last_cfg_metadata = last_cfg_path.metadata().unwrap(); let first_modified = last_cfg_metadata.modified().unwrap(); // not write to file when config is the equivalent of last one. persist_config(&cfg).unwrap(); last_cfg_metadata = last_cfg_path.metadata().unwrap(); assert_eq!(last_cfg_metadata.modified().unwrap(), first_modified); // write to file when config is the inequivalent of last one. cfg.log_level = slog::Level::Warning.into(); persist_config(&cfg).unwrap(); last_cfg_metadata = last_cfg_path.metadata().unwrap(); assert_ne!(last_cfg_metadata.modified().unwrap(), first_modified); } #[test] fn test_persist_cfg() { let dir = Builder::new().prefix("test_persist_cfg").tempdir().unwrap(); let path_buf = dir.path().join(LAST_CONFIG_FILE); let file = path_buf.as_path(); let (s1, s2) = ("/xxx/wal_dir".to_owned(), "/yyy/wal_dir".to_owned()); let mut tikv_cfg = TikvConfig::default(); tikv_cfg.rocksdb.wal_dir = s1.clone(); tikv_cfg.raftdb.wal_dir = s2.clone(); tikv_cfg.write_to_file(file).unwrap(); let cfg_from_file = TikvConfig::from_file(file, None).unwrap_or_else(|e| { panic!( "invalid auto generated configuration file {}, err {}", file.display(), e ); }); assert_eq!(cfg_from_file.rocksdb.wal_dir, s1); assert_eq!(cfg_from_file.raftdb.wal_dir, s2); // write critical config when exist. tikv_cfg.rocksdb.wal_dir = s2.clone(); tikv_cfg.raftdb.wal_dir = s1.clone(); tikv_cfg.write_to_file(file).unwrap(); let cfg_from_file = TikvConfig::from_file(file, None).unwrap_or_else(|e| { panic!( "invalid auto generated configuration file {}, err {}", file.display(), e ); }); assert_eq!(cfg_from_file.rocksdb.wal_dir, s2); assert_eq!(cfg_from_file.raftdb.wal_dir, s1); } #[test] fn test_flatten_cfg() { let mut cfg = TikvConfig::default(); cfg.server.labels.insert("zone".into(), "test".into()); cfg.raft_store.raft_log_gc_count_limit = Some(123); let flattened = to_flatten_config_info(&cfg); let mut expected = HashMap::new(); let mut labels = Map::new(); labels.insert("zone".into(), Value::String("test".into())); expected.insert("server.labels", Value::Object(labels)); expected.insert( "raftstore.raft-log-gc-count-limit", Value::Number(123.into()), ); for v in &flattened { let obj = v.as_object().unwrap(); if let Some(v) = expected.get(&obj["Name"].as_str().unwrap()) { assert_eq!(v, &obj["ValueInFile"]); } else { assert!(!obj.contains_key("ValueInFile")); } } } #[test] fn test_create_parent_dir_if_missing() { let root_path = Builder::new() .prefix("test_create_parent_dir_if_missing") .tempdir() .unwrap(); let path = root_path.path().join("not_exist_dir"); let mut tikv_cfg = TikvConfig::default(); tikv_cfg.storage.data_dir = path.as_path().to_str().unwrap().to_owned(); persist_config(&tikv_cfg).unwrap(); } #[test] fn test_keepalive_check() { let mut tikv_cfg = TikvConfig::default(); tikv_cfg.pd.endpoints = vec!["".to_owned()]; let dur = tikv_cfg.raft_store.raft_heartbeat_interval(); tikv_cfg.server.grpc_keepalive_time = ReadableDuration(dur); tikv_cfg.validate().unwrap_err(); tikv_cfg.server.grpc_keepalive_time = ReadableDuration(dur * 2); tikv_cfg.validate().unwrap(); } #[test] fn test_block_size() { let mut tikv_cfg = TikvConfig::default(); tikv_cfg.pd.endpoints = vec!["".to_owned()]; tikv_cfg.rocksdb.defaultcf.block_size = ReadableSize::gb(10); tikv_cfg.rocksdb.lockcf.block_size = ReadableSize::gb(10); tikv_cfg.rocksdb.writecf.block_size = ReadableSize::gb(10); tikv_cfg.rocksdb.raftcf.block_size = ReadableSize::gb(10); tikv_cfg.raftdb.defaultcf.block_size = ReadableSize::gb(10); tikv_cfg.validate().unwrap_err(); tikv_cfg.rocksdb.defaultcf.block_size = ReadableSize::kb(10); tikv_cfg.rocksdb.lockcf.block_size = ReadableSize::kb(10); tikv_cfg.rocksdb.writecf.block_size = ReadableSize::kb(10); tikv_cfg.rocksdb.raftcf.block_size = ReadableSize::kb(10); tikv_cfg.raftdb.defaultcf.block_size = ReadableSize::kb(10); tikv_cfg.validate().unwrap(); } #[test] fn test_parse_log_level() { #[derive(Serialize, Deserialize, Debug)] struct LevelHolder { v: LogLevel, } let legal_cases = vec![ ("fatal", Level::Critical), ("error", Level::Error), ("warn", Level::Warning), ("debug", Level::Debug), ("trace", Level::Trace), ("info", Level::Info), ]; for (serialized, deserialized) in legal_cases { let holder = LevelHolder { v: deserialized.into(), }; let res_string = toml::to_string(&holder).unwrap(); let exp_string = format!("v = \"{}\"\n", serialized); assert_eq!(res_string, exp_string); let res_value: LevelHolder = toml::from_str(&exp_string).unwrap(); assert_eq!(res_value.v, deserialized.into()); } let compatibility_cases = vec![("warning", Level::Warning), ("critical", Level::Critical)]; for (serialized, deserialized) in compatibility_cases { let variant_string = format!("v = \"{}\"\n", serialized); let res_value: LevelHolder = toml::from_str(&variant_string).unwrap(); assert_eq!(res_value.v, deserialized.into()); } let illegal_cases = vec!["foobar", ""]; for case in illegal_cases { let string = format!("v = \"{}\"\n", case); toml::from_str::(&string).unwrap_err(); } } #[test] fn test_to_config_change() { assert_eq!( to_change_value("10h", &ConfigValue::Duration(0)).unwrap(), ConfigValue::from(ReadableDuration::hours(10)) ); assert_eq!( to_change_value("100MB", &ConfigValue::Size(0)).unwrap(), ConfigValue::from(ReadableSize::mb(100)) ); assert_eq!( to_change_value("10000", &ConfigValue::U64(0)).unwrap(), ConfigValue::from(10000u64) ); assert_eq!( to_change_value( "42%", &ConfigValue::from(config::AbsoluteOrPercentSize::Percent(70.0)) ) .unwrap(), ConfigValue::from(config::AbsoluteOrPercentSize::Percent(42.0)) ); let old = TikvConfig::default(); let mut incoming = TikvConfig::default(); incoming.coprocessor.region_split_keys = Some(10000); incoming.gc.max_write_bytes_per_sec = ReadableSize::mb(100); incoming.rocksdb.defaultcf.block_cache_size = ReadableSize::mb(500); incoming.storage.io_rate_limit.import_priority = file_system::IoPriority::High; let diff = old.diff(&incoming); let mut change = HashMap::new(); change.insert( "coprocessor.region-split-keys".to_owned(), "10000".to_owned(), ); change.insert("gc.max-write-bytes-per-sec".to_owned(), "100MB".to_owned()); change.insert( "rocksdb.defaultcf.block-cache-size".to_owned(), "500MB".to_owned(), ); change.insert( "storage.io-rate-limit.import-priority".to_owned(), "high".to_owned(), ); let res = to_config_change(change).unwrap(); assert_eq!(diff, res); let mut change = HashMap::new(); change.insert("kvengine.ia.mem-cap".to_owned(), "256MB".to_owned()); change.insert("kvengine.ia.disk-cap".to_owned(), "42%".to_owned()); let mut updated = TikvConfig::default(); updated.update(to_config_change(change).unwrap()).unwrap(); assert_eq!( updated.kvengine.ia.mem_cap, config::AbsoluteOrPercentSize::Abs(ReadableSize::mb(256)) ); assert_eq!( updated.kvengine.ia.disk_cap, config::AbsoluteOrPercentSize::Percent(42.0) ); // illegal cases let cases = vec![ // wrong value type ("gc.max-write-bytes-per-sec".to_owned(), "10s".to_owned()), ( "pessimistic-txn.wait-for-lock-timeout".to_owned(), "1MB".to_owned(), ), // missing or unknown config fields ("xxx.yyy".to_owned(), "12".to_owned()), ( "rocksdb.defaultcf.block-cache-size.xxx".to_owned(), "50MB".to_owned(), ), ("rocksdb.xxx.block-cache-size".to_owned(), "50MB".to_owned()), ("rocksdb.block-cache-size".to_owned(), "50MB".to_owned()), // not support change config ( "raftstore.raft-heartbeat-ticks".to_owned(), "100".to_owned(), ), ("raftstore.prevote".to_owned(), "false".to_owned()), ]; for (name, value) in cases { let mut change = HashMap::new(); change.insert(name, value); to_config_change(change).unwrap_err(); } let mut change = HashMap::new(); change.insert( "resource-control.max-read-cpu-ratio".to_owned(), "0.9".to_owned(), ); let change = to_config_change(change).unwrap(); let ConfigValue::Module(resource_control) = &change["resource_control"] else { panic!("resource-control change should be a module"); }; assert_eq!( resource_control.get("max_read_cpu_ratio"), Some(&ConfigValue::F64(0.9)) ); let mut change = HashMap::new(); change.insert( "readpool.unified.keyspace-fairness.max-read-cpu-ratio".to_owned(), "0.3".to_owned(), ); to_config_change(change).unwrap(); } #[test] fn test_to_toml_encode() { let mut change = HashMap::new(); change.insert( "raftstore.pd-heartbeat-tick-interval".to_owned(), "1h".to_owned(), ); change.insert( "coprocessor.region-split-keys".to_owned(), "10000".to_owned(), ); change.insert("gc.max-write-bytes-per-sec".to_owned(), "100MB".to_owned()); change.insert( "rocksdb.defaultcf.titan.blob-run-mode".to_owned(), "read-only".to_owned(), ); change.insert("raftstore.apply_pool_size".to_owned(), "7".to_owned()); change.insert("raftstore.store-pool-size".to_owned(), "17".to_owned()); change.insert("kvengine.ia.disk-cap".to_owned(), "42%".to_owned()); let res = to_toml_encode(change).unwrap(); assert_eq!( res.get("raftstore.pd-heartbeat-tick-interval"), Some(&"\"1h\"".to_owned()) ); assert_eq!( res.get("coprocessor.region-split-keys"), Some(&"10000".to_owned()) ); assert_eq!( res.get("gc.max-write-bytes-per-sec"), Some(&"\"100MB\"".to_owned()) ); assert_eq!( res.get("rocksdb.defaultcf.titan.blob-run-mode"), Some(&"\"read-only\"".to_owned()) ); assert_eq!(res.get("raftstore.apply-pool-size"), Some(&"7".to_owned())); assert_eq!(res.get("raftstore.store-pool-size"), Some(&"17".to_owned())); assert_eq!(res.get("kvengine.ia.disk-cap"), Some(&"\"42%\"".to_owned())); } #[test] fn test_change_resolved_ts_config() { use crossbeam::channel; pub struct TestConfigManager(channel::Sender); impl ConfigManager for TestConfigManager { fn dispatch(&mut self, change: ConfigChange) -> online_config::Result<()> { self.0.send(change).unwrap(); Ok(()) } } let (cfg, _dir) = TikvConfig::with_tmp().unwrap(); let cfg_controller = ConfigController::new(cfg); let (tx, rx) = channel::unbounded(); cfg_controller.register(Module::ResolvedTs, Box::new(TestConfigManager(tx))); // Return error if try to update not support config or unknow config cfg_controller .update_config("resolved-ts.enable", "false") .unwrap_err(); cfg_controller .update_config("resolved-ts.scan-lock-pool-size", "10") .unwrap_err(); cfg_controller .update_config("resolved-ts.xxx", "false") .unwrap_err(); let mut resolved_ts_cfg = cfg_controller.get_current().resolved_ts; // Default value assert_eq!( resolved_ts_cfg.advance_ts_interval, ReadableDuration::secs(1) ); // Update `advance-ts-interval` to 100ms cfg_controller .update_config("resolved-ts.advance-ts-interval", "100ms") .unwrap(); resolved_ts_cfg.update(rx.recv().unwrap()).unwrap(); assert_eq!( resolved_ts_cfg.advance_ts_interval, ReadableDuration::millis(100) ); // Return error if try to update `advance-ts-interval` to an invalid value cfg_controller .update_config("resolved-ts.advance-ts-interval", "0m") .unwrap_err(); assert_eq!( resolved_ts_cfg.advance_ts_interval, ReadableDuration::millis(100) ); // Update `advance-ts-interval` to 3s cfg_controller .update_config("resolved-ts.advance-ts-interval", "3s") .unwrap(); resolved_ts_cfg.update(rx.recv().unwrap()).unwrap(); assert_eq!( resolved_ts_cfg.advance_ts_interval, ReadableDuration::secs(3) ); } #[test] fn test_change_kvengine_ia_rate_limit_config() { use crossbeam::channel; pub struct TestConfigManager(channel::Sender); impl ConfigManager for TestConfigManager { fn dispatch(&mut self, change: ConfigChange) -> online_config::Result<()> { self.0.send(change).unwrap(); Ok(()) } } let (cfg, _dir) = TikvConfig::with_tmp().unwrap(); let cfg_controller = ConfigController::new(cfg); let (tx, rx) = channel::unbounded(); cfg_controller.register(Module::Kvengine, Box::new(TestConfigManager(tx))); cfg_controller .update_config("kvengine.ia.remote-read-max-throughput", "128MB") .unwrap(); let change = rx.recv().unwrap(); let Some(ConfigValue::Module(ia_change)) = change.get("ia") else { panic!("expect ia change, got {:?}", change); }; let remote_read_max_throughput: ReadableSize = ia_change .get("remote_read_max_throughput") .expect("expect remote_read_max_throughput") .clone() .into(); assert_eq!(remote_read_max_throughput, ReadableSize::mb(128)); cfg_controller .update_config("kvengine.ia.local-write-max-throughput", "64MB") .unwrap(); let change = rx.recv().unwrap(); let Some(ConfigValue::Module(ia_change)) = change.get("ia") else { panic!("expect ia change, got {:?}", change); }; let local_write_max_throughput: ReadableSize = ia_change .get("local_write_max_throughput") .expect("expect local_write_max_throughput") .clone() .into(); assert_eq!(local_write_max_throughput, ReadableSize::mb(64)); cfg_controller .update_config("kvengine.ia.io-rate-limiter-refill-period", "0ms") .unwrap_err(); cfg_controller .update_config("kvengine.ia.io-rate-limiter-refill-period", "200ms") .unwrap_err(); cfg_controller .update_config("kvengine.ia.mem-cap", "1GB") .unwrap(); let change = rx.recv().unwrap(); let Some(ConfigValue::Module(ia_change)) = change.get("ia") else { panic!("expect ia change, got {:?}", change); }; let mem_cap: config::AbsoluteOrPercentSize = ia_change .get("mem_cap") .expect("expect mem_cap") .clone() .try_into() .unwrap(); assert_eq!( mem_cap, config::AbsoluteOrPercentSize::Abs(ReadableSize::gb(1)) ); cfg_controller .update_config("kvengine.ia.disk-cap", "42%") .unwrap(); let change = rx.recv().unwrap(); let Some(ConfigValue::Module(ia_change)) = change.get("ia") else { panic!("expect ia change, got {:?}", change); }; let disk_cap: config::AbsoluteOrPercentSize = ia_change .get("disk_cap") .expect("expect disk_cap") .clone() .try_into() .unwrap(); assert_eq!(disk_cap, config::AbsoluteOrPercentSize::Percent(42.0)); cfg_controller .update_config("kvengine.ia.cache-cap-to-total-data-size-ratio", "0.5") .unwrap(); let change = rx.recv().unwrap(); let Some(ConfigValue::Module(ia_change)) = change.get("ia") else { panic!("expect ia change, got {:?}", change); }; assert_eq!( ia_change.get("cache_cap_to_total_data_size_ratio"), Some(&ConfigValue::F64(0.5)) ); cfg_controller .update_config("kvengine.ia.cache-capacity-shrink-step", "2GiB") .unwrap(); let change = rx.recv().unwrap(); let Some(ConfigValue::Module(ia_change)) = change.get("ia") else { panic!("expect ia change, got {:?}", change); }; assert_eq!( ia_change.get("cache_capacity_shrink_step"), Some(&ConfigValue::Size(ReadableSize::gb(2).0)) ); } #[test] fn test_change_logconfig() { let (cfg, _dir) = TikvConfig::with_tmp().unwrap(); let cfg_controller = ConfigController::new(cfg); cfg_controller.register(Module::Log, Box::new(LogConfigManager)); // Test log level changes cfg_controller.update_config("log.level", "warn").unwrap(); assert_eq!(get_log_level().unwrap(), Level::Warning); assert_eq!( cfg_controller.get_current().log.level, LogLevel(Level::Warning) ); cfg_controller .update_config("log.level", "invalid") .unwrap_err(); assert_eq!( cfg_controller.get_current().log.level, LogLevel(Level::Warning) ); // Test txn-info-logging changes // Set to false explicitly cfg_controller .update_config("log.txn-info-logging", "false") .unwrap(); assert!(!txn_info_logging_enabled()); assert!(!cfg_controller.get_current().log.txn_info_logging); // Enable txn_info_logging cfg_controller .update_config("log.txn-info-logging", "true") .unwrap(); assert!(txn_info_logging_enabled()); assert!(cfg_controller.get_current().log.txn_info_logging); // Disable txn_info_logging again cfg_controller .update_config("log.txn-info-logging", "false") .unwrap(); assert!(!txn_info_logging_enabled()); assert!(!cfg_controller.get_current().log.txn_info_logging); } #[test] fn test_dispatch_titan_blob_run_mode_config() { let mut cfg = TikvConfig::default(); let mut incoming = cfg.clone(); cfg.rocksdb.defaultcf.titan.blob_run_mode = BlobRunMode::Normal; incoming.rocksdb.defaultcf.titan.blob_run_mode = BlobRunMode::Fallback; let diff = cfg .rocksdb .defaultcf .titan .diff(&incoming.rocksdb.defaultcf.titan); assert_eq!(diff.len(), 1); let diff = config_value_to_string(diff.into_iter().collect()); assert_eq!(diff.len(), 1); assert_eq!(diff[0].0.as_str(), "blob_run_mode"); assert_eq!(diff[0].1.as_str(), "fallback"); } #[test] fn test_change_quota_config() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.quota.foreground_cpu_time = 1000; cfg.quota.foreground_write_bandwidth = ReadableSize::mb(128); cfg.quota.foreground_read_bandwidth = ReadableSize::mb(256); cfg.quota.background_cpu_time = 1000; cfg.quota.background_write_bandwidth = ReadableSize::mb(128); cfg.quota.background_read_bandwidth = ReadableSize::mb(256); cfg.quota.max_delay_duration = ReadableDuration::secs(1); cfg.validate().unwrap(); let quota_limiter = Arc::new(QuotaLimiter::new( cfg.quota.foreground_cpu_time, cfg.quota.foreground_write_bandwidth, cfg.quota.foreground_read_bandwidth, cfg.quota.background_cpu_time, cfg.quota.background_write_bandwidth, cfg.quota.background_read_bandwidth, cfg.quota.max_delay_duration, false, )); let cfg_controller = ConfigController::new(cfg.clone()); cfg_controller.register( Module::Quota, Box::new(QuotaLimitConfigManager::new(quota_limiter.clone())), ); cfg_controller .update_config("quota.max-delay-duration", "213504d") .unwrap_err(); assert_eq!(cfg_controller.get_current(), cfg); cfg_controller .update_config("quota.foreground-cpu-time", "2000") .unwrap(); cfg.quota.foreground_cpu_time = 2000; assert_eq!(cfg_controller.get_current(), cfg); assert_eq!(quota_limiter.cputime_limiter(true), 2_000_000.0); cfg_controller .update_config("quota.foreground-write-bandwidth", "256MB") .unwrap(); cfg.quota.foreground_write_bandwidth = ReadableSize::mb(256); assert_eq!(cfg_controller.get_current(), cfg); let mut sample = quota_limiter.new_sample(true); sample.add_read_bytes(ReadableSize::mb(32).0 as usize); let should_delay = block_on(quota_limiter.consume_sample(sample, true)); assert_eq!(should_delay, Duration::from_millis(125)); cfg_controller .update_config("quota.foreground-read-bandwidth", "512MB") .unwrap(); cfg.quota.foreground_read_bandwidth = ReadableSize::mb(512); assert_eq!(cfg_controller.get_current(), cfg); let mut sample = quota_limiter.new_sample(true); sample.add_write_bytes(ReadableSize::mb(128).0 as usize); let should_delay = block_on(quota_limiter.consume_sample(sample, true)); assert_eq!(should_delay, Duration::from_millis(500)); cfg_controller .update_config("quota.background-cpu-time", "2000") .unwrap(); cfg.quota.background_cpu_time = 2000; assert_eq!(cfg_controller.get_current(), cfg); assert_eq!(quota_limiter.cputime_limiter(false), 2_000_000.0); cfg_controller .update_config("quota.background-write-bandwidth", "256MB") .unwrap(); cfg.quota.background_write_bandwidth = ReadableSize::mb(256); assert_eq!(cfg_controller.get_current(), cfg); let mut sample = quota_limiter.new_sample(false); sample.add_read_bytes(ReadableSize::mb(32).0 as usize); let should_delay = block_on(quota_limiter.consume_sample(sample, false)); assert_eq!(should_delay, Duration::from_millis(125)); cfg_controller .update_config("quota.background-read-bandwidth", "512MB") .unwrap(); cfg.quota.background_read_bandwidth = ReadableSize::mb(512); assert_eq!(cfg_controller.get_current(), cfg); let mut sample = quota_limiter.new_sample(false); sample.add_write_bytes(ReadableSize::mb(128).0 as usize); let should_delay = block_on(quota_limiter.consume_sample(sample, false)); assert_eq!(should_delay, Duration::from_millis(500)); cfg_controller .update_config("quota.max-delay-duration", "50ms") .unwrap(); cfg.quota.max_delay_duration = ReadableDuration::millis(50); assert_eq!(cfg_controller.get_current(), cfg); let mut sample = quota_limiter.new_sample(true); sample.add_write_bytes(ReadableSize::mb(128).0 as usize); let should_delay = block_on(quota_limiter.consume_sample(sample, true)); assert_eq!(should_delay, Duration::from_millis(50)); let mut sample = quota_limiter.new_sample(false); sample.add_write_bytes(ReadableSize::mb(128).0 as usize); let should_delay = block_on(quota_limiter.consume_sample(sample, false)); assert_eq!(should_delay, Duration::from_millis(50)); cfg_controller .update_config("quota.enable-auto-tune", "true") .unwrap(); cfg.quota.enable_auto_tune = true; assert_eq!(cfg_controller.get_current(), cfg); assert!(quota_limiter.auto_tune_enabled()); } #[test] fn test_legacy_admission_config_round_trip() { let cfg: TikvConfig = toml::from_str( r#" [quota] foreground-cpu-time = 1000 foreground-read-bandwidth = "256MB" max-delay-duration = "250ms" [overload] enable = true max-keys = 100000000 max-data-size = "8GB" "#, ) .unwrap(); assert_eq!(cfg.quota.foreground_cpu_time, 1000); assert_eq!(cfg.quota.foreground_read_bandwidth, ReadableSize::mb(256)); assert_eq!(cfg.quota.max_delay_duration, ReadableDuration::millis(250)); assert!(cfg.overload.enable); assert_eq!(cfg.overload.max_keys, 100_000_000); assert_eq!(cfg.overload.max_data_size, ReadableSize::gb(8)); let encoded = toml::to_string_pretty(&cfg).unwrap(); assert!(encoded.contains("[quota]")); assert!(encoded.contains("[overload]")); } struct DummyCfgMgr; impl ConfigManager for DummyCfgMgr { fn dispatch(&mut self, _change: ConfigChange) -> Result<(), Box> { Ok(()) } } #[test] fn test_change_server_config() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.validate().unwrap(); let cfg_controller = ConfigController::new(cfg.clone()); let version_tracker = Arc::new(VersionTrack::new(cfg.server.clone())); cfg_controller.register( Module::Server, Box::new(ServerConfigManager::new( version_tracker.clone(), ResourceQuota::new(None), Box::new(DummyCfgMgr), )), ); let check_cfg = |cfg: &TikvConfig| { assert_eq!(&cfg_controller.get_current(), cfg); assert_eq!(&*version_tracker.value(), &cfg.server); }; cfg_controller .update_config("server.max-grpc-send-msg-len", "10000") .unwrap(); cfg.server.max_grpc_send_msg_len = 10000; check_cfg(&cfg); cfg_controller .update_config("server.raft-msg-max-batch-size", "32") .unwrap(); cfg.server.raft_msg_max_batch_size = 32; assert_eq!(cfg_controller.get_current(), cfg); check_cfg(&cfg); cfg_controller .update_config("server.dfs-file-read-concurrency", "23") .unwrap(); cfg.server.dfs_file_read_concurrency = 23; check_cfg(&cfg); cfg_controller .update_config("server.dfs-file-read-concurrency", "0") .unwrap_err(); check_cfg(&cfg); } #[test] fn test_change_coprocessor_endpoint_config() { use crossbeam::channel; pub struct TestConfigManager(channel::Sender); impl ConfigManager for TestConfigManager { fn dispatch(&mut self, change: ConfigChange) -> online_config::Result<()> { self.0.send(change).unwrap(); Ok(()) } } let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.validate().unwrap(); let cfg_controller = ConfigController::new(cfg.clone()); let version_tracker = Arc::new(VersionTrack::new(cfg.server.clone())); let (cop_tx, cop_rx) = channel::unbounded(); let cop_manager = TestConfigManager(cop_tx); cfg_controller.register( Module::Server, Box::new(ServerConfigManager::new( version_tracker.clone(), ResourceQuota::new(None), Box::new(cop_manager), )), ); cfg_controller .update_config("server.end-point-memory-quota", "32MB") .unwrap(); let mut change = cop_rx.try_recv().unwrap(); let quota = change.remove("end_point_memory_quota").unwrap(); let cap: ReadableSize = quota.into(); assert_eq!(cap, ReadableSize::mb(32)); cfg.server.end_point_memory_quota = ReadableSize::mb(32); assert_eq!(cfg_controller.get_current(), cfg); } #[test] fn test_config_adjustments_are_idempotent() { // Repeated compatibility and CPU-limit adjustments should have the same // effect as running them once. let mut c = TikvConfig::default(); let mut cfg = c.clone(); c.compatible_adjust(); c.adjust_cpu_limited_worker_counts(); c.validate().unwrap(); for _ in 0..10 { cfg.compatible_adjust(); cfg.adjust_cpu_limited_worker_counts(); cfg.validate().unwrap(); assert_eq!(c, cfg); } } #[test] fn test_split_config_is_validated_on_startup() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.split.qps_threshold = 10; cfg.split.sample_num = 10; // Legacy cross-field combinations remain accepted while the feature is // disabled. cfg.validate().unwrap(); cfg.split.enable_load_based_split = true; let err = cfg.validate().unwrap_err().to_string(); assert!(err.contains("sample_num should be less than qps_threshold")); cfg.split.qps_threshold = 0; cfg.validate().unwrap(); } #[test] fn test_split_config_is_validated_on_online_update() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.split.qps_threshold = 10; cfg.split.sample_num = 1; cfg.validate().unwrap(); let cfg_controller = ConfigController::new(cfg.clone()); cfg_controller .update_config("split.sample-num", "10") .unwrap(); assert_eq!(cfg_controller.get_current().split.sample_num, 10); let err = cfg_controller .update_config("split.enable-load-based-split", "true") .unwrap_err() .to_string(); assert!(err.contains("sample_num should be less than qps_threshold")); assert!(!cfg_controller.get_current().split.enable_load_based_split); cfg_controller .update_config("split.qps-threshold", "0") .unwrap(); assert_eq!(cfg_controller.get_current().split.qps_threshold, 0); cfg_controller .update_config("split.enable-load-based-split", "true") .unwrap(); assert!(cfg_controller.get_current().split.enable_load_based_split); } #[test] fn test_adjust_cpu_limited_worker_counts() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); let cpu_quota = SysQuota::cpu_cores_quota(); let read_pool_limit = cmp::max(UNIFIED_READPOOL_MIN_CONCURRENCY, cpu_quota as usize); let scheduler_pool_limit = cmp::max( crate::storage::config::SCHED_MIN_WORKER_POOL_SIZE, cpu_quota as usize, ); let rfstore_pool_limit = cmp::max((cpu_quota.ceil() as usize).div_ceil(2), 2); let aux_worker_limit = cpu_quota.ceil() as usize; cfg.readpool.unified.max_thread_count = read_pool_limit + 1; cfg.storage.scheduler_worker_pool_size = scheduler_pool_limit + 1; cfg.storage.scheduler_background_worker_pool_size = scheduler_pool_limit + 1; cfg.raft_store.apply_batch_system.pool_size = rfstore_pool_limit + 1; cfg.raft_store.store_batch_system.pool_size = rfstore_pool_limit + 1; cfg.raft_store.aux_worker_count = aux_worker_limit + 1; cfg.validate().unwrap_err(); cfg.adjust_cpu_limited_worker_counts(); cfg.validate().unwrap(); assert_eq!(cfg.readpool.unified.max_thread_count, read_pool_limit); assert_eq!(cfg.storage.scheduler_worker_pool_size, scheduler_pool_limit); assert_eq!( cfg.storage.scheduler_background_worker_pool_size, scheduler_pool_limit ); assert_eq!( cfg.raft_store.apply_batch_system.pool_size, rfstore_pool_limit ); assert_eq!( cfg.raft_store.store_batch_system.pool_size, rfstore_pool_limit ); assert_eq!(cfg.raft_store.aux_worker_count, aux_worker_limit); } #[test] fn test_online_config_adjusts_cpu_limited_worker_counts() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.adjust_cpu_limited_worker_counts(); cfg.validate().unwrap(); let cpu_quota = SysQuota::cpu_cores_quota(); let read_pool_limit = cmp::max(UNIFIED_READPOOL_MIN_CONCURRENCY, cpu_quota as usize); let scheduler_pool_limit = cmp::max( crate::storage::config::SCHED_MIN_WORKER_POOL_SIZE, cpu_quota as usize, ); let rfstore_pool_limit = cmp::max((cpu_quota.ceil() as usize).div_ceil(2), 2); let aux_worker_limit = cpu_quota.ceil() as usize; let mut change = HashMap::new(); change.insert( "readpool.unified.max-thread-count".to_owned(), (read_pool_limit + 1).to_string(), ); change.insert( "storage.scheduler-worker-pool-size".to_owned(), (scheduler_pool_limit + 1).to_string(), ); change.insert( "raftstore.apply-pool-size".to_owned(), (rfstore_pool_limit + 1).to_string(), ); change.insert( "raftstore.store-pool-size".to_owned(), (rfstore_pool_limit + 1).to_string(), ); change.insert( "raftstore.aux-worker-count".to_owned(), (aux_worker_limit + 1).to_string(), ); let controller = ConfigController::new(cfg); controller.update(change).unwrap(); let current = controller.get_current(); assert_eq!(current.readpool.unified.max_thread_count, read_pool_limit); assert_eq!( current.storage.scheduler_worker_pool_size, scheduler_pool_limit ); assert_eq!( current.raft_store.apply_batch_system.pool_size, rfstore_pool_limit ); assert_eq!( current.raft_store.store_batch_system.pool_size, rfstore_pool_limit ); assert_eq!(current.raft_store.aux_worker_count, aux_worker_limit); } #[test] fn test_readpool_compatible_adjust_config() { let content = r#" [readpool.storage] [readpool.coprocessor] "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.compatible_adjust(); assert_eq!(cfg.readpool.storage.use_unified_pool, Some(true)); assert_eq!(cfg.readpool.coprocessor.use_unified_pool, Some(true)); let content = r#" [readpool.storage] stack-size = "1MB" [readpool.coprocessor] normal-concurrency = 1 "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.compatible_adjust(); assert_eq!(cfg.readpool.storage.use_unified_pool, Some(false)); assert_eq!(cfg.readpool.coprocessor.use_unified_pool, Some(false)); } #[test] fn test_deprecated_ia_kv_size_discount_is_accepted_but_not_serialized() { let mut temp_config_file = tempfile::NamedTempFile::new().unwrap(); let temp_config_writer = temp_config_file.as_file_mut(); temp_config_writer .write_all( br#" [raftstore] ia-kv-size-discount = 0.5 "#, ) .unwrap(); temp_config_writer.sync_data().unwrap(); let mut unrecognized_keys = Vec::new(); let mut cfg = TikvConfig::from_file(temp_config_file.path(), Some(&mut unrecognized_keys)).unwrap(); assert_eq!(unrecognized_keys, Vec::::new()); assert_eq!(cfg.raft_store.ia_kv_size_discount, 0.5); let serialized = toml::to_string_pretty(&cfg).unwrap(); assert!(!serialized.contains("ia-kv-size-discount")); cfg.compatible_adjust(); assert_eq!(cfg.raft_store.ia_kv_size_discount, 0.0); } #[test] fn test_unrecognized_config_keys() { let mut temp_config_file = tempfile::NamedTempFile::new().unwrap(); let temp_config_writer = temp_config_file.as_file_mut(); temp_config_writer .write_all( br#" log-level = "debug" log-fmt = "json" [readpool.unified] min-threads-count = 5 stack-size = "20MB" [import] num_threads = 4 [gcc] batch-keys = 1024 [[security.encryption.master-keys]] type = "file" "#, ) .unwrap(); temp_config_writer.sync_data().unwrap(); let mut unrecognized_keys = Vec::new(); let _ = TikvConfig::from_file(temp_config_file.path(), Some(&mut unrecognized_keys)); assert_eq!( unrecognized_keys, vec![ "log-fmt".to_owned(), "readpool.unified.min-threads-count".to_owned(), "import.num_threads".to_owned(), "gcc".to_owned(), "security.encryption.master-keys".to_owned(), ], ); } #[test] #[ignore = "CSE incompatibility; see #6158"] fn test_validate_tikv_config() { let mut cfg = TikvConfig::default(); cfg.validate().unwrap(); let default_region_split_check_diff = cfg.raft_store.region_split_check_diff().0; cfg.raft_store.region_split_check_diff = Some(ReadableSize(cfg.raft_store.region_split_check_diff().0 + 1)); cfg.validate().unwrap(); assert_eq!( cfg.raft_store.region_split_check_diff().0, default_region_split_check_diff + 1 ); // Test validating memory_usage_limit when it's greater than max. cfg.memory_usage_limit = Some(ReadableSize(SysQuota::memory_limit_in_bytes() * 2)); cfg.validate().unwrap_err(); // Test memory_usage_limit is based on block cache size if it's not configured. cfg.memory_usage_limit = None; cfg.storage.block_cache.capacity = Some(ReadableSize(3 * GIB)); cfg.validate().unwrap(); assert_eq!(cfg.memory_usage_limit.unwrap(), ReadableSize(5 * GIB)); // Test memory_usage_limit will fallback to system memory capacity with huge // block cache. cfg.memory_usage_limit = None; let system = SysQuota::memory_limit_in_bytes(); cfg.storage.block_cache.capacity = Some(ReadableSize(system * 3 / 4)); cfg.validate().unwrap(); assert_eq!(cfg.memory_usage_limit.unwrap(), ReadableSize(system)); } #[test] fn test_keyspace_fairness_rejects_merged_scheduler_pool() { let mut cfg = TikvConfig::default(); cfg.readpool.storage.use_unified_pool = Some(true); cfg.readpool.coprocessor.use_unified_pool = Some(true); cfg.readpool.unified.use_tokio = false; cfg.readpool.unified.keyspace_fairness.enabled = true; cfg.resource_control.enabled = false; cfg.storage.use_separated_scheduler_pool = false; let err = cfg.validate().unwrap_err().to_string(); assert!(err.contains("scheduler pool is merged with read pool")); cfg.storage.use_separated_scheduler_pool = true; cfg.validate().unwrap(); let mut legacy_gate_cfg = TikvConfig::default(); legacy_gate_cfg.readpool.storage.use_unified_pool = Some(false); legacy_gate_cfg.readpool.coprocessor.use_unified_pool = Some(false); legacy_gate_cfg.readpool.unified.use_tokio = false; legacy_gate_cfg.readpool.unified.keyspace_fairness.enabled = true; legacy_gate_cfg.resource_control.enabled = false; legacy_gate_cfg.storage.use_separated_scheduler_pool = false; let err = legacy_gate_cfg.validate().unwrap_err().to_string(); assert!(err.contains("scheduler pool is merged with read pool")); } #[test] fn test_memory_profile_auto_dump_config_validate() { let mut cfg = TikvConfig::default(); cfg.memory_profile_auto_dump.usage_threshold = 0.0; assert!(cfg.validate().is_err()); let mut cfg = TikvConfig::default(); cfg.memory_profile_auto_dump.usage_threshold = 0.6; cfg.memory_profile_auto_dump.rearm_usage_threshold = 0.6; assert!(cfg.validate().is_err()); let mut cfg = TikvConfig::default(); cfg.memory_profile_auto_dump.usage_threshold = 0.6; cfg.memory_profile_auto_dump.rearm_usage_threshold = 0.5; cfg.validate().unwrap(); } #[test] fn test_validate_tikv_wal_config() { let tmp_dir = tempfile::Builder::new().tempdir().unwrap(); let tmp_path = tmp_dir.path().to_path_buf(); macro_rules! tmp_path_string_generate { ($base:expr, $($sub:expr),+) => {{ let mut path: ::std::path::PathBuf = $base.clone(); $( path.push($sub); )* String::from(path.to_str().unwrap()) }} } { let mut cfg = TikvConfig::default(); cfg.validate().unwrap(); } { let mut cfg = TikvConfig::default(); cfg.storage.data_dir = tmp_path_string_generate!(tmp_path, "data"); cfg.raft_store.raftdb_path = tmp_path_string_generate!(tmp_path, "data", "db"); cfg.validate().unwrap_err(); } { let mut cfg = TikvConfig::default(); cfg.storage.data_dir = tmp_path_string_generate!(tmp_path, "data", "kvdb"); cfg.raft_store.raftdb_path = tmp_path_string_generate!(tmp_path, "data", "raftdb", "db"); cfg.rocksdb.wal_dir = tmp_path_string_generate!(tmp_path, "data", "raftdb", "db"); cfg.validate().unwrap_err(); } { let mut cfg = TikvConfig::default(); cfg.storage.data_dir = tmp_path_string_generate!(tmp_path, "data", "kvdb"); cfg.raft_store.raftdb_path = tmp_path_string_generate!(tmp_path, "data", "raftdb", "db"); cfg.raftdb.wal_dir = tmp_path_string_generate!(tmp_path, "data", "kvdb", "db"); cfg.validate().unwrap_err(); } { let mut cfg = TikvConfig::default(); cfg.rocksdb.wal_dir = tmp_path_string_generate!(tmp_path, "data", "wal"); cfg.raftdb.wal_dir = tmp_path_string_generate!(tmp_path, "data", "wal"); cfg.validate().unwrap_err(); } { let mut cfg = TikvConfig::default(); cfg.storage.data_dir = tmp_path_string_generate!(tmp_path, "data", "kvdb"); cfg.raft_store.raftdb_path = tmp_path_string_generate!(tmp_path, "data", "raftdb", "db"); cfg.rocksdb.wal_dir = tmp_path_string_generate!(tmp_path, "data", "kvdb", "db"); cfg.raftdb.wal_dir = tmp_path_string_generate!(tmp_path, "data", "raftdb", "db"); cfg.validate().unwrap(); } } #[test] fn test_background_job_limits() { // cpu num = 1 assert_eq!( get_background_job_limits_impl( 1, // cpu_num &KVDB_DEFAULT_BACKGROUND_JOB_LIMITS ), BackgroundJobLimits { max_background_jobs: 2, max_background_flushes: 1, max_sub_compactions: 1, max_titan_background_gc: 1, } ); assert_eq!( get_background_job_limits_impl( 1, // cpu_num &RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS ), BackgroundJobLimits { max_background_jobs: 2, max_background_flushes: 1, max_sub_compactions: 1, max_titan_background_gc: 1, } ); // cpu num = 2 assert_eq!( get_background_job_limits_impl( 2, // cpu_num &KVDB_DEFAULT_BACKGROUND_JOB_LIMITS ), BackgroundJobLimits { max_background_jobs: 2, max_background_flushes: 1, max_sub_compactions: 1, max_titan_background_gc: 2, } ); assert_eq!( get_background_job_limits_impl( 2, // cpu_num &RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS ), BackgroundJobLimits { max_background_jobs: 2, max_background_flushes: 1, max_sub_compactions: 1, max_titan_background_gc: 2, } ); // cpu num = 4 assert_eq!( get_background_job_limits_impl( 4, // cpu_num &KVDB_DEFAULT_BACKGROUND_JOB_LIMITS ), BackgroundJobLimits { max_background_jobs: 3, max_background_flushes: 1, max_sub_compactions: 1, max_titan_background_gc: 4, } ); assert_eq!( get_background_job_limits_impl( 4, // cpu_num &RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS ), BackgroundJobLimits { max_background_jobs: 3, max_background_flushes: 1, max_sub_compactions: 1, max_titan_background_gc: 4, } ); // cpu num = 8 assert_eq!( get_background_job_limits_impl( 8, // cpu_num &KVDB_DEFAULT_BACKGROUND_JOB_LIMITS ), BackgroundJobLimits { max_background_jobs: 7, max_background_flushes: 2, max_sub_compactions: 3, max_titan_background_gc: 4, } ); assert_eq!( get_background_job_limits_impl( 8, // cpu_num &RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS ), RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS, ); // cpu num = 16 assert_eq!( get_background_job_limits_impl( 16, // cpu_num &KVDB_DEFAULT_BACKGROUND_JOB_LIMITS ), KVDB_DEFAULT_BACKGROUND_JOB_LIMITS, ); assert_eq!( get_background_job_limits_impl( 16, // cpu_num &RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS ), RAFTDB_DEFAULT_BACKGROUND_JOB_LIMITS, ); } #[test] fn test_missing_and_partial_raftstore_preserve_apply_pool_default() { let default_pool_size = TikvConfig::default() .raft_store .apply_batch_system .pool_size; let without_raftstore: TikvConfig = toml::from_str("").unwrap(); assert_eq!( without_raftstore.raft_store.apply_batch_system.pool_size, default_pool_size ); let partial_raftstore: TikvConfig = toml::from_str( r#" [raftstore] raft-log-gc-threshold = 99 "#, ) .unwrap(); assert_eq!(partial_raftstore.raft_store.raft_log_gc_threshold, 99); assert_eq!( partial_raftstore.raft_store.apply_batch_system.pool_size, default_pool_size ); } #[test] fn test_config_template_is_valid() { let template_config = std::include_str!("../etc/config-template.toml") .lines() .map(|l| l.strip_prefix('#').unwrap_or(l)) .join("\n"); let mut cfg: TikvConfig = toml::from_str(&template_config).unwrap(); cfg.validate().unwrap(); } #[test] #[ignore = "CSE incompatibility; see #6158"] fn test_config_template_no_superfluous_keys() { let template_config = std::include_str!("../etc/config-template.toml") .lines() .map(|l| l.strip_prefix('#').unwrap_or(l)) .join("\n"); let mut deserializer = toml::Deserializer::new(&template_config); let mut unrecognized_keys = Vec::new(); let _: TikvConfig = serde_ignored::deserialize(&mut deserializer, |key| { unrecognized_keys.push(key.to_string()) }) .unwrap(); // Don't use `is_empty()` so we see which keys are superfluous on failure. assert_eq!(unrecognized_keys, Vec::::new()); } #[test] #[ignore = "CSE incompatibility; see #6158"] fn test_config_template_matches_default() { let template_config = std::include_str!("../etc/config-template.toml") .lines() .map(|l| l.strip_prefix('#').unwrap_or(l)) .join("\n"); let mut cfg: TikvConfig = toml::from_str(&template_config).unwrap(); let mut default_cfg = TikvConfig::default(); // Some default values are computed based on the environment. // Because we can't set config values for these in `config-template.toml`, we // will handle them manually. cfg.readpool.unified.max_thread_count = default_cfg.readpool.unified.max_thread_count; cfg.readpool.storage.high_concurrency = default_cfg.readpool.storage.high_concurrency; cfg.readpool.storage.normal_concurrency = default_cfg.readpool.storage.normal_concurrency; cfg.readpool.storage.low_concurrency = default_cfg.readpool.storage.low_concurrency; cfg.readpool.coprocessor.high_concurrency = default_cfg.readpool.coprocessor.high_concurrency; cfg.readpool.coprocessor.normal_concurrency = default_cfg.readpool.coprocessor.normal_concurrency; cfg.readpool.coprocessor.low_concurrency = default_cfg.readpool.coprocessor.low_concurrency; cfg.server.grpc_memory_pool_quota = default_cfg.server.grpc_memory_pool_quota; cfg.server.grpc_concurrency = default_cfg.server.grpc_concurrency; cfg.server.grpc_raft_conn_num = default_cfg.server.grpc_raft_conn_num; cfg.server.background_thread_count = default_cfg.server.background_thread_count; cfg.server.end_point_max_concurrency = default_cfg.server.end_point_max_concurrency; cfg.storage.scheduler_worker_pool_size = default_cfg.storage.scheduler_worker_pool_size; cfg.raft_store.apply_batch_system.pool_size = default_cfg.raft_store.apply_batch_system.pool_size; cfg.rocksdb.max_background_jobs = default_cfg.rocksdb.max_background_jobs; cfg.rocksdb.max_background_flushes = default_cfg.rocksdb.max_background_flushes; cfg.rocksdb.max_sub_compactions = default_cfg.rocksdb.max_sub_compactions; cfg.rocksdb.titan.max_background_gc = default_cfg.rocksdb.titan.max_background_gc; cfg.raftdb.max_background_jobs = default_cfg.raftdb.max_background_jobs; cfg.raftdb.max_background_flushes = default_cfg.raftdb.max_background_flushes; cfg.raftdb.max_sub_compactions = default_cfg.raftdb.max_sub_compactions; cfg.raftdb.titan.max_background_gc = default_cfg.raftdb.titan.max_background_gc; cfg.backup.num_threads = default_cfg.backup.num_threads; cfg.backup_stream.num_threads = default_cfg.backup_stream.num_threads; // There is another set of config values that we can't directly compare: // When the default values are `None`, but are then resolved to `Some(_)` later // on. default_cfg.readpool.storage.adjust_use_unified_pool(); default_cfg.readpool.coprocessor.adjust_use_unified_pool(); default_cfg.security.redact_info_log = Some(false); default_cfg.coprocessor.region_max_size = Some(default_cfg.coprocessor.region_max_size()); default_cfg.coprocessor.region_max_keys = Some(default_cfg.coprocessor.region_max_keys()); default_cfg.coprocessor.region_split_keys = Some(default_cfg.coprocessor.region_split_keys()); default_cfg.raft_store.raft_log_gc_size_limit = Some(default_cfg.coprocessor.region_split_size * 3 / 4); default_cfg.raft_store.raft_log_gc_count_limit = Some(default_cfg.coprocessor.region_split_size * 3 / 4 / ReadableSize::kb(1)); default_cfg.raft_store.region_split_check_diff = Some(default_cfg.coprocessor.region_split_size / 16); // Other special cases. cfg.pd.retry_max_count = default_cfg.pd.retry_max_count; // Both -1 and isize::MAX are the same. cfg.storage.block_cache.capacity = None; // Either `None` and a value is computed or `Some(_)` fixed value. cfg.memory_usage_limit = None; cfg.rocksdb.defaultcf.level0_slowdown_writes_trigger = None; cfg.rocksdb.defaultcf.level0_stop_writes_trigger = None; cfg.rocksdb.defaultcf.soft_pending_compaction_bytes_limit = None; cfg.rocksdb.defaultcf.hard_pending_compaction_bytes_limit = None; cfg.rocksdb.writecf.level0_slowdown_writes_trigger = None; cfg.rocksdb.writecf.level0_stop_writes_trigger = None; cfg.rocksdb.writecf.soft_pending_compaction_bytes_limit = None; cfg.rocksdb.writecf.hard_pending_compaction_bytes_limit = None; cfg.rocksdb.lockcf.level0_slowdown_writes_trigger = None; cfg.rocksdb.lockcf.level0_stop_writes_trigger = None; cfg.rocksdb.lockcf.soft_pending_compaction_bytes_limit = None; cfg.rocksdb.lockcf.hard_pending_compaction_bytes_limit = None; cfg.rocksdb.raftcf.level0_slowdown_writes_trigger = None; cfg.rocksdb.raftcf.level0_stop_writes_trigger = None; cfg.rocksdb.raftcf.soft_pending_compaction_bytes_limit = None; cfg.rocksdb.raftcf.hard_pending_compaction_bytes_limit = None; cfg.raftdb.defaultcf.level0_slowdown_writes_trigger = None; cfg.raftdb.defaultcf.level0_stop_writes_trigger = None; cfg.raftdb.defaultcf.soft_pending_compaction_bytes_limit = None; cfg.raftdb.defaultcf.hard_pending_compaction_bytes_limit = None; assert_eq!(cfg, default_cfg); } #[test] #[ignore = "CSE incompatibility; see #6158"] fn test_compatibility_with_old_config_template() { let mut buf = Vec::new(); let resp = reqwest::blocking::get( "https://raw.githubusercontent.com/tikv/tikv/master/etc/config-template.toml", ); match resp { Ok(mut resp) => { std::io::copy(&mut resp, &mut buf).expect("failed to copy content"); let template_config = std::str::from_utf8(&buf) .unwrap() .lines() .map(|l| l.strip_prefix('#').unwrap_or(l)) .join("\n"); let _: TikvConfig = toml::from_str(&template_config).unwrap(); } Err(e) => { if e.is_timeout() { println!("warn: fail to download latest config template due to timeout"); } else { panic!("fail to download latest config template"); } } } } #[test] fn test_cdc() { let content = r#" [cdc] "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); // old-value-cache-size is deprecated, 0 must not report error. let content = r#" [cdc] old-value-cache-size = 0 "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); let content = r#" [cdc] min-ts-interval = "0s" "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); let content = r#" [cdc] incremental-scan-threads = 0 "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); let content = r#" [cdc] incremental-scan-concurrency = 0 "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); let content = r#" [cdc] incremental-scan-concurrency = 1 incremental-scan-threads = 2 "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); } #[test] fn test_module_from_str() { let cases = vec![ ("readpool", Module::Readpool), ("server", Module::Server), ("metric", Module::Metric), ("raft_store", Module::Raftstore), ("coprocessor", Module::Coprocessor), ("pd", Module::Pd), ("split", Module::Split), ("rocksdb", Module::Rocksdb), ("raft_engine", Module::RaftEngine), ("storage", Module::Storage), ("security", Module::Security), ("import", Module::Import), ("backup", Module::Backup), ("backup_stream", Module::BackupStream), ("pessimistic_txn", Module::PessimisticTxn), ("gc", Module::Gc), ("cdc", Module::Cdc), ("resolved_ts", Module::ResolvedTs), ("resource_metering", Module::ResourceMetering), ("kvengine", Module::Kvengine), ("unknown", Module::Unknown("unknown".to_string())), ]; for (name, module) in cases { assert_eq!(Module::from(name), module); } } #[test] fn test_numeric_enum_serializing() { let normal_string_config = r#" compaction-style = 1 "#; let config: DefaultCfConfig = toml::from_str(normal_string_config).unwrap(); assert_eq!(config.compaction_style, DBCompactionStyle::Universal); // Test if we support string value let normal_string_config = r#" compaction-style = "universal" "#; let config: DefaultCfConfig = toml::from_str(normal_string_config).unwrap(); assert_eq!(config.compaction_style, DBCompactionStyle::Universal); assert!( toml::to_string(&config) .unwrap() .contains("compaction-style = 1") ); let bad_string_config = r#" compaction-style = "level1" "#; let r = panic_hook::recover_safe(|| { let _: DefaultCfConfig = toml::from_str(bad_string_config).unwrap(); }); r.unwrap_err(); let bad_string_config = r#" compaction-style = 4 "#; let r = panic_hook::recover_safe(|| { let _: DefaultCfConfig = toml::from_str(bad_string_config).unwrap(); }); r.unwrap_err(); // rate-limiter-mode default values is 2 let config_str = r#" rate-limiter-mode = 1 "#; let config: DbConfig = toml::from_str(config_str).unwrap(); assert_eq!(config.rate_limiter_mode, DBRateLimiterMode::ReadOnly); assert!( toml::to_string(&config) .unwrap() .contains("rate-limiter-mode = 1") ); } #[test] fn test_serde_to_online_config() { let cases = vec![ ( "raftstore.store_pool_size", "raft_store.store_batch_system.pool_size", ), ( "raftstore.store-pool-size", "raft_store.store_batch_system.pool_size", ), ( "raftstore.store_max_batch_size", "raft_store.store_batch_system.max_batch_size", ), ( "raftstore.store-max-batch-size", "raft_store.store_batch_system.max_batch_size", ), ( "raftstore.apply_pool_size", "raft_store.apply_batch_system.pool_size", ), ( "raftstore.apply-pool-size", "raft_store.apply_batch_system.pool_size", ), ( "raftstore.apply_max_batch_size", "raft_store.apply_batch_system.max_batch_size", ), ( "raftstore.apply-max-batch-size", "raft_store.apply_batch_system.max_batch_size", ), ( "raftstore.store_io_pool_size", "raft_store.store_io_pool_size", ), ( "raftstore.store-io-pool-size", "raft_store.store_io_pool_size", ), ( "raftstore.apply_yield_duration", "raft_store.apply_yield_duration", ), ( "raftstore.apply-yield-duration", "raft_store.apply_yield_duration", ), ( "raftstore.raft_store_max_leader_lease", "raft_store.raft_store_max_leader_lease", ), ( "raftstore.txn-file-lock-active-resolve.threshold", "raft_store.txn_file_lock_active_resolve.threshold", ), ( "raftstore.txn-file-lock-active-resolve.concurrency", "raft_store.txn_file_lock_active_resolve.concurrency", ), ]; for (name, res) in cases { assert_eq!(serde_to_online_config(name.into()).as_str(), res); } } #[test] fn test_flow_control_override() { let content = r#" [storage.flow-control] enable = true l0-files-threshold = 77 soft-pending-compaction-bytes-limit = "777GB" "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); assert_eq!( cfg.rocksdb.defaultcf.level0_slowdown_writes_trigger, Some(77) ); assert_eq!( cfg.rocksdb.defaultcf.soft_pending_compaction_bytes_limit, Some(ReadableSize::gb(777)) ); // Override with default values if flow control is disabled. let content = r#" [storage.flow-control] enable = false l0-files-threshold = 77 soft-pending-compaction-bytes-limit = "777GB" [rocksdb.defaultcf] level0-slowdown-writes-trigger = 888 soft-pending-compaction-bytes-limit = "888GB" [rocksdb.writecf] "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); assert_eq!( cfg.rocksdb.defaultcf.level0_slowdown_writes_trigger, Some(888) ); assert_eq!( cfg.rocksdb.defaultcf.soft_pending_compaction_bytes_limit, Some(ReadableSize::gb(888)) ); matches!(cfg.rocksdb.writecf.level0_slowdown_writes_trigger, Some(v) if v != 77); matches!(cfg.rocksdb.writecf.soft_pending_compaction_bytes_limit, Some(v) if v != ReadableSize::gb(777)); // Do not override when RocksDB configurations are specified. let content = r#" [storage.flow-control] enable = true l0-files-threshold = 77 soft-pending-compaction-bytes-limit = "777GB" [rocksdb.defaultcf] level0-slowdown-writes-trigger = 66 soft-pending-compaction-bytes-limit = "666GB" "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); assert_eq!( cfg.rocksdb.defaultcf.level0_slowdown_writes_trigger, Some(66) ); assert_eq!( cfg.rocksdb.defaultcf.soft_pending_compaction_bytes_limit, Some(ReadableSize::gb(666)) ); // Cannot specify larger configurations for RocksDB. let content = r#" [storage.flow-control] enable = true l0-files-threshold = 1 soft-pending-compaction-bytes-limit = "1GB" [rocksdb.defaultcf] level0-slowdown-writes-trigger = 88 soft-pending-compaction-bytes-limit = "888GB" "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); assert_eq!( cfg.rocksdb.defaultcf.level0_slowdown_writes_trigger, Some(1) ); assert_eq!( cfg.rocksdb.defaultcf.soft_pending_compaction_bytes_limit, Some(ReadableSize::gb(1)) ); } #[test] fn test_kv_engine() { let content = r#" [kvengine] enable-crossbeam-memtable = true per-keyspace-configs = [ {keyspace = 1, split-size-factor = 1.5}, {keyspace = 2, enable-blob = true}, ] "#; let mut cfg: TikvConfig = toml::from_str(content).unwrap(); cfg.validate().unwrap(); assert!(cfg.kvengine.enable_crossbeam_memtable); let per_keyspace_configs = cfg.kvengine.get_per_keyspace_configs(); assert_eq!(per_keyspace_configs.len(), 2); assert_eq!(per_keyspace_configs[&1].keyspace, 1); assert_eq!(per_keyspace_configs[&1].split_size_factor, 1.5); assert!(!per_keyspace_configs[&1].enable_blob); assert_eq!(per_keyspace_configs[&2].keyspace, 2); assert_eq!(per_keyspace_configs[&2].split_size_factor, 1.0); assert!(per_keyspace_configs[&2].enable_blob); } #[test] fn test_large_raft_entry_requires_crossbeam_memtable() { let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.kvengine.enable_crossbeam_memtable = false; cfg.raft_store.raft_entry_max_size = ReadableSize::mb(17); let err = cfg.validate().unwrap_err(); assert!( err.to_string() .contains("kvengine.enable-crossbeam-memtable = true") ); let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.kvengine.enable_crossbeam_memtable = false; cfg.raft_store.raft_entry_max_size = ReadableSize::mb(16); cfg.validate().unwrap(); let controller = ConfigController::new(cfg); assert!( controller .update_config("raftstore.raft-entry-max-size", "17MiB") .is_err() ); let (mut cfg, _dir) = TikvConfig::with_tmp().unwrap(); cfg.kvengine.enable_crossbeam_memtable = true; cfg.validate().unwrap(); let controller = ConfigController::new(cfg); controller .update_config("raftstore.raft-entry-max-size", "128MiB") .unwrap(); assert_eq!( controller.get_current().raft_store.raft_entry_max_size, ReadableSize::mb(128) ); } }