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

Merge pull request #405 from Entropy-Xu/codex/async-cleanup-records

Browse Source 
Add async request body cleanup records
This commit is contained in:
fawney19
2026-05-10 01:12:12 +08:00
committed by GitHub
parent 209322b499 a81053e6ff
commit 3c1456706a
16 changed files with 1793 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
apps/aether-gateway/src/control/route/admin/system_families.rs
View File

@@ -95,6 +95,14 @@ pub(super) fn classify_admin_system_family_route(
"admin:system",
false,
))
} else if method == http::Method::GET && normalized_path == "/api/admin/system/cleanup/runs" {
Some(classified(
"admin_proxy",
"system_manage",
"cleanup_runs",
"admin:system",
false,
))
} else if method == http::Method::POST && normalized_path == "/api/admin/system/purge/config" {
Some(classified(
"admin_proxy",
@@ -139,6 +147,16 @@ pub(super) fn classify_admin_system_family_route(
"admin:system",
false,
))
} else if method == http::Method::POST
&& normalized_path == "/api/admin/system/purge/request-bodies/task"
{
Some(classified(
"admin_proxy",
"system_manage",
"purge_request_bodies_task",
"admin:system",
false,
))
} else if method == http::Method::POST && normalized_path == "/api/admin/system/purge/stats" {
Some(classified(
"admin_proxy",

23
apps/aether-gateway/src/control/tests/admin_core.rs
View File

@@ -172,6 +172,10 @@ fn classifies_admin_system_maintenance_write_routes_as_admin_proxy_route() {
"/api/admin/system/purge/request-bodies",
"purge_request_bodies",
),
(
"/api/admin/system/purge/request-bodies/task",
"purge_request_bodies_task",
),
("/api/admin/system/purge/stats", "purge_stats"),
];
@@ -246,6 +250,25 @@ fn classifies_admin_system_stats_as_admin_proxy_route() {
assert!(!decision.is_execution_runtime_candidate());
}
#[test]
fn classifies_admin_system_cleanup_runs_as_admin_proxy_route() {
let headers = headers(&[]);
let uri: Uri = "/api/admin/system/cleanup/runs"
.parse()
.expect("uri should parse");
let decision =
classify_control_route(&http::Method::GET, &uri, &headers).expect("route should classify");
assert_eq!(decision.route_class.as_deref(), Some("admin_proxy"));
assert_eq!(decision.route_family.as_deref(), Some("system_manage"));
assert_eq!(decision.route_kind.as_deref(), Some("cleanup_runs"));
assert_eq!(
decision.auth_endpoint_signature.as_deref(),
Some("admin:system")
);
assert!(!decision.is_execution_runtime_candidate());
}
#[test]
fn classifies_admin_system_settings_set_as_admin_proxy_route() {
let headers = headers(&[]);

10
apps/aether-gateway/src/data/state/core.rs
View File

@@ -461,4 +461,14 @@ impl GatewayDataState {
None => Ok(aether_data::repository::system::AdminSystemPurgeSummary::default()),
}
}
pub(crate) async fn purge_admin_request_bodies_batch(
&self,
batch_size: usize,
) -> Result<aether_data::repository::system::AdminSystemPurgeSummary, DataLayerError> {
match self.backends.as_ref() {
Some(backends) => backends.purge_admin_request_bodies_batch(batch_size).await,
None => Ok(aether_data::repository::system::AdminSystemPurgeSummary::default()),
}
}
}

119
apps/aether-gateway/src/handlers/admin/system/core/system_routes.rs
View File

@@ -18,7 +18,6 @@ use crate::handlers::admin::system::shared::settings::{
};
use crate::handlers::admin::system::shared::smtp::build_admin_smtp_test_payload;
use crate::GatewayError;
use aether_data::repository::system::AdminSystemPurgeTarget;
use axum::{
body::{Body, Bytes},
http,
@@ -26,6 +25,7 @@ use axum::{
Json,
};
use serde_json::json;
use std::time::Instant;
pub(super) async fn maybe_build_local_admin_core_system_response(
state: &AdminAppState<'_>,
@@ -194,15 +194,31 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
)));
}
if let Some((target, action, object_type, object_id)) =
admin_system_purge_target_for_route_kind(decision.route_kind.as_deref())
if decision.route_kind.as_deref() == Some("cleanup_runs")
&& request_method == http::Method::GET
&& request_path == "/api/admin/system/cleanup/runs"
{
let records = crate::maintenance::list_admin_cleanup_run_records(&state.app().data)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
return Ok(Some(Json(json!({ "items": records })).into_response()));
}
if let Some((task_kind, action, object_type, object_id)) =
admin_system_purge_task_for_route_kind(decision.route_kind.as_deref())
{
if request_method != http::Method::POST {
return Ok(None);
}
let task = crate::maintenance::start_admin_system_purge_task(state.cloned_app(), task_kind)
.await?;
return Ok(Some(attach_admin_audit_response(
Json(build_admin_system_purge_payload(state, target).await?).into_response(),
"admin_system_data_purged",
Json(json!({
"message": task.message.clone(),
"task": task,
}))
.into_response(),
"admin_system_purge_task_started",
action,
object_type,
object_id,
@@ -435,100 +451,60 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
Ok(None)
}
fn admin_system_purge_target_for_route_kind(
fn admin_system_purge_task_for_route_kind(
route_kind: Option<&str>,
) -> Option<(
AdminSystemPurgeTarget,
crate::maintenance::AdminCleanupTaskKind,
&'static str,
&'static str,
&'static str,
)> {
match route_kind {
Some("purge_config") => Some((
AdminSystemPurgeTarget::Config,
"purge_system_config",
crate::maintenance::AdminCleanupTaskKind::Config,
"purge_system_config_async",
"system_config",
"global",
)),
Some("purge_users") => Some((
AdminSystemPurgeTarget::Users,
"purge_non_admin_users",
crate::maintenance::AdminCleanupTaskKind::Users,
"purge_non_admin_users_async",
"users",
"non_admin",
)),
Some("purge_usage") => Some((
AdminSystemPurgeTarget::Usage,
"purge_usage_records",
crate::maintenance::AdminCleanupTaskKind::Usage,
"purge_usage_records_async",
"usage",
"all",
)),
Some("purge_audit_logs") => Some((
AdminSystemPurgeTarget::AuditLogs,
"purge_audit_logs",
crate::maintenance::AdminCleanupTaskKind::AuditLogs,
"purge_audit_logs_async",
"audit_logs",
"all",
)),
Some("purge_request_bodies") => Some((
AdminSystemPurgeTarget::RequestBodies,
"purge_request_bodies",
Some("purge_request_bodies") | Some("purge_request_bodies_task") => Some((
crate::maintenance::AdminCleanupTaskKind::RequestBodies,
"purge_request_bodies_async",
"request_bodies",
"all",
)),
Some("purge_stats") => Some((AdminSystemPurgeTarget::Stats, "purge_stats", "stats", "all")),
Some("purge_stats") => Some((
crate::maintenance::AdminCleanupTaskKind::Stats,
"purge_stats_async",
"stats",
"all",
)),
_ => None,
}
}
async fn build_admin_system_purge_payload(
state: &AdminAppState<'_>,
target: AdminSystemPurgeTarget,
) -> Result<serde_json::Value, GatewayError> {
let summary = state.purge_admin_system_data(target).await?;
let total = summary.total();
let affected = summary.affected.clone();
if target == AdminSystemPurgeTarget::Stats {
let rebuild = state.rebuild_admin_stats_once().await?;
let message = if rebuild.capped {
format!(
"统计聚合已清空，已重建 {} 个小时桶和 {} 个日桶，仍有历史统计待后台任务继续重建",
rebuild.hourly_buckets, rebuild.daily_buckets
)
} else {
format!(
"统计聚合已清空并重建，删除 {} 行，重建 {} 个小时桶和 {} 个日桶",
total, rebuild.hourly_buckets, rebuild.daily_buckets
)
};
return Ok(json!({
"message": message,
"deleted": affected,
"rebuilt": {
"hourly_buckets": rebuild.hourly_buckets,
"daily_buckets": rebuild.daily_buckets,
"capped": rebuild.capped,
},
}));
}
let (message, count_key) = match target {
AdminSystemPurgeTarget::Config => ("系统配置已清空", "deleted"),
AdminSystemPurgeTarget::Users => ("非管理员用户已清空", "deleted"),
AdminSystemPurgeTarget::Usage => ("使用记录已清空", "deleted"),
AdminSystemPurgeTarget::AuditLogs => ("审计日志已清空", "deleted"),
AdminSystemPurgeTarget::RequestBodies => ("请求/响应体已清空", "cleaned"),
AdminSystemPurgeTarget::Stats => unreachable!("stats handled above"),
};
Ok(json!({
"message": format!("{message}，影响 {} 行", total),
count_key: affected,
}))
}
async fn build_admin_system_cleanup_payload(
state: &AdminAppState<'_>,
) -> Result<serde_json::Value, GatewayError> {
let started_at_unix_secs = chrono::Utc::now().timestamp().max(0) as u64;
let started_at = Instant::now();
let summary = state.run_admin_system_cleanup_once().await?;
let cleaned = json!({
"audit_logs": summary.audit_logs_deleted,
@@ -558,6 +534,17 @@ async fn build_admin_system_cleanup_payload(
.saturating_add(summary.usage.keys_cleaned)
.saturating_add(summary.usage.records_deleted);
crate::maintenance::record_completed_cleanup_run(
&state.app().data,
"system_cleanup",
"manual",
started_at_unix_secs,
started_at,
cleaned.clone(),
format!("系统清理已执行，影响 {total} 项"),
)
.await;
Ok(json!({
"message": format!("系统清理已执行，影响 {} 项", total),
"cleaned": cleaned,

18
apps/aether-gateway/src/maintenance/mod.rs
View File

@@ -4,8 +4,9 @@ mod tests;
pub(crate) use runtime::{
cancel_proxy_upgrade_rollout, clear_proxy_upgrade_rollout_conflicts,
inspect_proxy_upgrade_rollout, perform_oauth_token_refresh_once, perform_pool_quota_probe_once,
perform_provider_checkin_once, rebuild_admin_stats_once, record_proxy_upgrade_traffic_success,
inspect_proxy_upgrade_rollout, list_admin_cleanup_run_records,
perform_oauth_token_refresh_once, perform_pool_quota_probe_once, perform_provider_checkin_once,
rebuild_admin_stats_once, record_completed_cleanup_run, record_proxy_upgrade_traffic_success,
restore_proxy_upgrade_rollout_skipped_nodes, retry_proxy_upgrade_rollout_node,
run_admin_system_cleanup_once, skip_proxy_upgrade_rollout_node, spawn_audit_cleanup_worker,
spawn_db_maintenance_worker, spawn_gemini_file_mapping_cleanup_worker,
@@ -15,10 +16,11 @@ pub(crate) use runtime::{
spawn_proxy_upgrade_rollout_worker, spawn_request_candidate_cleanup_worker,
spawn_stats_aggregation_worker, spawn_stats_hourly_aggregation_worker,
spawn_usage_cleanup_worker, spawn_wallet_daily_usage_aggregation_worker,
start_proxy_upgrade_rollout, AdminStatsRebuildSummary, AdminSystemCleanupSummary,
OAuthTokenRefreshRunSummary, PoolQuotaProbeRunSummary, ProviderCheckinRunSummary,
ProxyUpgradeRolloutCancelSummary, ProxyUpgradeRolloutConflictClearSummary,
ProxyUpgradeRolloutNodeActionSummary, ProxyUpgradeRolloutProbeConfig,
ProxyUpgradeRolloutSkippedRestoreSummary, ProxyUpgradeRolloutStatus,
ProxyUpgradeRolloutTrackedNodeState,
start_admin_request_body_cleanup_task, start_admin_system_purge_task,
start_proxy_upgrade_rollout, AdminCleanupRunRecord, AdminCleanupTaskKind,
AdminStatsRebuildSummary, AdminSystemCleanupSummary, OAuthTokenRefreshRunSummary,
PoolQuotaProbeRunSummary, ProviderCheckinRunSummary, ProxyUpgradeRolloutCancelSummary,
ProxyUpgradeRolloutConflictClearSummary, ProxyUpgradeRolloutNodeActionSummary,
ProxyUpgradeRolloutProbeConfig, ProxyUpgradeRolloutSkippedRestoreSummary,
ProxyUpgradeRolloutStatus, ProxyUpgradeRolloutTrackedNodeState,
};

7
apps/aether-gateway/src/maintenance/runtime.rs
View File

@@ -8,6 +8,8 @@ use crate::{AppState, GatewayError};
#[path = "runtime/audit_cleanup.rs"]
mod audit_cleanup;
#[path = "runtime/cleanup_runs.rs"]
mod cleanup_runs;
#[path = "runtime/config.rs"]
mod config;
#[path = "runtime/db_maintenance.rs"]
@@ -49,6 +51,11 @@ pub(crate) use aether_data_contracts::repository::usage::{
UsageCleanupSummary, UsageCleanupWindow,
};
use audit_cleanup::*;
pub(crate) use cleanup_runs::{
list_admin_cleanup_run_records, record_completed_cleanup_run, record_failed_cleanup_run,
start_admin_request_body_cleanup_task, start_admin_system_purge_task, AdminCleanupRunRecord,
AdminCleanupTaskKind,
};
use config::*;
use db_maintenance::*;
pub(crate) use oauth_token_refresh::{

518
apps/aether-gateway/src/maintenance/runtime/cleanup_runs.rs Normal file
View File

@@ -0,0 +1,518 @@
use std::time::Instant;
use aether_data::repository::system::{AdminSystemPurgeSummary, AdminSystemPurgeTarget};
use aether_data_contracts::DataLayerError;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tracing::{info, warn};
use crate::data::GatewayDataState;
use crate::{AppState, GatewayError};
use super::{now_unix_secs, system_config_usize};
const CLEANUP_RUN_HISTORY_KEY: &str = "admin_cleanup_run_history";
const CLEANUP_RUN_HISTORY_LIMIT: usize = 50;
const REQUEST_BODY_PROGRESS_UPDATE_BATCHES: usize = 10;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct AdminCleanupRunRecord {
pub(crate) id: String,
pub(crate) kind: String,
pub(crate) trigger: String,
pub(crate) status: String,
pub(crate) message: String,
pub(crate) started_at_unix_secs: u64,
pub(crate) completed_at_unix_secs: Option<u64>,
pub(crate) duration_ms: Option<u64>,
pub(crate) summary: Value,
pub(crate) error: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum AdminCleanupTaskKind {
Config,
Users,
RequestBodies,
Usage,
AuditLogs,
Stats,
}
impl AdminCleanupTaskKind {
fn record_kind(self) -> &'static str {
match self {
Self::Config => "config_purge",
Self::Users => "users_purge",
Self::RequestBodies => "request_bodies",
Self::Usage => "usage_purge",
Self::AuditLogs => "audit_logs_purge",
Self::Stats => "stats_purge",
}
}
fn target(self) -> Option<AdminSystemPurgeTarget> {
match self {
Self::RequestBodies => None,
Self::Config => Some(AdminSystemPurgeTarget::Config),
Self::Users => Some(AdminSystemPurgeTarget::Users),
Self::Usage => Some(AdminSystemPurgeTarget::Usage),
Self::AuditLogs => Some(AdminSystemPurgeTarget::AuditLogs),
Self::Stats => Some(AdminSystemPurgeTarget::Stats),
}
}
fn start_message(self, batch_size: Option<usize>) -> String {
match self {
Self::Config => "系统配置后台清空已开始".to_string(),
Self::Users => "非管理员用户后台清空已开始".to_string(),
Self::RequestBodies => format!(
"请求/响应体后台清理已开始，每批 {} 条",
batch_size.unwrap_or(1)
),
Self::Usage => "使用记录后台清空已开始".to_string(),
Self::AuditLogs => "审计日志后台清空已开始".to_string(),
Self::Stats => "统计聚合后台清空和重建已开始".to_string(),
}
}
fn failure_message(self) -> &'static str {
match self {
Self::Config => "系统配置后台清空失败",
Self::Users => "非管理员用户后台清空失败",
Self::RequestBodies => "请求/响应体后台清理失败",
Self::Usage => "使用记录后台清空失败",
Self::AuditLogs => "审计日志后台清空失败",
Self::Stats => "统计聚合后台清空或重建失败",
}
}
}
pub(crate) async fn list_admin_cleanup_run_records(
data: &GatewayDataState,
) -> Result<Vec<AdminCleanupRunRecord>, DataLayerError> {
let Some(value) = data
.find_system_config_value(CLEANUP_RUN_HISTORY_KEY)
.await?
else {
return Ok(Vec::new());
};
Ok(parse_cleanup_run_records(value))
}
pub(crate) async fn start_admin_request_body_cleanup_task(
app: AppState,
) -> Result<AdminCleanupRunRecord, GatewayError> {
start_admin_system_purge_task(app, AdminCleanupTaskKind::RequestBodies).await
}
pub(crate) async fn start_admin_system_purge_task(
app: AppState,
kind: AdminCleanupTaskKind,
) -> Result<AdminCleanupRunRecord, GatewayError> {
let data = app.data.clone();
let records = list_admin_cleanup_run_records(&data)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
if let Some(existing) = records
.iter()
.find(|record| record.kind == kind.record_kind() && record.status == "processing")
.cloned()
{
return Ok(existing);
}
let batch_size = if kind == AdminCleanupTaskKind::RequestBodies {
Some(
system_config_usize(&data, "cleanup_batch_size", 1_000)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?
.max(1),
)
} else {
None
};
let started_at = now_unix_secs();
let record = AdminCleanupRunRecord {
id: uuid::Uuid::new_v4().to_string(),
kind: kind.record_kind().to_string(),
trigger: "manual".to_string(),
status: "processing".to_string(),
message: kind.start_message(batch_size),
started_at_unix_secs: started_at,
completed_at_unix_secs: None,
duration_ms: None,
summary: initial_task_summary(kind, batch_size),
error: None,
};
record_cleanup_run(&data, record.clone())
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
match kind {
AdminCleanupTaskKind::RequestBodies => {
tokio::spawn(run_request_body_cleanup_task(
data,
record.clone(),
batch_size.unwrap_or(1),
));
}
AdminCleanupTaskKind::Usage
| AdminCleanupTaskKind::AuditLogs
| AdminCleanupTaskKind::Config
| AdminCleanupTaskKind::Users
| AdminCleanupTaskKind::Stats => {
tokio::spawn(run_admin_system_purge_task(app, record.clone(), kind));
}
}
Ok(record)
}
pub(crate) async fn record_completed_cleanup_run(
data: &GatewayDataState,
kind: &str,
trigger: &str,
started_at_unix_secs: u64,
started_at: Instant,
summary: Value,
message: impl Into<String>,
) {
let record = AdminCleanupRunRecord {
id: uuid::Uuid::new_v4().to_string(),
kind: kind.to_string(),
trigger: trigger.to_string(),
status: "completed".to_string(),
message: message.into(),
started_at_unix_secs,
completed_at_unix_secs: Some(now_unix_secs()),
duration_ms: Some(
started_at
.elapsed()
.as_millis()
.try_into()
.unwrap_or(u64::MAX),
),
summary,
error: None,
};
if let Err(err) = record_cleanup_run(data, record).await {
warn!(error = %err, kind, "failed to record cleanup run");
}
}
pub(crate) async fn record_failed_cleanup_run(
data: &GatewayDataState,
kind: &str,
trigger: &str,
started_at_unix_secs: u64,
started_at: Instant,
error: &DataLayerError,
) {
let record = AdminCleanupRunRecord {
id: uuid::Uuid::new_v4().to_string(),
kind: kind.to_string(),
trigger: trigger.to_string(),
status: "failed".to_string(),
message: "清理执行失败".to_string(),
started_at_unix_secs,
completed_at_unix_secs: Some(now_unix_secs()),
duration_ms: Some(
started_at
.elapsed()
.as_millis()
.try_into()
.unwrap_or(u64::MAX),
),
summary: json!({}),
error: Some(error.to_string()),
};
if let Err(err) = record_cleanup_run(data, record).await {
warn!(error = %err, kind, "failed to record failed cleanup run");
}
}
async fn run_request_body_cleanup_task(
data: std::sync::Arc<GatewayDataState>,
initial_record: AdminCleanupRunRecord,
batch_size: usize,
) {
let started_at = Instant::now();
let mut total = AdminSystemPurgeSummary::default();
let mut batches = 0usize;
loop {
match data.purge_admin_request_bodies_batch(batch_size).await {
Ok(batch) if batch.total() == 0 => break,
Ok(batch) => {
batches = batches.saturating_add(1);
total.merge(&batch);
if batches.is_multiple_of(REQUEST_BODY_PROGRESS_UPDATE_BATCHES) {
let progress = request_body_cleanup_record(
&initial_record,
"processing",
format!(
"请求/响应体后台清理中，已处理 {} 批，影响 {} 行",
batches,
total.total()
),
batches,
batch_size,
&total,
Some(started_at),
None,
);
if let Err(err) = record_cleanup_run(&data, progress).await {
warn!(error = %err, "failed to update request body cleanup progress");
}
}
tokio::task::yield_now().await;
}
Err(err) => {
let failed = request_body_cleanup_record(
&initial_record,
"failed",
"请求/响应体后台清理失败".to_string(),
batches,
batch_size,
&total,
Some(started_at),
Some(err.to_string()),
);
if let Err(record_err) = record_cleanup_run(&data, failed).await {
warn!(error = %record_err, "failed to record request body cleanup failure");
}
warn!(error = %err, "request body cleanup task failed");
return;
}
}
}
let completed = request_body_cleanup_record(
&initial_record,
"completed",
format!(
"请求/响应体后台清理完成，影响 {} 行，共 {} 批",
total.total(),
batches
),
batches,
batch_size,
&total,
Some(started_at),
None,
);
if let Err(err) = record_cleanup_run(&data, completed).await {
warn!(error = %err, "failed to record request body cleanup completion");
}
info!(
event_name = "request_body_cleanup_task_completed",
log_type = "ops",
worker = "request_body_cleanup_task",
batches,
affected = total.total(),
"gateway finished request body cleanup task"
);
}
async fn run_admin_system_purge_task(
app: AppState,
initial_record: AdminCleanupRunRecord,
kind: AdminCleanupTaskKind,
) {
let started_at = Instant::now();
let data = app.data.clone();
match run_admin_system_purge_task_once(&app, kind).await {
Ok((summary, message)) => {
let completed = cleanup_task_record(
&initial_record,
"completed",
message,
summary,
Some(started_at),
None,
);
if let Err(err) = record_cleanup_run(&data, completed).await {
warn!(error = %err, "failed to record admin system purge task completion");
}
info!(
event_name = "admin_system_purge_task_completed",
log_type = "ops",
worker = "admin_system_purge_task",
kind = ?kind,
"gateway finished admin system purge task"
);
}
Err(err) => {
let failed = cleanup_task_record(
&initial_record,
"failed",
kind.failure_message().to_string(),
json!({}),
Some(started_at),
Some(format!("{err:?}")),
);
if let Err(record_err) = record_cleanup_run(&data, failed).await {
warn!(error = %record_err, "failed to record admin system purge task failure");
}
warn!(error = ?err, kind = ?kind, "admin system purge task failed");
}
}
}
async fn run_admin_system_purge_task_once(
app: &AppState,
kind: AdminCleanupTaskKind,
) -> Result<(Value, String), GatewayError> {
let target = kind
.target()
.expect("non-request-body cleanup task should map to purge target");
let purge = app.purge_admin_system_data(target).await?;
let deleted_total = purge.total();
let affected = purge.affected.clone();
if kind == AdminCleanupTaskKind::Stats {
let rebuild = app.rebuild_admin_stats_once().await?;
let message = if rebuild.capped {
format!(
"统计聚合后台清空完成，已重建 {} 个小时桶和 {} 个日桶，仍有历史统计待后台任务继续重建",
rebuild.hourly_buckets, rebuild.daily_buckets
)
} else {
format!(
"统计聚合后台清空并重建完成，删除 {} 行，重建 {} 个小时桶和 {} 个日桶",
deleted_total, rebuild.hourly_buckets, rebuild.daily_buckets
)
};
return Ok((
json!({
"deleted": affected,
"rebuilt": {
"hourly_buckets": rebuild.hourly_buckets,
"daily_buckets": rebuild.daily_buckets,
"capped": rebuild.capped,
},
"total": deleted_total,
}),
message,
));
}
let message = match kind {
AdminCleanupTaskKind::Config => format!("系统配置后台清空完成，影响 {deleted_total} 行"),
AdminCleanupTaskKind::Users => format!("非管理员用户后台清空完成，影响 {deleted_total} 行"),
AdminCleanupTaskKind::Usage => format!("使用记录后台清空完成，影响 {deleted_total} 行"),
AdminCleanupTaskKind::AuditLogs => format!("审计日志后台清空完成，影响 {deleted_total} 行"),
AdminCleanupTaskKind::RequestBodies | AdminCleanupTaskKind::Stats => unreachable!(),
};
Ok((
json!({
"deleted": affected,
"total": deleted_total,
}),
message,
))
}
fn initial_task_summary(kind: AdminCleanupTaskKind, batch_size: Option<usize>) -> Value {
match kind {
AdminCleanupTaskKind::RequestBodies => json!({
"batch_size": batch_size.unwrap_or(1),
"batches": 0,
"cleaned": {},
}),
AdminCleanupTaskKind::Stats => json!({
"deleted": {},
"rebuilt": {},
}),
AdminCleanupTaskKind::Config
| AdminCleanupTaskKind::Users
| AdminCleanupTaskKind::Usage
| AdminCleanupTaskKind::AuditLogs => json!({
"deleted": {},
}),
}
}
fn cleanup_task_record(
initial: &AdminCleanupRunRecord,
status: &str,
message: String,
summary: Value,
started_at: Option<Instant>,
error: Option<String>,
) -> AdminCleanupRunRecord {
let completed = matches!(status, "completed" | "failed");
AdminCleanupRunRecord {
id: initial.id.clone(),
kind: initial.kind.clone(),
trigger: initial.trigger.clone(),
status: status.to_string(),
message,
started_at_unix_secs: initial.started_at_unix_secs,
completed_at_unix_secs: completed.then(now_unix_secs),
duration_ms: started_at
.map(|value| value.elapsed().as_millis().try_into().unwrap_or(u64::MAX)),
summary,
error,
}
}
fn request_body_cleanup_record(
initial: &AdminCleanupRunRecord,
status: &str,
message: String,
batches: usize,
batch_size: usize,
total: &AdminSystemPurgeSummary,
started_at: Option<Instant>,
error: Option<String>,
) -> AdminCleanupRunRecord {
let completed = matches!(status, "completed" | "failed");
AdminCleanupRunRecord {
id: initial.id.clone(),
kind: initial.kind.clone(),
trigger: initial.trigger.clone(),
status: status.to_string(),
message,
started_at_unix_secs: initial.started_at_unix_secs,
completed_at_unix_secs: completed.then(now_unix_secs),
duration_ms: started_at
.map(|value| value.elapsed().as_millis().try_into().unwrap_or(u64::MAX)),
summary: json!({
"batch_size": batch_size,
"batches": batches,
"cleaned": total.affected,
"total": total.total(),
}),
error,
}
}
async fn record_cleanup_run(
data: &GatewayDataState,
record: AdminCleanupRunRecord,
) -> Result<(), DataLayerError> {
let mut records = list_admin_cleanup_run_records(data).await?;
records.retain(|existing| existing.id != record.id);
records.insert(0, record);
records.truncate(CLEANUP_RUN_HISTORY_LIMIT);
let value = serde_json::to_value(records).map_err(|err| {
DataLayerError::UnexpectedValue(format!("invalid cleanup run history: {err}"))
})?;
data.upsert_system_config_entry(
CLEANUP_RUN_HISTORY_KEY,
&value,
Some("最近的系统清理执行记录"),
)
.await?;
Ok(())
}
fn parse_cleanup_run_records(value: Value) -> Vec<AdminCleanupRunRecord> {
value
.as_array()
.into_iter()
.flat_map(|items| items.iter())
.filter_map(|item| serde_json::from_value::<AdminCleanupRunRecord>(item.clone()).ok())
.collect()
}

35
apps/aether-gateway/src/maintenance/runtime/request_candidate_cleanup.rs
View File

@@ -1,9 +1,14 @@
use aether_data_contracts::DataLayerError;
use serde_json::json;
use std::time::Instant;
use tracing::info;
use crate::data::GatewayDataState;
use super::{now_unix_secs, system_config_bool, system_config_u64, system_config_usize};
use super::{
now_unix_secs, record_completed_cleanup_run, record_failed_cleanup_run, system_config_bool,
system_config_u64, system_config_usize,
};
pub(crate) async fn cleanup_request_candidates_once(
data: &GatewayDataState,
@@ -47,7 +52,33 @@ pub(crate) async fn cleanup_request_candidates_once(
pub(super) async fn run_request_candidate_cleanup_once(
data: &GatewayDataState,
) -> Result<(), DataLayerError> {
let deleted = cleanup_request_candidates_once(data).await?;
let started_at_unix_secs = now_unix_secs();
let started_at = Instant::now();
let deleted = match cleanup_request_candidates_once(data).await {
Ok(deleted) => deleted,
Err(err) => {
record_failed_cleanup_run(
data,
"request_candidate_cleanup",
"auto",
started_at_unix_secs,
started_at,
&err,
)
.await;
return Err(err);
}
};
record_completed_cleanup_run(
data,
"request_candidate_cleanup",
"auto",
started_at_unix_secs,
started_at,
json!({ "request_candidates_deleted": deleted }),
format!("候选记录自动清理完成，删除 {deleted} 行"),
)
.await;
if deleted > 0 {
info!(deleted, "gateway deleted expired request candidates");
}

80
apps/aether-gateway/src/maintenance/runtime/runners.rs
View File

@@ -1,4 +1,6 @@
use aether_data_contracts::DataLayerError;
use serde_json::json;
use std::time::Instant;
use tracing::{info, warn};
use crate::data::GatewayDataState;
@@ -8,15 +10,41 @@ use super::{
advance_proxy_upgrade_rollout_once, cleanup_audit_logs_once,
cleanup_expired_gemini_file_mappings_once, cleanup_proxy_node_metrics_once,
cleanup_request_candidates_once, cleanup_stale_pending_requests_once,
cleanup_stale_proxy_nodes_once, collect_proxy_upgrade_rollout_probes,
cleanup_stale_proxy_nodes_once, collect_proxy_upgrade_rollout_probes, now_unix_secs,
perform_db_maintenance_once, perform_provider_checkin_once, perform_stats_aggregation_once,
perform_stats_hourly_aggregation_once, perform_usage_cleanup_once,
perform_wallet_daily_usage_aggregation_once, record_proxy_upgrade_traffic_success,
summarize_database_pool,
perform_wallet_daily_usage_aggregation_once, record_completed_cleanup_run,
record_failed_cleanup_run, record_proxy_upgrade_traffic_success, summarize_database_pool,
};
pub(super) async fn run_audit_cleanup_once(data: &GatewayDataState) -> Result<(), DataLayerError> {
let deleted = cleanup_audit_logs_once(data).await?;
let started_at_unix_secs = now_unix_secs();
let started_at = Instant::now();
let deleted = match cleanup_audit_logs_once(data).await {
Ok(deleted) => deleted,
Err(err) => {
record_failed_cleanup_run(
data,
"audit_cleanup",
"auto",
started_at_unix_secs,
started_at,
&err,
)
.await;
return Err(err);
}
};
record_completed_cleanup_run(
data,
"audit_cleanup",
"auto",
started_at_unix_secs,
started_at,
json!({ "audit_logs_deleted": deleted }),
format!("审计日志自动清理完成，删除 {deleted} 行"),
)
.await;
if deleted > 0 {
info!(
event_name = "audit_cleanup_completed",
@@ -204,7 +232,49 @@ pub(super) async fn run_stats_aggregation_once(
}
pub(super) async fn run_usage_cleanup_once(data: &GatewayDataState) -> Result<(), DataLayerError> {
let summary = perform_usage_cleanup_once(data).await?;
let started_at_unix_secs = now_unix_secs();
let started_at = Instant::now();
let summary = match perform_usage_cleanup_once(data).await {
Ok(summary) => summary,
Err(err) => {
record_failed_cleanup_run(
data,
"usage_cleanup",
"auto",
started_at_unix_secs,
started_at,
&err,
)
.await;
return Err(err);
}
};
record_completed_cleanup_run(
data,
"usage_cleanup",
"auto",
started_at_unix_secs,
started_at,
json!({
"body_externalized": summary.body_externalized,
"legacy_body_refs_migrated": summary.legacy_body_refs_migrated,
"body_cleaned": summary.body_cleaned,
"header_cleaned": summary.header_cleaned,
"keys_cleaned": summary.keys_cleaned,
"records_deleted": summary.records_deleted,
}),
format!(
"请求记录自动清理完成，影响 {} 项",
summary
.body_externalized
.saturating_add(summary.legacy_body_refs_migrated)
.saturating_add(summary.body_cleaned)
.saturating_add(summary.header_cleaned)
.saturating_add(summary.keys_cleaned)
.saturating_add(summary.records_deleted)
),
)
.await;
if summary.body_externalized > 0
|| summary.legacy_body_refs_migrated > 0
|| summary.body_cleaned > 0

21
crates/aether-data/src/backend/maintenance.rs
View File

@@ -194,6 +194,16 @@ impl DataBackends {
None => Ok(AdminSystemPurgeSummary::default()),
}
}
pub async fn purge_admin_request_bodies_batch(
&self,
batch_size: usize,
) -> Result<AdminSystemPurgeSummary, DataLayerError> {
match self.sql_backend() {
Some(backend) => backend.purge_admin_request_bodies_batch(batch_size).await,
None => Ok(AdminSystemPurgeSummary::default()),
}
}
}
impl PostgresBackend {
@@ -494,4 +504,15 @@ impl<'a> SqlBackendRef<'a> {
Self::Sqlite(sqlite) => sqlite.purge_admin_system_data(target).await,
}
}
async fn purge_admin_request_bodies_batch(
self,
batch_size: usize,
) -> Result<AdminSystemPurgeSummary, DataLayerError> {
match self {
Self::Postgres(postgres) => postgres.purge_admin_request_bodies_batch(batch_size).await,
Self::Mysql(mysql) => mysql.purge_admin_request_bodies_batch(batch_size).await,
Self::Sqlite(sqlite) => sqlite.purge_admin_request_bodies_batch(batch_size).await,
}
}
}

93
crates/aether-data/src/backend/sqlite.rs
View File

@@ -434,6 +434,99 @@ VALUES
assert_eq!(admin_exists, 1);
}
#[tokio::test]
async fn admin_system_request_bodies_purge_clears_inline_usage_body_fields() {
let config = SqlDatabaseConfig {
driver: DatabaseDriver::Sqlite,
url: "sqlite::memory:".to_string(),
pool: SqlPoolConfig {
max_connections: 1,
..SqlPoolConfig::default()
},
};
let backend = SqliteBackend::from_config(config).expect("backend should build");
run_sqlite_migrations(backend.pool())
.await
.expect("sqlite migrations should run");
for (column, ty) in [
("request_body", "TEXT"),
("response_body", "TEXT"),
("provider_request_body", "TEXT"),
("client_response_body", "TEXT"),
("request_body_compressed", "BLOB"),
("response_body_compressed", "BLOB"),
("provider_request_body_compressed", "BLOB"),
("client_response_body_compressed", "BLOB"),
] {
sqlx::query(&format!(r#"ALTER TABLE "usage" ADD COLUMN {column} {ty}"#))
.execute(backend.pool())
.await
.expect("legacy body column should be added");
}
sqlx::query(
r#"
INSERT INTO "usage" (
request_id,
provider_name,
model,
request_body,
response_body,
provider_request_body,
client_response_body,
request_body_compressed,
response_body_compressed,
provider_request_body_compressed,
client_response_body_compressed,
created_at_unix_ms
)
VALUES (
'request-1',
'openai',
'gpt-4.1',
'client request',
'provider response',
'provider request',
'client response',
X'01',
X'02',
X'03',
X'04',
1
)
"#,
)
.execute(backend.pool())
.await
.expect("usage row should insert");
let summary = backend
.purge_admin_system_data(AdminSystemPurgeTarget::RequestBodies)
.await
.expect("request body purge should run");
assert_eq!(summary.affected.get("usage_body_fields_cleaned"), Some(&1));
let remaining: i64 = sqlx::query_scalar(
r#"
SELECT COUNT(*)
FROM "usage"
WHERE request_body IS NOT NULL
OR response_body IS NOT NULL
OR provider_request_body IS NOT NULL
OR client_response_body IS NOT NULL
OR request_body_compressed IS NOT NULL
OR response_body_compressed IS NOT NULL
OR provider_request_body_compressed IS NOT NULL
OR client_response_body_compressed IS NOT NULL
"#,
)
.fetch_one(backend.pool())
.await
.expect("remaining body count should load");
assert_eq!(remaining, 0);
}
async fn sqlite_count(pool: &sqlx::SqlitePool, table: &str) -> i64 {
let sql = format!("SELECT COUNT(*) FROM \"{table}\"");
sqlx::query_scalar::<_, i64>(&sql)

684
crates/aether-data/src/backend/system.rs
View File

@@ -90,6 +90,20 @@ impl PostgresBackend {
Ok(summary)
}
pub async fn purge_admin_request_bodies_batch(
&self,
batch_size: usize,
) -> Result<AdminSystemPurgeSummary, DataLayerError> {
if batch_size == 0 {
return Ok(AdminSystemPurgeSummary::default());
}
let mut tx = self.pool().begin().await.map_postgres_err()?;
let mut summary = AdminSystemPurgeSummary::default();
purge_postgres_request_bodies_batch(&mut tx, batch_size, &mut summary).await?;
tx.commit().await.map_postgres_err()?;
Ok(summary)
}
pub async fn find_system_config_value(
&self,
key: &str,
@@ -195,6 +209,20 @@ impl MysqlBackend {
Ok(summary)
}
pub async fn purge_admin_request_bodies_batch(
&self,
batch_size: usize,
) -> Result<AdminSystemPurgeSummary, DataLayerError> {
if batch_size == 0 {
return Ok(AdminSystemPurgeSummary::default());
}
let mut tx = self.pool().begin().await.map_sql_err()?;
let mut summary = AdminSystemPurgeSummary::default();
purge_mysql_request_bodies_batch(&mut tx, batch_size, &mut summary).await?;
tx.commit().await.map_sql_err()?;
Ok(summary)
}
pub async fn find_system_config_value(
&self,
key: &str,
@@ -335,6 +363,20 @@ impl SqliteBackend {
Ok(summary)
}
pub async fn purge_admin_request_bodies_batch(
&self,
batch_size: usize,
) -> Result<AdminSystemPurgeSummary, DataLayerError> {
if batch_size == 0 {
return Ok(AdminSystemPurgeSummary::default());
}
let mut tx = self.pool().begin().await.map_sql_err()?;
let mut summary = AdminSystemPurgeSummary::default();
purge_sqlite_request_bodies_batch(&mut tx, batch_size, &mut summary).await?;
tx.commit().await.map_sql_err()?;
Ok(summary)
}
pub async fn find_system_config_value(
&self,
key: &str,
@@ -519,6 +561,17 @@ const ADMIN_USAGE_CHILD_TABLES: &[&str] = &[
"usage_settlement_snapshots",
];
const USAGE_BODY_FIELD_COLUMNS: &[&str] = &[
"request_body",
"response_body",
"provider_request_body",
"client_response_body",
"request_body_compressed",
"response_body_compressed",
"provider_request_body_compressed",
"client_response_body_compressed",
];
const ADMIN_USER_SCOPED_TABLES: &[&str] = &[
"stats_user_daily_cost_savings_model_provider",
"stats_user_daily_cost_savings_model",
@@ -678,9 +731,10 @@ WHERE request_count <> 0
}
AdminSystemPurgeTarget::RequestBodies => {
pg_delete_table(tx, "usage_body_blobs", summary).await?;
pg_execute_if_table(
pg_execute_if_table_has_columns(
tx,
"usage",
USAGE_BODY_FIELD_COLUMNS,
"usage_body_fields_cleaned",
r#"
UPDATE public.usage
@@ -1038,6 +1092,33 @@ WHERE request_count <> 0
}
AdminSystemPurgeTarget::RequestBodies => {
mysql_delete_table(tx, "usage_body_blobs", summary).await?;
mysql_execute_if_table_has_columns(
tx,
"usage",
USAGE_BODY_FIELD_COLUMNS,
"usage_body_fields_cleaned",
r#"
UPDATE `usage`
SET request_body = NULL,
response_body = NULL,
provider_request_body = NULL,
client_response_body = NULL,
request_body_compressed = NULL,
response_body_compressed = NULL,
provider_request_body_compressed = NULL,
client_response_body_compressed = NULL
WHERE request_body IS NOT NULL
OR response_body IS NOT NULL
OR provider_request_body IS NOT NULL
OR client_response_body IS NOT NULL
OR request_body_compressed IS NOT NULL
OR response_body_compressed IS NOT NULL
OR provider_request_body_compressed IS NOT NULL
OR client_response_body_compressed IS NOT NULL
"#,
summary,
)
.await?;
mysql_execute_if_table(
tx,
"usage_http_audits",
@@ -1333,6 +1414,33 @@ WHERE request_count <> 0
}
AdminSystemPurgeTarget::RequestBodies => {
sqlite_delete_table(tx, "usage_body_blobs", summary).await?;
sqlite_execute_if_table_has_columns(
tx,
"usage",
USAGE_BODY_FIELD_COLUMNS,
"usage_body_fields_cleaned",
r#"
UPDATE "usage"
SET request_body = NULL,
response_body = NULL,
provider_request_body = NULL,
client_response_body = NULL,
request_body_compressed = NULL,
response_body_compressed = NULL,
provider_request_body_compressed = NULL,
client_response_body_compressed = NULL
WHERE request_body IS NOT NULL
OR response_body IS NOT NULL
OR provider_request_body IS NOT NULL
OR client_response_body IS NOT NULL
OR request_body_compressed IS NOT NULL
OR response_body_compressed IS NOT NULL
OR provider_request_body_compressed IS NOT NULL
OR client_response_body_compressed IS NOT NULL
"#,
summary,
)
.await?;
sqlite_execute_if_table(
tx,
"usage_http_audits",
@@ -1506,6 +1614,304 @@ WHERE user_id IN ({non_admin_users})
Ok(())
}
async fn purge_postgres_request_bodies_batch(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
batch_size: usize,
summary: &mut AdminSystemPurgeSummary,
) -> Result<(), DataLayerError> {
let limit = i64::try_from(batch_size).unwrap_or(i64::MAX);
pg_execute_batch_if_table(
tx,
"usage_body_blobs",
"usage_body_blobs",
r#"
WITH doomed AS (
SELECT body_ref
FROM public.usage_body_blobs
ORDER BY body_ref ASC
LIMIT $1
)
DELETE FROM public.usage_body_blobs AS blobs
USING doomed
WHERE blobs.body_ref = doomed.body_ref
"#,
summary,
limit,
)
.await?;
pg_execute_batch_if_table_has_columns(
tx,
"usage",
USAGE_BODY_FIELD_COLUMNS,
"usage_body_fields_cleaned",
r#"
WITH batch AS (
SELECT request_id
FROM public.usage
WHERE request_body IS NOT NULL
OR response_body IS NOT NULL
OR provider_request_body IS NOT NULL
OR client_response_body IS NOT NULL
OR request_body_compressed IS NOT NULL
OR response_body_compressed IS NOT NULL
OR provider_request_body_compressed IS NOT NULL
OR client_response_body_compressed IS NOT NULL
ORDER BY created_at_unix_ms ASC, request_id ASC
LIMIT $1
)
UPDATE public.usage AS usage_rows
SET request_body = NULL,
response_body = NULL,
provider_request_body = NULL,
client_response_body = NULL,
request_body_compressed = NULL,
response_body_compressed = NULL,
provider_request_body_compressed = NULL,
client_response_body_compressed = NULL
FROM batch
WHERE usage_rows.request_id = batch.request_id
"#,
summary,
limit,
)
.await?;
pg_execute_batch_if_table(
tx,
"usage_http_audits",
"usage_http_audit_body_refs_cleaned",
r#"
WITH batch AS (
SELECT request_id
FROM public.usage_http_audits
WHERE request_body_ref IS NOT NULL
OR provider_request_body_ref IS NOT NULL
OR response_body_ref IS NOT NULL
OR client_response_body_ref IS NOT NULL
OR request_body_state IS NOT NULL
OR provider_request_body_state IS NOT NULL
OR response_body_state IS NOT NULL
OR client_response_body_state IS NOT NULL
OR body_capture_mode <> 'none'
ORDER BY request_id ASC
LIMIT $1
)
UPDATE public.usage_http_audits AS audits
SET request_body_ref = NULL,
provider_request_body_ref = NULL,
response_body_ref = NULL,
client_response_body_ref = NULL,
request_body_state = NULL,
provider_request_body_state = NULL,
response_body_state = NULL,
client_response_body_state = NULL,
body_capture_mode = 'none',
updated_at = NOW()
FROM batch
WHERE audits.request_id = batch.request_id
"#,
summary,
limit,
)
.await?;
Ok(())
}
async fn purge_mysql_request_bodies_batch(
tx: &mut sqlx::Transaction<'_, sqlx::MySql>,
batch_size: usize,
summary: &mut AdminSystemPurgeSummary,
) -> Result<(), DataLayerError> {
let limit = i64::try_from(batch_size).unwrap_or(i64::MAX);
mysql_execute_batch_if_table(
tx,
"usage_body_blobs",
"usage_body_blobs",
r#"
DELETE FROM usage_body_blobs
WHERE body_ref IN (
SELECT body_ref FROM (
SELECT body_ref
FROM usage_body_blobs
ORDER BY body_ref ASC
LIMIT ?
) AS doomed
)
"#,
summary,
limit,
)
.await?;
mysql_execute_batch_if_table_has_columns(
tx,
"usage",
USAGE_BODY_FIELD_COLUMNS,
"usage_body_fields_cleaned",
r#"
UPDATE `usage`
SET request_body = NULL,
response_body = NULL,
provider_request_body = NULL,
client_response_body = NULL,
request_body_compressed = NULL,
response_body_compressed = NULL,
provider_request_body_compressed = NULL,
client_response_body_compressed = NULL
WHERE request_id IN (
SELECT request_id FROM (
SELECT request_id
FROM `usage`
WHERE request_body IS NOT NULL
OR response_body IS NOT NULL
OR provider_request_body IS NOT NULL
OR client_response_body IS NOT NULL
OR request_body_compressed IS NOT NULL
OR response_body_compressed IS NOT NULL
OR provider_request_body_compressed IS NOT NULL
OR client_response_body_compressed IS NOT NULL
ORDER BY created_at_unix_ms ASC, request_id ASC
LIMIT ?
) AS batch
)
"#,
summary,
limit,
)
.await?;
mysql_execute_batch_if_table(
tx,
"usage_http_audits",
"usage_http_audit_body_refs_cleaned",
r#"
UPDATE usage_http_audits
SET request_body_ref = NULL,
provider_request_body_ref = NULL,
response_body_ref = NULL,
client_response_body_ref = NULL,
request_body_state = NULL,
provider_request_body_state = NULL,
response_body_state = NULL,
client_response_body_state = NULL,
body_capture_mode = 'none'
WHERE request_id IN (
SELECT request_id FROM (
SELECT request_id
FROM usage_http_audits
WHERE request_body_ref IS NOT NULL
OR provider_request_body_ref IS NOT NULL
OR response_body_ref IS NOT NULL
OR client_response_body_ref IS NOT NULL
OR request_body_state IS NOT NULL
OR provider_request_body_state IS NOT NULL
OR response_body_state IS NOT NULL
OR client_response_body_state IS NOT NULL
OR body_capture_mode <> 'none'
ORDER BY request_id ASC
LIMIT ?
) AS batch
)
"#,
summary,
limit,
)
.await?;
Ok(())
}
async fn purge_sqlite_request_bodies_batch(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
batch_size: usize,
summary: &mut AdminSystemPurgeSummary,
) -> Result<(), DataLayerError> {
let limit = i64::try_from(batch_size).unwrap_or(i64::MAX);
sqlite_execute_batch_if_table(
tx,
"usage_body_blobs",
"usage_body_blobs",
r#"
DELETE FROM usage_body_blobs
WHERE body_ref IN (
SELECT body_ref
FROM usage_body_blobs
ORDER BY body_ref ASC
LIMIT ?
)
"#,
summary,
limit,
)
.await?;
sqlite_execute_batch_if_table_has_columns(
tx,
"usage",
USAGE_BODY_FIELD_COLUMNS,
"usage_body_fields_cleaned",
r#"
UPDATE "usage"
SET request_body = NULL,
response_body = NULL,
provider_request_body = NULL,
client_response_body = NULL,
request_body_compressed = NULL,
response_body_compressed = NULL,
provider_request_body_compressed = NULL,
client_response_body_compressed = NULL
WHERE request_id IN (
SELECT request_id
FROM "usage"
WHERE request_body IS NOT NULL
OR response_body IS NOT NULL
OR provider_request_body IS NOT NULL
OR client_response_body IS NOT NULL
OR request_body_compressed IS NOT NULL
OR response_body_compressed IS NOT NULL
OR provider_request_body_compressed IS NOT NULL
OR client_response_body_compressed IS NOT NULL
ORDER BY created_at_unix_ms ASC, request_id ASC
LIMIT ?
)
"#,
summary,
limit,
)
.await?;
sqlite_execute_batch_if_table(
tx,
"usage_http_audits",
"usage_http_audit_body_refs_cleaned",
r#"
UPDATE usage_http_audits
SET request_body_ref = NULL,
provider_request_body_ref = NULL,
response_body_ref = NULL,
client_response_body_ref = NULL,
request_body_state = NULL,
provider_request_body_state = NULL,
response_body_state = NULL,
client_response_body_state = NULL,
body_capture_mode = 'none'
WHERE request_id IN (
SELECT request_id
FROM usage_http_audits
WHERE request_body_ref IS NOT NULL
OR provider_request_body_ref IS NOT NULL
OR response_body_ref IS NOT NULL
OR client_response_body_ref IS NOT NULL
OR request_body_state IS NOT NULL
OR provider_request_body_state IS NOT NULL
OR response_body_state IS NOT NULL
OR client_response_body_state IS NOT NULL
OR body_capture_mode <> 'none'
ORDER BY request_id ASC
LIMIT ?
)
"#,
summary,
limit,
)
.await?;
Ok(())
}
async fn pg_delete_table(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
table: &str,
@@ -1586,6 +1992,69 @@ async fn pg_execute_if_table(
Ok(())
}
async fn pg_execute_if_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
table: &str,
columns: &[&str],
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
) -> Result<(), DataLayerError> {
if !pg_table_has_columns(tx, checked_sql_identifier(table)?, columns).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.execute(&mut **tx)
.await
.map_postgres_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn pg_execute_batch_if_table(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
table: &str,
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
limit: i64,
) -> Result<(), DataLayerError> {
if !pg_table_exists(tx, checked_sql_identifier(table)?).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.bind(limit)
.execute(&mut **tx)
.await
.map_postgres_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn pg_execute_batch_if_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
table: &str,
columns: &[&str],
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
limit: i64,
) -> Result<(), DataLayerError> {
if !pg_table_has_columns(tx, checked_sql_identifier(table)?, columns).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.bind(limit)
.execute(&mut **tx)
.await
.map_postgres_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn pg_table_exists(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
table: &str,
@@ -1598,6 +2067,38 @@ async fn pg_table_exists(
.map_postgres_err()
}
async fn pg_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
table: &str,
columns: &[&str],
) -> Result<bool, DataLayerError> {
let table = checked_sql_identifier(table)?;
if !pg_table_exists(tx, table).await? {
return Ok(false);
}
for column in columns {
let column = checked_sql_identifier(column)?;
let exists: bool = sqlx::query_scalar(
"SELECT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = $1
AND column_name = $2
)",
)
.bind(table)
.bind(column)
.fetch_one(&mut **tx)
.await
.map_postgres_err()?;
if !exists {
return Ok(false);
}
}
Ok(true)
}
async fn mysql_delete_table(
tx: &mut sqlx::Transaction<'_, sqlx::MySql>,
table: &str,
@@ -1678,6 +2179,69 @@ async fn mysql_execute_if_table(
Ok(())
}
async fn mysql_execute_if_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::MySql>,
table: &str,
columns: &[&str],
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
) -> Result<(), DataLayerError> {
if !mysql_table_has_columns(tx, checked_sql_identifier(table)?, columns).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.execute(&mut **tx)
.await
.map_sql_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn mysql_execute_batch_if_table(
tx: &mut sqlx::Transaction<'_, sqlx::MySql>,
table: &str,
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
limit: i64,
) -> Result<(), DataLayerError> {
if !mysql_table_exists(tx, checked_sql_identifier(table)?).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.bind(limit)
.execute(&mut **tx)
.await
.map_sql_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn mysql_execute_batch_if_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::MySql>,
table: &str,
columns: &[&str],
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
limit: i64,
) -> Result<(), DataLayerError> {
if !mysql_table_has_columns(tx, checked_sql_identifier(table)?, columns).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.bind(limit)
.execute(&mut **tx)
.await
.map_sql_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn mysql_table_exists(
tx: &mut sqlx::Transaction<'_, sqlx::MySql>,
table: &str,
@@ -1693,6 +2257,36 @@ async fn mysql_table_exists(
Ok(total > 0)
}
async fn mysql_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::MySql>,
table: &str,
columns: &[&str],
) -> Result<bool, DataLayerError> {
let table = checked_sql_identifier(table)?;
if !mysql_table_exists(tx, table).await? {
return Ok(false);
}
for column in columns {
let column = checked_sql_identifier(column)?;
let total: i64 = sqlx::query_scalar(
"SELECT COUNT(*)
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND table_name = ?
AND column_name = ?",
)
.bind(table)
.bind(column)
.fetch_one(&mut **tx)
.await
.map_sql_err()?;
if total == 0 {
return Ok(false);
}
}
Ok(true)
}
async fn sqlite_delete_table(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table: &str,
@@ -1773,6 +2367,69 @@ async fn sqlite_execute_if_table(
Ok(())
}
async fn sqlite_execute_if_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table: &str,
columns: &[&str],
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
) -> Result<(), DataLayerError> {
if !sqlite_table_has_columns(tx, checked_sql_identifier(table)?, columns).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.execute(&mut **tx)
.await
.map_sql_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn sqlite_execute_batch_if_table(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table: &str,
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
limit: i64,
) -> Result<(), DataLayerError> {
if !sqlite_table_exists(tx, checked_sql_identifier(table)?).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.bind(limit)
.execute(&mut **tx)
.await
.map_sql_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn sqlite_execute_batch_if_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table: &str,
columns: &[&str],
key: &str,
sql: &str,
summary: &mut AdminSystemPurgeSummary,
limit: i64,
) -> Result<(), DataLayerError> {
if !sqlite_table_has_columns(tx, checked_sql_identifier(table)?, columns).await? {
return Ok(());
}
let rows = sqlx::query(sql)
.bind(limit)
.execute(&mut **tx)
.await
.map_sql_err()?
.rows_affected();
summary.add(key, rows);
Ok(())
}
async fn sqlite_table_exists(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table: &str,
@@ -1787,6 +2444,31 @@ async fn sqlite_table_exists(
Ok(total > 0)
}
async fn sqlite_table_has_columns(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table: &str,
columns: &[&str],
) -> Result<bool, DataLayerError> {
let table = checked_sql_identifier(table)?;
if !sqlite_table_exists(tx, table).await? {
return Ok(false);
}
for column in columns {
let column = checked_sql_identifier(column)?;
let total: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info(?) WHERE name = ?")
.bind(table)
.bind(column)
.fetch_one(&mut **tx)
.await
.map_sql_err()?;
if total == 0 {
return Ok(false);
}
}
Ok(true)
}
fn current_unix_secs() -> u64 {
chrono::Utc::now().timestamp().max(0) as u64
}

6
crates/aether-data/src/repository/system.rs
View File

@@ -41,6 +41,12 @@ impl AdminSystemPurgeSummary {
*self.affected.entry(key.into()).or_insert(0) += count;
}
pub fn merge(&mut self, other: &Self) {
for (key, count) in &other.affected {
self.add(key.clone(), *count);
}
}
pub fn total(&self) -> u64 {
self.affected.values().copied().sum()
}

51
frontend/src/api/admin.ts
View File

@@ -236,6 +236,28 @@ export interface EmailTemplateResetResponse {
}
}
export interface CleanupRunRecord {
id: string
kind: string
trigger: string
status: 'processing' | 'completed' | 'failed'
message: string
started_at_unix_secs: number
completed_at_unix_secs: number | null
duration_ms: number | null
summary: Record<string, unknown>
error: string | null
}
export interface CleanupRunListResponse {
items: CleanupRunRecord[]
}
export interface CleanupTaskResponse {
message: string
task: CleanupRunRecord
}
// 检查更新响应
export interface CheckUpdateResponse {
current_version: string
@@ -1051,12 +1073,29 @@ export const adminApi = {
},
// 数据清空
purgeConfig: () => purge<{ message: string; deleted: Record<string, number> }>('config'),
purgeUsers: () => purge<{ message: string; deleted: Record<string, number> }>('users'),
purgeUsage: () => purge<{ message: string; deleted: Record<string, number> }>('usage'),
purgeAuditLogs: () => purge<{ message: string; deleted: Record<string, number> }>('audit-logs'),
purgeRequestBodies: () => purge<{ message: string; cleaned: Record<string, number> }>('request-bodies'),
purgeStats: () => purge<{ message: string }>('stats'),
purgeConfig: () => purge<CleanupTaskResponse>('config'),
purgeUsers: () => purge<CleanupTaskResponse>('users'),
async purgeUsage(): Promise<CleanupTaskResponse> {
const response = await apiClient.post<CleanupTaskResponse>('/api/admin/system/purge/usage')
return response.data
},
async purgeAuditLogs(): Promise<CleanupTaskResponse> {
const response = await apiClient.post<CleanupTaskResponse>('/api/admin/system/purge/audit-logs')
return response.data
},
purgeRequestBodies: () => purge<CleanupTaskResponse>('request-bodies'),
async purgeRequestBodiesAsync(): Promise<CleanupTaskResponse> {
const response = await apiClient.post<CleanupTaskResponse>('/api/admin/system/purge/request-bodies/task')
return response.data
},
async purgeStats(): Promise<CleanupTaskResponse> {
const response = await apiClient.post<CleanupTaskResponse>('/api/admin/system/purge/stats')
return response.data
},
async getCleanupRuns(): Promise<CleanupRunListResponse> {
const response = await apiClient.get<CleanupRunListResponse>('/api/admin/system/cleanup/runs')
return response.data
},
async getTimeSeries(params?: {
start_date?: string

185
frontend/src/views/admin/system-settings/CleanupPolicySection.vue
View File

@@ -275,10 +275,104 @@
<p>7. <strong>代理指标</strong>: 仅保留 1m/1h 聚合桶，清理任务按批次删除过期桶</p>
</div>
</div>
<div class="mt-4 border border-border rounded-lg overflow-hidden">
<div class="flex items-center justify-between px-4 py-3 border-b border-border">
<div>
<h4 class="text-sm font-medium">
最近清理记录
</h4>
<p class="text-xs text-muted-foreground">
自动清理、手动系统清理和请求体后台任务的执行结果
</p>
</div>
<Button
variant="outline"
size="sm"
:disabled="cleanupRunsLoading"
@click="loadCleanupRuns"
>
<RefreshCw
class="w-3.5 h-3.5 mr-1.5"
:class="{ 'animate-spin': cleanupRunsLoading }"
/>
刷新
</Button>
</div>
<div
v-if="cleanupRuns.length === 0 && !cleanupRunsLoading"
class="px-4 py-6 text-sm text-muted-foreground"
>
暂无清理记录
</div>
<div
v-else
class="overflow-x-auto"
>
<table class="w-full text-sm">
<thead class="bg-muted/30 text-xs text-muted-foreground">
<tr>
<th class="px-4 py-2 text-left font-medium">
时间
</th>
<th class="px-4 py-2 text-left font-medium">
类型
</th>
<th class="px-4 py-2 text-left font-medium">
来源
</th>
<th class="px-4 py-2 text-left font-medium">
状态
</th>
<th class="px-4 py-2 text-left font-medium">
结果
</th>
<th class="px-4 py-2 text-right font-medium">
耗时
</th>
</tr>
</thead>
<tbody>
<tr
v-for="run in cleanupRuns"
:key="run.id"
class="border-t border-border"
>
<td class="px-4 py-2 whitespace-nowrap">
{{ formatRunTime(run.started_at_unix_secs) }}
</td>
<td class="px-4 py-2 whitespace-nowrap">
{{ cleanupKindLabel(run.kind) }}
</td>
<td class="px-4 py-2 whitespace-nowrap text-muted-foreground">
{{ run.trigger === 'manual' ? '手动' : '自动' }}
</td>
<td class="px-4 py-2 whitespace-nowrap">
<span :class="cleanupStatusClass(run.status)">
{{ cleanupStatusLabel(run.status) }}
</span>
</td>
<td class="px-4 py-2 min-w-[18rem]">
<div>{{ run.error || run.message }}</div>
<div class="text-xs text-muted-foreground">
{{ cleanupSummaryText(run.summary) }}
</div>
</td>
<td class="px-4 py-2 text-right whitespace-nowrap text-muted-foreground">
{{ formatDuration(run.duration_ms) }}
</td>
</tr>
</tbody>
</table>
</div>
</div>
</CardSection>
</template>
<script setup lang="ts">
import { onBeforeUnmount, onMounted, ref } from 'vue'
import { RefreshCw } from 'lucide-vue-next'
import { adminApi, type CleanupRunRecord } from '@/api/admin'
import Button from '@/components/ui/button.vue'
import Input from '@/components/ui/input.vue'
import Label from '@/components/ui/label.vue'
@@ -317,4 +411,95 @@ defineEmits<{
'update:proxyNodeMetrics1hRetentionDays': [value: number]
'update:proxyNodeMetricsCleanupBatchSize': [value: number]
}>()
const cleanupRuns = ref<CleanupRunRecord[]>([])
const cleanupRunsLoading = ref(false)
let cleanupRunsTimer: ReturnType<typeof window.setInterval> | null = null
async function loadCleanupRuns() {
cleanupRunsLoading.value = true
try {
const response = await adminApi.getCleanupRuns()
cleanupRuns.value = response.items.slice(0, 10)
} finally {
cleanupRunsLoading.value = false
}
}
function cleanupKindLabel(kind: string): string {
const labels: Record<string, string> = {
usage_cleanup: '请求记录',
audit_cleanup: '审计日志',
request_candidate_cleanup: '候选记录',
request_bodies: '请求体',
config_purge: '配置清空',
users_purge: '用户清空',
usage_purge: '使用记录清空',
audit_logs_purge: '审计日志清空',
stats_purge: '统计聚合清空',
system_cleanup: '系统清理',
}
return labels[kind] || kind
}
function cleanupStatusLabel(status: string): string {
if (status === 'processing') return '执行中'
if (status === 'failed') return '失败'
return '完成'
}
function cleanupStatusClass(status: string): string {
if (status === 'processing') return 'text-amber-500'
if (status === 'failed') return 'text-destructive'
return 'text-emerald-500'
}
function formatRunTime(value: number): string {
if (!value) return '-'
return new Date(value * 1000).toLocaleString()
}
function formatDuration(value: number | null): string {
if (value === null || value === undefined) return '-'
if (value < 1000) return `${value}ms`
return `${(value / 1000).toFixed(1)}s`
}
function cleanupSummaryText(summary: Record<string, unknown>): string {
const total = typeof summary.total === 'number' ? summary.total : null
if (total !== null) return `影响 ${total} 行`
const entries = Object.entries(summary)
.filter(([, value]) => typeof value === 'number' && value > 0)
.map(([key, value]) => `${summaryLabel(key)} ${value}`)
return entries.length > 0 ? entries.join(' / ') : '无数据变更'
}
function summaryLabel(key: string): string {
const labels: Record<string, string> = {
body_externalized: '压缩',
legacy_body_refs_migrated: '迁移',
body_cleaned: '清体',
header_cleaned: '清头',
keys_cleaned: 'Key',
records_deleted: '删记录',
audit_logs_deleted: '删日志',
request_candidates_deleted: '删候选',
}
return labels[key] || key
}
onMounted(() => {
void loadCleanupRuns()
cleanupRunsTimer = window.setInterval(() => {
void loadCleanupRuns()
}, 15_000)
})
onBeforeUnmount(() => {
if (cleanupRunsTimer) {
window.clearInterval(cleanupRunsTimer)
cleanupRunsTimer = null
}
})
</script>

26
frontend/src/views/admin/system-settings/DataManagementSection.vue
View File

@@ -62,55 +62,55 @@ const purgeItems: PurgeItem[] = [
{
key: 'config',
title: '清空配置',
description: '删除所有提供商、端点、API Key 和模型配置',
description: '后台删除所有提供商、端点、API Key 和模型配置',
buttonText: '清空配置',
icon: markRaw(Settings),
confirmMessage: '确定要清空所有提供商配置吗？这将删除所有提供商、端点、API Key 和模型配置，操作不可逆。',
confirmMessage: '确定要后台清空所有提供商配置吗？这将删除所有提供商、端点、API Key 和模型配置，操作不可逆。',
action: () => adminApi.purgeConfig(),
},
{
key: 'users',
title: '清空用户',
description: '删除所有非管理员用户及其 API Keys',
description: '后台删除所有非管理员用户及其 API Keys',
buttonText: '清空用户',
icon: markRaw(Users),
confirmMessage: '确定要清空所有非管理员用户吗？管理员账户将被保留，操作不可逆。',
confirmMessage: '确定要后台清空所有非管理员用户吗？管理员账户将被保留，操作不可逆。',
action: () => adminApi.purgeUsers(),
},
{
key: 'usage',
title: '清空使用记录',
description: '删除全部使用记录和请求候选记录',
description: '后台清空全部使用记录和请求候选记录',
buttonText: '清空记录',
icon: markRaw(BarChart3),
confirmMessage: '确定要清空全部使用记录吗？所有请求统计数据将被永久删除，操作不可逆。',
confirmMessage: '确定要后台清空全部使用记录吗？所有请求统计数据将被永久删除，操作不可逆。',
action: () => adminApi.purgeUsage(),
},
{
key: 'audit-logs',
title: '清空审计日志',
description: '删除全部审计日志记录',
description: '后台删除全部审计日志记录',
buttonText: '清空日志',
icon: markRaw(Shield),
confirmMessage: '确定要清空全部审计日志吗？所有安全事件记录将被永久删除，操作不可逆。',
confirmMessage: '确定要后台清空全部审计日志吗？所有安全事件记录将被永久删除，操作不可逆。',
action: () => adminApi.purgeAuditLogs(),
},
{
key: 'request-bodies',
title: '清空请求体',
description: '清空所有请求/响应体数据，保留统计信息',
description: '后台分批清空所有请求/响应体数据，保留统计信息',
buttonText: '清空请求体',
icon: markRaw(FileText),
confirmMessage: '确定要清空全部请求体吗？请求/响应内容将被清除，但 token 和成本等统计信息会保留，操作不可逆。',
action: () => adminApi.purgeRequestBodies(),
confirmMessage: '确定要后台清空全部请求体吗？请求/响应内容将被分批清除，但 token 和成本等统计信息会保留，操作不可逆。',
action: () => adminApi.purgeRequestBodiesAsync(),
},
{
key: 'stats',
title: '清空统计聚合',
description: '删除统计聚合数据，保留原始使用记录；统计可从原始使用记录重新构建',
description: '后台删除统计聚合数据并重建，保留原始使用记录',
buttonText: '清空统计聚合',
icon: markRaw(PieChart),
confirmMessage: '确定要清空全部统计聚合数据吗？原始使用记录会保留，仪表盘和累计统计可从原始记录重新构建。',
confirmMessage: '确定要后台清空全部统计聚合数据吗？原始使用记录会保留，仪表盘和累计统计会从原始记录重新构建。',
action: () => adminApi.purgeStats(),
},
]