Files
RustProxy/src/cache/dashboard.rs
T
2026-10-17 22:33:00 +02:00

435 lines
18 KiB
Rust

impl Cache {
pub fn view(&self) -> View {
let settings = self
.tier_settings
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone();
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
self.counters
.expired
.fetch_add(store.sweep() as u64, Ordering::Relaxed);
let n = |value: &AtomicU64| value.load(Ordering::Relaxed);
let hits = n(&self.counters.hits);
let misses = n(&self.counters.misses);
let cold = self.cold.lock().unwrap_or_else(|e| e.into_inner());
let fill_bytes_active = n(&self.fill_bytes_active);
let cold_queue_bytes = n(&self.cold_queue_bytes);
let cold_write_bytes_active = n(&self.cold_write_bytes_active);
let cold_read_bytes_active = n(&self.cold_read_bytes_active);
let transient_bytes = n(self.transient_bytes_active.as_ref());
View {
hits,
misses,
bypasses: n(&self.counters.bypasses),
stored: n(&self.counters.stored),
evictions: n(&self.counters.evictions),
expired: n(&self.counters.expired),
rejected_fills: n(&self.counters.rejected_fills),
memory_pressure_rejections: n(&self.counters.memory_pressure_rejections),
hot_eviction_limited: n(&self.counters.hot_eviction_limited),
purges: n(&self.counters.purges),
entries: store.entries.len(),
resident_bytes: store.bytes,
hot_logical_bytes: store.logical_bytes,
hot_allocated_bytes: store.bytes,
hot_limit_bytes: settings.max_bytes,
transient_limit_bytes: settings.transient_max_bytes,
cold_queue_limit_bytes: settings.cold_queue_max_bytes,
cold_entries: cold.entries.len(),
cold_bytes: cold.bytes,
cold_limit_bytes: settings.cold_max_bytes,
cold_limit_entries: settings.cold_max_entries,
cold_hits: n(&self.counters.cold_hits),
cold_promotions: n(&self.counters.cold_promotions),
hot_demotions: n(&self.counters.hot_demotions),
cold_evictions: n(&self.counters.cold_evictions),
cold_errors: n(&self.counters.cold_errors),
cold_queue_dropped: n(&self.counters.cold_queue_dropped),
collapsed_fills: n(&self.counters.collapsed_fills),
collapse_timeouts: n(&self.counters.collapse_timeouts),
collapse_waiters: n(&self.collapse_waiters),
purge_entries: n(&self.counters.purge_entries),
bans_created: n(&self.counters.bans_created),
bans_removed: n(&self.counters.bans_removed),
ban_matches: n(&self.counters.ban_matches),
active_bans: self.active_bans().len(),
proactive_scheduled: n(&self.counters.proactive_scheduled),
proactive_started: n(&self.counters.proactive_started),
proactive_skipped: n(&self.counters.proactive_skipped),
proactive_inflight: n(&self.proactive_inflight),
fills_active: FILL_SLOTS - self.fills.available_permits(),
fill_bytes_active,
cold_queue_items: n(&self.cold_queue_items),
cold_queue_bytes,
cold_write_bytes_active,
cold_read_bytes_active,
transient_bytes,
hit_ratio: if hits + misses > 0 {
hits as f64 / (hits + misses) as f64
} else {
0.0
},
}
}
pub fn prometheus(&self) -> String {
let view = self.view();
let mut text = String::new();
for (name, kind, value) in [
("hits_total", "counter", view.hits),
("misses_total", "counter", view.misses),
("bypasses_total", "counter", view.bypasses),
("stored_total", "counter", view.stored),
("evictions_total", "counter", view.evictions),
("expired_total", "counter", view.expired),
("rejected_fills_total", "counter", view.rejected_fills),
(
"memory_pressure_rejections_total",
"counter",
view.memory_pressure_rejections,
),
(
"hot_eviction_limited_total",
"counter",
view.hot_eviction_limited,
),
("purges_total", "counter", view.purges),
("purge_entries_total", "counter", view.purge_entries),
("collapsed_fills_total", "counter", view.collapsed_fills),
("collapse_timeouts_total", "counter", view.collapse_timeouts),
("collapse_waiters", "gauge", view.collapse_waiters),
("bans_created_total", "counter", view.bans_created),
("bans_removed_total", "counter", view.bans_removed),
("ban_matches_total", "counter", view.ban_matches),
("active_bans", "gauge", view.active_bans as u64),
("proactive_scheduled_total", "counter", view.proactive_scheduled),
("proactive_started_total", "counter", view.proactive_started),
("proactive_skipped_total", "counter", view.proactive_skipped),
("proactive_inflight", "gauge", view.proactive_inflight),
("entries", "gauge", view.entries as u64),
("resident_bytes", "gauge", view.resident_bytes as u64),
("hot_logical_bytes", "gauge", view.hot_logical_bytes as u64),
("hot_allocated_bytes", "gauge", view.hot_allocated_bytes as u64),
("hot_limit_bytes", "gauge", view.hot_limit_bytes as u64),
(
"transient_limit_bytes",
"gauge",
view.transient_limit_bytes as u64,
),
(
"cold_queue_limit_bytes",
"gauge",
view.cold_queue_limit_bytes as u64,
),
("cold_entries", "gauge", view.cold_entries as u64),
("cold_bytes", "gauge", view.cold_bytes),
("cold_limit_bytes", "gauge", view.cold_limit_bytes),
("cold_limit_entries", "gauge", view.cold_limit_entries as u64),
("cold_hits_total", "counter", view.cold_hits),
("cold_promotions_total", "counter", view.cold_promotions),
("hot_demotions_total", "counter", view.hot_demotions),
("cold_evictions_total", "counter", view.cold_evictions),
("cold_errors_total", "counter", view.cold_errors),
(
"cold_queue_dropped_total",
"counter",
view.cold_queue_dropped,
),
("fills_active", "gauge", view.fills_active as u64),
("fill_bytes_active", "gauge", view.fill_bytes_active),
("cold_queue_items", "gauge", view.cold_queue_items),
("cold_queue_bytes", "gauge", view.cold_queue_bytes),
(
"cold_write_bytes_active",
"gauge",
view.cold_write_bytes_active,
),
(
"cold_read_bytes_active",
"gauge",
view.cold_read_bytes_active,
),
("transient_bytes", "gauge", view.transient_bytes),
] {
text.push_str(&format!(
"# TYPE rust_proxy_cache_{name} {kind}\nrust_proxy_cache_{name} {value}\n"
));
}
text.push_str(&format!(
"# TYPE rust_proxy_cache_hit_ratio gauge\nrust_proxy_cache_hit_ratio {:.6}\n",
view.hit_ratio
));
text
}
pub fn bypass(&self) {
self.counters.bypasses.fetch_add(1, Ordering::Relaxed);
}
/// Clear all keys and cancel publication of older in-flight fills.
pub fn purge(&self, revision: Option<u64>) -> usize {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
let mut count = store.entries.len();
store.entries.clear();
store.variants.clear();
store.lru.clear();
store.bytes = 0;
store.logical_bytes = 0;
store.generation = store.generation.wrapping_add(1);
store.scope_generations.clear();
store.key_generations.clear();
if let Some(revision) = revision {
store.revision = revision;
self.revision.store(revision, Ordering::Relaxed);
store.domains.clear();
}
drop(store);
self.cancel_all_fill_flights();
let settings = self
.tier_settings
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone();
{
let mut cold = self.cold.lock().unwrap_or_else(|e| e.into_inner());
count += cold.entries.len();
cold.entries.clear();
cold.variants.clear();
cold.lru.clear();
cold.bytes = 0;
cold.sequence = 0;
}
if let (Some(root), Some(tx)) = (self.cold_root_for(&settings), &self.cold_tx) {
let _ = tx.try_send(ColdCommand::PurgeDirectory(root));
}
self.counters.purges.fetch_add(1, Ordering::Relaxed);
self.counters.purge_entries.fetch_add(count as u64, Ordering::Relaxed);
count
}
/// Purge one vhost without evicting other hosts. Generation bumps also stop
/// an already streaming MISS from repopulating the just-purged scope.
pub fn purge_scope(&self, scope: &str) -> usize {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
let keys: Vec<_> = store
.entries
.iter()
.filter(|(_, e)| e.scope == scope)
.map(|(k, _)| *k)
.collect();
let mut count = keys.len();
for key in keys {
store.remove(&key);
}
let generation = store.scope_generations.entry(scope.to_owned()).or_default();
*generation = generation.wrapping_add(1);
drop(store);
self.cancel_scope_fill_flights(scope);
let mut delete = Vec::new();
{
let mut cold = self.cold.lock().unwrap_or_else(|e| e.into_inner());
let keys: Vec<_> = cold
.entries
.iter()
.filter(|(_, entry)| entry.scope == scope)
.map(|(key, _)| *key)
.collect();
count += keys.len();
for key in keys {
if let Some(entry) = cold.remove(&key) {
delete.extend([entry.meta_path, entry.body_path]);
}
}
}
if let (Some(tx), false) = (&self.cold_tx, delete.is_empty()) {
let _ = tx.try_send(ColdCommand::Delete(delete));
}
self.counters.purges.fetch_add(1, Ordering::Relaxed);
self.counters.purge_entries.fetch_add(count as u64, Ordering::Relaxed);
count
}
pub fn scope_view(&self, scope: &str) -> ScopeView {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
self.counters
.expired
.fetch_add(store.sweep() as u64, Ordering::Relaxed);
let mut view = ScopeView {
entries: 0,
resident_bytes: 0,
cold_entries: 0,
cold_bytes: 0,
};
for entry in store.entries.values().filter(|entry| entry.scope == scope) {
view.entries += 1;
view.resident_bytes += entry.cost;
}
drop(store);
let cold = self.cold.lock().unwrap_or_else(|e| e.into_inner());
for entry in cold.entries.values().filter(|entry| entry.scope == scope) {
view.cold_entries += 1;
view.cold_bytes = view.cold_bytes.saturating_add(entry.disk_bytes);
}
view
}
pub fn record(&self, scope: &str, domain: &str, outcome: &str, reason: &'static str) {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
if let Some(c) = store.domain(scope, domain) {
match outcome {
"HIT" => c.hits += 1,
"MISS" => c.misses += 1,
_ => c.bypasses += 1,
}
*c.reasons.entry(reason.into()).or_default() += 1;
}
}
pub fn record_reason(&self, scope: &str, domain: &str, reason: &'static str) {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
if let Some(c) = store.domain(scope, domain) {
*c.reasons.entry(reason.into()).or_default() += 1;
}
}
pub fn dashboard(&self, scope: Option<&str>, offset: usize, limit: usize) -> Dashboard {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
self.counters
.expired
.fetch_add(store.sweep() as u64, Ordering::Relaxed);
let revision = store.revision;
let mut groups: BTreeMap<(String, String), DomainView> = store
.domains
.iter()
.filter(|((vhost, _), _)| scope.is_none_or(|s| s == vhost))
.map(|((vhost, domain), counters)| {
(
(vhost.clone(), domain.clone()),
DomainView {
vhost: vhost.clone(),
domain: domain.clone(),
entries: 0,
logical_bytes: 0,
resident_bytes: 0,
hot_entries: 0,
cold_entries: 0,
cold_bytes: 0,
hit_ratio: if counters.hits + counters.misses > 0 {
counters.hits as f64 / (counters.hits + counters.misses) as f64
} else {
0.0
},
counters: counters.clone(),
},
)
})
.collect();
let mut objects = Vec::new();
for key in store.lru.values().rev() {
let Some(e) = store.entries.get(key) else {
continue;
};
let age = e.initial_age.saturating_add(e.received.elapsed()).as_secs();
// This is an inventory of retained objects, not a freshness lookup.
// Expired representations still occupy RAM and may be revalidated.
if !scope.is_none_or(|s| s == e.scope) {
continue;
}
let row = groups
.entry((e.scope.clone(), e.domain.clone()))
.or_insert_with(|| DomainView {
vhost: e.scope.clone(),
domain: e.domain.clone(),
entries: 0,
logical_bytes: 0,
resident_bytes: 0,
hot_entries: 0,
cold_entries: 0,
cold_bytes: 0,
hit_ratio: 0.0,
counters: DomainCounters::default(),
});
row.entries += 1;
row.hot_entries += 1;
row.logical_bytes += e.body.len();
row.resident_bytes += e.cost;
objects.push(ObjectView {
vhost: e.scope.clone(),
scheme: e.scheme.clone(),
authority: e.authority.clone(),
domain: e.domain.clone(),
path: e.path.clone(),
target: e.target.clone(),
cache_key: key_hex(key),
tags: e.tags.clone(),
transformed: e.transformed,
body_bytes: e.body.len(),
resident_bytes: e.cost,
storage_bytes: 0,
tier: "hot",
hits: e.hits,
age_seconds: age,
resident_age_seconds: hot_resident_age(e).as_secs(),
remaining_ttl_seconds: e.ttl.saturating_sub(age),
});
}
drop(store);
let cold = self.cold.lock().unwrap_or_else(|e| e.into_inner());
for key in cold.lru.values().rev() {
let Some(e) = cold.entries.get(key) else {
continue;
};
if e.revision != revision || !scope.is_none_or(|s| s == e.scope) {
continue;
}
let age = cold_age(e);
let row = groups
.entry((e.scope.clone(), e.domain.clone()))
.or_insert_with(|| DomainView {
vhost: e.scope.clone(),
domain: e.domain.clone(),
entries: 0,
logical_bytes: 0,
resident_bytes: 0,
hot_entries: 0,
cold_entries: 0,
cold_bytes: 0,
hit_ratio: 0.0,
counters: DomainCounters::default(),
});
row.entries += 1;
row.cold_entries += 1;
row.cold_bytes = row.cold_bytes.saturating_add(e.disk_bytes);
objects.push(ObjectView {
vhost: e.scope.clone(),
scheme: e.scheme.clone(),
authority: e.authority.clone(),
domain: e.domain.clone(),
path: e.path.clone(),
target: e.target.clone(),
cache_key: key_hex(key),
tags: e.tags.clone(),
transformed: e.transformed,
body_bytes: e.body_bytes,
resident_bytes: 0,
storage_bytes: e.disk_bytes,
tier: "cold",
hits: e.hits,
age_seconds: age,
resident_age_seconds: cold_resident_age(e),
remaining_ttl_seconds: e.ttl.saturating_sub(age),
});
}
let total = objects.len();
let matching_body_bytes = objects.iter().map(|entry| entry.body_bytes).sum();
let matching_resident_bytes = objects.iter().map(|entry| entry.resident_bytes).sum();
let matching_cold_bytes = objects.iter().map(|entry| entry.storage_bytes).sum();
let limit = limit.clamp(1, 200);
let objects = objects.into_iter().skip(offset).take(limit).collect();
Dashboard {
domains: groups.into_values().collect(),
objects,
matching_objects: total,
matching_body_bytes,
matching_resident_bytes,
matching_cold_bytes,
offset,
limit,
statistics_revision: revision,
}
}
}