i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore(proxy): guard metrics retention cleanup

Browse Source
This commit is contained in:
fawney19
2026-05-08 22:57:00 +08:00
parent a703acd1fe
commit e21fb58479
16 changed files with 562 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
crates/aether-data/src/repository/proxy_nodes/memory.rs
View File

@@ -840,17 +840,33 @@ impl ProxyNodeWriteRepository for InMemoryProxyNodeRepository {
&self,
retain_1m_from_unix_secs: u64,
retain_1h_from_unix_secs: u64,
delete_limit: usize,
) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
let delete_limit = delete_limit.max(1);
let mut metrics_1m = self.metrics_1m.write().expect("proxy node repository lock");
let before_1m = metrics_1m.len();
metrics_1m.retain(|(_, bucket_start), _| *bucket_start >= retain_1m_from_unix_secs);
let deleted_1m_rows = before_1m.saturating_sub(metrics_1m.len());
let expired_1m_keys = metrics_1m
.keys()
.filter(|(_, bucket_start)| *bucket_start < retain_1m_from_unix_secs)
.take(delete_limit)
.cloned()
.collect::<Vec<_>>();
let deleted_1m_rows = expired_1m_keys
.iter()
.filter(|key| metrics_1m.remove(key).is_some())
.count();
drop(metrics_1m);
let mut metrics_1h = self.metrics_1h.write().expect("proxy node repository lock");
let before_1h = metrics_1h.len();
metrics_1h.retain(|(_, bucket_start), _| *bucket_start >= retain_1h_from_unix_secs);
let deleted_1h_rows = before_1h.saturating_sub(metrics_1h.len());
let expired_1h_keys = metrics_1h
.keys()
.filter(|(_, bucket_start)| *bucket_start < retain_1h_from_unix_secs)
.take(delete_limit)
.cloned()
.collect::<Vec<_>>();
let deleted_1h_rows = expired_1h_keys
.iter()
.filter(|key| metrics_1h.remove(key).is_some())
.count();
Ok(ProxyNodeMetricsCleanupSummary {
deleted_1m_rows,

44
crates/aether-data/src/repository/proxy_nodes/mysql.rs
View File

@@ -1031,22 +1031,38 @@ WHERE is_manual = 0
&self,
retain_1m_from_unix_secs: u64,
retain_1h_from_unix_secs: u64,
delete_limit: usize,
) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
let deleted_1m =
sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE bucket_start_unix_secs < ?")
.bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
let delete_limit_i64 = i64::try_from(delete_limit.max(1)).unwrap_or(i64::MAX);
let deleted_1m = sqlx::query(
r#"
DELETE FROM proxy_node_metrics_1m
WHERE bucket_start_unix_secs < ?
ORDER BY bucket_start_unix_secs ASC
LIMIT ?
"#,
)
.bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
.bind(delete_limit_i64)
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
let deleted_1h =
sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE bucket_start_unix_secs < ?")
.bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
let deleted_1h = sqlx::query(
r#"
DELETE FROM proxy_node_metrics_1h
WHERE bucket_start_unix_secs < ?
ORDER BY bucket_start_unix_secs ASC
LIMIT ?
"#,
)
.bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
.bind(delete_limit_i64)
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
Ok(ProxyNodeMetricsCleanupSummary {
deleted_1m_rows: deleted_1m,

58
crates/aether-data/src/repository/proxy_nodes/postgres.rs
View File

@@ -1448,22 +1448,52 @@ WHERE id = $1
&self,
retain_1m_from_unix_secs: u64,
retain_1h_from_unix_secs: u64,
delete_limit: usize,
) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
let deleted_1m =
sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE bucket_start_unix_secs < $1")
.bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
.execute(&self.pool)
.await
.map_postgres_err()?
.rows_affected() as usize;
let delete_limit_i64 = i64::try_from(delete_limit.max(1)).unwrap_or(i64::MAX);
let deleted_1m = sqlx::query(
r#"
WITH expired AS (
SELECT node_id, bucket_start_unix_secs
FROM proxy_node_metrics_1m
WHERE bucket_start_unix_secs < $1
ORDER BY bucket_start_unix_secs ASC
LIMIT $2
)
DELETE FROM proxy_node_metrics_1m metrics
USING expired
WHERE metrics.node_id = expired.node_id
AND metrics.bucket_start_unix_secs = expired.bucket_start_unix_secs
"#,
)
.bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
.bind(delete_limit_i64)
.execute(&self.pool)
.await
.map_postgres_err()?
.rows_affected() as usize;
let deleted_1h =
sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE bucket_start_unix_secs < $1")
.bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
.execute(&self.pool)
.await
.map_postgres_err()?
.rows_affected() as usize;
let deleted_1h = sqlx::query(
r#"
WITH expired AS (
SELECT node_id, bucket_start_unix_secs
FROM proxy_node_metrics_1h
WHERE bucket_start_unix_secs < $1
ORDER BY bucket_start_unix_secs ASC
LIMIT $2
)
DELETE FROM proxy_node_metrics_1h metrics
USING expired
WHERE metrics.node_id = expired.node_id
AND metrics.bucket_start_unix_secs = expired.bucket_start_unix_secs
"#,
)
.bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
.bind(delete_limit_i64)
.execute(&self.pool)
.await
.map_postgres_err()?
.rows_affected() as usize;
Ok(ProxyNodeMetricsCleanupSummary {
deleted_1m_rows: deleted_1m,

54
crates/aether-data/src/repository/proxy_nodes/sqlite.rs
View File

@@ -1034,22 +1034,46 @@ WHERE is_manual = 0
&self,
retain_1m_from_unix_secs: u64,
retain_1h_from_unix_secs: u64,
delete_limit: usize,
) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
let deleted_1m =
sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE bucket_start_unix_secs < ?")
.bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
let delete_limit_i64 = i64::try_from(delete_limit.max(1)).unwrap_or(i64::MAX);
let deleted_1m = sqlx::query(
r#"
DELETE FROM proxy_node_metrics_1m
WHERE (node_id, bucket_start_unix_secs) IN (
SELECT node_id, bucket_start_unix_secs
FROM proxy_node_metrics_1m
WHERE bucket_start_unix_secs < ?
ORDER BY bucket_start_unix_secs ASC
LIMIT ?
)
"#,
)
.bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
.bind(delete_limit_i64)
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
let deleted_1h =
sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE bucket_start_unix_secs < ?")
.bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
let deleted_1h = sqlx::query(
r#"
DELETE FROM proxy_node_metrics_1h
WHERE (node_id, bucket_start_unix_secs) IN (
SELECT node_id, bucket_start_unix_secs
FROM proxy_node_metrics_1h
WHERE bucket_start_unix_secs < ?
ORDER BY bucket_start_unix_secs ASC
LIMIT ?
)
"#,
)
.bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
.bind(delete_limit_i64)
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected() as usize;
Ok(ProxyNodeMetricsCleanupSummary {
deleted_1m_rows: deleted_1m,
@@ -1627,7 +1651,7 @@ VALUES ('node-1', 'registered', 'ok', 3)
);
let cleanup = repository
.cleanup_proxy_node_metrics(now.saturating_add(1), now.saturating_add(1))
.cleanup_proxy_node_metrics(now.saturating_add(1), now.saturating_add(1), 10)
.await
.expect("cleanup should run");
assert_eq!(cleanup.deleted_1m_rows, 1);

1
crates/aether-data/src/repository/proxy_nodes/types.rs
View File

@@ -717,6 +717,7 @@ pub trait ProxyNodeWriteRepository: Send + Sync {
&self,
retain_1m_from_unix_secs: u64,
retain_1h_from_unix_secs: u64,
delete_limit: usize,
) -> Result<ProxyNodeMetricsCleanupSummary, crate::DataLayerError>;
}