diff --git a/backend/crates/gateway-admin/Cargo.toml b/backend/crates/gateway-admin/Cargo.toml index c58b46804..cf91e69d2 100644 --- a/backend/crates/gateway-admin/Cargo.toml +++ b/backend/crates/gateway-admin/Cargo.toml @@ -28,7 +28,7 @@ tracing = "0.1.44" uuid = { version = "1.23.3", features = ["v7"] } cron = "0.17" chrono-tz = "0.10" -tokio = { version = "1.52", features = ["rt", "time", "macros"] } +tokio = { version = "1.52", features = ["rt", "time", "macros", "sync"] } [dev-dependencies] serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] } diff --git a/backend/crates/gateway-admin/src/model/observability.rs b/backend/crates/gateway-admin/src/model/observability.rs index c36774fc2..a6bdc171c 100644 --- a/backend/crates/gateway-admin/src/model/observability.rs +++ b/backend/crates/gateway-admin/src/model/observability.rs @@ -1,6 +1,6 @@ //! 用量、成本、健康与错误诊断的 UTC 语义事实。 -use std::{num::NonZeroU32, str::FromStr}; +use std::str::FromStr; use chrono::{DateTime, TimeDelta, Timelike as _, Utc}; @@ -38,22 +38,6 @@ impl TimeRange { } } -/// 从一开始的观测页码。 -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct PageNumber(NonZeroU32); - -impl PageNumber { - #[must_use] - pub const fn new(value: NonZeroU32) -> Self { - Self(value) - } - - #[must_use] - pub const fn get(self) -> u32 { - self.0.get() - } -} - /// 观测记录的稳定键集游标。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct ObservabilityCursor { @@ -147,8 +131,8 @@ pub struct UsageQuery { pub range: TimeRange, pub filter: UsageFilter, pub cursor: Option, - pub page: PageNumber, pub page_size: PageSize, + pub include_total: bool, } /// 运维错误过滤条件。 @@ -175,8 +159,8 @@ pub struct OpsErrorQuery { pub range: TimeRange, pub filter: OpsErrorFilter, pub cursor: Option, - pub page: PageNumber, pub page_size: PageSize, + pub include_total: bool, } /// 用量诊断维度。 @@ -545,13 +529,60 @@ pub struct DashboardTotals { #[derive(Debug, Clone, PartialEq)] pub struct DashboardObservation { pub range: TimeRange, - pub requests: RequestMetrics, - pub attempts: AttemptMetrics, pub totals: DashboardTotals, pub provider_accounts: AccountPoolMetrics, pub trend: Vec, pub account_usage: Vec, - pub recent_requests: Vec, + pub recent_requests: Vec, +} + +/// 使用记录表格的窄读模型。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UsageListRecord { + pub id: String, + pub endpoint: String, + pub client_transport: String, + pub requested_model_id: Option, + pub provider_kind: Option, + pub provider_account_ref: Option, + pub provider_account_name: Option, + pub provider_account_email: Option, + pub provider_account_authentication_kind: Option, + pub upstream_model_id: Option, + pub upstream_transport: Option, + pub service_tier: Option, + pub input_tokens: Option, + pub output_tokens: Option, + pub cached_tokens: Option, + pub cache_write_tokens: Option, + pub reasoning_tokens: Option, + pub image_input_tokens: Option, + pub image_output_tokens: Option, + pub total_tokens: Option, + pub cost_source: String, + pub cost_amount: Option, + pub cost_currency: Option, + pub billing: Option, + pub transport_decision_wait_ms: Option, + pub connect_ms: Option, + pub headers_ms: Option, + pub first_event_ms: Option, + pub first_reasoning_ms: Option, + pub first_text_ms: Option, + pub first_token_ms: Option, + pub provider_processing_ms: Option, + pub latency_ms: Option, + pub admission_decision_ms: Option, + pub account_selection_wait_ms: Option, + pub capacity_used_slots: Option, + pub capacity_total_slots: Option, + pub client_ip: Option, + pub user_agent: Option, + pub reasoning_effort: Option, + pub reasoning_preset: Option, + pub subagent_kind: Option, + pub compact: bool, + pub started_at: DateTime, } /// 一次完整模型请求的公共观测记录。 @@ -635,8 +666,8 @@ pub struct UsageRecord { /// 用量分页结果。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct UsagePage { - pub items: Vec, - pub total: u64, + pub items: Vec, + pub total: Option, pub next_cursor: Option, } @@ -764,7 +795,7 @@ pub struct OpsError { #[derive(Debug, Clone, PartialEq, Eq)] pub struct OpsErrorPage { pub items: Vec, - pub total: u64, + pub total: Option, pub next_cursor: Option, } diff --git a/backend/crates/gateway-admin/src/use_case/observability.rs b/backend/crates/gateway-admin/src/use_case/observability.rs index e227fbf06..e397ac074 100644 --- a/backend/crates/gateway-admin/src/use_case/observability.rs +++ b/backend/crates/gateway-admin/src/use_case/observability.rs @@ -1,9 +1,10 @@ //! 观测查询、趋势、费用覆盖与健康阈值规则。 use std::{ - collections::{BTreeMap, btree_map::Entry}, + collections::{BTreeMap, HashMap, btree_map::Entry}, str::FromStr, sync::Arc, + time::{Duration as StdDuration, Instant}, }; use async_trait::async_trait; @@ -39,6 +40,58 @@ const HEALTH_TIMELINE_SLOTS: i64 = 24 * 4; const HEALTH_TIMELINE_MIN_SAMPLE_SIZE: u64 = 10; const HEALTH_TIMELINE_UNAVAILABLE_FAILURE_THRESHOLD: u64 = 3; const HEALTH_TIMELINE_STABLE_RELIABILITY: f64 = 99.0; +const DASHBOARD_QUERY_BUCKET_SECONDS: i64 = 2; +const DASHBOARD_QUERY_CACHE_RETENTION: StdDuration = StdDuration::from_secs(2); + +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +struct DashboardQueryKey { + start: DateTime, + end_bucket: i64, +} + +impl DashboardQueryKey { + fn new(range: TimeRange) -> Self { + Self { + start: range.start, + end_bucket: range + .end + .timestamp() + .div_euclid(DASHBOARD_QUERY_BUCKET_SECONDS), + } + } +} + +struct DashboardQueryEntry { + created_at: Instant, + result: Arc>>, +} + +#[derive(Default)] +struct DashboardQueryCache { + entries: tokio::sync::Mutex>, +} + +impl DashboardQueryCache { + async fn result_cell( + &self, + key: DashboardQueryKey, + ) -> Arc>> { + let now = Instant::now(); + let mut entries = self.entries.lock().await; + entries.retain(|_, entry| { + now.duration_since(entry.created_at) < DASHBOARD_QUERY_CACHE_RETENTION + }); + Arc::clone( + &entries + .entry(key) + .or_insert_with(|| DashboardQueryEntry { + created_at: now, + result: Arc::new(tokio::sync::OnceCell::new()), + }) + .result, + ) + } +} /// API 消费的观测控制面服务。 #[async_trait] @@ -75,6 +128,7 @@ pub(crate) struct DefaultObservabilityService { store: Arc, settings: Arc, providers: ProviderAdminRegistry, + dashboard_queries: DashboardQueryCache, } impl DefaultObservabilityService { @@ -88,16 +142,13 @@ impl DefaultObservabilityService { store, settings, providers, + dashboard_queries: DashboardQueryCache::default(), } } -} -#[async_trait] -impl ObservabilityService for DefaultObservabilityService { - async fn dashboard_summary( + async fn load_dashboard_summary( &self, range: TimeRange, - kind: TrendKind, ) -> Result { let observed_at = Utc::now(); let (mut observation, settings, runtime_slots) = futures::try_join!( @@ -106,7 +157,7 @@ impl ObservabilityService for DefaultObservabilityService { self.store.dashboard_runtime_slots(observed_at), ) .map_err(|error| map_store_error(error, "dashboard"))?; - self.enrich_billing(&mut observation.recent_requests); + self.enrich_list_billing(&mut observation.recent_requests); self.enrich_dashboard_quotas(&mut observation.account_usage) .await; let today_start = china_day_start(observation.range.end); @@ -119,11 +170,18 @@ impl ObservabilityService for DefaultObservabilityService { observation.totals.cached_tokens, observation.totals.input_tokens, ); - let average_first_token_latency_ms = average( - observation.requests.first_token_latency_sum_ms, - observation.requests.first_token_latency_count, - ); - let trend = trend(kind, observation.trend.clone())?; + let (first_token_latency_sum_ms, first_token_latency_count) = observation + .trend + .iter() + .fold((0_u64, 0_u64), |(sum, count), point| { + ( + sum.saturating_add(point.metrics.first_token_latency_sum_ms), + count.saturating_add(point.metrics.first_token_latency_count), + ) + }); + let average_first_token_latency_ms = + average(first_token_latency_sum_ms, first_token_latency_count); + let trend = trend(TrendKind::Usage, observation.trend.clone())?; let health_timeline = health_timeline_at(&observation.trend, Utc::now()); let wire_profiles = self.providers.dashboard_wire_profiles(); let max_concurrent_per_account = u64::from(settings.max_concurrent_per_account); @@ -161,6 +219,26 @@ impl ObservabilityService for DefaultObservabilityService { wire_profiles, }) } +} + +#[async_trait] +impl ObservabilityService for DefaultObservabilityService { + async fn dashboard_summary( + &self, + range: TimeRange, + kind: TrendKind, + ) -> Result { + let cell = self + .dashboard_queries + .result_cell(DashboardQueryKey::new(range)) + .await; + let mut result = cell + .get_or_init(|| self.load_dashboard_summary(range)) + .await + .clone()?; + result.trend = trend(kind, result.observation.trend.clone())?; + Ok(result) + } async fn dashboard_trend( &self, @@ -181,7 +259,7 @@ impl ObservabilityService for DefaultObservabilityService { .list_usage_records(query) .await .map_err(|error| map_store_error(error, "usage records"))?; - self.enrich_billing(&mut page.items); + self.enrich_list_billing(&mut page.items); Ok(page) } @@ -194,7 +272,7 @@ impl ObservabilityService for DefaultObservabilityService { .usage_record_detail(request_id) .await .map_err(|error| map_store_error(error, "usage record"))?; - self.enrich_billing(std::slice::from_mut(&mut detail.request)); + self.enrich_detail_billing(std::slice::from_mut(&mut detail.request)); Ok(detail) } @@ -659,7 +737,39 @@ impl DefaultObservabilityService { /// 逐条尽力把可校验的总额升级为完整分解；单条脏数据（非法 Provider kind、 /// 不支持的来源或费用规则失败）只保留该条已存的总额，不影响整页返回。 - fn enrich_billing(&self, records: &mut [crate::model::observability::UsageRecord]) { + fn enrich_list_billing(&self, records: &mut [crate::model::observability::UsageListRecord]) { + for record in records { + let Some(UsageBilling::Total { source, total }) = record.billing.as_ref() else { + continue; + }; + if !matches!(source.as_str(), "calculated" | "provider_reported") { + continue; + } + let (Some(provider), Some(upstream_model_id)) = ( + record.provider_kind.as_deref(), + record.upstream_model_id.as_ref(), + ) else { + continue; + }; + let Ok(provider_kind) = ProviderKind::new(provider.to_owned()) else { + continue; + }; + let input = ProviderBillingInput { + upstream_model_id: upstream_model_id.clone(), + service_tier: record.service_tier.clone(), + input_tokens: record.input_tokens, + output_tokens: record.output_tokens, + cached_tokens: record.cached_tokens, + cache_write_tokens: record.cache_write_tokens, + total: total.clone(), + }; + if let Ok(Some(breakdown)) = self.providers.calculated_billing(&provider_kind, &input) { + record.billing = Some(UsageBilling::Calculated(Box::new(breakdown))); + } + } + } + + fn enrich_detail_billing(&self, records: &mut [crate::model::observability::UsageRecord]) { for record in records { let Some(UsageBilling::Total { source, total }) = record.billing.as_ref() else { continue; diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index f3163d8d2..6fb251e53 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -1,7 +1,10 @@ use std::{ - num::NonZeroU32, str::FromStr as _, - sync::{Arc, Mutex}, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration as StdDuration, }; use async_trait::async_trait; @@ -14,10 +17,10 @@ use gateway_admin::{ observability::{ AccountPoolMetrics, AttemptMetrics, CostCoverage, CurrencyCost, DashboardObservation, DashboardRuntimeSlots, DiagnosticDimension, DiagnosticObservation, Granularity, - HealthStatus, LatencyPercentiles, OpsErrorPage, OpsErrorQuery, PageNumber, - PercentileMilliseconds, RequestMetricPoint, RequestMetrics, RequestOutcome, TimeRange, - TrendKind, UsageBilling, UsageCalculatedBillingFact, UsageDetail, UsageFilter, - UsageOverview, UsagePage, UsageQuery, UsageRecord, china_day_start, + HealthStatus, LatencyPercentiles, OpsErrorPage, OpsErrorQuery, PercentileMilliseconds, + RequestMetricPoint, RequestMetrics, TimeRange, TrendKind, UsageBilling, + UsageCalculatedBillingFact, UsageDetail, UsageFilter, UsageListRecord, UsageOverview, + UsagePage, UsageQuery, china_day_start, }, settings::{ AdminApiKey, AdminApiKeyMutation, ReplaceRuntimeSettings, RotationStrategy, @@ -105,6 +108,40 @@ async fn health_timeline_should_keep_exactly_china_day_quarter_hour_slots() { })); } +#[tokio::test] +async fn dashboard_summary_should_derive_first_token_average_from_trend() { + let now = Utc::now(); + let current_slot = quarter_hour_start(now); + let store = Arc::new(FixtureObservabilityStore::new(observation_range(now))); + store.replace_trend(vec![ + health_metric_point( + current_slot - Duration::minutes(15), + RequestMetrics { + first_token_latency_sum_ms: 120, + first_token_latency_count: 2, + ..RequestMetrics::default() + }, + ), + health_metric_point( + current_slot, + RequestMetrics { + first_token_latency_sum_ms: 180, + first_token_latency_count: 1, + ..RequestMetrics::default() + }, + ), + ]); + let services = observability_services(store).await; + + let summary = services + .observability() + .dashboard_summary(observation_range(now), TrendKind::Usage) + .await + .expect("dashboard summary"); + + assert_eq!(summary.average_first_token_latency_ms, Some(100)); +} + #[tokio::test] async fn health_timeline_should_match_legacy_status_precedence_and_thresholds() { let now = Utc::now(); @@ -160,11 +197,15 @@ async fn health_timeline_should_match_legacy_status_precedence_and_thresholds() ), ]; - for (metrics, expected_status, expected_reliability) in fixtures { + for (index, (metrics, expected_status, expected_reliability)) in + fixtures.into_iter().enumerate() + { store.replace_trend(vec![health_metric_point(current_slot, metrics)]); + let range_end = + now + Duration::seconds(i64::try_from(index).expect("fixture index fits i64") * 2); let timeline = services .observability() - .dashboard_summary(observation_range(now), TrendKind::Usage) + .dashboard_summary(observation_range(range_end), TrendKind::Usage) .await .expect("dashboard summary") .health_timeline; @@ -220,6 +261,53 @@ async fn dashboard_summary_should_share_one_current_observation_time_across_runt assert!(summary_observed_at.is_some_and(|value| value > historical_end)); } +#[tokio::test] +async fn dashboard_summary_should_coalesce_concurrent_requests_in_one_short_window() { + let end_bucket = Utc::now().timestamp().div_euclid(2) * 2; + let first_end = DateTime::from_timestamp(end_bucket, 100_000_000).expect("first end"); + let second_end = DateTime::from_timestamp(end_bucket, 900_000_000).expect("second end"); + let first_range = TimeRange::new(china_day_start(first_end), first_end).expect("first range"); + let second_range = + TimeRange::new(china_day_start(second_end), second_end).expect("second range"); + let store = Arc::new(FixtureObservabilityStore::new(first_range)); + store.set_dashboard_delay(StdDuration::from_millis(20)); + let services = observability_services(store.clone()).await; + let observability = services.observability(); + + let (first, second) = tokio::join!( + observability.dashboard_summary(first_range, TrendKind::Usage), + observability.dashboard_summary(second_range, TrendKind::Errors), + ); + + first.expect("first dashboard summary"); + second.expect("second dashboard summary"); + assert_eq!(store.dashboard_summary_calls(), 1); +} + +#[tokio::test] +async fn dashboard_summary_should_reload_after_the_short_window_changes() { + let end_bucket = Utc::now().timestamp().div_euclid(2) * 2; + let first_end = DateTime::from_timestamp(end_bucket, 100_000_000).expect("first end"); + let second_end = first_end + Duration::seconds(2); + let first_range = TimeRange::new(china_day_start(first_end), first_end).expect("first range"); + let second_range = + TimeRange::new(china_day_start(second_end), second_end).expect("second range"); + let store = Arc::new(FixtureObservabilityStore::new(first_range)); + let services = observability_services(store.clone()).await; + let observability = services.observability(); + + observability + .dashboard_summary(first_range, TrendKind::Usage) + .await + .expect("first dashboard summary"); + observability + .dashboard_summary(second_range, TrendKind::Usage) + .await + .expect("second dashboard summary"); + + assert_eq!(store.dashboard_summary_calls(), 2); +} + #[tokio::test] async fn observability_services_should_calculate_usage_insights_and_diagnostic_shares() { let now = Utc::now(); @@ -457,7 +545,9 @@ struct FixtureObservabilityStore { runtime_slots: Mutex>, summary_observed_at: Mutex>>, slots_observed_at: Mutex>>, - usage_records: Mutex>, + usage_records: Mutex>, + dashboard_delay: Mutex, + dashboard_summary_calls: AtomicUsize, } impl FixtureObservabilityStore { @@ -476,9 +566,19 @@ impl FixtureObservabilityStore { summary_observed_at: Mutex::new(None), slots_observed_at: Mutex::new(None), usage_records: Mutex::new(Vec::new()), + dashboard_delay: Mutex::new(StdDuration::ZERO), + dashboard_summary_calls: AtomicUsize::new(0), } } + fn set_dashboard_delay(&self, delay: StdDuration) { + *self.dashboard_delay.lock().expect("dashboard delay") = delay; + } + + fn dashboard_summary_calls(&self) -> usize { + self.dashboard_summary_calls.load(Ordering::Relaxed) + } + fn replace_trend(&self, trend: Vec) { *self.trend.lock().expect("trend") = trend; } @@ -512,7 +612,7 @@ impl FixtureObservabilityStore { ) } - fn replace_usage_records(&self, records: Vec) { + fn replace_usage_records(&self, records: Vec) { *self.usage_records.lock().expect("usage records") = records; } } @@ -524,14 +624,17 @@ impl ObservabilityStore for FixtureObservabilityStore { range: TimeRange, observed_at: DateTime, ) -> AdminStoreResult { + self.dashboard_summary_calls.fetch_add(1, Ordering::Relaxed); *self .summary_observed_at .lock() .expect("summary observed at") = Some(observed_at); + let dashboard_delay = *self.dashboard_delay.lock().expect("dashboard delay"); + if !dashboard_delay.is_zero() { + tokio::time::sleep(dashboard_delay).await; + } Ok(DashboardObservation { range, - requests: RequestMetrics::default(), - attempts: AttemptMetrics::default(), totals: Default::default(), provider_accounts: AccountPoolMetrics::default(), trend: self.trend.lock().expect("trend").clone(), @@ -577,7 +680,7 @@ impl ObservabilityStore for FixtureObservabilityStore { let total = u64::try_from(items.len()).unwrap_or(u64::MAX); Ok(UsagePage { items, - total, + total: Some(total), next_cursor: None, }) } @@ -681,8 +784,8 @@ fn usage_query(now: DateTime) -> UsageQuery { range: observation_range(now), filter: UsageFilter::default(), cursor: None, - page: PageNumber::new(NonZeroU32::new(1).expect("page number")), page_size: PageSize::new(50).expect("page size"), + include_total: true, } } @@ -691,16 +794,9 @@ fn total_record( provider_kind: Option<&str>, source: &str, now: DateTime, -) -> UsageRecord { - UsageRecord { +) -> UsageListRecord { + UsageListRecord { id: id.to_owned(), - client_api_key_ref: "key_billing".to_owned(), - config_revision: 1, - routing_scope: "all".to_owned(), - routing_group_refs: Vec::new(), - routing_group_names_snapshot: Vec::new(), - protocol: "openai".to_owned(), - operation: "generate".to_owned(), endpoint: "/v1/responses".to_owned(), client_transport: "http_sse".to_owned(), requested_model_id: Some("gpt-5.5".to_owned()), @@ -711,23 +807,7 @@ fn total_record( provider_account_authentication_kind: None, upstream_model_id: Some("gpt-5.5".to_owned()), upstream_transport: None, - http_version: None, - websocket_pool: None, service_tier: None, - provider_metadata_json: None, - attempt_count: 1, - upstream_send_state: "sent".to_owned(), - downstream_committed_at: None, - outcome: RequestOutcome::Succeeded, - client_status_code: Some(200), - upstream_status_code: Some(200), - client_response_id: None, - upstream_request_id: None, - upstream_response_id: None, - error_kind: None, - provider_error_code: None, - error_message: None, - retry_after_ms: None, input_tokens: Some(800), output_tokens: Some(200), cached_tokens: Some(0), @@ -764,14 +844,9 @@ fn total_record( user_agent: None, reasoning_effort: None, reasoning_preset: None, - request_kind: None, subagent_kind: None, compact: false, - image_generation_requested: false, - image_generation_succeeded: None, started_at: now, - deadline_at: now, - completed_at: Some(now), } } diff --git a/backend/crates/gateway-api/src/admin/observability/mod.rs b/backend/crates/gateway-api/src/admin/observability/mod.rs index a1b0a4305..6f8746ccc 100644 --- a/backend/crates/gateway-api/src/admin/observability/mod.rs +++ b/backend/crates/gateway-api/src/admin/observability/mod.rs @@ -1,6 +1,6 @@ //! Dashboard、用量与诊断查询的 wire 映射和固定路由。 -use std::{collections::BTreeMap, num::NonZeroU32}; +use std::collections::BTreeMap; use axum::{Router, extract::State, http::StatusCode, response::IntoResponse, routing::get}; use base64::engine::general_purpose::URL_SAFE_NO_PAD; @@ -11,7 +11,7 @@ use serde_json::Value; use super::presenter::{format_compact_number, format_decimal_currency, format_number}; use super::{ - AdminAuth, AdminEnvelope, AdminError, AdminQuery, AdminResponse, AdminSessionState, PageMeta, + AdminAuth, AdminEnvelope, AdminError, AdminQuery, AdminResponse, AdminSessionState, WireValidationError, wire::map_admin_service_error, }; diff --git a/backend/crates/gateway-api/src/admin/observability/presenter.rs b/backend/crates/gateway-api/src/admin/observability/presenter.rs index 2e87850c9..b94496a27 100644 --- a/backend/crates/gateway-api/src/admin/observability/presenter.rs +++ b/backend/crates/gateway-api/src/admin/observability/presenter.rs @@ -146,6 +146,43 @@ pub(crate) fn token_details(record: &domain::UsageRecord) -> TokenDetailsView { } } +pub(crate) fn usage_list_token_details(record: &domain::UsageListRecord) -> TokenDetailsView { + TokenDetailsView { + input_tokens: record.input_tokens, + output_tokens: record.output_tokens, + cached_tokens: record.cached_tokens, + cache_write_tokens: record.cache_write_tokens, + reasoning_tokens: record.reasoning_tokens, + image_input_tokens: record.image_input_tokens, + image_output_tokens: record.image_output_tokens, + total_tokens: record.total_tokens, + input_tokens_display: record + .input_tokens + .map_or_else(|| "-".to_owned(), format_number), + output_tokens_display: record + .output_tokens + .map_or_else(|| "-".to_owned(), format_number), + cached_tokens_display: record + .cached_tokens + .map_or_else(|| "-".to_owned(), format_compact_number), + cache_write_tokens_display: record + .cache_write_tokens + .map_or_else(|| "-".to_owned(), format_compact_number), + reasoning_tokens_display: record + .reasoning_tokens + .map_or_else(|| "-".to_owned(), format_number), + image_input_tokens_display: record + .image_input_tokens + .map_or_else(|| "-".to_owned(), format_number), + image_output_tokens_display: record + .image_output_tokens + .map_or_else(|| "-".to_owned(), format_number), + total_tokens_display: record + .total_tokens + .map_or_else(|| "-".to_owned(), format_number), + } +} + pub(crate) fn format_money(cost: &domain::CurrencyCost) -> String { format_decimal_currency(cost.amount.as_str(), &cost.currency) } @@ -207,6 +244,57 @@ pub(crate) fn billing_view(billing: Option<&domain::UsageBilling>) -> Option UsageListRecordView { + let token_details = usage_list_token_details(&record); + let billing = billing_view(record.billing.as_ref()); + let created_at_display = china_datetime(&record.started_at); + let model = record + .upstream_model_id + .clone() + .or_else(|| record.requested_model_id.clone()); + UsageListRecordView { + id: record.id, + provider: record.provider_kind, + authentication_kind: record.provider_account_authentication_kind, + account_id: record.provider_account_ref, + account_email: record.provider_account_email, + account_name: record.provider_account_name, + route: record.endpoint, + model, + requested_model: record.requested_model_id, + upstream_model: record.upstream_model_id, + service_tier: record.service_tier, + client_transport: record.client_transport, + upstream_transport: record.upstream_transport, + reasoning_effort: record.reasoning_effort, + reasoning_preset: record.reasoning_preset, + subagent_kind: record.subagent_kind, + compact: record.compact, + token_details, + billing, + latency_details: UsageLatencyDetailsView { + admission_decision_ms: record.admission_decision_ms, + account_selection_wait_ms: record.account_selection_wait_ms, + capacity_used_slots: record.capacity_used_slots, + capacity_total_slots: record.capacity_total_slots, + transport_decision_wait_ms: record.transport_decision_wait_ms, + ws_connect_ms: record.connect_ms, + upstream_headers_ms: record.headers_ms, + first_event_ms: record.first_event_ms, + first_reasoning_ms: record.first_reasoning_ms, + first_text_ms: record.first_text_ms, + first_token_ms: record.first_token_ms, + openai_processing_ms: record.provider_processing_ms, + }, + first_token_latency_ms: record.first_token_ms, + latency_ms: record.latency_ms, + created_at: record.started_at, + created_at_display, + client_ip: record.client_ip, + user_agent: record.user_agent, + } +} + pub(crate) fn usage_record_view(record: domain::UsageRecord) -> UsageRecordView { let tokens = token_details(&record); let billing = billing_view(record.billing.as_ref()); @@ -427,25 +515,12 @@ pub(crate) fn usage_detail_view(detail: domain::UsageDetail) -> UsageRecordDetai } } -pub(crate) fn page_meta(page: u32, page_size: u16, total: u64) -> PageMeta { - let page_size_u64 = u64::from(page_size); - let total_pages = total.saturating_add(page_size_u64 - 1) / page_size_u64; - PageMeta::new( - page, - u32::from(page_size), - total, - u32::try_from(total_pages).unwrap_or(u32::MAX), - ) -} - pub(crate) fn usage_page_view( page: domain::UsagePage, - page_number: u32, - page_size: u16, -) -> Result, WireValidationError> { +) -> Result, WireValidationError> { Ok(PageData { - items: page.items.into_iter().map(usage_record_view).collect(), - page: page_meta(page_number, page_size, page.total), + items: page.items.into_iter().map(usage_list_record_view).collect(), + total: page.total, next_cursor: page .next_cursor .as_ref() @@ -493,12 +568,10 @@ pub(crate) fn ops_error_view(error: domain::OpsError) -> OpsErrorView { pub(crate) fn ops_page_view( page: domain::OpsErrorPage, - page_number: u32, - page_size: u16, ) -> Result, WireValidationError> { Ok(PageData { items: page.items.into_iter().map(ops_error_view).collect(), - page: page_meta(page_number, page_size, page.total), + total: page.total, next_cursor: page .next_cursor .as_ref() @@ -833,7 +906,10 @@ pub(crate) fn dashboard_view( health_timeline: health_timeline_view(health_timeline), wire_profiles: wire_profiles.into_iter().map(wire_profile_view).collect(), account_usage: account_usage_views, - usage_records: recent_requests.into_iter().map(usage_record_view).collect(), + usage_records: recent_requests + .into_iter() + .map(usage_list_record_view) + .collect(), pool_summary: DashboardPoolSummaryView { total: provider_accounts.total, normal: provider_accounts.normal, diff --git a/backend/crates/gateway-api/src/admin/observability/query.rs b/backend/crates/gateway-api/src/admin/observability/query.rs index 83deb06b9..aedf62905 100644 --- a/backend/crates/gateway-api/src/admin/observability/query.rs +++ b/backend/crates/gateway-api/src/admin/observability/query.rs @@ -31,7 +31,6 @@ impl DashboardQuery { #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct UsageQuery { - pub page: Option, pub page_size: Option, pub cursor: Option, pub kind: Option, @@ -54,16 +53,12 @@ pub struct UsageQuery { impl UsageQuery { /// 校验分页字段，不改变仓储层的分页 owner。 - pub fn validate_page(&self) -> Result<(u32, u16), WireValidationError> { - let page = self.page.unwrap_or(1); - if page == 0 { - return Err(WireValidationError::new("page")); - } + pub fn validate_page_size(&self) -> Result { let page_size = self.page_size.unwrap_or(DEFAULT_PAGE_SIZE); if page_size == 0 || page_size > MAX_PAGE_SIZE { return Err(WireValidationError::new("pageSize")); } - Ok((page, page_size)) + Ok(page_size) } /// 校验游标的 wire 边界；编码和排序语义由应用层负责。 @@ -110,7 +105,6 @@ impl DiagnosticsQuery { #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct OpsQuery { - pub page: Option, pub page_size: Option, pub cursor: Option, pub kind: Option, @@ -135,16 +129,12 @@ pub struct OpsQuery { impl OpsQuery { /// 校验分页字段。 - pub fn validate_page(&self) -> Result<(u32, u16), WireValidationError> { - let page = self.page.unwrap_or(1); - if page == 0 { - return Err(WireValidationError::new("page")); - } + pub fn validate_page_size(&self) -> Result { let page_size = self.page_size.unwrap_or(DEFAULT_PAGE_SIZE); if page_size == 0 || page_size > MAX_PAGE_SIZE { return Err(WireValidationError::new("pageSize")); } - Ok((page, page_size)) + Ok(page_size) } /// 校验游标的 wire 边界。 @@ -345,37 +335,23 @@ pub(crate) fn usage_filter(query: &UsageQuery) -> Result Result<(domain::UsageQuery, u32, u16), WireValidationError> { - let (page, page_size) = query.validate_page()?; +pub(crate) fn usage_command(query: &UsageQuery) -> Result { + let page_size = query.validate_page_size()?; query.validate_cursor()?; - let page_number = NonZeroU32::new(page) - .map(domain::PageNumber::new) - .ok_or_else(|| WireValidationError::new("page"))?; let page_size_value = DomainPageSize::new(page_size).map_err(|_| WireValidationError::new("pageSize"))?; - Ok(( - domain::UsageQuery { - range: usage_range(query.start_time.as_deref(), query.end_time.as_deref())?, - filter: usage_filter(query)?, - cursor: decode_observability_cursor(query.cursor.as_deref())?, - page: page_number, - page_size: page_size_value, - }, - page, - page_size, - )) + Ok(domain::UsageQuery { + range: usage_range(query.start_time.as_deref(), query.end_time.as_deref())?, + filter: usage_filter(query)?, + cursor: decode_observability_cursor(query.cursor.as_deref())?, + page_size: page_size_value, + include_total: query.cursor.is_none(), + }) } -pub(crate) fn ops_command( - query: &OpsQuery, -) -> Result<(domain::OpsErrorQuery, u32, u16), WireValidationError> { - let (page, page_size) = query.validate_page()?; +pub(crate) fn ops_command(query: &OpsQuery) -> Result { + let page_size = query.validate_page_size()?; query.validate_cursor()?; - let page_number = NonZeroU32::new(page) - .map(domain::PageNumber::new) - .ok_or_else(|| WireValidationError::new("page"))?; let page_size_value = DomainPageSize::new(page_size).map_err(|_| WireValidationError::new("pageSize"))?; let status_code = parse_status( @@ -384,31 +360,27 @@ pub(crate) fn ops_command( .or(query.client_status_code) .or(query.status_code), )?; - Ok(( - domain::OpsErrorQuery { - range: usage_range(query.start_time.as_deref(), query.end_time.as_deref())?, - filter: domain::OpsErrorFilter { - client_api_key_ref: non_empty(query.client_api_key_id.clone()), - request_id: non_empty(query.request_id.clone()), - provider_kind: non_empty(query.provider.clone()), - provider_account_ref: non_empty(query.account_id.clone()), - operation: non_empty(query.route.clone()).or_else(|| non_empty(query.kind.clone())), - model: non_empty(query.model.clone()), - 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()), - failure_kind: non_empty(query.failure_class.clone()), - status_code, - search: non_empty(query.search.clone()), - }, - cursor: decode_observability_cursor(query.cursor.as_deref())?, - page: page_number, - page_size: page_size_value, + Ok(domain::OpsErrorQuery { + range: usage_range(query.start_time.as_deref(), query.end_time.as_deref())?, + filter: domain::OpsErrorFilter { + client_api_key_ref: non_empty(query.client_api_key_id.clone()), + request_id: non_empty(query.request_id.clone()), + provider_kind: non_empty(query.provider.clone()), + provider_account_ref: non_empty(query.account_id.clone()), + operation: non_empty(query.route.clone()).or_else(|| non_empty(query.kind.clone())), + model: non_empty(query.model.clone()), + 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()), + failure_kind: non_empty(query.failure_class.clone()), + status_code, + search: non_empty(query.search.clone()), }, - page, - page_size, - )) + cursor: decode_observability_cursor(query.cursor.as_deref())?, + page_size: page_size_value, + include_total: query.cursor.is_none(), + }) } pub(crate) fn decode_observability_cursor( diff --git a/backend/crates/gateway-api/src/admin/observability/routes.rs b/backend/crates/gateway-api/src/admin/observability/routes.rs index c7b784b35..a5a4fbc2d 100644 --- a/backend/crates/gateway-api/src/admin/observability/routes.rs +++ b/backend/crates/gateway-api/src/admin/observability/routes.rs @@ -84,14 +84,14 @@ pub(crate) async fn usage_records( where S: AdminSessionState + Send + Sync, { - let (command, page, page_size) = usage_command(&query).map_err(map_wire_error)?; + let command = usage_command(&query).map_err(map_wire_error)?; let result = state .admin_services() .observability() .usage_records(command) .await .map_err(map_service_error)?; - let data = usage_page_view(result, page, page_size).map_err(|_| AdminError::internal())?; + let data = usage_page_view(result).map_err(|_| AdminError::internal())?; Ok(AdminResponse::new(StatusCode::OK, AdminEnvelope::ok(data))) } @@ -200,13 +200,13 @@ pub(crate) async fn ops_errors( where S: AdminSessionState + Send + Sync, { - let (command, page, page_size) = ops_command(&query).map_err(map_wire_error)?; + let command = ops_command(&query).map_err(map_wire_error)?; let result = state .admin_services() .observability() .ops_errors(command) .await .map_err(map_service_error)?; - let data = ops_page_view(result, page, page_size).map_err(|_| AdminError::internal())?; + let data = ops_page_view(result).map_err(|_| AdminError::internal())?; Ok(AdminResponse::new(StatusCode::OK, AdminEnvelope::ok(data))) } diff --git a/backend/crates/gateway-api/src/admin/observability/wire.rs b/backend/crates/gateway-api/src/admin/observability/wire.rs index 960c1655a..1b159ca63 100644 --- a/backend/crates/gateway-api/src/admin/observability/wire.rs +++ b/backend/crates/gateway-api/src/admin/observability/wire.rs @@ -15,7 +15,7 @@ pub struct CursorWire { #[serde(rename_all = "camelCase")] pub struct PageData { pub items: Vec, - pub page: PageMeta, + pub total: Option, pub next_cursor: Option, } @@ -77,7 +77,39 @@ pub struct BillingView { pub multiplier_display: String, } -/// 单条逻辑请求展示。 +/// 使用记录表格的窄展示。 +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct UsageListRecordView { + pub id: String, + pub provider: Option, + pub authentication_kind: Option, + pub account_id: Option, + pub account_email: Option, + pub account_name: Option, + pub route: String, + pub model: Option, + pub requested_model: Option, + pub upstream_model: Option, + pub service_tier: Option, + pub client_transport: String, + pub upstream_transport: Option, + pub reasoning_effort: Option, + pub reasoning_preset: Option, + pub subagent_kind: Option, + pub compact: bool, + pub token_details: TokenDetailsView, + pub billing: Option, + pub latency_details: UsageLatencyDetailsView, + pub first_token_latency_ms: Option, + pub latency_ms: Option, + pub created_at: DateTime, + pub created_at_display: String, + pub client_ip: Option, + pub user_agent: Option, +} + +/// 单条逻辑请求详情展示。 #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct UsageRecordView { @@ -526,7 +558,7 @@ pub struct DashboardDataView { pub health_timeline: HealthTimelineView, pub wire_profiles: Vec, pub account_usage: Vec, - pub usage_records: Vec, + pub usage_records: Vec, pub pool_summary: DashboardPoolSummaryView, pub capacity_info: DashboardCapacityInfoView, pub rotation_strategy: String, diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 912cce76d..b64ffe2d9 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -32,7 +32,7 @@ use gateway_admin::{ observability::{ DashboardObservation, DecimalAmount, DiagnosticDimension, DiagnosticObservation, OpsError, OpsErrorPage, OpsErrorQuery, RequestMetricPoint, TimeRange, UsageDetail, - UsageFilter, UsageOverview, UsagePage, UsageQuery, UsageRecord, + UsageFilter, UsageListRecord, UsageOverview, UsagePage, UsageQuery, }, provider_credentials::{ AuthorizationCommit, AuthorizationStarted, CompleteAuthorization, CredentialDetails, @@ -88,7 +88,7 @@ pub(super) struct AdminTestFixture { pub services: AdminServices, pub auth: Arc, pub settings: Arc, - pub usage_records: Arc>>, + pub usage_records: Arc>>, pub usage_detail: Arc>>, pub diagnostics: Arc>>, pub ops_errors: Arc>>, @@ -661,7 +661,7 @@ impl ClientKeyStore for MemoryClientKeyStore { } struct UnusedStore { - usage_records: Arc>>, + usage_records: Arc>>, usage_detail: Arc>>, diagnostics: Arc>>, ops_errors: Arc>>, @@ -855,10 +855,10 @@ impl ObservabilityStore for UnusedStore { Err(unavailable("usage billing facts")) } - async fn list_usage_records(&self, _: UsageQuery) -> AdminStoreResult { + async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult { let items = self.usage_records.lock().expect("usage records").clone(); Ok(UsagePage { - total: items.len() as u64, + total: query.include_total.then_some(items.len() as u64), items, next_cursor: None, }) @@ -885,10 +885,10 @@ impl ObservabilityStore for UnusedStore { Ok(self.diagnostics.lock().expect("diagnostics").clone()) } - async fn list_ops_errors(&self, _: OpsErrorQuery) -> AdminStoreResult { + async fn list_ops_errors(&self, query: OpsErrorQuery) -> AdminStoreResult { let items = self.ops_errors.lock().expect("ops errors").clone(); Ok(OpsErrorPage { - total: items.len() as u64, + total: query.include_total.then_some(items.len() as u64), items, next_cursor: None, }) diff --git a/backend/crates/gateway-api/tests/admin/observability/query.rs b/backend/crates/gateway-api/tests/admin/observability/query.rs index 374594040..0e9e082f0 100644 --- a/backend/crates/gateway-api/tests/admin/observability/query.rs +++ b/backend/crates/gateway-api/tests/admin/observability/query.rs @@ -21,25 +21,28 @@ fn dashboard_query_should_reject_unknown_trend_kind() { #[test] fn usage_query_should_bound_page_size_and_cursor() { let query: UsageQuery = serde_json::from_value(json!({ - "page": 2, "pageSize": 100, "cursor": "opaque" })) .unwrap(); - assert_eq!(query.validate_page().unwrap(), (2, 100)); + assert_eq!(query.validate_page_size().unwrap(), 100); assert!(query.validate_cursor().is_ok()); } #[test] -fn usage_query_should_reject_zero_page() { - let query: UsageQuery = serde_json::from_value(json!({"page": 0})).unwrap(); - assert_eq!(query.validate_page().unwrap_err().field(), "page"); +fn usage_query_should_reject_removed_total_toggle() { + assert!(serde_json::from_value::(json!({"includeTotal": true})).is_err()); +} + +#[test] +fn usage_query_should_reject_removed_offset_page_contract() { + assert!(serde_json::from_value::(json!({"page": 1})).is_err()); } #[test] fn ops_query_should_reject_page_size_above_terminal_limit() { let query: OpsQuery = serde_json::from_value(json!({"pageSize": 101})).unwrap(); - assert_eq!(query.validate_page().unwrap_err().field(), "pageSize"); + assert_eq!(query.validate_page_size().unwrap_err().field(), "pageSize"); } #[test] diff --git a/backend/crates/gateway-api/tests/admin/observability/response.rs b/backend/crates/gateway-api/tests/admin/observability/response.rs index 63b3a2c17..1e0377434 100644 --- a/backend/crates/gateway-api/tests/admin/observability/response.rs +++ b/backend/crates/gateway-api/tests/admin/observability/response.rs @@ -2,7 +2,6 @@ use chrono::{TimeZone, Utc}; use gateway_admin::model::observability::DesktopReleaseStatus; -use gateway_api::admin::PageMeta; use gateway_api::admin::observability::{ BillingView, CostCoverageView, CursorWire, DashboardAccountRequestBucketView, DashboardAccountUsageView, DashboardDesktopReleaseStatusView, DashboardWireAttributeView, @@ -15,11 +14,11 @@ use serde_json::json; fn usage_page_should_keep_terminal_camel_case_shape() { let data = PageData { items: vec![json!({"id": "request_1"})], - page: PageMeta::new(1, 50, 1, 1), + total: Some(1), next_cursor: Some("cursor".to_owned()), }; let value = serde_json::to_value(data).unwrap(); - assert_eq!(value["page"]["pageSize"], 50); + assert_eq!(value["total"], 1); assert_eq!(value["nextCursor"], "cursor"); } @@ -209,7 +208,7 @@ async fn dashboard_summary_should_include_quota_exhaustion_in_unavailable_headli }; use chrono::{Duration, Utc}; use gateway_admin::model::observability::{ - AccountPoolMetrics, AttemptMetrics, DashboardObservation, RequestMetrics, TimeRange, + AccountPoolMetrics, DashboardObservation, TimeRange, }; use gateway_api::admin::observability; use tower::ServiceExt as _; @@ -224,8 +223,6 @@ async fn dashboard_summary_should_include_quota_exhaustion_in_unavailable_headli .lock() .expect("dashboard observation") = Some(DashboardObservation { range: TimeRange::new(now - Duration::hours(1), now).expect("dashboard range"), - requests: RequestMetrics::default(), - attempts: AttemptMetrics::default(), totals: Default::default(), provider_accounts: AccountPoolMetrics { total: 1002, @@ -293,8 +290,7 @@ async fn dashboard_summary_should_use_lifetime_totals_for_card_footers() { }; use chrono::{Duration, Utc}; use gateway_admin::model::observability::{ - AccountPoolMetrics, AttemptMetrics, DashboardObservation, DashboardTotals, RequestMetrics, - TimeRange, + AccountPoolMetrics, DashboardObservation, DashboardTotals, TimeRange, }; use gateway_api::admin::observability; use tower::ServiceExt as _; @@ -304,20 +300,11 @@ async fn dashboard_summary_should_use_lifetime_totals_for_card_footers() { let fixture = AdminTestFixture::new().await; fixture.auth.insert_session("valid-session"); let now = Utc::now(); - let range_requests = RequestMetrics { - request_count: 2, - input_tokens: 8, - cached_tokens: 2, - total_tokens: 12, - ..RequestMetrics::default() - }; *fixture .dashboard_observation .lock() .expect("dashboard observation") = Some(DashboardObservation { range: TimeRange::new(now - Duration::hours(1), now).expect("dashboard range"), - requests: range_requests, - attempts: AttemptMetrics::default(), totals: DashboardTotals { request_count: 42, input_tokens: 800, @@ -374,8 +361,7 @@ async fn dashboard_summary_should_default_to_current_china_day() { }; use chrono::{Duration, Utc}; use gateway_admin::model::observability::{ - AccountPoolMetrics, AttemptMetrics, DashboardObservation, RequestMetrics, TimeRange, - china_day_start, + AccountPoolMetrics, DashboardObservation, TimeRange, china_day_start, }; use gateway_api::admin::observability; use tower::ServiceExt as _; @@ -390,8 +376,6 @@ async fn dashboard_summary_should_default_to_current_china_day() { .lock() .expect("dashboard observation") = Some(DashboardObservation { range: TimeRange::new(now - Duration::hours(1), now).expect("dashboard range"), - requests: RequestMetrics::default(), - attempts: AttemptMetrics::default(), totals: Default::default(), provider_accounts: AccountPoolMetrics::default(), trend: Vec::new(), @@ -798,7 +782,7 @@ async fn usage_route_should_forward_a_bounded_unknown_outcome_filter() { } #[tokio::test] -async fn usage_route_should_expose_image_and_websocket_facts() { +async fn usage_route_should_expose_table_facts_without_detail_payload() { use std::str::FromStr as _; use axum::{ @@ -807,8 +791,7 @@ async fn usage_route_should_expose_image_and_websocket_facts() { }; use chrono::Utc; use gateway_admin::model::observability::{ - CalculatedBillingBreakdown, CurrencyCost, DecimalAmount, RequestOutcome, UsageBilling, - UsageRecord, + CalculatedBillingBreakdown, CurrencyCost, DecimalAmount, UsageBilling, UsageListRecord, }; use gateway_api::admin::observability; use tower::ServiceExt as _; @@ -825,15 +808,8 @@ async fn usage_route_should_expose_image_and_websocket_facts() { .usage_records .lock() .expect("usage records") - .push(UsageRecord { + .push(UsageListRecord { id: "request_endpoint".to_owned(), - client_api_key_ref: "key_endpoint".to_owned(), - config_revision: 1, - routing_scope: "groups".to_owned(), - routing_group_refs: vec!["grp_usage".to_owned()], - routing_group_names_snapshot: vec!["Usage group".to_owned()], - protocol: "openai".to_owned(), - operation: "generate".to_owned(), endpoint: "/v1/responses".to_owned(), client_transport: "websocket".to_owned(), requested_model_id: Some("grok-4.5".to_owned()), @@ -844,30 +820,7 @@ async fn usage_route_should_expose_image_and_websocket_facts() { provider_account_authentication_kind: Some("oauth".to_owned()), upstream_model_id: Some("grok-4.5".to_owned()), upstream_transport: Some("http_sse".to_owned()), - http_version: Some("h2".to_owned()), - websocket_pool: Some("reuse".to_owned()), service_tier: Some("default".to_owned()), - provider_metadata_json: Some( - serde_json::json!({ - "effectiveModel": "grok-4.5", - "requestSummary": {"inputItemsCount": 2}, - "transport": "must-not-overwrite-core", - }) - .to_string(), - ), - attempt_count: 1, - upstream_send_state: "sent".to_owned(), - downstream_committed_at: None, - outcome: RequestOutcome::Succeeded, - client_status_code: Some(200), - upstream_status_code: Some(200), - client_response_id: Some("resp_usage_metadata".to_owned()), - upstream_request_id: Some("upstream_usage_metadata".to_owned()), - upstream_response_id: None, - error_kind: None, - provider_error_code: None, - error_message: None, - retry_after_ms: None, input_tokens: Some(1), output_tokens: Some(1), cached_tokens: Some(0), @@ -895,31 +848,26 @@ async fn usage_route_should_expose_image_and_websocket_facts() { multiplier_percent: 100, }, ))), - transport_decision_wait_ms: Some(7), - connect_ms: Some(11), - headers_ms: Some(13), + transport_decision_wait_ms: Some(4), + connect_ms: Some(5), + headers_ms: Some(7), first_event_ms: Some(17), - first_reasoning_ms: Some(19), - first_text_ms: Some(23), - first_token_ms: Some(19), - provider_processing_ms: Some(29), + first_reasoning_ms: Some(18), + first_text_ms: Some(19), + first_token_ms: Some(20), + provider_processing_ms: Some(21), latency_ms: Some(31), - admission_decision_ms: Some(3), - account_selection_wait_ms: Some(5), - capacity_used_slots: Some(7), - capacity_total_slots: Some(10), - client_ip: Some("127.0.0.1".to_owned()), - user_agent: Some("usage-metadata-test".to_owned()), + admission_decision_ms: Some(1), + account_selection_wait_ms: Some(2), + capacity_used_slots: Some(3), + capacity_total_slots: Some(8), + client_ip: Some("192.0.2.10".to_owned()), + user_agent: Some("codex-cli/1.0".to_owned()), reasoning_effort: Some("max".to_owned()), reasoning_preset: Some("ultra".to_owned()), - request_kind: Some("review".to_owned()), subagent_kind: Some("worker".to_owned()), compact: true, - image_generation_requested: true, - image_generation_succeeded: Some(true), started_at: Utc::now(), - deadline_at: Utc::now(), - completed_at: Some(Utc::now()), }); let response = observability::router::() .with_state(fixture.state()) @@ -959,9 +907,6 @@ async fn usage_route_should_expose_image_and_websocket_facts() { assert_eq!( serde_json::json!({ "route": value["data"]["items"][0]["route"], - "routingScope": value["data"]["items"][0]["routingScope"], - "routingGroupRefs": value["data"]["items"][0]["routingGroupRefs"], - "routingGroupNamesSnapshot": value["data"]["items"][0]["routingGroupNamesSnapshot"], "serviceTier": value["data"]["items"][0]["serviceTier"], "accountId": value["data"]["items"][0]["accountId"], "accountName": value["data"]["items"][0]["accountName"], @@ -969,33 +914,22 @@ async fn usage_route_should_expose_image_and_websocket_facts() { "authenticationKind": value["data"]["items"][0]["authenticationKind"], "imageInputTokens": value["data"]["items"][0]["tokenDetails"]["imageInputTokens"], "imageOutputTokens": value["data"]["items"][0]["tokenDetails"]["imageOutputTokens"], - "websocketPool": value["data"]["items"][0]["websocketPool"], - "imageGenerationRequested": value["data"]["items"][0]["imageGenerationRequested"], - "imageGenerationSucceeded": value["data"]["items"][0]["imageGenerationSucceeded"], "requestedModel": value["data"]["items"][0]["requestedModel"], "upstreamModel": value["data"]["items"][0]["upstreamModel"], - "clientIp": value["data"]["items"][0]["clientIp"], - "userAgent": value["data"]["items"][0]["userAgent"], "reasoningEffort": value["data"]["items"][0]["reasoningEffort"], "reasoningPreset": value["data"]["items"][0]["reasoningPreset"], - "requestKind": value["data"]["items"][0]["requestKind"], "subagentKind": value["data"]["items"][0]["subagentKind"], "compact": value["data"]["items"][0]["compact"], "clientTransport": value["data"]["items"][0]["clientTransport"], "upstreamTransport": value["data"]["items"][0]["upstreamTransport"], - "transportDecisionWaitMs": value["data"]["items"][0]["latencyDetails"]["transportDecisionWaitMs"], - "wsConnectMs": value["data"]["items"][0]["latencyDetails"]["wsConnectMs"], - "firstReasoningMs": value["data"]["items"][0]["latencyDetails"]["firstReasoningMs"], - "firstTextMs": value["data"]["items"][0]["latencyDetails"]["firstTextMs"], - "firstTokenMs": value["data"]["items"][0]["latencyDetails"]["firstTokenMs"], - "effectiveModel": value["data"]["items"][0]["metadata"]["effectiveModel"], - "requestSummary": value["data"]["items"][0]["metadata"]["requestSummary"], + "latencyDetails": value["data"]["items"][0]["latencyDetails"], + "firstTokenLatencyMs": value["data"]["items"][0]["firstTokenLatencyMs"], + "latencyMs": value["data"]["items"][0]["latencyMs"], + "clientIp": value["data"]["items"][0]["clientIp"], + "userAgent": value["data"]["items"][0]["userAgent"], }), serde_json::json!({ "route": "/v1/responses", - "routingScope": "groups", - "routingGroupRefs": ["grp_usage"], - "routingGroupNamesSnapshot": ["Usage group"], "serviceTier": "default", "accountId": "acct_snapshot", "accountName": "Snapshot Alpha", @@ -1003,27 +937,33 @@ async fn usage_route_should_expose_image_and_websocket_facts() { "authenticationKind": "oauth", "imageInputTokens": 31, "imageOutputTokens": 9, - "websocketPool": {"kind": "reuse"}, - "imageGenerationRequested": true, - "imageGenerationSucceeded": true, "requestedModel": "grok-4.5", "upstreamModel": "grok-4.5", - "clientIp": "127.0.0.1", - "userAgent": "usage-metadata-test", "reasoningEffort": "max", "reasoningPreset": "ultra", - "requestKind": "review", "subagentKind": "worker", "compact": true, "clientTransport": "websocket", "upstreamTransport": "http_sse", - "transportDecisionWaitMs": 7, - "wsConnectMs": 11, - "firstReasoningMs": 19, - "firstTextMs": 23, - "firstTokenMs": 19, - "effectiveModel": "grok-4.5", - "requestSummary": {"inputItemsCount": 2}, + "latencyDetails": { + "admissionDecisionMs": 1, + "accountSelectionWaitMs": 2, + "capacityUsedSlots": 3, + "capacityTotalSlots": 8, + "transportDecisionWaitMs": 4, + "wsConnectMs": 5, + "upstreamHeadersMs": 7, + "firstEventMs": 17, + "firstReasoningMs": 18, + "firstTextMs": 19, + "firstTokenMs": 20, + "openaiProcessingMs": 21, + }, + "firstTokenLatencyMs": 20, + "latencyMs": 31, + "clientIp": "192.0.2.10", + "userAgent": "codex-cli/1.0", }) ); + assert!(value["data"]["items"][0].get("metadata").is_none()); } diff --git a/backend/crates/gateway-store/src/bundle.rs b/backend/crates/gateway-store/src/bundle.rs index ea1caef3e..7ff89117c 100644 --- a/backend/crates/gateway-store/src/bundle.rs +++ b/backend/crates/gateway-store/src/bundle.rs @@ -120,6 +120,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult { pool.clone(), Some(credential_leases.clone()), Some(Arc::clone(&cooldowns) as Arc), + observability_query_budget, )), Arc::new(AdminSettingsStoreAdapter { control_plane: postgres::PgControlPlaneRepository::new(pool.clone()), 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 6cd110ff7..92e396fcf 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -16,12 +16,21 @@ use crate::redis::{CredentialLeaseRepository as _, RedisCredentialLeaseRepositor pub struct PgObservabilityRepository { pool: PgPool, cooldowns: Option>, + query_budget: ObservabilityQueryBudget, } impl PgObservabilityRepository { #[must_use] - pub const fn new(pool: PgPool, cooldowns: Option>) -> Self { - Self { pool, cooldowns } + pub fn new( + pool: PgPool, + cooldowns: Option>, + query_budget: ObservabilityQueryBudget, + ) -> Self { + Self { + pool, + cooldowns, + query_budget, + } } /// 从账号事实和当前冷却一次性派生 Dashboard 五态；不在 SQL 中复制状态机。 @@ -29,8 +38,13 @@ impl PgObservabilityRepository { &self, observed_at: DateTime, ) -> StoreResult<(ProviderAccountMetrics, Vec)> { - let accounts = PgProviderAccountRepository::new(self.pool.clone()) - .list_provider_accounts(None, true) + let repository = PgProviderAccountRepository::new(self.pool.clone()); + let accounts = self + .query_budget + .run( + "load dashboard provider accounts", + repository.list_provider_accounts(None, true), + ) .await?; let now = observed_at.into(); let rate_limited_until = @@ -85,9 +99,10 @@ impl PgAdminObservabilityStore { pool: PgPool, runtime_signals: Option, cooldowns: Option>, + query_budget: ObservabilityQueryBudget, ) -> Self { Self { - repository: PgObservabilityRepository::new(pool, cooldowns), + repository: PgObservabilityRepository::new(pool, cooldowns, query_budget), runtime_signals, } } @@ -111,25 +126,33 @@ impl ObservabilityRepository for PgObservabilityRepository { ..UsageRecordFilter::default() }, cursor: None, - page: ObservabilityPageNumber::new(1)?, page_size: ObservabilityPageSize::new(10)?, + include_total: false, }; - // 每批最多三路并发，既缩短概览渲染路径，也不与数据面写路径争抢整个连接池。 - let (requests, attempts, totals) = futures::try_join!( - request_metrics(&self.pool, range, &filter), - attempt_metrics(&self.pool, range, &filter), - dashboard_totals(&self.pool), + // 每条 SQL 独立取一个全局观测槽位，避免整包预留造成队头阻塞。 + let (totals, (provider_accounts, _)) = futures::try_join!( + self.query_budget.run( + "load dashboard lifetime totals", + dashboard_totals(&self.pool) + ), + self.account_status_snapshot(observed_at), )?; - let (provider_accounts, _) = self.account_status_snapshot(observed_at).await?; let (trend, account_usage, recent_requests) = futures::try_join!( - request_metric_series(&self.pool, range, &filter), - provider_account_usage(&self.pool, account_usage_query), - async { Ok(list_usage_records(&self.pool, recent_query).await?.items) }, + self.query_budget.run( + "load dashboard request trend", + dashboard_request_metric_series(&self.pool, range, &filter), + ), + self.query_budget.run( + "load dashboard account usage", + provider_account_usage(&self.pool, account_usage_query), + ), + self.query_budget + .run("load dashboard recent requests", async { + Ok(list_usage_records(&self.pool, recent_query).await?.items) + }), )?; Ok(DashboardObservation { range, - requests, - attempts, totals, provider_accounts, trend, @@ -142,7 +165,12 @@ impl ObservabilityRepository for PgObservabilityRepository { &self, range: ObservabilityRange, ) -> StoreResult> { - request_metric_series(&self.pool, range, &UsageRecordFilter::default()).await + self.query_budget + .run( + "load dashboard request trend", + dashboard_request_metric_series(&self.pool, range, &UsageRecordFilter::default()), + ) + .await } async fn usage_trend( @@ -150,7 +178,12 @@ impl ObservabilityRepository for PgObservabilityRepository { range: ObservabilityRange, filter: UsageRecordFilter, ) -> StoreResult> { - request_metric_series(&self.pool, range, &filter).await + self.query_budget + .run( + "load usage request trend", + request_metric_series(&self.pool, range, &filter), + ) + .await } async fn usage_calculated_billing_facts( @@ -158,22 +191,39 @@ impl ObservabilityRepository for PgObservabilityRepository { range: ObservabilityRange, filter: UsageRecordFilter, ) -> StoreResult> { - calculated_usage_billing_facts(&self.pool, range, &filter).await + self.query_budget + .run( + "load calculated usage billing facts", + calculated_usage_billing_facts(&self.pool, range, &filter), + ) + .await } async fn provider_account_usage( &self, query: ProviderAccountUsageQuery, ) -> StoreResult> { - provider_account_usage(&self.pool, query).await + self.query_budget + .run( + "load provider account usage", + provider_account_usage(&self.pool, query), + ) + .await } async fn list_usage_records(&self, query: UsageRecordQuery) -> StoreResult { - list_usage_records(&self.pool, query).await + self.query_budget + .run("list usage records", list_usage_records(&self.pool, query)) + .await } async fn usage_record_detail(&self, request_id: &str) -> StoreResult { - usage_record_detail(&self.pool, request_id).await + self.query_budget + .run( + "load usage record detail", + usage_record_detail(&self.pool, request_id), + ) + .await } async fn usage_summary( @@ -182,9 +232,27 @@ impl ObservabilityRepository for PgObservabilityRepository { filter: UsageRecordFilter, ) -> StoreResult { filter.validate()?; - let requests = request_metrics(&self.pool, range, &filter).await?; - let attempts = attempt_metrics(&self.pool, range, &filter).await?; - let providers = provider_observations(&self.pool, range, &filter).await?; + let requests = self + .query_budget + .run( + "load usage request metrics", + request_metrics(&self.pool, range, &filter), + ) + .await?; + let attempts = self + .query_budget + .run( + "load usage attempt metrics", + attempt_metrics(&self.pool, range, &filter), + ) + .await?; + let providers = self + .query_budget + .run( + "load usage provider observations", + provider_observations(&self.pool, range, &filter), + ) + .await?; Ok(UsageOverview { range, requests, @@ -199,11 +267,18 @@ impl ObservabilityRepository for PgObservabilityRepository { filter: UsageRecordFilter, dimension: DiagnosticDimension, ) -> StoreResult> { - usage_diagnostics(&self.pool, range, &filter, dimension).await + self.query_budget + .run( + "load usage diagnostics", + usage_diagnostics(&self.pool, range, &filter, dimension), + ) + .await } async fn list_ops_errors(&self, query: OpsErrorQuery) -> StoreResult { - list_ops_errors(&self.pool, query).await + self.query_budget + .run("list ops errors", list_ops_errors(&self.pool, query)) + .await } } diff --git a/backend/crates/gateway-store/src/postgres/observability/mapping.rs b/backend/crates/gateway-store/src/postgres/observability/mapping.rs index bb294b593..ee42d675e 100644 --- a/backend/crates/gateway-store/src/postgres/observability/mapping.rs +++ b/backend/crates/gateway-store/src/postgres/observability/mapping.rs @@ -43,9 +43,9 @@ pub(crate) fn store_usage_query( range: store_range(query.range)?, filter: store_usage_filter(query.filter), cursor: query.cursor.map(store_cursor).transpose()?, - page: ObservabilityPageNumber::new(query.page.get()).map_err(observability_error)?, page_size: ObservabilityPageSize::new(query.page_size.get()) .map_err(observability_error)?, + include_total: query.include_total, }) } @@ -76,9 +76,9 @@ pub(crate) fn store_ops_error_query( range: store_range(query.range)?, filter: store_ops_error_filter(query.filter), cursor: query.cursor.map(store_cursor).transpose()?, - page: ObservabilityPageNumber::new(query.page.get()).map_err(observability_error)?, page_size: ObservabilityPageSize::new(query.page_size.get()) .map_err(observability_error)?, + include_total: query.include_total, }) } @@ -101,8 +101,6 @@ pub(crate) fn admin_dashboard_observation( ) -> AdminStoreResult { let DashboardObservation { range, - requests, - attempts, totals, provider_accounts, trend, @@ -111,8 +109,6 @@ pub(crate) fn admin_dashboard_observation( } = observation; Ok(admin_observability::DashboardObservation { range: admin_range(range), - requests: admin_request_metrics(requests)?, - attempts: admin_attempt_metrics(attempts)?, totals: admin_dashboard_totals(totals)?, provider_accounts: admin_account_pool_metrics(provider_accounts), trend: trend @@ -125,7 +121,7 @@ pub(crate) fn admin_dashboard_observation( .collect::>()?, recent_requests: recent_requests .into_iter() - .map(admin_usage_record) + .map(admin_usage_list_record) .collect::>()?, }) } @@ -411,7 +407,7 @@ pub(crate) fn admin_usage_page( items: page .items .into_iter() - .map(admin_usage_record) + .map(admin_usage_list_record) .collect::>()?, total: page.total, next_cursor: page.next_cursor.map(admin_cursor), @@ -427,6 +423,73 @@ pub(crate) fn admin_cursor( } } +pub(crate) fn admin_usage_list_record( + record: UsageListRecord, +) -> AdminStoreResult { + let billing = match (&record.cost_amount, &record.cost_currency) { + (Some(amount), Some(currency)) => Some(admin_observability::UsageBilling::Total { + source: record.cost_source.clone(), + total: admin_observability::CurrencyCost { + currency: currency.clone(), + amount: admin_decimal_amount(amount.clone())?, + }, + }), + (None, None) => None, + _ => { + return Err(observability_error(StoreError::InvalidData { + entity: "observability request billing", + message: "cost amount and currency must be present together".to_owned(), + })); + } + }; + Ok(admin_observability::UsageListRecord { + id: record.id, + endpoint: record.endpoint, + client_transport: record.client_transport, + requested_model_id: record.requested_model_id, + provider_kind: record.provider_kind, + provider_account_ref: record.provider_account_ref, + provider_account_name: record.provider_account_name, + provider_account_email: record.provider_account_email, + provider_account_authentication_kind: record.provider_account_authentication_kind, + upstream_model_id: record.upstream_model_id, + upstream_transport: record.upstream_transport, + service_tier: record.service_tier, + input_tokens: record.input_tokens, + output_tokens: record.output_tokens, + cached_tokens: record.cached_tokens, + cache_write_tokens: record.cache_write_tokens, + reasoning_tokens: record.reasoning_tokens, + image_input_tokens: record.image_input_tokens, + image_output_tokens: record.image_output_tokens, + total_tokens: record.total_tokens, + cost_source: record.cost_source, + cost_amount: admin_optional_decimal_amount(record.cost_amount)?, + cost_currency: record.cost_currency, + billing, + transport_decision_wait_ms: record.transport_decision_wait_ms, + connect_ms: record.connect_ms, + headers_ms: record.headers_ms, + first_event_ms: record.first_event_ms, + first_reasoning_ms: record.first_reasoning_ms, + first_text_ms: record.first_text_ms, + first_token_ms: record.first_token_ms, + provider_processing_ms: record.provider_processing_ms, + latency_ms: record.latency_ms, + admission_decision_ms: record.admission_decision_ms, + account_selection_wait_ms: record.account_selection_wait_ms, + capacity_used_slots: record.capacity_used_slots, + capacity_total_slots: record.capacity_total_slots, + client_ip: record.client_ip, + user_agent: record.user_agent, + reasoning_effort: record.reasoning_effort, + reasoning_preset: record.reasoning_preset, + subagent_kind: record.subagent_kind, + compact: record.compact, + started_at: record.started_at, + }) +} + pub(crate) fn admin_usage_record( record: UsageRecord, ) -> AdminStoreResult { @@ -674,6 +737,56 @@ pub(crate) fn admin_ops_error(error: OpsErrorRecord) -> admin_observability::Ops } } +pub(crate) fn usage_list_record_from_row( + row: &sqlx::postgres::PgRow, +) -> StoreResult { + Ok(UsageListRecord { + id: get(row, "id")?, + endpoint: get(row, "endpoint")?, + client_transport: get(row, "client_transport")?, + requested_model_id: get(row, "requested_model_id")?, + provider_kind: get(row, "provider_kind")?, + provider_account_ref: get(row, "provider_account_ref")?, + provider_account_name: get(row, "provider_account_name")?, + provider_account_email: get(row, "provider_account_email")?, + provider_account_authentication_kind: get(row, "provider_account_authentication_kind")?, + upstream_model_id: get(row, "upstream_model_id")?, + upstream_transport: get(row, "upstream_transport")?, + service_tier: get(row, "service_tier")?, + input_tokens: optional_unsigned(row, "input_tokens")?, + output_tokens: optional_unsigned(row, "output_tokens")?, + cached_tokens: optional_unsigned(row, "cached_tokens")?, + cache_write_tokens: optional_unsigned(row, "cache_write_tokens")?, + reasoning_tokens: optional_unsigned(row, "reasoning_tokens")?, + image_input_tokens: optional_unsigned(row, "image_input_tokens")?, + image_output_tokens: optional_unsigned(row, "image_output_tokens")?, + total_tokens: optional_unsigned(row, "total_tokens")?, + cost_source: get(row, "cost_source")?, + cost_amount: optional_decimal(row, "cost_amount")?, + cost_currency: get(row, "cost_currency")?, + transport_decision_wait_ms: optional_unsigned(row, "transport_decision_wait_ms")?, + connect_ms: optional_unsigned(row, "connect_ms")?, + headers_ms: optional_unsigned(row, "headers_ms")?, + first_event_ms: optional_unsigned(row, "first_event_ms")?, + first_reasoning_ms: optional_unsigned(row, "first_reasoning_ms")?, + first_text_ms: optional_unsigned(row, "first_text_ms")?, + first_token_ms: optional_unsigned(row, "first_token_ms")?, + provider_processing_ms: optional_unsigned(row, "provider_processing_ms")?, + latency_ms: optional_unsigned(row, "latency_ms")?, + admission_decision_ms: optional_unsigned(row, "admission_decision_ms")?, + account_selection_wait_ms: optional_unsigned(row, "account_selection_wait_ms")?, + capacity_used_slots: optional_unsigned(row, "capacity_used_slots")?, + capacity_total_slots: optional_unsigned(row, "capacity_total_slots")?, + client_ip: get(row, "client_ip")?, + user_agent: get(row, "user_agent")?, + reasoning_effort: get(row, "reasoning_effort")?, + reasoning_preset: get(row, "reasoning_preset")?, + subagent_kind: get(row, "subagent_kind")?, + compact: get(row, "compact")?, + started_at: get(row, "started_at")?, + }) +} + pub(crate) fn usage_record_from_row(row: &sqlx::postgres::PgRow) -> StoreResult { Ok(UsageRecord { id: get(row, "id")?, diff --git a/backend/crates/gateway-store/src/postgres/observability/model.rs b/backend/crates/gateway-store/src/postgres/observability/model.rs index b70bf7f3b..c09f5223d 100644 --- a/backend/crates/gateway-store/src/postgres/observability/model.rs +++ b/backend/crates/gateway-store/src/postgres/observability/model.rs @@ -43,23 +43,6 @@ impl ObservabilityPageSize { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct ObservabilityPageNumber(u32); - -impl ObservabilityPageNumber { - pub fn new(value: u32) -> StoreResult { - if value == 0 { - return Err(invalid("page number must be positive")); - } - Ok(Self(value)) - } - - #[must_use] - pub const fn get(self) -> u32 { - self.0 - } -} - #[derive(Debug, Clone, PartialEq, Eq)] pub struct ObservabilityCursor { pub observed_at: DateTime, @@ -137,8 +120,8 @@ pub struct UsageRecordQuery { pub range: ObservabilityRange, pub filter: UsageRecordFilter, pub cursor: Option, - pub page: ObservabilityPageNumber, pub page_size: ObservabilityPageSize, + pub include_total: bool, } #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -201,8 +184,8 @@ pub struct OpsErrorQuery { pub range: ObservabilityRange, pub filter: OpsErrorFilter, pub cursor: Option, - pub page: ObservabilityPageNumber, pub page_size: ObservabilityPageSize, + pub include_total: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -493,13 +476,59 @@ pub struct DashboardTotals { #[derive(Debug, Clone, PartialEq, Eq)] pub struct DashboardObservation { pub range: ObservabilityRange, - pub requests: RequestMetrics, - pub attempts: AttemptMetrics, pub totals: DashboardTotals, pub provider_accounts: ProviderAccountMetrics, pub trend: Vec, pub account_usage: Vec, - pub recent_requests: Vec, + pub recent_requests: Vec, +} + +/// 使用记录列表所需的窄投影；完整执行、路由和客户端详情按 ID 单独读取。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UsageListRecord { + pub id: String, + pub endpoint: String, + pub client_transport: String, + pub requested_model_id: Option, + pub provider_kind: Option, + pub provider_account_ref: Option, + pub provider_account_name: Option, + pub provider_account_email: Option, + pub provider_account_authentication_kind: Option, + pub upstream_model_id: Option, + pub upstream_transport: Option, + pub service_tier: Option, + pub input_tokens: Option, + pub output_tokens: Option, + pub cached_tokens: Option, + pub cache_write_tokens: Option, + pub reasoning_tokens: Option, + pub image_input_tokens: Option, + pub image_output_tokens: Option, + pub total_tokens: Option, + pub cost_source: String, + pub cost_amount: Option, + pub cost_currency: Option, + pub transport_decision_wait_ms: Option, + pub connect_ms: Option, + pub headers_ms: Option, + pub first_event_ms: Option, + pub first_reasoning_ms: Option, + pub first_text_ms: Option, + pub first_token_ms: Option, + pub provider_processing_ms: Option, + pub latency_ms: Option, + pub admission_decision_ms: Option, + pub account_selection_wait_ms: Option, + pub capacity_used_slots: Option, + pub capacity_total_slots: Option, + pub client_ip: Option, + pub user_agent: Option, + pub reasoning_effort: Option, + pub reasoning_preset: Option, + pub subagent_kind: Option, + pub compact: bool, + pub started_at: DateTime, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -579,8 +608,8 @@ pub struct UsageRecord { #[derive(Debug, Clone, PartialEq, Eq)] pub struct UsageRecordPage { - pub items: Vec, - pub total: u64, + pub items: Vec, + pub total: Option, pub next_cursor: Option, } @@ -694,7 +723,7 @@ pub struct OpsErrorRecord { #[derive(Debug, Clone, PartialEq, Eq)] pub struct OpsErrorPage { pub items: Vec, - pub total: u64, + pub total: Option, pub next_cursor: Option, } diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/accounts.rs b/backend/crates/gateway-store/src/postgres/observability/queries/accounts.rs index 1c2fb1627..c0d5b6fb8 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/accounts.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/accounts.rs @@ -1,4 +1,4 @@ -//! 账号用量查询族。 +//! Provider 账号用量查询族。 use super::super::*; @@ -14,32 +14,18 @@ pub(crate) async fn provider_account_usage( } let mut statement = QueryBuilder::::new( - "select pa.id, pa.provider_kind, pa.authentication_kind, pa.name, pa.email, - pa.plan_type, - count(mr.id)::bigint as request_count, - count(mr.id) filter (where mr.outcome = 'succeeded')::bigint as success_count, - sum(mr.input_tokens)::bigint as input_tokens, - sum(mr.output_tokens)::bigint as output_tokens, - sum(mr.cached_tokens)::bigint as cached_tokens, - sum(mr.cache_write_tokens)::bigint as cache_write_tokens, - sum(mr.reasoning_tokens)::bigint as reasoning_tokens, - sum(mr.image_input_tokens)::bigint as image_input_tokens, - sum(mr.image_output_tokens)::bigint as image_output_tokens, - count(mr.id) filter (where mr.image_generation_succeeded is true)::bigint - as image_request_count, - count(mr.id) filter (where mr.image_generation_succeeded is false)::bigint - as image_request_failed_count, - sum(mr.total_tokens)::bigint as total_tokens, - count(mr.id) filter (where mr.cost_source = 'provider_reported')::bigint - as provider_reported_count, - count(mr.id) filter (where mr.cost_source = 'calculated')::bigint - as calculated_count, - count(mr.id) filter (where mr.cost_source = 'unavailable')::bigint - as unavailable_count, - max(mr.started_at) as last_used_at - from provider_accounts pa - left join model_requests mr - on mr.provider_account_ref = pa.id and mr.started_at >= ", + "with matched as ( + select pa.id, pa.provider_kind, pa.authentication_kind, pa.name, pa.email, + pa.plan_type, mr.id as request_id, + coalesce(mr.upstream_model_id, mr.requested_model_id) as model, + mr.cost_currency, mr.outcome, mr.input_tokens, mr.output_tokens, + mr.cached_tokens, mr.cache_write_tokens, mr.reasoning_tokens, + mr.image_input_tokens, mr.image_output_tokens, + mr.image_generation_succeeded, mr.total_tokens, mr.cost_source, + mr.cost_amount, mr.started_at + from provider_accounts pa + left join model_requests mr + on mr.provider_account_ref = pa.id and mr.started_at >= ", ); statement.push_bind(query.range.start); statement.push(" and mr.started_at < "); @@ -51,24 +37,101 @@ pub(crate) async fn provider_account_usage( statement.push("::text[])"); } statement.push( - " group by pa.id, pa.provider_kind, pa.authentication_kind, pa.name, pa.email, - pa.plan_type", + "), aggregated as ( + select id, provider_kind, authentication_kind, name, email, plan_type, + model, cost_currency, + grouping(model)::integer as model_grouping, + grouping(cost_currency)::integer as currency_grouping, + count(request_id)::bigint as request_count, + count(request_id) filter (where outcome = 'succeeded')::bigint + as success_count, + sum(input_tokens)::bigint as input_tokens, + sum(output_tokens)::bigint as output_tokens, + sum(cached_tokens)::bigint as cached_tokens, + sum(cache_write_tokens)::bigint as cache_write_tokens, + sum(reasoning_tokens)::bigint as reasoning_tokens, + sum(image_input_tokens)::bigint as image_input_tokens, + sum(image_output_tokens)::bigint as image_output_tokens, + count(request_id) filter (where image_generation_succeeded is true)::bigint + as image_request_count, + count(request_id) filter (where image_generation_succeeded is false)::bigint + as image_request_failed_count, + sum(total_tokens)::bigint as total_tokens, + count(request_id) filter (where cost_source = 'provider_reported')::bigint + as provider_reported_count, + count(request_id) filter (where cost_source = 'calculated')::bigint + as calculated_count, + count(request_id) filter (where cost_source = 'unavailable')::bigint + as unavailable_count, + max(started_at) as last_used_at, + sum(cost_amount)::text as amount + from matched + group by id, provider_kind, authentication_kind, name, email, plan_type, + grouping sets ((), (cost_currency), (model), (model, cost_currency)) + ), selected_accounts as ( + select id, + row_number() over (order by last_used_at desc nulls last, name, id) + as sort_position + from aggregated + where model_grouping = 1 and currency_grouping = 1", ); if query.account_ids.is_none() { - statement.push(" having count(mr.id) > 0"); + statement.push(" and request_count > 0"); } - statement.push(" order by max(mr.started_at) desc nulls last, pa.name, pa.id limit "); + statement.push(" order by last_used_at desc nulls last, name, id limit "); statement.push_bind(i64::from(query.limit)); + statement.push( + ") + select aggregated.* + from aggregated + join selected_accounts selected using (id) + order by selected.sort_position, model_grouping desc, currency_grouping desc, + model nulls last, cost_currency nulls last", + ); let rows = statement .build() .fetch_all(pool) .await .map_err(|_| postgres_unavailable("load provider account usage"))?; - let mut observations = rows - .iter() - .map(provider_account_from_row) - .collect::>>()?; + let mut observations = Vec::with_capacity(usize::from(query.limit)); + let mut costs = HashMap::>::new(); + let mut models = HashMap::>::new(); + let mut model_costs = HashMap::<(String, String), Vec>::new(); + for row in &rows { + let model_grouping: i32 = get(row, "model_grouping")?; + let currency_grouping: i32 = get(row, "currency_grouping")?; + match (model_grouping, currency_grouping) { + (1, 1) => observations.push(provider_account_from_row(row)?), + (1, 0) if get::>(row, "cost_currency")?.is_some() => { + costs + .entry(get(row, "id")?) + .or_default() + .push(cost_from_row(row)?); + } + (0, 1) if get::>(row, "model")?.is_some() => { + models + .entry(get(row, "id")?) + .or_default() + .push(provider_account_model_from_row(row)?); + } + (0, 0) + if get::>(row, "model")?.is_some() + && get::>(row, "cost_currency")?.is_some() => + { + model_costs + .entry((get(row, "id")?, get(row, "model")?)) + .or_default() + .push(cost_from_row(row)?); + } + (0 | 1, 0 | 1) => {} + _ => { + return Err(postgres_unavailable( + "decode provider account usage grouping", + )); + } + } + } if observations.is_empty() { return Ok(observations); } @@ -81,52 +144,23 @@ pub(crate) async fn provider_account_usage( } else { HashMap::new() }; - let account_costs = account_costs(pool, query.range, &account_ids).await?; - let model_rows = account_model_rows(pool, query.range, &account_ids).await?; - let model_costs = account_model_costs(pool, query.range, &account_ids).await?; - let account_positions = observations - .iter() - .enumerate() - .map(|(index, item)| (item.account_id.clone(), index)) - .collect::>(); for observation in &mut observations { - observation.costs = account_costs - .get(&observation.account_id) - .cloned() - .unwrap_or_default(); + observation.costs = costs.remove(&observation.account_id).unwrap_or_default(); observation.request_buckets = request_buckets .remove(&observation.account_id) .unwrap_or_default(); - } - for row in model_rows { - let account_id: String = get(&row, "provider_account_ref")?; - let Some(position) = account_positions.get(&account_id).copied() else { - continue; - }; - let model: String = get(&row, "model")?; - observations[position] - .models - .push(ProviderAccountModelUsageObservation { - costs: model_costs - .get(&(account_id, model.clone())) - .cloned() - .unwrap_or_default(), - model, - request_count: unsigned(&row, "request_count")?, - success_count: unsigned(&row, "success_count")?, - input_tokens: optional_unsigned(&row, "input_tokens")?, - output_tokens: optional_unsigned(&row, "output_tokens")?, - cached_tokens: optional_unsigned(&row, "cached_tokens")?, - cache_write_tokens: optional_unsigned(&row, "cache_write_tokens")?, - reasoning_tokens: optional_unsigned(&row, "reasoning_tokens")?, - image_input_tokens: optional_unsigned(&row, "image_input_tokens")?, - image_output_tokens: optional_unsigned(&row, "image_output_tokens")?, - image_request_count: unsigned(&row, "image_request_count")?, - image_request_failed_count: unsigned(&row, "image_request_failed_count")?, - total_tokens: optional_unsigned(&row, "total_tokens")?, - cost_coverage: coverage_from_row(&row)?, - last_used_at: get(&row, "last_used_at")?, - }); + observation.models = models.remove(&observation.account_id).unwrap_or_default(); + for model in &mut observation.models { + model.costs = model_costs + .remove(&(observation.account_id.clone(), model.model.clone())) + .unwrap_or_default(); + } + observation.models.sort_by(|left, right| { + right + .request_count + .cmp(&left.request_count) + .then_with(|| left.model.cmp(&right.model)) + }); } Ok(observations) } @@ -145,7 +179,8 @@ pub(crate) async fn provider_account_request_buckets( and mr.started_at >= $1 and mr.started_at < $3 and mr.outcome = 'succeeded' and mr.downstream_committed_at is not null - and mr.client_status_code between 200 and 399 + and ((mr.client_transport = 'websocket' and mr.client_status_code is null) + or mr.client_status_code between 200 and 399) group by mr.provider_account_ref, bucket_index order by mr.provider_account_ref, bucket_index", ) @@ -191,121 +226,27 @@ pub(crate) async fn provider_account_request_buckets( Ok(timelines) } -pub(crate) async fn account_costs( - pool: &PgPool, - range: ObservabilityRange, - account_ids: &[String], -) -> StoreResult>> { - let rows = sqlx::query( - "select mr.provider_account_ref, mr.cost_currency, sum(mr.cost_amount)::text as amount - from model_requests mr - where mr.provider_account_ref = any($1::text[]) - and mr.started_at >= $2 and mr.started_at < $3 - and mr.outcome = 'succeeded' - and mr.downstream_committed_at is not null - and mr.client_status_code between 200 and 399 - and mr.cost_amount is not null and mr.cost_currency is not null - group by mr.provider_account_ref, mr.cost_currency - order by mr.provider_account_ref, mr.cost_currency", - ) - .bind(account_ids) - .bind(range.start) - .bind(range.end) - .fetch_all(pool) - .await - .map_err(|_| postgres_unavailable("load provider account costs"))?; - let mut costs = HashMap::>::new(); - for row in rows { - costs - .entry(get(&row, "provider_account_ref")?) - .or_default() - .push(cost_from_row(&row)?); - } - Ok(costs) -} - -pub(crate) async fn account_model_rows( - pool: &PgPool, - range: ObservabilityRange, - account_ids: &[String], -) -> StoreResult> { - sqlx::query( - "select mr.provider_account_ref, - coalesce(mr.upstream_model_id, mr.requested_model_id) as model, - count(*)::bigint as request_count, - count(*) filter (where mr.outcome = 'succeeded')::bigint as success_count, - sum(mr.input_tokens)::bigint as input_tokens, - sum(mr.output_tokens)::bigint as output_tokens, - sum(mr.cached_tokens)::bigint as cached_tokens, - sum(mr.cache_write_tokens)::bigint as cache_write_tokens, - sum(mr.reasoning_tokens)::bigint as reasoning_tokens, - sum(mr.image_input_tokens)::bigint as image_input_tokens, - sum(mr.image_output_tokens)::bigint as image_output_tokens, - count(*) filter (where mr.image_generation_succeeded is true)::bigint - as image_request_count, - count(*) filter (where mr.image_generation_succeeded is false)::bigint - as image_request_failed_count, - sum(mr.total_tokens)::bigint as total_tokens, - count(*) filter (where mr.cost_source = 'provider_reported')::bigint - as provider_reported_count, - count(*) filter (where mr.cost_source = 'calculated')::bigint - as calculated_count, - count(*) filter (where mr.cost_source = 'unavailable')::bigint - as unavailable_count, - max(mr.started_at) as last_used_at - from model_requests mr - where mr.provider_account_ref = any($1::text[]) - and mr.started_at >= $2 and mr.started_at < $3 - and mr.outcome = 'succeeded' - and mr.downstream_committed_at is not null - and mr.client_status_code between 200 and 399 - and coalesce(mr.upstream_model_id, mr.requested_model_id) is not null - group by mr.provider_account_ref, coalesce(mr.upstream_model_id, mr.requested_model_id) - order by mr.provider_account_ref, request_count desc, model", - ) - .bind(account_ids) - .bind(range.start) - .bind(range.end) - .fetch_all(pool) - .await - .map_err(|_| postgres_unavailable("load provider account model usage")) -} - -pub(crate) async fn account_model_costs( - pool: &PgPool, - range: ObservabilityRange, - account_ids: &[String], -) -> StoreResult>> { - let rows = sqlx::query( - "select mr.provider_account_ref, - coalesce(mr.upstream_model_id, mr.requested_model_id) as model, - mr.cost_currency, sum(mr.cost_amount)::text as amount - from model_requests mr - where mr.provider_account_ref = any($1::text[]) - and mr.started_at >= $2 and mr.started_at < $3 - and mr.outcome = 'succeeded' - and mr.downstream_committed_at is not null - and mr.client_status_code between 200 and 399 - and mr.cost_amount is not null and mr.cost_currency is not null - and coalesce(mr.upstream_model_id, mr.requested_model_id) is not null - group by mr.provider_account_ref, coalesce(mr.upstream_model_id, mr.requested_model_id), - mr.cost_currency - order by mr.provider_account_ref, model, mr.cost_currency", - ) - .bind(account_ids) - .bind(range.start) - .bind(range.end) - .fetch_all(pool) - .await - .map_err(|_| postgres_unavailable("load provider account model costs"))?; - let mut costs = HashMap::<(String, String), Vec>::new(); - for row in rows { - costs - .entry((get(&row, "provider_account_ref")?, get(&row, "model")?)) - .or_default() - .push(cost_from_row(&row)?); - } - Ok(costs) +fn provider_account_model_from_row( + row: &sqlx::postgres::PgRow, +) -> StoreResult { + Ok(ProviderAccountModelUsageObservation { + model: get(row, "model")?, + request_count: unsigned(row, "request_count")?, + success_count: unsigned(row, "success_count")?, + input_tokens: optional_unsigned(row, "input_tokens")?, + output_tokens: optional_unsigned(row, "output_tokens")?, + cached_tokens: optional_unsigned(row, "cached_tokens")?, + cache_write_tokens: optional_unsigned(row, "cache_write_tokens")?, + reasoning_tokens: optional_unsigned(row, "reasoning_tokens")?, + image_input_tokens: optional_unsigned(row, "image_input_tokens")?, + image_output_tokens: optional_unsigned(row, "image_output_tokens")?, + image_request_count: unsigned(row, "image_request_count")?, + image_request_failed_count: unsigned(row, "image_request_failed_count")?, + total_tokens: optional_unsigned(row, "total_tokens")?, + cost_coverage: coverage_from_row(row)?, + costs: Vec::new(), + last_used_at: get(row, "last_used_at")?, + }) } pub(crate) fn provider_account_from_row( diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs b/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs index 5fdb8899d..d29674369 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs @@ -133,6 +133,23 @@ pub(crate) async fn request_metric_series( pool: &PgPool, range: ObservabilityRange, filter: &UsageRecordFilter, +) -> StoreResult> { + request_metric_series_inner(pool, range, filter, true).await +} + +pub(crate) async fn dashboard_request_metric_series( + pool: &PgPool, + range: ObservabilityRange, + filter: &UsageRecordFilter, +) -> StoreResult> { + request_metric_series_inner(pool, range, filter, false).await +} + +async fn request_metric_series_inner( + pool: &PgPool, + range: ObservabilityRange, + filter: &UsageRecordFilter, + load_costs: bool, ) -> StoreResult> { filter.validate()?; let granularity = granularity_for(range); @@ -252,10 +269,12 @@ pub(crate) async fn request_metric_series( }, ); } - let bucket_costs = request_costs_by_bucket(pool, range, filter, granularity).await?; - for (bucket, costs) in bucket_costs { - if let Some(point) = points.get_mut(&bucket) { - point.costs = costs; + if load_costs { + let bucket_costs = request_costs_by_bucket(pool, range, filter, granularity).await?; + for (bucket, costs) in bucket_costs { + if let Some(point) = points.get_mut(&bucket) { + point.costs = costs; + } } } fill_metric_gaps(range, granularity, points) diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs index 32405df28..7a4ef1564 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs @@ -2,84 +2,71 @@ use super::super::*; -pub(crate) const OPS_ERRORS_CTE: &str = "with errors as ( - select 'model_request'::text as source, - mr.id as event_id, mr.id as request_id, - nullif(mr.attempt_count, 0) as attempt_index, - mr.client_api_key_ref, 'model_request'::text as component, mr.operation, - mr.endpoint, - mr.provider_kind, - mr.provider_account_ref, - mr.provider_account_name_snapshot as provider_account_name, - mr.provider_account_email_snapshot as provider_account_email, - mr.provider_account_authentication_kind_snapshot - as provider_account_authentication_kind, - mr.upstream_model_id, mr.upstream_transport, - coalesce(mr.error_kind, 'failed') as failure_kind, - mr.client_status_code, mr.upstream_status_code, - mr.provider_error_code, mr.client_response_id, mr.upstream_request_id, - mr.latency_ms, coalesce(mr.error_message, mr.error_kind, 'request failed') as message, - 1::integer as occurrence_count, - coalesce(mr.completed_at, mr.started_at) as occurred_at, - 'model_request:' || mr.id as stable_sort_id - from model_requests mr where mr.outcome = 'failed' - union all - select 'ops_event'::text, oe.id, oe.model_request_id, oe.attempt_index, - mr.client_api_key_ref, oe.component, oe.operation, - null::text as endpoint, - oe.provider_kind, oe.provider_account_ref, - oe.provider_account_name_snapshot as provider_account_name, - oe.provider_account_email_snapshot as provider_account_email, - oe.provider_account_authentication_kind_snapshot - as provider_account_authentication_kind, - oe.upstream_model_id, null::text, oe.failure_kind, - null::integer as client_status_code, oe.status_code as upstream_status_code, - oe.provider_error_code, mr.client_response_id, oe.upstream_request_id, - oe.latency_ms, oe.message, oe.occurrence_count, oe.created_at, - 'ops_event:' || oe.id - from ops_events oe left join model_requests mr on mr.id = oe.model_request_id - )"; +const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source, + mr.id as event_id, mr.id as request_id, + nullif(mr.attempt_count, 0) as attempt_index, + mr.client_api_key_ref, 'model_request'::text as component, mr.operation, + mr.endpoint, mr.provider_kind, mr.provider_account_ref, + mr.provider_account_name_snapshot as provider_account_name, + mr.provider_account_email_snapshot as provider_account_email, + mr.provider_account_authentication_kind_snapshot + as provider_account_authentication_kind, + mr.upstream_model_id, mr.upstream_transport, + coalesce(mr.error_kind, 'failed') as failure_kind, + mr.client_status_code, mr.upstream_status_code, + mr.provider_error_code, mr.client_response_id, mr.upstream_request_id, + mr.latency_ms, + coalesce(mr.error_message, mr.error_kind, 'request failed') as message, + 1::integer as occurrence_count, mr.completed_at as occurred_at, + 'model_request:' || mr.id as stable_sort_id +from model_requests mr +where mr.outcome = 'failed'"; + +const OPS_EVENT_SELECT: &str = "select 'ops_event'::text as source, + oe.id as event_id, oe.model_request_id as request_id, oe.attempt_index, + mr.client_api_key_ref, oe.component, oe.operation, + null::text as endpoint, oe.provider_kind, oe.provider_account_ref, + oe.provider_account_name_snapshot as provider_account_name, + oe.provider_account_email_snapshot as provider_account_email, + oe.provider_account_authentication_kind_snapshot + as provider_account_authentication_kind, + oe.upstream_model_id, null::text as upstream_transport, oe.failure_kind, + null::integer as client_status_code, oe.status_code as upstream_status_code, + oe.provider_error_code, mr.client_response_id, oe.upstream_request_id, + oe.latency_ms, oe.message, oe.occurrence_count, oe.created_at as occurred_at, + 'ops_event:' || oe.id as stable_sort_id +from ops_events oe +left join model_requests mr on mr.id = oe.model_request_id +where true"; pub(crate) async fn list_ops_errors( pool: &PgPool, query: OpsErrorQuery, ) -> StoreResult { query.filter.validate()?; - let total = count_ops_errors(pool, query.range, &query.filter).await?; - let mut statement = QueryBuilder::::new(OPS_ERRORS_CTE); - statement.push( - " select source, event_id, request_id, attempt_index, client_api_key_ref, - component, operation, endpoint, provider_kind, provider_account_ref, - provider_account_name, provider_account_email, - provider_account_authentication_kind, - upstream_model_id, upstream_transport, - failure_kind, client_status_code, upstream_status_code, - provider_error_code, client_response_id, - upstream_request_id, latency_ms, message, occurrence_count, occurred_at, - stable_sort_id - from errors e where e.occurred_at >= ", + let total = if query.include_total { + Some(count_ops_errors(pool, query.range, &query.filter).await?) + } else { + None + }; + let mut statement = QueryBuilder::::new("select * from ("); + statement.push(REQUEST_ERROR_SELECT); + push_request_error_predicates( + &mut statement, + query.range, + &query.filter, + query.cursor.as_ref(), ); - statement.push_bind(query.range.start); - statement.push(" and e.occurred_at < "); - statement.push_bind(query.range.end); - push_ops_filter(&mut statement, &query.filter); - if let Some(cursor) = &query.cursor { - statement.push(" and (e.occurred_at, e.stable_sort_id) < ("); - statement.push_bind(cursor.observed_at); - statement.push(", "); - statement.push_bind(cursor.stable_id.clone()); - statement.push(")"); - } - statement.push(" order by e.occurred_at desc, e.stable_sort_id desc limit "); + statement.push(" union all "); + statement.push(OPS_EVENT_SELECT); + push_ops_event_predicates( + &mut statement, + query.range, + &query.filter, + query.cursor.as_ref(), + ); + statement.push(") e order by occurred_at desc, stable_sort_id desc limit "); statement.push_bind(i64::from(query.page_size.get()) + 1); - if query.cursor.is_none() { - let offset = u64::from(query.page.get() - 1) - .checked_mul(u64::from(query.page_size.get())) - .and_then(|value| i64::try_from(value).ok()) - .ok_or_else(|| invalid("ops error page offset is too large"))?; - statement.push(" offset "); - statement.push_bind(offset); - } let rows = statement .build() .fetch_all(pool) @@ -113,12 +100,15 @@ pub(crate) async fn count_ops_errors( range: ObservabilityRange, filter: &OpsErrorFilter, ) -> StoreResult { - let mut statement = QueryBuilder::::new(OPS_ERRORS_CTE); - statement.push(" select count(*)::bigint from errors e where e.occurred_at >= "); - statement.push_bind(range.start); - statement.push(" and e.occurred_at < "); - statement.push_bind(range.end); - push_ops_filter(&mut statement, filter); + let mut statement = QueryBuilder::::new( + "select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where mr.outcome = 'failed'", + ); + push_request_error_predicates(&mut statement, range, filter, None); + statement.push( + " union all select count(*)::bigint as source_count from ops_events oe left join model_requests mr on mr.id = oe.model_request_id where true", + ); + push_ops_event_predicates(&mut statement, range, filter, None); + statement.push(") counts"); let total = statement .build_query_scalar::() .fetch_one(pool) @@ -127,45 +117,163 @@ pub(crate) async fn count_ops_errors( to_u64(total) } -pub(crate) fn push_ops_filter(statement: &mut QueryBuilder, filter: &OpsErrorFilter) { +fn push_request_error_predicates( + statement: &mut QueryBuilder, + range: ObservabilityRange, + filter: &OpsErrorFilter, + cursor: Option<&ObservabilityCursor>, +) { + push_range(statement, "mr.completed_at", range); for (column, value) in [ - ("client_api_key_ref", &filter.client_api_key_ref), - ("request_id", &filter.request_id), - ("provider_account_ref", &filter.provider_account_ref), - ("provider_kind", &filter.provider_kind), - ("operation", &filter.operation), - ("upstream_transport", &filter.transport), - ("upstream_request_id", &filter.upstream_request_id), - ("failure_kind", &filter.failure_kind), + ("mr.client_api_key_ref", &filter.client_api_key_ref), + ("mr.id", &filter.request_id), + ("mr.provider_account_ref", &filter.provider_account_ref), + ("mr.provider_kind", &filter.provider_kind), + ("mr.operation", &filter.operation), + ("mr.upstream_transport", &filter.transport), + ("mr.upstream_request_id", &filter.upstream_request_id), ] { - if let Some(value) = value { - statement.push(format!(" and e.{column} = ")); - statement.push_bind(value.clone()); - } + push_text_equality(statement, column, value); } - if let Some(value) = &filter.response_id { - statement.push(" and e.client_response_id = "); - statement.push_bind(value.as_bytes().to_vec()); - } - if let Some(model) = &filter.model { - statement.push(" and e.upstream_model_id = "); - statement.push_bind(model.clone()); + if let Some(value) = &filter.failure_kind { + statement.push(" and coalesce(mr.error_kind, 'failed') = "); + statement.push_bind(value.clone()); } + push_response_id_filter(statement, "mr.client_response_id", filter); + push_text_equality(statement, "mr.upstream_model_id", &filter.model); if let Some(index) = filter.attempt_index { - statement.push(" and e.attempt_index = "); + statement.push(" and nullif(mr.attempt_count, 0) = "); statement.push_bind(i32::try_from(index).unwrap_or(i32::MAX)); } if let Some(status) = filter.status_code { - statement.push(" and e.upstream_status_code = "); + statement.push(" and mr.upstream_status_code = "); statement.push_bind(i32::from(status)); } if let Some(search) = &filter.search { - statement.push( - " and lower(concat_ws(' ', e.event_id, e.request_id, e.client_api_key_ref, - e.component, e.operation, e.provider_kind, e.provider_account_ref, - e.upstream_model_id, e.failure_kind, e.provider_error_code, - e.upstream_request_id, e.message)) like ", + push_prefix_search( + statement, + &[ + "mr.id", + "mr.client_api_key_ref", + "mr.provider_account_ref", + "mr.upstream_request_id", + "mr.provider_error_code", + ], + search, ); - statement.push_bind(format!("%{}%", search.to_lowercase())); + } + push_cursor( + statement, + "mr.completed_at", + "'model_request:' || mr.id", + cursor, + ); +} + +fn push_ops_event_predicates( + statement: &mut QueryBuilder, + range: ObservabilityRange, + filter: &OpsErrorFilter, + cursor: Option<&ObservabilityCursor>, +) { + push_range(statement, "oe.created_at", range); + for (column, value) in [ + ("mr.client_api_key_ref", &filter.client_api_key_ref), + ("oe.model_request_id", &filter.request_id), + ("oe.provider_account_ref", &filter.provider_account_ref), + ("oe.provider_kind", &filter.provider_kind), + ("oe.operation", &filter.operation), + ("oe.upstream_request_id", &filter.upstream_request_id), + ("oe.failure_kind", &filter.failure_kind), + ] { + push_text_equality(statement, column, value); + } + // Ops events do not persist upstream transport. A transport filter therefore + // intentionally excludes this source instead of matching an unrelated request fact. + if filter.transport.is_some() { + statement.push(" and false"); + } + push_response_id_filter(statement, "mr.client_response_id", filter); + push_text_equality(statement, "oe.upstream_model_id", &filter.model); + if let Some(index) = filter.attempt_index { + statement.push(" and oe.attempt_index = "); + statement.push_bind(i32::try_from(index).unwrap_or(i32::MAX)); + } + if let Some(status) = filter.status_code { + statement.push(" and oe.status_code = "); + statement.push_bind(i32::from(status)); + } + if let Some(search) = &filter.search { + push_prefix_search( + statement, + &[ + "oe.id", + "oe.model_request_id", + "mr.client_api_key_ref", + "oe.provider_account_ref", + "oe.upstream_request_id", + "oe.provider_error_code", + ], + search, + ); + } + push_cursor(statement, "oe.created_at", "'ops_event:' || oe.id", cursor); +} + +fn push_range(statement: &mut QueryBuilder, column: &str, range: ObservabilityRange) { + statement.push(format!(" and {column} >= ")); + statement.push_bind(range.start); + statement.push(format!(" and {column} < ")); + statement.push_bind(range.end); +} + +fn push_text_equality( + statement: &mut QueryBuilder, + column: &str, + value: &Option, +) { + if let Some(value) = value { + statement.push(format!(" and {column} = ")); + statement.push_bind(value.clone()); + } +} + +fn push_response_id_filter( + statement: &mut QueryBuilder, + column: &str, + filter: &OpsErrorFilter, +) { + if let Some(value) = &filter.response_id { + statement.push(format!(" and {column} = ")); + statement.push_bind(value.as_bytes().to_vec()); + } +} + +fn push_prefix_search(statement: &mut QueryBuilder, columns: &[&str], value: &str) { + let pattern = literal_prefix_pattern(value); + statement.push(" and ("); + for (index, column) in columns.iter().enumerate() { + if index > 0 { + statement.push(" or "); + } + statement.push(format!("{column} like ")); + statement.push_bind(pattern.clone()); + statement.push(" escape '\\'"); + } + statement.push(")"); +} + +fn push_cursor( + statement: &mut QueryBuilder, + occurred_at: &str, + stable_sort_id: &str, + cursor: Option<&ObservabilityCursor>, +) { + if let Some(cursor) = cursor { + statement.push(format!(" and ({occurred_at}, {stable_sort_id}) < (")); + statement.push_bind(cursor.observed_at); + statement.push(", "); + statement.push_bind(cursor.stable_id.clone()); + statement.push(")"); } } 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 206812025..bc5856719 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs @@ -75,18 +75,33 @@ pub(crate) fn push_usage_filter( query.push_bind(value.clone()); } if let Some(value) = &filter.search { - query.push(format!( - " and lower(concat_ws(' ', {alias}.id, {alias}.client_api_key_ref, - {alias}.requested_model_id, {alias}.provider_kind, - {alias}.provider_account_ref, {alias}.upstream_model_id, - encode({alias}.client_response_id, 'escape'), {alias}.upstream_request_id, - {alias}.error_kind, {alias}.provider_error_code, - {alias}.error_message)) like " - )); - query.push_bind(format!("%{}%", value.to_lowercase())); + let pattern = literal_prefix_pattern(value); + query.push(format!(" and ({alias}.id like ")); + query.push_bind(pattern.clone()); + for column in [ + "client_api_key_ref", + "provider_account_ref", + "requested_model_id", + "upstream_model_id", + "upstream_request_id", + ] { + query.push(format!(" escape '\\' or {alias}.{column} like ")); + query.push_bind(pattern.clone()); + } + query.push(" escape '\\')"); } } +pub(crate) fn literal_prefix_pattern(value: &str) -> String { + format!( + "{}%", + value + .replace('\\', "\\\\") + .replace('%', "\\%") + .replace('_', "\\_") + ) +} + /// 只把已完整交付给客户端的成功响应投影为使用事实。 /// /// `model_requests` 同时承担执行审计：包括上游已发送、但尚未收到首个事件就断开的 @@ -104,7 +119,27 @@ pub(crate) fn push_completed_usage_fact_filter(query: &mut QueryBuilder= "); statement.push_bind(query.range.start); statement.push(" and mr.started_at < "); @@ -155,14 +194,6 @@ pub(crate) async fn list_usage_records( } statement.push(" order by mr.started_at desc, mr.id desc limit "); statement.push_bind(i64::from(query.page_size.get()) + 1); - if query.cursor.is_none() { - let offset = u64::from(query.page.get() - 1) - .checked_mul(u64::from(query.page_size.get())) - .and_then(|value| i64::try_from(value).ok()) - .ok_or_else(|| invalid("usage page offset is too large"))?; - statement.push(" offset "); - statement.push_bind(offset); - } let rows = statement .build() .fetch_all(pool) @@ -170,7 +201,7 @@ pub(crate) async fn list_usage_records( .map_err(|_| postgres_unavailable("list usage records"))?; let mut items = rows .iter() - .map(usage_record_from_row) + .map(usage_list_record_from_row) .collect::>>()?; let has_more = items.len() > usize::from(query.page_size.get()); if has_more { @@ -218,7 +249,7 @@ pub(crate) async fn usage_record_detail( ) -> StoreResult { require_nonempty("model request", "id", request_id)?; validate_text(request_id, MAX_FILTER_BYTES, "request ID")?; - let mut statement = QueryBuilder::::new(USAGE_RECORD_SELECT); + let mut statement = QueryBuilder::::new(USAGE_RECORD_DETAIL_SELECT); statement.push(" where mr.id = "); statement.push_bind(request_id.to_owned()); push_completed_usage_fact_filter(&mut statement, "mr"); @@ -343,29 +374,13 @@ pub(crate) async fn usage_diagnostics( ) -> StoreResult> { filter.validate()?; let dimension_sql = diagnostic_dimension_sql(dimension); - let mut statement = QueryBuilder::::new("select "); + let mut statement = QueryBuilder::::new("with matched as (select "); statement.push(dimension_sql); statement.push( - " as dimension_name, - count(*)::bigint as request_count, - count(*) filter (where mr.outcome = 'succeeded')::bigint as success_count, - count(*) filter (where mr.outcome = 'failed')::bigint as failure_count, - coalesce(sum(mr.attempt_count), 0)::bigint as attempt_count, - coalesce(sum(mr.total_tokens), 0)::bigint as total_tokens, - round(avg(mr.latency_ms))::bigint as average_latency_ms, - round(percentile_cont(0.95) within group (order by mr.latency_ms))::bigint - as latency_p95_ms, - round(percentile_cont(0.95) within group (order by mr.first_token_ms))::bigint - as first_token_p95_ms, - count(*) filter (where mr.outcome in ('cancelled', 'incomplete'))::bigint - as non_completion_count, - coalesce(sum(greatest(mr.attempt_count - 1, 0)), 0)::bigint as retry_count, - count(*) filter (where mr.cost_source = 'provider_reported')::bigint - as provider_reported_count, - count(*) filter (where mr.cost_source = 'calculated')::bigint - as calculated_count, - count(*) filter (where mr.cost_source = 'unavailable')::bigint - as unavailable_count + " as dimension_name, mr.outcome, mr.attempt_count, mr.total_tokens, + mr.latency_ms, mr.first_token_ms, mr.cost_source, mr.cost_amount, + mr.cost_currency, mr.downstream_committed_at, mr.client_transport, + mr.client_status_code from model_requests mr where mr.started_at >= ", ); statement.push_bind(range.start); @@ -373,17 +388,63 @@ pub(crate) async fn usage_diagnostics( statement.push_bind(range.end); push_diagnostic_dimension_filter(&mut statement, dimension); push_usage_filter(&mut statement, filter, "mr"); - statement.push(" group by dimension_name order by request_count desc, dimension_name limit "); + statement.push( + "), aggregated as ( + select dimension_name, cost_currency, + grouping(cost_currency)::integer as currency_grouping, + count(*)::bigint as request_count, + count(*) filter (where outcome = 'succeeded')::bigint as success_count, + count(*) filter (where outcome = 'failed')::bigint as failure_count, + coalesce(sum(attempt_count), 0)::bigint as attempt_count, + coalesce(sum(total_tokens), 0)::bigint as total_tokens, + round(avg(latency_ms))::bigint as average_latency_ms, + round(percentile_cont(0.95) within group (order by latency_ms))::bigint + as latency_p95_ms, + round(percentile_cont(0.95) within group (order by first_token_ms))::bigint + as first_token_p95_ms, + count(*) filter (where outcome in ('cancelled', 'incomplete'))::bigint + as non_completion_count, + coalesce(sum(greatest(attempt_count - 1, 0)), 0)::bigint as retry_count, + count(*) filter (where cost_source = 'provider_reported')::bigint + as provider_reported_count, + count(*) filter (where cost_source = 'calculated')::bigint + as calculated_count, + count(*) filter (where cost_source = 'unavailable')::bigint + as unavailable_count, + sum(cost_amount) filter ( + where outcome = 'succeeded' + and downstream_committed_at is not null + and ((client_transport = 'websocket' and client_status_code is null) + or client_status_code between 200 and 399) + )::text as amount + from matched + group by dimension_name, grouping sets ((), (cost_currency)) + ), selected_dimensions as ( + select dimension_name, + row_number() over (order by request_count desc, dimension_name) + as sort_position + from aggregated + where currency_grouping = 1 + order by request_count desc, dimension_name limit ", + ); statement.push_bind(DIAGNOSTIC_LIMIT); + statement.push( + ") + select aggregated.* + from aggregated + join selected_dimensions selected using (dimension_name) + order by selected.sort_position, currency_grouping desc, cost_currency nulls last", + ); let rows = statement .build() .fetch_all(pool) .await .map_err(|_| postgres_unavailable("load usage diagnostics"))?; - let mut observations = rows - .iter() - .map(|row| { - Ok(DiagnosticObservation { + let mut observations = Vec::with_capacity(DIAGNOSTIC_LIMIT as usize); + let mut costs = HashMap::>::new(); + for row in &rows { + match get::(row, "currency_grouping")? { + 1 => observations.push(DiagnosticObservation { key: get(row, "dimension_name")?, name: get(row, "dimension_name")?, request_count: unsigned(row, "request_count")?, @@ -398,9 +459,19 @@ pub(crate) async fn usage_diagnostics( retry_count: unsigned(row, "retry_count")?, cost_coverage: coverage_from_row(row)?, costs: Vec::new(), - }) - }) - .collect::>>()?; + }), + 0 if get::>(row, "cost_currency")?.is_some() + && get::>(row, "amount")?.is_some() => + { + costs + .entry(get(row, "dimension_name")?) + .or_default() + .push(cost_from_row(row)?); + } + 0 => {} + _ => return Err(postgres_unavailable("decode usage diagnostic grouping")), + } + } let mut display_names = match dimension { DiagnosticDimension::Account => { diagnostic_account_display_names( @@ -428,17 +499,7 @@ pub(crate) async fn usage_diagnostics( if let Some(name) = display_names.remove(&observation.key) { observation.name = name; } - } - let positions = observations - .iter() - .enumerate() - .map(|(index, item)| (item.key.clone(), index)) - .collect::>(); - let costs = diagnostic_costs(pool, range, filter, dimension_sql, dimension).await?; - for (key, values) in costs { - if let Some(position) = positions.get(&key).copied() { - observations[position].costs = values; - } + observation.costs = costs.remove(&observation.key).unwrap_or_default(); } Ok(observations) } @@ -451,17 +512,27 @@ pub(crate) async fn diagnostic_account_display_names( return Ok(HashMap::new()); } let rows = sqlx::query( - "select provider_account_ref, + "select requested.account_id as provider_account_ref, coalesce( - (array_agg(provider_account_email_snapshot order by started_at desc) - filter (where provider_account_email_snapshot is not null))[1], - (array_agg(provider_account_name_snapshot order by started_at desc) - filter (where provider_account_name_snapshot is not null))[1], - provider_account_ref + ( + select request.provider_account_email_snapshot + from model_requests request + where request.provider_account_ref = requested.account_id + and request.provider_account_email_snapshot is not null + order by request.started_at desc, request.id desc + limit 1 + ), + ( + select request.provider_account_name_snapshot + from model_requests request + where request.provider_account_ref = requested.account_id + and request.provider_account_name_snapshot is not null + order by request.started_at desc, request.id desc + limit 1 + ), + requested.account_id ) as display_name - from model_requests - where provider_account_ref = any($1::text[]) - group by provider_account_ref", + from unnest($1::text[]) as requested(account_id)", ) .bind(account_ids) .fetch_all(pool) @@ -496,44 +567,6 @@ pub(crate) async fn diagnostic_api_key_display_names( Ok(display_names) } -pub(crate) async fn diagnostic_costs( - pool: &PgPool, - range: ObservabilityRange, - filter: &UsageRecordFilter, - dimension_sql: &'static str, - dimension: DiagnosticDimension, -) -> StoreResult>> { - let mut statement = QueryBuilder::::new("select "); - statement.push(dimension_sql); - statement.push( - " as dimension_name, mr.cost_currency, sum(mr.cost_amount)::text as amount - from model_requests mr where mr.started_at >= ", - ); - statement.push_bind(range.start); - statement.push(" and mr.started_at < "); - statement.push_bind(range.end); - statement.push(" and mr.cost_amount is not null and mr.cost_currency is not null"); - push_completed_usage_fact_filter(&mut statement, "mr"); - push_diagnostic_dimension_filter(&mut statement, dimension); - push_usage_filter(&mut statement, filter, "mr"); - statement.push( - " group by dimension_name, mr.cost_currency order by dimension_name, mr.cost_currency", - ); - let rows = statement - .build() - .fetch_all(pool) - .await - .map_err(|_| postgres_unavailable("load diagnostic costs"))?; - let mut result = HashMap::>::new(); - for row in rows { - result - .entry(get(&row, "dimension_name")?) - .or_default() - .push(cost_from_row(&row)?); - } - Ok(result) -} - pub(crate) fn diagnostic_dimension_sql(dimension: DiagnosticDimension) -> &'static str { match dimension { DiagnosticDimension::Provider => "coalesce(mr.provider_kind, 'unrouted')", diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs index a245170a4..c71d1d188 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs @@ -31,7 +31,7 @@ impl PgAdminAccountStore { Self { pool: pool.clone(), accounts: PgProviderAccountRepository::new(pool.clone()), - observability: PgObservabilityRepository::new(pool.clone(), None), + observability: PgObservabilityRepository::new(pool.clone(), None, query_budget.clone()), control_plane: PgControlPlaneRepository::new(pool), cooldowns, query_budget, diff --git a/backend/crates/gateway-store/tests/postgres/mod.rs b/backend/crates/gateway-store/tests/postgres/mod.rs index 91139a869..981eea290 100644 --- a/backend/crates/gateway-store/tests/postgres/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/mod.rs @@ -1,7 +1,8 @@ use std::{str::FromStr, time::Duration}; use gateway_store::postgres::{ - ObservabilityQueryBudget, PgAdminAccountStore, connect_and_migrate, + ObservabilityQueryBudget, PgAdminAccountStore, PgAdminObservabilityStore, + PgObservabilityRepository, connect_and_migrate, }; use sqlx::{ ConnectOptions as _, PgPool, @@ -39,6 +40,14 @@ pub(super) fn observability_query_budget() -> ObservabilityQueryBudget { .expect("valid test observability query budget") } +pub(super) fn observability_repository(pool: &PgPool) -> PgObservabilityRepository { + PgObservabilityRepository::new(pool.clone(), None, observability_query_budget()) +} + +pub(super) fn admin_observability_store(pool: &PgPool) -> PgAdminObservabilityStore { + PgAdminObservabilityStore::new(pool.clone(), None, None, observability_query_budget()) +} + pub(super) fn admin_account_store(pool: &PgPool) -> PgAdminAccountStore { PgAdminAccountStore::new(pool.clone(), None, observability_query_budget()) } diff --git a/backend/crates/gateway-store/tests/postgres/observability.rs b/backend/crates/gateway-store/tests/postgres/observability.rs index ed2fce54c..94b57dd6a 100644 --- a/backend/crates/gateway-store/tests/postgres/observability.rs +++ b/backend/crates/gateway-store/tests/postgres/observability.rs @@ -1,6 +1,5 @@ use std::{ collections::BTreeMap, - num::NonZeroU32, sync::Arc, time::{Duration, SystemTime}, }; @@ -19,13 +18,15 @@ use gateway_core::{ }, }; use gateway_store::postgres::{ - DiagnosticDimension, ObservabilityPageNumber, ObservabilityPageSize, ObservabilityRange, - ObservabilityRepository, OpsErrorFilter, OpsErrorQuery, PgAdminObservabilityStore, - PgObservabilityRepository, ProviderAccountUsageQuery, UsageRecordFilter, UsageRecordQuery, + DiagnosticDimension, ObservabilityPageSize, ObservabilityRange, ObservabilityRepository, + OpsErrorFilter, OpsErrorQuery, PgAdminObservabilityStore, PgObservabilityRepository, + ProviderAccountUsageQuery, UsageRecordFilter, UsageRecordQuery, }; use sqlx::PgPool; -use super::TestDatabase; +use super::{ + TestDatabase, admin_observability_store, observability_query_budget, observability_repository, +}; #[test] fn observability_range_rejects_empty_window() { @@ -96,7 +97,7 @@ async fn observability_preserves_and_filters_opaque_response_ids() { .expect("persist opaque response IDs"); let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) .expect("observability range"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let records = repository .list_usage_records(UsageRecordQuery { @@ -106,24 +107,116 @@ async fn observability_preserves_and_filters_opaque_response_ids() { ..UsageRecordFilter::default() }, cursor: None, - page: ObservabilityPageNumber::new(1).expect("page"), page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("filter opaque response ID"); - assert_eq!(records.total, 1); + assert_eq!(records.total, Some(1)); + assert_eq!(records.items[0].id, "req_observe_success"); + let detail = repository + .usage_record_detail("req_observe_success") + .await + .expect("usage detail with opaque response IDs"); assert_eq!( - records.items[0].client_response_id.as_deref(), + detail.request.client_response_id.as_deref(), Some(response_id.as_str()) ); assert_eq!( - records.items[0].upstream_response_id.as_deref(), + detail.request.upstream_response_id.as_deref(), Some(response_id.as_str()) ); database.close().await; } +#[tokio::test] +async fn usage_page_should_skip_exact_count_when_total_is_not_requested() { + let Some(database) = TestDatabase::create("usage_page_without_total").await else { + return; + }; + let now = Utc::now(); + seed_observability_facts(&database.pool, now) + .await + .expect("seed observability facts"); + let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) + .expect("observability range"); + + let page = observability_repository(&database.pool) + .list_usage_records(UsageRecordQuery { + range, + filter: UsageRecordFilter::default(), + cursor: None, + page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: false, + }) + .await + .expect("usage page without total"); + + assert_eq!(page.total, None); + database.close().await; +} + +#[tokio::test] +async fn usage_search_should_match_literal_prefix_instead_of_substring() { + let Some(database) = TestDatabase::create("usage_literal_prefix_search").await else { + return; + }; + let now = Utc::now(); + seed_observability_facts(&database.pool, now) + .await + .expect("seed observability facts"); + let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) + .expect("observability range"); + + let page = observability_repository(&database.pool) + .list_usage_records(UsageRecordQuery { + range, + filter: UsageRecordFilter { + search: Some("observe_success".to_owned()), + ..UsageRecordFilter::default() + }, + cursor: None, + page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: true, + }) + .await + .expect("usage substring search"); + + assert_eq!(page.total, Some(0)); + database.close().await; +} + +#[tokio::test] +async fn ops_search_should_treat_sql_wildcards_as_literals() { + let Some(database) = TestDatabase::create("ops_literal_wildcard_search").await else { + return; + }; + let now = Utc::now(); + seed_observability_facts(&database.pool, now) + .await + .expect("seed observability facts"); + let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) + .expect("observability range"); + + let page = observability_repository(&database.pool) + .list_ops_errors(OpsErrorQuery { + range, + filter: OpsErrorFilter { + search: Some("req%".to_owned()), + ..OpsErrorFilter::default() + }, + cursor: None, + page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: true, + }) + .await + .expect("ops literal wildcard search"); + + assert_eq!(page.total, Some(0)); + database.close().await; +} + #[tokio::test] async fn dashboard_account_metrics_should_partition_account_statuses() { let Some(database) = TestDatabase::create("dashboard_account_metrics").await else { @@ -161,7 +254,7 @@ async fn dashboard_account_metrics_should_partition_account_statuses() { .execute(&database.pool) .await .expect("seed account metric states"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) .expect("dashboard range"); @@ -246,8 +339,11 @@ async fn dashboard_account_metrics_with_cooldowns_should_only_reclassify_eligibl test_cooldown("acct_metrics_unknown", 6), test_cooldown("acct_metrics_stale_cooldown", 3), ]); - let repository = - PgObservabilityRepository::new(database.pool.clone(), Some(Arc::new(cooldowns))); + let repository = PgObservabilityRepository::new( + database.pool.clone(), + Some(Arc::new(cooldowns)), + observability_query_budget(), + ); let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) .expect("dashboard range"); @@ -371,7 +467,7 @@ async fn calculated_usage_billing_facts_keep_only_completed_calculated_costs() { .expect("seed calculated billing facts"); let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) .expect("observability range"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let facts = repository .usage_calculated_billing_facts(range, UsageRecordFilter::default()) @@ -385,7 +481,7 @@ async fn calculated_usage_billing_facts_keep_only_completed_calculated_costs() { assert_eq!(facts[0].service_tier.as_deref(), Some("priority")); assert_eq!(facts[0].total.amount.as_str(), "1.25"); - let store = PgAdminObservabilityStore::new(database.pool.clone(), None, None); + let store = admin_observability_store(&database.pool); let facts = store .usage_calculated_billing_facts( admin_observability::TimeRange::new(range.start, range.end) @@ -413,24 +509,37 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det let range = admin_observability::TimeRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) .expect("admin observability range"); - let store = PgAdminObservabilityStore::new(database.pool.clone(), None, None); + let store = admin_observability_store(&database.pool); let dashboard = store .dashboard_summary(range, now) .await .expect("admin dashboard summary"); assert_eq!(dashboard.range, range); - assert_eq!(dashboard.requests.request_count, 3); - assert_eq!(dashboard.requests.success_count, 2); - assert_eq!(dashboard.requests.failure_count, 1); - assert_eq!(dashboard.requests.first_token_latency_sum_ms, 120); - assert_eq!(dashboard.requests.latency_sum_ms, 900); - assert_eq!(dashboard.requests.min_latency_ms, Some(900)); - assert_eq!(dashboard.requests.max_latency_ms, Some(900)); - assert_eq!(dashboard.attempts.attempt_count, 4); - assert_eq!(dashboard.attempts.cost_coverage.provider_reported_count, 1); - assert_eq!(dashboard.attempts.cost_coverage.unavailable_count, 0); - assert_eq!(dashboard.attempts.costs[0].amount.as_str(), "1.25"); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.request_count) + .sum::(), + 3 + ); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.first_token_latency_sum_ms) + .sum::(), + 120 + ); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.latency_sum_ms) + .sum::(), + 900 + ); assert_eq!(dashboard.provider_accounts.total, 1); assert_eq!(dashboard.account_usage[0].request_count, 1); assert_eq!(dashboard.account_usage[0].request_buckets.len(), 2); @@ -447,10 +556,7 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det dashboard.recent_requests[0].service_tier.as_deref(), Some("priority") ); - assert_eq!( - dashboard.recent_requests[0].outcome, - admin_observability::RequestOutcome::Succeeded, - ); + assert_eq!(dashboard.recent_requests[0].id, "req_observe_success"); assert_eq!( dashboard.recent_requests[0] .cost_amount @@ -478,6 +584,14 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det .sum::(), 1, ); + assert!( + dashboard + .trend + .iter() + .chain(&dashboard_trend) + .all(|point| point.costs.is_empty()), + "Dashboard 趋势不应触发未展示的成本聚合", + ); let trend = store .usage_trend( @@ -512,12 +626,12 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det range, filter: admin_observability::UsageFilter::default(), cursor: None, - page: admin_page(1), page_size: PageSize::new(1).expect("page size"), + include_total: true, }) .await .expect("first usage page"); - assert_eq!(first_page.total, 1); + assert_eq!(first_page.total, Some(1)); assert_eq!(first_page.items.len(), 1); assert!(first_page.next_cursor.is_none()); assert_eq!( @@ -544,23 +658,14 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det search: Some("req_observe_success".to_owned()), }, cursor: None, - page: admin_page(1), page_size: PageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("fully filtered usage page"); - assert_eq!(filtered.total, 1); - assert_eq!(filtered.items[0].routing_scope, "groups"); - assert_eq!(filtered.items[0].routing_group_refs, ["grp_history"]); - assert_eq!( - filtered.items[0].routing_group_names_snapshot, - ["Historical group"] - ); + assert_eq!(filtered.total, Some(1)); + assert_eq!(filtered.items[0].id, "req_observe_success"); assert_eq!(filtered.items[0].service_tier.as_deref(), Some("priority")); - assert_eq!( - filtered.items[0].upstream_request_id.as_deref(), - Some("upstream_req_success") - ); let other_outcome = store .list_usage_records(admin_observability::UsageQuery { @@ -573,12 +678,12 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det ..admin_observability::UsageFilter::default() }, cursor: None, - page: admin_page(1), page_size: PageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("other outcome filter should reach PostgreSQL"); - assert_eq!(other_outcome.total, 0); + assert_eq!(other_outcome.total, Some(0)); assert!(other_outcome.items.is_empty()); let detail = store @@ -596,6 +701,10 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det detail.request.routing_group_names_snapshot, ["Historical group"] ); + assert_eq!( + detail.request.upstream_request_id.as_deref(), + Some("upstream_req_success") + ); assert_eq!(detail.attempts.len(), 1); assert_eq!( detail.attempts[0].outcome, @@ -645,16 +754,16 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det model: Some("upstream-model".to_owned()), failure_kind: Some("rate_limited".to_owned()), status_code: Some(429), - search: Some("limited".to_owned()), + search: Some("req_observe_failed".to_owned()), ..admin_observability::OpsErrorFilter::default() }, cursor: None, - page: admin_page(1), page_size: PageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("admin ops errors"); - assert_eq!(errors.total, 2); + assert_eq!(errors.total, Some(2)); assert!(errors.items.iter().all(|item| item.occurred_at <= now)); for filter in [ @@ -688,21 +797,17 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det range, filter, cursor: None, - page: admin_page(1), page_size: PageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("fully forwarded ops filter"); - assert_eq!(page.total, 0); + assert_eq!(page.total, Some(0)); } database.close().await; } -fn admin_page(value: u32) -> admin_observability::PageNumber { - admin_observability::PageNumber::new(NonZeroU32::new(value).expect("positive page")) -} - #[tokio::test] async fn dashboard_summary_totals_include_history_outside_selected_range() { let Some(database) = TestDatabase::create("observability_dashboard_totals").await else { @@ -723,13 +828,20 @@ async fn dashboard_summary_totals_include_history_outside_selected_range() { .expect("move historical request outside selected range"); let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) .expect("observability range"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let dashboard = repository .dashboard_summary(range, now) .await .expect("dashboard summary"); - assert_eq!(dashboard.requests.request_count, 2); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.request_count) + .sum::(), + 2 + ); assert!(dashboard.account_usage.is_empty()); let explicit_account_usage = repository @@ -774,51 +886,92 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact .expect("seed observability facts"); let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) .expect("observability range"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let dashboard = repository .dashboard_summary(range, now) .await .expect("dashboard summary"); - assert_eq!(dashboard.requests.request_count, 3); - assert_eq!(dashboard.requests.success_count, 2); - assert_eq!(dashboard.requests.failure_count, 1); - assert_eq!(dashboard.requests.caller_error_count, 0); - assert_eq!(dashboard.requests.cache_eligible_request_count, 1); - assert_eq!(dashboard.requests.cache_hit_request_count, 1); - assert_eq!(dashboard.requests.cache_hit_request_rate(), Some(1.0)); assert_eq!( dashboard - .requests - .latency_percentiles - .p50_ms + .trend + .iter() + .map(|point| point.metrics.request_count) + .sum::(), + 3 + ); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.success_count) + .sum::(), + 2 + ); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.failure_count) + .sum::(), + 1 + ); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.caller_error_count) + .sum::(), + 0 + ); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.cache_eligible_request_count) + .sum::(), + 1 + ); + assert_eq!( + dashboard + .trend + .iter() + .map(|point| point.metrics.cache_hit_request_count) + .sum::(), + 1 + ); + assert_eq!( + dashboard + .trend + .iter() + .find_map(|point| point.metrics.latency_percentiles.p50_ms) .expect("latency p50") .as_f64(), 900.0 ); assert_eq!( dashboard - .requests - .latency_percentiles - .p95_ms + .trend + .iter() + .find_map(|point| point.metrics.latency_percentiles.p95_ms) .expect("latency p95") .as_f64(), 900.0 ); assert_eq!( dashboard - .requests - .latency_percentiles - .p99_ms + .trend + .iter() + .find_map(|point| point.metrics.latency_percentiles.p99_ms) .expect("latency p99") .as_f64(), 900.0 ); assert_eq!( dashboard - .requests - .first_token_latency_percentiles - .p50_ms + .trend + .iter() + .find_map(|point| point.metrics.first_token_latency_percentiles.p50_ms) .expect("first token p50") .as_f64(), 120.0 @@ -900,12 +1053,12 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact ..UsageRecordFilter::default() }, cursor: None, - page: ObservabilityPageNumber::new(1).expect("page"), page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("usage records"); - assert_eq!(usage_page.total, 1); + assert_eq!(usage_page.total, Some(1)); let successful_image = usage_page .items .iter() @@ -913,25 +1066,27 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact .expect("successful image usage record"); assert_eq!( ( - successful_image.websocket_pool.as_deref(), successful_image.service_tier.as_deref(), successful_image .provider_account_authentication_kind .as_deref(), successful_image.image_input_tokens, successful_image.image_output_tokens, - successful_image.image_generation_requested, - successful_image.image_generation_succeeded, ), + (Some("priority"), Some("oauth"), Some(31), Some(9)) + ); + + let successful_detail = repository + .usage_record_detail("req_observe_success") + .await + .expect("successful usage detail"); + assert_eq!( ( - Some("reuse"), - Some("priority"), - Some("oauth"), - Some(31), - Some(9), - true, - Some(true) - ) + successful_detail.request.websocket_pool.as_deref(), + successful_detail.request.image_generation_requested, + successful_detail.request.image_generation_succeeded, + ), + (Some("reuse"), true, Some(true)) ); assert!( @@ -1003,12 +1158,12 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact range, filter: OpsErrorFilter::default(), cursor: None, - page: ObservabilityPageNumber::new(1).expect("page"), page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("ops errors"); - assert_eq!(errors.total, 2); + assert_eq!(errors.total, Some(2)); let request_error = errors .items .iter() diff --git a/backend/crates/gateway-store/tests/postgres/snapshots.rs b/backend/crates/gateway-store/tests/postgres/snapshots.rs index 352415e01..14cb11daa 100644 --- a/backend/crates/gateway-store/tests/postgres/snapshots.rs +++ b/backend/crates/gateway-store/tests/postgres/snapshots.rs @@ -1,13 +1,13 @@ use chrono::{DateTime, TimeDelta, Utc}; use gateway_store::postgres::{ DiagnosticDimension, ModelRequestAttemptStart, ModelRequestRepository, NewModelRequest, - ObservabilityPageNumber, ObservabilityPageSize, ObservabilityRange, ObservabilityRepository, - OpsErrorFilter, OpsErrorQuery, OpsEvent, OpsEventLevel, OpsEventRepository, PgExecutionStore, - PgObservabilityRepository, PgOpsEventRepository, UsageRecordFilter, UsageRecordQuery, + ObservabilityPageSize, ObservabilityRange, ObservabilityRepository, OpsErrorFilter, + OpsErrorQuery, OpsEvent, OpsEventLevel, OpsEventRepository, PgExecutionStore, + PgOpsEventRepository, UsageRecordFilter, UsageRecordQuery, }; use sqlx::PgPool; -use super::TestDatabase; +use super::{TestDatabase, observability_repository}; #[tokio::test] async fn usage_facts_should_accept_statusless_websocket_but_reject_statusless_http() { @@ -46,7 +46,7 @@ async fn usage_facts_should_accept_statusless_websocket_but_reject_statusless_ht .await; finalize_request_without_client_status(&database.pool, "req_statusless_http", started_at).await; - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let page = repository .list_usage_records(usage_query(started_at, UsageRecordFilter::default())) .await @@ -59,9 +59,15 @@ async fn usage_facts_should_accept_statusless_websocket_but_reject_statusless_ht assert_eq!( (page.total, request_ids), - (1, vec!["req_statusless_websocket"]) + (Some(1), vec!["req_statusless_websocket"]) ); + let overview = repository + .usage_summary(range_around(started_at), UsageRecordFilter::default()) + .await + .expect("summarize statusless transport usage"); + assert_eq!(overview.attempts.cost_coverage.unavailable_count, 1); + database.close().await; } @@ -95,7 +101,7 @@ async fn request_snapshots_should_survive_account_deletion() { .await .expect("delete provider account"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let page = repository .list_usage_records(usage_query(started_at, UsageRecordFilter::default())) .await @@ -210,7 +216,7 @@ async fn attempts_should_keep_their_own_account_snapshots() { .await .expect("delete provider accounts"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let detail = repository .usage_record_detail("req_snap_ab") .await @@ -332,14 +338,14 @@ async fn ops_errors_should_keep_request_and_event_snapshots_after_account_deleti .await .expect("delete provider accounts"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let errors = repository .list_ops_errors(OpsErrorQuery { range: range_around(started_at), filter: OpsErrorFilter::default(), cursor: None, - page: ObservabilityPageNumber::new(1).expect("page"), page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: true, }) .await .expect("list ops errors after deletion"); @@ -432,7 +438,7 @@ async fn diagnostics_should_group_same_email_accounts_by_stable_ref() { .await .expect("delete provider accounts"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let diagnostics = repository .usage_diagnostics( range_around(started_at), @@ -504,7 +510,7 @@ async fn diagnostics_should_fallback_to_name_then_ref_for_missing_snapshots() { .expect("insert legacy request without live account"); finalize_request(&database.pool, "req_snap_legacy", started_at).await; - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let diagnostics = repository .usage_diagnostics( range_around(started_at), @@ -526,6 +532,66 @@ async fn diagnostics_should_fallback_to_name_then_ref_for_missing_snapshots() { database.close().await; } +#[tokio::test] +async fn diagnostics_should_prefer_the_latest_non_null_email_snapshot() { + let Some(database) = TestDatabase::create("snapshot_diagnostics_latest_email").await else { + return; + }; + let started_at = Utc::now(); + seed_account( + &database.pool, + "acct_snapshot_history", + "Older Name", + Some("history@example.invalid"), + "oauth", + started_at, + ) + .await; + let store = PgExecutionStore::new(database.pool.clone()); + store + .insert_model_request_with_first_attempt( + new_request("req_snapshot_history_old", started_at), + attempt("req_snapshot_history_old", 1, "acct_snapshot_history"), + ) + .await + .expect("insert older account snapshot"); + finalize_request(&database.pool, "req_snapshot_history_old", started_at).await; + sqlx::query( + "update provider_accounts + set name = 'Newer Name', email = null, updated_at = now() + where id = 'acct_snapshot_history'", + ) + .execute(&database.pool) + .await + .expect("update account snapshot source"); + store + .insert_model_request_with_first_attempt( + new_request( + "req_snapshot_history_new", + started_at + chrono::Duration::seconds(1), + ), + attempt("req_snapshot_history_new", 1, "acct_snapshot_history"), + ) + .await + .expect("insert newer account snapshot"); + finalize_request(&database.pool, "req_snapshot_history_new", started_at).await; + + let diagnostics = observability_repository(&database.pool) + .usage_diagnostics( + range_around(started_at), + UsageRecordFilter::default(), + DiagnosticDimension::Account, + ) + .await + .expect("account diagnostics"); + + assert_eq!(diagnostics.len(), 1); + assert_eq!(diagnostics[0].key, "acct_snapshot_history"); + assert_eq!(diagnostics[0].name, "history@example.invalid"); + assert_eq!(diagnostics[0].request_count, 2); + database.close().await; +} + #[tokio::test] async fn failure_diagnostics_should_only_include_errored_requests() { let Some(database) = TestDatabase::create("snapshot_failure_dimension").await else { @@ -570,7 +636,7 @@ async fn failure_diagnostics_should_only_include_errored_requests() { .await .expect("mark request failed"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let diagnostics = repository .usage_diagnostics( range_around(started_at), @@ -628,7 +694,7 @@ async fn api_key_diagnostics_should_display_key_name_and_fallback_to_ref() { .await .expect("set request b latency"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let diagnostics = repository .usage_diagnostics( range_around(started_at), @@ -695,7 +761,7 @@ async fn renaming_account_should_not_rewrite_historical_snapshots() { .await .expect("rename provider account"); - let repository = PgObservabilityRepository::new(database.pool.clone(), None); + let repository = observability_repository(&database.pool); let page = repository .list_usage_records(usage_query(started_at, UsageRecordFilter::default())) .await @@ -765,8 +831,8 @@ fn usage_query(started_at: DateTime, filter: UsageRecordFilter) -> UsageRec range: range_around(started_at), filter, cursor: None, - page: ObservabilityPageNumber::new(1).expect("page"), page_size: ObservabilityPageSize::new(10).expect("page size"), + include_total: true, } } diff --git a/frontend/src/api/modules/dashboard.ts b/frontend/src/api/modules/dashboard.ts index 7d9464617..19fad8e7e 100644 --- a/frontend/src/api/modules/dashboard.ts +++ b/frontend/src/api/modules/dashboard.ts @@ -1,4 +1,4 @@ -import type { UsageRecord } from './usage' +import type { UsageListRecord } from './usage' import request from '../request' export type DashboardTrendKind = 'usage' | 'latency' | 'errors' @@ -170,7 +170,7 @@ export interface DashboardSummaryResponse { healthTimeline: DashboardHealthTimeline wireProfiles: DashboardWireProfile[] accountUsage: DashboardAccountUsage[] - usageRecords: UsageRecord[] + usageRecords: UsageListRecord[] poolSummary: DashboardPoolSummary capacityInfo: DashboardCapacityInfo rotationStrategy: string diff --git a/frontend/src/api/modules/usage.ts b/frontend/src/api/modules/usage.ts index cc3775fc7..11800b913 100644 --- a/frontend/src/api/modules/usage.ts +++ b/frontend/src/api/modules/usage.ts @@ -1,12 +1,5 @@ import request from '../request' -export interface UsagePageMeta { - page: number - pageSize: number - total: number - totalPages: number -} - export interface UsageTokenDetails { inputTokens: number | null outputTokens: number | null @@ -73,6 +66,35 @@ export interface UsageRecordMetadata { [key: string]: unknown } +export interface UsageListRecord { + id: string + provider: string | null + authenticationKind: string | null + accountId: string | null + accountEmail: string | null + accountName: string | null + route: string + model: string | null + requestedModel: string | null + upstreamModel: string | null + serviceTier: string | null + clientTransport: string + upstreamTransport: string | null + reasoningEffort: string | null + reasoningPreset: string | null + subagentKind: string | null + compact: boolean + tokenDetails: UsageTokenDetails + billing: UsageBilling | null + latencyDetails: UsageLatencyDetails + firstTokenLatencyMs: number | null + latencyMs: number | null + createdAt: string + createdAtDisplay: string + clientIp: string | null + userAgent: string | null +} + export interface UsageRecord { id: string requestId: string @@ -171,9 +193,10 @@ export type UsageRecordDetail = UsageRecord & { } export interface UsageRecordsResponse { - items: UsageRecord[] - page: UsagePageMeta - nextCursor: string | null + items: UsageListRecord[] + currentPage: number + pageSize: number + total: number } export interface OpsErrorMetadata { @@ -209,8 +232,9 @@ export interface OpsError { export interface OpsErrorsResponse { items: OpsError[] - page: UsagePageMeta - nextCursor: string | null + currentPage: number + pageSize: number + total: number } export interface UsageSummaryResponse { @@ -370,36 +394,110 @@ interface UsageRangeQuery { search?: string } -type UsagePagedQuery = UsageRangeQuery & { - page: number +interface PageQuery { + currentPage: number pageSize: number } -type OpsErrorPagedQuery = UsagePagedQuery & { +type UsagePageQuery = UsageRangeQuery & PageQuery + +type OpsErrorPageQuery = UsagePageQuery & { failureClass?: string route?: string } +interface CursorPageResponse { + items: Item[] + total: number | null + nextCursor: string | null +} + +type CursorPageQuery = Query & { + pageSize: number + cursor?: string +} + interface UsageDetailQuery { id: string } type UsageDiagnosticsQuery = UsageRangeQuery & { dimension: string } -export function getUsageRecords(data: UsagePagedQuery) { - return request({ - url: '/api/admin/usage/records', - method: 'GET', - params: data, - }) +export function createUsageRecordsPager() { + return createPageLoader(data => + request>({ + url: '/api/admin/usage/records', + method: 'GET', + params: data, + })) } -export function getOpsErrors(data: OpsErrorPagedQuery) { - return request({ - url: '/api/admin/operations/errors', - method: 'GET', - params: data, - }) +export function createOpsErrorsPager() { + return createPageLoader>(data => + request>({ + url: '/api/admin/operations/errors', + method: 'GET', + params: data, + })) +} + +function createPageLoader( + fetchPage: (query: CursorPageQuery) => Promise>, +) { + const cursors = new Map([[1, undefined]]) + let queryKey = '' + let total = 0 + + function reset(nextQueryKey: string) { + cursors.clear() + cursors.set(1, undefined) + queryKey = nextQueryKey + total = 0 + } + + async function load(data: Query & PageQuery) { + const { currentPage, pageSize, ...query } = data + const nextQueryKey = JSON.stringify([pageSize, query]) + if (queryKey !== nextQueryKey) + reset(nextQueryKey) + + const requestedPage = Math.max(1, Math.trunc(currentPage)) + let page = [...cursors.keys()].reduce( + (known, candidate) => candidate <= requestedPage && candidate > known ? candidate : known, + 1, + ) + let result: CursorPageResponse + + while (page < requestedPage) { + result = await fetchPage({ + ...(query as Query), + pageSize, + cursor: cursors.get(page), + }) + if (result.total != null) + total = result.total + if (!result.nextCursor) + return { items: result.items, currentPage: page, pageSize, total } + cursors.set(page + 1, result.nextCursor) + page += 1 + } + + result = await fetchPage({ + ...(query as Query), + pageSize, + cursor: cursors.get(page), + }) + if (result.total != null) + total = result.total + if (result.nextCursor) + cursors.set(page + 1, result.nextCursor) + else + cursors.delete(page + 1) + + return { items: result.items, currentPage: page, pageSize, total } + } + + return { load } } export function getUsageRecordDetail(data: UsageDetailQuery) { diff --git a/frontend/src/views/dashboard/composables/useDashboard.ts b/frontend/src/views/dashboard/composables/useDashboard.ts index b88f704c6..81f6a24fa 100644 --- a/frontend/src/views/dashboard/composables/useDashboard.ts +++ b/frontend/src/views/dashboard/composables/useDashboard.ts @@ -8,7 +8,6 @@ import { getDashboardSummary, getDashboardTrend } from '@/api' import { withMinimumDuration } from '@/utils/async' import { formatDateTime } from '@/utils/date' import { formatCompactNumber, formatInteger } from '@/utils/number' -import { normalizeUsageRecord } from '@/views/usage/utils/records' export function useDashboard() { const activeTrendKind = shallowRef(normalizeDashboardTrendKind('usage')) @@ -215,7 +214,7 @@ export function dashboardSnapshotView(summary: DashboardSummary | null) { healthTimeline: summary?.healthTimeline ?? emptyHealthTimeline, accountUsage: (summary?.accountUsage ?? []).map(accountUsageItem), wireProfiles: summary?.wireProfiles ?? [], - usageRecords: (summary?.usageRecords ?? []).map(normalizeUsageRecord), + usageRecords: summary?.usageRecords ?? [], poolSummary: summary?.poolSummary ?? null, capacityInfo: summary?.capacityInfo ?? emptyCapacityInfo, rotationStrategy: summary?.rotationStrategy ?? null, diff --git a/frontend/src/views/theme/fixtures/dashboard.ts b/frontend/src/views/theme/fixtures/dashboard.ts index 195b3e15f..4ae683f39 100644 --- a/frontend/src/views/theme/fixtures/dashboard.ts +++ b/frontend/src/views/theme/fixtures/dashboard.ts @@ -3,7 +3,7 @@ import type { DashboardSummaryResponse, DashboardTrendPoint, } from '@/api/modules/dashboard' -import type { UsageRecord } from '@/api/modules/usage' +import type { UsageListRecord } from '@/api/modules/usage' interface PreviewUsageRecordOptions { id: string @@ -104,16 +104,13 @@ function previewHealthStatus(index: number): DashboardHealthStatus { return 'stable' } -function previewUsageRecord(options: PreviewUsageRecordOptions): UsageRecord { +function previewUsageRecord(options: PreviewUsageRecordOptions): UsageListRecord { const totalTokens = options.inputTokens + options.outputTokens const reasoningTokens = Math.round(options.outputTokens * 0.32) const stream = options.clientTransport !== 'http_json' return { id: options.id, - requestId: `req_${options.id}`, - clientApiKeyId: 'preview-client-key', - kind: 'request.completed', provider: options.provider, authenticationKind: options.authenticationKind, accountId: `account_${options.id}`, @@ -124,51 +121,11 @@ function previewUsageRecord(options: PreviewUsageRecordOptions): UsageRecord { requestedModel: options.model, upstreamModel: options.model, serviceTier: options.provider === 'openai' ? 'priority' : 'standard', - statusCode: 200, clientTransport: options.clientTransport, upstreamTransport: options.upstreamTransport, - protocol: stream ? 'responses' : 'chat_completions', - httpVersion: options.upstreamTransport === 'websocket' ? null : 'HTTP/2', - clientStatusCode: options.clientTransport === 'websocket' ? null : 200, - upstreamStatusCode: options.upstreamTransport === 'websocket' ? null : 200, - websocketPool: options.upstreamTransport === 'websocket' ? { kind: 'warm' } : null, - imageGenerationRequested: false, - imageGenerationSucceeded: null, - latencyDetails: { - transportDecisionWaitMs: 4, - upstreamHeadersMs: Math.max(24, options.firstTokenMs - 36), - firstEventMs: options.firstTokenMs, - firstReasoningMs: options.firstTokenMs + 18, - firstTextMs: options.firstTokenMs + 64, - firstTokenMs: options.firstTokenMs, - openaiProcessingMs: options.provider === 'openai' ? options.latencyMs - 72 : undefined, - }, - attemptIndex: 0, - attemptCount: 1, - responseId: `resp_${options.id}`, - upstreamRequestId: `up_${options.id}`, - latencyMs: options.latencyMs, - firstTokenMs: options.firstTokenMs, - inputTokens: options.inputTokens, - outputTokens: options.outputTokens, - cachedTokens: options.cachedTokens, - cacheWriteTokens: 0, - reasoningTokens, - imageInputTokens: 0, - imageOutputTokens: 0, - message: '请求成功', - metadata: { - stream, - apiKind: stream ? 'responses' : 'chat', - }, - createdAt: options.createdAt, - createdAtDisplay: options.createdAtDisplay, - clientIp: options.clientIp, - userAgent: options.userAgent, reasoningEffort: options.reasoningEffort, reasoningPreset: options.reasoningEffort, compact: options.compact ?? false, - requestKind: stream ? 'responses' : 'chat', subagentKind: null, tokenDetails: { inputTokens: options.inputTokens, @@ -202,12 +159,21 @@ function previewUsageRecord(options: PreviewUsageRecordOptions): UsageRecord { serviceTierDisplay: options.provider === 'openai' ? 'Fast' : 'Standard', multiplierDisplay: '1.0x', }, - costs: [{ currency: 'USD', estimatedAmount: options.estimatedCost }], - costCoverage: { known: 1, partial: 0, unknown: 0, notBillable: 0 }, + latencyDetails: { + transportDecisionWaitMs: 4, + upstreamHeadersMs: Math.max(24, options.firstTokenMs - 36), + firstEventMs: options.firstTokenMs, + firstReasoningMs: options.firstTokenMs + 18, + firstTextMs: options.firstTokenMs + 64, + firstTokenMs: options.firstTokenMs, + openaiProcessingMs: options.provider === 'openai' ? options.latencyMs - 72 : undefined, + }, firstTokenLatencyMs: options.firstTokenMs, - firstTokenLatencyMsDisplay: `${options.firstTokenMs} ms`, - latencyMsDisplay: `${options.latencyMs} ms`, - logicalOutcome: 'success', + latencyMs: options.latencyMs, + createdAt: options.createdAt, + createdAtDisplay: options.createdAtDisplay, + clientIp: options.clientIp, + userAgent: options.userAgent, } } diff --git a/frontend/src/views/usage/components/OpsErrorDetailModal.vue b/frontend/src/views/usage/components/OpsErrorDetailModal.vue index f9dfe53de..3dab19be7 100644 --- a/frontend/src/views/usage/components/OpsErrorDetailModal.vue +++ b/frontend/src/views/usage/components/OpsErrorDetailModal.vue @@ -1,5 +1,5 @@ @@ -61,7 +59,7 @@ function errorSummary(record: Awaited>['items'][