+1

ufos/src/main.rs

Reviewed

··· 162 162 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); 163 163 loop { 164 164 interval.tick().await; 165 165 + read_store.update_metrics(); 165 166 match read_store.get_consumer_info().await { 166 167 Err(e) => log::warn!("failed to get jetstream consumer info: {e:?}"), 167 168 Ok(ConsumerInfo::Jetstream {

+2

ufos/src/storage.rs

Reviewed

··· 104 104 pub trait StoreReader: Send + Sync { 105 105 fn name(&self) -> String; 106 106 107 107 + fn update_metrics(&self) {} 108 108 + 107 109 async fn get_storage_stats(&self) -> StorageResult<serde_json::Value>; 108 110 109 111 async fn get_consumer_info(&self) -> StorageResult<ConsumerInfo>;

+27 -1

ufos/src/storage_fjall.rs

Reviewed

··· 23 23 Batch as FjallBatch, Config, Keyspace, PartitionCreateOptions, PartitionHandle, Snapshot, 24 24 }; 25 25 use jetstream::events::Cursor; 26 26 - use metrics::{counter, describe_counter, describe_histogram, histogram, Unit}; 26 26 + use lsm_tree::AbstractTree; 27 27 + use metrics::{ 28 28 + counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram, Unit, 29 29 + }; 27 30 use std::collections::{HashMap, HashSet}; 28 31 use std::iter::Peekable; 29 32 use std::ops::Bound; ··· 227 230 feeds: feeds.clone(), 228 231 records: records.clone(), 229 232 rollups: rollups.clone(), 233 233 + queues: queues.clone(), 230 234 }; 235 235 + reader.describe_metrics(); 231 236 let writer = FjallWriter { 232 237 bg_taken: Arc::new(AtomicBool::new(false)), 233 238 keyspace, ··· 250 255 feeds: PartitionHandle, 251 256 records: PartitionHandle, 252 257 rollups: PartitionHandle, 258 258 + queues: PartitionHandle, 253 259 } 254 260 255 261 /// An iterator that knows how to skip over deleted/invalidated records ··· 381 387 type CollectionSerieses = HashMap<Nsid, Vec<CountsValue>>; 382 388 383 389 impl FjallReader { 390 390 + fn describe_metrics(&self) { 391 391 + describe_gauge!( 392 392 + "storage_fjall_l0_run_count", 393 393 + Unit::Count, 394 394 + "number of L0 runs in a partition" 395 395 + ); 396 396 + } 397 397 + 384 398 fn get_storage_stats(&self) -> StorageResult<serde_json::Value> { 385 399 let rollup_cursor = 386 400 get_static_neu::<NewRollupCursorKey, NewRollupCursorValue>(&self.global)? ··· 999 1013 impl StoreReader for FjallReader { 1000 1014 fn name(&self) -> String { 1001 1015 "fjall storage v2".into() 1016 1016 + } 1017 1017 + fn update_metrics(&self) { 1018 1018 + gauge!("storage_fjall_l0_run_count", "partition" => "global") 1019 1019 + .set(self.global.tree.l0_run_count() as f64); 1020 1020 + gauge!("storage_fjall_l0_run_count", "partition" => "feeds") 1021 1021 + .set(self.feeds.tree.l0_run_count() as f64); 1022 1022 + gauge!("storage_fjall_l0_run_count", "partition" => "records") 1023 1023 + .set(self.records.tree.l0_run_count() as f64); 1024 1024 + gauge!("storage_fjall_l0_run_count", "partition" => "rollups") 1025 1025 + .set(self.rollups.tree.l0_run_count() as f64); 1026 1026 + gauge!("storage_fjall_l0_run_count", "partition" => "queues") 1027 1027 + .set(self.queues.tree.l0_run_count() as f64); 1002 1028 } 1003 1029 async fn get_storage_stats(&self) -> StorageResult<serde_json::Value> { 1004 1030 let s = self.clone();