1use std::sync::{Arc, OnceLock};
16
17use prometheus::core::{AtomicU64, Collector, Desc, GenericCounter};
18use prometheus::{
19 Gauge, Histogram, HistogramVec, IntGauge, IntGaugeVec, Opts, Registry, exponential_buckets,
20 histogram_opts, proto, register_histogram_vec_with_registry, register_histogram_with_registry,
21 register_int_counter_vec_with_registry, register_int_gauge_vec_with_registry,
22 register_int_gauge_with_registry,
23};
24use risingwave_common::config::MetricLevel;
25use risingwave_common::metrics::{
26 RelabeledCounterVec, RelabeledGuardedHistogramVec, RelabeledGuardedIntCounterVec,
27 RelabeledGuardedIntGaugeVec, RelabeledHistogramVec, RelabeledMetricVec, UintGauge,
28};
29use risingwave_common::monitor::GLOBAL_METRICS_REGISTRY;
30use risingwave_common::{
31 register_guarded_histogram_vec_with_registry, register_guarded_int_counter_vec_with_registry,
32 register_guarded_int_gauge_vec_with_registry,
33};
34use thiserror_ext::AsReport;
35use tracing::warn;
36
37#[derive(Debug, Clone)]
43pub struct HummockStateStoreMetrics {
44 pub bloom_filter_true_negative_counts: RelabeledGuardedIntCounterVec,
45 pub bloom_filter_check_counts: RelabeledGuardedIntCounterVec,
46 pub iter_merge_sstable_counts: RelabeledHistogramVec,
47 pub vnode_pruning_counts: RelabeledGuardedIntCounterVec,
48 pub sst_store_block_request_counts: RelabeledGuardedIntCounterVec,
49 pub iter_scan_key_counts: RelabeledGuardedIntCounterVec,
50 pub get_shared_buffer_hit_counts: RelabeledCounterVec,
51 pub remote_read_time: RelabeledHistogramVec,
52 pub iter_fetch_meta_duration: RelabeledGuardedHistogramVec,
53 pub iter_fetch_meta_cache_unhits: IntGauge,
54 pub iter_slow_fetch_meta_cache_unhits: IntGauge,
55
56 pub vector_object_request_counts: RelabeledGuardedIntCounterVec,
57 pub vector_request_stats: RelabeledGuardedHistogramVec,
58 pub vector_hnsw_graph_level_node_count: RelabeledGuardedIntGaugeVec,
59 pub vector_index_file_count: RelabeledGuardedIntGaugeVec,
60 pub vector_index_file_size: RelabeledGuardedIntGaugeVec,
61
62 pub read_req_bloom_filter_positive_counts: RelabeledGuardedIntCounterVec,
63 pub read_req_positive_but_non_exist_counts: RelabeledGuardedIntCounterVec,
64 pub read_req_check_bloom_filter_counts: RelabeledGuardedIntCounterVec,
65
66 pub write_batch_tuple_counts: RelabeledGuardedIntCounterVec,
67 pub write_batch_duration: RelabeledGuardedHistogramVec,
68 pub write_batch_size: RelabeledGuardedHistogramVec,
69
70 pub spill_task_counts_from_unsealed: GenericCounter<AtomicU64>,
72 pub spill_task_size_from_unsealed: GenericCounter<AtomicU64>,
74
75 pub uploader_uploading_task_size: UintGauge,
77 pub uploader_uploading_task_count: IntGauge,
78 pub uploader_imm_size: UintGauge,
79 pub uploader_upload_task_latency: Histogram,
80 pub uploader_syncing_epoch_count: IntGauge,
81 pub uploader_wait_poll_latency: Histogram,
82 pub uploader_per_table_imm_size: RelabeledGuardedIntGaugeVec,
83 pub uploader_per_table_imm_count: RelabeledGuardedIntGaugeVec,
84
85 pub per_table_imm_size: RelabeledGuardedIntGaugeVec,
87 pub per_table_imm_count: RelabeledGuardedIntGaugeVec,
88 pub mem_table_spill_counts: RelabeledGuardedIntCounterVec,
89 pub old_value_size: RelabeledGuardedIntGaugeVec,
90
91 pub block_efficiency_histogram: Histogram,
93
94 pub event_handler_pending_event: IntGaugeVec,
95 pub event_handler_latency: HistogramVec,
96
97 pub safe_version_hit: GenericCounter<AtomicU64>,
98 pub safe_version_miss: GenericCounter<AtomicU64>,
99 pub table_change_log_fetch_latency: Histogram,
100 pub table_change_log_cache_hit: GenericCounter<AtomicU64>,
101 pub table_change_log_cache_miss: GenericCounter<AtomicU64>,
102}
103
104pub static GLOBAL_HUMMOCK_STATE_STORE_METRICS: OnceLock<HummockStateStoreMetrics> = OnceLock::new();
105
106pub fn global_hummock_state_store_metrics(metric_level: MetricLevel) -> HummockStateStoreMetrics {
107 GLOBAL_HUMMOCK_STATE_STORE_METRICS
108 .get_or_init(|| HummockStateStoreMetrics::new(&GLOBAL_METRICS_REGISTRY, metric_level))
109 .clone()
110}
111
112impl HummockStateStoreMetrics {
113 pub fn new(registry: &Registry, metric_level: MetricLevel) -> Self {
114 let time_buckets = exponential_buckets(0.01, 10.0, 7).unwrap();
116
117 let state_store_read_time_buckets = exponential_buckets(0.001, 10.0, 5).unwrap();
119
120 let bloom_filter_true_negative_counts = register_guarded_int_counter_vec_with_registry!(
121 "state_store_bloom_filter_true_negative_counts",
122 "Total number of sstables that have been considered true negative by SST filters",
123 &["table_id", "type"],
124 registry
125 )
126 .unwrap();
127 let bloom_filter_true_negative_counts = RelabeledMetricVec::with_metric_level(
128 MetricLevel::Debug,
129 bloom_filter_true_negative_counts,
130 metric_level,
131 );
132
133 let bloom_filter_check_counts = register_guarded_int_counter_vec_with_registry!(
134 "state_store_bloom_filter_check_counts",
135 "Total number of read request to check SST filters",
136 &["table_id", "type"],
137 registry
138 )
139 .unwrap();
140 let bloom_filter_check_counts = RelabeledMetricVec::with_metric_level(
141 MetricLevel::Debug,
142 bloom_filter_check_counts,
143 metric_level,
144 );
145
146 let opts = histogram_opts!(
148 "state_store_iter_merge_sstable_counts",
149 "Number of child iterators merged into one MergeIterator",
150 vec![1.0, 10.0, 100.0, 1000.0, 10000.0]
151 );
152 let iter_merge_sstable_counts =
153 register_histogram_vec_with_registry!(opts, &["table_id", "type"], registry).unwrap();
154 let iter_merge_sstable_counts = RelabeledHistogramVec::with_metric_level(
155 MetricLevel::Debug,
156 iter_merge_sstable_counts,
157 metric_level,
158 );
159
160 let vnode_pruning_counts = register_guarded_int_counter_vec_with_registry!(
161 "state_store_vnode_pruning_counts",
162 "Total number of SST pruning operations by vnode key range hints",
163 &["table_id", "operation", "result"],
164 registry
165 )
166 .unwrap();
167
168 let vnode_pruning_counts = RelabeledGuardedIntCounterVec::with_metric_level(
169 MetricLevel::Debug,
170 vnode_pruning_counts,
171 metric_level,
172 );
173
174 let sst_store_block_request_counts = register_guarded_int_counter_vec_with_registry!(
176 "state_store_sst_store_block_request_counts",
177 "Total number of sst block requests that have been issued to sst store",
178 &["table_id", "type"],
179 registry
180 )
181 .unwrap();
182 let sst_store_block_request_counts = RelabeledGuardedIntCounterVec::with_metric_level(
183 MetricLevel::Info,
184 sst_store_block_request_counts,
185 metric_level,
186 );
187
188 let iter_scan_key_counts = register_guarded_int_counter_vec_with_registry!(
189 "state_store_iter_scan_key_counts",
190 "Total number of keys read by iterator",
191 &["table_id", "type"],
192 registry
193 )
194 .unwrap();
195 let iter_scan_key_counts = RelabeledGuardedIntCounterVec::with_metric_level(
196 MetricLevel::Info,
197 iter_scan_key_counts,
198 metric_level,
199 );
200
201 let get_shared_buffer_hit_counts = register_int_counter_vec_with_registry!(
202 "state_store_get_shared_buffer_hit_counts",
203 "Total number of get requests that have been fulfilled by shared buffer",
204 &["table_id"],
205 registry
206 )
207 .unwrap();
208 let get_shared_buffer_hit_counts = RelabeledCounterVec::with_metric_level(
209 MetricLevel::Debug,
210 get_shared_buffer_hit_counts,
211 metric_level,
212 );
213
214 let opts = histogram_opts!(
215 "state_store_remote_read_time_per_task",
216 "Total time of operations which read from remote storage when enable prefetch",
217 time_buckets.clone(),
218 );
219 let remote_read_time =
220 register_histogram_vec_with_registry!(opts, &["table_id"], registry).unwrap();
221 let remote_read_time = RelabeledHistogramVec::with_metric_level(
222 MetricLevel::Debug,
223 remote_read_time,
224 metric_level,
225 );
226
227 let opts = histogram_opts!(
228 "state_store_iter_fetch_meta_duration",
229 "Histogram of iterator fetch SST meta time that have been issued to state store",
230 state_store_read_time_buckets.clone(),
231 );
232 let iter_fetch_meta_duration =
233 register_guarded_histogram_vec_with_registry!(opts, &["table_id"], registry).unwrap();
234 let iter_fetch_meta_duration = RelabeledGuardedHistogramVec::with_metric_level(
235 MetricLevel::Info,
236 iter_fetch_meta_duration,
237 metric_level,
238 );
239
240 let iter_fetch_meta_cache_unhits = register_int_gauge_with_registry!(
241 "state_store_iter_fetch_meta_cache_unhits",
242 "Number of SST meta cache unhit during one iterator meta fetch",
243 registry
244 )
245 .unwrap();
246
247 let iter_slow_fetch_meta_cache_unhits = register_int_gauge_with_registry!(
248 "state_store_iter_slow_fetch_meta_cache_unhits",
249 "Number of SST meta cache unhit during a iterator meta fetch which is slow (costs >5 seconds)",
250 registry
251 )
252 .unwrap();
253
254 let opts = histogram_opts!(
255 "state_store_table_change_log_fetch_latency",
256 "Latency of fetching table change logs",
257 state_store_read_time_buckets,
258 );
259 let table_change_log_fetch_latency =
260 register_histogram_with_registry!(opts, registry).unwrap();
261
262 let table_change_log_cache_counts = register_int_counter_vec_with_registry!(
263 "state_store_table_change_log_cache_counts",
264 "Total number of table change log cache lookups",
265 &["result"],
266 registry
267 )
268 .unwrap();
269 let table_change_log_cache_hit = table_change_log_cache_counts.with_label_values(&["hit"]);
270 let table_change_log_cache_miss =
271 table_change_log_cache_counts.with_label_values(&["miss"]);
272
273 let vector_object_request_counts = register_guarded_int_counter_vec_with_registry!(
275 "state_store_vector_object_request_counts",
276 "Metrics about vector object requests that have been issued",
277 &["table_id", "type", "mode"],
278 registry
279 )
280 .unwrap();
281 let vector_object_request_counts = RelabeledGuardedIntCounterVec::with_metric_level(
282 MetricLevel::Critical,
283 vector_object_request_counts,
284 metric_level,
285 );
286
287 let opts = histogram_opts!(
288 "state_store_vector_request_stats",
289 "Metrics about vector requests",
290 exponential_buckets(100.0, 10.0, 5).unwrap(),
291 );
292
293 let vector_request_stats = register_guarded_histogram_vec_with_registry!(
294 opts,
295 &["table_id", "type", "mode", "top_n", "ef"],
296 registry
297 )
298 .unwrap();
299 let vector_request_stats = RelabeledGuardedHistogramVec::with_metric_level(
300 MetricLevel::Critical,
301 vector_request_stats,
302 metric_level,
303 );
304
305 let vector_hnsw_graph_level_node_count = register_guarded_int_gauge_vec_with_registry!(
306 "state_store_vector_hnsw_graph_level_node_count",
307 "Number of nodes in each level of hnsw graph",
308 &["table_id", "level"],
309 registry
310 )
311 .unwrap();
312 let vector_hnsw_graph_level_node_count = RelabeledGuardedIntGaugeVec::with_metric_level(
313 MetricLevel::Critical,
314 vector_hnsw_graph_level_node_count,
315 metric_level,
316 );
317
318 let vector_index_file_count = register_guarded_int_gauge_vec_with_registry!(
319 "state_store_vector_index_file_count",
320 "Number of vector file",
321 &["table_id"],
322 registry
323 )
324 .unwrap();
325 let vector_index_file_count = RelabeledGuardedIntGaugeVec::with_metric_level(
326 MetricLevel::Critical,
327 vector_index_file_count,
328 metric_level,
329 );
330
331 let vector_index_file_size = register_guarded_int_gauge_vec_with_registry!(
332 "state_store_vector_index_file_size",
333 "total size of vector index file",
334 &["table_id", "type"],
335 registry
336 )
337 .unwrap();
338 let vector_index_file_size = RelabeledGuardedIntGaugeVec::with_metric_level(
339 MetricLevel::Critical,
340 vector_index_file_size,
341 metric_level,
342 );
343
344 let write_batch_tuple_counts = register_guarded_int_counter_vec_with_registry!(
346 "state_store_write_batch_tuple_counts",
347 "Total number of batched write kv pairs requests that have been issued to state store",
348 &["table_id"],
349 registry
350 )
351 .unwrap();
352 let write_batch_tuple_counts = RelabeledGuardedIntCounterVec::with_metric_level(
353 MetricLevel::Debug,
354 write_batch_tuple_counts,
355 metric_level,
356 );
357
358 let opts = histogram_opts!(
359 "state_store_write_batch_duration",
360 "Total time of batched write that have been issued to state store. With shared buffer on, this is the latency writing to the shared buffer",
361 time_buckets.clone()
362 );
363 let write_batch_duration =
364 register_guarded_histogram_vec_with_registry!(opts, &["table_id"], registry).unwrap();
365 let write_batch_duration = RelabeledGuardedHistogramVec::with_metric_level(
366 MetricLevel::Debug,
367 write_batch_duration,
368 metric_level,
369 );
370
371 let opts = histogram_opts!(
372 "state_store_write_batch_size",
373 "Total size of batched write that have been issued to state store",
374 exponential_buckets(256.0, 16.0, 7).unwrap() );
376 let write_batch_size =
377 register_guarded_histogram_vec_with_registry!(opts, &["table_id"], registry).unwrap();
378 let write_batch_size = RelabeledGuardedHistogramVec::with_metric_level(
379 MetricLevel::Debug,
380 write_batch_size,
381 metric_level,
382 );
383
384 let spill_task_counts = register_int_counter_vec_with_registry!(
385 "state_store_spill_task_counts",
386 "Total number of started spill tasks",
387 &["uploader_stage"],
388 registry
389 )
390 .unwrap();
391
392 let spill_task_size = register_int_counter_vec_with_registry!(
393 "state_store_spill_task_size",
394 "Total task of started spill tasks",
395 &["uploader_stage"],
396 registry
397 )
398 .unwrap();
399
400 let uploader_uploading_task_size = UintGauge::new(
401 "state_store_uploader_uploading_task_size",
402 "Total size of uploader uploading tasks",
403 )
404 .unwrap();
405 registry
406 .register(Box::new(uploader_uploading_task_size.clone()))
407 .unwrap();
408
409 let uploader_uploading_task_count = register_int_gauge_with_registry!(
410 "state_store_uploader_uploading_task_count",
411 "Total number of uploader uploading tasks",
412 registry
413 )
414 .unwrap();
415
416 let uploader_imm_size = UintGauge::new(
417 "state_store_uploader_imm_size",
418 "Total size of imms tracked by uploader",
419 )
420 .unwrap();
421 registry
422 .register(Box::new(uploader_imm_size.clone()))
423 .unwrap();
424
425 let opts = histogram_opts!(
426 "state_store_uploader_upload_task_latency",
427 "Latency of uploader uploading tasks",
428 time_buckets
429 );
430
431 let uploader_upload_task_latency =
432 register_histogram_with_registry!(opts, registry).unwrap();
433
434 let opts = histogram_opts!(
435 "state_store_uploader_wait_poll_latency",
436 "Latency of upload uploading task being polled after finish",
437 exponential_buckets(0.001, 5.0, 7).unwrap(), );
439
440 let uploader_wait_poll_latency = register_histogram_with_registry!(opts, registry).unwrap();
441
442 let uploader_syncing_epoch_count = register_int_gauge_with_registry!(
443 "state_store_uploader_syncing_epoch_count",
444 "Total number of syncing epoch",
445 registry
446 )
447 .unwrap();
448
449 let uploader_per_table_imm_size = register_guarded_int_gauge_vec_with_registry!(
450 "state_store_uploader_per_table_imm_size",
451 "Total uploader-tracked imm size per table",
452 &["table_id"],
453 registry
454 )
455 .unwrap();
456
457 let uploader_per_table_imm_size = RelabeledGuardedIntGaugeVec::with_metric_level(
458 MetricLevel::Debug,
459 uploader_per_table_imm_size,
460 metric_level,
461 );
462
463 let uploader_per_table_imm_count = register_guarded_int_gauge_vec_with_registry!(
464 "state_store_uploader_per_table_imm_count",
465 "Total uploader-tracked imm count per table",
466 &["table_id"],
467 registry
468 )
469 .unwrap();
470
471 let uploader_per_table_imm_count = RelabeledGuardedIntGaugeVec::with_metric_level(
472 MetricLevel::Debug,
473 uploader_per_table_imm_count,
474 metric_level,
475 );
476
477 let per_table_imm_size = register_guarded_int_gauge_vec_with_registry!(
478 "state_store_per_table_imm_size",
479 "Total imm size per table",
480 &["table_id", "fragment_id"],
481 registry
482 )
483 .unwrap();
484
485 let per_table_imm_size = RelabeledGuardedIntGaugeVec::with_metric_level_relabel_n(
486 MetricLevel::Debug,
487 per_table_imm_size,
488 metric_level,
489 1,
490 );
491
492 let per_table_imm_count = register_guarded_int_gauge_vec_with_registry!(
493 "state_store_per_table_imm_count",
494 "Total imm count per table",
495 &["table_id"],
496 registry
497 )
498 .unwrap();
499
500 let per_table_imm_count = RelabeledGuardedIntGaugeVec::with_metric_level(
501 MetricLevel::Debug,
502 per_table_imm_count,
503 metric_level,
504 );
505
506 let read_req_bloom_filter_positive_counts =
507 register_guarded_int_counter_vec_with_registry!(
508 "state_store_read_req_bloom_filter_positive_counts",
509 "Total number of read request with at least one SST filter check returns positive",
510 &["table_id", "type"],
511 registry
512 )
513 .unwrap();
514 let read_req_bloom_filter_positive_counts =
515 RelabeledGuardedIntCounterVec::with_metric_level_relabel_n(
516 MetricLevel::Info,
517 read_req_bloom_filter_positive_counts,
518 metric_level,
519 1,
520 );
521
522 let read_req_positive_but_non_exist_counts = register_guarded_int_counter_vec_with_registry!(
523 "state_store_read_req_positive_but_non_exist_counts",
524 "Total number of read request on non-existent key/prefix with at least one SST filter check returns positive",
525 &["table_id", "type"],
526 registry
527 )
528 .unwrap();
529 let read_req_positive_but_non_exist_counts =
530 RelabeledGuardedIntCounterVec::with_metric_level(
531 MetricLevel::Info,
532 read_req_positive_but_non_exist_counts,
533 metric_level,
534 );
535
536 let read_req_check_bloom_filter_counts = register_guarded_int_counter_vec_with_registry!(
537 "state_store_read_req_check_bloom_filter_counts",
538 "Total number of read request that checks an SST filter with a prefix hint",
539 &["table_id", "type"],
540 registry
541 )
542 .unwrap();
543
544 let read_req_check_bloom_filter_counts = RelabeledGuardedIntCounterVec::with_metric_level(
545 MetricLevel::Info,
546 read_req_check_bloom_filter_counts,
547 metric_level,
548 );
549
550 let mem_table_spill_counts = register_guarded_int_counter_vec_with_registry!(
551 "state_store_mem_table_spill_counts",
552 "Total number of mem table spill occurs for one table",
553 &["table_id"],
554 registry
555 )
556 .unwrap();
557
558 let mem_table_spill_counts = RelabeledGuardedIntCounterVec::with_metric_level(
559 MetricLevel::Info,
560 mem_table_spill_counts,
561 metric_level,
562 );
563
564 let old_value_size = register_guarded_int_gauge_vec_with_registry!(
565 "state_store_old_value_size",
566 "The size of old value",
567 &["table_id"],
568 registry
569 )
570 .unwrap();
571
572 let old_value_size = RelabeledGuardedIntGaugeVec::with_metric_level(
573 MetricLevel::Info,
574 old_value_size,
575 metric_level,
576 );
577
578 let opts = histogram_opts!(
579 "block_efficiency_histogram",
580 "Access ratio of in-memory block.",
581 exponential_buckets(0.001, 2.0, 11).unwrap(),
582 );
583 let block_efficiency_histogram = register_histogram_with_registry!(opts, registry).unwrap();
584
585 let event_handler_pending_event = register_int_gauge_vec_with_registry!(
586 "state_store_event_handler_pending_event",
587 "The number of sent but unhandled events",
588 &["event_type"],
589 registry,
590 )
591 .unwrap();
592
593 let opts = histogram_opts!(
594 "state_store_event_handler_latency",
595 "Latency to handle event",
596 exponential_buckets(0.001, 5.0, 7).unwrap(), );
598
599 let event_handler_latency =
600 register_histogram_vec_with_registry!(opts, &["event_type"], registry).unwrap();
601
602 let safe_version_hit = GenericCounter::new(
603 "state_store_safe_version_hit",
604 "The total count of a safe version that can be retrieved successfully",
605 )
606 .unwrap();
607 registry
608 .register(Box::new(safe_version_hit.clone()))
609 .unwrap();
610
611 let safe_version_miss = GenericCounter::new(
612 "state_store_safe_version_miss",
613 "The total count of a safe version that cannot be retrieved",
614 )
615 .unwrap();
616 registry
617 .register(Box::new(safe_version_miss.clone()))
618 .unwrap();
619
620 Self {
621 bloom_filter_true_negative_counts,
622 bloom_filter_check_counts,
623 iter_merge_sstable_counts,
624 vnode_pruning_counts,
625 sst_store_block_request_counts,
626 iter_scan_key_counts,
627 get_shared_buffer_hit_counts,
628 remote_read_time,
629 iter_fetch_meta_duration,
630 iter_fetch_meta_cache_unhits,
631 iter_slow_fetch_meta_cache_unhits,
632 vector_object_request_counts,
633 vector_request_stats,
634 vector_hnsw_graph_level_node_count,
635 vector_index_file_count,
636 vector_index_file_size,
637 read_req_bloom_filter_positive_counts,
638 read_req_positive_but_non_exist_counts,
639 read_req_check_bloom_filter_counts,
640 write_batch_tuple_counts,
641 write_batch_duration,
642 write_batch_size,
643 spill_task_counts_from_unsealed: spill_task_counts.with_label_values(&["unsealed"]),
644 spill_task_size_from_unsealed: spill_task_size.with_label_values(&["unsealed"]),
645 uploader_uploading_task_size,
646 uploader_uploading_task_count,
647 uploader_imm_size,
648 uploader_upload_task_latency,
649 uploader_syncing_epoch_count,
650 uploader_wait_poll_latency,
651 uploader_per_table_imm_size,
652 uploader_per_table_imm_count,
653 per_table_imm_size,
654 per_table_imm_count,
655 mem_table_spill_counts,
656 old_value_size,
657
658 block_efficiency_histogram,
659 event_handler_pending_event,
660 event_handler_latency,
661 safe_version_hit,
662 safe_version_miss,
663 table_change_log_fetch_latency,
664 table_change_log_cache_hit,
665 table_change_log_cache_miss,
666 }
667 }
668
669 pub fn unused() -> Self {
670 global_hummock_state_store_metrics(MetricLevel::Disabled)
671 }
672}
673
674pub trait MemoryCollector: Sync + Send {
675 fn get_meta_memory_usage(&self) -> u64;
676 fn get_data_memory_usage(&self) -> u64;
677 fn get_vector_meta_memory_usage(&self) -> u64;
678 fn get_vector_data_memory_usage(&self) -> u64;
679 fn get_uploading_memory_usage(&self) -> u64;
680 fn get_prefetch_memory_usage(&self) -> usize;
681 fn get_meta_cache_memory_usage_ratio(&self) -> f64;
682 fn get_block_cache_memory_usage_ratio(&self) -> f64;
683 fn get_vector_meta_cache_memory_usage_ratio(&self) -> f64;
684 fn get_vector_data_cache_memory_usage_ratio(&self) -> f64;
685 fn get_shared_buffer_usage_ratio(&self) -> f64;
686}
687
688#[derive(Clone)]
689struct StateStoreCollector {
690 memory_collector: Arc<dyn MemoryCollector>,
691 collectors: Vec<Arc<dyn Collector>>,
692 block_cache_size: IntGauge,
693 meta_cache_size: IntGauge,
694 vector_data_cache_size: IntGauge,
695 vector_meta_cache_size: IntGauge,
696 uploading_memory_size: IntGauge,
697 prefetch_memory_size: IntGauge,
698 meta_cache_usage_ratio: Gauge,
699 block_cache_usage_ratio: Gauge,
700 vector_data_cache_usage_ratio: Gauge,
701 vector_meta_cache_usage_ratio: Gauge,
702 uploading_memory_usage_ratio: Gauge,
703}
704
705impl StateStoreCollector {
706 pub fn new(memory_collector: Arc<dyn MemoryCollector>) -> Self {
707 let mut collectors = Vec::new();
708
709 let block_cache_size = IntGauge::with_opts(Opts::new(
710 "state_store_block_cache_size",
711 "the size of cache for data block cache",
712 ))
713 .unwrap();
714 collectors.push(Arc::new(block_cache_size.clone()) as _);
715
716 let block_cache_usage_ratio = Gauge::with_opts(Opts::new(
717 "state_store_block_cache_usage_ratio",
718 "the ratio of block cache to it's pre-allocated memory",
719 ))
720 .unwrap();
721 collectors.push(Arc::new(block_cache_usage_ratio.clone()) as _);
722
723 let meta_cache_size = IntGauge::with_opts(Opts::new(
724 "state_store_meta_cache_size",
725 "the size of cache for meta file cache",
726 ))
727 .unwrap();
728 collectors.push(Arc::new(meta_cache_size.clone()) as _);
729
730 let meta_cache_usage_ratio = Gauge::with_opts(Opts::new(
731 "state_store_meta_cache_usage_ratio",
732 "the ratio of meta cache to it's pre-allocated memory",
733 ))
734 .unwrap();
735 collectors.push(Arc::new(meta_cache_usage_ratio.clone()) as _);
736
737 let vector_data_cache_size = IntGauge::with_opts(Opts::new(
738 "state_store_vector_data_cache_size",
739 "the size of cache for vector data file cache",
740 ))
741 .unwrap();
742 collectors.push(Arc::new(vector_data_cache_size.clone()) as _);
743
744 let vector_data_cache_usage_ratio = Gauge::with_opts(Opts::new(
745 "state_store_vector_data_cache_usage_ratio",
746 "the ratio of vector data cache to it's pre-allocated memory",
747 ))
748 .unwrap();
749 collectors.push(Arc::new(vector_data_cache_usage_ratio.clone()) as _);
750
751 let vector_meta_cache_size = IntGauge::with_opts(Opts::new(
752 "state_store_vector_meta_cache_size",
753 "the size of cache for vector meta file cache",
754 ))
755 .unwrap();
756 collectors.push(Arc::new(vector_meta_cache_size.clone()) as _);
757
758 let vector_meta_cache_usage_ratio = Gauge::with_opts(Opts::new(
759 "state_store_vector_meta_cache_usage_ratio",
760 "the ratio of vector meta cache to it's pre-allocated memory",
761 ))
762 .unwrap();
763 collectors.push(Arc::new(vector_meta_cache_usage_ratio.clone()) as _);
764
765 let uploading_memory_size = IntGauge::with_opts(Opts::new(
766 "uploading_memory_size",
767 "the size of uploading SSTs memory usage",
768 ))
769 .unwrap();
770 collectors.push(Arc::new(uploading_memory_size.clone()) as _);
771
772 let uploading_memory_usage_ratio = Gauge::with_opts(Opts::new(
773 "state_store_uploading_memory_usage_ratio",
774 "the ratio of uploading SSTs memory usage to it's pre-allocated memory",
775 ))
776 .unwrap();
777 collectors.push(Arc::new(uploading_memory_usage_ratio.clone()) as _);
778
779 let prefetch_memory_size = IntGauge::with_opts(Opts::new(
780 "state_store_prefetch_memory_size",
781 "the size of prefetch memory usage",
782 ))
783 .unwrap();
784 collectors.push(Arc::new(prefetch_memory_size.clone()) as _);
785
786 Self {
787 memory_collector,
788 collectors,
789 block_cache_size,
790 meta_cache_size,
791 vector_data_cache_size,
792 vector_meta_cache_size,
793 uploading_memory_size,
794 prefetch_memory_size,
795 meta_cache_usage_ratio,
796 block_cache_usage_ratio,
797
798 vector_data_cache_usage_ratio,
799 vector_meta_cache_usage_ratio,
800 uploading_memory_usage_ratio,
801 }
802 }
803}
804
805impl Collector for StateStoreCollector {
806 fn desc(&self) -> Vec<&Desc> {
807 self.collectors.iter().flat_map(|c| c.desc()).collect()
808 }
809
810 fn collect(&self) -> Vec<proto::MetricFamily> {
811 self.block_cache_size
812 .set(self.memory_collector.get_data_memory_usage() as i64);
813 self.meta_cache_size
814 .set(self.memory_collector.get_meta_memory_usage() as i64);
815 self.vector_data_cache_size
816 .set(self.memory_collector.get_vector_data_memory_usage() as _);
817 self.vector_meta_cache_size
818 .set(self.memory_collector.get_vector_meta_memory_usage() as _);
819 self.uploading_memory_size
820 .set(self.memory_collector.get_uploading_memory_usage() as i64);
821 self.prefetch_memory_size
822 .set(self.memory_collector.get_prefetch_memory_usage() as i64);
823 self.meta_cache_usage_ratio
824 .set(self.memory_collector.get_meta_cache_memory_usage_ratio());
825 self.block_cache_usage_ratio
826 .set(self.memory_collector.get_block_cache_memory_usage_ratio());
827 self.vector_meta_cache_usage_ratio.set(
828 self.memory_collector
829 .get_vector_meta_cache_memory_usage_ratio(),
830 );
831 self.vector_data_cache_usage_ratio.set(
832 self.memory_collector
833 .get_vector_data_cache_memory_usage_ratio(),
834 );
835 self.uploading_memory_usage_ratio
836 .set(self.memory_collector.get_shared_buffer_usage_ratio());
837 self.collectors.iter().flat_map(|c| c.collect()).collect()
839 }
840}
841
842pub fn monitor_cache(memory_collector: Arc<dyn MemoryCollector>) {
843 let collector = Box::new(StateStoreCollector::new(memory_collector));
844 if let Err(e) = GLOBAL_METRICS_REGISTRY.register(collector) {
845 warn!(
846 "unable to monitor cache. May have been registered if in all-in-one deployment: {}",
847 e.as_report()
848 );
849 }
850}