use tracing::warn;
use crate::tenant::error::{Result, TenantError};
use crate::tenant::registry::NamespaceRegistry;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QuotaResource {
SeriesCount,
IngestionRate,
StorageBytes,
Measurements,
}
impl std::fmt::Display for QuotaResource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::SeriesCount => write!(f, "series_count"),
Self::IngestionRate => write!(f, "ingestion_rate"),
Self::StorageBytes => write!(f, "storage_bytes"),
Self::Measurements => write!(f, "measurements"),
}
}
}
#[derive(Debug)]
pub struct QuotaEnforcer<'a> {
registry: &'a NamespaceRegistry,
}
impl<'a> QuotaEnforcer<'a> {
#[must_use]
pub fn new(registry: &'a NamespaceRegistry) -> Self {
Self { registry }
}
#[doc(hidden)]
#[deprecated(
since = "0.2.0",
note = "Non-atomic (TOCTOU). Use `check_and_increment_write` instead."
)]
pub fn check_write(&self, namespace: &str, new_series: u64, new_points: u64) -> Result<()> {
let state = self.registry.get(namespace)?;
let quota = &state.info.quota;
let usage = &state.usage;
let projected_series = usage.series_count + new_series;
Self::emit_ratio(
namespace,
QuotaResource::SeriesCount,
projected_series,
quota.max_series_count,
);
if projected_series > quota.max_series_count {
warn!(
namespace,
current = usage.series_count,
new = new_series,
limit = quota.max_series_count,
"series count quota exceeded"
);
return Err(TenantError::QuotaExceeded {
namespace: namespace.to_string(),
resource: QuotaResource::SeriesCount.to_string(),
current: usage.series_count,
limit: quota.max_series_count,
});
}
#[allow(clippy::cast_precision_loss)]
let projected_rate = usage.ingestion_rate + new_points as f64;
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
let projected_rate_u64 = projected_rate as u64;
Self::emit_ratio(
namespace,
QuotaResource::IngestionRate,
projected_rate_u64,
quota.max_ingestion_rate,
);
#[allow(clippy::cast_precision_loss)]
let rate_limit = quota.max_ingestion_rate as f64;
if projected_rate > rate_limit {
warn!(
namespace,
current_rate = usage.ingestion_rate,
new_points,
limit = quota.max_ingestion_rate,
"ingestion rate quota exceeded"
);
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
let current = usage.ingestion_rate as u64;
return Err(TenantError::QuotaExceeded {
namespace: namespace.to_string(),
resource: QuotaResource::IngestionRate.to_string(),
current,
limit: quota.max_ingestion_rate,
});
}
const STORAGE_QUOTA_MARGIN: f64 = 0.95;
let effective_limit = (quota.max_storage_bytes as f64 * STORAGE_QUOTA_MARGIN) as u64;
Self::emit_ratio(
namespace,
QuotaResource::StorageBytes,
usage.storage_bytes,
quota.max_storage_bytes,
);
if usage.storage_bytes > effective_limit {
return Err(TenantError::QuotaExceeded {
namespace: namespace.to_string(),
resource: QuotaResource::StorageBytes.to_string(),
current: usage.storage_bytes,
limit: quota.max_storage_bytes,
});
}
Ok(())
}
pub fn check_create_measurement(&self, namespace: &str) -> Result<()> {
let state = self.registry.get(namespace)?;
let quota = &state.info.quota;
let usage = &state.usage;
let projected = u64::from(usage.measurements) + 1;
Self::emit_ratio(
namespace,
QuotaResource::Measurements,
projected,
u64::from(quota.max_measurements),
);
if projected > u64::from(quota.max_measurements) {
return Err(TenantError::QuotaExceeded {
namespace: namespace.to_string(),
resource: QuotaResource::Measurements.to_string(),
current: u64::from(usage.measurements),
limit: u64::from(quota.max_measurements),
});
}
Ok(())
}
pub fn check_and_increment_write(
&self,
namespace: &str,
new_series: u64,
new_points: u64,
) -> Result<()> {
self.registry
.check_and_increment_write(namespace, new_series, new_points)
}
pub fn check_and_increment_measurement(&self, namespace: &str) -> Result<()> {
self.registry.check_and_increment_measurement(namespace)
}
#[must_use]
pub fn check_all(&self, namespace: &str) -> Vec<QuotaResource> {
let Ok(state) = self.registry.get(namespace) else {
return Vec::new();
};
let quota = &state.info.quota;
let usage = &state.usage;
let mut violations = Vec::new();
if usage.series_count > quota.max_series_count {
violations.push(QuotaResource::SeriesCount);
}
#[allow(clippy::cast_precision_loss)]
if usage.ingestion_rate > quota.max_ingestion_rate as f64 {
violations.push(QuotaResource::IngestionRate);
}
if usage.storage_bytes > quota.max_storage_bytes {
violations.push(QuotaResource::StorageBytes);
}
if u64::from(usage.measurements) > u64::from(quota.max_measurements) {
violations.push(QuotaResource::Measurements);
}
violations
}
#[must_use]
pub fn usage_ratio(current: u64, limit: u64) -> f64 {
if limit == 0 {
return if current > 0 { f64::MAX } else { 0.0 };
}
#[allow(clippy::cast_precision_loss)]
let ratio = current as f64 / limit as f64;
ratio
}
fn emit_ratio(namespace: &str, resource: QuotaResource, current: u64, limit: u64) {
let ratio = Self::usage_ratio(current, limit);
metrics::gauge!(
"chronix_namespace_usage_ratio",
"namespace" => namespace.to_string(),
"resource" => resource.to_string(),
)
.set(ratio);
}
}
#[cfg(test)]
#[allow(deprecated)]
mod tests {
use super::*;
use chronix_core::{NamespaceId, NamespaceQuota, NamespaceUsage};
fn setup_registry() -> NamespaceRegistry {
let registry = NamespaceRegistry::new();
let id = NamespaceId::new("test-ns").unwrap();
let quota = NamespaceQuota {
max_series_count: 100,
max_ingestion_rate: 1000,
max_storage_bytes: 1_000_000,
max_measurements: 10,
max_request_rps: 0,
max_request_burst: 0,
};
registry
.create_namespace(id, "test", "admin", quota)
.unwrap();
registry
}
#[test]
fn write_within_quota_passes() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
enforcer.check_write("test-ns", 5, 100).unwrap();
}
#[test]
fn write_exceeds_series_quota() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
let err = enforcer.check_write("test-ns", 101, 10).unwrap_err();
assert!(matches!(
err,
TenantError::QuotaExceeded {
resource,
..
} if resource == "series_count"
));
}
#[test]
fn write_exceeds_ingestion_rate() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
let err = enforcer.check_write("test-ns", 0, 1001).unwrap_err();
assert!(matches!(
err,
TenantError::QuotaExceeded {
resource,
..
} if resource == "ingestion_rate"
));
}
#[test]
fn write_exceeds_storage_quota() {
let registry = setup_registry();
let usage = NamespaceUsage {
storage_bytes: 1_000_001,
..Default::default()
};
registry.update_usage("test-ns", usage).unwrap();
let enforcer = QuotaEnforcer::new(®istry);
let err = enforcer.check_write("test-ns", 0, 1).unwrap_err();
assert!(matches!(
err,
TenantError::QuotaExceeded {
resource,
..
} if resource == "storage_bytes"
));
}
#[test]
fn measurement_quota_enforcement() {
let registry = setup_registry();
let usage = NamespaceUsage {
measurements: 10,
..Default::default()
};
registry.update_usage("test-ns", usage).unwrap();
let enforcer = QuotaEnforcer::new(®istry);
let err = enforcer.check_create_measurement("test-ns").unwrap_err();
assert!(matches!(
err,
TenantError::QuotaExceeded {
resource,
..
} if resource == "measurements"
));
}
#[test]
fn check_all_returns_violations() {
let registry = setup_registry();
let usage = NamespaceUsage {
series_count: 200, storage_bytes: 2_000_000, ..Default::default()
};
registry.update_usage("test-ns", usage).unwrap();
let enforcer = QuotaEnforcer::new(®istry);
let violations = enforcer.check_all("test-ns");
assert_eq!(violations.len(), 2);
assert!(violations.contains(&QuotaResource::SeriesCount));
assert!(violations.contains(&QuotaResource::StorageBytes));
}
#[test]
fn check_nonexistent_namespace() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
let err = enforcer.check_write("nope", 1, 1).unwrap_err();
assert!(matches!(err, TenantError::NamespaceNotFound(_)));
}
#[test]
fn usage_ratio_calculation() {
assert!((QuotaEnforcer::usage_ratio(50, 100) - 0.5).abs() < f64::EPSILON);
assert!((QuotaEnforcer::usage_ratio(100, 100) - 1.0).abs() < f64::EPSILON);
assert!((QuotaEnforcer::usage_ratio(150, 100) - 1.5).abs() < f64::EPSILON);
assert!((QuotaEnforcer::usage_ratio(0, 100) - 0.0).abs() < f64::EPSILON);
assert!((QuotaEnforcer::usage_ratio(0, 0) - 0.0).abs() < f64::EPSILON);
}
#[test]
fn usage_ratio_zero_limit_json_safe() {
let r = QuotaEnforcer::usage_ratio(100, 0);
assert_eq!(r, f64::MAX);
assert!(r.is_finite());
let r = QuotaEnforcer::usage_ratio(0, 0);
assert_eq!(r, 0.0);
}
#[test]
fn quota_resource_display() {
assert_eq!(QuotaResource::SeriesCount.to_string(), "series_count");
assert_eq!(QuotaResource::IngestionRate.to_string(), "ingestion_rate");
assert_eq!(QuotaResource::StorageBytes.to_string(), "storage_bytes");
assert_eq!(QuotaResource::Measurements.to_string(), "measurements");
}
#[test]
fn default_namespace_has_generous_quotas() {
let registry = NamespaceRegistry::new();
let enforcer = QuotaEnforcer::new(®istry);
enforcer.check_write("default", 100, 10_000).unwrap();
}
#[test]
fn check_and_increment_write_updates_usage() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
enforcer
.check_and_increment_write("test-ns", 10, 50)
.unwrap();
let usage = registry.get_usage("test-ns").unwrap();
assert_eq!(usage.series_count, 10);
}
#[test]
fn check_and_increment_write_rejects_over_quota() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
let err = enforcer
.check_and_increment_write("test-ns", 101, 10)
.unwrap_err();
assert!(matches!(
err,
TenantError::QuotaExceeded { resource, .. }
if resource == "series_count"
));
let usage = registry.get_usage("test-ns").unwrap();
assert_eq!(usage.series_count, 0);
}
#[test]
fn check_and_increment_write_sequential_accumulation() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
enforcer
.check_and_increment_write("test-ns", 50, 0)
.unwrap();
enforcer
.check_and_increment_write("test-ns", 50, 0)
.unwrap();
let err = enforcer
.check_and_increment_write("test-ns", 1, 0)
.unwrap_err();
assert!(matches!(err, TenantError::QuotaExceeded { .. }));
let usage = registry.get_usage("test-ns").unwrap();
assert_eq!(usage.series_count, 100);
}
#[test]
fn check_and_increment_measurement_works() {
let registry = setup_registry();
let enforcer = QuotaEnforcer::new(®istry);
for _ in 0..10 {
enforcer.check_and_increment_measurement("test-ns").unwrap();
}
let err = enforcer
.check_and_increment_measurement("test-ns")
.unwrap_err();
assert!(matches!(err, TenantError::QuotaExceeded { .. }));
let usage = registry.get_usage("test-ns").unwrap();
assert_eq!(usage.measurements, 10);
}
#[test]
fn check_and_increment_concurrent_safety() {
use std::sync::Arc;
use std::thread;
let registry = Arc::new(NamespaceRegistry::new());
let id = NamespaceId::new("concurrent-ns").unwrap();
let quota = NamespaceQuota {
max_series_count: 100,
max_ingestion_rate: 1_000_000,
max_storage_bytes: 1_000_000_000,
max_measurements: 1000,
max_request_rps: 0,
max_request_burst: 0,
};
registry
.create_namespace(id, "test", "admin", quota)
.unwrap();
let handles: Vec<_> = (0..20)
.map(|_| {
let reg = Arc::clone(®istry);
thread::spawn(move || {
let mut success = 0u64;
for _ in 0..10 {
if reg.check_and_increment_write("concurrent-ns", 1, 0).is_ok() {
success += 1;
}
}
success
})
})
.collect();
let total_success: u64 = handles.into_iter().map(|h| h.join().unwrap()).sum();
assert_eq!(total_success, 100);
let usage = registry.get_usage("concurrent-ns").unwrap();
assert_eq!(usage.series_count, 100);
}
}