risingwave_meta/hummock/
metrics_utils.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use 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        // sub level stat
218        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    // table change log
489    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
515// Triggers a report on compact_pending_bytes_needed
516pub 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    // compact_pending_bytes
524    // we don't actually generate a compaction task here so developer config can be ignored.
525    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        // compact_level_compression_ratio
542        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}