i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-10-03 16:07:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(stats): scope group usage by providers and add ungrouped view

Browse Source
This commit is contained in:
wangpengxiang
2026-09-21 12:49:48 +08:00
parent ba7c9f8b27
commit a9a7c64e5d
15 changed files with 622 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/handlers/admin/observability/mod.rs
+107
View File
@@ -31,6 +31,16 @@ pub(crate) async fn resolve_usage_user_group_scope(
return Ok(Err("user group data is unavailable".to_string()));
}
if group_id == UNGROUPED_USAGE_ID {
let ids = ungrouped_usage_users(state)
.await?
.into_iter()
.filter(|user| include_inactive || user.is_active)
.filter(|user| !exclude_admin || !user.role.eq_ignore_ascii_case("admin"))
.map(|user| user.id)
.collect();
return Ok(Ok(Some(ids)));
}
match state
.resolve_usage_user_group_member_ids(&group_id, include_inactive, exclude_admin)
.await?
@@ -39,3 +49,100 @@ pub(crate) async fn resolve_usage_user_group_scope(
None => Ok(Err("user_group_id does not exist".to_string())),
}
}
/// Reserved statistics-only scope; never a permission group.
pub(crate) const UNGROUPED_USAGE_ID: &str = "__ungrouped__";
pub(crate) async fn ungrouped_usage_users(
state: &crate::handlers::admin::request::AdminAppState<'_>,
) -> Result<Vec<aether_data::repository::users::StoredUserSummary>, crate::GatewayError> {
use aether_data::repository::users::UserExportListQuery;
let mut users = Vec::new();
let mut skip = 0;
loop {
let page = state
.list_export_users_page(&UserExportListQuery {
skip,
limit: 500,
..Default::default()
})
.await?;
let count = page.len();
if count == 0 {
break;
}
let ids = page.into_iter().map(|user| user.id).collect::<Vec<_>>();
let grouped = state
.list_user_group_memberships_by_user_ids(&ids)
.await?
.into_iter()
.map(|membership| membership.user_id)
.collect::<std::collections::BTreeSet<_>>();
let ids = ids
.into_iter()
.filter(|id| !grouped.contains(id))
.collect::<Vec<_>>();
users.extend(
state
.list_users_by_ids(&ids)
.await?
.into_iter()
.filter(|user| !user.is_deleted),
);
skip += count;
if count < 500 {
break;
}
}
Ok(users)
}
/// Current group provider policy, resolved to the provider-name dimension used by usage rollups.
/// None is unrestricted; Some(empty) deliberately matches no usage.
pub(crate) async fn usage_group_provider_names(
state: &crate::handlers::admin::request::AdminAppState<'_>,
group: &aether_data::repository::users::StoredUserGroup,
) -> Result<Option<Vec<String>>, crate::GatewayError> {
if matches!(
group.allowed_providers_mode.as_str(),
"unrestricted" | "inherit"
) {
return Ok(None);
}
if group.allowed_providers_mode != "specific" {
return Ok(Some(Vec::new()));
}
let allowed = group.allowed_providers.as_deref().unwrap_or_default();
let providers = state.list_provider_catalog_providers(false).await?;
let mut names = providers
.into_iter()
.filter(|provider| {
allowed.iter().any(|value| {
let value = value.trim();
value.eq_ignore_ascii_case(&provider.id)
|| value.eq_ignore_ascii_case(&provider.name)
|| value.eq_ignore_ascii_case(&provider.provider_type)
})
})
.map(|provider| provider.name)
.collect::<Vec<_>>();
names.sort();
names.dedup();
Ok(Some(names))
}
pub(crate) async fn resolve_usage_group_provider_names(
state: &crate::handlers::admin::request::AdminAppState<'_>,
query: Option<&str>,
) -> Result<Option<Vec<String>>, crate::GatewayError> {
let Some(id) = crate::handlers::admin::shared::query_param_value(query, "user_group_id") else {
return Ok(None);
};
if id == UNGROUPED_USAGE_ID {
return Ok(None);
}
let Some(group) = state.find_user_group_by_id(&id).await? else {
return Ok(Some(Vec::new()));
};
usage_group_provider_names(state, &group).await
}
apps/aether-gateway/src/handlers/admin/observability/monitoring/activity.rs
+1
View File
@@ -169,6 +169,7 @@ pub(super) async fn build_admin_monitoring_system_status_response(
let today_usage = state
.summarize_usage_audits(&UsageAuditSummaryQuery {
provider_names: None,
created_from_unix_secs: today_start.timestamp().max(0) as u64,
created_until_unix_secs: now_unix_secs.saturating_add(1),
user_id: None,
apps/aether-gateway/src/handlers/admin/observability/stats/analytics_routes.rs
+7
View File
@@ -110,6 +110,7 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response(
};
let current_summary = state
.summarize_usage_audits(&UsageAuditSummaryQuery {
provider_names: None,
created_from_unix_secs: current_from_unix_secs,
created_until_unix_secs: current_until_unix_secs,
..Default::default()
@@ -117,6 +118,7 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response(
.await?;
let comparison_summary = state
.summarize_usage_audits(&UsageAuditSummaryQuery {
provider_names: None,
created_from_unix_secs: comparison_from_unix_secs,
created_until_unix_secs: comparison_until_unix_secs,
..Default::default()
@@ -318,6 +320,11 @@ pub(super) async fn maybe_build_local_admin_stats_analytics_response(
};
let buckets = state
.summarize_usage_time_series(&UsageTimeSeriesQuery {
provider_names: super::super::resolve_usage_group_provider_names(
state,
request_context.query_string(),
)
.await?,
created_from_unix_secs,
created_until_unix_secs,
granularity: query_granularity,
apps/aether-gateway/src/handlers/admin/observability/stats/cost_routes.rs
+1
View File
@@ -72,6 +72,7 @@ pub(super) async fn maybe_build_local_admin_stats_cost_response(
};
let buckets = state
.summarize_usage_time_series(&UsageTimeSeriesQuery {
provider_names: None,
created_from_unix_secs,
created_until_unix_secs,
granularity: UsageTimeSeriesGranularity::Day,
apps/aether-gateway/src/handlers/admin/observability/stats/leaderboard_routes.rs
+65 -34
View File
@@ -78,6 +78,7 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
};
let summaries = state
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
provider_names: None,
created_from_unix_secs,
created_until_unix_secs,
group_by: UsageLeaderboardGroupBy::Model,
@@ -156,6 +157,7 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
};
let summaries = state
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
provider_names: None,
created_from_unix_secs,
created_until_unix_secs,
group_by: UsageLeaderboardGroupBy::ApiKey,
@@ -283,34 +285,6 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
)));
};
let summaries = state
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
created_from_unix_secs,
created_until_unix_secs,
group_by: UsageLeaderboardGroupBy::User,
user_id: None,
user_ids: None,
provider_name: filters.provider_name,
model: filters.model,
})
.await?;
let user_ids = summaries
.iter()
.map(|item| item.group_key.clone())
.collect::<Vec<_>>();
let user_metadata = load_user_leaderboard_metadata(state, &user_ids).await?;
let user_usage = build_user_leaderboard_items_from_summaries(
&summaries,
&user_metadata,
state.has_auth_user_data_reader(),
state.has_user_data_reader(),
include_inactive,
exclude_admin,
)
.into_iter()
.map(|item| (item.id.clone(), item))
.collect::<BTreeMap<_, _>>();
let mut leaderboard = Vec::new();
let mut member_counts = BTreeMap::new();
let mut active_member_counts = BTreeMap::new();
@@ -321,13 +295,38 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
.iter()
.filter(|member| !member.is_deleted && member.is_active)
.count();
let scoped_user_ids = members
let user_ids = members
.iter()
.filter(|member| !member.is_deleted)
.filter(|member| include_inactive || member.is_active)
.filter(|member| !exclude_admin || !member.role.eq_ignore_ascii_case("admin"))
.map(|member| member.user_id.as_str())
.collect::<BTreeSet<_>>();
.map(|member| member.user_id.clone())
.collect::<Vec<_>>();
let summaries = state
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
created_from_unix_secs,
created_until_unix_secs,
group_by: UsageLeaderboardGroupBy::User,
user_id: None,
user_ids: Some(user_ids),
provider_names: super::super::usage_group_provider_names(state, &group).await?,
provider_name: filters.provider_name.clone(),
model: filters.model.clone(),
})
.await?;
let user_ids = summaries
.iter()
.map(|row| row.group_key.clone())
.collect::<Vec<_>>();
let metadata = load_user_leaderboard_metadata(state, &user_ids).await?;
let users = build_user_leaderboard_items_from_summaries(
&summaries,
&metadata,
state.has_auth_user_data_reader(),
state.has_user_data_reader(),
include_inactive,
exclude_admin,
);
let mut item = AdminStatsLeaderboardItem {
id: group.id.clone(),
name: group.name,
@@ -335,17 +334,47 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
tokens: 0,
cost: 0.0,
};
for user_id in scoped_user_ids {
if let Some(user) = user_usage.get(user_id) {
for user in users {
item.requests = item.requests.saturating_add(user.requests);
item.tokens = item.tokens.saturating_add(user.tokens);
item.cost += user.cost;
}
}
member_counts.insert(group.id.clone(), member_count);
active_member_counts.insert(group.id, active_member_count);
leaderboard.push(item);
}
let ungrouped = super::super::ungrouped_usage_users(state).await?;
let id = super::super::UNGROUPED_USAGE_ID.to_string();
member_counts.insert(id.clone(), ungrouped.len());
active_member_counts.insert(
id.clone(),
ungrouped.iter().filter(|user| user.is_active).count(),
);
let user_ids = ungrouped
.into_iter()
.filter(|user| include_inactive || user.is_active)
.filter(|user| !exclude_admin || !user.role.eq_ignore_ascii_case("admin"))
.map(|user| user.id)
.collect();
let rows = state
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
created_from_unix_secs,
created_until_unix_secs,
group_by: UsageLeaderboardGroupBy::User,
user_id: None,
user_ids: Some(user_ids),
provider_names: None,
provider_name: filters.provider_name.clone(),
model: filters.model.clone(),
})
.await?;
leaderboard.push(AdminStatsLeaderboardItem {
id,
name: "Ungrouped".to_string(),
requests: rows.iter().map(|row| row.request_count).sum(),
tokens: rows.iter().map(|row| row.total_tokens).sum(),
cost: rows.iter().map(|row| row.total_cost_usd).sum(),
});
leaderboard.sort_by(|left, right| compare_leaderboard_items(metric, order, left, right));
return Ok(Some(build_admin_stats_user_group_leaderboard_response(
@@ -422,6 +451,8 @@ pub(super) async fn maybe_build_local_admin_stats_leaderboard_response(
};
let summaries = state
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
provider_names: super::super::resolve_usage_group_provider_names(state, query)
.await?,
created_from_unix_secs,
created_until_unix_secs,
group_by: UsageLeaderboardGroupBy::User,
apps/aether-gateway/src/handlers/admin/observability/usage/summary_routes.rs
+2
View File
@@ -717,6 +717,8 @@ pub(super) async fn maybe_build_local_admin_usage_summary_response(
};
let summary = state
.summarize_usage_audits(&UsageAuditSummaryQuery {
provider_names: super::super::resolve_usage_group_provider_names(state, query)
.await?,
created_from_unix_secs,
created_until_unix_secs,
user_id: query_param_value(query, "user_id"),
apps/aether-gateway/src/tests/control/admin/stats.rs
+222 -1
View File
@@ -2031,7 +2031,12 @@ async fn gateway_aggregates_admin_stats_by_current_user_group_membership() {
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["attribution"], "current_membership");
assert_eq!(payload["total"], 1);
assert_eq!(payload["total"], 2);
let mut payload = payload;
payload["items"]
.as_array_mut()
.unwrap()
.retain(|item| item["id"] == group.id);
assert_eq!(payload["items"][0]["id"], group.id);
assert_eq!(payload["items"][0]["name"], "Engineering");
assert_eq!(payload["items"][0]["requests"], 2);
@@ -2067,6 +2072,222 @@ async fn gateway_aggregates_admin_stats_by_current_user_group_membership() {
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_group_usage_intersects_members_and_providers_with_shared_overlap() {
let rows = vec![
sample_usage_row(
"g",
"g",
Some("user-1"),
None,
None,
"Gemini",
"model",
10,
2,
0.4,
0.4,
DAY_1_UNIX_SECS,
),
sample_usage_row(
"s",
"s",
Some("user-1"),
None,
None,
"Shared",
"model",
10,
2,
0.35,
0.35,
DAY_1_UNIX_SECS,
),
sample_usage_row(
"o",
"o",
Some("user-1"),
None,
None,
"Other",
"model",
10,
2,
1.5,
1.5,
DAY_1_UNIX_SECS,
),
sample_usage_row(
"x",
"x",
Some("user-2"),
None,
None,
"Gemini",
"model",
10,
2,
10.0,
10.0,
DAY_1_UNIX_SECS,
),
];
let users = InMemoryUserReadRepository::seed_auth_users([
sample_auth_user("user-1", "alice", "user", true),
sample_auth_user("user-2", "bob", "user", true),
]);
let export_users = users.list_export_users().await.unwrap();
let users = users.with_export_users(export_users);
let mut group_ids = Vec::new();
for (name, allowed, mode) in [
(
"Gemini group",
vec!["provider-gemini", "Shared", "provider-shared"],
"specific",
),
("Other group", vec!["Other", "shared-type"], "specific"),
("Denied", vec!["Gemini"], "deny_all"),
("Empty", vec![], "specific"),
("Inherited", vec![], "inherit"),
("Unrestricted", vec![], "unrestricted"),
] {
let group = users
.create_user_group(UpsertUserGroupRecord {
name: name.to_string(),
description: None,
priority: 0,
allowed_providers: Some(allowed.into_iter().map(str::to_string).collect()),
allowed_providers_mode: mode.to_string(),
allowed_api_formats: None,
allowed_api_formats_mode: "inherit".to_string(),
allowed_models: None,
allowed_models_mode: "inherit".to_string(),
rate_limit: None,
rate_limit_mode: "inherit".to_string(),
})
.await
.unwrap()
.unwrap();
users
.replace_user_group_members(&group.id, &["user-1".to_string()])
.await
.unwrap();
group_ids.push(group.id);
}
let mut shared = sample_provider("provider-shared", "Shared", 0);
shared.provider_type = "shared-type".to_string();
let providers = InMemoryProviderCatalogReadRepository::seed(
vec![
sample_provider("provider-gemini", "Gemini", 0),
shared,
sample_provider("provider-other", "Other", 0),
],
vec![],
vec![],
);
let data = GatewayDataState::with_usage_reader_for_tests(Arc::new(
InMemoryUsageReadRepository::seed(rows),
))
.with_user_reader(Arc::new(users))
.with_provider_catalog_reader(Arc::new(providers));
let gateway = build_router_with_state(AppState::new().unwrap().with_data_state_for_tests(data));
let (url, handle) = start_server(gateway).await;
let client = reqwest::Client::new();
let range = "start_date=2024-03-21&end_date=2024-03-21&tz_offset_minutes=0";
let paths = [
(
format!("usage/stats?{range}&user_group_id=__ungrouped__"),
Some(10.0),
),
(
format!("stats/time-series?{range}&granularity=day&user_group_id=__ungrouped__"),
Some(10.0),
),
(
format!("stats/leaderboard/users?{range}&metric=cost&user_group_id=__ungrouped__"),
Some(10.0),
),
(
format!("stats/leaderboard/user-groups?{range}&metric=cost"),
None,
),
(
format!("usage/stats?{range}&user_group_id={}", group_ids[0]),
Some(0.75),
),
(
format!(
"stats/time-series?{range}&granularity=day&user_group_id={}",
group_ids[1]
),
Some(1.85),
),
(
format!(
"stats/leaderboard/users?{range}&metric=cost&user_group_id={}",
group_ids[0]
),
Some(0.75),
),
(format!("usage/stats?{range}&user_id=user-1"), Some(2.25)),
(
format!(
"usage/stats?{range}&user_group_id={}&provider=Other",
group_ids[0]
),
Some(0.0),
),
];
for (path, expected) in paths {
let response = admin_request(client.get(format!("{url}/api/admin/{path}")))
.send()
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK, "{path}");
let body: serde_json::Value = response.json().await.unwrap();
if let Some(expected) = expected {
let value = if path.starts_with("stats/time-series") {
&body[0]["total_cost"]
} else if path.starts_with("stats/leaderboard") {
&body["items"][0]["cost"]
} else {
&body["total_cost"]
};
assert!(
(value.as_f64().unwrap() - expected).abs() < 1e-9,
"{path}: {body}"
);
} else {
let items = body["items"].as_array().unwrap();
assert_eq!(items.len(), 7);
let ungrouped = items
.iter()
.find(|item| item["id"] == "__ungrouped__")
.unwrap();
assert_eq!(ungrouped["cost"], 10.0);
assert_eq!(ungrouped["member_count"], 1);
assert_eq!(ungrouped["active_member_count"], 1);
for (id, cost, requests) in [
(&group_ids[0], 0.75, 2),
(&group_ids[1], 1.85, 2),
(&group_ids[2], 0.0, 0),
(&group_ids[3], 0.0, 0),
(&group_ids[4], 2.25, 3),
(&group_ids[5], 2.25, 3),
] {
let row = items
.iter()
.find(|item| item["id"].as_str() == Some(id.as_str()))
.unwrap();
assert!((row["cost"].as_f64().unwrap() - cost).abs() < 1e-9);
assert_eq!(row["requests"], requests);
}
}
}
handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_stats_leaderboard_users_locally_without_usage_reader() {
let (upstream_url, upstream_hits, upstream_handle) =
crates/aether-data/adapters/postgres/src/usage/mod.rs
+93 -3
View File
@@ -3715,6 +3715,7 @@ FROM "#,
created_until_unix_secs: u64,
user_id: Option<&str>,
user_ids: Option<&[String]>,
provider_names: Option<&[String]>,
provider_name: Option<&str>,
model: Option<&str>,
) -> Result<StoredUsageAuditSummary, DataLayerError> {
@@ -3770,6 +3771,12 @@ FROM usage_billing_facts AS "usage"
.push_bind(created_until_unix_secs as f64)
.push("::double precision)");
push_usage_user_scope(&mut builder, "\"usage\".user_id", user_id, user_ids);
if let Some(names) = provider_names {
builder
.push(" AND \"usage\".provider_name = ANY(")
.push_bind(names.to_vec())
.push("::text[])");
}
if let Some(provider_name) = provider_name {
builder.push(if has_where { " AND " } else { " WHERE " });
has_where = true;
@@ -3796,13 +3803,17 @@ FROM usage_billing_facts AS "usage"
&self,
query: &UsageAuditSummaryQuery,
) -> Result<StoredUsageAuditSummary, DataLayerError> {
if query.provider_name.is_some() || query.model.is_some() {
// Provider rollups lack cache-cost/error detail required by this response.
// Keep these scopes on canonical facts rather than inventing missing daily totals.
if query.provider_names.is_some() || query.provider_name.is_some() || query.model.is_some()
{
return self
.summarize_usage_audits_raw(
query.created_from_unix_secs,
query.created_until_unix_secs,
query.user_id.as_deref(),
query.user_ids.as_deref(),
query.provider_names.as_deref(),
query.provider_name.as_deref(),
query.model.as_deref(),
)
@@ -3815,6 +3826,7 @@ FROM usage_billing_facts AS "usage"
query.created_until_unix_secs,
query.user_id.as_deref(),
query.user_ids.as_deref(),
query.provider_names.as_deref(),
None,
None,
)
@@ -3831,6 +3843,7 @@ FROM usage_billing_facts AS "usage"
query.created_until_unix_secs,
query.user_id.as_deref(),
query.user_ids.as_deref(),
query.provider_names.as_deref(),
None,
None,
)
@@ -3846,6 +3859,7 @@ FROM usage_billing_facts AS "usage"
dashboard_utc_to_unix_secs(raw_end),
query.user_id.as_deref(),
query.user_ids.as_deref(),
query.provider_names.as_deref(),
None,
None,
)
@@ -3872,6 +3886,7 @@ FROM usage_billing_facts AS "usage"
dashboard_utc_to_unix_secs(raw_end),
query.user_id.as_deref(),
query.user_ids.as_deref(),
query.provider_names.as_deref(),
None,
None,
)
@@ -6821,6 +6836,12 @@ FROM usage_billing_facts AS "usage"
query.user_id.as_deref(),
query.user_ids.as_deref(),
);
if let Some(names) = query.provider_names.as_ref() {
builder
.push(" AND \"usage\".provider_name = ANY(")
.push_bind(names.clone())
.push("::text[])");
}
if let Some(provider_name) = query.provider_name.as_deref() {
builder.push(if has_where { " AND " } else { " WHERE " });
has_where = true;
@@ -6861,6 +6882,7 @@ FROM usage_billing_facts AS "usage"
end_day_utc: DateTime<Utc>,
user_id: Option<&str>,
user_ids: Option<&[String]>,
provider_names: Option<&[String]>,
) -> Result<Vec<StoredUsageTimeSeriesBucket>, DataLayerError> {
if start_day_utc >= end_day_utc {
return Ok(Vec::new());
@@ -6881,7 +6903,9 @@ SELECT
AS total_response_time_ms
FROM "#,
);
builder.push(if scoped_to_users {
builder.push(if provider_names.is_some() {
"stats_user_daily_provider"
} else if scoped_to_users {
"stats_user_daily"
} else {
"stats_daily"
@@ -6894,6 +6918,12 @@ FROM "#,
if scoped_to_users {
push_usage_user_scope(&mut builder, "user_id", user_id, user_ids);
}
if let Some(names) = provider_names {
builder
.push(" AND provider_name = ANY(")
.push_bind(names.to_vec())
.push("::text[])");
}
builder.push(" GROUP BY date ORDER BY date ASC");
let mut rows = builder.build().fetch(&self.pool);
@@ -6979,6 +7009,7 @@ WHERE is_complete IS TRUE
&mut grouped,
self.summarize_usage_time_series_raw(
&UsageTimeSeriesQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end),
granularity: UsageTimeSeriesGranularity::Day,
@@ -7000,6 +7031,7 @@ WHERE is_complete IS TRUE
aggregate_end,
query.user_id.as_deref(),
query.user_ids.as_deref(),
query.provider_names.as_deref(),
)
.await?,
);
@@ -7007,6 +7039,7 @@ WHERE is_complete IS TRUE
&mut grouped,
self.summarize_usage_time_series_raw(
&UsageTimeSeriesQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(aggregate_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(aggregate_end),
granularity: UsageTimeSeriesGranularity::Day,
@@ -7025,6 +7058,7 @@ WHERE is_complete IS TRUE
&mut grouped,
self.summarize_usage_time_series_raw(
&UsageTimeSeriesQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end),
granularity: UsageTimeSeriesGranularity::Day,
@@ -7044,7 +7078,10 @@ WHERE is_complete IS TRUE
}
}
if query.user_id.is_none() && query.user_ids.is_none() && query.tz_offset_minutes % 60 == 0
if query.provider_names.is_none()
&& query.user_id.is_none()
&& query.user_ids.is_none()
&& query.tz_offset_minutes % 60 == 0
{
if let Some(cutoff_utc) = self.read_stats_hourly_cutoff().await? {
let start_utc = dashboard_unix_secs_to_utc(query.created_from_unix_secs);
@@ -7057,6 +7094,7 @@ WHERE is_complete IS TRUE
&mut grouped,
self.summarize_usage_time_series_raw(
&UsageTimeSeriesQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end),
granularity: query.granularity,
@@ -7085,6 +7123,7 @@ WHERE is_complete IS TRUE
&mut grouped,
self.summarize_usage_time_series_raw(
&UsageTimeSeriesQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(aggregate_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(aggregate_end),
granularity: query.granularity,
@@ -7103,6 +7142,7 @@ WHERE is_complete IS TRUE
&mut grouped,
self.summarize_usage_time_series_raw(
&UsageTimeSeriesQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end),
granularity: query.granularity,
@@ -7149,6 +7189,7 @@ WHERE "usage".created_at >= TO_TIMESTAMP($1::double precision)
AND ($4::varchar IS NULL OR "usage".provider_name = $4)
AND ($5::varchar IS NULL OR "usage".model = $5)
AND ($6::text[] IS NULL OR "usage".user_id::text = ANY($6))
AND ($7::text[] IS NULL OR "usage".provider_name = ANY($7))
GROUP BY group_key
ORDER BY group_key ASC
"#,
@@ -7163,6 +7204,7 @@ ORDER BY group_key ASC
.bind(query.provider_name.as_deref())
.bind(query.model.as_deref())
.bind(query.user_ids.clone())
.bind(query.provider_names.clone())
.fetch(&self.pool);
let mut items = Vec::new();
while let Some(row) = rows.try_next().await.map_postgres_err()? {
@@ -7177,6 +7219,51 @@ ORDER BY group_key ASC
end_day_utc: DateTime<Utc>,
query: &UsageLeaderboardQuery,
) -> Result<Option<Vec<StoredUsageLeaderboardSummary>>, DataLayerError> {
if let Some(names) = query.provider_names.as_ref() {
let group_key = match query.group_by {
UsageLeaderboardGroupBy::User => "user_id",
UsageLeaderboardGroupBy::Model => "model",
UsageLeaderboardGroupBy::ApiKey => return Ok(None),
};
let mut builder = QueryBuilder::<Postgres>::new(format!(
r#"
SELECT
{group_key} AS group_key,
MAX(NULLIF(BTRIM(username), '')) AS legacy_name,
COALESCE(SUM(total_requests), 0)::BIGINT AS request_count,
COALESCE(SUM(total_tokens), 0)::BIGINT AS total_tokens,
COALESCE(SUM(total_cost), 0)::DOUBLE PRECISION AS total_cost_usd
FROM stats_user_daily_model_provider
WHERE date >=
"#
));
builder
.push_bind(start_day_utc)
.push(" AND date < ")
.push_bind(end_day_utc);
builder
.push(" AND provider_name = ANY(")
.push_bind(names.clone())
.push("::text[])");
push_usage_user_scope(
&mut builder,
"user_id",
query.user_id.as_deref(),
query.user_ids.as_deref(),
);
if let Some(provider) = query.provider_name.as_ref() {
builder
.push(" AND provider_name = ")
.push_bind(provider.clone());
}
if let Some(model) = query.model.as_ref() {
builder.push(" AND model = ").push_bind(model.clone());
}
builder.push(format!(" GROUP BY {group_key} ORDER BY {group_key}"));
return fetch_usage_leaderboard_query(builder.build(), &self.pool)
.await
.map(Some);
}
let items = match query.group_by {
UsageLeaderboardGroupBy::Model => {
let mut builder = if let Some(user_id) = query.user_id.as_deref() {
@@ -7452,6 +7539,7 @@ WHERE stats_daily_api_key.date >=
absorb_usage_leaderboard_rows(
&mut grouped,
self.summarize_usage_leaderboard_raw(&UsageLeaderboardQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end),
..query.clone()
@@ -7474,6 +7562,7 @@ WHERE stats_daily_api_key.date >=
absorb_usage_leaderboard_rows(
&mut grouped,
self.summarize_usage_leaderboard_raw(&UsageLeaderboardQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(aggregate_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(aggregate_end),
..query.clone()
@@ -7487,6 +7576,7 @@ WHERE stats_daily_api_key.date >=
absorb_usage_leaderboard_rows(
&mut grouped,
self.summarize_usage_leaderboard_raw(&UsageLeaderboardQuery {
provider_names: query.provider_names.clone(),
created_from_unix_secs: dashboard_utc_to_unix_secs(raw_start),
created_until_unix_secs: dashboard_utc_to_unix_secs(raw_end),
..query.clone()
crates/aether-data/contracts/src/repository/usage/types.rs
+9
View File
@@ -1060,6 +1060,9 @@ pub struct UsageAuditAggregationQuery {
#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub struct UsageAuditSummaryQuery {
/// Optional provider-name allowlist, intersected with provider_name; empty matches nothing.
#[serde(default)]
pub provider_names: Option<Vec<String>>,
pub created_from_unix_secs: u64,
pub created_until_unix_secs: u64,
pub user_id: Option<String>,
@@ -1453,6 +1456,9 @@ pub enum UsageTimeSeriesGranularity {
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct UsageTimeSeriesQuery {
/// Optional provider-name allowlist, intersected with provider_name; empty matches nothing.
#[serde(default)]
pub provider_names: Option<Vec<String>>,
pub created_from_unix_secs: u64,
pub created_until_unix_secs: u64,
pub granularity: UsageTimeSeriesGranularity,
@@ -1487,6 +1493,9 @@ pub enum UsageLeaderboardGroupBy {
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct UsageLeaderboardQuery {
/// Optional provider-name allowlist, intersected with provider_name; empty matches nothing.
#[serde(default)]
pub provider_names: Option<Vec<String>>,
pub created_from_unix_secs: u64,
pub created_until_unix_secs: u64,
pub group_by: UsageLeaderboardGroupBy,
crates/aether-data/runtime/src/lifecycle/migrate/tests.rs
+1
View File
@@ -2382,6 +2382,7 @@ INSERT INTO public.stats_daily_api_key (
.expect("API key daily aggregate fixtures should be inserted");
let leaderboard_query = UsageLeaderboardQuery {
provider_names: None,
created_from_unix_secs: u64::try_from(stats_day.timestamp())
.expect("historical stats day should be nonnegative"),
created_until_unix_secs: u64::try_from((stats_day + chrono::Duration::days(1)).timestamp())
crates/aether-data/runtime/src/repository/usage/memory.rs
+21
View File
@@ -502,6 +502,13 @@ fn usage_matches_summary_query(
return false;
}
}
if query
.provider_names
.as_ref()
.is_some_and(|names| !names.contains(&item.provider_name))
{
return false;
}
if let Some(provider_name) = query.provider_name.as_deref() {
if item.provider_name != provider_name {
return false;
@@ -538,6 +545,13 @@ fn usage_matches_time_series_query(
return false;
}
}
if query
.provider_names
.as_ref()
.is_some_and(|names| !names.contains(&item.provider_name))
{
return false;
}
if let Some(provider_name) = query.provider_name.as_deref() {
if item.provider_name != provider_name {
return false;
@@ -918,6 +932,13 @@ fn usage_matches_leaderboard_query(
return false;
}
}
if query
.provider_names
.as_ref()
.is_some_and(|names| !names.contains(&item.provider_name))
{
return false;
}
if let Some(provider_name) = query.provider_name.as_deref() {
if item.provider_name != provider_name {
return false;
crates/aether-data/runtime/src/repository/usage/memory/tests.rs
+78
View File
@@ -457,6 +457,7 @@ async fn unmetered_session_audit_counts_lifecycle_without_token_or_cost_contribu
let summary = repository
.summarize_usage_audits(&UsageAuditSummaryQuery {
provider_names: None,
created_from_unix_secs: 0,
created_until_unix_secs: 1_000,
..UsageAuditSummaryQuery::default()
@@ -2213,6 +2214,7 @@ async fn dashboard_and_leaderboard_total_tokens_use_effective_cache_aware_tokens
let leaderboard = repository
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
provider_names: None,
created_from_unix_secs: 1_711_000_000,
created_until_unix_secs: 1_711_000_001,
group_by: UsageLeaderboardGroupBy::User,
@@ -2239,6 +2241,7 @@ async fn usage_analytics_filters_by_multiple_user_ids() {
let summary = repository
.summarize_usage_audits(&UsageAuditSummaryQuery {
provider_names: None,
created_from_unix_secs: 1_711_000_000,
created_until_unix_secs: 1_711_000_001,
user_ids: Some(scoped_user_ids.clone()),
@@ -2250,6 +2253,7 @@ async fn usage_analytics_filters_by_multiple_user_ids() {
let buckets = repository
.summarize_usage_time_series(&UsageTimeSeriesQuery {
provider_names: None,
created_from_unix_secs: 1_711_000_000,
created_until_unix_secs: 1_711_000_001,
granularity: UsageTimeSeriesGranularity::Day,
@@ -2271,6 +2275,7 @@ async fn usage_analytics_filters_by_multiple_user_ids() {
let leaderboard = repository
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
provider_names: None,
created_from_unix_secs: 1_711_000_000,
created_until_unix_secs: 1_711_000_001,
group_by: UsageLeaderboardGroupBy::User,
@@ -2285,6 +2290,79 @@ async fn usage_analytics_filters_by_multiple_user_ids() {
assert!(leaderboard.iter().all(|item| item.group_key != "user-3"));
}
#[tokio::test]
async fn usage_analytics_intersects_provider_allowlist_and_user_scope() {
let mut a = sample_usage("allowed", 1_711_000_000);
a.provider_name = "Gemini".to_string();
a.user_id = Some("user-1".to_string());
let mut b = a.clone();
b.request_id = "other-provider".to_string();
b.provider_name = "Other".to_string();
let mut c = a.clone();
c.request_id = "other-user".to_string();
c.user_id = Some("user-2".to_string());
let repository = InMemoryUsageReadRepository::seed(vec![a, b, c]);
for (names, provider, count) in [
(
Some(vec!["Gemini".to_string(), "Gemini".to_string()]),
None,
1,
),
(Some(vec![]), None, 0),
(
Some(vec!["Gemini".to_string()]),
Some("Other".to_string()),
0,
),
(None, None, 2),
] {
let summary = repository
.summarize_usage_audits(&UsageAuditSummaryQuery {
created_from_unix_secs: 1_711_000_000,
created_until_unix_secs: 1_711_000_001,
user_ids: Some(vec!["user-1".to_string()]),
provider_names: names.clone(),
provider_name: provider.clone(),
..Default::default()
})
.await
.unwrap();
assert_eq!(summary.total_requests, count);
let buckets = repository
.summarize_usage_time_series(&UsageTimeSeriesQuery {
created_from_unix_secs: 1_711_000_000,
created_until_unix_secs: 1_711_000_001,
user_id: None,
user_ids: Some(vec!["user-1".to_string()]),
provider_names: names.clone(),
provider_name: provider.clone(),
model: None,
granularity: UsageTimeSeriesGranularity::Day,
tz_offset_minutes: 0,
})
.await
.unwrap();
assert_eq!(
buckets.iter().map(|row| row.total_requests).sum::<u64>(),
count
);
let rows = repository
.summarize_usage_leaderboard(&UsageLeaderboardQuery {
created_from_unix_secs: 1_711_000_000,
created_until_unix_secs: 1_711_000_001,
user_id: None,
user_ids: Some(vec!["user-1".to_string()]),
provider_names: names,
provider_name: provider,
model: None,
group_by: UsageLeaderboardGroupBy::User,
})
.await
.unwrap();
assert_eq!(rows.iter().map(|row| row.request_count).sum::<u64>(), count);
}
}
#[tokio::test]
async fn summarizes_provider_api_key_last_used_at_in_seconds() {
let repository = InMemoryUsageReadRepository::seed(vec![
frontend/src/components/stats/LeaderboardTable.vue
+1 -1
View File
@@ -77,7 +77,7 @@
<TableCell class="font-medium">
{{ item.rank }}
</TableCell>
<TableCell>{{ item.name }}</TableCell>
<TableCell>{{ showMemberCount && item.id === '__ungrouped__' ? t('userStats.ungrouped') : item.name }}</TableCell>
<TableCell
v-if="showMemberCount"
class="text-right"
frontend/src/i18n/messages.ts
+2
View File
@@ -350,6 +350,7 @@ export const messages = {
'userStats.description': '按用户或当前用户组成员查看排行榜与使用趋势',
'userStats.scope.placeholder': '统计维度',
'userStats.scope.user': '按用户',
'userStats.ungrouped': '未分组',
'userStats.scope.userGroup': '按用户组',
'userStats.search.user': '搜索用户',
'userStats.search.userGroup': '搜索用户组',
@@ -834,6 +835,7 @@ export const messages = {
'userStats.description': 'View leaderboards and usage trends by user or current user-group membership',
'userStats.scope.placeholder': 'Statistics scope',
'userStats.scope.user': 'By user',
'userStats.ungrouped': 'Ungrouped',
'userStats.scope.userGroup': 'By user group',
'userStats.search.user': 'Search users',
'userStats.search.userGroup': 'Search user groups',
frontend/src/views/admin/UserStats.vue
+9 -6
View File
@@ -297,7 +297,8 @@ let ready = false
const allEntities = computed<SelectableEntity[]>(() => scope.value === 'user'
? users.value.map(user => ({ id: user.id, name: user.username || user.email || user.id }))
: userGroups.value.map(group => ({ id: group.id, name: group.name })))
: [...userGroups.value.map(group => ({ id: group.id, name: group.name })),
{ id: '__ungrouped__', name: t('userStats.ungrouped') }])
const selectedEntityId = computed({
get: () => scope.value === 'user' ? selectedUserId.value : selectedUserGroupId.value,
@@ -417,7 +418,7 @@ async function loadPanels() {
limit: PAGE_SIZE
})
: Promise.resolve({ items: [] })
const groupMembersPromise: Promise<UserGroupMember[]> = scope.value === 'user_group'
const groupMembersPromise: Promise<UserGroupMember[]> = scope.value === 'user_group' && selectedId !== '__ungrouped__'
? usersApi.listUserGroupMembers(selectedId)
: Promise.resolve([])
@@ -433,10 +434,12 @@ async function loadPanels() {
series.value = primarySeries
comparisonSeries.value = compareSeries
memberLeaderboard.value = members.items
groupMemberCount.value = groupMembers.filter(member => !member.is_deleted).length
activeGroupMemberCount.value = groupMembers.filter(
member => !member.is_deleted && member.is_active
).length
const ungroupedUsers = users.value.filter(user => user.groups?.length === 0)
groupMemberCount.value = selectedId === '__ungrouped__'
? ungroupedUsers.length : groupMembers.filter(member => !member.is_deleted).length
activeGroupMemberCount.value = selectedId === '__ungrouped__'
? ungroupedUsers.filter(user => user.is_active).length
: groupMembers.filter(member => !member.is_deleted && member.is_active).length
} finally {
if (requestId === panelRequestId) {
summaryLoading.value = false