1use std::collections::{BTreeMap, HashMap, HashSet};
16use std::sync::Arc;
17use std::sync::atomic::Ordering;
18use std::time::{SystemTime, UNIX_EPOCH};
19
20use itertools::{Itertools, enumerate};
21use prometheus::IntGauge;
22use prometheus::core::{AtomicU64, GenericCounter};
23use risingwave_hummock_sdk::compact_task::CompactTask;
24use risingwave_hummock_sdk::compaction_group::hummock_version_ext::object_size_map;
25use risingwave_hummock_sdk::level::Levels;
26use risingwave_hummock_sdk::table_stats::PbTableStatsMap;
27use risingwave_hummock_sdk::version::HummockVersion;
28use risingwave_hummock_sdk::{CompactionGroupId, HummockContextId, HummockVersionId};
29use risingwave_pb::hummock::write_limits::WriteLimit;
30use risingwave_pb::hummock::{
31 CompactionConfig, HummockPinnedVersion, HummockVersionStats, LevelType,
32};
33
34use super::compaction::selector::DynamicLevelSelectorCore;
35use super::compaction::{CompactionDeveloperConfig, get_compression_algorithm};
36use crate::hummock::checkpoint::HummockVersionCheckpoint;
37use crate::hummock::compaction::CompactStatus;
38use crate::rpc::metrics::MetaMetrics;
39
40pub struct LocalTableMetrics {
41 total_key_count: IntGauge,
42 total_key_size: IntGauge,
43 total_value_size: IntGauge,
44 write_throughput: GenericCounter<AtomicU64>,
45 cal_count: usize,
46 write_size: u64,
47}
48
49const MIN_FLUSH_COUNT: usize = 16;
50const MIN_FLUSH_DATA_SIZE: u64 = 128 * 1024 * 1024;
51
52impl LocalTableMetrics {
53 pub fn inc_write_throughput(&mut self, val: u64) {
54 self.write_size += val;
55 self.cal_count += 1;
56 if self.write_size > MIN_FLUSH_DATA_SIZE || self.cal_count > MIN_FLUSH_COUNT {
57 self.write_throughput.inc_by(self.write_size / 1024 / 1024);
58 self.write_size = 0;
59 self.cal_count = 0;
60 }
61 }
62}
63
64pub fn get_or_create_local_table_stat<'a>(
65 metrics: &MetaMetrics,
66 table_id: u32,
67 local_metrics: &'a mut HashMap<u32, LocalTableMetrics>,
68) -> &'a mut LocalTableMetrics {
69 local_metrics.entry(table_id).or_insert_with(|| {
70 let table_label = format!("{}", table_id);
71 LocalTableMetrics {
72 total_key_count: metrics
73 .version_stats
74 .with_label_values(&[&table_label, "total_key_count"]),
75 total_key_size: metrics
76 .version_stats
77 .with_label_values(&[&table_label, "total_key_size"]),
78 total_value_size: metrics
79 .version_stats
80 .with_label_values(&[&table_label, "total_value_size"]),
81 write_throughput: metrics
82 .table_write_throughput
83 .with_label_values(&[&table_label]),
84 cal_count: 0,
85 write_size: 0,
86 }
87 })
88}
89
90pub fn trigger_local_table_stat(
91 metrics: &MetaMetrics,
92 local_metrics: &mut HashMap<u32, LocalTableMetrics>,
93 version_stats: &HummockVersionStats,
94 table_stats_change: &PbTableStatsMap,
95) {
96 for (table_id, stats) in table_stats_change {
97 if stats.total_key_size == 0 && stats.total_value_size == 0 && stats.total_key_count == 0 {
98 continue;
99 }
100 let table_metrics = get_or_create_local_table_stat(metrics, *table_id, local_metrics);
101 if let Some(table_stats) = version_stats.table_stats.get(table_id) {
102 table_metrics
103 .total_key_count
104 .set(table_stats.total_key_count);
105 table_metrics
106 .total_value_size
107 .set(table_stats.total_value_size);
108 table_metrics.total_key_size.set(table_stats.total_key_size);
109 }
110 }
111}
112
113pub fn trigger_mv_stat(
114 metrics: &MetaMetrics,
115 version_stats: &HummockVersionStats,
116 mv_id_to_all_table_ids: Vec<(u32, Vec<u32>)>,
117) {
118 metrics.materialized_view_stats.reset();
119 for (mv_id, all_table_ids) in mv_id_to_all_table_ids {
120 let total_size = all_table_ids
121 .iter()
122 .filter_map(|&table_id| version_stats.table_stats.get(&table_id))
123 .map(|stats| stats.total_key_size + stats.total_value_size)
124 .sum();
125
126 metrics
127 .materialized_view_stats
128 .with_label_values(&[&mv_id.to_string(), "materialized_view_total_size"])
129 .set(total_size);
130 }
131}
132
133pub fn trigger_sst_stat(
134 metrics: &MetaMetrics,
135 compact_status: Option<&CompactStatus>,
136 current_version: &HummockVersion,
137 compaction_group_id: CompactionGroupId,
138) {
139 let level_sst_cnt = |level_idx: usize| {
140 let mut sst_num = 0;
141 current_version.level_iter(compaction_group_id, |level| {
142 if level.level_idx == level_idx as u32 {
143 sst_num += level.table_infos.len();
144 }
145 true
146 });
147 sst_num
148 };
149 let level_sst_size = |level_idx: usize| {
150 let mut level_sst_size = 0;
151 current_version.level_iter(compaction_group_id, |level| {
152 if level.level_idx == level_idx as u32 {
153 level_sst_size += level.total_file_size;
154 }
155 true
156 });
157 level_sst_size / 1024
158 };
159
160 let mut compacting_task_stat: BTreeMap<(usize, usize), usize> = BTreeMap::default();
161 for idx in 0..current_version.num_levels(compaction_group_id) {
162 let sst_num = level_sst_cnt(idx);
163 let level_label = build_level_metrics_label(compaction_group_id, idx);
164 metrics
165 .level_sst_num
166 .with_label_values(&[&level_label])
167 .set(sst_num as i64);
168 metrics
169 .level_file_size
170 .with_label_values(&[&level_label])
171 .set(level_sst_size(idx) as i64);
172 if let Some(compact_status) = compact_status {
173 let compact_cnt = compact_status.level_handlers[idx].pending_file_count();
174 metrics
175 .level_compact_cnt
176 .with_label_values(&[&level_label])
177 .set(compact_cnt as i64);
178
179 let compacting_task = compact_status.level_handlers[idx].pending_tasks();
180 let mut pending_task_ids: HashSet<u64> = HashSet::default();
181 for task in compacting_task {
182 if pending_task_ids.contains(&task.task_id) {
183 continue;
184 }
185
186 if idx != 0 && idx == task.target_level as usize {
187 continue;
188 }
189
190 let key = (idx, task.target_level as usize);
191 let count = compacting_task_stat.entry(key).or_insert(0);
192 *count += 1;
193
194 pending_task_ids.insert(task.task_id);
195 }
196 }
197 }
198
199 tracing::debug!("LSM Compacting STAT {:?}", compacting_task_stat);
200 for ((select, target), compacting_task_count) in &compacting_task_stat {
201 let label_str =
202 build_compact_task_stat_metrics_label(compaction_group_id, *select, *target);
203 metrics
204 .level_compact_task_cnt
205 .with_label_values(&[&label_str])
206 .set(*compacting_task_count as _);
207 }
208
209 if compacting_task_stat.is_empty() {
210 if let Some(levels) = current_version.levels.get(&compaction_group_id) {
211 let max_level = levels.levels.len();
212 remove_compacting_task_stat(metrics, compaction_group_id, max_level);
213 }
214 }
215
216 {
217 let overlapping_level_label =
219 build_compact_task_l0_stat_metrics_label(compaction_group_id, true, false);
220 let non_overlap_level_label =
221 build_compact_task_l0_stat_metrics_label(compaction_group_id, false, false);
222 let partition_level_label =
223 build_compact_task_l0_stat_metrics_label(compaction_group_id, true, true);
224
225 let overlapping_sst_num = current_version
226 .levels
227 .get(&compaction_group_id)
228 .map(|level| {
229 level
230 .l0
231 .sub_levels
232 .iter()
233 .filter(|sub_level| sub_level.level_type == LevelType::Overlapping)
234 .count()
235 })
236 .unwrap_or(0);
237
238 let non_overlap_sst_num = current_version
239 .levels
240 .get(&compaction_group_id)
241 .map(|level| {
242 level
243 .l0
244 .sub_levels
245 .iter()
246 .filter(|sub_level| sub_level.level_type == LevelType::Nonoverlapping)
247 .count()
248 })
249 .unwrap_or(0);
250
251 let partition_level_num = current_version
252 .levels
253 .get(&compaction_group_id)
254 .map(|level| {
255 level
256 .l0
257 .sub_levels
258 .iter()
259 .filter(|sub_level| {
260 sub_level.level_type == LevelType::Nonoverlapping
261 && sub_level.vnode_partition_count > 0
262 })
263 .count()
264 })
265 .unwrap_or(0);
266 metrics
267 .level_sst_num
268 .with_label_values(&[&overlapping_level_label])
269 .set(overlapping_sst_num as i64);
270
271 metrics
272 .level_sst_num
273 .with_label_values(&[&non_overlap_level_label])
274 .set(non_overlap_sst_num as i64);
275
276 metrics
277 .level_sst_num
278 .with_label_values(&[&partition_level_label])
279 .set(partition_level_num as i64);
280 }
281
282 let previous_time = metrics.time_after_last_observation.load(Ordering::Relaxed);
283 let current_time = SystemTime::now()
284 .duration_since(UNIX_EPOCH)
285 .unwrap()
286 .as_secs();
287 if current_time > 600 + previous_time
288 && metrics
289 .time_after_last_observation
290 .compare_exchange(
291 previous_time,
292 current_time,
293 Ordering::Relaxed,
294 Ordering::Relaxed,
295 )
296 .is_ok()
297 {
298 if let Some(compact_status) = compact_status {
299 for (idx, level_handler) in enumerate(compact_status.level_handlers.iter()) {
300 let sst_num = level_sst_cnt(idx);
301 let sst_size = level_sst_size(idx);
302 let compact_cnt = level_handler.pending_file_count();
303 tracing::info!(
304 "Level {} has {} SSTs, the total size of which is {}KB, while {} of those are being compacted to bottom levels",
305 idx,
306 sst_num,
307 sst_size,
308 compact_cnt,
309 );
310 }
311 }
312 }
313}
314
315pub fn trigger_epoch_stat(metrics: &MetaMetrics, version: &HummockVersion) {
316 metrics.max_committed_epoch.set(
317 version
318 .state_table_info
319 .info()
320 .values()
321 .map(|i| i.committed_epoch)
322 .max()
323 .unwrap_or(0) as _,
324 );
325 metrics.min_committed_epoch.set(
326 version
327 .state_table_info
328 .info()
329 .values()
330 .map(|i| i.committed_epoch)
331 .min()
332 .unwrap_or(0) as _,
333 );
334}
335
336pub fn remove_compaction_group_in_sst_stat(
337 metrics: &MetaMetrics,
338 compaction_group_id: CompactionGroupId,
339 max_level: usize,
340) {
341 let mut idx = 0;
342 loop {
343 let level_label = build_level_metrics_label(compaction_group_id, idx);
344 let should_continue = metrics
345 .level_sst_num
346 .remove_label_values(&[&level_label])
347 .is_ok();
348
349 metrics
350 .level_file_size
351 .remove_label_values(&[&level_label])
352 .ok();
353
354 metrics
355 .level_compact_cnt
356 .remove_label_values(&[&level_label])
357 .ok();
358 if !should_continue {
359 break;
360 }
361 idx += 1;
362 }
363
364 let overlapping_level_label = build_level_l0_metrics_label(compaction_group_id, true);
365 let non_overlap_level_label = build_level_l0_metrics_label(compaction_group_id, false);
366 metrics
367 .level_sst_num
368 .remove_label_values(&[&overlapping_level_label])
369 .ok();
370 metrics
371 .level_sst_num
372 .remove_label_values(&[&non_overlap_level_label])
373 .ok();
374
375 remove_compacting_task_stat(metrics, compaction_group_id, max_level);
376 remove_split_stat(metrics, compaction_group_id);
377 remove_compact_task_metrics(metrics, compaction_group_id, max_level);
378 remove_compaction_group_stat(metrics, compaction_group_id);
379}
380
381pub fn remove_compacting_task_stat(
382 metrics: &MetaMetrics,
383 compaction_group_id: CompactionGroupId,
384 max_level: usize,
385) {
386 for select_level in 0..=max_level {
387 for target_level in 0..=max_level {
388 let label_str = build_compact_task_stat_metrics_label(
389 compaction_group_id,
390 select_level,
391 target_level,
392 );
393 metrics
394 .level_compact_task_cnt
395 .remove_label_values(&[&label_str])
396 .ok();
397 }
398 }
399}
400
401pub fn remove_split_stat(metrics: &MetaMetrics, compaction_group_id: CompactionGroupId) {
402 let label_str = compaction_group_id.to_string();
403 metrics
404 .state_table_count
405 .remove_label_values(&[&label_str])
406 .ok();
407
408 metrics
409 .branched_sst_count
410 .remove_label_values(&[&label_str])
411 .ok();
412}
413
414pub fn remove_compaction_group_stat(metrics: &MetaMetrics, compaction_group_id: CompactionGroupId) {
415 let label_str = compaction_group_id.to_string();
416 metrics
417 .compaction_group_size
418 .remove_label_values(&[&label_str])
419 .ok();
420 metrics
421 .compaction_group_file_count
422 .remove_label_values(&[&label_str])
423 .ok();
424 metrics
425 .compaction_group_throughput
426 .remove_label_values(&[&label_str])
427 .ok();
428
429 metrics
430 .split_compaction_group_count
431 .remove_label_values(&[&label_str])
432 .ok();
433
434 metrics
435 .merge_compaction_group_count
436 .remove_label_values(&[&label_str])
437 .ok();
438}
439
440pub fn trigger_pin_unpin_version_state(
441 metrics: &MetaMetrics,
442 pinned_versions: &BTreeMap<HummockContextId, HummockPinnedVersion>,
443) {
444 if let Some(m) = pinned_versions.values().map(|v| v.min_pinned_id).min() {
445 metrics.min_pinned_version_id.set(m as i64);
446 } else {
447 metrics
448 .min_pinned_version_id
449 .set(HummockVersionId::MAX.to_u64() as _);
450 }
451}
452
453pub fn trigger_gc_stat(
454 metrics: &MetaMetrics,
455 checkpoint: &HummockVersionCheckpoint,
456 min_pinned_version_id: HummockVersionId,
457) {
458 let current_version_object_size_map = object_size_map(&checkpoint.version);
459 let current_version_object_size = current_version_object_size_map.values().sum::<u64>();
460 let current_version_object_count = current_version_object_size_map.len();
461 let mut old_version_object_size = 0;
462 let mut old_version_object_count = 0;
463 let mut stale_object_size = 0;
464 let mut stale_object_count = 0;
465 checkpoint.stale_objects.iter().for_each(|(id, objects)| {
466 if *id <= min_pinned_version_id {
467 stale_object_size += objects.total_file_size;
468 stale_object_count += objects.id.len() as u64;
469 } else {
470 old_version_object_size += objects.total_file_size;
471 old_version_object_count += objects.id.len() as u64;
472 }
473 });
474 metrics
475 .current_version_object_size
476 .set(current_version_object_size as _);
477 metrics
478 .current_version_object_count
479 .set(current_version_object_count as _);
480 metrics
481 .old_version_object_size
482 .set(old_version_object_size as _);
483 metrics
484 .old_version_object_count
485 .set(old_version_object_count as _);
486 metrics.stale_object_size.set(stale_object_size as _);
487 metrics.stale_object_count.set(stale_object_count as _);
488 for (table_id, logs) in &checkpoint.version.table_change_log {
490 let object_count = logs
491 .iter()
492 .map(|l| l.old_value.len() + l.new_value.len())
493 .sum::<usize>();
494 let object_size = logs
495 .iter()
496 .map(|l| {
497 l.old_value
498 .iter()
499 .chain(l.new_value.iter())
500 .map(|s| s.file_size as usize)
501 .sum::<usize>()
502 })
503 .sum::<usize>();
504 metrics
505 .table_change_log_object_count
506 .with_label_values(&[&format!("{table_id}")])
507 .set(object_count as _);
508 metrics
509 .table_change_log_object_size
510 .with_label_values(&[&format!("{table_id}")])
511 .set(object_size as _);
512 }
513}
514
515pub fn trigger_lsm_stat(
517 metrics: &MetaMetrics,
518 compaction_config: Arc<CompactionConfig>,
519 levels: &Levels,
520 compaction_group_id: CompactionGroupId,
521) {
522 let group_label = compaction_group_id.to_string();
523 let dynamic_level_core = DynamicLevelSelectorCore::new(
526 compaction_config.clone(),
527 Arc::new(CompactionDeveloperConfig::default()),
528 );
529 let ctx = dynamic_level_core.calculate_level_base_size(levels);
530 {
531 let compact_pending_bytes_needed =
532 dynamic_level_core.compact_pending_bytes_needed_with_ctx(levels, &ctx);
533
534 metrics
535 .compact_pending_bytes
536 .with_label_values(&[&group_label])
537 .set(compact_pending_bytes_needed as _);
538 }
539
540 {
541 let level_compression_ratio = levels
543 .levels
544 .iter()
545 .map(|level| {
546 let ratio = if level.uncompressed_file_size == 0 {
547 0.0
548 } else {
549 level.total_file_size as f64 / level.uncompressed_file_size as f64
550 };
551
552 (level.level_idx, ratio)
553 })
554 .collect_vec();
555
556 for (level_index, compression_ratio) in level_compression_ratio {
557 let compression_algorithm_label = get_compression_algorithm(
558 compaction_config.as_ref(),
559 ctx.base_level,
560 level_index as usize,
561 );
562
563 metrics
564 .compact_level_compression_ratio
565 .with_label_values(&[
566 &group_label,
567 &level_index.to_string(),
568 &compression_algorithm_label,
569 ])
570 .set(compression_ratio);
571 }
572 }
573}
574
575pub fn trigger_write_stop_stats(
576 metrics: &MetaMetrics,
577 write_limit: &HashMap<CompactionGroupId, WriteLimit>,
578) {
579 metrics.write_stop_compaction_groups.reset();
580 for cg in write_limit.keys() {
581 metrics
582 .write_stop_compaction_groups
583 .with_label_values(&[&cg.to_string()])
584 .set(1);
585 }
586}
587
588pub fn trigger_split_stat(metrics: &MetaMetrics, version: &HummockVersion) {
589 let branched_ssts = version.build_branched_sst_info();
590
591 for compaction_group_id in version.levels.keys() {
592 let group_label = compaction_group_id.to_string();
593 metrics
594 .state_table_count
595 .with_label_values(&[&group_label])
596 .set(
597 version
598 .state_table_info
599 .compaction_group_member_table_ids(*compaction_group_id)
600 .len() as _,
601 );
602
603 let branched_sst_count: usize = branched_ssts
604 .values()
605 .map(|branched_map| {
606 branched_map
607 .keys()
608 .filter(|group_id| *group_id == compaction_group_id)
609 .count()
610 })
611 .sum();
612
613 metrics
614 .branched_sst_count
615 .with_label_values(&[&group_label])
616 .set(branched_sst_count as _);
617 }
618}
619
620pub fn build_level_metrics_label(compaction_group_id: u64, level_idx: usize) -> String {
621 format!("cg{}_L{}", compaction_group_id, level_idx)
622}
623
624pub fn build_level_l0_metrics_label(compaction_group_id: u64, overlapping: bool) -> String {
625 if overlapping {
626 format!("cg{}_l0_sub_overlapping", compaction_group_id)
627 } else {
628 format!("cg{}_l0_sub_non_overlap", compaction_group_id)
629 }
630}
631
632pub fn build_compact_task_stat_metrics_label(
633 compaction_group_id: u64,
634 select_level: usize,
635 target_level: usize,
636) -> String {
637 format!(
638 "cg{} L{} -> L{}",
639 compaction_group_id, select_level, target_level
640 )
641}
642
643pub fn build_compact_task_l0_stat_metrics_label(
644 compaction_group_id: u64,
645 overlapping: bool,
646 partition: bool,
647) -> String {
648 if partition {
649 format!("cg{}_l0_sub_partition", compaction_group_id)
650 } else if overlapping {
651 format!("cg{}_l0_sub_overlapping", compaction_group_id)
652 } else {
653 format!("cg{}_l0_sub_non_overlap", compaction_group_id)
654 }
655}
656
657pub fn build_compact_task_level_type_metrics_label(
658 select_level: usize,
659 target_level: usize,
660) -> String {
661 format!("L{}->L{}", select_level, target_level)
662}
663
664pub fn remove_compact_task_metrics(
665 metrics: &MetaMetrics,
666 compaction_group_id: CompactionGroupId,
667 max_level: usize,
668) {
669 for select_level in 0..=max_level {
670 for target_level in 0..=max_level {
671 let level_type_label =
672 build_compact_task_level_type_metrics_label(select_level, target_level);
673 let should_continue = metrics
674 .l0_compact_level_count
675 .remove_label_values(&[&compaction_group_id.to_string(), &level_type_label])
676 .is_ok();
677
678 metrics
679 .compact_task_size
680 .remove_label_values(&[&compaction_group_id.to_string(), &level_type_label])
681 .ok();
682
683 metrics
684 .compact_task_size
685 .remove_label_values(&[
686 &compaction_group_id.to_string(),
687 &format!("{} uncompressed", level_type_label),
688 ])
689 .ok();
690 if !should_continue {
691 break;
692 }
693 }
694 }
695}
696
697pub fn trigger_compact_tasks_stat(
698 metrics: &MetaMetrics,
699 compact_tasks: &[CompactTask],
700 compact_status: &BTreeMap<CompactionGroupId, CompactStatus>,
701 current_version: &HummockVersion,
702) {
703 let mut task_status_label_map = HashMap::new();
704 let mut task_type_label_map = HashMap::new();
705 let mut group_label_map = HashMap::new();
706
707 for task in compact_tasks {
708 let task_status_label = task_status_label_map
709 .entry(task.task_status)
710 .or_insert_with(|| task.task_status.as_str_name().to_owned());
711
712 let task_type_label = task_type_label_map
713 .entry(task.task_type)
714 .or_insert_with(|| task.task_type.as_str_name().to_owned());
715
716 let group_label = group_label_map
717 .entry(task.compaction_group_id)
718 .or_insert_with(|| task.compaction_group_id.to_string());
719
720 metrics
721 .compact_frequency
722 .with_label_values(&["normal", group_label, task_type_label, task_status_label])
723 .inc();
724 }
725
726 group_label_map.keys().for_each(|group_id| {
727 trigger_sst_stat(
728 metrics,
729 compact_status.get(group_id),
730 current_version,
731 *group_id,
732 );
733 });
734}