diff --git a/.gitignore b/.gitignore index a13f02972..504aefc29 100644 --- a/.gitignore +++ b/.gitignore @@ -34,6 +34,7 @@ Thumbs.db # Local tools and worktrees /.codegraph/ +/.playwright-cli/ /.codex-ws-audit/ /.reasonix/ /.claude diff --git a/backend/crates/gateway-admin/src/model/analytics.rs b/backend/crates/gateway-admin/src/model/analytics.rs index 678a08b52..51e6793b8 100644 --- a/backend/crates/gateway-admin/src/model/analytics.rs +++ b/backend/crates/gateway-admin/src/model/analytics.rs @@ -2,7 +2,9 @@ use chrono::{DateTime, Utc}; -use super::observability::{DecimalAmount, RequestMetricPoint, UsageOverview}; +use super::observability::{ + DecimalAmount, RequestMetricPoint, UsageConcurrencyPoint, UsageOverview, +}; /// 数据可见性状态，不以缺失投影冒充零统计 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] @@ -57,9 +59,10 @@ pub struct UsageProjectedCostPoint { } /// 同一数据库快照内的完整洞察输入 -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq)] pub struct UsageInsightsObservation { pub overview: UsageOverview, pub trend: Vec, pub costs: Vec, + pub concurrency: Vec, } diff --git a/backend/crates/gateway-admin/src/model/observability.rs b/backend/crates/gateway-admin/src/model/observability.rs index 719f0dc77..0970e8149 100644 --- a/backend/crates/gateway-admin/src/model/observability.rs +++ b/backend/crates/gateway-admin/src/model/observability.rs @@ -1260,38 +1260,44 @@ pub struct UsageInsightsHealth { pub points: Vec, } +/// 一个时间桶内按重叠时长加权的逻辑请求并发 +#[derive(Debug, Clone, PartialEq)] +pub struct UsageConcurrencyPoint { + pub bucket_start: DateTime, + pub average_concurrency: f64, + pub peak_concurrency: u64, +} + +/// 请求并发与调度等待的负载指标 +#[derive(Debug, Clone, PartialEq)] +pub struct UsageInsightsLoad { + pub average_concurrency: f64, + pub peak_concurrency: u64, + pub admission_decision_p95_ms: Option, + pub account_selection_wait_p95_ms: Option, +} + /// 洞察性能时间点 -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq)] pub struct UsageInsightsPerformancePoint { pub bucket_start: DateTime, pub latency_percentiles: LatencyPercentiles, pub first_token_latency_percentiles: LatencyPercentiles, - pub admission_decision_percentiles: LatencyPercentiles, - pub account_selection_wait_percentiles: LatencyPercentiles, pub output_throughput_p10: Option, pub output_throughput_p50: Option, pub output_throughput_p90: Option, - pub capacity_utilization_avg_basis_points: Option, - pub capacity_utilization_p95_basis_points: Option, + pub load: UsageInsightsLoad, } -/// 洞察性能汇总与可观测覆盖率 +/// 洞察性能汇总 #[derive(Debug, Clone, PartialEq)] pub struct UsageInsightsPerformance { pub latency_percentiles: LatencyPercentiles, pub first_token_latency_percentiles: LatencyPercentiles, - pub latency_coverage: f64, - pub first_token_coverage: f64, - pub admission_decision_percentiles: LatencyPercentiles, - pub account_selection_wait_percentiles: LatencyPercentiles, - pub admission_decision_coverage: f64, - pub account_selection_wait_coverage: f64, pub output_throughput_p10: Option, pub output_throughput_p50: Option, pub output_throughput_p90: Option, - pub capacity_utilization_avg_basis_points: Option, - pub capacity_utilization_p95_basis_points: Option, - pub capacity_coverage: f64, + pub load: UsageInsightsLoad, pub points: Vec, } diff --git a/backend/crates/gateway-admin/src/use_case/observability/mod.rs b/backend/crates/gateway-admin/src/use_case/observability/mod.rs index 818332de9..0ee5a7192 100644 --- a/backend/crates/gateway-admin/src/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/observability/mod.rs @@ -20,9 +20,10 @@ use crate::{ HealthTimeline, HealthTimelinePoint, OpsErrorPage, OpsErrorQuery, ProviderBillingInput, RequestMetricPoint, RequestMetrics, TimeRange, Trend, TrendKind, TrendPoint, TrendSummary, UsageBilling, UsageBillingProjection, UsageCalculatedBillingFact, - UsageDetail, UsageFilter, UsageInsights, UsageInsightsCost, UsageInsightsCostPoint, - UsageInsightsHealth, UsageInsightsHealthPoint, UsageInsightsPerformance, - UsageInsightsPerformancePoint, UsageOverview, UsagePage, UsageQuery, UsageSummary, + UsageConcurrencyPoint, UsageDetail, UsageFilter, UsageInsights, UsageInsightsCost, + UsageInsightsCostPoint, UsageInsightsHealth, UsageInsightsHealthPoint, + UsageInsightsLoad, UsageInsightsPerformance, UsageInsightsPerformancePoint, + UsageOverview, UsagePage, UsageQuery, UsageSummary, }, provider_credentials::ProviderQuotaRequest, }, @@ -304,7 +305,7 @@ impl ObservabilityService for DefaultObservabilityService { } let overview = observation.overview; let trend = observation.trend; - build_usage_insights(overview, trend, standard_costs) + build_usage_insights(overview, trend, standard_costs, observation.concurrency) } async fn diagnostics( @@ -385,6 +386,7 @@ fn build_usage_insights( overview: UsageOverview, trend: Vec, standard_costs: UsageCostScenarios, + concurrency: Vec, ) -> Result { let granularity = trend.first().map_or( crate::model::observability::Granularity::FifteenMinutes, @@ -432,49 +434,69 @@ fn build_usage_insights( }) .collect(), }; + let concurrency_by_bucket = concurrency + .iter() + .map(|point| (point.bucket_start, point)) + .collect::>(); + let mut weighted_concurrency_micros = 0.0; + let mut effective_micros = 0_i64; + for (index, point) in concurrency.iter().enumerate() { + let start = point.bucket_start.max(overview.range.start); + let end = concurrency + .get(index + 1) + .map_or(overview.range.end, |next| next.bucket_start) + .min(overview.range.end); + let micros = (end - start).num_microseconds().unwrap_or(0).max(0); + effective_micros = effective_micros.saturating_add(micros); + weighted_concurrency_micros += point.average_concurrency * micros as f64; + } + let average_concurrency = if effective_micros > 0 { + weighted_concurrency_micros / effective_micros as f64 + } else { + 0.0 + }; + let peak_concurrency = concurrency + .iter() + .map(|point| point.peak_concurrency) + .max() + .unwrap_or(0); let performance = UsageInsightsPerformance { latency_percentiles: requests.latency_percentiles, first_token_latency_percentiles: requests.first_token_latency_percentiles, - latency_coverage: rate_or_zero(requests.latency_count, requests.request_count), - first_token_coverage: rate_or_zero( - requests.first_token_latency_count, - requests.request_count, - ), - admission_decision_percentiles: requests.admission_decision_percentiles, - account_selection_wait_percentiles: requests.account_selection_wait_percentiles, - admission_decision_coverage: rate_or_zero( - requests.admission_decision_count, - requests.request_count, - ), - account_selection_wait_coverage: rate_or_zero( - requests.account_selection_wait_count, - requests.request_count, - ), output_throughput_p10: requests.output_throughput_p10, output_throughput_p50: requests.output_throughput_p50, output_throughput_p90: requests.output_throughput_p90, - capacity_utilization_avg_basis_points: requests.capacity_utilization_avg_basis_points, - capacity_utilization_p95_basis_points: requests.capacity_utilization_p95_basis_points, - capacity_coverage: rate_or_zero(requests.capacity_sample_count, requests.request_count), + load: UsageInsightsLoad { + average_concurrency, + peak_concurrency, + admission_decision_p95_ms: requests.admission_decision_percentiles.p95_ms, + account_selection_wait_p95_ms: requests.account_selection_wait_percentiles.p95_ms, + }, points: trend .iter() - .map(|point| UsageInsightsPerformancePoint { - bucket_start: point.bucket_start, - latency_percentiles: point.metrics.latency_percentiles, - first_token_latency_percentiles: point.metrics.first_token_latency_percentiles, - admission_decision_percentiles: point.metrics.admission_decision_percentiles, - account_selection_wait_percentiles: point - .metrics - .account_selection_wait_percentiles, - output_throughput_p10: point.metrics.output_throughput_p10, - output_throughput_p50: point.metrics.output_throughput_p50, - output_throughput_p90: point.metrics.output_throughput_p90, - capacity_utilization_avg_basis_points: point - .metrics - .capacity_utilization_avg_basis_points, - capacity_utilization_p95_basis_points: point - .metrics - .capacity_utilization_p95_basis_points, + .map(|point| { + let concurrency = concurrency_by_bucket.get(&point.bucket_start); + UsageInsightsPerformancePoint { + bucket_start: point.bucket_start, + latency_percentiles: point.metrics.latency_percentiles, + first_token_latency_percentiles: point.metrics.first_token_latency_percentiles, + output_throughput_p10: point.metrics.output_throughput_p10, + output_throughput_p50: point.metrics.output_throughput_p50, + output_throughput_p90: point.metrics.output_throughput_p90, + load: UsageInsightsLoad { + average_concurrency: concurrency + .map_or(0.0, |point| point.average_concurrency), + peak_concurrency: concurrency.map_or(0, |point| point.peak_concurrency), + admission_decision_p95_ms: point + .metrics + .admission_decision_percentiles + .p95_ms, + account_selection_wait_p95_ms: point + .metrics + .account_selection_wait_percentiles + .p95_ms, + }, + } }) .collect(), }; diff --git a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs index 4c4b02a46..997b86784 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs @@ -22,8 +22,9 @@ use gateway_admin::{ DiagnosticDimension, DiagnosticObservation, DiagnosticsObservation, Granularity, HealthStatus, LatencyPercentiles, ObservabilityPageSize, OpsErrorPage, OpsErrorQuery, PercentileMilliseconds, RequestMetricPoint, RequestMetrics, TimeRange, TrendKind, - UsageBilling, UsageDetail, UsageFilter, UsageInsightsObservation, UsageListRecord, - UsageOverview, UsagePage, UsageProjectedCostPoint, UsageQuery, UsageSummaryObservation, + UsageBilling, UsageConcurrencyPoint, UsageDetail, UsageFilter, + UsageInsightsObservation, UsageListRecord, UsageOverview, UsagePage, + UsageProjectedCostPoint, UsageQuery, UsageSummaryObservation, }, settings::{ AdminApiKey, AdminApiKeyMutation, ReplaceRuntimeSettings, RotationStrategy, @@ -678,10 +679,26 @@ async fn observability_services_should_calculate_usage_insights_and_diagnostic_s assert_eq!(insights.health.success_rate, 0.75); assert_eq!(insights.health.completion_rate, 1.0); assert_eq!(insights.granularity, Granularity::FifteenMinutes); - assert_eq!(insights.performance.latency_coverage, 0.5); - assert_eq!(insights.performance.admission_decision_coverage, 1.0); - assert_eq!(insights.performance.account_selection_wait_coverage, 0.8); - assert_eq!(insights.performance.capacity_coverage, 0.8); + assert_eq!( + insights + .performance + .load + .admission_decision_p95_ms + .map(|value| value.as_f64()), + Some(5.0) + ); + assert_eq!( + insights + .performance + .load + .account_selection_wait_p95_ms + .map(|value| value.as_f64()), + Some(40.0) + ); + assert_eq!( + insights.performance.points[0].load, + insights.performance.load + ); assert_eq!(insights.performance.output_throughput_p50, Some(50)); assert_eq!(insights.cost.tokens_per_request, 100.0); assert_eq!( @@ -742,6 +759,52 @@ async fn observability_services_should_calculate_usage_insights_and_diagnostic_s assert_eq!(diagnostics.items[1].request_share, 0.25); } +#[tokio::test] +async fn usage_insights_should_weight_concurrency_by_effective_bucket_duration() { + let bucket_start = DateTime::parse_from_rfc3339("2026-09-01T10:00:00Z") + .unwrap() + .with_timezone(&Utc); + let range = TimeRange::new( + bucket_start + Duration::minutes(5), + bucket_start + Duration::minutes(20), + ) + .unwrap(); + let store = Arc::new(FixtureObservabilityStore::new(range)); + store.replace_trend(vec![ + health_metric_point(bucket_start, RequestMetrics::default()), + health_metric_point( + bucket_start + Duration::minutes(15), + RequestMetrics::default(), + ), + ]); + *store.concurrency.lock().unwrap() = vec![ + UsageConcurrencyPoint { + bucket_start, + average_concurrency: 1.0, + peak_concurrency: 1, + }, + UsageConcurrencyPoint { + bucket_start: bucket_start + Duration::minutes(15), + average_concurrency: 3.0, + peak_concurrency: 4, + }, + ]; + + let insights = observability_services(store) + .await + .observability() + .usage_insights(range, UsageFilter::default()) + .await + .unwrap(); + + assert!((insights.performance.load.average_concurrency - 5.0 / 3.0).abs() < 1e-10); + assert_eq!(insights.performance.load.peak_concurrency, 4); + assert_eq!(insights.performance.points[0].load.average_concurrency, 1.0); + assert_eq!(insights.performance.points[0].load.peak_concurrency, 1); + assert_eq!(insights.performance.points[1].load.average_concurrency, 3.0); + assert_eq!(insights.performance.points[1].load.peak_concurrency, 4); +} + #[tokio::test] async fn diagnostics_should_preserve_request_order_when_a_small_sample_fails() { let now = Utc::now(); @@ -987,6 +1050,7 @@ struct FixtureObservabilityStore { trend: Mutex>, overview: Mutex, projected_costs: Mutex>, + concurrency: Mutex>, insights_fail: AtomicBool, summary_fail: AtomicBool, diagnostics: Mutex, @@ -1014,6 +1078,7 @@ impl FixtureObservabilityStore { providers: Vec::new(), }), projected_costs: Mutex::new(Vec::new()), + concurrency: Mutex::new(Vec::new()), insights_fail: AtomicBool::new(false), summary_fail: AtomicBool::new(false), diagnostics: Mutex::new(DiagnosticsObservation::default()), @@ -1144,6 +1209,7 @@ impl ObservabilityStore for FixtureObservabilityStore { .lock() .expect("projected costs") .clone(), + concurrency: self.concurrency.lock().expect("concurrency").clone(), }) } diff --git a/backend/crates/gateway-api/src/admin/observability/presenter.rs b/backend/crates/gateway-api/src/admin/observability/presenter.rs index e5ddedcd2..ed4235d4f 100644 --- a/backend/crates/gateway-api/src/admin/observability/presenter.rs +++ b/backend/crates/gateway-api/src/admin/observability/presenter.rs @@ -41,10 +41,6 @@ pub(crate) fn display_rate(value: f64) -> String { } } -fn basis_points_rate(value: u64) -> f64 { - value as f64 / 10_000.0 -} - pub(crate) fn outcome_name(outcome: &domain::RequestOutcome) -> &str { outcome.as_str() } @@ -1050,6 +1046,17 @@ pub(crate) fn usage_summary_view( } } +fn overview_load_view(load: &domain::UsageInsightsLoad) -> OverviewLoadView { + OverviewLoadView { + average_concurrency: load.average_concurrency, + peak_concurrency: load.peak_concurrency, + admission_decision_p95_ms: load.admission_decision_p95_ms.map(|value| value.as_f64()), + account_selection_wait_p95_ms: load + .account_selection_wait_p95_ms + .map(|value| value.as_f64()), + } +} + pub(crate) fn usage_insights_view( insights: domain::UsageInsights, time: crate::time::TimePresenter, @@ -1092,31 +1099,10 @@ pub(crate) fn usage_insights_view( .first_token_latency_percentiles .p99_ms .map(|value| value.as_f64()), - admission_decision_p50_ms: point - .admission_decision_percentiles - .p50_ms - .map(|value| value.as_f64()), - admission_decision_p95_ms: point - .admission_decision_percentiles - .p95_ms - .map(|value| value.as_f64()), - account_selection_wait_p50_ms: point - .account_selection_wait_percentiles - .p50_ms - .map(|value| value.as_f64()), - account_selection_wait_p95_ms: point - .account_selection_wait_percentiles - .p95_ms - .map(|value| value.as_f64()), output_throughput_p10: point.output_throughput_p10, output_throughput_p50: point.output_throughput_p50, output_throughput_p90: point.output_throughput_p90, - capacity_utilization: point - .capacity_utilization_avg_basis_points - .map(basis_points_rate), - capacity_utilization_p95: point - .capacity_utilization_p95_basis_points - .map(basis_points_rate), + load: overview_load_view(&point.load), }) .collect(); let cost_points = insights @@ -1185,42 +1171,10 @@ pub(crate) fn usage_insights_view( .first_token_latency_percentiles .p99_ms .map(|value| value.as_f64()), - admission_decision_p50_ms: insights - .performance - .admission_decision_percentiles - .p50_ms - .map(|value| value.as_f64()), - admission_decision_p95_ms: insights - .performance - .admission_decision_percentiles - .p95_ms - .map(|value| value.as_f64()), - account_selection_wait_p50_ms: insights - .performance - .account_selection_wait_percentiles - .p50_ms - .map(|value| value.as_f64()), - account_selection_wait_p95_ms: insights - .performance - .account_selection_wait_percentiles - .p95_ms - .map(|value| value.as_f64()), output_throughput_p10: insights.performance.output_throughput_p10, output_throughput_p50: insights.performance.output_throughput_p50, output_throughput_p90: insights.performance.output_throughput_p90, - capacity_utilization: insights - .performance - .capacity_utilization_avg_basis_points - .map(basis_points_rate), - capacity_utilization_p95: insights - .performance - .capacity_utilization_p95_basis_points - .map(basis_points_rate), - latency_coverage: insights.performance.latency_coverage, - first_token_coverage: insights.performance.first_token_coverage, - admission_decision_coverage: insights.performance.admission_decision_coverage, - account_selection_wait_coverage: insights.performance.account_selection_wait_coverage, - capacity_coverage: insights.performance.capacity_coverage, + load: overview_load_view(&insights.performance.load), points: performance_points, }, cost: OverviewCostView { diff --git a/backend/crates/gateway-api/src/admin/observability/wire.rs b/backend/crates/gateway-api/src/admin/observability/wire.rs index 8cabcc265..ceb49536d 100644 --- a/backend/crates/gateway-api/src/admin/observability/wire.rs +++ b/backend/crates/gateway-api/src/admin/observability/wire.rs @@ -648,6 +648,16 @@ pub struct OverviewHealthView { pub points: Vec, } +/// 并发与调度等待的负载指标 +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct OverviewLoadView { + pub average_concurrency: f64, + pub peak_concurrency: u64, + pub admission_decision_p95_ms: Option, + pub account_selection_wait_p95_ms: Option, +} + /// 洞察性能趋势点 #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] @@ -660,15 +670,10 @@ pub struct OverviewPerformancePointView { pub first_token_p50_ms: Option, pub first_token_p95_ms: Option, pub first_token_p99_ms: Option, - pub admission_decision_p50_ms: Option, - pub admission_decision_p95_ms: Option, - pub account_selection_wait_p50_ms: Option, - pub account_selection_wait_p95_ms: Option, pub output_throughput_p10: Option, pub output_throughput_p50: Option, pub output_throughput_p90: Option, - pub capacity_utilization: Option, - pub capacity_utilization_p95: Option, + pub load: OverviewLoadView, } /// 洞察性能摘要 @@ -681,20 +686,10 @@ pub struct OverviewPerformanceView { pub first_token_p50_ms: Option, pub first_token_p95_ms: Option, pub first_token_p99_ms: Option, - pub admission_decision_p50_ms: Option, - pub admission_decision_p95_ms: Option, - pub account_selection_wait_p50_ms: Option, - pub account_selection_wait_p95_ms: Option, pub output_throughput_p10: Option, pub output_throughput_p50: Option, pub output_throughput_p90: Option, - pub capacity_utilization: Option, - pub capacity_utilization_p95: Option, - pub latency_coverage: f64, - pub first_token_coverage: f64, - pub admission_decision_coverage: f64, - pub account_selection_wait_coverage: f64, - pub capacity_coverage: f64, + pub load: OverviewLoadView, pub points: Vec, } diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 796ea6624..63dd63c92 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -697,6 +697,7 @@ pub(super) struct MemoryClientKeyStore(Arc>>); pub(super) struct MemoryObservations { pub summary: Option, pub trend: Option>, + pub concurrency: Vec, pub summaries: Vec<(TimeRange, UsageFilter)>, pub trends: Vec<(TimeRange, UsageFilter)>, pub records: Vec, @@ -1526,6 +1527,7 @@ impl ObservabilityStore for UnusedStore { .clone() .ok_or_else(|| unavailable("usage trend"))?, costs: Vec::new(), + concurrency: data.concurrency.clone(), }, ) } diff --git a/backend/crates/gateway-api/tests/admin/observability/response.rs b/backend/crates/gateway-api/tests/admin/observability/response.rs index 738326453..caa5f883f 100644 --- a/backend/crates/gateway-api/tests/admin/observability/response.rs +++ b/backend/crates/gateway-api/tests/admin/observability/response.rs @@ -1286,6 +1286,102 @@ async fn calendar_query_rejects_an_anchor_outside_the_local_calendar() { assert!(fixture.dashboard_summary_range.lock().unwrap().is_none()); } +#[tokio::test] +async fn usage_insights_maps_load_summary_and_points_to_camel_case() { + use crate::admin::{AdminTestFixture, AdminTestState}; + use axum::{ + body::{Body, to_bytes}, + http::{Request, StatusCode, header}, + }; + use chrono::Duration; + use gateway_admin::model::observability::{ + Granularity, LatencyPercentiles, PercentileMilliseconds, RequestMetricPoint, + RequestMetrics, TimeRange, UsageConcurrencyPoint, UsageOverview, + }; + use gateway_api::admin::observability; + use tower::ServiceExt as _; + + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let now = Utc::now(); + let range = TimeRange::new(now - Duration::minutes(15), now).unwrap(); + let bucket = range.start; + { + let mut observations = fixture.observations.lock().unwrap(); + observations.summary = Some(UsageOverview { + analytics: Default::default(), + range, + requests: RequestMetrics { + admission_decision_percentiles: LatencyPercentiles { + p95_ms: Some(PercentileMilliseconds::new(12.5).unwrap()), + ..Default::default() + }, + account_selection_wait_percentiles: LatencyPercentiles { + p95_ms: Some(PercentileMilliseconds::new(128.75).unwrap()), + ..Default::default() + }, + ..Default::default() + }, + attempts: Default::default(), + providers: Vec::new(), + }); + observations.trend = Some(vec![RequestMetricPoint { + bucket_start: bucket, + granularity: Granularity::FifteenMinutes, + metrics: RequestMetrics { + admission_decision_percentiles: LatencyPercentiles { + p95_ms: Some(PercentileMilliseconds::new(2.5).unwrap()), + ..Default::default() + }, + ..Default::default() + }, + cost_coverage: Default::default(), + costs: Vec::new(), + }]); + observations.concurrency = vec![UsageConcurrencyPoint { + bucket_start: bucket, + average_concurrency: 1.25, + peak_concurrency: 3, + }]; + } + + let response = observability::router::() + .with_state(fixture.state()) + .oneshot( + Request::builder() + .uri("/api/admin/usage/insights/overview") + .header(header::COOKIE, "cpr_session=valid-session") + .header("x-request-id", "req_usage_insights_load") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), 1024 * 1024).await.unwrap(); + let value: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(status, StatusCode::OK, "{value}"); + let performance = &value["data"]["performance"]; + assert_eq!( + performance["load"], + json!({ + "averageConcurrency": 1.25, + "peakConcurrency": 3, + "admissionDecisionP95Ms": 12.5, + "accountSelectionWaitP95Ms": 128.75, + }) + ); + assert_eq!( + performance["points"][0]["load"], + json!({ + "averageConcurrency": 1.25, + "peakConcurrency": 3, + "admissionDecisionP95Ms": 2.5, + "accountSelectionWaitP95Ms": null, + }) + ); +} + #[tokio::test] async fn observability_queries_reject_group_filters_and_analytics_search() { use crate::admin::{AdminTestFixture, AdminTestState}; 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 0e5161ce3..8a37c5f7c 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -294,7 +294,7 @@ impl PgObservabilityRepository { "load usage insights snapshot", |mut connection| async move { let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?; - let (mut overview, trend, costs) = snapshot + let (mut overview, trend, costs, concurrency) = snapshot .insights(range, &filter, granularity, self.timezone) .await?; overview.analytics = snapshot.metadata(); @@ -303,6 +303,7 @@ impl PgObservabilityRepository { overview: admin_usage_overview(overview), trend: trend.into_iter().map(admin_request_metric_point).collect(), costs, + concurrency, }) }, ) diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs index 04c4e904f..5038811f3 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs @@ -339,6 +339,7 @@ impl<'a> AnalyticsSnapshot<'a> { UsageOverview, Vec, Vec, + Vec, )> { let states = self .load(range, filter, None, Some(granularity), timezone, true) @@ -354,6 +355,9 @@ impl<'a> AnalyticsSnapshot<'a> { }) }) .collect::>>()?; + let concurrency = self + .concurrency(range, filter, granularity, timezone) + .await?; // 趋势已包含可合并的完整指标和情景费用，概览直接复用同一份状态 // 精确分位数不可跨桶合并，明细回退时仍按完整区间单独计算概览 let overview = if states.values().any(MetricState::has_exact_percentiles) { @@ -362,7 +366,98 @@ impl<'a> AnalyticsSnapshot<'a> { self.overview_from_state(range, filter, merge_states(states.into_values())?) .await? }; - Ok((overview, trend, costs)) + Ok((overview, trend, costs, concurrency)) + } + + async fn concurrency( + &mut self, + range: ObservabilityRange, + filter: &UsageRecordFilter, + granularity: ObservationGranularity, + timezone: gateway_core::time::DeploymentTimeZone, + ) -> StoreResult> { + filter.validate()?; + self.require_visible(Some(range))?; + let boundaries = if granularity == ObservationGranularity::Day { + calendar_buckets(range, timezone)? + } else { + let seconds = granularity.seconds(); + let mut bucket = + DateTime::from_timestamp(range.start.timestamp().div_euclid(seconds) * seconds, 0) + .ok_or_else(|| invalid("invalid concurrency bucket"))?; + let step = TimeDelta::seconds(seconds); + let mut boundaries = Vec::new(); + while bucket < range.end { + boundaries.push(bucket); + bucket = bucket + .checked_add_signed(step) + .ok_or_else(|| invalid("invalid next concurrency bucket"))?; + } + boundaries + }; + // 按生命周期分开筛选，使终态请求复用 completed_at 索引，运行中请求沿用独立索引 + let mut query = QueryBuilder::::new( + "with buckets as materialized (select greatest(bucket_start,", + ); + query + .push_bind(range.start) + .push("::timestamptz) as effective_start,least(coalesce(lead(bucket_start) over(order by bucket_start),") + .push_bind(range.end) + .push("::timestamptz),") + .push_bind(range.end) + .push("::timestamptz) as bucket_end,bucket_start from unnest(") + .push_bind(boundaries) + .push("::timestamptz[]) input(bucket_start)), matching as (select mr.started_at,mr.completed_at from model_request_observations mr where mr.outcome <> 'running' and mr.completed_at > ") + .push_bind(range.start) + .push(" and mr.started_at < ") + .push_bind(range.end); + push_analytics_filter(&mut query, filter, "mr"); + query + .push(" union all select mr.started_at,") + .push_bind(range.end) + .push("::timestamptz as completed_at from model_request_observations mr where mr.outcome = 'running' and mr.started_at < ") + .push_bind(range.end); + push_analytics_filter(&mut query, filter, "mr"); + // 每个请求只产生两个事件，桶边界只切分时间轴，避免请求与所有桶逐一连接 + // 同时刻先合并增减，再计算下一段并发，遵循左闭右开的执行区间 + query + .push("), events as (select e.occurred_at,e.delta,null::timestamptz as bucket_start from matching m cross join lateral (values (greatest(m.started_at,") + .push_bind(range.start) + .push("::timestamptz),1::bigint),(least(m.completed_at,") + .push_bind(range.end) + .push("::timestamptz),-1::bigint)) e(occurred_at,delta) union all select effective_start,0::bigint,bucket_start from buckets union all select ") + .push_bind(range.end) + .push("::timestamptz,0::bigint,null::timestamptz), + event_points as ( + select occurred_at,sum(delta)::bigint as delta,max(bucket_start) as bucket_start + from events group by occurred_at + ), timeline as ( + select occurred_at,sum(delta) over timeline_order::bigint as concurrency, + max(bucket_start) over timeline_order as bucket_start, + lead(occurred_at) over(order by occurred_at) as next_occurred_at + from event_points + window timeline_order as (order by occurred_at rows between unbounded preceding and current row) + ) + select b.bucket_start, + coalesce(sum(t.concurrency*extract(epoch from t.next_occurred_at-t.occurred_at)) + /nullif(extract(epoch from b.bucket_end-b.effective_start),0),0)::float8 as average_concurrency, + coalesce(max(t.concurrency),0)::bigint as peak_concurrency + from buckets b left join timeline t on t.bucket_start=b.bucket_start and t.next_occurred_at>t.occurred_at + group by b.bucket_start,b.effective_start,b.bucket_end order by b.bucket_start"); + let rows = query + .build() + .fetch_all(&mut *self.tx) + .await + .map_err(|error| postgres_unavailable("aggregate request concurrency", error))?; + rows.iter() + .map(|row| { + Ok(admin_observability::UsageConcurrencyPoint { + bucket_start: get(row, "bucket_start")?, + average_concurrency: get(row, "average_concurrency")?, + peak_concurrency: unsigned(row, "peak_concurrency")?, + }) + }) + .collect() } pub async fn diagnostics( diff --git a/backend/crates/gateway-store/tests/postgres/analytics.rs b/backend/crates/gateway-store/tests/postgres/analytics.rs index dc9fea1ad..a7c497091 100644 --- a/backend/crates/gateway-store/tests/postgres/analytics.rs +++ b/backend/crates/gateway-store/tests/postgres/analytics.rs @@ -1691,6 +1691,156 @@ async fn daily_rank_counts_follow_rebuilds_rollbacks_deletions_and_dirty_days() database.close().await; } +#[tokio::test] +async fn insights_concurrency_counts_logical_request_overlap_once_per_bucket() { + let Some(database) = TestDatabase::create("analytics_concurrency").await else { + return; + }; + let start = DateTime::parse_from_rfc3339("2026-09-01T10:00:00Z") + .unwrap() + .with_timezone(&Utc); + for (id, from, to) in [ + ("concurrency_serial_a", 1, Some(4)), + ("concurrency_serial_b", 4, Some(6)), + ("concurrency_cross_bucket", 10, Some(25)), + ("concurrency_overlap", 12, Some(14)), + ("concurrency_running", 30, None), + ("concurrency_xai", 5, Some(20)), + ("concurrency_instant", 7, Some(7)), + ] { + seed(&database.pool, id, start, &[]).await; + sqlx::query("update model_requests set started_at=$2,completed_at=$3,recovery_lease_expires_at=$2+interval '1 hour',provider_kind=case when id='concurrency_xai' then 'xai' else 'openai' end,provider_account_ref='account_concurrency',upstream_transport='http_sse',upstream_send_state='sent',attempt_count=case when id='concurrency_serial_a' then 3 else 1 end,outcome=case when $3 is null then 'running' else 'succeeded' end where id=$1") + .bind(id) + .bind(start + TimeDelta::minutes(from)) + .bind(to.map(|minute| start + TimeDelta::minutes(minute))) + .execute(&database.pool) + .await + .unwrap(); + } + drain_analytics(&database.pool).await; + let repo = observability_repository(&database.pool); + use gateway_admin::model::observability::Granularity::{FifteenMinutes, Hour}; + + for (provider, from, to, granularity, expected) in [ + ( + Some("openai"), + 0, + 60, + FifteenMinutes, + vec![(0.8, 2), (2.0 / 3.0, 1), (1.0, 1), (1.0, 1)], + ), + (Some("openai"), 0, 10, FifteenMinutes, vec![(0.5, 1)]), + (Some("openai"), 4, 6, FifteenMinutes, vec![(1.0, 1)]), + ( + Some("openai"), + 3, + 17, + FifteenMinutes, + vec![(5.0 / 6.0, 2), (1.0, 1)], + ), + (Some("openai"), 35, 40, FifteenMinutes, vec![(1.0, 1)]), + (Some("openai"), -15, 0, FifteenMinutes, vec![(0.0, 0)]), + (Some("openai"), 0, 60, Hour, vec![(13.0 / 15.0, 2)]), + ( + Some("xai"), + 0, + 60, + FifteenMinutes, + vec![(2.0 / 3.0, 1), (1.0 / 3.0, 1), (0.0, 0), (0.0, 0)], + ), + ( + Some("xai"), + 10, + 17, + FifteenMinutes, + vec![(1.0, 1), (1.0, 1)], + ), + ( + None, + 0, + 60, + FifteenMinutes, + vec![(22.0 / 15.0, 3), (1.0, 2), (1.0, 1), (1.0, 1)], + ), + ] { + let insights = repo + .usage_insights_observation( + ObservabilityRange::new( + start + TimeDelta::minutes(from), + start + TimeDelta::minutes(to), + ) + .unwrap(), + UsageRecordFilter { + provider_kind: provider.map(str::to_owned), + ..Default::default() + }, + granularity, + ) + .await + .unwrap(); + assert_eq!(insights.concurrency.len(), expected.len()); + for (point, (average, peak)) in insights.concurrency.iter().zip(expected) { + assert!( + (point.average_concurrency - average).abs() < 1e-10, + "{provider:?} {from}..{to}: {point:?}" + ); + assert_eq!(point.peak_concurrency, peak, "{provider:?} {from}..{to}"); + } + } + database.close().await; +} + +#[tokio::test] +async fn insights_concurrency_weights_calendar_days_across_dst() { + let Some(database) = TestDatabase::create("analytics_concurrency_days").await else { + return; + }; + let repo = + observability_repository(&database.pool).with_timezone("America/New_York".parse().unwrap()); + for (date, hours) in [("2026-03-08T05:00:00Z", 23), ("2026-11-01T04:00:00Z", 25)] { + let start = DateTime::parse_from_rfc3339(date) + .unwrap() + .with_timezone(&Utc); + let next_day = start + TimeDelta::hours(hours); + for (suffix, from, to) in [("completed", -1, Some(1)), ("running", 12, None)] { + let id = format!("concurrency_{date}_{suffix}"); + seed(&database.pool, &id, start, &[]).await; + sqlx::query("update model_requests set started_at=$2,completed_at=$3,recovery_lease_expires_at=$2+interval '2 days',outcome=case when $3 is null then 'running' else 'succeeded' end,requested_model_id=$4 where id=$1") + .bind(id) + .bind(start + TimeDelta::hours(from)) + .bind(to.map(|hour| start + TimeDelta::hours(hour))) + .bind(date) + .execute(&database.pool) + .await + .unwrap(); + } + drain_analytics(&database.pool).await; + let insights = repo + .usage_insights_observation( + ObservabilityRange::new(start, next_day + TimeDelta::hours(6)).unwrap(), + UsageRecordFilter { + model: Some(date.to_owned()), + ..Default::default() + }, + gateway_admin::model::observability::Granularity::Day, + ) + .await + .unwrap(); + assert_eq!(insights.concurrency.len(), 2); + assert_eq!(insights.concurrency[0].bucket_start, start); + assert!( + (insights.concurrency[0].average_concurrency - (hours - 11) as f64 / hours as f64) + .abs() + < 1e-10 + ); + assert_eq!(insights.concurrency[0].peak_concurrency, 1); + assert_eq!(insights.concurrency[1].bucket_start, next_day); + assert_eq!(insights.concurrency[1].average_concurrency, 1.0); + assert_eq!(insights.concurrency[1].peak_concurrency, 1); + } + database.close().await; +} + #[tokio::test] async fn sql_edge_histograms_preserve_repeated_samples_and_power_of_two_boundaries() { let Some(database) = TestDatabase::create("analytics_sql_histogram_boundaries").await else { diff --git a/docs/api.md b/docs/api.md index 3d697f936..878a77571 100644 --- a/docs/api.md +++ b/docs/api.md @@ -2124,7 +2124,7 @@ errorCode, errorMessage, startedAt, completedAt, expiresAt, createdAt, updatedAt `points[]` 提供趋势图与首页卡片使用的请求数、Tokens、缓存比例、错误数、成功率、 `latencyValue`、首 Token P95、总延迟 P95 和输出吞吐 P50，以及时间标签与展示值 -内部调度分位数和容量分布通过用量洞察查询 +请求并发与内部调度等待通过用量洞察的 `performance.load` 查询 ### Dashboard 客户端身份 @@ -2222,6 +2222,19 @@ OpenAI 推理强度优先取 `input` 中最后一条 `type: "configuration_updat 已登记的尝试即使尚未收到首个上游事件，也会保留真实 attempt。 鉴权、解析、路由和准入等入口拒绝不属于该范围；请求观测仍是可能延迟或丢弃的异步投影 +性能洞察的 `performance.load` 和 `performance.points[].load` 使用相同的负载结构，分别表示查询范围汇总与时间桶指标： + +| 字段 | 含义 | +| --- | --- | +| `averageConcurrency` | 按有效时长加权的平均请求并发 | +| `peakConcurrency` | 最大重叠请求数 | +| `admissionDecisionP95Ms` | 准入判定耗时 P95，单位毫秒，无样本为 `null` | +| `accountSelectionWaitP95Ms` | 账号选择等待耗时 P95，单位毫秒，无样本为 `null` | + +并发按逻辑请求的执行区间计算，不因重试 attempt 重复计数。 +区间从 `started_at` 到 `completed_at`，执行中请求按查询范围末端仍在处理计算；与时间桶重叠的请求参与该桶， +平均值按重叠时长加权，峰值取桶内最大重叠请求数，无请求的时间桶为零。全范围平均按有效查询时长加权，峰值取各桶最大值 + 汇总与洞察中的请求数与 outcome 分布覆盖筛选范围内全部请求；token、缓存、延迟与成本聚合仅统计 已完整交付客户端的成功推理响应。OpenAI 的 `generate: false` 连接与上下文准备记录归类为 `requestKind: "prewarm"`，不进入用量列表、账号用量或额度预测的 Token / 费用覆盖统计，但仍可按 diff --git a/frontend/src/api/modules/usage.ts b/frontend/src/api/modules/usage.ts index a48b00b23..a45eea578 100644 --- a/frontend/src/api/modules/usage.ts +++ b/frontend/src/api/modules/usage.ts @@ -256,47 +256,32 @@ export interface UsageOverviewHealth { points: UsageOverviewHealthPoint[] } -export interface UsageOverviewPerformancePoint { - bucket: string - label: string - latencyP50Ms: number | null - latencyP95Ms: number | null - latencyP99Ms: number | null - firstTokenP50Ms: number | null - firstTokenP95Ms: number | null - firstTokenP99Ms: number | null - admissionDecisionP50Ms: number | null +export interface UsageOverviewLoad { + averageConcurrency: number + peakConcurrency: number admissionDecisionP95Ms: number | null - accountSelectionWaitP50Ms: number | null accountSelectionWaitP95Ms: number | null - outputThroughputP10: number | null - outputThroughputP50: number | null - outputThroughputP90: number | null - capacityUtilization: number | null - capacityUtilizationP95: number | null } -export interface UsageOverviewPerformance { +export interface UsageOverviewPerformanceMetrics { latencyP50Ms: number | null latencyP95Ms: number | null latencyP99Ms: number | null firstTokenP50Ms: number | null firstTokenP95Ms: number | null firstTokenP99Ms: number | null - admissionDecisionP50Ms: number | null - admissionDecisionP95Ms: number | null - accountSelectionWaitP50Ms: number | null - accountSelectionWaitP95Ms: number | null outputThroughputP10: number | null outputThroughputP50: number | null outputThroughputP90: number | null - capacityUtilization: number | null - capacityUtilizationP95: number | null - latencyCoverage: number - firstTokenCoverage: number - admissionDecisionCoverage: number - accountSelectionWaitCoverage: number - capacityCoverage: number + load: UsageOverviewLoad +} + +export interface UsageOverviewPerformancePoint extends UsageOverviewPerformanceMetrics { + bucket: string + label: string +} + +export interface UsageOverviewPerformance extends UsageOverviewPerformanceMetrics { points: UsageOverviewPerformancePoint[] } diff --git a/frontend/src/i18n/locales/en/usage.json b/frontend/src/i18n/locales/en/usage.json index f0d4a3b89..3fd7ec253 100644 --- a/frontend/src/i18n/locales/en/usage.json +++ b/frontend/src/i18n/locales/en/usage.json @@ -61,21 +61,21 @@ }, "performance": { "ttftHelp": "TTFT: time to first token", - "scheduling": "Scheduling", + "concurrency": "Concurrency", + "averageConcurrency": "Average concurrency", + "peakConcurrency": "Peak concurrency", + "load": "Load", + "waitTime": "Duration", "admissionP95": "Admission P95", "accountSelectionP95": "Selection P95", - "capacityP95": "Capacity P95", - "decisionCoverage": "Decision coverage", - "accountSelectionCoverage": "Selection coverage", - "capacityCoverage": "Capacity coverage", "title": "Performance", - "description": "Latency, throughput, scheduling", + "description": "Total latency, TTFT, rate, and load", "performanceIndicators": "Performance indicators", "loading": "Loading performance", "noTtftSamples": "No TTFT samples in this range", "noThroughput": "No throughput samples in this range", - "noScheduling": "No scheduling or capacity samples in this range", - "noLatency": "No latency samples in this range" + "noLatency": "No latency samples in this range", + "noLoad": "No load data in this range" }, "reasoning": { "subagentRequest": "Subagent request", diff --git a/frontend/src/i18n/locales/zh-Hans/usage.json b/frontend/src/i18n/locales/zh-Hans/usage.json index 288dd42c7..b9bab0dde 100644 --- a/frontend/src/i18n/locales/zh-Hans/usage.json +++ b/frontend/src/i18n/locales/zh-Hans/usage.json @@ -61,21 +61,21 @@ }, "performance": { "ttftHelp": "TTFT：首个 Token 耗时", - "scheduling": "调度", + "concurrency": "并发", + "averageConcurrency": "平均并发", + "peakConcurrency": "峰值并发", + "load": "负载", + "waitTime": "耗时", "admissionP95": "准入 P95", - "accountSelectionP95": "选号 P95", - "capacityP95": "容量 P95", - "decisionCoverage": "判定覆盖", - "accountSelectionCoverage": "选号覆盖", - "capacityCoverage": "容量覆盖", + "accountSelectionP95": "选号等待 P95", "title": "响应性能", - "description": "延迟、速率与调度分位", + "description": "总耗时、首个 Token 耗时、速率与负载", "performanceIndicators": "性能指标", "loading": "正在加载性能数据", "noTtftSamples": "当前范围暂无 TTFT 样本", "noThroughput": "当前范围暂无速率样本", - "noScheduling": "当前范围暂无调度或容量样本", - "noLatency": "当前范围暂无延迟样本" + "noLatency": "当前范围暂无延迟样本", + "noLoad": "当前范围暂无负载数据" }, "reasoning": { "subagentRequest": "子代理请求", @@ -119,7 +119,7 @@ "subscription": "订阅", "platformType": "平台" }, - "outputThroughput": "输出速率", + "outputThroughput": "速率", "noValidSample": "无有效样本", "notReachable": "不可达", "unstable": "不稳定", diff --git a/frontend/src/i18n/locales/zh-Hant/usage.json b/frontend/src/i18n/locales/zh-Hant/usage.json index 945c6e786..4bc49b0ac 100644 --- a/frontend/src/i18n/locales/zh-Hant/usage.json +++ b/frontend/src/i18n/locales/zh-Hant/usage.json @@ -61,21 +61,21 @@ }, "performance": { "ttftHelp": "TTFT：首個 Token 耗時", - "scheduling": "調度", + "concurrency": "並發", + "averageConcurrency": "平均並發", + "peakConcurrency": "峰值並發", + "load": "負載", + "waitTime": "耗時", "admissionP95": "准入 P95", - "accountSelectionP95": "選號 P95", - "capacityP95": "容量 P95", - "decisionCoverage": "判定覆蓋", - "accountSelectionCoverage": "選號覆蓋", - "capacityCoverage": "容量覆蓋", + "accountSelectionP95": "選號等待 P95", "title": "響應性能", - "description": "延遲、速率與調度分位", + "description": "總耗時、首個 Token 耗時、速率與負載", "performanceIndicators": "性能指標", "loading": "正在加載性能數據", "noTtftSamples": "當前範圍暫無 TTFT 樣本", "noThroughput": "當前範圍暫無速率樣本", - "noScheduling": "當前範圍暫無調度或容量樣本", - "noLatency": "當前範圍暫無延遲樣本" + "noLatency": "當前範圍暫無延遲樣本", + "noLoad": "當前範圍暫無負載資料" }, "reasoning": { "subagentRequest": "子代理請求", @@ -119,7 +119,7 @@ "subscription": "訂閱", "platformType": "平臺" }, - "outputThroughput": "輸出速率", + "outputThroughput": "速率", "noValidSample": "無有效樣本", "notReachable": "不可達", "unstable": "不穩定", diff --git a/frontend/src/views/usage/components/UsageCostCard.vue b/frontend/src/views/usage/components/UsageCostCard.vue index f5830cf60..2ceb36eca 100644 --- a/frontend/src/views/usage/components/UsageCostCard.vue +++ b/frontend/src/views/usage/components/UsageCostCard.vue @@ -265,7 +265,9 @@ function costAmount(value: string | null, pointIndex: number) { class="@container/cost min-h-90 xl:h-full" >