From a23afbf2f570e4dc1def60b26d2ad852396ce3a1 Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 23 Oct 2026 01:08:57 +0800 Subject: [PATCH] perf: streamline usage queries and batch account reads Use literal contains filters for usage names and models, align Key Usage filtering, and remove unused usage query parameters. Optimize summary aggregation, record counting, quota reads, and credential cooldown lookups without adding indexes. Co-authored-by: Codex --- .../gateway-admin/src/model/observability.rs | 54 ++- .../src/model/provider_credentials.rs | 7 + .../gateway-admin/src/ports/provider.rs | 24 +- .../crates/gateway-admin/src/ports/store.rs | 4 +- .../src/use_case/accounts/mod.rs | 117 ++++--- .../gateway-admin/src/use_case/key_usage.rs | 2 +- .../src/use_case/observability/mod.rs | 8 +- .../src/use_case/observability/store_cache.rs | 6 +- .../tests/use_case/accounts/mod.rs | 31 +- .../gateway-admin/tests/use_case/mod.rs | 8 +- .../tests/use_case/observability/mod.rs | 23 +- .../src/admin/observability/presenter.rs | 6 +- .../src/admin/observability/query.rs | 76 ++-- .../src/admin/observability/routes.rs | 14 +- backend/crates/gateway-api/tests/admin/mod.rs | 11 +- .../tests/admin/observability/query.rs | 5 +- .../tests/admin/observability/response.rs | 108 +++++- .../crates/gateway-api/tests/key_usage/mod.rs | 9 +- .../postgres/observability/admin_adapter.rs | 17 +- .../observability/analytics/accounts.rs | 2 +- .../observability/analytics/aggregate.rs | 113 +++++- .../observability/analytics/contributions.rs | 3 + .../observability/analytics/metrics.rs | 19 + .../postgres/observability/analytics/query.rs | 108 +++++- .../src/postgres/observability/mapping.rs | 3 + .../src/postgres/observability/model.rs | 20 ++ .../postgres/observability/queries/usage.rs | 235 ++++++++++--- .../gateway-store/tests/postgres/analytics.rs | 301 +++++++++++++++- .../tests/postgres/execution/mod.rs | 36 +- .../tests/postgres/observability/mod.rs | 35 +- .../gateway-store/tests/postgres/snapshots.rs | 44 +++ .../tests/postgres/usage_data/mod.rs | 7 +- .../tests/postgres/usage_paging.rs | 326 ++++++++++++++++-- backend/crates/providers/openai/src/admin.rs | 88 ++++- .../openai/src/credential/quota/service.rs | 18 +- .../openai/src/credential/selector.rs | 14 +- .../crates/providers/openai/tests/admin.rs | 175 +++++++++- .../credential/contract/cooldown_reads.rs | 255 ++++++++++++++ .../openai/tests/credential/contract/mod.rs | 1 + .../crates/providers/openai/tests/support.rs | 18 + docs/api.md | 19 +- frontend/src/api/modules/usage.ts | 11 +- frontend/src/views/key-usage/index.vue | 16 +- .../views/usage/components/UsageFilters.vue | 21 +- .../usage/composables/useUsageRecordsTable.ts | 37 +- frontend/src/views/usage/index.vue | 6 +- 46 files changed, 2140 insertions(+), 321 deletions(-) create mode 100644 backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs diff --git a/backend/crates/gateway-admin/src/model/observability.rs b/backend/crates/gateway-admin/src/model/observability.rs index 24f978e06..538c5ef0d 100644 --- a/backend/crates/gateway-admin/src/model/observability.rs +++ b/backend/crates/gateway-admin/src/model/observability.rs @@ -157,12 +157,15 @@ impl OtherRequestOutcome { /// 用量记录过滤条件 #[derive(Debug, Clone, Default, PartialEq, Eq, Hash)] pub struct UsageFilter { + pub client_api_key_name: Option, + pub provider_account_name: Option, pub client_api_key_ref: Option, pub request_id: Option, pub provider_account_ref: Option, pub operation: Option, pub provider_kind: Option, pub model: Option, + pub model_contains: Option, pub outcome: Option, pub status_code: Option, pub transport: Option, @@ -506,6 +509,46 @@ pub struct RequestMetrics { pub cache_hit_request_count: u64, } +/// 摘要实际消费的请求计数与延迟样本，不承诺分位数或分布 +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RequestSummaryMetrics { + pub request_count: u64, + pub success_count: u64, + pub failure_count: u64, + pub cancelled_count: u64, + pub incomplete_count: u64, + pub caller_error_count: u64, + pub input_tokens: u64, + pub output_tokens: u64, + pub cached_input_tokens: u64, + pub cache_write_input_tokens: u64, + pub reasoning_output_tokens: u64, + pub total_tokens: u64, + pub latency_sum_ms: u64, + pub latency_count: u64, +} + +impl From<&RequestMetrics> for RequestSummaryMetrics { + fn from(metrics: &RequestMetrics) -> Self { + Self { + request_count: metrics.request_count, + success_count: metrics.success_count, + failure_count: metrics.failure_count, + cancelled_count: metrics.cancelled_count, + incomplete_count: metrics.incomplete_count, + caller_error_count: metrics.caller_error_count, + input_tokens: metrics.input_tokens, + output_tokens: metrics.output_tokens, + cached_input_tokens: metrics.cached_input_tokens, + cache_write_input_tokens: metrics.cache_write_input_tokens, + reasoning_output_tokens: metrics.reasoning_output_tokens, + total_tokens: metrics.total_tokens, + latency_sum_ms: metrics.latency_sum_ms, + latency_count: metrics.latency_count, + } + } +} + /// 上游 attempt 聚合指标 #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct AttemptMetrics { @@ -893,10 +936,19 @@ pub struct UsageOverview { pub providers: Vec, } +/// 用量摘要所需的窄聚合事实 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UsageSummaryObservation { + pub analytics: AnalyticsMetadata, + pub range: TimeRange, + pub requests: RequestSummaryMetrics, + pub attempts: AttemptMetrics, +} + /// 用量摘要的用例结果 #[derive(Debug, Clone, PartialEq, Eq)] pub struct UsageSummary { - pub overview: UsageOverview, + pub observation: UsageSummaryObservation, pub average_latency_ms: Option, } diff --git a/backend/crates/gateway-admin/src/model/provider_credentials.rs b/backend/crates/gateway-admin/src/model/provider_credentials.rs index a8a04185a..d3f5a5879 100644 --- a/backend/crates/gateway-admin/src/model/provider_credentials.rs +++ b/backend/crates/gateway-admin/src/model/provider_credentials.rs @@ -820,6 +820,13 @@ pub struct ProviderQuotaRequest { pub rolling_usage: Option, } +/// 当前目录页已读取的账号事实；批量投影不触发上游刷新 +#[derive(Debug, Clone, Copy)] +pub struct ProviderQuotaReadRequest<'a> { + pub account: &'a AccountRecord, + pub rolling_usage: Option<&'a AccountUsage>, +} + /// Provider 已解析的 quota 结果及其不透明差异字段 #[derive(Debug, Clone, PartialEq)] pub struct ProviderQuota { diff --git a/backend/crates/gateway-admin/src/ports/provider.rs b/backend/crates/gateway-admin/src/ports/provider.rs index f88d57694..6f3808576 100644 --- a/backend/crates/gateway-admin/src/ports/provider.rs +++ b/backend/crates/gateway-admin/src/ports/provider.rs @@ -18,8 +18,9 @@ use crate::model::provider_credentials::{ PrepareCredentialImport, PrepareCredentialRefresh, PrepareCredentialRotation, PreparedAuthorizationCommit, PreparedCredentialImport, PreparedCredentialRotation, ProviderExport, ProviderExportCredentialInput, ProviderModelCatalogDocument, ProviderModels, - ProviderProfileAvatar, ProviderProfileStatistics, ProviderQuota, ProviderQuotaRequest, - ProviderResetCreditResult, ProviderResetCredits, ProviderSubscription, explicit_plan_type, + ProviderProfileAvatar, ProviderProfileStatistics, ProviderQuota, ProviderQuotaReadRequest, + ProviderQuotaRequest, ProviderResetCreditResult, ProviderResetCredits, ProviderSubscription, + explicit_plan_type, }; use crate::model::{ provider_credentials::{ProviderDocument, ProviderQuotaWindow}, @@ -289,6 +290,25 @@ pub trait ProviderAdmin: Send + Sync { request: ProviderQuotaRequest, ) -> Result; + /// 只读投影当前页额度，返回值与输入逐项对应，单个失败不影响其他账号 + /// 没有批量存储能力的 Provider 继续使用原有单账号语义 + async fn read_quotas( + &self, + requests: &[ProviderQuotaReadRequest<'_>], + ) -> Vec> { + futures::future::join_all(requests.iter().map(|request| async { + let account_id = ProviderAccountId::new(request.account.id.clone()) + .map_err(|_| ProviderAdminError::new(ProviderAdminErrorKind::Invalid))?; + self.quota(ProviderQuotaRequest { + account_id, + refresh: false, + rolling_usage: request.rolling_usage.cloned(), + }) + .await + })) + .await + } + /// 历史观测的协议字段仅由具体 Provider 解释；不支持时保留累计估算 fn quota_forecast_observation( &self, diff --git a/backend/crates/gateway-admin/src/ports/store.rs b/backend/crates/gateway-admin/src/ports/store.rs index a8127e8c4..840dcbdc9 100644 --- a/backend/crates/gateway-admin/src/ports/store.rs +++ b/backend/crates/gateway-admin/src/ports/store.rs @@ -30,7 +30,7 @@ use crate::model::{ DashboardObservation, DashboardQuery, DiagnosticDimension, DiagnosticsObservation, Granularity, OpsErrorPage, OpsErrorQuery, RequestMetricPoint, TimeRange, UsageBillingProjection, UsageCalculatedBillingFact, UsageDetail, UsageFilter, - UsageInsightsObservation, UsageOverview, UsagePage, UsageQuery, + UsageInsightsObservation, UsagePage, UsageQuery, UsageSummaryObservation, }, provider_credentials::{ AuthorizationCommit, CredentialDetails, CredentialImportCommit, CredentialImportResult, @@ -485,7 +485,7 @@ pub trait ObservabilityStore: Send + Sync { &self, range: TimeRange, filter: UsageFilter, - ) -> AdminStoreResult; + ) -> AdminStoreResult; async fn usage_diagnostics( &self, diff --git a/backend/crates/gateway-admin/src/use_case/accounts/mod.rs b/backend/crates/gateway-admin/src/use_case/accounts/mod.rs index 6971d757d..df25f857d 100644 --- a/backend/crates/gateway-admin/src/use_case/accounts/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/accounts/mod.rs @@ -30,8 +30,9 @@ use crate::{ AccountDirectoryItem, AccountDirectoryPage, AccountExportBundle, AccountPersonalInfo, AccountRefreshResult, AccountUsagePeriod, ConsumeProviderResetCredit, PrepareCredentialRefresh, ProviderModelCatalogDocument, ProviderModels, - ProviderProfileAvatar, ProviderQuota, ProviderQuotaRequest, ProviderQuotaWindow, - ProviderResetCreditResult, ProviderResetCredits, QuotaLocalUsageAttribution, + ProviderProfileAvatar, ProviderQuota, ProviderQuotaReadRequest, ProviderQuotaRequest, + ProviderQuotaWindow, ProviderResetCreditResult, ProviderResetCredits, + QuotaLocalUsageAttribution, }, quota_forecast::{ AccountQuotaForecastReport, account_quota_forecasts, quota_forecast_source_window, @@ -236,6 +237,78 @@ impl DefaultAccountsService { ) } + async fn read_quotas( + providers: &ProviderAdminRegistry, + items: &[AccountPageItem], + rolling_usage: &BTreeMap, + ) -> Vec { + let mut groups = BTreeMap::<_, Vec>::new(); + for (index, item) in items.iter().enumerate() { + groups + .entry(&item.account.provider_kind) + .or_default() + .push(index); + } + let batches = groups.into_iter().map(|(kind, indices)| async move { + let provider = match providers.require(kind) { + Ok(provider) => provider, + Err(error) => { + tracing::warn!( + provider = %kind, + error = %error, + "account directory provider is not registered; showing empty quota" + ); + return Vec::new(); + } + }; + let requests = indices + .iter() + .map(|&index| { + let account = &items[index].account; + ProviderQuotaReadRequest { + account, + rolling_usage: rolling_usage.get(&account.id), + } + }) + .collect::>(); + let results = provider.read_quotas(&requests).await; + if results.len() != indices.len() { + tracing::warn!( + provider = %kind, + expected = indices.len(), + actual = results.len(), + "account directory quota batch has invalid length; showing empty quota" + ); + return Vec::new(); + } + indices + .into_iter() + .zip(results) + .map(|(index, result)| { + let quota = result.unwrap_or_else(|error| { + tracing::warn!( + account_id = %items[index].account.id, + error = %error, + "account directory quota projection failed; showing empty quota" + ); + empty_quota() + }); + (index, quota) + }) + .collect::>() + }); + // Provider 可独立完成，回填原页顺序并保留未注册或失败账号的位置 + let mut quotas = vec![empty_quota(); items.len()]; + for (index, quota) in futures::future::join_all(batches) + .await + .into_iter() + .flatten() + { + quotas[index] = quota; + } + quotas + } + async fn load_account( &self, account_id: &ProviderAccountId, @@ -487,45 +560,7 @@ impl AccountsService for DefaultAccountsService { .into_iter() .map(|usage| (usage.account_id.clone(), usage)) .collect::>(); - let mut quotas = futures::future::join_all(page.items.iter().map(|item| async { - let account = &item.account; - let account_id = ProviderAccountId::new(account.id.clone()) - .map_err(|_| AdminError::invalid("Provider 账号 ID 不合法"))?; - // 单个账号的 quota 投影失败（Provider 未注册或 quota 读取失败）不拖垮整页： - // 该账号降级为空额度投影，其余账号与页面状态照常返回 - let provider = match providers.require(&account.provider_kind) { - Ok(provider) => provider, - Err(error) => { - tracing::warn!( - account_id = %account.id, - error = %error, - "account directory provider is not registered; showing empty quota" - ); - return Ok(empty_quota()); - } - }; - match provider - .quota(ProviderQuotaRequest { - account_id, - refresh: false, - rolling_usage: rolling_usage.get(&account.id).cloned(), - }) - .await - { - Ok(quota) => Ok(quota), - Err(error) => { - tracing::warn!( - account_id = %account.id, - error = %error, - "account directory quota projection failed; showing empty quota" - ); - Ok(empty_quota()) - } - } - })) - .await - .into_iter() - .collect::, AdminError>>()?; + let mut quotas = Self::read_quotas(providers, &page.items, &rolling_usage).await; self.attach_quota_local_usage(&page.items, &mut quotas) .await?; let mut api_key_usage = self.load_api_key_usage(&page.items).await?; diff --git a/backend/crates/gateway-admin/src/use_case/key_usage.rs b/backend/crates/gateway-admin/src/use_case/key_usage.rs index de65dbfaf..117171627 100644 --- a/backend/crates/gateway-admin/src/use_case/key_usage.rs +++ b/backend/crates/gateway-admin/src/use_case/key_usage.rs @@ -106,7 +106,7 @@ impl DefaultKeyUsageService { fn usage_filter(id: &ClientApiKeyId, model: Option) -> UsageFilter { UsageFilter { client_api_key_ref: Some(id.as_str().to_owned()), - model, + model_contains: model, ..UsageFilter::default() } } diff --git a/backend/crates/gateway-admin/src/use_case/observability/mod.rs b/backend/crates/gateway-admin/src/use_case/observability/mod.rs index 4fad8294e..818332de9 100644 --- a/backend/crates/gateway-admin/src/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/observability/mod.rs @@ -267,17 +267,17 @@ impl ObservabilityService for DefaultObservabilityService { range: TimeRange, filter: UsageFilter, ) -> Result { - let overview = self + let observation = self .store .usage_summary(range, filter) .await .map_err(|error| map_store_error(error, "usage summary"))?; let average_latency_ms = average( - overview.requests.latency_sum_ms, - overview.requests.latency_count, + observation.requests.latency_sum_ms, + observation.requests.latency_count, ); Ok(UsageSummary { - overview, + observation, average_latency_ms, }) } diff --git a/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs b/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs index b8828179b..cd07a5fde 100644 --- a/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs +++ b/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs @@ -41,7 +41,7 @@ impl AnalyticsValue for DashboardObservation { &mut self.analytics } } -impl AnalyticsValue for UsageOverview { +impl AnalyticsValue for UsageSummaryObservation { fn metadata(&mut self) -> &mut AnalyticsMetadata { &mut self.analytics } @@ -177,7 +177,7 @@ type UsageKey = (TimeRange, UsageFilter); pub(crate) struct CachedObservabilityStore { inner: Arc, dashboard: Cache, - summaries: Cache, + summaries: Cache, insights: Cache<(UsageKey, Granularity), UsageInsightsObservation>, diagnostics: Cache<(UsageKey, DiagnosticDimension, u16), DiagnosticsObservation>, } @@ -254,7 +254,7 @@ impl ObservabilityStore for CachedObservabilityStore { &self, range: TimeRange, filter: UsageFilter, - ) -> AdminStoreResult { + ) -> AdminStoreResult { self.summaries .get((range, filter.clone()), self.inner.as_ref(), TTL, || { self.inner.usage_summary(range, filter) diff --git a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs index a20ecebf2..3838ed63a 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs @@ -2098,11 +2098,23 @@ async fn accounts_list_should_degrade_quota_failure_to_empty_window_without_drop provider_data: None, }); let failing = FakeProviderAdmin::new("xai", events.clone()); + let mut xai_quota = failing.quota.lock().expect("provider quota").clone(); + xai_quota.observed_at = Some(Utc::now()); + failing.set_quota(xai_quota); failing.fail_next_quota(ProviderAdminErrorKind::Invalid); let store = FakeAccountStore::new("openai", events.clone()); let mut xai_account = account_record("xai"); xai_account.id = "acct_bad_xai".to_owned(); - store.set_accounts(vec![account_record("openai"), xai_account]); + let mut healthy_xai = xai_account.clone(); + healthy_xai.id = "acct_healthy_xai".to_owned(); + let mut second_openai = account_record("openai"); + second_openai.id = "acct_second_openai".to_owned(); + store.set_accounts(vec![ + account_record("openai"), + xai_account, + second_openai, + healthy_xai, + ]); let services = super::AdminHarness::new() .accounts(store.clone()) .settings(Arc::new(StaticSettingsStore::default())) @@ -2126,7 +2138,18 @@ async fn accounts_list_should_degrade_quota_failure_to_empty_window_without_drop .await .expect("one failing quota must not fail the directory page"); - assert_eq!(page.items.len(), 2); + assert_eq!( + page.items + .iter() + .map(|item| item.account.id.as_str()) + .collect::>(), + [ + "acct_test", + "acct_bad_xai", + "acct_second_openai", + "acct_healthy_xai" + ] + ); let by_id = page .items .iter() @@ -2139,7 +2162,9 @@ async fn accounts_list_should_degrade_quota_failure_to_empty_window_without_drop assert!(degraded.quota.windows.is_empty()); assert!(degraded.quota.observed_at.is_none()); assert!(degraded.quota.provider_data.is_none()); - assert_eq!(page.summary.total, 2); + assert!(by_id["acct_healthy_xai"].quota.observed_at.is_some()); + assert_eq!(by_id["acct_second_openai"].quota.windows.len(), 1); + assert_eq!(page.summary.total, 4); } #[tokio::test] diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index bb45b5806..c28fb035a 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -49,7 +49,7 @@ use gateway_admin::{ DashboardDesktopRelease, DashboardObservation, DashboardQuery, DashboardWireAttribute, DashboardWireProfile, DashboardWireTarget, DesktopReleaseStatus, DiagnosticDimension, DiagnosticsObservation, Granularity, OpsErrorPage, OpsErrorQuery, RequestMetricPoint, - TimeRange, UsageDetail, UsageFilter, UsageOverview, UsagePage, UsageQuery, + TimeRange, UsageDetail, UsageFilter, UsagePage, UsageQuery, UsageSummaryObservation, }, provider_credentials::{ AuthorizationCommit, AuthorizationStarted, CompleteAuthorization, CredentialDetails, @@ -833,7 +833,11 @@ impl ObservabilityStore for UnavailableStore { Err(unavailable("usage detail")) } - async fn usage_summary(&self, _: TimeRange, _: UsageFilter) -> AdminStoreResult { + async fn usage_summary( + &self, + _: TimeRange, + _: UsageFilter, + ) -> AdminStoreResult { Err(unavailable("usage summary")) } diff --git a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs index b9b0d8a02..f55ad8dc9 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs @@ -23,7 +23,7 @@ use gateway_admin::{ HealthStatus, LatencyPercentiles, ObservabilityPageSize, OpsErrorPage, OpsErrorQuery, PercentileMilliseconds, RequestMetricPoint, RequestMetrics, TimeRange, TrendKind, UsageBilling, UsageDetail, UsageFilter, UsageInsightsObservation, UsageListRecord, - UsageOverview, UsagePage, UsageProjectedCostPoint, UsageQuery, + UsageOverview, UsagePage, UsageProjectedCostPoint, UsageQuery, UsageSummaryObservation, }, settings::{ AdminApiKey, AdminApiKeyMutation, ReplaceRuntimeSettings, RotationStrategy, @@ -1166,7 +1166,7 @@ impl ObservabilityStore for FixtureObservabilityStore { &self, range: TimeRange, filter: UsageFilter, - ) -> AdminStoreResult { + ) -> AdminStoreResult { self.usage_summary_filters .lock() .expect("summary filters") @@ -1178,9 +1178,12 @@ impl ObservabilityStore for FixtureObservabilityStore { if self.summary_fail.load(Ordering::SeqCst) { return Err(super::unavailable("projection summary failed")); } - Ok(UsageOverview { + let overview = self.overview.lock().expect("overview").clone(); + Ok(UsageSummaryObservation { range, - ..self.overview.lock().expect("overview").clone() + analytics: overview.analytics, + requests: (&overview.requests).into(), + attempts: overview.attempts, }) } @@ -1504,8 +1507,8 @@ async fn usage_summary_cache_should_coalesce_identical_queries_and_invalidate_de .usage_summary(range, filter) .await .expect("generation refresh"); - assert_eq!(refreshed.overview.analytics.generation, 1); - assert_eq!(refreshed.overview.requests.request_count, 7); + assert_eq!(refreshed.observation.analytics.generation, 1); + assert_eq!(refreshed.observation.requests.request_count, 7); assert_eq!(store.usage_summary_filters.lock().unwrap().len(), 2); } @@ -1525,13 +1528,17 @@ async fn usage_summary_cache_should_coalesce_when_generation_advances_during_the }, ); assert_eq!( - first.expect("first summary").overview.analytics.generation, + first + .expect("first summary") + .observation + .analytics + .generation, 1 ); assert_eq!( second .expect("coalesced summary") - .overview + .observation .analytics .generation, 1 diff --git a/backend/crates/gateway-api/src/admin/observability/presenter.rs b/backend/crates/gateway-api/src/admin/observability/presenter.rs index 9035ef183..8d5d813af 100644 --- a/backend/crates/gateway-api/src/admin/observability/presenter.rs +++ b/backend/crates/gateway-api/src/admin/observability/presenter.rs @@ -49,7 +49,7 @@ pub(crate) fn outcome_name(outcome: &domain::RequestOutcome) -> &str { outcome.as_str() } -pub(crate) fn request_metrics_view(metrics: &domain::RequestMetrics) -> RequestMetricsView { +fn request_summary_metrics_view(metrics: &domain::RequestSummaryMetrics) -> RequestMetricsView { RequestMetricsView { request_count: metrics.request_count, success_count: metrics.success_count, @@ -1050,7 +1050,7 @@ pub(crate) fn usage_summary_view( summary: domain::UsageSummary, time: crate::time::TimePresenter, ) -> UsageSummaryView { - let overview = summary.overview; + let overview = summary.observation; UsageSummaryView { analytics: analytics_view(overview.analytics, time), total_requests: format_compact_number(overview.requests.request_count), @@ -1060,7 +1060,7 @@ pub(crate) fn usage_summary_view( cache_write_input_tokens: format_compact_number(overview.requests.cache_write_input_tokens), total_tokens: format_compact_number(overview.requests.total_tokens), average_latency_ms: display_duration(summary.average_latency_ms), - logical_requests: request_metrics_view(&overview.requests), + logical_requests: request_summary_metrics_view(&overview.requests), attempts: attempt_metrics_view(&overview.attempts), } } diff --git a/backend/crates/gateway-api/src/admin/observability/query.rs b/backend/crates/gateway-api/src/admin/observability/query.rs index 065df1a89..6319fd8f0 100644 --- a/backend/crates/gateway-api/src/admin/observability/query.rs +++ b/backend/crates/gateway-api/src/admin/observability/query.rs @@ -25,27 +25,28 @@ impl DashboardQuery { } } -/// 逻辑请求列表、汇总和洞察查询参数 +/// 请求明细的字段筛选、时间范围与分页参数 #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct UsageQuery { + pub client_api_key_name: Option, + pub account_name: Option, pub status: Option, pub current_page: Option, pub page_size: Option, - pub kind: Option, - pub outcome: Option, - pub client_api_key_id: Option, pub provider: Option, - pub request_id: Option, - pub account_id: Option, - pub route: Option, pub model: Option, - pub status_code: Option, - pub transport: Option, - pub attempt_index: Option, - pub response_id: Option, - pub upstream_request_id: Option, - pub search: Option, + pub period: Option, + pub as_of: Option, + pub start_time: Option, + pub end_time: Option, +} + +/// 用量汇总与洞察只按时间和 Provider 查询 +#[derive(Clone, Default, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct UsageAnalyticsQuery { + pub provider: Option, pub period: Option, pub as_of: Option, pub start_time: Option, @@ -258,16 +259,6 @@ pub(crate) fn domain_diagnostic_dimension( } } -pub(crate) fn request_outcome( - value: Option, -) -> Result, WireValidationError> { - non_empty(value) - .map(|value| { - domain::RequestOutcome::new(value).map_err(|_| WireValidationError::new("outcome")) - }) - .transpose() -} - pub(crate) fn usage_range( start: Option<&str>, end: Option<&str>, @@ -288,39 +279,14 @@ pub(crate) fn dashboard_today_range( crate::time::query_range(start, end, period, as_of, timezone, "today") } -pub(crate) fn usage_filter(query: &UsageQuery) -> Result { - let outcome = query.outcome.clone().or_else(|| { - non_empty(query.kind.clone()).filter(|value| { - matches!( - value.as_str(), - "running" | "succeeded" | "failed" | "cancelled" | "incomplete" - ) - }) - }); - Ok(domain::UsageFilter { - client_api_key_ref: non_empty(query.client_api_key_id.clone()), - request_id: non_empty(query.request_id.clone()), - provider_account_ref: non_empty(query.account_id.clone()), - operation: non_empty(query.route.clone()), +pub(crate) fn usage_filter(query: &UsageQuery) -> domain::UsageFilter { + domain::UsageFilter { + client_api_key_name: non_empty(query.client_api_key_name.clone()), + provider_account_name: non_empty(query.account_name.clone()), provider_kind: non_empty(query.provider.clone()), - model: non_empty(query.model.clone()), - outcome: request_outcome(outcome)?, - status_code: parse_status(query.status_code)?, - transport: non_empty(query.transport.clone()), - attempt_index: parse_attempt_index(query.attempt_index)?, - response_id: non_empty(query.response_id.clone()), - upstream_request_id: non_empty(query.upstream_request_id.clone()), - search: non_empty(query.search.clone()), - }) -} - -pub(crate) fn analytics_filter( - query: &UsageQuery, -) -> Result { - if query.search.is_some() { - return Err(WireValidationError::new("search")); + model_contains: non_empty(query.model.clone()), + ..Default::default() } - usage_filter(query) } pub(crate) fn usage_command( @@ -344,7 +310,7 @@ pub(crate) fn usage_command( query.as_of, timezone, )?, - filter: usage_filter(query)?, + filter: usage_filter(query), current_page, page_size: page_size_value, }) diff --git a/backend/crates/gateway-api/src/admin/observability/routes.rs b/backend/crates/gateway-api/src/admin/observability/routes.rs index 1d42f31cf..902a853ee 100644 --- a/backend/crates/gateway-api/src/admin/observability/routes.rs +++ b/backend/crates/gateway-api/src/admin/observability/routes.rs @@ -142,7 +142,7 @@ where pub(crate) async fn usage_records_summary( _auth: AdminAuth, State(state): State, - AdminQuery(query): AdminQuery, + AdminQuery(query): AdminQuery, ) -> Result where S: SessionState + Send + Sync, @@ -155,7 +155,10 @@ where state.admin_services().timezone(), ) .map_err(map_wire_error)?; - let filter = analytics_filter(&query).map_err(map_wire_error)?; + let filter = domain::UsageFilter { + provider_kind: non_empty(query.provider), + ..Default::default() + }; let result = state .admin_services() .observability() @@ -174,7 +177,7 @@ where pub(crate) async fn usage_insights_overview( _auth: AdminAuth, State(state): State, - AdminQuery(query): AdminQuery, + AdminQuery(query): AdminQuery, ) -> Result where S: SessionState + Send + Sync, @@ -188,7 +191,10 @@ where state.admin_services().timezone(), ) .map_err(map_wire_error)?; - let filter = analytics_filter(&query).map_err(map_wire_error)?; + let filter = domain::UsageFilter { + provider_kind: non_empty(query.provider), + ..Default::default() + }; let result = state .admin_services() .observability() diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 7869a8616..181872666 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -41,7 +41,7 @@ use gateway_admin::{ DashboardObservation, DashboardQuery, DecimalAmount, DiagnosticDimension, DiagnosticsObservation, Granularity, OpsError, OpsErrorPage, OpsErrorQuery, RequestMetricPoint, TimeRange, UsageDetail, UsageFilter, UsageListRecord, - UsageOverview, UsagePage, UsageQuery, + UsageOverview, UsagePage, UsageQuery, UsageSummaryObservation, }, provider_credentials::{ AuthorizationCommit, AuthorizationStarted, CompleteAuthorization, CredentialDetails, @@ -1547,12 +1547,17 @@ impl ObservabilityStore for UnusedStore { &self, range: TimeRange, filter: UsageFilter, - ) -> AdminStoreResult { + ) -> AdminStoreResult { let mut data = self.observations.lock().expect("observations"); data.summaries.push((range, filter)); data.summary .clone() - .map(|summary| UsageOverview { range, ..summary }) + .map(|summary| UsageSummaryObservation { + range, + analytics: summary.analytics, + requests: (&summary.requests).into(), + attempts: summary.attempts, + }) .ok_or_else(|| unavailable("usage summary")) } diff --git a/backend/crates/gateway-api/tests/admin/observability/query.rs b/backend/crates/gateway-api/tests/admin/observability/query.rs index b3cf8a680..214317fb6 100644 --- a/backend/crates/gateway-api/tests/admin/observability/query.rs +++ b/backend/crates/gateway-api/tests/admin/observability/query.rs @@ -1,8 +1,8 @@ //! 查询 wire 校验的固定合同测试 use gateway_api::admin::observability::{ - DashboardQuery, DiagnosticDimension, DiagnosticsQuery, OpsQuery, TrendKind, UsageQuery, - parse_attempt_index, parse_datetime, parse_status, + DashboardQuery, DiagnosticDimension, DiagnosticsQuery, OpsQuery, TrendKind, + UsageAnalyticsQuery, UsageQuery, parse_attempt_index, parse_datetime, parse_status, }; use serde_json::json; @@ -33,6 +33,7 @@ fn observability_queries_should_reject_unknown_fields() { let query = json!({"unknownFilter": true}); assert!(serde_json::from_value::(query.clone()).is_err()); assert!(serde_json::from_value::(query.clone()).is_err()); + assert!(serde_json::from_value::(query.clone()).is_err()); assert!(serde_json::from_value::(query.clone()).is_err()); assert!(serde_json::from_value::(query).is_err()); } diff --git a/backend/crates/gateway-api/tests/admin/observability/response.rs b/backend/crates/gateway-api/tests/admin/observability/response.rs index bea75a206..c4343041c 100644 --- a/backend/crates/gateway-api/tests/admin/observability/response.rs +++ b/backend/crates/gateway-api/tests/admin/observability/response.rs @@ -3,6 +3,100 @@ use chrono::Utc; use serde_json::json; +#[tokio::test] +async fn usage_summary_keeps_request_attempt_cost_and_average_latency_wire_fields() { + use crate::admin::{AdminTestFixture, AdminTestState}; + use axum::{ + body::{Body, to_bytes}, + http::{Request, StatusCode, header}, + }; + use chrono::Duration; + use gateway_admin::model::observability::{ + AttemptMetrics, CostCoverage, CurrencyCost, RequestMetrics, TimeRange, UsageOverview, + }; + use gateway_api::admin::observability; + use tower::ServiceExt as _; + + for (sum, count, display) in [(404, 3, "134 ms"), (0, 1, "0 ms"), (0, 0, "—")] { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let now = Utc::now(); + fixture.observations.lock().unwrap().summary = Some(UsageOverview { + analytics: gateway_admin::model::observability::AnalyticsMetadata { + generated_at: now, + ..Default::default() + }, + range: TimeRange::new(now - Duration::hours(1), now).unwrap(), + requests: RequestMetrics { + request_count: 5, + success_count: 4, + cancelled_count: 1, + caller_error_count: 1, + input_tokens: 400, + output_tokens: 80, + cached_input_tokens: 30, + cache_write_input_tokens: 7, + reasoning_output_tokens: 9, + total_tokens: 480, + latency_sum_ms: sum, + latency_count: count, + ..Default::default() + }, + attempts: AttemptMetrics { + attempt_count: 6, + success_count: 4, + failure_count: 1, + cancelled_count: 1, + rate_limited_count: 1, + cost_coverage: CostCoverage { + provider_reported_count: 3, + unavailable_count: 1, + ..Default::default() + }, + costs: vec![CurrencyCost { + currency: "USD".into(), + amount: "1.25".parse().unwrap(), + }], + ..Default::default() + }, + providers: Vec::new(), + }); + let response = observability::router::() + .with_state(fixture.state()) + .oneshot( + Request::builder() + .uri("/api/admin/usage/records/summary?provider=openai") + .header(header::COOKIE, "cpr_session=valid-session") + .header("x-request-id", "req_usage_summary_projection") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), 1024 * 1024).await.unwrap(); + let value: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(status, StatusCode::OK, "{value}"); + assert_eq!( + value["data"], + json!({ + "analytics": value["data"]["analytics"], + "totalRequests": "5", "inputTokens": "400", "outputTokens": "80", + "cachedInputTokens": "30", "cacheWriteInputTokens": "7", "totalTokens": "480", + "averageLatencyMs": display, + "logicalRequests": {"requestCount":5,"successCount":4,"failureCount":0,"cancelledCount":1,"incompleteCount":0,"callerErrorCount":1,"inputTokens":400,"outputTokens":80,"cachedInputTokens":30,"cacheWriteInputTokens":7,"reasoningOutputTokens":9,"totalTokens":480}, + "attempts": {"attemptCount":6,"successCount":4,"failureCount":1,"cancelledCount":1,"incompleteCount":0,"rateLimitedCount":1,"authFailureCount":0,"provider5xxCount":0,"costCoverage":{"known":3,"partial":0,"unknown":1,"notBillable":0},"costs":[{"currency":"USD","estimatedAmount":"1.25"}]} + }) + ); + let observations = fixture.observations.lock().unwrap(); + assert_eq!(observations.summaries.len(), 1); + assert_eq!( + observations.summaries[0].1.provider_kind.as_deref(), + Some("openai") + ); + } +} + #[tokio::test] async fn dashboard_summary_should_include_quota_exhaustion_in_unavailable_headline_count() { use axum::{ @@ -725,7 +819,7 @@ pub(super) fn usage_record_with_account( } #[tokio::test] -async fn usage_route_should_forward_a_bounded_unknown_outcome_filter() { +async fn usage_route_should_forward_name_and_model_contains_filters() { use axum::{ body::Body, http::{Request, StatusCode, header}, @@ -741,7 +835,7 @@ async fn usage_route_should_forward_a_bounded_unknown_outcome_filter() { .with_state(fixture.state()) .oneshot( Request::builder() - .uri("/api/admin/usage/records?outcome=provider_future_state") + .uri("/api/admin/usage/records?model=%20coding%20&clientApiKeyName=%20Key%20%25%20&accountName=%20Account%20Name%20") .header(header::COOKIE, "cpr_session=valid-session") .header("x-request-id", "req_usage_other_outcome") .body(Body::empty()) @@ -751,6 +845,16 @@ async fn usage_route_should_forward_a_bounded_unknown_outcome_filter() { .expect("usage response"); assert_eq!(response.status(), StatusCode::OK); + let observations = fixture.observations.lock().unwrap(); + let filter = &observations.records.last().expect("usage query").filter; + assert_eq!(filter.client_api_key_name.as_deref(), Some("Key %")); + assert_eq!(filter.model_contains.as_deref(), Some("coding")); + assert_eq!( + filter.provider_account_name.as_deref(), + Some("Account Name") + ); + assert!(filter.client_api_key_ref.is_none()); + assert!(filter.provider_account_ref.is_none()); } #[tokio::test] diff --git a/backend/crates/gateway-api/tests/key_usage/mod.rs b/backend/crates/gateway-api/tests/key_usage/mod.rs index 3044cd112..0856ec83f 100644 --- a/backend/crates/gateway-api/tests/key_usage/mod.rs +++ b/backend/crates/gateway-api/tests/key_usage/mod.rs @@ -228,11 +228,14 @@ async fn overview_scopes_every_query_and_projects_only_key_visible_fields() { assert!(filter.provider_account_ref.is_none()); assert!(filter.provider_kind.is_none()); } - assert_eq!(observations.summaries[0].1.model.as_deref(), Some("coding")); + assert_eq!( + observations.summaries[0].1.model_contains.as_deref(), + Some("coding") + ); let health = observations .trends .iter() - .find(|(_, filter)| filter.model.is_none()) + .find(|(_, filter)| filter.model_contains.is_none()) .unwrap(); assert_eq!( health.0.start, @@ -346,7 +349,7 @@ async fn records_keep_pagination_and_hide_admin_and_upstream_data() { ]) { assert_eq!(query.status, status); assert_eq!(query.filter.client_api_key_ref.as_deref(), Some("key-42")); - assert_eq!(query.filter.model.as_deref(), Some("coding")); + assert_eq!(query.filter.model_contains.as_deref(), Some("coding")); assert!(query.filter.provider_account_ref.is_none()); assert!(query.filter.provider_kind.is_none()); } diff --git a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs index e250e4400..0e5161ce3 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -355,16 +355,16 @@ impl PgObservabilityRepository { &self, range: ObservabilityRange, filter: UsageRecordFilter, - ) -> StoreResult { + ) -> StoreResult { self.query_budget .run( &self.pool, "load usage analytics", |mut connection| async move { let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?; - let overview = snapshot.overview(range, &filter).await?; + let summary = snapshot.summary(range, &filter).await?; snapshot.finish().await?; - Ok(overview) + Ok(summary) }, ) .await @@ -536,8 +536,8 @@ impl AdminObservabilityStore for PgAdminObservabilityStore { &self, range: admin_observability::TimeRange, filter: admin_observability::UsageFilter, - ) -> AdminStoreResult { - let overview = self + ) -> AdminStoreResult { + let summary = self .repository .usage_summary( store_range(range).map_err(observability_error)?, @@ -545,7 +545,12 @@ impl AdminObservabilityStore for PgAdminObservabilityStore { ) .await .map_err(observability_error)?; - Ok(admin_usage_overview(overview)) + Ok(admin_observability::UsageSummaryObservation { + analytics: summary.analytics, + range: admin_range(summary.range), + requests: summary.requests, + attempts: admin_attempt_metrics(summary.attempts), + }) } async fn usage_diagnostics( diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/accounts.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/accounts.rs index 023519a2d..df64e3b66 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/accounts.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/accounts.rs @@ -51,7 +51,7 @@ impl AnalyticsSnapshot<'_> { timezone: Default::default(), account_models: Some(by_model), selection: super::query::ReadSelection { - distributions: false, + distributions: super::query::DistributionRead::None, costs: true, keys: Some(account_ids), counters: None, diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/aggregate.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/aggregate.rs index 4a26fbf6c..c076b39f7 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/aggregate.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/aggregate.rs @@ -9,28 +9,47 @@ pub(super) fn push_fact_aggregation( exact_percentiles: bool, read_costs: bool, ) { - push_fact_aggregation_selected(query, exact_percentiles, read_costs, true, None, None); + push_fact_aggregation_selected( + query, + exact_percentiles, + read_costs, + DistributionRead::Full, + None, + None, + ); } pub(super) fn push_fact_aggregation_selected( query: &mut QueryBuilder, exact_percentiles: bool, read_costs: bool, - read_distributions: bool, + distributions: DistributionRead, counter_names: Option<&[&str]>, distribution_names: Option<&[&str]>, ) { + if read_costs && matches!(distributions, DistributionRead::Statistics) { + push_fact_statistics_by_currency(query, counter_names, distribution_names); + return; + } + let read_distributions = !matches!(distributions, DistributionRead::None); + let statistics_only = matches!(distributions, DistributionRead::Statistics); let counters = selected_counter_expressions(counter_names); query.push(", counted as (select output_bucket,name,max(started_at) filter(where is_completed_usage) as last_used_at,"); push_counter_object(query, &counters); query.push(" as counts"); - if exact_percentiles && read_distributions { + if read_distributions && (exact_percentiles || statistics_only) { query.push(","); - push_exact_distributions(query, distribution_names); + push_distribution_statistics( + query, + distribution_names, + exact_percentiles && !statistics_only, + ); } query.push(" from matching group by output_bucket,name)"); if read_costs { - query.push(", cost_values as (select output_bucket,name,currency,sum(amount)::text as amount from matching cross join lateral (values (cost_currency,cost_amount),('standard_usd',standard_cost_amount),('no_cache_usd',no_cache_cost_amount)) cost(currency,amount) where is_completed_usage and currency is not null and amount is not null group by output_bucket,name,currency), merged_costs as (select output_bucket,name,jsonb_object_agg(currency,amount) as costs from cost_values group by output_bucket,name)"); + // 先按原币种合并金额，再展开费用场景，避免每条事实放大成三条聚合输入 + // 空币种仍参与场景费用，金额为空的实际费用继续在展开后排除 + query.push(", cost_groups as (select output_bucket,name,cost_currency,sum(cost_amount) as cost_amount,sum(standard_cost_amount) as standard_cost_amount,sum(no_cache_cost_amount) as no_cache_cost_amount from matching where is_completed_usage group by output_bucket,name,cost_currency), cost_values as (select output_bucket,name,currency,sum(amount)::text as amount from cost_groups cross join lateral (values (cost_currency,cost_amount),('standard_usd',standard_cost_amount),('no_cache_usd',no_cache_cost_amount)) cost(currency,amount) where currency is not null and amount is not null group by output_bucket,name,currency), merged_costs as (select output_bucket,name,jsonb_object_agg(currency,amount) as costs from cost_values group by output_bucket,name)"); } if !read_distributions { query.push(" select c.output_bucket as bucket_start,c.name,c.counts,c.last_used_at,'{}'::jsonb as distributions,'{}'::jsonb as exact_percentiles,"); @@ -45,7 +64,7 @@ pub(super) fn push_fact_aggregation_selected( } return; } - if exact_percentiles { + if exact_percentiles || statistics_only { query.push(" select c.output_bucket as bucket_start,c.name,c.counts,c.last_used_at,c.distributions,c.exact_percentiles,"); query.push(if read_costs { "coalesce(k.costs,'{}'::jsonb)" @@ -91,6 +110,30 @@ pub(super) fn push_fact_aggregation_selected( } } +fn push_fact_statistics_by_currency( + query: &mut QueryBuilder, + counter_names: Option<&[&str]>, + distribution_names: Option<&[&str]>, +) { + // 摘要的计数、延迟样本和费用共用一次事实扫描；只物化按币种缩减后的分组， + // 再沿用汇总合并规则，避免把大范围事实复制到临时结果或重复扫描 + query.push(", currency_groups as materialized (select output_bucket,name,cost_currency,max(started_at) filter(where is_completed_usage) as last_used_at,"); + push_counter_object(query, &selected_counter_expressions(counter_names)); + query.push(" as counts,"); + push_distribution_statistics(query, distribution_names, false); + query.push(",sum(cost_amount) filter(where is_completed_usage) as cost_amount,sum(standard_cost_amount) filter(where is_completed_usage) as standard_cost_amount,sum(no_cache_cost_amount) filter(where is_completed_usage) as no_cache_cost_amount from matching group by output_bucket,name,cost_currency)"); + // 空币种只排除实际费用，情景费用仍保留；同名费用项先求和，维持原有合并语义 + query.push(", currency_summaries as (select output_bucket,name,last_used_at,counts,distributions,coalesce(costs,'{}'::jsonb) as costs from currency_groups cross join lateral (select jsonb_object_agg(currency,amount::text) as costs from (select currency,sum(amount) as amount from (values (cost_currency,cost_amount),('standard_usd',standard_cost_amount),('no_cache_usd',no_cache_cost_amount)) cost(currency,amount) where currency is not null and amount is not null group by currency) merged) cost)"); + push_summary_aggregation_from( + query, + "currency_summaries", + DistributionRead::Statistics, + true, + counter_names, + distribution_names, + ); +} + // 事实聚合与汇总读取共用指标清单，防止新增计数只在其中一条路径生效 fn counter_expressions() -> Vec<(String, String)> { let mut counters = vec![ @@ -188,7 +231,11 @@ fn selected_counter_expressions(names: Option<&[&str]>) -> Vec<(String, String)> counters } -fn push_exact_distributions(query: &mut QueryBuilder, names: Option<&[&str]>) { +fn push_distribution_statistics( + query: &mut QueryBuilder, + names: Option<&[&str]>, + exact_percentiles: bool, +) { let samples = selected_sample_expressions(names); // 独立 ordered-set 聚合只排序实际样本，避免把每条事实展开为六条记录再排序 query.push("jsonb_build_object("); @@ -198,7 +245,12 @@ fn push_exact_distributions(query: &mut QueryBuilder, names: Option<&[ } query.push(format!("'{name}',jsonb_build_object('count',count({expression}),'sum',coalesce(sum({expression}),0),'minimum',min({expression}),'maximum',max({expression}),'zero',count(*) filter(where ({expression})=0),'bins','{{}}'::jsonb)")); } - query.push(") as distributions,jsonb_build_object("); + query.push(") as distributions,"); + if !exact_percentiles { + query.push("'{}'::jsonb as exact_percentiles"); + return; + } + query.push("jsonb_build_object("); for (index, (name, expression)) in samples.iter().enumerate() { if index > 0 { query.push(","); @@ -264,11 +316,31 @@ fn push_counter_object(query: &mut QueryBuilder, counters: &[(String, /// 输入 matching 的行来自互斥时间桶，合并后每个最终输出组只返回一行 pub(super) fn push_summary_aggregation( query: &mut QueryBuilder, - distributions: bool, + distribution_read: DistributionRead, costs: bool, counter_names: Option<&[&str]>, distribution_names: Option<&[&str]>, ) { + push_summary_aggregation_from( + query, + "matching", + distribution_read, + costs, + counter_names, + distribution_names, + ); +} + +// 来源只由本模块提供静态 CTE 名称，互斥时间桶和互斥币种分组共用合并规则 +fn push_summary_aggregation_from( + query: &mut QueryBuilder, + source: &'static str, + distribution_read: DistributionRead, + costs: bool, + counter_names: Option<&[&str]>, + distribution_names: Option<&[&str]>, +) { + let distributions = !matches!(distribution_read, DistributionRead::None); // 每个源汇总只展开一次固定指标，避免按指标键放大行数再排序 let counters = selected_counter_expressions(counter_names) .into_iter() @@ -279,7 +351,10 @@ pub(super) fn push_summary_aggregation( .collect::>(); query.push(", grouped as (select output_bucket,name,max(last_used_at) as last_used_at,"); push_counter_object(query, &counters); - query.push(" as counts from matching cross join lateral jsonb_to_record(counts) as metric("); + query + .push(" as counts from ") + .push(source) + .push(" cross join lateral jsonb_to_record(counts) as metric("); query.push( counters .iter() @@ -289,11 +364,17 @@ pub(super) fn push_summary_aggregation( ); query.push(") group by output_bucket,name)"); if costs { - query.push(", cost_values as (select output_bucket,name,c.key,sum(c.value::numeric)::text as value from matching cross join lateral jsonb_each_text(costs) c group by output_bucket,name,c.key), merged_costs as (select output_bucket,name,jsonb_object_agg(key,value) as costs from cost_values group by output_bucket,name)"); + query + .push(", cost_values as (select output_bucket,name,c.key,sum(c.value::numeric)::text as value from ") + .push(source) + .push(" cross join lateral jsonb_each_text(costs) c group by output_bucket,name,c.key), merged_costs as (select output_bucket,name,jsonb_object_agg(key,value) as costs from cost_values group by output_bucket,name)"); } if distributions { // 先按最终输出组归并分布，再在每组内合并直方图键，避免全局展开后大排序 - query.push(", samples as (select output_bucket,name,d.key as metric,d.value from matching cross join lateral jsonb_each(distributions) d"); + query + .push(", samples as (select output_bucket,name,d.key as metric,d.value from ") + .push(source) + .push(" cross join lateral jsonb_each(distributions) d"); if let Some(names) = distribution_names { query .push(" where d.key=any(") @@ -305,7 +386,13 @@ pub(super) fn push_summary_aggregation( ) .push("::text[])"); } - query.push("), sample_stats as (select output_bucket,name,metric,sum((value->>'count')::numeric) as count,sum((value->>'sum')::double precision) as sum,min((value->>'minimum')::double precision) as minimum,max((value->>'maximum')::double precision) as maximum,sum((value->>'zero')::numeric) as zero,array_agg(value->'bins') as bins from samples group by output_bucket,name,metric), merged_distributions as (select s.output_bucket,s.name,jsonb_object_agg(s.metric,jsonb_build_object('count',s.count,'sum',s.sum,'minimum',s.minimum,'maximum',s.maximum,'zero',s.zero,'bins',coalesce(b.bins,'{}'::jsonb))) as distributions from sample_stats s cross join lateral(select jsonb_object_agg(key,count) as bins from(select b.key,sum(b.value::numeric) as count from unnest(s.bins) input(value) cross join lateral jsonb_each_text(input.value) b group by b.key) merged)b group by s.output_bucket,s.name)"); + query.push("), sample_stats as (select output_bucket,name,metric,sum((value->>'count')::numeric) as count,sum((value->>'sum')::double precision) as sum,min((value->>'minimum')::double precision) as minimum,max((value->>'maximum')::double precision) as maximum,sum((value->>'zero')::numeric) as zero"); + if matches!(distribution_read, DistributionRead::Full) { + query.push(",array_agg(value->'bins') as bins from samples group by output_bucket,name,metric), merged_distributions as (select s.output_bucket,s.name,jsonb_object_agg(s.metric,jsonb_build_object('count',s.count,'sum',s.sum,'minimum',s.minimum,'maximum',s.maximum,'zero',s.zero,'bins',coalesce(b.bins,'{}'::jsonb))) as distributions from sample_stats s cross join lateral(select jsonb_object_agg(key,count) as bins from(select b.key,sum(b.value::numeric) as count from unnest(s.bins) input(value) cross join lateral jsonb_each_text(input.value) b group by b.key) merged)b group by s.output_bucket,s.name)"); + } else { + // 只合并均值所需的样本统计，已发布直方图的桶不参与摘要读取 + query.push(" from samples group by output_bucket,name,metric), merged_distributions as (select output_bucket,name,jsonb_object_agg(metric,jsonb_build_object('count',count,'sum',sum,'minimum',minimum,'maximum',maximum,'zero',zero,'bins','{}'::jsonb)) as distributions from sample_stats group by output_bucket,name)"); + } } query.push(" select g.output_bucket as bucket_start,g.name,g.last_used_at,g.counts,"); query.push(if costs { diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/contributions.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/contributions.rs index 479cb3052..27b9c578d 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/contributions.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/contributions.rs @@ -37,6 +37,9 @@ impl AnalyticsSnapshot<'_> { || filter.status_code.is_some() || filter.outcome.is_some() || filter.search.is_some() + || filter.model_contains.is_some() + || filter.client_api_key_name.is_some() + || filter.provider_account_name.is_some() || matches!( dimension, Some(DiagnosticDimension::Failure | DiagnosticDimension::Status) diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/metrics.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/metrics.rs index c5e40dfa8..717b276b5 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/metrics.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/metrics.rs @@ -320,6 +320,25 @@ impl MetricState { p99_ms: percentile(0.99)?, }) } + pub fn request_summary_metrics(&self) -> admin_observability::RequestSummaryMetrics { + admin_observability::RequestSummaryMetrics { + request_count: self.count("request_count"), + success_count: self.count("success_count"), + failure_count: self.count("failure_count"), + cancelled_count: self.count("cancelled_count"), + incomplete_count: self.count("incomplete_count"), + caller_error_count: self.count("caller_error_count"), + input_tokens: self.count("input_tokens"), + output_tokens: self.count("output_tokens"), + cached_input_tokens: self.count("cached_input_tokens"), + cache_write_input_tokens: self.count("cache_write_input_tokens"), + reasoning_output_tokens: self.count("reasoning_output_tokens"), + total_tokens: self.count("total_tokens"), + latency_sum_ms: self.count("latency_ms_sum"), + latency_count: self.histogram("latency_ms").count, + } + } + pub fn request_metrics(&self) -> StoreResult { let latency = self.histogram("latency_ms"); let first = self.histogram("first_token_ms"); diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs index 860f90f55..04c4e904f 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs @@ -9,9 +9,17 @@ pub(crate) struct AnalyticsSnapshot<'a> { approximate: bool, } +#[derive(Clone, Copy)] +pub(super) enum DistributionRead { + None, + // 只供窄摘要读取样本统计；不把缺少桶的状态用于百分位或持久化 + Statistics, + Full, +} + #[derive(Clone, Copy)] pub(super) struct ReadSelection<'a> { - pub(super) distributions: bool, + pub(super) distributions: DistributionRead, pub(super) costs: bool, pub(super) keys: Option<&'a [String]>, pub(super) counters: Option<&'a [&'a str]>, @@ -157,7 +165,7 @@ impl<'a> AnalyticsSnapshot<'a> { None, Default::default(), ReadSelection { - distributions: false, + distributions: DistributionRead::None, costs: false, keys: None, counters: Some(&[ @@ -174,6 +182,62 @@ impl<'a> AnalyticsSnapshot<'a> { merge_states(states.into_values())?.request_metrics() } + pub async fn summary( + &mut self, + range: ObservabilityRange, + filter: &UsageRecordFilter, + ) -> StoreResult { + // 摘要只需要均值的样本数，不排序分位数、展开直方图或再按 Provider 聚合 + let states = self + .load_selected( + range, + filter, + None, + None, + Default::default(), + ReadSelection { + distributions: DistributionRead::Statistics, + costs: true, + keys: None, + counters: Some(&[ + "request_count", + "success_count", + "failure_count", + "cancelled_count", + "incomplete_count", + "caller_error_count", + "input_tokens", + "output_tokens", + "cached_input_tokens", + "cache_write_input_tokens", + "reasoning_output_tokens", + "total_tokens", + "latency_ms_sum", + "attempt_count", + "attempt_success_count", + "attempt_failure_count", + "attempt_cancelled_count", + "attempt_incomplete_count", + "rate_limited_count", + "auth_failure_count", + "provider_5xx_count", + "provider_reported_count", + "calculated_count", + "unavailable_count", + ]), + distribution_names: Some(&["latency_ms"]), + }, + ) + .await?; + let state = merge_states(states.into_values())?; + Ok(UsageSummaryObservation { + analytics: self.metadata(), + range, + requests: state.request_summary_metrics(), + attempts: state.attempts()?, + }) + } + pub async fn overview( &mut self, range: ObservabilityRange, @@ -201,7 +265,7 @@ impl<'a> AnalyticsSnapshot<'a> { None, Default::default(), ReadSelection { - distributions: false, + distributions: DistributionRead::None, costs: false, keys: None, counters: Some(&[ @@ -254,7 +318,7 @@ impl<'a> AnalyticsSnapshot<'a> { Some(granularity), timezone, ReadSelection { - distributions: true, + distributions: DistributionRead::Full, costs: include_costs, keys: None, counters: None, @@ -335,7 +399,7 @@ impl<'a> AnalyticsSnapshot<'a> { None, Default::default(), ReadSelection { - distributions: true, + distributions: DistributionRead::Full, costs: true, keys: selected_keys, counters: Some(&[ @@ -631,7 +695,11 @@ impl<'a> AnalyticsSnapshot<'a> { granularity, timezone, ReadSelection { - distributions: read_distributions, + distributions: if read_distributions { + DistributionRead::Full + } else { + DistributionRead::None + }, costs: true, keys: None, counters: None, @@ -679,12 +747,13 @@ impl<'a> AnalyticsSnapshot<'a> { selection: ReadSelection<'_>, ) -> StoreResult, String), MetricState>> { let ReadSelection { - distributions: read_distributions, + distributions, costs: read_costs, keys: selected_keys, counters, distribution_names, } = selection; + let read_distributions = !matches!(distributions, DistributionRead::None); filter.validate()?; self.require_visible(Some(range))?; let selected = collection_for(filter, dimension); @@ -735,12 +804,12 @@ impl<'a> AnalyticsSnapshot<'a> { } else { "'{}'::jsonb as costs" }; - let distributions = if read_distributions { + let distribution_sql = if read_distributions { "distributions" } else { "'{}'::jsonb as distributions" }; - query.push(format!(" as output_bucket,{name} as name,counts,{costs},{distributions},last_used_at from {table} a where collection=")) + query.push(format!(" as output_bucket,{name} as name,counts,{costs},{distribution_sql},last_used_at from {table} a where collection=")) .push_bind(collection).push(" and schema_version=").push_bind(SCHEMA_VERSION) .push(" and histogram_version=").push_bind(HISTOGRAM_VERSION); push_summary_buckets(&mut query, &buckets, seconds); @@ -764,7 +833,7 @@ impl<'a> AnalyticsSnapshot<'a> { query.push(")"); aggregate::push_summary_aggregation( &mut query, - read_distributions, + distributions, read_costs, counters, distribution_names, @@ -857,7 +926,7 @@ impl<'a> AnalyticsSnapshot<'a> { &mut query, exact, read_costs, - read_distributions, + distributions, counters, distribution_names, ); @@ -868,7 +937,10 @@ impl<'a> AnalyticsSnapshot<'a> { .map_err(|error| postgres_unavailable("aggregate analytics facts", error))? { let key = (get(&row, "bucket_start")?, get(&row, "name")?); - let state = MetricState::decode_aggregate(&row, exact)?; + let state = MetricState::decode_aggregate( + &row, + exact && matches!(distributions, DistributionRead::Full), + )?; if exact { result.insert(key, state); } else { @@ -953,6 +1025,9 @@ pub(super) fn collection_for( || filter.response_id.is_some() || filter.upstream_request_id.is_some() || filter.search.is_some() + || filter.model_contains.is_some() + || filter.client_api_key_name.is_some() + || filter.provider_account_name.is_some() { return None; } @@ -1004,10 +1079,17 @@ pub(crate) fn push_analytics_filter( ) { let mut typed = filter.clone(); typed.search = None; + typed.client_api_key_name = None; + typed.provider_account_name = None; push_usage_filter(query, &typed, alias); - if filter.search.is_some() { + if filter.search.is_some() + || filter.client_api_key_name.is_some() + || filter.provider_account_name.is_some() + { query.push(format!(" and exists(select 1 from model_request_observations searched where searched.id={alias}.id")); let text = UsageRecordFilter { + client_api_key_name: filter.client_api_key_name.clone(), + provider_account_name: filter.provider_account_name.clone(), search: filter.search.clone(), ..Default::default() }; diff --git a/backend/crates/gateway-store/src/postgres/observability/mapping.rs b/backend/crates/gateway-store/src/postgres/observability/mapping.rs index 97293b4c4..adc96d1f2 100644 --- a/backend/crates/gateway-store/src/postgres/observability/mapping.rs +++ b/backend/crates/gateway-store/src/postgres/observability/mapping.rs @@ -17,6 +17,9 @@ pub(crate) fn store_range( pub(crate) fn store_usage_filter(filter: admin_observability::UsageFilter) -> UsageRecordFilter { UsageRecordFilter { + model_contains: filter.model_contains, + client_api_key_name: filter.client_api_key_name, + provider_account_name: filter.provider_account_name, client_api_key_ref: filter.client_api_key_ref, request_id: filter.request_id, provider_account_ref: filter.provider_account_ref, diff --git a/backend/crates/gateway-store/src/postgres/observability/model.rs b/backend/crates/gateway-store/src/postgres/observability/model.rs index 6ef4301fe..a1feb501f 100644 --- a/backend/crates/gateway-store/src/postgres/observability/model.rs +++ b/backend/crates/gateway-store/src/postgres/observability/model.rs @@ -39,12 +39,15 @@ pub(crate) fn observability_page_offset( #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct UsageRecordFilter { + pub client_api_key_name: Option, + pub provider_account_name: Option, pub client_api_key_ref: Option, pub request_id: Option, pub provider_account_ref: Option, pub operation: Option, pub provider_kind: Option, pub model: Option, + pub model_contains: Option, pub outcome: Option, pub status_code: Option, pub transport: Option, @@ -57,6 +60,14 @@ pub struct UsageRecordFilter { impl UsageRecordFilter { pub fn validate(&self) -> StoreResult<()> { for (value, field) in [ + ( + self.client_api_key_name.as_deref(), + "client API key name filter", + ), + ( + self.provider_account_name.as_deref(), + "provider account name filter", + ), (self.client_api_key_ref.as_deref(), "client API key filter"), (self.request_id.as_deref(), "request ID filter"), ( @@ -66,6 +77,7 @@ impl UsageRecordFilter { (self.operation.as_deref(), "operation filter"), (self.provider_kind.as_deref(), "provider filter"), (self.model.as_deref(), "model filter"), + (self.model_contains.as_deref(), "model contains filter"), (self.outcome.as_deref(), "outcome filter"), (self.transport.as_deref(), "transport filter"), ( @@ -345,6 +357,14 @@ pub struct UsageOverview { pub providers: Vec, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UsageSummaryObservation { + pub analytics: admin_observability::AnalyticsMetadata, + pub range: ObservabilityRange, + pub requests: admin_observability::RequestSummaryMetrics, + pub attempts: AttemptMetrics, +} + #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct DiagnosticsObservation { pub analytics: admin_observability::AnalyticsMetadata, diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs index fc195ef05..4ef32e36e 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs @@ -11,6 +11,19 @@ pub(crate) fn push_usage_filter( filter: &UsageRecordFilter, alias: &str, ) { + for (column, value) in [ + ("client_api_key_name_snapshot", &filter.client_api_key_name), + ( + "provider_account_name_snapshot", + &filter.provider_account_name, + ), + ] { + if let Some(value) = value { + query.push(format!(" and {alias}.{column} like ")); + query.push_bind(literal_contains_pattern(value)); + query.push(" escape '\\'"); + } + } if let Some(value) = &filter.client_api_key_ref { query.push(format!(" and {alias}.client_api_key_ref = ")); query.push_bind(value.clone()); @@ -38,6 +51,14 @@ pub(crate) fn push_usage_filter( query.push_bind(value.clone()); query.push(")"); } + if let Some(value) = &filter.model_contains { + let pattern = literal_contains_pattern(value); + query.push(format!(" and ({alias}.requested_model_id like ")); + query.push_bind(pattern.clone()); + query.push(format!(" escape '\\' or {alias}.upstream_model_id like ")); + query.push_bind(pattern); + query.push(" escape '\\')"); + } if let Some(value) = &filter.outcome { query.push(format!(" and {alias}.outcome = ")); query.push_bind(value.clone()); @@ -102,6 +123,10 @@ pub(crate) fn push_usage_filter( } } +fn literal_contains_pattern(value: &str) -> String { + format!("%{}", literal_prefix_pattern(value)) +} + pub(crate) fn literal_prefix_pattern(value: &str) -> String { format!( "{}%", @@ -176,11 +201,7 @@ pub(crate) async fn list_usage_records( let items = list_usage_record_items(&mut *snapshot.tx, &query, None).await?; let total = if items.len() < usize::from(query.page_size.get()) { items.len() as u64 - } else if query.status == admin_observability::UsageRecordStatus::Success - && let Some(counts) = snapshot - .completed_request_counts(query.range, &query.filter) - .await? - { + } else if let Some(counts) = projected_usage_counts(&mut snapshot, &query).await? { counts.values().try_fold(0_u64, |sum, count| { sum.checked_add(*count) .ok_or_else(|| invalid("usage count overflows u64")) @@ -190,21 +211,18 @@ pub(crate) async fn list_usage_records( }; (total, items) } else { - let (total, page) = if query.status == admin_observability::UsageRecordStatus::Success - && let Some(counts) = snapshot - .completed_request_counts(query.range, &query.filter) - .await? - { - time_page_from_segments( - counts.into_iter().rev(), - query.range, - observability_page_offset(query.current_page, query.page_size)?, - query.page_size, - TimeDelta::minutes(15), - )? - } else { - usage_time_page(&mut snapshot.tx, &query).await? - }; + let (total, page) = + if let Some(counts) = projected_usage_counts(&mut snapshot, &query).await? { + time_page_from_segments( + counts.into_iter().rev(), + query.range, + observability_page_offset(query.current_page, query.page_size)?, + query.page_size, + TimeDelta::minutes(15), + )? + } else { + usage_time_page(&mut snapshot.tx, &query).await? + }; let items = match page { Some(page) => list_usage_record_items_at(&mut *snapshot.tx, &query, page).await?, None => Vec::new(), @@ -220,6 +238,48 @@ pub(crate) async fn list_usage_records( }) } +async fn projected_usage_counts( + snapshot: &mut AnalyticsSnapshot<'_>, + query: &UsageRecordQuery, +) -> StoreResult, u64>>> { + use admin_observability::UsageRecordStatus; + if query.status == UsageRecordStatus::Error { + return Ok(None); + } + let Some(mut counts) = snapshot + .completed_request_counts(query.range, &query.filter) + .await? + else { + return Ok(None); + }; + if query.status == UsageRecordStatus::All { + // 成功计数已校正未投影差量；其他状态直接读同一快照源表，保留恢复记录和执行中请求 + let mut statement = QueryBuilder::::new( + "select date_bin(interval '15 minutes',mr.started_at,timestamptz 'epoch') as bucket_start, + count(*)::bigint as request_count + from model_request_observations mr where mr.started_at >= ", + ); + statement.push_bind(query.range.start); + statement.push(" and mr.started_at < "); + statement.push_bind(query.range.end); + statement.push(" and mr.outcome in ('running','failed','cancelled','incomplete')"); + push_non_success_count_filter(&mut statement, &query.filter); + statement.push(" group by bucket_start"); + let rows = statement + .build() + .fetch_all(&mut *snapshot.tx) + .await + .map_err(|source| postgres_unavailable("count non-success usage records", source))?; + for row in rows { + let count = counts.entry(get(&row, "bucket_start")?).or_default(); + *count = count + .checked_add(to_u64(get(&row, "request_count")?)?) + .ok_or_else(|| invalid("usage count overflows u64"))?; + } + } + Ok(Some(counts)) +} + pub(crate) async fn list_usage_record_items<'e, E: sqlx::Executor<'e, Database = Postgres>>( executor: E, query: &UsageRecordQuery, @@ -357,17 +417,7 @@ async fn usage_time_page( query: &UsageRecordQuery, ) -> StoreResult<(u64, Option)> { let offset = observability_page_offset(query.current_page, query.page_size)?; - let mut statement = QueryBuilder::::new( - "select date_bin(interval '1 minute', mr.started_at, timestamptz 'epoch') as bucket_start, - count(*)::bigint as request_count - from model_request_observations mr where mr.started_at >= ", - ); - statement.push_bind(query.range.start); - statement.push(" and mr.started_at < "); - statement.push_bind(query.range.end); - push_usage_record_status(&mut statement, query.status); - push_usage_filter(&mut statement, &query.filter, "mr"); - statement.push(" group by bucket_start order by bucket_start desc"); + let mut statement = usage_counts_query(query.range, &query.filter, query.status, true); let rows = statement .build() .fetch_all(connection) @@ -382,14 +432,7 @@ pub(crate) async fn count_usage_records<'e, E: sqlx::Executor<'e, Database = Pos filter: &UsageRecordFilter, status: admin_observability::UsageRecordStatus, ) -> StoreResult { - let mut statement = QueryBuilder::::new( - "select count(*)::bigint from model_request_observations mr where mr.started_at >= ", - ); - statement.push_bind(range.start); - statement.push(" and mr.started_at < "); - statement.push_bind(range.end); - push_usage_record_status(&mut statement, status); - push_usage_filter(&mut statement, filter, "mr"); + let mut statement = usage_counts_query(range, filter, status, false); let total = statement .build_query_scalar::() .fetch_one(executor) @@ -398,6 +441,120 @@ pub(crate) async fn count_usage_records<'e, E: sqlx::Executor<'e, Database = Pos to_u64(total) } +fn usage_counts_query( + range: ObservabilityRange, + filter: &UsageRecordFilter, + status: admin_observability::UsageRecordStatus, + by_minute: bool, +) -> QueryBuilder { + use admin_observability::UsageRecordStatus; + let split = status == UsageRecordStatus::All; + let mut statement = QueryBuilder::::new(if split { + if by_minute { + "select bucket_start,sum(request_count)::bigint as request_count from (" + } else { + "select sum(request_count)::bigint as request_count from (" + } + } else { + "" + }); + // 成功交付和非成功互斥，分别使用交付部分索引和 outcome 索引，保留恢复记录 + for branch in 0..if split { 2 } else { 1 } { + if branch > 0 { + statement.push(" union all "); + } + statement.push("select "); + if by_minute { + statement.push( + "date_bin(interval '1 minute',mr.started_at,timestamptz 'epoch') as bucket_start,", + ); + } + statement.push("count(*)::bigint as request_count from model_request_observations mr where mr.started_at >= "); + statement.push_bind(range.start); + statement.push(" and mr.started_at < "); + statement.push_bind(range.end); + if branch > 0 { + // 与 model_requests_outcome_ck 允许的全部非成功状态一致 + statement.push(" and mr.outcome in ('running','failed','cancelled','incomplete')"); + } else { + push_usage_record_status( + &mut statement, + if split { + UsageRecordStatus::Success + } else { + status + }, + ); + } + if branch > 0 || status == UsageRecordStatus::Error { + push_non_success_count_filter(&mut statement, filter); + } else { + push_usage_filter(&mut statement, filter, "mr"); + } + if by_minute { + statement.push(" group by bucket_start"); + } + } + if split { + statement.push(") counts"); + if by_minute { + statement.push(" group by bucket_start"); + } + } + if by_minute { + statement.push(" order by bucket_start desc"); + } + statement +} + +fn push_non_success_count_filter( + statement: &mut QueryBuilder, + filter: &UsageRecordFilter, +) { + let model_filter = UsageRecordFilter { + provider_kind: filter.provider_kind.clone(), + model: filter.model.clone(), + outcome: filter.outcome.clone(), + ..Default::default() + }; + let Some(model) = filter.model.as_ref().filter(|_| *filter == model_filter) else { + push_usage_filter(statement, filter, "mr"); + return; + }; + let outcome_only = UsageRecordFilter { + outcome: filter.outcome.clone(), + ..Default::default() + }; + push_usage_filter(statement, &outcome_only, "mr"); + // outcome 索引没有模型或供应方字段；用同一记录的三列等值探测现有索引，避免大批错误回宽表 + // 其他筛选仍需读取源行，继续使用原谓词；两个 EXISTS 的 OR 不会重复计算双模型匹配 + statement.push(" and ("); + for (index, column) in ["requested_model_id", "upstream_model_id"] + .iter() + .enumerate() + { + if index > 0 { + statement.push(" or "); + } + statement.push(format!( + "exists(select 1 from model_requests modeled where modeled.{column}=" + )); + statement.push_bind(model.clone()); + statement.push(" and modeled.started_at=mr.started_at and modeled.id=mr.id"); + if let Some(provider) = &filter.provider_kind { + // 放在模型候选内，避免外层同表 EXISTS 被合并为 mr.provider_kind 而重新回宽表 + statement.push( + " and exists(select 1 from model_requests provided where provided.provider_kind=", + ); + statement.push_bind(provider.clone()); + statement + .push(" and provided.started_at=modeled.started_at and provided.id=modeled.id)"); + } + statement.push(")"); + } + statement.push(")"); +} + pub(crate) async fn usage_record_detail( connection: &mut PgConnection, request_id: &str, diff --git a/backend/crates/gateway-store/tests/postgres/analytics.rs b/backend/crates/gateway-store/tests/postgres/analytics.rs index d5e410664..1c4fb8124 100644 --- a/backend/crates/gateway-store/tests/postgres/analytics.rs +++ b/backend/crates/gateway-store/tests/postgres/analytics.rs @@ -21,6 +21,245 @@ pub(super) async fn seed(pool: &sqlx::PgPool, id: &str, start: DateTime, gr .bind(id).bind(start).bind(observation).execute(pool).await.expect("seed analytic request"); } +#[tokio::test] +async fn usage_summary_preserves_counts_latency_samples_and_costs_across_read_paths() { + let Some(database) = TestDatabase::create("analytics_summary_projection").await else { + return; + }; + let start = DateTime::parse_from_rfc3339("2026-09-01T10:00:00Z") + .unwrap() + .with_timezone(&Utc); + for (id, minute, latency, currency, amount, key, provider, outcome) in [ + ( + "summary_zero", + 5, + Some(0_i64), + Some("USD"), + Some("0"), + "key_analytics", + "openai", + "succeeded", + ), + ( + "summary_usd", + 20, + Some(101), + Some("USD"), + Some("1.25"), + "key_analytics", + "openai", + "succeeded", + ), + ( + "summary_cny", + 35, + None, + Some("CNY"), + Some("2.50"), + "key_analytics", + "openai", + "succeeded", + ), + ( + "summary_unknown", + 50, + Some(303), + None, + None, + "key_analytics", + "openai", + "succeeded", + ), + ( + "summary_cancelled", + 65, + Some(999), + Some("USD"), + Some("777"), + "key_analytics", + "openai", + "cancelled", + ), + ( + "summary_outside", + 130, + Some(999), + Some("USD"), + Some("999"), + "key_analytics", + "openai", + "succeeded", + ), + ( + "summary_other_key", + 5, + Some(999), + Some("USD"), + Some("99"), + "key_other", + "openai", + "succeeded", + ), + ( + "summary_other_provider", + 5, + Some(999), + Some("USD"), + Some("99"), + "key_analytics", + "xai", + "succeeded", + ), + ] { + seed(&database.pool, id, start + TimeDelta::minutes(minute), &[]).await; + sqlx::query("update model_requests set provider_kind=$2,provider_account_ref='account_summary',client_api_key_ref=$3,upstream_transport='http_sse',attempt_count=1,upstream_send_state='sent',outcome=$4,client_status_code=case when $4='cancelled' then 499 else 200 end,cost_source=case when $5::text is null then 'unavailable' else 'provider_reported' end,cost_currency=$5,cost_amount=$6::text::numeric,request_observation_json=jsonb_set(request_observation_json,'{timings,local}',jsonb_strip_nulls(jsonb_build_object('latencyMs',$7::bigint))) where id=$1") + .bind(id).bind(provider).bind(key).bind(outcome).bind(currency).bind(amount).bind(latency) + .execute(&database.pool).await.unwrap(); + } + drain_analytics(&database.pool).await; + let repo = observability_repository(&database.pool); + let full = ObservabilityRange::new(start, start + TimeDelta::hours(2)).unwrap(); + let edges = ObservabilityRange::new( + start + TimeDelta::minutes(4), + start + TimeDelta::minutes(66), + ) + .unwrap(); + let filter = UsageRecordFilter { + provider_kind: Some("openai".into()), + client_api_key_ref: Some("key_analytics".into()), + ..Default::default() + }; + let exact_filter = UsageRecordFilter { + transport: Some("http_sse".into()), + ..filter.clone() + }; + for (range, filter, approximate) in [ + (full, filter.clone(), true), + (edges, filter.clone(), true), + (full, exact_filter.clone(), false), + ] { + let summary = repo.usage_summary(range, filter.clone()).await.unwrap(); + let overview = repo + .usage_insights_observation( + range, + filter, + gateway_admin::model::observability::Granularity::Hour, + ) + .await + .unwrap() + .overview; + assert_eq!(summary.requests, (&overview.requests).into()); + assert_eq!( + ( + summary.requests.request_count, + summary.requests.success_count, + summary.requests.cancelled_count + ), + (5, 4, 1) + ); + assert_eq!( + ( + summary.requests.latency_sum_ms, + summary.requests.latency_count + ), + (404, 3) + ); + assert_eq!(summary.requests.total_tokens, 480); + assert_eq!(summary.attempts.attempt_count, 5); + assert_eq!(summary.attempts.cost_coverage.provider_reported_count, 3); + assert_eq!(summary.attempts.cost_coverage.unavailable_count, 1); + assert_eq!(summary.attempts.costs, overview.attempts.costs); + assert_eq!( + summary + .attempts + .costs + .iter() + .map(|cost| (cost.currency.as_str(), cost.amount.as_str())) + .collect::>(), + vec![("CNY", "2.5"), ("USD", "1.25")] + ); + assert_eq!(summary.analytics.percentiles_approximate, approximate); + } + let empty = repo + .usage_summary( + full, + UsageRecordFilter { + client_api_key_ref: Some("key_missing".into()), + ..filter.clone() + }, + ) + .await + .unwrap(); + assert_eq!(empty.requests, Default::default()); + assert!(empty.attempts.costs.is_empty()); + let other = repo + .usage_summary( + full, + UsageRecordFilter { + client_api_key_ref: Some("key_other".into()), + ..filter + }, + ) + .await + .unwrap(); + assert_eq!(other.requests.request_count, 1); + assert_eq!(other.attempts.costs[0].amount.as_str(), "99"); + + // 事实摘要按币种合并时，零毫秒是一个样本，缺少延迟和空结果均不能补造样本 + for (minute, request_count, latency_count) in [(5, 1, 1), (35, 1, 0), (90, 0, 0)] { + let range = ObservabilityRange::new( + start + TimeDelta::minutes(minute), + start + TimeDelta::minutes(minute + 1), + ) + .unwrap(); + let summary = repo + .usage_summary(range, exact_filter.clone()) + .await + .unwrap(); + assert_eq!(summary.requests.request_count, request_count); + assert_eq!(summary.requests.latency_count, latency_count); + assert_eq!(summary.requests.latency_sum_ms, 0); + } + + // 单独验证事实费用聚合：空实际金额不造零币种，空币种仍保留情景费用 + sqlx::query("update request_analytics_facts set cost_amount=case when id='summary_zero' then null else cost_amount end,standard_cost_amount=case id when 'summary_zero' then 0.1 when 'summary_usd' then 0.2 when 'summary_cny' then 0.3 when 'summary_unknown' then 0.4 when 'summary_cancelled' then 999 end,no_cache_cost_amount=case id when 'summary_zero' then 0.2 when 'summary_usd' then 0.4 when 'summary_cny' then 0.6 when 'summary_unknown' then 0.8 when 'summary_cancelled' then 999 end") + .execute(&database.pool).await.unwrap(); + let summary = repo + .usage_summary(full, exact_filter.clone()) + .await + .unwrap(); + let insights = repo + .usage_insights_observation( + full, + exact_filter, + gateway_admin::model::observability::Granularity::Hour, + ) + .await + .unwrap(); + assert_eq!( + insights.costs[0].standard_amount.as_ref().unwrap().as_str(), + "1" + ); + assert_eq!( + insights.costs[0].no_cache_amount.as_ref().unwrap().as_str(), + "2" + ); + assert_eq!( + insights + .overview + .attempts + .costs + .iter() + .map(|cost| (cost.currency.as_str(), cost.amount.as_str())) + .collect::>(), + vec![("CNY", "2.5"), ("USD", "1.25")] + ); + assert!(insights.costs[1].standard_amount.is_none()); + assert_eq!(summary.requests, (&insights.overview.requests).into()); + assert_eq!(summary.attempts.costs, insights.overview.attempts.costs); + database.close().await; +} + #[tokio::test] async fn provider_counts_match_for_rollups_boundaries_and_filtered_facts() { let Some(database) = TestDatabase::create("analytics_provider_counts").await else { @@ -58,7 +297,15 @@ async fn provider_counts_match_for_rollups_boundaries_and_filtered_facts() { }, ), ] { - let summary = repo.usage_summary(range, filter.clone()).await.unwrap(); + let summary = repo + .usage_insights_observation( + range, + filter.clone(), + gateway_admin::model::observability::Granularity::Hour, + ) + .await + .map(|snapshot| snapshot.overview) + .unwrap(); assert_eq!(summary.requests.request_count, 3); let providers = summary .providers @@ -159,7 +406,15 @@ async fn insights_share_bucket_states_without_merging_exact_percentiles() { ..Default::default() }, ] { - let expected = repo.usage_summary(range, filter.clone()).await.unwrap(); + let expected = repo + .usage_insights_observation( + range, + filter.clone(), + gateway_admin::model::observability::Granularity::Hour, + ) + .await + .map(|snapshot| snapshot.overview) + .unwrap(); if filter.search.is_some() { assert_eq!( expected @@ -850,16 +1105,26 @@ async fn persisted_histograms_merge_long_tails_and_distinguish_empty_from_zero() drain_analytics(&database.pool).await; let repo = observability_repository(&database.pool); let range = ObservabilityRange::new(start, start + TimeDelta::hours(1)).unwrap(); - let approximate = repo.usage_summary(range, Default::default()).await.unwrap(); + let approximate = repo + .usage_insights_observation( + range, + Default::default(), + gateway_admin::model::observability::Granularity::Hour, + ) + .await + .map(|snapshot| snapshot.overview) + .unwrap(); let exact = repo - .usage_summary( + .usage_insights_observation( range, UsageRecordFilter { search: Some("model_".to_owned()), ..Default::default() }, + gateway_admin::model::observability::Granularity::Hour, ) .await + .map(|snapshot| snapshot.overview) .unwrap(); assert!(approximate.analytics.percentiles_approximate); assert!(!exact.analytics.percentiles_approximate); @@ -887,23 +1152,27 @@ async fn persisted_histograms_merge_long_tails_and_distinguish_empty_from_zero() assert!((actual - reference).abs() <= reference * 0.03); } let zero = repo - .usage_summary( + .usage_insights_observation( ObservabilityRange::new(start, start + TimeDelta::minutes(10)).unwrap(), Default::default(), + gateway_admin::model::observability::Granularity::Hour, ) .await + .map(|snapshot| snapshot.overview) .unwrap(); assert_eq!( zero.requests.latency_percentiles.p95_ms.unwrap().as_f64(), 0.0 ); let empty = repo - .usage_summary( + .usage_insights_observation( ObservabilityRange::new(start + TimeDelta::hours(2), start + TimeDelta::hours(3)) .unwrap(), Default::default(), + gateway_admin::model::observability::Granularity::Hour, ) .await + .map(|snapshot| snapshot.overview) .unwrap(); assert!(empty.requests.latency_percentiles.p95_ms.is_none()); database.close().await; @@ -1448,10 +1717,26 @@ async fn sql_edge_histograms_match_persisted_power_of_two_boundaries() { drain_analytics(&database.pool).await; let repo = observability_repository(&database.pool); let range = ObservabilityRange::new(start, start + TimeDelta::hours(1)).unwrap(); - let persisted = repo.usage_summary(range, Default::default()).await.unwrap(); + let persisted = repo + .usage_insights_observation( + range, + Default::default(), + gateway_admin::model::observability::Granularity::Hour, + ) + .await + .unwrap() + .overview; sqlx::query("insert into request_analytics_pending_buckets(bucket_start,oldest_pending_at) values($1,now())") .bind(start).execute(&database.pool).await.unwrap(); - let edge = repo.usage_summary(range, Default::default()).await.unwrap(); + let edge = repo + .usage_insights_observation( + range, + Default::default(), + gateway_admin::model::observability::Granularity::Hour, + ) + .await + .unwrap() + .overview; assert_eq!(edge.requests, persisted.requests); assert_eq!(edge.attempts, persisted.attempts); database.close().await; diff --git a/backend/crates/gateway-store/tests/postgres/execution/mod.rs b/backend/crates/gateway-store/tests/postgres/execution/mod.rs index 7efbf0a32..bee0edf67 100644 --- a/backend/crates/gateway-store/tests/postgres/execution/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/execution/mod.rs @@ -23,9 +23,9 @@ use gateway_core::policy::ClientApiKeyId; use gateway_core::routing::{AccountRoutingSnapshot, ConfigRevision, PublicModelId}; use gateway_core::upstream::UpstreamSendState; use gateway_store::postgres::{ - AttemptMetrics, ModelRequestAttemptStart, ModelRequestRepository, NewModelRequest, - ObservabilityPageSize, ObservabilityRange, OpsErrorFilter, OpsErrorQuery, PgExecutionStore, - UsageRecordFilter, UsageRecordQuery, + ModelRequestAttemptStart, ModelRequestRepository, NewModelRequest, ObservabilityPageSize, + ObservabilityRange, OpsErrorFilter, OpsErrorQuery, PgExecutionStore, UsageRecordFilter, + UsageRecordQuery, }; use serde_json::{Value, json}; use sqlx::PgPool; @@ -1997,8 +1997,13 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates let repository = observability_repository(&database.pool); super::drain_analytics(&database.pool).await; let before = repository - .usage_summary(zero_attempt_range(&request), UsageRecordFilter::default()) + .usage_insights_observation( + zero_attempt_range(&request), + UsageRecordFilter::default(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .expect("baseline usage"); assert_eq!(before.requests.total_tokens, 120); assert!(!before.attempts.costs.is_empty()); @@ -2011,8 +2016,13 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates .expect("finalize failure"); super::drain_analytics(&database.pool).await; let after = repository - .usage_summary(zero_attempt_range(&request), UsageRecordFilter::default()) + .usage_insights_observation( + zero_attempt_range(&request), + UsageRecordFilter::default(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .expect("usage after failure"); assert_eq!( after.requests.request_count, @@ -2031,14 +2041,16 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates assert_eq!(after.requests.latency_count, before.requests.latency_count); assert_eq!(after.attempts, before.attempts); let successes = repository - .usage_summary( + .usage_insights_observation( zero_attempt_range(&request), UsageRecordFilter { outcome: Some("succeeded".to_owned()), ..Default::default() }, + gateway_admin::model::observability::Granularity::Hour, ) .await + .map(|snapshot| snapshot.overview) .expect("successful aggregation unchanged"); assert_eq!(successes.range, before.range); assert_eq!(successes.requests, before.requests); @@ -2061,14 +2073,22 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates ..Default::default() }; let failed = repository - .usage_summary(zero_attempt_range(&request), filter.clone()) + .usage_insights_observation( + zero_attempt_range(&request), + filter.clone(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .expect("failed request metrics"); assert_eq!(failed.requests.request_count, 1); assert_eq!(failed.requests.failure_count, 1); assert_eq!(failed.requests.total_tokens, 0); assert_eq!(failed.requests.latency_count, 0); - assert_eq!(failed.attempts, AttemptMetrics::default()); + assert_eq!( + failed.attempts, + admin_observability::AttemptMetrics::default() + ); let page = repository .list_usage_records(UsageRecordQuery { status: Default::default(), diff --git a/backend/crates/gateway-store/tests/postgres/observability/mod.rs b/backend/crates/gateway-store/tests/postgres/observability/mod.rs index abdb17342..43ab3fe36 100644 --- a/backend/crates/gateway-store/tests/postgres/observability/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/observability/mod.rs @@ -153,8 +153,13 @@ async fn output_throughput_uses_official_response_duration_without_local_fallbac super::drain_analytics(&database.pool).await; let summary = repository - .usage_summary(range, UsageRecordFilter::default()) + .usage_insights_observation( + range, + UsageRecordFilter::default(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .expect("throughput summary"); assert_percentile( summary.requests.output_throughput_p10.map(|v| v as f64), @@ -1161,8 +1166,13 @@ async fn recovered_continuation_failure_should_be_visible_in_ops_but_hidden_from super::drain_analytics(&database.pool).await; let overview = repository - .usage_summary(range, UsageRecordFilter::default()) + .usage_insights_observation( + range, + UsageRecordFilter::default(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .expect("usage summary without recovered intermediates"); assert_eq!(overview.requests.request_count, 2); assert_eq!(overview.requests.failure_count, 0); @@ -1758,6 +1768,9 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det status: Default::default(), range, filter: admin_observability::UsageFilter { + model_contains: None, + client_api_key_name: None, + provider_account_name: None, client_api_key_ref: Some("key_observe".to_owned()), request_id: Some("req_observe_success".to_owned()), provider_account_ref: Some("acct_observe".to_owned()), @@ -1841,8 +1854,13 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det super::drain_analytics(&database.pool).await; let overview = store - .usage_summary(range, admin_observability::UsageFilter::default()) + .usage_insights_observation( + range, + admin_observability::UsageFilter::default(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .expect("admin usage overview"); assert_eq!(overview.range, range); assert_eq!(overview.providers[0].provider_kind, "openai"); @@ -2324,8 +2342,13 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact super::drain_analytics(&database.pool).await; let overview = repository - .usage_summary(range, UsageRecordFilter::default()) + .usage_insights_observation( + range, + UsageRecordFilter::default(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .expect("usage summary"); assert_eq!(overview.attempts.attempt_count, 4); assert_eq!(overview.attempts.failure_count, 2); @@ -2343,14 +2366,16 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact super::drain_analytics(&database.pool).await; let succeeded = repository - .usage_summary( + .usage_insights_observation( range, UsageRecordFilter { outcome: Some("succeeded".to_owned()), ..UsageRecordFilter::default() }, + gateway_admin::model::observability::Granularity::Hour, ) .await + .map(|snapshot| snapshot.overview) .expect("filtered usage summary"); assert_eq!(succeeded.requests.cache_eligible_request_count, 1); assert_eq!(succeeded.requests.cache_hit_request_count, 1); diff --git a/backend/crates/gateway-store/tests/postgres/snapshots.rs b/backend/crates/gateway-store/tests/postgres/snapshots.rs index 7c8276932..19516108e 100644 --- a/backend/crates/gateway-store/tests/postgres/snapshots.rs +++ b/backend/crates/gateway-store/tests/postgres/snapshots.rs @@ -1446,5 +1446,49 @@ async fn queued_history_keeps_frozen_identity_after_configuration_deletion() { "Original note".to_owned() ) ); + finalize_request(&database.pool, "queued_history", started_at).await; + let repository = observability_repository(&database.pool); + for published in [false, true] { + if published { + super::drain_analytics(&database.pool).await; + } + for (key_name, account_name, expected) in [ + (Some("Original Key"), None, 1), + (None, Some("Original Account"), 1), + (Some("Original Key"), Some("Original Account"), 1), + (Some("Original"), None, 1), + (Some("ginal K"), None, 1), + (None, Some("ginal Acc"), 1), + (Some("queued_key"), None, 0), + (None, Some("queued_account"), 0), + (Some("Original Account"), None, 0), + (None, Some("Original Key"), 0), + (Some("Original%"), None, 0), + ] { + for status in [ + admin_observability::UsageRecordStatus::Success, + admin_observability::UsageRecordStatus::All, + ] { + let mut query = usage_query( + started_at, + UsageRecordFilter { + client_api_key_name: key_name.map(str::to_owned), + provider_account_name: account_name.map(str::to_owned), + ..Default::default() + }, + ); + query.status = status; + let page = repository.list_usage_records(query).await.unwrap(); + assert_eq!( + page.total, expected, + "published={published}, key={key_name:?}, account={account_name:?}, status={status:?}" + ); + assert_eq!(page.items.len() as u64, expected); + if expected == 1 { + assert_eq!(page.items[0].id, "queued_history"); + } + } + } + } database.close().await; } diff --git a/backend/crates/gateway-store/tests/postgres/usage_data/mod.rs b/backend/crates/gateway-store/tests/postgres/usage_data/mod.rs index 00f8f0c41..3c315dbad 100644 --- a/backend/crates/gateway-store/tests/postgres/usage_data/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/usage_data/mod.rs @@ -530,8 +530,13 @@ async fn unassigned_correction_changes_totals_without_inventing_dimension_member let repo = observability_repository(&db.pool); let range = ObservabilityRange::new(start(), start() + TimeDelta::days(1)).unwrap(); let summary = repo - .usage_summary(range, NativeFilter::default()) + .usage_insights_observation( + range, + NativeFilter::default(), + gateway_admin::model::observability::Granularity::Hour, + ) .await + .map(|snapshot| snapshot.overview) .unwrap(); assert_eq!( ( diff --git a/backend/crates/gateway-store/tests/postgres/usage_paging.rs b/backend/crates/gateway-store/tests/postgres/usage_paging.rs index 6c0dda31b..1a251c0a9 100644 --- a/backend/crates/gateway-store/tests/postgres/usage_paging.rs +++ b/backend/crates/gateway-store/tests/postgres/usage_paging.rs @@ -10,6 +10,8 @@ use super::{TestDatabase, analytics::seed, observability_repository}; #[tokio::test] async fn projected_page_counts_correct_pending_moves_deletions_and_model_or() { + use gateway_admin::model::observability::UsageRecordStatus; + let Some(database) = TestDatabase::create("usage_projected_pages").await else { return; }; @@ -35,10 +37,10 @@ async fn projected_page_counts_correct_pending_moves_deletions_and_model_or() { .execute(&database.pool) .await .unwrap(); - sqlx::query("update model_requests set outcome='failed' where id='projected_21'") - .execute(&database.pool) - .await - .unwrap(); + sqlx::query("update model_requests set outcome='failed',upstream_model_id='model_b',recovery_request_id='projected_22',recovered_at=$1,recovery_attempt_count=1,request_observation_json=jsonb_set(request_observation_json,'{recovery}',jsonb_build_object('retryDelayMs',0,'totalLatencyMs',1)) where id='projected_21'") + .bind(start + TimeDelta::minutes(22)).execute(&database.pool).await.unwrap(); + sqlx::query("update model_requests set outcome=case id when 'projected_23' then 'running' when 'projected_24' then 'cancelled' else 'incomplete' end,completed_at=case when id='projected_23' then null else completed_at end,requested_model_id=case when id in ('projected_24','projected_25') then null else requested_model_id end,upstream_model_id=case when id='projected_25' then 'model_b' else upstream_model_id end where id in ('projected_23','projected_24','projected_25')") + .execute(&database.pool).await.unwrap(); seed( &database.pool, "projected_new", @@ -56,37 +58,39 @@ async fn projected_page_counts_correct_pending_moves_deletions_and_model_or() { ) .unwrap(); for model in [None, Some("model_a"), Some("model_b")] { - // 真值只读取实际源版本，不依赖被测汇总或分页辅助函数 - let expected:Vec = sqlx::query_scalar("select id from model_requests where started_at >= $1 and started_at < $2 and outcome='succeeded' and ($3::text is null or requested_model_id=$3 or upstream_model_id=$3) order by started_at desc,id desc") - .bind(range.start).bind(range.end).bind(model).fetch_all(&database.pool).await.unwrap(); - for current_page in [1, 3, 6, 8] { - let page = repository - .list_usage_records(UsageRecordQuery { - status: Default::default(), - range, - filter: UsageRecordFilter { - client_api_key_ref: Some("key_analytics".to_owned()), - model: model.map(str::to_owned), - ..Default::default() - }, - current_page, - page_size: ObservabilityPageSize::new(7).unwrap(), - }) - .await - .unwrap(); - assert_eq!( - page.total, - expected.len() as u64, - "stage={stage}, edge={edge}, model={model:?}" - ); - assert_eq!( - page.items.iter().map(|row| &row.id).collect::>(), - expected - .iter() - .skip((current_page as usize - 1) * 7) - .take(7) - .collect::>() - ); + for status in [UsageRecordStatus::Success, UsageRecordStatus::All] { + // 夹具中成功记录均已交付；真值只读源版本，不依赖投影或分页辅助函数 + let expected:Vec = sqlx::query_scalar("select id from model_requests where started_at >= $1 and started_at < $2 and ($4::bool or outcome='succeeded') and ($3::text is null or requested_model_id=$3 or upstream_model_id=$3) order by started_at desc,id desc") + .bind(range.start).bind(range.end).bind(model).bind(status == UsageRecordStatus::All).fetch_all(&database.pool).await.unwrap(); + for current_page in [1, 3, 6, 8] { + let page = repository + .list_usage_records(UsageRecordQuery { + status, + range, + filter: UsageRecordFilter { + client_api_key_ref: Some("key_analytics".to_owned()), + model: model.map(str::to_owned), + ..Default::default() + }, + current_page, + page_size: ObservabilityPageSize::new(7).unwrap(), + }) + .await + .unwrap(); + assert_eq!( + page.total, + expected.len() as u64, + "stage={stage}, edge={edge}, model={model:?}, status={status:?}" + ); + assert_eq!( + page.items.iter().map(|row| &row.id).collect::>(), + expected + .iter() + .skip((current_page as usize - 1) * 7) + .take(7) + .collect::>() + ); + } } } } @@ -284,6 +288,258 @@ async fn ops_pages_merge_sources_across_minutes_with_stable_ties() { database.close().await; } +#[tokio::test] +async fn all_status_counts_keep_filters_delivery_rules_and_time_boundaries() { + use gateway_admin::model::observability::UsageRecordStatus; + + let Some(database) = TestDatabase::create("usage_all_counts").await else { + return; + }; + let start = chrono::DateTime::parse_from_rfc3339("2026-09-01T10:00:00Z") + .unwrap() + .with_timezone(&Utc); + for (id, seconds, outcome) in [ + ("count_success", 0, "succeeded"), + ("count_failed", 10, "failed"), + ("count_running", 60, "running"), + ("count_cancelled", 70, "cancelled"), + ("count_incomplete", 120, "incomplete"), + ("count_success_ws", 130, "succeeded"), + ("count_prewarm", 140, "succeeded"), + ("count_uncommitted", 145, "succeeded"), + ("count_before", -1, "failed"), + ("count_end", 180, "failed"), + ] { + seed(&database.pool, id, start + TimeDelta::seconds(seconds), &[]).await; + sqlx::query("update model_requests set outcome=$2,provider_kind='openai',completed_at=case when $2='running' then null else completed_at end,input_tokens=case when $2='succeeded' then input_tokens else null end,output_tokens=case when $2='succeeded' then output_tokens else null end,total_tokens=case when $2='succeeded' then total_tokens else null end where id=$1") + .bind(id).bind(outcome).execute(&database.pool).await.unwrap(); + } + sqlx::query("update model_requests set recovery_request_id='count_success',recovered_at=$1,recovery_attempt_count=1,request_observation_json=jsonb_set(request_observation_json,'{recovery}',jsonb_build_object('retryDelayMs',1000,'totalLatencyMs',2000)) where id='count_failed'") + .bind(start + TimeDelta::minutes(2)).execute(&database.pool).await.unwrap(); + sqlx::query("update model_requests set upstream_model_id='model_a' where id='count_failed'") + .execute(&database.pool) + .await + .unwrap(); + sqlx::query("update model_requests set requested_model_id=null,upstream_model_id='model_b' where id='count_incomplete'") + .execute(&database.pool).await.unwrap(); + sqlx::query("update model_requests set client_api_key_ref='key_other',requested_model_id='model_b' where id='count_running'") + .execute(&database.pool).await.unwrap(); + sqlx::query("update model_requests set client_transport='websocket',client_status_code=null,requested_model_id='model_b' where id='count_success_ws'") + .execute(&database.pool).await.unwrap(); + sqlx::query("update model_requests set request_kind='prewarm' where id='count_prewarm'") + .execute(&database.pool) + .await + .unwrap(); + sqlx::query( + "update model_requests set downstream_committed_at=null where id='count_uncommitted'", + ) + .execute(&database.pool) + .await + .unwrap(); + // 恢复错误仍属于列表；尚未投影的源记录也必须参与精确计数 + let repo = observability_repository(&database.pool); + let range = ObservabilityRange::new(start, start + TimeDelta::minutes(3)).unwrap(); + let all = vec![ + "count_success_ws", + "count_incomplete", + "count_cancelled", + "count_running", + "count_failed", + "count_success", + ]; + let model_a = vec!["count_cancelled", "count_failed", "count_success"]; + for (filter, expected) in [ + (UsageRecordFilter::default(), all.clone()), + ( + UsageRecordFilter { + search: Some("model_".into()), + ..Default::default() + }, + all, + ), + ( + UsageRecordFilter { + client_api_key_ref: Some("key_analytics".into()), + ..Default::default() + }, + vec![ + "count_success_ws", + "count_incomplete", + "count_cancelled", + "count_failed", + "count_success", + ], + ), + ( + UsageRecordFilter { + model: Some("model_a".into()), + ..Default::default() + }, + model_a.clone(), + ), + ( + UsageRecordFilter { + provider_kind: Some("openai".into()), + model: Some("model_a".into()), + ..Default::default() + }, + model_a.clone(), + ), + ( + UsageRecordFilter { + provider_kind: Some("other".into()), + model: Some("model_a".into()), + ..Default::default() + }, + vec![], + ), + ( + UsageRecordFilter { + provider_kind: Some("openai".into()), + model: Some("model_a".into()), + outcome: Some("failed".into()), + ..Default::default() + }, + vec!["count_failed"], + ), + ( + UsageRecordFilter { + search: Some("model_a".into()), + ..Default::default() + }, + model_a, + ), + ( + UsageRecordFilter { + search: Some("no_such_model".into()), + ..Default::default() + }, + vec![], + ), + ( + UsageRecordFilter { + model: Some("model_b".into()), + ..Default::default() + }, + vec!["count_success_ws", "count_incomplete", "count_running"], + ), + ( + UsageRecordFilter { + provider_kind: Some("openai".into()), + model: Some("model_b".into()), + ..Default::default() + }, + vec!["count_success_ws", "count_incomplete", "count_running"], + ), + ( + UsageRecordFilter { + outcome: Some("failed".into()), + ..Default::default() + }, + vec!["count_failed"], + ), + ( + UsageRecordFilter { + request_id: Some("count_failed".into()), + ..Default::default() + }, + vec!["count_failed"], + ), + ( + UsageRecordFilter { + transport: Some("websocket".into()), + ..Default::default() + }, + vec!["count_success_ws"], + ), + ( + UsageRecordFilter { + provider_kind: Some("other".into()), + ..Default::default() + }, + vec![], + ), + ] { + for status in [UsageRecordStatus::All, UsageRecordStatus::Error] { + let expected = expected + .iter() + .copied() + .filter(|id| { + status == UsageRecordStatus::All + || ["count_failed", "count_incomplete"].contains(id) + }) + .collect::>(); + let mut actual = Vec::new(); + for current_page in 1..=4 { + let page = repo + .list_usage_records(UsageRecordQuery { + status, + range, + filter: filter.clone(), + current_page, + page_size: ObservabilityPageSize::new(2).unwrap(), + }) + .await + .unwrap(); + assert_eq!( + page.total, + expected.len() as u64, + "{status:?} {filter:?} page {current_page}" + ); + actual.extend(page.items.into_iter().map(|item| item.id)); + } + assert_eq!(actual, expected, "{status:?} {filter:?}"); + } + } + super::drain_analytics(&database.pool).await; + // 聚合已经发布，但跨维筛选仍走源表；已恢复错误不能因汇总不包含它而漏计 + let page = repo + .list_usage_records(UsageRecordQuery { + status: UsageRecordStatus::All, + range, + filter: UsageRecordFilter { + client_api_key_ref: Some("key_analytics".into()), + transport: Some("http_sse".into()), + model: Some("model_a".into()), + ..Default::default() + }, + current_page: 2, + page_size: ObservabilityPageSize::new(2).unwrap(), + }) + .await + .unwrap(); + assert_eq!(page.total, 3); + assert_eq!(page.items[0].id, "count_success"); + // 发布成功聚合后，供应方与模型仍共同限制源表非成功记录 + for (provider, total, expected) in [("openai", 3, vec!["count_success"]), ("other", 0, vec![])] + { + let page = repo + .list_usage_records(UsageRecordQuery { + status: UsageRecordStatus::All, + range, + filter: UsageRecordFilter { + provider_kind: Some(provider.into()), + model: Some("model_a".into()), + ..Default::default() + }, + current_page: 2, + page_size: ObservabilityPageSize::new(2).unwrap(), + }) + .await + .unwrap(); + assert_eq!(page.total, total, "{provider}"); + assert_eq!( + page.items + .into_iter() + .map(|item| item.id) + .collect::>(), + expected, + "{provider}", + ); + } + database.close().await; +} + #[tokio::test] async fn request_status_pages_share_one_request_timeline() { use gateway_admin::{ diff --git a/backend/crates/providers/openai/src/admin.rs b/backend/crates/providers/openai/src/admin.rs index f4392bd61..18f5d462d 100644 --- a/backend/crates/providers/openai/src/admin.rs +++ b/backend/crates/providers/openai/src/admin.rs @@ -1,5 +1,6 @@ //! OpenAI 管理边界：Provider preparation 与 Redis OAuth pending 适配 +use std::collections::BTreeMap; use std::sync::Arc; use std::time::SystemTime; @@ -24,9 +25,10 @@ use gateway_admin::model::provider_credentials::{ ProviderExportCredentialInput, ProviderModel, ProviderModelCatalogDocument, ProviderModels, ProviderProfileActivityInsights, ProviderProfileAvatar, ProviderProfileAvatarStreamError, ProviderProfileDailyUsage, ProviderProfileInvocation, ProviderProfileStatistics, - ProviderProfileStatisticsSummary, ProviderQuota, ProviderQuotaCredits, ProviderQuotaRequest, - ProviderQuotaWindow, ProviderQuotaWindowRole, ProviderResetCredit, ProviderResetCreditResult, - ProviderResetCredits, ProviderSubscription, QuotaLocalUsageAttribution, + ProviderProfileStatisticsSummary, ProviderQuota, ProviderQuotaCredits, + ProviderQuotaReadRequest, ProviderQuotaRequest, ProviderQuotaWindow, ProviderQuotaWindowRole, + ProviderResetCredit, ProviderResetCreditResult, ProviderResetCredits, ProviderSubscription, + QuotaLocalUsageAttribution, }; use gateway_admin::model::quota_forecast_sampling::QuotaForecastObservation; use gateway_admin::ports::provider::{ProviderAdmin, ProviderAdminError, ProviderAdminErrorKind}; @@ -707,7 +709,7 @@ impl ProviderAdmin for OpenAiAdminProvider { if refresh { return Err(provider_admin_error(ProviderAdminErrorKind::Unsupported)); } - return Ok(project_quota(None, &account)); + return Ok(project_quota(None, account.quota().is_exhausted())); } let snapshot = if refresh { let snapshot = Some( @@ -720,11 +722,81 @@ impl ProviderAdmin for OpenAiAdminProvider { snapshot } else { self.quota - .read_account(&account_id) + .read_snapshot_for(&account) .await .map_err(map_quota_error)? }; - Ok(project_quota(snapshot, &account)) + Ok(project_quota(snapshot, account.quota().is_exhausted())) + } + + async fn read_quotas( + &self, + requests: &[ProviderQuotaReadRequest<'_>], + ) -> Vec> { + let identities = requests + .iter() + .map(|request| { + if request.account.provider_kind != self.provider_kind { + return Err(provider_admin_error(ProviderAdminErrorKind::NotFound)); + } + let id = ProviderAccountId::new(request.account.id.clone()) + .map_err(|_| provider_admin_error(ProviderAdminErrorKind::Invalid))?; + let revision = + CredentialRevision::new(request.account.credential_revision.get()) + .map_err(|_| provider_admin_error(ProviderAdminErrorKind::Invalid))?; + Ok((id, revision)) + }) + .collect::>(); + let ids = requests + .iter() + .zip(&identities) + .filter(|(request, _)| { + request.account.authentication_kind + != crate::credential::CODEX_AUTHENTICATION_KIND_API_KEY + }) + .filter_map(|(_, identity)| identity.as_ref().ok().map(|(id, _)| id.clone())) + .collect::>(); + let observations = if ids.is_empty() { + Ok(BTreeMap::new()) + } else { + self.accounts + .get_quotas(&ids) + .await + .map_err(|error| map_quota_error(error.into())) + .map(|values| { + values + .into_iter() + .map(|observation| (observation.account_id.clone(), observation)) + .collect::>() + }) + }; + requests + .iter() + .zip(identities) + .map(|(request, identity)| { + let (id, revision) = identity?; + let snapshot = if request.account.authentication_kind + == crate::credential::CODEX_AUTHENTICATION_KIND_API_KEY + { + None + } else { + observations + .as_ref() + .map_err(Clone::clone)? + .get(&id) + .map(|observation| { + self.quota.project_observation(&id, revision, observation) + }) + .transpose() + .map_err(map_quota_error)? + .flatten() + }; + Ok(project_quota( + snapshot, + request.account.quota.is_exhausted(), + )) + }) + .collect() } fn quota_forecast_observation( @@ -1170,12 +1242,12 @@ fn empty_quota() -> ProviderQuota { fn project_quota( snapshot: Option, - account: &ProviderAccount, + limit_reached: bool, ) -> ProviderQuota { let mut quota = snapshot .map(project_quota_snapshot) .unwrap_or_else(empty_quota); - quota.limit_reached = account.quota().is_exhausted(); + quota.limit_reached = limit_reached; quota } diff --git a/backend/crates/providers/openai/src/credential/quota/service.rs b/backend/crates/providers/openai/src/credential/quota/service.rs index b714196cc..161ecb70f 100644 --- a/backend/crates/providers/openai/src/credential/quota/service.rs +++ b/backend/crates/providers/openai/src/credential/quota/service.rs @@ -469,7 +469,7 @@ impl CodexCredentialQuotaService { self.read_snapshot_for(&account).await } - async fn read_snapshot_for( + pub(crate) async fn read_snapshot_for( &self, account: &ProviderAccount, ) -> Result, CodexCredentialQuotaError> { @@ -486,9 +486,17 @@ impl CodexCredentialQuotaService { else { return Ok(None); }; - if observation.account_id != *account_id - || observation.expected_revision != account.revision() - { + self.project_observation(account_id, account.revision(), &observation) + } + + /// 单账号与目录批量读取共用 revision 校验、窗口解释及调度信号更新 + pub(crate) fn project_observation( + &self, + account_id: &ProviderAccountId, + revision: CredentialRevision, + observation: &QuotaObservation, + ) -> Result, CodexCredentialQuotaError> { + if observation.account_id != *account_id || observation.expected_revision != revision { return Err(CodexCredentialQuotaError::RevisionConflict); } if observation.quota.expose_to_provider().is_empty() { @@ -497,7 +505,7 @@ impl CodexCredentialQuotaService { let observed_at = observation.observed_at; let snapshot = parse_account_quota_snapshot( account_id.clone(), - account.revision(), + revision, observed_at, &Value::Object(observation.quota.expose_to_provider().clone()), )? diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index edb811abf..586f43537 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -49,6 +49,7 @@ const CLOUDFLARE_PATH_BLOCK_THRESHOLD: u32 = 3; const SESSION_AFFINITY_TIMEOUT: Duration = Duration::from_millis(100); const CYBER_POLICY_SESSION_TTL: Duration = Duration::from_secs(60 * 60); const MAX_ACCOUNT_SNAPSHOT_RETRIES: u32 = 3; +const COOLDOWN_READ_CONCURRENCY: usize = 128; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum CodexAccountFailure { @@ -671,9 +672,16 @@ impl CodexCredentialSelector { } let mut rate_limits = HashMap::with_capacity(accounts.len()); if !diagnostic { - for account in &accounts { - let until = self.quota.cooldown(account.id()).await.unwrap_or(None); - rate_limits.insert(account.id().clone(), until); + // 同一批候选并发读取实时冷却，避免逐账号累加 Redis 往返 + // 分批限制在途读取，取消选号时直接释放本批 future，不启动后续批次 + for batch in accounts.chunks(COOLDOWN_READ_CONCURRENCY) { + rate_limits.extend( + futures::future::join_all(batch.iter().map(|account| async { + let until = self.quota.cooldown(account.id()).await.unwrap_or(None); + (account.id().clone(), until) + })) + .await, + ); } } let account_ids = accounts diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 94fa4a923..bea4c5b90 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -16,8 +16,8 @@ use gateway_admin::model::provider_credentials::{ AuthorizationMutationTarget, AuthorizationOwnerBinding, CompleteAuthorization, ConsumeProviderResetCredit, PendingAuthorizationMutation, PrepareCredentialImport, PrepareCredentialRefresh, PrepareCredentialRotation, ProviderDocument, - ProviderExportCredentialInput, ProviderQuotaRequest, ProviderQuotaWindowRole, - QuotaLocalUsageAttribution, + ProviderExportCredentialInput, ProviderQuotaReadRequest, ProviderQuotaRequest, + ProviderQuotaWindowRole, QuotaLocalUsageAttribution, }; use gateway_admin::model::{MutationActor, MutationContext, Revision}; use gateway_admin::ports::provider::ProviderAdminErrorKind; @@ -807,6 +807,168 @@ async fn openai_reauthorization_pending_payload_reuses_the_account_installation_ assert!(target.get("expected_credential_revision").is_none()); } +#[tokio::test] +async fn openai_admin_reads_a_full_quota_page_in_one_store_call() { + let store = Arc::new(MemoryAccountStore::default()); + let mut records = Vec::new(); + for index in 0..200 { + let id = format!("acct_admin_quota_page_{index}"); + store + .seed_oauth_credential(ImportCodexOAuthCredential { + account_id: id.clone(), + name: id.clone(), + secret: secret("synthetic-quota-page-access"), + verified_account: profile(&format!("chatgpt-quota-page-{index}")), + next_refresh_at: None, + enabled: false, + }) + .await; + records.push(account_record(&store.account(&id).expect("stored account"))); + } + let config = valid_config(); + let bundle = provider_openai::initialize( + config.config.clone(), + provider_ports_with(Arc::clone(&store), Arc::new(TestOAuthPending::default())), + ) + .await + .expect("OpenAI bundle"); + let requests = records + .iter() + .map(|account| ProviderQuotaReadRequest { + account, + rolling_usage: None, + }) + .collect::>(); + let account_reads = store.account_reads(); + let credential_loads = store.credential_loads(); + let quota_reads = store.quota_reads(); + + let results = bundle.admin_provider().read_quotas(&requests).await; + + assert_eq!(results.len(), records.len()); + assert!( + results + .into_iter() + .all(|result| result.expect("cached quota").windows.is_empty()) + ); + assert_eq!(store.quota_reads() - quota_reads, 1); + assert_eq!(store.account_reads(), account_reads); + assert_eq!(store.credential_loads(), credential_loads); +} + +#[tokio::test] +async fn openai_admin_batch_quota_isolates_stale_and_invalid_observations() { + let store = Arc::new(MemoryAccountStore::default()); + let mut records = Vec::new(); + for suffix in ["healthy", "stale", "invalid", "missing"] { + let id = format!("acct_admin_quota_batch_{suffix}"); + store + .seed_oauth_credential(ImportCodexOAuthCredential { + account_id: id.clone(), + name: id.clone(), + secret: secret("synthetic-quota-batch-access"), + verified_account: profile(&format!("chatgpt-quota-batch-{suffix}")), + next_refresh_at: None, + enabled: false, + }) + .await; + let account = store.account(&id).expect("stored account"); + if suffix != "missing" { + let used = if suffix == "invalid" { + json!("invalid") + } else { + json!(25) + }; + store + .compare_and_swap_quota(QuotaObservation { + account_id: account.id().clone(), + expected_revision: account.revision(), + observed_at: SystemTime::now(), + plan_type: None, + state: QuotaState::allowed(SystemTime::now()), + quota: OpaqueProviderData::new( + json!({ + "rate_limit": { "primary_window": { + "used_percent": used, + "reset_at": Utc::now().timestamp() + 3600, + "limit_window_seconds": 18000 + }} + }) + .as_object() + .expect("quota object") + .clone(), + ), + }) + .await + .expect("store quota"); + } + let mut record = account_record(&store.account(&id).expect("stored quota facts")); + if suffix == "stale" { + record.credential_revision = + Revision::new(record.credential_revision.get() + 1).unwrap(); + } + records.push(record); + } + store + .seed_api_key( + "acct_admin_quota_batch_api_key", + "http://127.0.0.1:9".to_owned(), + provider_openai::credential::ResponsesTransport::Http, + ) + .await; + records.push(account_record( + &store.account("acct_admin_quota_batch_api_key").unwrap(), + )); + let mut foreign = records[0].clone(); + foreign.provider_kind = ProviderKind::new("xai").unwrap(); + records.push(foreign); + let config = valid_config(); + let bundle = provider_openai::initialize( + config.config.clone(), + provider_ports_with(Arc::clone(&store), Arc::new(TestOAuthPending::default())), + ) + .await + .expect("OpenAI bundle"); + let requests = records + .iter() + .map(|account| ProviderQuotaReadRequest { + account, + rolling_usage: None, + }) + .collect::>(); + let quota_reads = store.quota_reads(); + let account_reads = store.account_reads(); + + let results = bundle.admin_provider().read_quotas(&requests).await; + + assert_eq!(results.len(), 6); + assert_eq!( + results[0].as_ref().unwrap().windows[0].used_percent, + Some(25.0) + ); + assert_eq!( + results[1].as_ref().unwrap_err().kind(), + ProviderAdminErrorKind::Conflict + ); + assert_eq!( + results[2].as_ref().unwrap_err().kind(), + ProviderAdminErrorKind::Invalid + ); + assert!(results[3].as_ref().unwrap().windows.is_empty()); + assert!(results[4].as_ref().unwrap().windows.is_empty()); + assert_eq!( + results[5].as_ref().unwrap_err().kind(), + ProviderAdminErrorKind::NotFound + ); + assert_eq!(store.quota_reads() - quota_reads, 1); + assert_eq!(store.account_reads(), account_reads); + + let api_only = bundle.admin_provider().read_quotas(&requests[4..5]).await; + assert!(api_only[0].as_ref().unwrap().windows.is_empty()); + assert!(bundle.admin_provider().read_quotas(&[]).await.is_empty()); + assert_eq!(store.quota_reads() - quota_reads, 1); +} + #[tokio::test] async fn openai_admin_provider_projects_cached_quota_models_and_canonical_export() { let store = Arc::new(MemoryAccountStore::default()); @@ -1429,6 +1591,15 @@ async fn openai_admin_preserves_expired_window_usage_and_exhaustion_attribution( }) .await .expect("project quota"); + let record = account_record(&store.account(account_id).expect("stored quota facts")); + let batch = bundle + .admin_provider() + .read_quotas(&[ProviderQuotaReadRequest { + account: &record, + rolling_usage: None, + }]) + .await; + assert_eq!(batch[0].as_ref().expect("batch quota"), &projected); assert_eq!(projected.limit_reached, exhausted); // 账号接口还会归一化耗尽展示；过期周窗口不能把触顶错误转移到短期窗口 projected.apply_limit_reached_display(); diff --git a/backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs b/backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs new file mode 100644 index 000000000..7bdf2318c --- /dev/null +++ b/backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs @@ -0,0 +1,255 @@ +//! 验证选号冷却读取的并发边界、取消与冻结和读取失败语义 + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use gateway_core::provider_ports::{ + ProviderCooldown, ProviderCooldownKind, ProviderStoreError, ProviderStoreErrorKind, +}; +use tokio::sync::Notify; + +use super::*; + +#[derive(Default)] +struct ReadActivity { + started: AtomicUsize, + active: AtomicUsize, + peak: AtomicUsize, + changed: Notify, +} + +struct ActiveRead(Arc); + +impl ActiveRead { + fn begin(activity: Arc) -> Self { + activity.started.fetch_add(1, Ordering::SeqCst); + let active = activity.active.fetch_add(1, Ordering::SeqCst) + 1; + activity.peak.fetch_max(active, Ordering::SeqCst); + activity.changed.notify_one(); + Self(activity) + } +} + +impl Drop for ActiveRead { + fn drop(&mut self) { + self.0.active.fetch_sub(1, Ordering::SeqCst); + } +} + +fn pool_attempt(store: &Arc, count: usize) -> AttemptContext { + let provider = ProviderKind::new("openai").unwrap(); + let accounts = (0..count) + .map(|index| { + let id = format!("acct_{index:04}"); + create_account(store, &id, "synthetic-access-token"); + ( + ProviderAccountId::new(id).unwrap(), + RuntimeAccount::new(provider.clone(), BTreeSet::new()), + ) + }) + .collect(); + AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_cooldown_reads").unwrap(), + ClientApiKeyId::new("key_cooldown_reads").unwrap(), + ), + NonZeroU32::new(1).unwrap(), + SystemTime::now() + Duration::from_secs(30), + account_policy(), + AccountAttemptContext::new(BTreeSet::new(), None, None).with_account_scope(Arc::new( + FrozenAccountScope::new( + Arc::new(RuntimeAccountDirectory::new(accounts)), + ClientRoutingScope::all_accounts(), + ), + )), + None, + CancellationToken::new(), + ) +} + +#[tokio::test(start_paused = true)] +async fn selector_overlaps_cooldown_reads_with_bounded_in_flight_work() { + for count in [10_usize, 100, 1000] { + let store = Arc::new(MemoryAccountStore::default()); + let attempt = pool_attempt(&store, count); + let activity = Arc::new(ReadActivity::default()); + let observed = activity.clone(); + let cooldowns = Arc::new(MemoryCooldownPort::new().with_read_hook(Arc::new(move |_| { + let activity = observed.clone(); + Box::pin(async move { + let _read = ActiveRead::begin(activity); + tokio::time::sleep(Duration::from_millis(10)).await; + Ok(()) + }) + }))); + let selector = selector_with_runtime( + &store, + Arc::new(TestLeaseCoordinator::default()), + Arc::new(MemorySessionAffinity::default()), + Arc::new(AccountFeedbackStats::default()), + cooldowns, + ); + let request_url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let started = tokio::time::Instant::now(); + let selected = selector + .select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &request_url, + attempt: &attempt, + session_affinity_key: None, + }) + .await + .expect("available account"); + let elapsed = started.elapsed(); + assert!(selected.account_id().as_str().starts_with("acct_")); + assert_eq!(activity.started.load(Ordering::SeqCst), count); + assert_eq!(activity.active.load(Ordering::SeqCst), 0); + assert_eq!(activity.peak.load(Ordering::SeqCst), count.min(128)); + // 使用虚拟时间证明等待被合并，不把 CI 的 CPU 调度抖动当成性能回归 + assert!(elapsed <= Duration::from_millis(20 * count.div_ceil(128) as u64)); + println!( + "candidates={count} simulated_rtt_ms=10 elapsed={elapsed:?} peak_reads={}", + activity.peak.load(Ordering::SeqCst) + ); + } +} + +#[tokio::test(start_paused = true)] +async fn dropping_selection_cancels_pending_reads_without_starting_later_batches() { + let store = Arc::new(MemoryAccountStore::default()); + let attempt = pool_attempt(&store, 1000); + let activity = Arc::new(ReadActivity::default()); + let observed = activity.clone(); + let cooldowns = Arc::new(MemoryCooldownPort::new().with_read_hook(Arc::new(move |_| { + let activity = observed.clone(); + Box::pin(async move { + let _read = ActiveRead::begin(activity); + std::future::pending().await + }) + }))); + let leases = Arc::new(TestLeaseCoordinator::default()); + let selector = selector_with_runtime( + &store, + leases.clone(), + Arc::new(MemorySessionAffinity::default()), + Arc::new(AccountFeedbackStats::default()), + cooldowns, + ); + let request_url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let request = SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &request_url, + attempt: &attempt, + session_affinity_key: None, + }; + let mut selection = Box::pin(selector.select(&request)); + let pending_batch = async { + while activity.started.load(Ordering::SeqCst) < 128 { + activity.changed.notified().await; + } + }; + tokio::select! { + result = &mut selection => panic!("reads must remain pending: {result:?}"), + result = tokio::time::timeout(Duration::from_secs(1), pending_batch) => { + result.expect("first batch started"); + } + } + drop(selection); + assert_eq!(activity.started.load(Ordering::SeqCst), 128); + assert_eq!(activity.active.load(Ordering::SeqCst), 0); + assert!(leases.requests.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn concurrent_reads_preserve_cooldown_expiry_probe_freeze_and_error_fallback() { + let now = SystemTime::now(); + for (kind, until, fail_read, expected) in [ + ( + ProviderCooldownKind::RateLimit, + now + Duration::from_secs(60), + false, + "acct_second", + ), + ( + ProviderCooldownKind::RateLimit, + now - Duration::from_secs(60), + false, + "acct_first", + ), + ( + ProviderCooldownKind::CapacityFreeze, + now + Duration::from_secs(60), + false, + "acct_second", + ), + ( + ProviderCooldownKind::CapacityFreeze, + now - Duration::from_secs(60), + false, + "acct_first", + ), + ( + ProviderCooldownKind::CapacityFreezeProbe, + now - Duration::from_secs(60), + false, + "acct_second", + ), + ( + ProviderCooldownKind::RateLimit, + now + Duration::from_secs(60), + true, + "acct_first", + ), + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + store.set_scheduling("acct_first", None, AccountWeight::new(100).unwrap()); + store.set_scheduling("acct_second", None, AccountWeight::new(1).unwrap()); + let account = store.account("acct_first").unwrap(); + let cooldown = + ProviderCooldown::new_with_kind(account.id().clone(), account.revision(), until, kind); + let cooldowns = Arc::new( + MemoryCooldownPort::new().with_read_hook(Arc::new(move |id| { + Box::pin(async move { + tokio::task::yield_now().await; + if fail_read && id.as_str() == "acct_first" { + Err(ProviderStoreError::new( + ProviderStoreErrorKind::Unavailable, + "test cooldown read", + )) + } else { + Ok(()) + } + }) + })), + ); + cooldowns.put_if_later(cooldown.clone()).await.unwrap(); + let selector = selector_with_runtime( + &store, + Arc::new(TestLeaseCoordinator::default()), + Arc::new(MemorySessionAffinity::default()), + Arc::new(AccountFeedbackStats::default()), + cooldowns.clone(), + ); + let request_url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let attempt = attempt(BTreeSet::new()); + let selected = selector + .select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &request_url, + attempt: &attempt, + session_affinity_key: None, + }) + .await + .unwrap(); + assert_eq!( + selected.account_id().as_str(), + expected, + "{kind:?} failure={fail_read}" + ); + assert_eq!( + cooldowns.cooldowns.lock().unwrap().get(account.id()), + Some(&cooldown) + ); + } +} diff --git a/backend/crates/providers/openai/tests/credential/contract/mod.rs b/backend/crates/providers/openai/tests/credential/contract/mod.rs index 38e092d0d..1e7bb4333 100644 --- a/backend/crates/providers/openai/tests/credential/contract/mod.rs +++ b/backend/crates/providers/openai/tests/credential/contract/mod.rs @@ -1,6 +1,7 @@ //! OpenAI 凭据合同测试入口，以及凭据编码、持久化与选号约束测试 mod capacity; +mod cooldown_reads; use std::collections::{BTreeMap, BTreeSet}; use std::num::NonZeroU32; diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index 050f8cff9..647280f57 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -52,6 +52,7 @@ type CredentialLoadHook = dyn for<'a> Fn( #[derive(Default)] pub(crate) struct MemoryAccountStore { accounts: Mutex>, + account_reads: AtomicUsize, quota_reads: AtomicUsize, fail_provider_listing: AtomicBool, pending_account_reads: bool, @@ -182,6 +183,10 @@ impl MemoryAccountStore { self.quota_reads.load(Ordering::SeqCst) } + pub(crate) fn account_reads(&self) -> usize { + self.account_reads.load(Ordering::SeqCst) + } + pub(crate) fn has_quota(&self, id: &str) -> bool { let Some(id) = ProviderAccountId::new(id).ok() else { return false; @@ -244,6 +249,7 @@ impl ProviderAccountStore for MemoryAccountStore { &self, account: &ProviderAccountId, ) -> Result, StoreError> { + self.account_reads.fetch_add(1, Ordering::SeqCst); if self.pending_account_reads { futures::future::pending::<()>().await; } @@ -1143,6 +1149,9 @@ pub(crate) fn account_policy() -> gateway_core::account::AccountSelectionPolicy .with_openai_account_affinity(gateway_core::account::AccountAffinity::Strict) } +type CooldownReadHook = + dyn Fn(ProviderAccountId) -> BoxFuture<'static, Result<(), ProviderStoreError>> + Send + Sync; + /// 内存 `ProviderCooldownPort`：实现 `read`/`put_if_later` 与容量失败计数 /// 证据（openai selector 429 冷却与容量熔断路径用），scope 变体测试不涉及， /// 返回占位 @@ -1150,6 +1159,7 @@ pub(crate) fn account_policy() -> gateway_core::account::AccountSelectionPolicy pub(crate) struct MemoryCooldownPort { pub(crate) cooldowns: Arc>>, capacity_failures: Arc>>, + before_read: Option>, } impl MemoryCooldownPort { @@ -1158,6 +1168,11 @@ impl MemoryCooldownPort { Self::default() } + pub(crate) fn with_read_hook(mut self, hook: Arc) -> Self { + self.before_read = Some(hook); + self + } + /// 测试断言用：窗口内累计失败次数与在途峰值 #[must_use] pub(crate) fn capacity_evidence(&self, account_id: &ProviderAccountId) -> Option<(u32, u32)> { @@ -1193,6 +1208,9 @@ impl ProviderCooldownPort for MemoryCooldownPort { account_id: &'a ProviderAccountId, ) -> BoxFuture<'a, Result, ProviderStoreError>> { Box::pin(async move { + if let Some(hook) = &self.before_read { + hook(account_id.clone()).await?; + } Ok(self .cooldowns .lock() diff --git a/docs/api.md b/docs/api.md index 270341277..46b5d9492 100644 --- a/docs/api.md +++ b/docs/api.md @@ -680,7 +680,7 @@ call 与账号的绑定保存在网关进程内，未连接时一小时过期； | `GET` | `/api/key-usage/config` | 无 | 当前 Key 的客户端配置凭据 | | `GET` | `/api/key-usage/version` | 无 | “关于”弹窗使用的当前版本号和提交号 | -用量查询支持[页面时间合同](#页面时间合同)，默认 `period=7d`，一次最多 31 天。模型按完整名称匹配； +用量查询支持[页面时间合同](#页面时间合同)，默认 `period=7d`，一次最多 31 天。模型按字面量包含匹配，区分大小写，通配符按普通字符处理 不接受 Key ID、账号、Provider 等范围参数或其他未知字段。页码默认 1，每页默认 20，允许 1–100 条； `kind` 为 `success`（默认）、`error` 或 `all`，沿用使用统计的请求结果筛选：错误包括失败和未完成请求， 全部还包括取消与进行中的请求；一次请求只占一条记录。分页响应为 `{ items, currentPage, pageSize, total }` @@ -2007,9 +2007,9 @@ Dashboard 的 `accountUsage[]` 由后端提供 `usageWindow`、`metricLabel`、` Dashboard 默认 `period=today`，用量、诊断和错误查询默认 `period=7d`，范围参数遵循[页面时间合同](#页面时间合同)。 Dashboard 返回原始 `asOf` 及 `asOfDisplay`，作为当前数据的查询锚点 -请求明细列表可组合页码、时间范围、Provider、Client Key、账号、模型、route、transport、状态码、 -request/response/upstream ID、outcome 与搜索文本。`search` 仅供明细及错误列表使用； -汇总、洞察和诊断不接受搜索参数，管理观测接口不接受路由分组筛选。页面搜索不改变顶部统计范围 +请求明细列表接受 `currentPage`、`pageSize`、时间范围、`provider`、`clientApiKeyName`、`accountName`、`model` 和 `status`。 +汇总和洞察只接受时间范围与 `provider`；诊断另接受 `dimension`、`model` 和 `statusCode`。 +各接口拒绝未知参数，明细筛选不改变顶部统计范围 请求列表的 `status` 可取 `success`、`error`、`all`，默认 `success`；未知值返回 `400`。 `success` 沿用成功用量记录范围，`error` 包含最终 `failed`、`incomplete` 的请求， @@ -2052,14 +2052,11 @@ OpenAI 优先采用服务端 `openai-model` / `x-openai-model` 报告（流内 缺少终态声明时保留首次声明。这些值仅表示上游报告，不作为模型真实性证明，也不参与路由、 聚合或本地计价模型选择。没有保存上游模型报告的记录保持未知 -请求记录列表的 `search` 使用字面量前缀匹配，支持 Client Key ID / 名称、 -账号 ID、账号邮箱与名称、请求 / 上游模型 ID。密钥名称不区分大小写，其他字段区分大小写。 -密钥名称、账号邮箱与名称按请求记录的历史快照检索， -不随当前密钥或账号修改、删除而改变；`%`、`_` 和 `\` 均按普通字符处理，不作为搜索通配符 +请求明细的 `clientApiKeyName`、`accountName` 按字面量包含匹配请求保存的历史名称快照，区分大小写， +不解析为 ID，也不依赖当前配置是否存在。`model` 按字面量包含匹配请求模型或上游模型，`%`、`_` 和 `\` 按普通字符处理 -请求记录的 `requestId`、`upstreamRequestId` 参数按模型执行 ID 或上游 ID 精确查询； -搜索框不匹配请求 ID，只有入口 ID 时需结合时间与入口日志定位。 -请求记录和错误列表支持按密钥名称搜索，不支持密钥值或可见前缀搜索 +运维错误列表的 `search` 使用字面量前缀匹配，不支持密钥值或可见前缀搜索。 +请求详情通过 `/api/admin/usage/records/detail?id=...` 查询 ### 记录范围与统计口径 diff --git a/frontend/src/api/modules/usage.ts b/frontend/src/api/modules/usage.ts index 8e3ba1cc3..322fb4ab7 100644 --- a/frontend/src/api/modules/usage.ts +++ b/frontend/src/api/modules/usage.ts @@ -376,8 +376,6 @@ interface UsageRangeQuery { startTime: string endTime: string provider?: string - model?: string - statusCode?: number } interface PageQuery { @@ -385,13 +383,18 @@ interface PageQuery { pageSize: number } -type UsagePageQuery = UsageRangeQuery & PageQuery & { search?: string, status?: UsageRecordStatus } +type UsagePageQuery = UsageRangeQuery & PageQuery & { + clientApiKeyName?: string + accountName?: string + model?: string + status?: UsageRecordStatus +} interface UsageDetailQuery { id: string } -type UsageDiagnosticsQuery = UsageRangeQuery & { dimension: string } +type UsageDiagnosticsQuery = UsageRangeQuery & { dimension: string, model?: string, statusCode?: number } type DiagnosticsDownloadParams = { requestId: string } | { errorId: string } diff --git a/frontend/src/views/key-usage/index.vue b/frontend/src/views/key-usage/index.vue index 682b51104..0c4a2baee 100644 --- a/frontend/src/views/key-usage/index.vue +++ b/frontend/src/views/key-usage/index.vue @@ -1,6 +1,6 @@