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