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