perf(usage): heatmap 改为数据库端按天聚合查询

将 admin 和 user heatmap 从逐条加载 usage audit 记录后在应用层聚合,
改为通过 SQL GROUP BY DATE 在数据库端直接按天汇总, 大幅减少数据传输量。

新增 UsageDailyHeatmapQuery / StoredUsageDailySummary 类型,
在 trait、SQL、内存实现中均补齐 summarize_usage_daily_heatmap 方法。

同时优化 docker-compose: postgres 增加空闲事务超时与 keepalive 参数,
gateway 增加健康检查配置。
This commit is contained in:
fawney19
2026-04-15 02:11:32 +08:00
parent 05fbbac493
commit 98ad1172b0
11 changed files with 289 additions and 54 deletions

View File

@@ -10,8 +10,8 @@ use serde_json::Value;
use super::{
strip_deprecated_usage_display_fields, usage_can_recover_terminal_failure,
StoredProviderApiKeyUsageSummary, StoredProviderUsageSummary, StoredProviderUsageWindow,
StoredRequestUsageAudit, UpsertUsageRecord, UsageAuditListQuery, UsageReadRepository,
UsageWriteRepository,
StoredRequestUsageAudit, StoredUsageDailySummary, UpsertUsageRecord, UsageAuditListQuery,
UsageDailyHeatmapQuery, UsageReadRepository, UsageWriteRepository,
};
use crate::DataLayerError;
@@ -335,6 +335,70 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
Ok(summary)
}
async fn summarize_usage_daily_heatmap(
&self,
query: &UsageDailyHeatmapQuery,
) -> Result<Vec<StoredUsageDailySummary>, DataLayerError> {
let items = self.by_request_id.read().expect("usage repository lock");
let mut daily = BTreeMap::<String, (u64, u64, f64, f64)>::new();
for item in items.values() {
if item.created_at_unix_ms < query.created_from_unix_secs {
continue;
}
if let Some(user_id) = &query.user_id {
if item.user_id.as_deref() != Some(user_id) {
continue;
}
}
if query.admin_mode {
if item.status == "pending" || item.status == "streaming" {
continue;
}
} else if item.billing_status != "settled" || item.total_cost_usd <= 0.0 {
continue;
}
let ts = i64::try_from(item.created_at_unix_ms).unwrap_or_default();
let Some(dt) = chrono::DateTime::<chrono::Utc>::from_timestamp(ts, 0) else {
continue;
};
let date_key = dt.date_naive().to_string();
let entry = daily.entry(date_key).or_insert((0, 0, 0.0, 0.0));
entry.0 += 1;
let cache_creation = if item.cache_creation_input_tokens == 0
&& (item.cache_creation_ephemeral_5m_input_tokens
+ item.cache_creation_ephemeral_1h_input_tokens)
> 0
{
item.cache_creation_ephemeral_5m_input_tokens
+ item.cache_creation_ephemeral_1h_input_tokens
} else {
item.cache_creation_input_tokens
};
entry.1 += item.input_tokens
+ item.output_tokens
+ cache_creation
+ item.cache_read_input_tokens;
entry.2 += item.total_cost_usd;
entry.3 += item.actual_total_cost_usd;
}
let mut result: Vec<_> = daily
.into_iter()
.map(
|(date, (requests, total_tokens, total_cost_usd, actual_total_cost_usd))| {
StoredUsageDailySummary {
date,
requests,
total_tokens,
total_cost_usd,
actual_total_cost_usd,
}
},
)
.collect();
result.sort_by(|a, b| a.date.cmp(&b.date));
Ok(result)
}
}
fn detach_usage_body(

View File

@@ -4,8 +4,8 @@ mod sql;
#[allow(unused_imports)]
pub(crate) use aether_data_contracts::repository::usage::{
StoredProviderApiKeyUsageSummary, StoredProviderUsageSummary, StoredProviderUsageWindow,
StoredRequestUsageAudit, UpsertUsageRecord, UsageAuditListQuery, UsageReadRepository,
UsageRepository, UsageWriteRepository,
StoredRequestUsageAudit, StoredUsageDailySummary, UpsertUsageRecord, UsageAuditListQuery,
UsageDailyHeatmapQuery, UsageReadRepository, UsageRepository, UsageWriteRepository,
};
pub use memory::InMemoryUsageReadRepository;
pub use sql::SqlxUsageReadRepository;

View File

@@ -14,7 +14,8 @@ use uuid::Uuid;
use super::{
incoming_usage_can_recover_terminal_failure, strip_deprecated_usage_display_fields,
StoredProviderApiKeyUsageSummary, StoredProviderUsageSummary, StoredRequestUsageAudit,
UpsertUsageRecord, UsageAuditListQuery, UsageReadRepository, UsageWriteRepository,
StoredUsageDailySummary, UpsertUsageRecord, UsageAuditListQuery, UsageDailyHeatmapQuery,
UsageReadRepository, UsageWriteRepository,
};
use crate::postgres::PostgresTransactionRunner;
use crate::{error::SqlxResultExt, DataLayerError};
@@ -1209,6 +1210,67 @@ impl SqlxUsageReadRepository {
Ok(items)
}
pub async fn summarize_usage_daily_heatmap(
&self,
query: &UsageDailyHeatmapQuery,
) -> Result<Vec<StoredUsageDailySummary>, DataLayerError> {
let mut sql = String::from(
r#"SELECT
DATE("usage".created_at) AS day,
COUNT(*)::BIGINT AS requests,
COALESCE(SUM("usage".input_tokens + "usage".output_tokens
+ CASE
WHEN COALESCE("usage".cache_creation_input_tokens, 0) = 0
AND (COALESCE("usage".cache_creation_input_tokens_5m, 0) + COALESCE("usage".cache_creation_input_tokens_1h, 0)) > 0
THEN COALESCE("usage".cache_creation_input_tokens_5m, 0) + COALESCE("usage".cache_creation_input_tokens_1h, 0)
ELSE COALESCE("usage".cache_creation_input_tokens, 0)
END
+ COALESCE("usage".cache_read_input_tokens, 0)), 0)::BIGINT AS total_tokens,
COALESCE(SUM(CAST("usage".total_cost_usd AS DOUBLE PRECISION)), 0) AS total_cost_usd,
COALESCE(SUM(CAST("usage".actual_total_cost_usd AS DOUBLE PRECISION)), 0) AS actual_total_cost_usd
FROM "usage"
WHERE "usage".created_at >= TO_TIMESTAMP($1::double precision)"#,
);
if query.admin_mode {
sql.push_str(" AND \"usage\".status NOT IN ('pending', 'streaming')");
} else {
sql.push_str(
" AND \"usage\".billing_status = 'settled' AND CAST(\"usage\".total_cost_usd AS DOUBLE PRECISION) > 0",
);
}
let mut bind_index = 2;
if query.user_id.is_some() {
sql.push_str(&format!(" AND \"usage\".user_id = ${bind_index}"));
bind_index += 1;
}
let _ = bind_index;
sql.push_str(" GROUP BY day ORDER BY day ASC");
let mut q = sqlx::query(&sql).bind(query.created_from_unix_secs as f64);
if let Some(user_id) = &query.user_id {
q = q.bind(user_id.clone());
}
let mut rows = q.fetch(&self.pool);
let mut items = Vec::new();
while let Some(row) = rows.try_next().await.map_postgres_err()? {
let day: chrono::NaiveDate = row.try_get("day").map_postgres_err()?;
let requests: i64 = row.try_get("requests").map_postgres_err()?;
let total_tokens: i64 = row.try_get("total_tokens").map_postgres_err()?;
let total_cost_usd: f64 = row.try_get("total_cost_usd").map_postgres_err()?;
let actual_total_cost_usd: f64 =
row.try_get("actual_total_cost_usd").map_postgres_err()?;
items.push(StoredUsageDailySummary {
date: day.to_string(),
requests: requests as u64,
total_tokens: total_tokens as u64,
total_cost_usd,
actual_total_cost_usd,
});
}
Ok(items)
}
pub async fn list_recent_usage_audits(
&self,
user_id: Option<&str>,
@@ -1703,6 +1765,13 @@ impl UsageReadRepository for SqlxUsageReadRepository {
) -> Result<StoredProviderUsageSummary, DataLayerError> {
Self::summarize_provider_usage_since(self, provider_id, since_unix_secs).await
}
async fn summarize_usage_daily_heatmap(
&self,
query: &UsageDailyHeatmapQuery,
) -> Result<Vec<StoredUsageDailySummary>, DataLayerError> {
Self::summarize_usage_daily_heatmap(self, query).await
}
}
#[async_trait]