// Copyright 2022 TiKV Project Authors. Licensed under Apache-2.0. // Modifications Copyright 2022 PingCAP, Inc. use std::time::{Duration, Instant}; use lazy_static::lazy_static; use prometheus::*; use prometheus_static_metric::*; use crate::*; pub fn flush_engine_properties(kv_engine: &Engine, _name: &str) { let kv_all_shard_stats = kv_engine.get_all_shard_stats(); Engine::update_region_huge_table_bytes_metrics( &kv_all_shard_stats, kv_engine.opts.max_mem_table_size, ); let EngineStats { num_shards, num_initial_flushed_shard, num_active_shards, num_compacting_shards, num_pending_compaction_shards, num_has_del_prefixes_shards, num_inner_key_shards, ready_destroy_range_shards, mem_tables_count, mem_tables_size: _, // Exported by PdRunner::handle_store_heartbeat l0_tables_count, l0_tables_size, blob_tables_count, in_use_blob_size, total_blob_size, partial_l0_count, partial_blob_count, partial_ln_count, cf_level_files: cf_level_num_files, cf_level_sizes: cf_level_total_sizes, index_size, filter_size, open_files, max_ts, entries, old_entries, tombs, kv_size, total_data_size: _, txn_file_locks, columnar_levels, vector_indexes, fts_levels: _, // TODO: Export FTS levels if needed. tops: _, // TODO: Export to metrics. ia: StorageClassStats { num_shards: ia_num_shards, num_tables: ia_num_tables, data_size: _, // Exported by PdRunner::handle_store_heartbeat kv_size: _, // Exported by PdRunner::handle_store_heartbeat }, } = kv_engine.get_engine_stats(kv_all_shard_stats); ENGINE_SHARDS.set(num_shards as i64); ENGINE_INITIAL_FLUSHED_SHARDS.set(num_initial_flushed_shard as i64); ENGINE_ACTIVE_SHARDS.set(num_active_shards as i64); ENGINE_COMPACTING_SHARDS.set(num_compacting_shards as i64); ENGINE_PENDING_COMPACTION_SHARDS.set(num_pending_compaction_shards as i64); ENGINE_HAS_DEL_PREFIXES_SHARDS.set(num_has_del_prefixes_shards as i64); ENGINE_INNER_KEY_SHARDS.set(num_inner_key_shards as i64); ENGINE_READY_DESTROY_RANGE_SHARDS.set(ready_destroy_range_shards.len() as i64); ENGINE_MEM_TABLES_COUNT.set(mem_tables_count as i64); ENGINE_L0_TABLES_COUNT.set(l0_tables_count as i64); ENGINE_L0_TABLES_SIZE.set(l0_tables_size as i64); ENGINE_BLOB_TABLES_COUNT.set(blob_tables_count as i64); ENGINE_IN_USE_BLOB_SIZE.set(in_use_blob_size as i64); ENGINE_TOTAL_BLOB_SIZE.set(total_blob_size as i64); ENGINE_PARTIAL_L0_COUNT.set(partial_l0_count as i64); ENGINE_PARTIAL_BLOB_COUNT.set(partial_blob_count as i64); ENGINE_PARTIAL_LN_COUNT.set(partial_ln_count as i64); for (cf, level_files) in cf_level_num_files.iter().enumerate() { for (level, num_files) in level_files.iter().enumerate() { ENGINE_CF_LEVEL_NUM_FILES .with_label_values(&[CF_NAMES[cf], &level.to_string()]) .set(*num_files as i64); } } for (cf, level_size) in cf_level_total_sizes.iter().enumerate() { for (level, total_size) in level_size.iter().enumerate() { ENGINE_CF_LEVEL_TOTAL_SIZES .with_label_values(&[CF_NAMES[cf], &level.to_string()]) .set(*total_size as i64); } } ENGINE_INDEX_SIZE.set(index_size as i64); ENGINE_FILTER_SIZE.set(filter_size as i64); ENGINE_OPEN_FILES.set(open_files); ENGINE_MAX_TS.set(max_ts as i64); ENGINE_ENTRIES.set(entries as i64); ENGINE_OLD_ENTRIES.set(old_entries as i64); ENGINE_TOMBSTONES.set(tombs as i64); ENGINE_KV_SIZE.set(kv_size as i64); ENGINE_TXN_FILE_LOCKS.set(txn_file_locks as i64); for (level, stats) in columnar_levels.iter().enumerate() { ENGINE_COLUMNAR_FILES_COUNT .with_label_values(&[&level.to_string()]) .set(stats.num_files as i64); ENGINE_COLUMNAR_DATA_SIZE .with_label_values(&[&level.to_string()]) .set(stats.data_size as i64); } ENGINE_VECTOR_INDEX_FILE_COUNT.set(vector_indexes.num_files as i64); ENGINE_VECTOR_INDEX_DATA_SIZE.set(vector_indexes.data_size as i64); ENGINE_IA_SHARDS.set(ia_num_shards as i64); ENGINE_IA_SST_TABLES.set(ia_num_tables as i64); } make_static_metric! { pub label_enum WriteFlowType { keys, bytes, } pub struct WriteFlowVec: Histogram { "type" => WriteFlowType, } pub label_enum IaIoType { remote_read, local_write, } pub struct IaIoThrottleDurationVec: Histogram { "type" => IaIoType, } pub struct IaIoThroughputVec: IntCounter { "type" => IaIoType, } pub struct IaIoThroughputCapVec: IntGauge { "type" => IaIoType, } } make_auto_flush_static_metric! { pub label_enum ChangeSetType { clear_columnar, columnar_compaction, compaction, destroy_range, flush, ingest_files, initial_flush, major_compaction, restore_shard, trim_over_bound, truncate_ts, update_schema_meta, update_vector_index, update_storage_class, fts_update, } pub struct ChangeSetVec: LocalIntCounter { "type" => ChangeSetType, } pub label_enum SeekType { seek, next, prev, } pub struct SeekDurationVec: LocalHistogram { "type" => SeekType, } pub label_enum SSTCryptType { encrypt_data_blocks, decrypt_data_block, } pub struct SSTCryptDurationVec: LocalHistogram { "type" => SSTCryptType, } } lazy_static! { pub static ref ENGINE_APPLY_CHANGE_SET_VEC: IntCounterVec = register_int_counter_vec!( "kvengine_apply_change_set_total", "Total number of kvengine apply change set.", &["type"] ) .unwrap(); pub static ref ENGINE_APPLY_CHANGE_SET_COUNTER: ChangeSetVec = auto_flush_from!(ENGINE_APPLY_CHANGE_SET_VEC, ChangeSetVec); pub static ref ENGINE_SHARDS: IntGauge = register_int_gauge!( "kv_engine_shards_total", "Total number of shards in the engine", ) .unwrap(); pub static ref ENGINE_INITIAL_FLUSHED_SHARDS: IntGauge = register_int_gauge!( "kv_engine_initial_flushed_shards", "Number of initial flushed shards in the engine", ) .unwrap(); pub static ref ENGINE_ACTIVE_SHARDS: IntGauge = register_int_gauge!( "kv_engine_active_shards", "Number of active shards in the engine", ) .unwrap(); pub static ref ENGINE_COMPACTING_SHARDS: IntGauge = register_int_gauge!( "kv_engine_compacting_shards", "Number of compacting shards in the engine", ) .unwrap(); pub static ref ENGINE_PENDING_COMPACTION_SHARDS: IntGauge = register_int_gauge!( "kv_engine_pending_compaction_shards", "Number of pending compaction shards in the engine", ) .unwrap(); pub static ref ENGINE_HAS_DEL_PREFIXES_SHARDS: IntGauge = register_int_gauge!( "kv_engine_has_del_prefixes_shards", "Number of shards with delete prefixes in the engine", ) .unwrap(); pub static ref ENGINE_INNER_KEY_SHARDS: IntGauge = register_int_gauge!( "kv_engine_inner_key_shards", "Number of shards with inner keys in the engine", ) .unwrap(); pub static ref ENGINE_READY_DESTROY_RANGE_SHARDS: IntGauge = register_int_gauge!( "kv_engine_ready_destroy_range_shards", "Number of shards ready to destroy range in the engine", ) .unwrap(); pub static ref ENGINE_MEM_TABLES_COUNT: IntGauge = register_int_gauge!( "kv_engine_mem_tables_count", "Number of memory tables in the engine", ) .unwrap(); pub static ref ENGINE_L0_TABLES_COUNT: IntGauge = register_int_gauge!( "kv_engine_l0_tables_count", "Number of L0 tables in the engine", ) .unwrap(); pub static ref ENGINE_L0_TABLES_SIZE: IntGauge = register_int_gauge!( "kv_engine_l0_tables_size_bytes", "Total size of L0 tables in bytes", ) .unwrap(); pub static ref ENGINE_BLOB_TABLES_COUNT: IntGauge = register_int_gauge!( "kv_engine_blob_tables_count", "Number of blob tables in the engine", ) .unwrap(); pub static ref ENGINE_IN_USE_BLOB_SIZE: IntGauge = register_int_gauge!( "kv_engine_in_use_blob_size_bytes", "In-use blob size in bytes", ) .unwrap(); pub static ref ENGINE_TOTAL_BLOB_SIZE: IntGauge = register_int_gauge!( "kv_engine_total_blob_size_bytes", "Total blob size in bytes", ) .unwrap(); pub static ref ENGINE_PARTIAL_L0_COUNT: IntGauge = register_int_gauge!("kv_engine_partial_l0_count", "Number of partial L0 tables",).unwrap(); pub static ref ENGINE_PARTIAL_BLOB_COUNT: IntGauge = register_int_gauge!( "kv_engine_partial_blob_count", "Number of partial blob tables", ) .unwrap(); pub static ref ENGINE_PARTIAL_LN_COUNT: IntGauge = register_int_gauge!("kv_engine_partial_ln_count", "Number of partial Ln tables",).unwrap(); pub static ref ENGINE_CF_LEVEL_NUM_FILES: IntGaugeVec = register_int_gauge_vec!( "kv_engine_cf_level_num_files", "Number of files in each level", &["cf", "level"] ) .unwrap(); pub static ref ENGINE_CF_LEVEL_TOTAL_SIZES: IntGaugeVec = register_int_gauge_vec!( "kv_engine_cf_level_total_sizes_bytes", "Total sizes of each level in bytes", &["cf", "level"] ) .unwrap(); pub static ref ENGINE_INDEX_SIZE: IntGauge = register_int_gauge!("kv_engine_index_size_bytes", "Total index size in bytes",).unwrap(); pub static ref ENGINE_FILTER_SIZE: IntGauge = register_int_gauge!("kv_engine_filter_size_bytes", "Total filter size in bytes",).unwrap(); pub static ref ENGINE_MAX_TS: IntGauge = register_int_gauge!( "kv_engine_max_ts", "Maximum timestamp for PiTR completion check", ) .unwrap(); pub static ref ENGINE_ENTRIES: IntGauge = register_int_gauge!("kv_engine_entries", "Total number of entries in the engine",).unwrap(); pub static ref ENGINE_OLD_ENTRIES: IntGauge = register_int_gauge!( "kv_engine_old_entries", "Number of old entries in the engine", ) .unwrap(); pub static ref ENGINE_TOMBSTONES: IntGauge = register_int_gauge!("kv_engine_tombstones", "Number of tombstones in the engine",).unwrap(); pub static ref ENGINE_KV_SIZE: IntGauge = register_int_gauge!("kv_engine_kv_size_bytes", "Total KV size in bytes",).unwrap(); pub static ref ENGINE_TXN_FILE_LOCKS: IntGauge = register_int_gauge!( "kv_engine_txn_file_locks", "Number of transaction file locks", ) .unwrap(); pub static ref ENGINE_COLUMNAR_FILES_COUNT: IntGaugeVec = register_int_gauge_vec!( "kv_engine_columnar_files_count", "Number of columnar files in the engine", &["level"] ) .unwrap(); pub static ref ENGINE_COLUMNAR_DATA_SIZE: IntGaugeVec = register_int_gauge_vec!( "kv_engine_columnar_data_size_bytes", "Total size of columnar data in bytes", &["level"] ) .unwrap(); pub static ref ENGINE_VECTOR_INDEX_FILE_COUNT: IntGauge = register_int_gauge!( "kv_engine_vector_index_file_count", "Number of vector index files in the engine", ) .unwrap(); pub static ref ENGINE_VECTOR_INDEX_DATA_SIZE: IntGauge = register_int_gauge!( "kv_engine_vector_index_data_size_bytes", "Total size of vector indexes in bytes", ) .unwrap(); } lazy_static! { pub static ref ENGINE_GET_DURATION: Histogram = register_histogram!( "kv_engine_get_duration_seconds", "Bucketed histogram of KV Engine get duration", exponential_buckets(0.00005, 1.8, 26).unwrap() ) .unwrap(); pub static ref ENGINE_SEEK_DURATION_VEC: HistogramVec = register_histogram_vec!( "kv_engine_seek_duration_seconds", "Bucketed histogram of KV Engine seek duration", &["type"], exponential_buckets(0.00005, 1.8, 26).unwrap() ) .unwrap(); pub static ref ENGINE_SEEK_DURATION_STATIC: SeekDurationVec = auto_flush_from!(ENGINE_SEEK_DURATION_VEC, SeekDurationVec); pub static ref ENGINE_WRITE_DURATION: Histogram = register_histogram!( "kv_engine_write_duration_seconds", "Bucketed histogram of KV Engine write duration", exponential_buckets(0.00005, 1.8, 26).unwrap() ) .unwrap(); pub static ref ENGINE_WRITE_FLOW: WriteFlowVec = register_static_histogram_vec!( WriteFlowVec, "kv_engine_write_flow", "Bucketed histogram of KV Engine write flow", &["type"], exponential_buckets(8.0, 2.0, 20).unwrap() ) .unwrap(); pub static ref ENGINE_ARENA_GROW_DURATION_HISTOGRAM: Histogram = register_histogram!( "kv_engine_arena_grow_duration_seconds", "Bucketed histogram of KV Engine arena grow duration", exponential_buckets(0.00005, 1.8, 26).unwrap() ) .unwrap(); pub static ref ENGINE_CACHE_MISS: IntCounter = register_int_counter!("kv_engine_cache_miss", "kv engine cache miss",).unwrap(); pub static ref ENGINE_CACHE_LOOKUP: IntCounter = register_int_counter!( "kv_engine_cache_lookup", "Total number of block cache lookups", ) .unwrap(); pub static ref ENGINE_LEVEL_WRITE_VEC: IntCounterVec = register_int_counter_vec!( "kv_engine_level_write_bytes", "Write bytes of kvengine of each level", &["level"] ) .unwrap(); pub static ref ENGINE_OPEN_FILES: IntGauge = register_int_gauge!("kv_engine_open_files", "kv engine open files",).unwrap(); pub static ref ENGINE_LOAD_TABLE_FILES_ERROR: IntCounterVec = register_int_counter_vec!( "kv_engine_load_table_files_error", "Total number of kv engine load table files error", &["type"] ) .unwrap(); pub static ref ENGINE_LIMITER_THROTTLE_COUNTER: IntCounterVec = register_int_counter_vec!( "kv_engine_limiter_throttle_counter", "Counter when throttle happens.", &["limiter"] ) .unwrap(); pub static ref ENGINE_LIMITER_THROTTLE_ACTION_COUNTER: IntCounterVec = register_int_counter_vec!( "kv_engine_limiter_throttle_action_counter", "Counter when different throttling actions happen.", &["limiter", "type"] ) .unwrap(); pub static ref ENGINE_REGION_HUGE_MEM_TABLE_BYTES_HISTOGRAM: Histogram = register_histogram!( "kv_engine_region_huge_mem_table_bytes", "Histogram of huge mem table bytes for regions", exponential_buckets(1024.0 * 1024.0, 2.0, 20).unwrap() ) .unwrap(); pub static ref ENGINE_REGION_HUGE_L0_TABLE_BYTES_HISTOGRAM: Histogram = register_histogram!( "kv_engine_region_huge_l0_table_bytes", "Histogram of huge l0 table bytes for regions", exponential_buckets(1024.0 * 1024.0, 2.0, 20).unwrap() ) .unwrap(); pub static ref ENGINE_FREE_MEM_BYTES_HISTOGRAM: Histogram = register_histogram!( "kv_engine_free_mem_bytes", "Histogram of free mem bytes", exponential_buckets(1024.0 * 1024.0, 2.0, 20).unwrap() // 1MB ~ 1TB ) .unwrap(); pub static ref ENGINE_IA_MANAGER_SEGMENTS_DISK_SIZE: IntGauge = register_int_gauge!( "kv_engine_ia_manager_segments_disk_size", "Total disk usage size of IA manager segments", ) .unwrap(); pub static ref ENGINE_IA_MANAGER_SEGMENTS_MEMORY_SIZE: IntGauge = register_int_gauge!( "kv_engine_ia_manager_segments_memory_size", "Total memory size of IA manager segments", ) .unwrap(); pub static ref ENGINE_IA_MANAGER_SEGMENTS_MEMORY_CAPACITY: IntGauge = register_int_gauge!( "kv_engine_ia_manager_segments_memory_capacity", "Memory admission capacity of IA manager segments", ) .unwrap(); pub static ref ENGINE_IA_MANAGER_SEGMENTS_MEMORY_WAIT_DURATION: Histogram = register_histogram!( "kv_engine_ia_manager_segments_memory_wait_duration_seconds", "Histogram of wait duration when acquiring IA segment memory", exponential_buckets(0.001, 2.0, 20).unwrap() // 1ms ~ 524s ) .unwrap(); pub static ref ENGINE_IA_MAIN_QUEUE_CAPACITY: IntGauge = register_int_gauge!( "kv_engine_ia_main_queue_capacity", "Capacity of IA main queue", ) .unwrap(); pub static ref ENGINE_IA_SMALL_QUEUE_CAPACITY: IntGauge = register_int_gauge!( "kv_engine_ia_small_queue_capacity", "Capacity of IA small queue", ) .unwrap(); pub static ref ENGINE_DFS_LOAD_MEMORY_USAGE: IntGauge = register_int_gauge!( "kv_engine_dfs_load_memory_usage_bytes", "Required DFS load memory reservations in bytes", ) .unwrap(); pub static ref ENGINE_DFS_LOAD_MEMORY_WAITING_REQUIRED_LOADS: IntGauge = register_int_gauge!( "kv_engine_dfs_load_memory_waiting_required_loads", "Number of required DFS loads waiting for memory admission", ) .unwrap(); pub static ref ENGINE_DFS_LOAD_MEMORY_WAIT_DURATION: Histogram = register_histogram!( "kv_engine_dfs_load_memory_wait_duration_seconds", "Histogram of wait duration when acquiring DFS load memory", exponential_buckets(0.001, 2.0, 20).unwrap() // 1ms ~ 524s ) .unwrap(); pub static ref ENGINE_MATERIALIZATION_REMOTE_LOGICAL_READ_BYTES: IntCounterVec = register_int_counter_vec!( "kv_engine_materialization_remote_logical_read_bytes_total", "Total logical DFS payload bytes returned while materializing files by I/O type", &["io_type"] ) .unwrap(); pub static ref ENGINE_MATERIALIZATION_REMOTE_READ_DURATION: HistogramVec = register_histogram_vec!( "kv_engine_materialization_remote_read_duration_seconds", "DFS read duration while materializing files by I/O type and transport result", &["io_type", "result"], exponential_buckets(0.001, 2.0, 20).unwrap() // 1ms ~ 524s ) .unwrap(); pub static ref ENGINE_IA_REMOTE_READ_SEGMENT_DURATION_HISTOGRAM: Histogram = register_histogram!( "kv_engine_ia_remote_read_segment_duration_seconds", "Histogram of IA remote segment read duration", exponential_buckets(5e-5, 2.0, 20).unwrap() // 50us ~ 26s ) .unwrap(); pub static ref ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_HIT: IntCounter = register_int_counter!( "kv_engine_ia_remote_read_segment_cache_hit", "Counter of IA remote segment cache hit", ) .unwrap(); pub static ref ENGINE_IA_REMOTE_READ_SEGMENT_CACHE_MISS: IntCounter = register_int_counter!( "kv_engine_ia_remote_read_segment_cache_miss", "Counter of IA remote segment cache miss", ) .unwrap(); pub static ref ENGINE_IA_IO_THROTTLE_DURATION: IaIoThrottleDurationVec = register_static_histogram_vec!( IaIoThrottleDurationVec, "kv_engine_ia_io_throttle_duration_seconds", "Bucketed histogram of IA IO rate limiter throttle duration", &["type"], exponential_buckets(0.001, 2.0, 20).unwrap() ) .unwrap(); pub static ref ENGINE_IA_IO_THROUGHPUT_BYTES: IaIoThroughputVec = register_static_int_counter_vec!( IaIoThroughputVec, "kv_engine_ia_io_throughput_bytes", "Total IA IO bytes passing through the IA rate limiter", &["type"] ) .unwrap(); pub static ref ENGINE_IA_IO_THROUGHPUT_CAP: IaIoThroughputCapVec = register_static_int_gauge_vec!( IaIoThroughputCapVec, "kv_engine_ia_io_throughput_cap", "IA IO rate limiter throughput cap in bytes per second. Zero means unlimited.", &["type"] ) .unwrap(); pub static ref ENGINE_STORAGE_CLASS_TRANSITION_REGION_PENDING: IntGauge = register_int_gauge!( "kv_engine_ia_transition_region_pending", "Number of storage-class transition regions waiting for the per-store region permit", ) .unwrap(); pub static ref ENGINE_STORAGE_CLASS_TRANSITION_REGION_INFLIGHT: IntGauge = register_int_gauge!( "kv_engine_ia_transition_region_inflight", "Number of storage-class transition regions currently inflight", ) .unwrap(); pub static ref ENGINE_STORAGE_CLASS_TRANSITION_REGION_FINISHED: IntCounterVec = register_int_counter_vec!( "kv_engine_ia_transition_region_finished_total", "Counter of storage-class transition regions finished by result", &["result"], ) .unwrap(); pub static ref ENGINE_IA_TO_STANDARD_TRANSITION_REMOTE_READ_BYTES: IntCounter = register_int_counter!( "kv_engine_ia_to_standard_transition_remote_read_bytes_total", "Total remote read bytes admitted by the IA to Standard transition limiter. The value is based on DFS file size before the remote read starts.", ) .unwrap(); pub static ref ENGINE_IA_TO_STANDARD_TRANSITION_REMOTE_READ_THROTTLE_DURATION: Counter = register_counter!( "kv_engine_ia_to_standard_transition_remote_read_throttle_duration_seconds", "Total time spent throttling IA to Standard transition remote reads", ) .unwrap(); pub static ref ENGINE_IA_SYNC_READ_COUNTER: IntCounter = register_int_counter!( "kv_engine_ia_sync_read_counter", "Counter of sync read IA segments", ) .unwrap(); // `type`: reserved for more storage classes in the future. pub static ref ENGINE_STORAGE_CLASS_TRANSITION_COUNTER: IntCounterVec = register_int_counter_vec!( "kv_engine_sc_transition_counter", "Counter of storage class transition", &["type"], ) .unwrap(); pub static ref ENGINE_IA_SHARDS: IntGauge = register_int_gauge!( "kv_engine_ia_shards_total", "Total number of IA shards in the engine", ) .unwrap(); pub static ref ENGINE_IA_SST_TABLES: IntGauge = register_int_gauge!( "kv_engine_ia_sst_tables_total", "Total number of IA SST tables in the engine", ) .unwrap(); pub static ref ENGINE_COLUMNAR_TOO_MANY_UNCONVERTED_L0S: IntCounter = register_int_counter!( "kv_engine_columnar_too_many_unconverted_l0s", "Counter of columnar has too many unconverted L0s", ) .unwrap(); pub static ref ENGINE_REMOTE_COMPACT_EXCEED_MEMORY_LIMIT_COUNTER: IntCounter = register_int_counter!( "kv_engine_remote_compact_exceed_memory_limit_counter", "Total number of remote compaction requests that exceed memory limit", ).unwrap(); pub static ref ENGINE_VECTOR_INDEX_CACHE_HIT: IntCounter = register_int_counter!( "kv_engine_vector_index_cache_hit", "Total number of vector index cache hit", ).unwrap(); pub static ref ENGINE_VECTOR_INDEX_CACHE_MISS: IntCounter = register_int_counter!( "kv_engine_vector_index_cache_miss", "Total number of vector index cache miss", ).unwrap(); pub static ref ENGINE_COLUMNAR_FILE_CACHE_HIT: IntCounter = register_int_counter!( "kv_engine_columnar_file_cache_hit", "Total number of columnar file cache hit", ).unwrap(); pub static ref ENGINE_COLUMNAR_FILE_CACHE_MISS: IntCounter = register_int_counter!( "kv_engine_columnar_file_cache_miss", "Total number of columnar file cache miss", ).unwrap(); pub static ref ENGINE_VALUE_CACHE_CACHE_FILL: IntCounter = register_int_counter!( "kv_engine_value_cache_fill", "Total number of value cache fill", ).unwrap(); pub static ref ENGINE_VALUE_CACHE_CACHE_HIT: IntCounter = register_int_counter!( "kv_engine_value_cache_hit", "Total number of value cache hit", ).unwrap(); pub static ref ENGINE_DUPLICATED_CHANGE_SET_COUNTER: IntCounter = register_int_counter!( "kv_engine_duplicated_changeset_counter", "Total number of kvengine duplicated change set", ).unwrap(); pub static ref COLUMNAR_PREFETCH_HISTOGRAM: Histogram = register_histogram!( "kv_engine_columnar_prefetch_duration_seconds", "Bucketed histogram of columnar prefetch segments duration", exponential_buckets(0.1, 2.0, 20).unwrap() ) .unwrap(); pub static ref COLUMNAR_PREFETCH_CACHE_HIT_HISTOGRAM: Histogram = register_histogram!( "kv_engine_columnar_prefetch_cache_hit", "Bucketed histogram of columnar prefetch cache hit", linear_buckets(50.0, 2.0, 25).unwrap() ) .unwrap(); pub static ref COLUMNAR_FETCH_SNAPSHOT_HISTOGRAM: Histogram = register_histogram!( "kv_engine_columnar_fetch_snapshot_duration_seconds", "Bucketed histogram of columnar fetch snapshot duration", exponential_buckets(0.1, 2.0, 20).unwrap() ) .unwrap(); pub static ref COLUMNAR_FETCH_SNAPSHOT_RETRY_COUNTER: Counter = register_counter!( "kv_engine_columnar_fetch_snapshot_retry_count", "Count of columnar fetch snapshot retry", ) .unwrap(); pub static ref ENGINE_INGEST_LEVEL_HISTOGRAM: Histogram = register_histogram!( "kv_engine_ingest_level", "Histogram of levels where tables are ingested", vec![0.0, 1.0, 2.0, 3.0] ).unwrap(); pub static ref ENGINE_PREPARE_LOAD_REMOTE_FILE: IntCounter = register_int_counter!( "kv_engine_prepare_load_remote_file", "Total number of remote files loaded during preparing changeset" ).unwrap(); pub static ref ENGINE_PREPARE_USE_LOCAL_FILE: IntCounter = register_int_counter!( "kv_engine_prepare_use_local_file", "Total number local file hit during preparing changeset" ).unwrap(); pub static ref ENGINE_SST_CRYPT_DURATION_VEC: HistogramVec = register_histogram_vec!( "kv_engine_sst_crypt_duration_seconds", "Bucketed histogram of KV Engine SST encryption/decryption duration", &["type"], exponential_buckets(0.00005, 1.8, 26).unwrap() ).unwrap(); pub static ref ENGINE_SST_CRYPT_DURATION_STATIC: SSTCryptDurationVec = auto_flush_from!(ENGINE_SST_CRYPT_DURATION_VEC, SSTCryptDurationVec); } /// Records one high-level DFS read used to materialize a file locally. /// /// `bytes` is the logical payload returned by DFS, not physical object-store /// traffic. A successful transport read is recorded as success even if later /// file validation fails. pub(crate) fn observe_materialization_remote_read( io_type: file_system::IoType, bytes: Option, elapsed: Duration, ) { let result = if bytes.is_some() { "success" } else { "error" }; ENGINE_MATERIALIZATION_REMOTE_READ_DURATION .with_label_values(&[io_type.as_str(), result]) .observe(elapsed.as_secs_f64()); if let Some(bytes) = bytes { ENGINE_MATERIALIZATION_REMOTE_LOGICAL_READ_BYTES .with_label_values(&[io_type.as_str()]) .inc_by(bytes as u64); } } // Keep compaction metrics in their own `lazy_static!` block to keep the // TT-muncher recursion depth within the default limit. lazy_static! { pub static ref ENGINE_COMPACTION_TRIGGER_COUNTER: IntCounterVec = register_int_counter_vec!( "kv_engine_compaction_trigger_counter", "Total number of kvengine compaction with a label of compaction type triggered", &["type"] ).unwrap(); pub static ref ENGINE_COMPACTION_RESULT_COUNTER: IntCounterVec = register_int_counter_vec!( "kv_engine_compaction_result_counter", "Total number of kvengine compaction with a label of compaction result", &["result"] ).unwrap(); pub static ref ENGINE_COMPACTION_PROCESS_BYTES_TOTAL: IntCounterVec = register_int_counter_vec!( "kv_engine_compaction_process_bytes_total", "Total bytes processed by compaction", &["type", "kind", "format"] ) .unwrap(); pub static ref ENGINE_COMPACTION_PROCESS_ROWS_TOTAL: IntCounterVec = register_int_counter_vec!( "kv_engine_compaction_process_rows_total", "Total rows processed by compaction", &["type", "kind", "format"] ) .unwrap(); pub static ref ENGINE_COMPACTION_TASK_HANDLED_BYTES_THROUGHPUT: HistogramVec = register_histogram_vec!( "kv_engine_compaction_task_handled_bytes_throughput", "Per-task compaction handled input kv size (uncompressed) throughput in bytes per second", &["type"], exponential_buckets(1024.0, 2.0, 22).unwrap() ) .unwrap(); pub static ref ENGINE_COMPACTION_TASK_HANDLED_ROWS_THROUGHPUT: HistogramVec = register_histogram_vec!( "kv_engine_compaction_task_handled_rows_throughput", "Per-task compaction handled input rows throughput in rows per second", &["type"], exponential_buckets(1024.0, 2.0, 22).unwrap() ) .unwrap(); } // Keep file-GC metrics in their own `lazy_static!` block. The macro is a // TT-muncher whose recursion depth grows with the item count, so folding these // back into the big block above pushes `opts!` past the default recursion // limit. A separate block avoids a crate-wide `#![recursion_limit]` bump. lazy_static! { pub static ref ENGINE_FILE_GC_COMPACTION_TRIGGER_SKIP_COUNTER: IntCounterVec = register_int_counter_vec!( "kv_engine_file_gc_compaction_trigger_skip_counter", "Total number of file-level GC trigger checks that did not fire, by skip reason", &["reason"] ).unwrap(); pub static ref ENGINE_FILE_GC_COMPACTION_RATIO_HISTOGRAM: Histogram = register_histogram!( "kv_engine_file_gc_compaction_ratio", "Distribution of the file-level reclaimable-entries ratio at the moment a file-level GC compaction is triggered", // Anchor at 0.0 to keep resolution if the ratio threshold is // lowered. linear_buckets(0.0, 0.05, 21).unwrap() ).unwrap(); // File-GC obsolete-file distribution gauges, set by // collect_store_usage_stats. Each bucket value is the number of WRITE_CF // obsolete files whose attribute falls in that bucket. pub static ref ENGINE_FILE_GC_OBSOLETE_SIZE: IntGaugeVec = register_int_gauge_vec!( "kv_engine_file_gc_obsolete_size", "Number of WRITE_CF obsolete files bucketed by old-data-size; non-zero counts in the high buckets are few-but-large files holding large disk", &["size_bucket"] ).unwrap(); pub static ref ENGINE_FILE_GC_OBSOLETE_RATIO: IntGaugeVec = register_int_gauge_vec!( "kv_engine_file_gc_obsolete_ratio", "Number of WRITE_CF obsolete files bucketed by obsolete ratio; read file_gc_compaction_ratio as a line, files left of it are stuck on the ratio gate", &["ratio_bucket"] ).unwrap(); pub static ref ENGINE_FILE_GC_OBSOLETE_OVERLAP: IntGaugeVec = register_int_gauge_vec!( "kv_engine_file_gc_obsolete_overlap", "Number of WRITE_CF ratio-passing obsolete files bucketed by overlap ratio; read file_gc_compaction_max_overlap_ratio as a line, files right of it are stuck on the overlap gate", &["overlap_bucket"] ).unwrap(); } // Keep columnar meta cache metrics in their own `lazy_static!` block. // A separate block avoids a crate-wide `#![recursion_limit]` bump. lazy_static! { pub static ref ENGINE_COLUMNAR_META_CACHE_HIT: IntCounter = register_int_counter!( "kv_engine_columnar_meta_cache_hit", "Total number of columnar TableMeta cache hit", ) .unwrap(); pub static ref ENGINE_COLUMNAR_META_CACHE_MISS: IntCounter = register_int_counter!( "kv_engine_columnar_meta_cache_miss", "Total number of columnar TableMeta cache miss", ) .unwrap(); pub static ref ENGINE_COLUMNAR_META_CACHE_PARSE: IntCounter = register_int_counter!( "kv_engine_columnar_meta_cache_parse", "Successful TableMeta fills via columnar meta cache miss path", ) .unwrap(); pub static ref ENGINE_COLUMNAR_META_CACHE_ENTRIES: IntGauge = register_int_gauge!( "kv_engine_columnar_meta_cache_entries", "Approximate number of live columnar TableMeta cache entries in this process", ) .unwrap(); pub static ref ENGINE_COLUMNAR_META_CACHE_WEIGHTED_SIZE: IntGauge = register_int_gauge!( "kv_engine_columnar_meta_cache_weighted_size", "Approximate sum of weighter weights for live columnar TableMeta cache entries in this process", ) .unwrap(); pub static ref ENGINE_COLUMNAR_META_CACHE_REPORTED_SIZE_BYTES: IntCounter = register_int_counter!( "kv_engine_columnar_meta_cache_reported_size_bytes", "Sum of TableMeta.size on each successful meta cache fill", ) .unwrap(); } pub(crate) fn elapsed_secs(t: Instant) -> f64 { let d = Instant::now().saturating_duration_since(t); let nanos = f64::from(d.subsec_nanos()); d.as_secs() as f64 + (nanos / 1_000_000_000.0) } make_static_metric! { pub label_enum PrepareMetric { total_files, local_files, local_ia_files, local_auto_ia_files, remote_files, remote_ia_files, remote_auto_ia_files, tiny_meta_hit, tiny_meta_segment_offsets_hit, load_parent_shards, load_shards, } pub struct PrepareCounterVec: IntCounter { "metric" => PrepareMetric, } pub label_enum MetaPackActionMetric { pack, compact, } pub struct MetaPackActionCounterVec: IntCounter { "metric" => MetaPackActionMetric, } } lazy_static! { pub static ref PREPARE_COUNTER_VEC: PrepareCounterVec = register_static_int_counter_vec!( PrepareCounterVec, "kv_engine_prepare_counter", "Total number of prepare table file operations", &["metric"] ) .unwrap(); pub static ref META_PACK_ACTION_COUNTER_VEC: MetaPackActionCounterVec = register_static_int_counter_vec!( MetaPackActionCounterVec, "kv_engine_meta_packer_action_counter", "Total number of meta pack operations", &["metric"] ) .unwrap(); pub static ref META_PACKER_METAS_COUNT: IntGauge = register_int_gauge!( "kv_engine_meta_packer_metas_count", "Number of metas in meta packer", ) .unwrap(); pub static ref HALF_SPLIT_TABLES_HISTOGRAM: Histogram = register_histogram!( "kv_engine_half_split_tables", "Number of overlapping non-empty loaded WRITE SSTables examined for a half split", exponential_buckets(1.0, 2.0, 20).unwrap() ) .unwrap(); }