use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::SystemTime;
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use tracing::{debug, info};
use chronix_core::{NamespaceId, NamespaceQuota, NamespaceUsage};
use crate::tenant::error::{Result, TenantError};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NamespaceInfo {
pub id: NamespaceId,
pub description: String,
pub owner: String,
pub quota: NamespaceQuota,
pub created_at_ms: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NamespaceState {
pub info: NamespaceInfo,
pub usage: NamespaceUsage,
}
#[derive(Debug)]
pub struct NamespaceRegistry {
namespaces: DashMap<String, NamespaceState>,
persist_dir: Option<PathBuf>,
}
const SNAPSHOT_FILE: &str = "namespaces.json";
fn write_snapshot(dir: &Path, states: &[NamespaceState]) -> Result<()> {
let json = serde_json::to_string_pretty(states)
.map_err(|e| TenantError::InvalidConfig(format!("cannot serialize namespaces: {e}")))?;
let tmp_path = dir.join("namespaces.json.tmp");
let final_path = dir.join(SNAPSHOT_FILE);
std::fs::write(&tmp_path, &json)
.map_err(|e| TenantError::Persist(format!("cannot write namespace snapshot: {e}")))?;
let result = (|| -> Result<()> {
let f = std::fs::File::open(&tmp_path).map_err(|e| {
TenantError::Persist(format!("cannot open tmp snapshot for fsync: {e}"))
})?;
f.sync_all()
.map_err(|e| TenantError::Persist(format!("cannot fsync namespace snapshot: {e}")))?;
std::fs::rename(&tmp_path, &final_path)
.map_err(|e| TenantError::Persist(format!("cannot rename namespace snapshot: {e}")))?;
Ok(())
})();
if result.is_err() {
let _ = std::fs::remove_file(&tmp_path);
}
result?;
debug!("namespace snapshot saved");
Ok(())
}
impl NamespaceRegistry {
#[must_use]
pub fn new() -> Self {
let registry = Self {
namespaces: DashMap::new(),
persist_dir: None,
};
registry.ensure_default();
registry
}
pub fn open(dir: impl Into<PathBuf>) -> Result<Self> {
let dir = dir.into();
std::fs::create_dir_all(&dir)
.map_err(|e| TenantError::InvalidConfig(format!("cannot create namespace dir: {e}")))?;
let snapshot_path = dir.join(SNAPSHOT_FILE);
let registry = if snapshot_path.exists() {
let data = std::fs::read_to_string(&snapshot_path).map_err(|e| {
TenantError::InvalidConfig(format!("cannot read namespace snapshot: {e}"))
})?;
let states: Vec<NamespaceState> = serde_json::from_str(&data).map_err(|e| {
TenantError::InvalidConfig(format!("corrupt namespace snapshot: {e}"))
})?;
let namespaces = DashMap::new();
for state in states {
namespaces.insert(state.info.id.as_str().to_string(), state);
}
info!(count = namespaces.len(), "loaded namespace snapshot");
Self {
namespaces,
persist_dir: Some(dir),
}
} else {
Self {
namespaces: DashMap::new(),
persist_dir: Some(dir),
}
};
registry.ensure_default();
if !snapshot_path.exists() {
registry.save_snapshot_inner()?;
}
Ok(registry)
}
fn ensure_default(&self) {
if !self.namespaces.contains_key(NamespaceId::DEFAULT) {
let default_info = NamespaceInfo {
id: NamespaceId::default_namespace(),
description: "Default namespace".to_string(),
owner: "system".to_string(),
quota: NamespaceQuota::default(),
created_at_ms: now_ms(),
};
self.namespaces.insert(
NamespaceId::DEFAULT.to_string(),
NamespaceState {
info: default_info,
usage: NamespaceUsage::default(),
},
);
}
}
fn save_snapshot_inner(&self) -> Result<()> {
let Some(dir) = &self.persist_dir else {
return Ok(());
};
let states: Vec<NamespaceState> =
self.namespaces.iter().map(|r| r.value().clone()).collect();
write_snapshot(dir, &states)
}
pub fn create_namespace(
&self,
id: NamespaceId,
description: impl Into<String>,
owner: impl Into<String>,
quota: NamespaceQuota,
) -> Result<NamespaceInfo> {
let name = id.as_str().to_string();
use dashmap::mapref::entry::Entry;
match self.namespaces.entry(name.clone()) {
Entry::Occupied(_) => Err(TenantError::NamespaceAlreadyExists(name)),
Entry::Vacant(vacant) => {
let info = NamespaceInfo {
id,
description: description.into(),
owner: owner.into(),
quota,
created_at_ms: now_ms(),
};
let state = NamespaceState {
info: info.clone(),
usage: NamespaceUsage::default(),
};
vacant.insert(state);
info!(namespace = %name, "namespace created");
#[allow(clippy::cast_precision_loss)]
metrics::gauge!("chronix_namespace_count").set(self.namespaces.len() as f64);
if let Err(e) = self.save_snapshot_inner() {
self.namespaces.remove(&name);
#[allow(clippy::cast_precision_loss)]
metrics::gauge!("chronix_namespace_count").set(self.namespaces.len() as f64);
return Err(e);
}
Ok(info)
}
}
}
pub fn delete_namespace(&self, name: &str) -> Result<NamespaceInfo> {
if name == NamespaceId::DEFAULT {
return Err(TenantError::InvalidConfig(
"cannot delete the default namespace".to_string(),
));
}
let (_, state) = self
.namespaces
.remove(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
info!(namespace = %name, "namespace deleted");
#[allow(clippy::cast_precision_loss)]
metrics::gauge!("chronix_namespace_count").set(self.namespaces.len() as f64);
self.save_snapshot_inner()?;
Ok(state.info)
}
pub fn get(&self, name: &str) -> Result<NamespaceState> {
self.namespaces
.get(name)
.map(|r| r.value().clone())
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))
}
#[must_use]
pub fn exists(&self, name: &str) -> bool {
self.namespaces.contains_key(name)
}
#[must_use]
pub fn list_names(&self) -> Vec<String> {
self.namespaces.iter().map(|r| r.key().clone()).collect()
}
#[must_use]
pub fn list_all(&self) -> Vec<NamespaceState> {
self.namespaces.iter().map(|r| r.value().clone()).collect()
}
#[must_use]
pub fn count(&self) -> usize {
self.namespaces.len()
}
pub fn update_quota(&self, name: &str, quota: NamespaceQuota) -> Result<()> {
quota.validate().map_err(TenantError::InvalidConfig)?;
{
let mut entry = self
.namespaces
.get_mut(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
entry.value_mut().info.quota = quota;
}
debug!(namespace = %name, "quota updated");
self.save_snapshot_inner()?;
Ok(())
}
#[must_use]
pub fn get_usage(&self, name: &str) -> Option<NamespaceUsage> {
self.namespaces.get(name).map(|r| r.value().usage.clone())
}
pub fn update_usage(&self, name: &str, usage: NamespaceUsage) -> Result<()> {
let mut entry = self
.namespaces
.get_mut(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
entry.value_mut().usage = usage;
Ok(())
}
pub fn increment_series(&self, name: &str, count: u64) -> Result<()> {
let mut entry = self
.namespaces
.get_mut(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
entry.value_mut().usage.series_count =
entry.value().usage.series_count.saturating_add(count);
Ok(())
}
pub fn increment_storage(&self, name: &str, bytes: u64) -> Result<()> {
let mut entry = self
.namespaces
.get_mut(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
entry.value_mut().usage.storage_bytes =
entry.value().usage.storage_bytes.saturating_add(bytes);
Ok(())
}
pub fn increment_measurements(&self, name: &str, count: u32) -> Result<()> {
let mut entry = self
.namespaces
.get_mut(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
entry.value_mut().usage.measurements =
entry.value().usage.measurements.saturating_add(count);
Ok(())
}
pub fn check_and_increment_write(
&self,
name: &str,
new_series: u64,
new_points: u64,
) -> Result<()> {
let mut entry = self
.namespaces
.get_mut(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
let max_series = entry.value().info.quota.max_series_count;
let max_rate = entry.value().info.quota.max_ingestion_rate;
let max_storage = entry.value().info.quota.max_storage_bytes;
let current_series = entry.value().usage.series_count;
let current_storage = entry.value().usage.storage_bytes;
let projected_series = current_series.saturating_add(new_series);
if projected_series > max_series {
return Err(TenantError::QuotaExceeded {
namespace: name.to_string(),
resource: "series_count".to_string(),
current: current_series,
limit: max_series,
});
}
let now_s = now_ms() / 1000;
let last_window = entry.value().usage.ingestion_rate_window_s;
let current_rate = if last_window == now_s {
entry.value().usage.ingestion_rate
} else {
0.0
};
#[allow(clippy::cast_precision_loss)]
let projected_rate = current_rate + new_points as f64;
#[allow(clippy::cast_precision_loss)]
let rate_limit = max_rate as f64;
if projected_rate > rate_limit {
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
let current = current_rate as u64;
return Err(TenantError::QuotaExceeded {
namespace: name.to_string(),
resource: "ingestion_rate".to_string(),
current,
limit: max_rate,
});
}
if current_storage > max_storage {
return Err(TenantError::QuotaExceeded {
namespace: name.to_string(),
resource: "storage_bytes".to_string(),
current: current_storage,
limit: max_storage,
});
}
let state_mut = entry.value_mut();
state_mut.usage.series_count = projected_series;
state_mut.usage.ingestion_rate = projected_rate;
state_mut.usage.ingestion_rate_window_s = now_s;
Ok(())
}
pub fn check_and_increment_measurement(&self, name: &str) -> Result<()> {
let mut entry = self
.namespaces
.get_mut(name)
.ok_or_else(|| TenantError::NamespaceNotFound(name.to_string()))?;
let state = entry.value();
let projected = u64::from(state.usage.measurements) + 1;
let limit = u64::from(state.info.quota.max_measurements);
if projected > limit {
return Err(TenantError::QuotaExceeded {
namespace: name.to_string(),
resource: "measurements".to_string(),
current: u64::from(state.usage.measurements),
limit,
});
}
entry.value_mut().usage.measurements = state.usage.measurements.saturating_add(1);
Ok(())
}
#[must_use]
pub fn usage_ratios(&self) -> Vec<UsageRatio> {
let mut ratios = Vec::new();
for entry in &self.namespaces {
let ns = entry.key();
let state = entry.value();
let q = &state.info.quota;
let u = &state.usage;
ratios.push(UsageRatio {
namespace: ns.clone(),
resource: "series_count".to_string(),
current: u.series_count,
limit: q.max_series_count,
ratio: ratio(u.series_count, q.max_series_count),
});
ratios.push(UsageRatio {
namespace: ns.clone(),
resource: "storage_bytes".to_string(),
current: u.storage_bytes,
limit: q.max_storage_bytes,
ratio: ratio(u.storage_bytes, q.max_storage_bytes),
});
ratios.push(UsageRatio {
namespace: ns.clone(),
resource: "measurements".to_string(),
current: u64::from(u.measurements),
limit: u64::from(q.max_measurements),
ratio: ratio(u64::from(u.measurements), u64::from(q.max_measurements)),
});
}
ratios
}
pub fn emit_usage_metrics(&self) {
for r in self.usage_ratios() {
metrics::gauge!(
"chronix_namespace_usage_ratio",
"namespace" => r.namespace,
"resource" => r.resource,
)
.set(r.ratio);
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UsageRatio {
pub namespace: String,
pub resource: String,
pub current: u64,
pub limit: u64,
pub ratio: f64,
}
#[allow(clippy::cast_precision_loss)]
fn ratio(current: u64, limit: u64) -> f64 {
if limit == 0 {
return if current > 0 { f64::MAX } else { 0.0 };
}
current as f64 / limit as f64
}
impl Default for NamespaceRegistry {
fn default() -> Self {
Self::new()
}
}
fn now_ms() -> u64 {
#[allow(clippy::cast_possible_truncation)]
SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
#[must_use]
pub fn shared_registry() -> Arc<NamespaceRegistry> {
Arc::new(NamespaceRegistry::new())
}
#[cfg(test)]
mod tests {
#[test]
fn creating_a_namespace_is_durable_or_it_fails() {
let dir = tempfile::tempdir().unwrap();
let registry = NamespaceRegistry::open(dir.path()).unwrap();
std::fs::create_dir_all(dir.path().join("namespaces.json.tmp")).unwrap();
let err = registry.create_namespace(
NamespaceId::new("tenant-a").unwrap(),
"a tenant",
"ops",
NamespaceQuota::default(),
);
assert!(
err.is_err(),
"a namespace that cannot be persisted must not be acknowledged"
);
}
#[test]
fn changing_a_quota_is_durable_or_it_fails() {
let dir = tempfile::tempdir().unwrap();
let registry = NamespaceRegistry::open(dir.path()).unwrap();
registry
.create_namespace(
NamespaceId::new("tenant-a").unwrap(),
"a tenant",
"ops",
NamespaceQuota::default(),
)
.unwrap();
std::fs::create_dir_all(dir.path().join("namespaces.json.tmp")).unwrap();
let quota = NamespaceQuota {
max_series_count: 42,
..NamespaceQuota::default()
};
assert!(
registry.update_quota("tenant-a", quota).is_err(),
"a quota that cannot be persisted reverts at the next restart, so \
reporting it as applied is a setting that is accepted and ignored"
);
}
use super::*;
#[test]
fn default_namespace_exists_on_creation() {
let registry = NamespaceRegistry::new();
assert!(registry.exists(NamespaceId::DEFAULT));
assert_eq!(registry.count(), 1);
}
#[test]
fn create_and_get_namespace() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("team-alpha").unwrap();
let info = registry
.create_namespace(id, "Team Alpha", "admin", NamespaceQuota::default())
.unwrap();
assert_eq!(info.id.as_str(), "team-alpha");
assert_eq!(info.owner, "admin");
assert_eq!(registry.count(), 2);
let state = registry.get("team-alpha").unwrap();
assert_eq!(state.info.description, "Team Alpha");
}
#[test]
fn create_duplicate_fails() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("my-ns").unwrap();
registry
.create_namespace(id, "first", "admin", NamespaceQuota::default())
.unwrap();
let id2 = NamespaceId::new("my-ns").unwrap();
let err = registry
.create_namespace(id2, "second", "admin", NamespaceQuota::default())
.unwrap_err();
assert!(matches!(err, TenantError::NamespaceAlreadyExists(_)));
}
#[test]
fn delete_namespace() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("temp-ns").unwrap();
registry
.create_namespace(id, "temp", "admin", NamespaceQuota::default())
.unwrap();
assert_eq!(registry.count(), 2);
registry.delete_namespace("temp-ns").unwrap();
assert_eq!(registry.count(), 1);
assert!(!registry.exists("temp-ns"));
}
#[test]
fn cannot_delete_default_namespace() {
let registry = NamespaceRegistry::new();
let err = registry.delete_namespace("default").unwrap_err();
assert!(matches!(err, TenantError::InvalidConfig(_)));
}
#[test]
fn delete_nonexistent_fails() {
let registry = NamespaceRegistry::new();
let err = registry.delete_namespace("nope").unwrap_err();
assert!(matches!(err, TenantError::NamespaceNotFound(_)));
}
#[test]
fn list_namespaces() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("ns-a").unwrap();
registry
.create_namespace(id, "A", "admin", NamespaceQuota::default())
.unwrap();
let id = NamespaceId::new("ns-b").unwrap();
registry
.create_namespace(id, "B", "admin", NamespaceQuota::default())
.unwrap();
let mut names = registry.list_names();
names.sort();
assert_eq!(names, vec!["default", "ns-a", "ns-b"]);
}
#[test]
fn update_quota() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("quota-ns").unwrap();
registry
.create_namespace(id, "quota test", "admin", NamespaceQuota::default())
.unwrap();
let new_quota = NamespaceQuota {
max_series_count: 500,
max_ingestion_rate: 1000,
max_storage_bytes: 1024,
max_measurements: 10,
max_request_rps: 0,
max_request_burst: 0,
};
registry
.update_quota("quota-ns", new_quota.clone())
.unwrap();
let state = registry.get("quota-ns").unwrap();
assert_eq!(state.info.quota, new_quota);
}
#[test]
fn usage_tracking() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("usage-ns").unwrap();
registry
.create_namespace(id, "usage test", "admin", NamespaceQuota::default())
.unwrap();
registry.increment_series("usage-ns", 100).unwrap();
registry.increment_storage("usage-ns", 4096).unwrap();
registry.increment_measurements("usage-ns", 5).unwrap();
let usage = registry.get_usage("usage-ns").unwrap();
assert_eq!(usage.series_count, 100);
assert_eq!(usage.storage_bytes, 4096);
assert_eq!(usage.measurements, 5);
}
#[test]
fn shared_registry_is_arc() {
let reg = shared_registry();
let reg2 = reg.clone();
let id = NamespaceId::new("shared-test").unwrap();
reg.create_namespace(id, "test", "admin", NamespaceQuota::default())
.unwrap();
assert!(reg2.exists("shared-test"));
}
#[test]
fn namespace_id_validation() {
assert!(NamespaceId::new("valid-name").is_ok());
assert!(NamespaceId::new("a123").is_ok());
assert!(NamespaceId::new("").is_err()); assert!(NamespaceId::new("Invalid").is_err()); assert!(NamespaceId::new("has space").is_err()); assert!(NamespaceId::new("-leading").is_err()); assert!(NamespaceId::new("trailing-").is_err()); assert!(NamespaceId::new("a".repeat(64)).is_err()); assert!(NamespaceId::new("a".repeat(63)).is_ok()); }
#[test]
fn usage_ratios_empty_registry() {
let registry = NamespaceRegistry::new();
let ratios = registry.usage_ratios();
assert_eq!(ratios.len(), 3);
for r in &ratios {
assert_eq!(r.namespace, "default");
assert!((r.ratio - 0.0).abs() < f64::EPSILON);
}
}
#[test]
fn usage_ratios_with_usage() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("metrics-ns").unwrap();
let quota = NamespaceQuota {
max_series_count: 100,
max_ingestion_rate: 1000,
max_storage_bytes: 10_000,
max_measurements: 20,
max_request_rps: 0,
max_request_burst: 0,
};
registry
.create_namespace(id, "test", "admin", quota)
.unwrap();
registry.increment_series("metrics-ns", 50).unwrap();
registry.increment_storage("metrics-ns", 7500).unwrap();
registry.increment_measurements("metrics-ns", 10).unwrap();
let ratios = registry.usage_ratios();
let ns_ratios: Vec<_> = ratios
.iter()
.filter(|r| r.namespace == "metrics-ns")
.collect();
assert_eq!(ns_ratios.len(), 3);
let series = ns_ratios
.iter()
.find(|r| r.resource == "series_count")
.unwrap();
assert!((series.ratio - 0.5).abs() < f64::EPSILON);
assert_eq!(series.current, 50);
assert_eq!(series.limit, 100);
let storage = ns_ratios
.iter()
.find(|r| r.resource == "storage_bytes")
.unwrap();
assert!((storage.ratio - 0.75).abs() < f64::EPSILON);
let measurements = ns_ratios
.iter()
.find(|r| r.resource == "measurements")
.unwrap();
assert!((measurements.ratio - 0.5).abs() < f64::EPSILON);
}
#[test]
fn emit_usage_metrics_does_not_panic() {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("emit-ns").unwrap();
registry
.create_namespace(id, "test", "admin", NamespaceQuota::default())
.unwrap();
registry.increment_series("emit-ns", 42).unwrap();
registry.emit_usage_metrics();
}
#[test]
fn usage_ratio_zero_limit() {
let r = ratio(100, 0);
assert_eq!(r, f64::MAX);
let r = ratio(0, 0);
assert!((r - 0.0).abs() < f64::EPSILON);
}
#[test]
fn usage_ratio_over_quota() {
let r = ratio(200, 100);
assert!((r - 2.0).abs() < f64::EPSILON);
}
#[test]
fn open_creates_and_loads_snapshot() {
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path().join("namespaces");
{
let reg = NamespaceRegistry::open(&dir).unwrap();
let id = NamespaceId::new("test-ns").unwrap();
reg.create_namespace(id, "Test", "admin", NamespaceQuota::default())
.unwrap();
assert_eq!(reg.count(), 2);
}
{
let reg = NamespaceRegistry::open(&dir).unwrap();
assert_eq!(reg.count(), 2);
assert!(reg.exists("test-ns"));
assert!(reg.exists(NamespaceId::DEFAULT));
}
}
#[test]
fn open_delete_persists() {
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path().join("namespaces");
{
let reg = NamespaceRegistry::open(&dir).unwrap();
let id = NamespaceId::new("ephemeral").unwrap();
reg.create_namespace(id, "Ephemeral", "admin", NamespaceQuota::default())
.unwrap();
reg.delete_namespace("ephemeral").unwrap();
}
{
let reg = NamespaceRegistry::open(&dir).unwrap();
assert_eq!(reg.count(), 1);
assert!(!reg.exists("ephemeral"));
}
}
#[test]
fn open_quota_update_persists() {
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path().join("namespaces");
{
let reg = NamespaceRegistry::open(&dir).unwrap();
let id = NamespaceId::new("quotans").unwrap();
reg.create_namespace(id, "Quota Test", "admin", NamespaceQuota::default())
.unwrap();
reg.update_quota(
"quotans",
NamespaceQuota {
max_series_count: 42,
..NamespaceQuota::default()
},
)
.unwrap();
}
{
let reg = NamespaceRegistry::open(&dir).unwrap();
let state = reg.get("quotans").unwrap();
assert_eq!(state.info.quota.max_series_count, 42);
}
}
}