forked from
microcosm.blue/microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
use db stats in logs
...eventually put these properly in metrics
+10
-29
+7
-2
link_aggregator/src/main.rs
+7
-2
link_aggregator/src/main.rs
···
19
19
use server::serve;
20
20
#[cfg(feature = "rocks")]
21
21
use storage::RocksStorage;
22
-
use storage::{LinkReader, LinkStorage, MemStorage};
22
+
use storage::{LinkReader, LinkStorage, MemStorage, StorageStats};
23
23
24
24
const MONITOR_INTERVAL: time::Duration = time::Duration::from_secs(2);
25
25
···
173
173
}
174
174
175
175
'monitor: loop {
176
-
readable.summarize(qsize.load(Ordering::Relaxed));
176
+
let queue_size = qsize.load(Ordering::Relaxed);
177
+
match readable.get_stats() {
178
+
Ok(StorageStats { dids, targetables, linking_records }) => println!("queue: {queue_size},\tdids: {dids}, targetables: {targetables}, linking_records: {linking_records}"),
179
+
Err(e) => eprintln!("queue: {queue_size}, failed to get stats: {e:?}"),
180
+
}
181
+
177
182
process_collector.collect();
178
183
if let Some(ref p) = data_dir {
179
184
if let Ok(avail) = fs4::available_space(p) {
-12
link_aggregator/src/storage/mem_store.rs
-12
link_aggregator/src/storage/mem_store.rs
···
219
219
linking_records,
220
220
})
221
221
}
222
-
223
-
fn summarize(&self, qsize: u32) {
224
-
let data = self.0.lock().unwrap();
225
-
let dids = data.dids.len();
226
-
let targets = data.targets.len();
227
-
let target_paths: usize = data.targets.values().map(|paths| paths.len()).sum();
228
-
let links = data.links.len();
229
-
230
-
let sample_target = data.targets.keys().nth(data.targets.len() / 2);
231
-
let sample_path = sample_target.and_then(|t| data.targets.get(t).unwrap().keys().next());
232
-
println!("queue: {qsize}. {dids} dids, {targets} targets from {target_paths} paths, {links} links. sample: {sample_target:?} {sample_path:?}");
233
-
}
234
222
}
235
223
236
224
#[derive(Debug, PartialEq, Hash, Eq, Clone)]
+3
-8
link_aggregator/src/storage/mod.rs
+3
-8
link_aggregator/src/storage/mod.rs
···
21
21
pub struct StorageStats {
22
22
/// estimate of how many accounts we've seen create links. the _subjects_ of any links are not represented here.
23
23
/// for example: new user A follows users B and C. this count will only increment by one, for A.
24
-
dids: u64,
24
+
pub dids: u64,
25
25
26
26
/// estimate targets * distinct (collection, path)s to reference them.
27
27
/// distinct targets alone are currently challenging to estimate.
28
-
targetables: u64,
28
+
pub targetables: u64,
29
29
30
30
/// estimate of the count of atproto records seen that contain links.
31
31
/// records with multiple links are single-counted.
32
32
/// for LSM stores, deleted links don't decrement this, and updated records with any links will likely increment it.
33
-
linking_records: u64,
33
+
pub linking_records: u64,
34
34
}
35
35
36
36
pub trait LinkStorage: Send + Sync {
···
61
61
62
62
/// assume all stats are estimates, since exact counts are very challenging for LSMs
63
63
fn get_stats(&self) -> Result<StorageStats>;
64
-
65
-
// todo: remove it
66
-
fn summarize(&self, qsize: u32) {
67
-
println!("queue: {qsize}");
68
-
}
69
64
}
70
65
71
66
#[cfg(test)]
-7
link_aggregator/src/storage/rocks_store.rs
-7
link_aggregator/src/storage/rocks_store.rs
···
801
801
.map(|s| s.parse::<u64>())
802
802
.transpose()?
803
803
.unwrap_or(0);
804
-
eprintln!("lr {linking_records:?}");
805
804
Ok(StorageStats {
806
805
dids,
807
806
targetables,
808
807
linking_records,
809
808
})
810
-
}
811
-
812
-
fn summarize(&self, qsize: u32) {
813
-
let did_seq = self.did_id_table.base.id_seq.load(Ordering::SeqCst);
814
-
let target_seq = self.target_id_table.base.id_seq.load(Ordering::SeqCst);
815
-
println!("queue: {qsize}. did seq: {did_seq}, target seq: {target_seq}.");
816
809
}
817
810
}
818
811