// Copyright 2023 PingCAP, Inc. use std::{ collections::HashSet, io::Write as _, path::Path, sync::{Arc, Mutex, RwLock, atomic::Ordering}, time::Duration, }; use api_version::ApiV2; use cloud_encryption::KeyspaceEncryptionConfig; use cloud_worker::native_br::{NativeBrConfig, test_utils::NativeBrSvcClient}; use futures::executor::block_on; use kvengine::{ dfs::{self, DFSConfig, DFSConnOptions, FileType, new_dfs_from_config}, ia::util::IaConfig, metrics::{ENGINE_IA_SYNC_READ_COUNTER, ENGINE_REMOTE_COMPACT_EXCEED_MEMORY_LIMIT_COUNTER}, table::{ ChecksumType, blobtable::builder::BlobTableBuildOptions, schema_file::build_schema_file, }, }; use load_data::task::LoadDataConfig; use native_br::{backup, restore::RestoreConfig}; use pd_client::PdClient; use rand::prelude::*; use rfengine::RFENGINE_DFS_WORKER_BECOME_UNHEALTHY_COUNTER; use schema::schema::{StorageClass, StorageClassSpec}; use security::{SecurityConfig, SecurityManager}; use test_cloud_server::{ IA_DISK_CAP_DEF, IA_FREQ_UPDATE_INTERVAL_DEF, IA_MEM_CAP_DEF, ServerCluster, ServerClusterBuilder, TikvWorkerOptions, client::ClusterClientOptions, keyspace::make_row_key, oss::{ObjectStorageService, prepare_builtin_dfs, prepare_dfs}, tidb::TidbCluster, util::broadcast_schema_file_request_and_check, }; use test_pd_client::{PdClientExt, PdWrapper, TestPdControlServer}; use tikv::server::config::GrpcCompressionType; use tikv_util::{ config::{ReadableDuration, ReadableSize}, info, sys::SysQuota, time::Instant, warn, }; use tokio::runtime::Runtime; use txn_types::Key; use crate::{ TikvConfig, test_drop_table::*, test_load_data::*, test_native_br::*, test_txn_file::*, *, }; const MAX_IN_MEM_SIZE: usize = 10 * 1024; // 10KiB const FLUSH_BATCH_SIZE: usize = 4 * 1024; // 4KiB const INITIAL_KEYSPACE_COUNT: usize = 10; const ENABLE_BLOB_RATIO: f64 = 0.4; const BIG_REGION_SIZE_KEYSPACE_COUNT: usize = 2; const BIG_REGION_SIZE_FACTOR_OPTIONS: &[f64] = &[2.0, 3.0, 4.0]; const INITIAL_TABLE_COUNT: usize = 3; const TABLE_SCHEMA_ENABLE_RATIO: f64 = 0.5; const NODES_COUNT: usize = 4; const TIKV_WORKERS_COUNT: usize = 2; const RESTORE_CONCURRENCY: usize = 2; const LOAD_DATA_CONCURRENCY: usize = 2; const PERIODIC_BACKUP_INTERVAL: Duration = Duration::from_secs(4); const BACKUP_BATCH_INTERVAL: Duration = Duration::from_secs(2); const ON_DEMAND_BACKUP_INTERVAL_MIN: Duration = Duration::from_secs(10); const ON_DEMAND_BACKUP_INTERVAL_MAX: Duration = Duration::from_secs(25); const BACKUP_OPERATION_TIMEOUT: Duration = Duration::from_secs(30); const ONLINE_SNAPSHOT_RESTORE_RATIO_ENV: &str = "ONLINE_SNAPSHOT_RESTORE_RATIO"; const ONLINE_SNAPSHOT_RESTORE_RATIO_DEFAULT: f64 = 0.1; const KV_TARGET_FILE_SIZE: ReadableSize = ReadableSize::kb(16); const REGION_BUCKET_SIZE: ReadableSize = ReadableSize::kb(64); const COP_BLOCK_CACHE_SIZE: ReadableSize = ReadableSize::mb(4); // Small size to make eviction more frequent. #[test] fn test_random_all() { test_random_all_impl(false); } #[test] fn test_random_builtin_all() { test_random_all_impl(true); } fn test_random_all_impl(use_builtin_dfs: bool) { init_logger(); let prepare_time = Instant::now_coarse(); let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() .worker_threads(4) .thread_name("random-workload") .build() .unwrap(); let _guard = runtime.enter(); let mut switches = Switches::from_env(); if use_builtin_dfs { switches.ia_table_ratio = 0 as f64; } info!("switches: {:?}", switches); // Prepare. let (_temp_dir, oss, dfs_config) = if use_builtin_dfs { prepare_builtin_dfs("builtin_") } else { prepare_dfs("oss_") }; let security_conf = new_security_config(); let (mut cluster, _pd_control_server) = prepare_cluster( &runtime, dfs_config.clone(), &security_conf, NODES_COUNT, INITIAL_KEYSPACE_COUNT, &switches, &oss, use_builtin_dfs, ); let pd_client = cluster.get_pd_client(); let keyspace_manager = cluster.keyspace_manager().clone(); let load_data_config = { let tikv_config = cluster.get_node_config(cluster.get_nodes()[0]); let region_size = tikv_config.coprocessor.region_split_size.0 as usize; LoadDataConfig { kvpairs_worker_num: 2, building_worker_num: 2, max_in_mem_size: MAX_IN_MEM_SIZE, flush_batch_size: FLUSH_BATCH_SIZE, block_size: tikv_config.rocksdb.writecf.block_size.0 as usize, sst_file_size: tikv_config.rocksdb.writecf.target_file_size_base.0 as usize, region_size, coarse_split_size: region_size * 4, enable_checkpoint: false, rg_config: None, checksum_type: ChecksumType::Crc32, ..Default::default() } }; // Set the first service safe point for backup before starting workloads. let ts = block_on(pd_client.get_tso()).unwrap(); backup::update_service_safe_point(pd_client.as_ref(), ts.into_inner()).unwrap(); let dfs_conf = dfs_config.clone(); let s3fs = new_dfs_from_config(dfs_conf); // Start workloads & schedulers. let running = Running::new_start(); // Region move/merge depends on stable quorum. Pause those schedulers while a // store is being restarted to avoid false positives from transient loss of a // voter during the restart window. let schedule_lock = Arc::new(RwLock::new(())); let mut handles = vec![ spawn_merge(cluster.new_scheduler(), true, schedule_lock.clone()), spawn_transfer(cluster.new_scheduler()), spawn_move(cluster.new_scheduler(), schedule_lock.clone()), ]; if !use_builtin_dfs { handles.push(spawn_create_keyspace( cluster.get_pd_client(), keyspace_manager.clone(), s3fs.clone(), INITIAL_TABLE_COUNT, TABLE_SCHEMA_ENABLE_RATIO, switches.enable_cmek, TIMEOUT, )) } handles.push(spawn_major_compact( cluster.get_pd_client(), keyspace_manager.clone(), TIMEOUT, )); let restore_config = RestoreConfig { dfs: dfs_config.clone(), security: security_conf.clone(), timeout_wait_flush: ReadableDuration::minutes(3), /* Flush is slow when OSS has low rate * limit. */ timeout_restore_snapshot: ReadableDuration::secs(30), timeout_fetch_wal: ReadableDuration::secs(10), tolerate_err: 1, strict_tolerate: true, max_retry: 20, lower_memory: switches.restore_lower_memory, ..Default::default() }; if !use_builtin_dfs { let object_cache = cluster.create_object_cache_randomly(); let limiter = cluster.create_restore_limiter_randomly(runtime.handle().clone()); let active_source_restores = Arc::new(Mutex::new(HashSet::new())); for _ in 0..RESTORE_CONCURRENCY { handles.push(spawn_restore_keyspace( cluster.get_pd_client(), None, runtime.block_on(cluster.new_keyspace_client()), restore_config.clone(), keyspace_manager.clone(), s3fs.clone(), switches.enable_oss_chaos, object_cache.clone(), limiter.clone(), switches.enable_cmek, TIMEOUT, cluster.tikv_worker_endpoints(), pd_client.get_security_mgr(), switches.online_snapshot_restore_ratio, active_source_restores.clone(), )); } let load_data_task_timeout = Duration::from_secs(env_param("LOAD_DATA_TASK_TIMEOUT_SEC", 90)); for _i in 0..LOAD_DATA_CONCURRENCY { let load_data_config = load_data_config.clone(); handles.push(spawn_load_data( cluster.get_pd_client(), runtime.block_on(cluster.new_keyspace_client()), dfs_config.clone(), security_conf.clone(), load_data_config.clone(), keyspace_manager.clone(), load_data_task_timeout, Duration::from_secs(15), TIMEOUT, )); } } let mut async_handles = vec![]; if !use_builtin_dfs { let backup_config = backup::BackupConfig { dfs: dfs_config.clone(), backup_delay: ReadableDuration::secs(1), tolerate_err: 1, skip_keyspace_meta: true, ..Default::default() }; let mut br_client = NativeBrSvcClient::new( pd_client.get_cluster_id().unwrap(), vec![cluster.tikv_api_endpoint().to_owned()], Arc::new(SecurityManager::new(&security_conf).unwrap()), ) .unwrap(); br_client.set_timeout(BACKUP_OPERATION_TIMEOUT, BACKUP_OPERATION_TIMEOUT); async_handles.push(BackupWorkload::spawn( cluster.new_client(), pd_client.clone(), br_client, keyspace_manager.clone(), backup_config, s3fs.clone(), BackupWorkloadConfig { periodic_interval: PERIODIC_BACKUP_INTERVAL, batch_interval: BACKUP_BATCH_INTERVAL, instant_interval: Duration::from_secs(5), on_demand_interval_range: ON_DEMAND_BACKUP_INTERVAL_MIN ..=ON_DEMAND_BACKUP_INTERVAL_MAX, operation_timeout: BACKUP_OPERATION_TIMEOUT, workload_timeout: TIMEOUT, }, )); } async_handles.push(spawn_gc_worker( runtime.block_on(cluster.new_txn_client()), pd_client.clone(), keyspace_manager.clone(), TIMEOUT, )); for _ in 0..DROP_TABLE_CONCURRENCY { async_handles.push(spawn_drop_table( runtime.block_on(cluster.new_keyspace_client()), Duration::from_secs(10), TIMEOUT, )); } // Write workloads: for idx in 0..WRITE_CONCURRENCY { async_handles.push(spawn_keyspace_write( idx, runtime.block_on(cluster.new_keyspace_client()), TIMEOUT, )); } if !use_builtin_dfs { for i in 0..TXN_FILE_WRITE_CONCURRENCY { handles.push(spawn_txn_file_write( WRITE_CONCURRENCY, i, cluster.new_client_opt(ClusterClientOptions { txn_file_max_chunk_size: Some(TXN_CHUNK_MAX_SIZE), ..Default::default() }), keyspace_manager.clone(), switches.txn_file_pessimistic, TIMEOUT, )); } } if switches.enable_oss_chaos && !use_builtin_dfs { handles.push(spawn_oss_chaos(&oss, OSS_CHAOS_INTERVAL, running.clone())) } // Main loop. let start_time = Instant::now(); while start_time.saturating_elapsed() < TIMEOUT { // Restart nodes. // Always force stop when OSS chaos is enabled. Otherwise, it would take too // long to stop compact worker. random_node_restart_with_schedule_lock( &mut cluster, Some(schedule_lock.as_ref()), |_, _| {}, switches.enable_oss_chaos, ); } // Finish. info!("test finished, stopping all workers"); running.stop(); for handle in handles { handle.join().unwrap(); } runtime.block_on(async move { for handle in async_handles { handle.await.unwrap(); } }); info!("run all pending destroy range"); let txn_client = runtime.block_on(cluster.new_txn_client()); let all_keyspaces = keyspace_manager.get_all_keyspaces(); for keyspace_id in all_keyspaces { runtime .block_on(pick_and_run_pending_destroy_range( keyspace_id, u64::MAX, &keyspace_manager, &txn_client, )) .unwrap(); } // Read requests can still get data before destroy range is done by compaction. // So must wait for destroy range to finish before verifying. info!("wait destroy range"); cluster.wait_for_destroy_range(Duration::from_secs(30)); // Verify. info!("verify cluster"); let verified_records_count = runtime.block_on(verify_cluster( &mut cluster, use_builtin_dfs, switches.online_snapshot_restore_ratio, )); // Stop cluster. info!("stopping cluster"); cluster.stop(); // Statistics. let stats = WorkloadStats::collect(); let region_number = pd_client.get_regions_number(); let stdout = std::io::stdout(); writeln!( stdout.lock(), "{} TEST SUCCEED: elapsed {:?},{:?}, region {}, verified_records {}, {:?}", test_id(), prepare_time.saturating_elapsed(), start_time.saturating_elapsed(), region_number, verified_records_count, stats, ) .unwrap(); stdout.lock().flush().unwrap(); } fn prepare_cluster( runtime: &Runtime, mut dfs_config: DFSConfig, security_conf: &SecurityConfig, nodes_count: usize, initial_keyspace_count: usize, switches: &Switches, oss: &ObjectStorageService, use_builtin_dfs: bool, ) -> (ServerCluster, TestPdControlServer) { let mut rng = rand::thread_rng(); let nodes = alloc_node_id_vec(nodes_count); let cpu_cores = SysQuota::cpu_cores_quota() as usize; let big_region_size_keyspaces: HashSet = (1..=initial_keyspace_count as u32) .choose_multiple(&mut rng, BIG_REGION_SIZE_KEYSPACE_COUNT) .into_iter() .collect(); let per_keyspace_configs = (1..=initial_keyspace_count as u32) .map(|keyspace| kvengine::config::PerKeyspaceConfig { keyspace, split_size_factor: if big_region_size_keyspaces.contains(&keyspace) { *BIG_REGION_SIZE_FACTOR_OPTIONS .iter() .choose(&mut rng) .unwrap() } else { 1.0 }, enable_blob: switches.enable_blob && rng.gen_bool(ENABLE_BLOB_RATIO), }) .collect::>(); let mut dfs_max_retry_count = 9; let mut dfs_dispatch_timeout = ReadableDuration::secs(15); let mut rfengine_target_file_size = ReadableSize::mb(8); let mut dfs_worker_memory_limit = ReadableSize::mb(32); if switches.enable_oss_chaos { match rng.gen_range(0..=3) { 0 => { // Keep the limit small enough to exercise memory pressure, but leave // room for WAL chunks rebuilt concurrently after a store restart. // A limit equal to one target file can reject normal in-flight chunks // and make the worker unhealthy before it can upload a recovery snapshot. rfengine_target_file_size = ReadableSize::kb(512); dfs_worker_memory_limit = rfengine_target_file_size * 8; } 1 => { // To cover the scene of DFS put object failure. dfs_max_retry_count = 2; dfs_dispatch_timeout = ReadableDuration::secs(3); } _ => {} } } // TODO: use same `conn_options` for other workloads to cover scene of slow DFS. dfs_config.conn_options = DFSConnOptions { max_retry_count: dfs_max_retry_count, retry_sleep_interval: ReadableDuration::millis(100), dispatch_timeout: dfs_dispatch_timeout, read_body_timeout: ReadableDuration::secs(15), ..Default::default() }; // 10MB/s (Prod env: 10GB/s, 16MB/sst -> 640 sst/s) oss.set_max_read_bytes_per_sec(KV_TARGET_FILE_SIZE.0 as usize * 640); // 10MB/s (Prod env: 1GB/s) // TODO: adjust rate for oss chaos. oss.set_max_write_bytes_per_sec(KV_TARGET_FILE_SIZE.0 as usize * 640); info!("prepare_cluster"; "per_keyspace_configs" => ?per_keyspace_configs, "dfs_conn_opts" => ?dfs_config.conn_options, "rfengine_target_file_size" => ?rfengine_target_file_size, "dfs_worker_memory_limit" => ?dfs_worker_memory_limit, ); let dfs = dfs_config.clone(); let enable_ia = switches.ia_table_enabled(); let update_conf_fn = move |_, conf: &mut TikvConfig| { let mut rng = rand::thread_rng(); if switches.for_serverless { apply_serverless_tikv_config(conf); } conf.dfs = dfs.clone(); conf.dfs.allow_fallback_local = false; conf.server.grpc_concurrency = (cpu_cores / nodes_count).max(2); conf.server.grpc_compression_type = GrpcCompressionType::Gzip; conf.security = security_conf.clone(); conf.coprocessor.region_split_size = ReadableSize::kb(256); conf.coprocessor.region_bucket_size = REGION_BUCKET_SIZE; conf.raft_store.peer_stale_state_check_interval = ReadableDuration::secs(1); conf.raft_store.abnormal_leader_missing_duration = ReadableDuration::secs(3); conf.raft_store.max_leader_missing_duration = ReadableDuration::secs(5); conf.raft_store.notify_capacity = 4096; // Small capacity to trigger more eviction. let mut lightweight_backup = true; if use_builtin_dfs { conf.raft_store.local_file_gc_timeout = ReadableDuration::secs(120); conf.raft_store.local_file_gc_tick_interval = ReadableDuration::secs(10); lightweight_backup = false; } conf.rocksdb.writecf.block_size = ReadableSize::kb(4); conf.rocksdb.writecf.write_buffer_size = ReadableSize::kb(96); conf.rocksdb.writecf.target_file_size_base = KV_TARGET_FILE_SIZE; conf.rfengine.target_file_size = rfengine_target_file_size; conf.rfengine.rlog_file_size = rfengine_target_file_size / *[2, 8, 32].choose(&mut rng).unwrap(); conf.rfengine.batch_compression_threshold = ReadableSize::kb(rng.gen_range(0..2)); conf.rfengine.lightweight_backup = lightweight_backup; conf.rfengine.wal_chunk_target_file_size = rfengine_target_file_size / 8; conf.rfengine.dfs_worker_memory_limit = dfs_worker_memory_limit.into(); conf.rfengine.wal_secondary_dir = Path::new(&conf.rfengine.wal_sync_dir) .with_file_name("wal2") .to_string_lossy() .to_string(); conf.kvengine.compaction_tombs_count = 100; conf.kvengine.max_del_range_delay = ReadableDuration(Duration::from_secs(3)); conf.kvengine.flush_split_l0 = true; conf.kvengine.per_keyspace_configs = per_keyspace_configs.clone(); conf.kvengine.blob_table_build_options = BlobTableBuildOptions { min_blob_size: 16, block_size: 4 * 1024, // 4KB, same as writecf.block_size max_blob_table_size: 16 * 1024, // 16KB, same as target_file_size_base target_blob_table_size: 8 * 1024, // 8KB ..Default::default() }; conf.kvengine.value_cache_capacity = if switches.enable_value_cache { ReadableSize::mb(1).into() } else { 0.into() }; if enable_ia { conf.kvengine.ia = IaConfig { mem_cap: IA_MEM_CAP_DEF.into(), disk_cap: IA_DISK_CAP_DEF.into(), segment_size: conf.rocksdb.writecf.block_size.0 as i64 * 2, freq_update_interval: ReadableDuration(IA_FREQ_UPDATE_INTERVAL_DEF), ..Default::default() } } conf.storage.flow_control.enable = true; conf.storage.flow_control.min_region_speed_limit = ReadableSize::kb(16); conf.storage.flow_control.max_region_speed_limit = ReadableSize::mb(1); conf.storage.scheduler_worker_pool_size = cpu_cores; conf.storage.check_backup_ts = switches.txn_check_backup_ts; }; let pd_wrapper = PdWrapper::new_test(1, security_conf, None); let mut cluster = ServerClusterBuilder::new(nodes, update_conf_fn) .pd(pd_wrapper) .build(); let keyspace_manager = cluster.keyspace_manager().clone(); let pd_control_server = TestPdControlServer::start(runtime, cluster.get_pd_client(), move |keyspace_name| { keyspace_manager .get_all_keyspaces() .into_iter() .find_map(|keyspace_id| { let keyspace = keyspace_manager.get_keyspace_meta(keyspace_id)?; (keyspace.name() == keyspace_name).then_some(keyspace_id) }) }); if !use_builtin_dfs { let worker_ids = alloc_node_id_vec(TIKV_WORKERS_COUNT); cluster.start_tikv_workers( worker_ids, TikvWorkerOptions { kv_target_file_size: KV_TARGET_FILE_SIZE, cop_block_cache_size: COP_BLOCK_CACHE_SIZE, pd_control_endpoints: Some(vec![pd_control_server.endpoint().to_string()]), ..Default::default() }, ); cluster.start_tikv_api_worker( alloc_node_id(), NativeBrConfig { instant_backup_timeout: ReadableDuration(BACKUP_OPERATION_TIMEOUT), backup_interval: ReadableDuration::ZERO, // The on-demand client also has a 30s timeout. The default 15s batch // tick needs two ticks when backup_delay is set, leaving no time for // the backup itself or a retry after a concurrent store restart. backup_batch_interval: ReadableDuration::secs(1), backup_delay: ReadableDuration::secs(1), backup_skip_keyspace_meta: true, backup_tolerate_err: true, ..Default::default() }, ); } cluster.wait_region_replicated(&[], 3); let pd_client = cluster.get_pd_client(); pd_client.disable_default_operator(); // Split keyspaces. let mut keyspace_keys = vec![]; let mut keyspaces: Vec = vec![]; let mut data_keys = vec![]; let mut next_table_id = 1; // Table id is global unique. See `TableMeta::new`. for idx in 0..initial_keyspace_count { // New keyspace must allocated by keyspace manager to avoid conflicts. // TODO: test for default keyspace. let keyspace_id = cluster.keyspace_manager().new_keyspace_id(1); keyspaces.push(keyspace_id); keyspace_keys.push(ApiV2::get_txn_keyspace_prefix(keyspace_id)); keyspace_keys.push(ApiV2::get_txn_keyspace_prefix(keyspace_id + 1)); for _ in 0..INITIAL_TABLE_COUNT { let table_id = next_table_id; next_table_id += 1; for i in 0..rng.gen_range(0..10) { let key = make_row_key(keyspace_id, table_id, &gen_row_key_suffix(i * 100)); data_keys.push(Key::from_raw(&key).into_encoded()); } } // Random tests want to exercise both legacy and metadata paths in one run. // Split one keyspace range at a time so we never build a batch split with // mixed intents. let cfg = match idx { 0 => KeyspaceEncryptionConfig { enabled: true, bypass_mode_check: true, ..Default::default() }, 1 if switches.enable_cmek => KeyspaceEncryptionConfig { enabled: true, cmek_id: Some("random".into()), vendor: Some("test".into()), bypass_mode_check: true, ..Default::default() }, _ => random_keyspace_encryption_config(&mut rng, switches.enable_cmek), }; match pd_client.set_keyspace_encryption(keyspace_id, cfg) { Ok(_) => {} Err(err) => { panic!("set_keyspace_encryption failed: {:?}", err) } } let split_keys = [ ApiV2::get_txn_keyspace_prefix(keyspace_id), ApiV2::get_txn_keyspace_prefix(keyspace_id + 1), ] .iter() .map(|k| Key::from_raw(k).into_encoded()) .collect::>(); block_on(pd_client.split_regions_with_retry(split_keys, Duration::from_secs(60))).unwrap(); } keyspace_keys.dedup(); cluster.wait_pd_region_min_count(keyspace_keys.len() + 1); let keyspace_names = keyspaces .iter() .map(|&keyspace_id| TidbCluster::keyspace_name(keyspace_id as u16)) .collect(); let ia_table_ratio = switches.ia_table_ratio; let sc_spec_fn = Box::new(move |_| { if thread_rng().gen_bool(ia_table_ratio) { StorageClass::Ia.into() } else { StorageClassSpec::default() } }); cluster.keyspace_manager().create_keyspaces( &keyspaces, keyspace_names, &CreateKeyspaceOptions { table_count: INITIAL_TABLE_COUNT, schema_enable_ratio: TABLE_SCHEMA_ENABLE_RATIO, storage_class_spec_fn: sc_spec_fn, }, Some(&mut rng), ); KEYSPACE_COUNTER.store(initial_keyspace_count, Ordering::Relaxed); TABLE_COUNTER.store( initial_keyspace_count * INITIAL_TABLE_COUNT, Ordering::Relaxed, ); let data_keys_len = data_keys.len(); let res = block_on(pd_client.split_regions_with_retry(data_keys, Duration::from_secs(60))); if let Err(err) = res { warn!("split regions failed: {:?}", err); } let all_keyspaces = cluster.keyspace_manager().get_all_keyspaces(); let fs = cluster.get_dfs().unwrap(); let runtime = fs.get_runtime(); let stores = pd_client.get_all_stores(true).unwrap(); for keyspace_id in all_keyspaces { let keyspace_meta = cluster .keyspace_manager() .get_keyspace_meta(keyspace_id) .unwrap(); if !keyspace_meta.schemas().is_empty() { let schema_version = 10; let schema_data: Bytes = build_schema_file(keyspace_id, schema_version, keyspace_meta.schemas(), 0).into(); let schema_file_id = pd_client.alloc_id().unwrap(); runtime .block_on(fs.create( schema_file_id, schema_data.clone(), dfs::Options::default().with_type(FileType::Schema), )) .unwrap(); let schema_file = SchemaFile::open(Arc::new(InMemFile::new(schema_file_id, schema_data))).unwrap(); // Setup schema file to stores. runtime.block_on(broadcast_schema_file_request_and_check( &stores, keyspace_id, &schema_file, Duration::from_secs(30), )); } } // Scatter regions. let move_scheduler = cluster.new_scheduler(); for _ in 0..(keyspace_keys.len() + data_keys_len + 1) { move_scheduler.move_random_region(); } (cluster, pd_control_server) } async fn verify_cluster( cluster: &mut ServerCluster, use_builtin_dfs: bool, online_snapshot_restore_ratio: f64, ) -> usize /* records count in ref store */ { // Verify data. let mut handles = vec![]; for keyspace_id in cluster.keyspace_manager().ref_stores().all_keyspace_ids() { let mut client = cluster.new_keyspace_client().await; handles.push(tokio::spawn(async move { (keyspace_id, client.verify_keyspace(keyspace_id).await) })); } let mut records_cnt = 0; for handle in handles { let (keyspace_id, res) = handle.await.unwrap(); records_cnt += res.unwrap_or_else(|err| { panic!( "{} verify_keyspace_with_ref_store failed: {:?}", keyspace_id, err ); }); } assert!(records_cnt > 1000, "too few records in ref store"); // Check statistics. // Check after verify data, to ensure that PD heartbeat have updated region // stats. verify_cluster_stats(cluster, REGION_BUCKET_SIZE.0, Duration::from_secs(30)); verify_region_info_accessor(cluster); verify_rfengine_keyspace_id(cluster); if !use_builtin_dfs { check_br(online_snapshot_restore_ratio); check_load_data(); } check_gc(); check_drop_table(); // TODO: check storage class property. assert_eq!(ENGINE_IA_SYNC_READ_COUNTER.get(), 0); records_cnt } #[derive(Debug)] pub(crate) struct Switches { pub for_serverless: bool, pub ia_table_ratio: f64, pub enable_oss_chaos: bool, pub txn_check_backup_ts: bool, pub enable_value_cache: bool, pub enable_blob: bool, pub enable_cmek: bool, pub restore_lower_memory: bool, pub txn_file_pessimistic: bool, pub online_snapshot_restore_ratio: f64, } impl Switches { pub fn from_env() -> Self { let mut rng = thread_rng(); let for_serverless = env_switch("FOR_SERVERLESS"); let ia_table_ratio: f64 = env_param("IA_TABLE_RATIO", 0.5); let enable_oss_chaos = rng.gen_bool(env_param("OSS_CHAOS_RATIO", 0.2)); let txn_check_backup_ts = env_switch("TXN_CHECK_BACKUP_TS"); let enable_value_cache = env_switch("ENABLE_VALUE_CACHE"); let enable_blob = env_switch_opt("ENABLE_BLOB", 1); let enable_cmek = env_switch("ENABLE_CMEK"); let lower_memory = rng.gen_bool(0.5); let txn_file_pessimistic = env_switch_opt("TXN_FILE_PESSIMISTIC", 0); let online_snapshot_restore_ratio = env_param( ONLINE_SNAPSHOT_RESTORE_RATIO_ENV, ONLINE_SNAPSHOT_RESTORE_RATIO_DEFAULT, ); assert!( (0.0..=1.0).contains(&online_snapshot_restore_ratio), "{} must be between 0 and 1, got {}", ONLINE_SNAPSHOT_RESTORE_RATIO_ENV, online_snapshot_restore_ratio, ); Self { for_serverless, ia_table_ratio, enable_oss_chaos, txn_check_backup_ts, enable_value_cache, enable_blob, enable_cmek, restore_lower_memory: lower_memory, txn_file_pessimistic, online_snapshot_restore_ratio, } } pub fn ia_table_enabled(&self) -> bool { self.ia_table_ratio > 0.0 } } #[allow(dead_code)] #[derive(Debug)] struct WorkloadStats { write_count: usize, txn_file_write_count: usize, keyspace_count: usize, table_count: usize, drop_table_count: usize, merge_count: usize, move_count: usize, transfer_count: usize, node_restart: usize, backup_count: usize, on_demand_backup_count: usize, backup_tolerated_err_count: usize, restore_count: usize, online_snapshot_restore_count: usize, restore_tolerated_err_count: usize, broken_backup_count: usize, load_data_count: usize, manual_major_compact: usize, gc_resolved_locks: usize, remote_compact_exceed_memory_limit: u64, rfengine_dfs_worker_unhealthy: u64, } impl WorkloadStats { fn collect() -> Self { let write_count = WRITE_COUNTER.load(Ordering::SeqCst); let txn_file_write_count = TXN_FILE_WRITE_COUNTER.load(Ordering::SeqCst); let keyspace_count = KEYSPACE_COUNTER.load(Ordering::SeqCst); let table_count = TABLE_COUNTER.load(Ordering::SeqCst); let drop_table_count = DROP_TABLE_COUNTER.load(Ordering::SeqCst); let merge_count = MERGE_COUNTER.load(Ordering::SeqCst); let move_count = MOVE_COUNTER.load(Ordering::SeqCst); let transfer_count = TRANSFER_COUNTER.load(Ordering::SeqCst); let node_restart = NODE_RESTART_COUNTER.load(Ordering::SeqCst); let backup_count = BACKUP_COUNTER.load(Ordering::SeqCst); let on_demand_backup_count = ON_DEMAND_BACKUP_COUNTER.load(Ordering::SeqCst); let backup_tolerated_err_count = BACKUP_TOLERATED_ERR_COUNTER.load(Ordering::SeqCst); let restore_count = RESTORE_COUNTER.load(Ordering::SeqCst); let online_snapshot_restore_count = ONLINE_SNAPSHOT_RESTORE_COUNTER.load(Ordering::SeqCst); let restore_tolerated_err_count = RESTORE_TOLERATED_ERR_COUNTER.load(Ordering::SeqCst); let broken_backup_count = BROKEN_BACKUP_COUNTER.load(Ordering::SeqCst); let load_data_count = LOAD_DATA_COUNTER.load(Ordering::SeqCst); let manual_major_compact = MANUAL_MAJOR_COMPACT_COUNTER.load(Ordering::SeqCst); let gc_resolved_locks = GC_ADVANCE_SAFE_POINT_COUNTER.load(Ordering::SeqCst); let remote_compact_exceed_memory_limit = ENGINE_REMOTE_COMPACT_EXCEED_MEMORY_LIMIT_COUNTER.get(); let rfengine_dfs_worker_unhealthy = RFENGINE_DFS_WORKER_BECOME_UNHEALTHY_COUNTER.get(); Self { write_count, txn_file_write_count, keyspace_count, table_count, drop_table_count, merge_count, move_count, transfer_count, node_restart, backup_count, on_demand_backup_count, backup_tolerated_err_count, restore_count, online_snapshot_restore_count, restore_tolerated_err_count, broken_backup_count, load_data_count, manual_major_compact, gc_resolved_locks, remote_compact_exceed_memory_limit, rfengine_dfs_worker_unhealthy, } } }