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 @@