perf(usage): query insights by card and selected metric (#556)

* fix(usage): keep optional concurrency scans within a bounded budget

Skip unused concurrency work for Key usage and Dashboard trends. Preserve
other insights when exact concurrency exceeds its row, time, or slot budget.

Constraint: Preserve exact overlap semantics and one snapshot for usage metrics
Rejected: Add hourly peaks or truncate intervals | Neither preserves exact concurrency
Confidence: medium
Scope-risk: moderate
Directive: Keep unavailable concurrency distinct from a measured zero
Tested: Rust 1.99.1 rustfmt on changed Rust files; git diff --check; independent source review
Not-tested: PostgreSQL integration, Clippy, and frontend checks pending isolated CI
Co-authored-by: Codex <noreply@openai.com>

* refactor(analytics): limit concurrency changes to unused results

Keep Key usage and Dashboard skip behavior while retaining the existing concurrency calculation and query budget.

Co-authored-by: Codex <noreply@openai.com>

* perf(usage): query insight cards by selected metric

Split health, performance and cost reads by card and selected metric, with independent frontend requests and metric-aware caching. Only the load metric reads concurrency.

Co-authored-by: Codex <noreply@openai.com>

---------

Co-authored-by: Codex <noreply@openai.com>
Co-authored-by: Zyy <zyycn@users.noreply.github.com>
This commit is contained in:
think-back
2026-10-30 15:09:24 +08:00
committed by GitHub
co-authored by Codex Zyy
parent aa0ce52034
commit 3a2ccbc5e6
33 changed files with 1680 additions and 417 deletions
@@ -1352,6 +1352,84 @@ pub struct UsageInsights {
pub providers: Vec<ProviderObservation>,
}
/// 单次洞察查询所需的指标,持久化层据此选择计数、分布与费用状态
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum UsageInsightMetric {
Health,
Latency,
FirstToken,
Throughput,
Load,
Cost,
Savings,
Cache,
}
/// 单个指标的时间点,活动计数与指标共用读取快照
#[derive(Debug, Clone, PartialEq)]
pub struct UsageMetricPoint<T> {
pub bucket_start: DateTime<Utc>,
pub request_count: u64,
pub value: T,
}
/// 各指标复用时间序列形状,汇总与时间点保留各自的类型
#[derive(Debug, Clone, PartialEq)]
pub struct UsageMetricSeries<S, P = S> {
pub summary: S,
pub points: Vec<UsageMetricPoint<P>>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct UsageThroughputPercentiles {
pub p10: Option<u64>,
pub p50: Option<u64>,
pub p90: Option<u64>,
}
/// 成本卡片各视图共同显示的汇总
#[derive(Debug, Clone, PartialEq)]
pub struct UsageCostSummary {
pub estimated_cost: Option<DecimalAmount>,
pub cache_savings: Option<DecimalAmount>,
pub tier_premium: Option<DecimalAmount>,
pub cost_per_successful_request: Option<DecimalAmount>,
pub coverage: CostCoverage,
}
#[derive(Debug, Clone, PartialEq)]
pub struct UsageCostComparison {
pub estimated_cost: Option<DecimalAmount>,
pub no_cache_cost: Option<DecimalAmount>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct UsageCacheEfficiency {
pub input_tokens: u64,
pub cached_token_rate: f64,
pub cache_hit_request_rate: Option<f64>,
}
/// 只发布本次请求的指标,未查询的值不伪装成零
#[derive(Debug, Clone, PartialEq)]
pub enum UsageInsightData {
Health(UsageInsightsHealth),
Latency(UsageMetricSeries<LatencyPercentiles>),
FirstToken(UsageMetricSeries<LatencyPercentiles>),
Throughput(UsageMetricSeries<UsageThroughputPercentiles>),
Load(UsageMetricSeries<UsageInsightsLoad>),
Cost(UsageMetricSeries<UsageCostSummary, UsageCostComparison>),
Savings(UsageMetricSeries<UsageCostSummary, Option<DecimalAmount>>),
Cache(UsageMetricSeries<UsageCostSummary, UsageCacheEfficiency>),
}
#[derive(Debug, Clone, PartialEq)]
pub struct UsageInsight {
pub analytics: AnalyticsMetadata,
pub granularity: Granularity,
pub data: UsageInsightData,
}
/// 已完成分母计算的诊断项
#[derive(Debug, Clone, PartialEq)]
pub struct DiagnosticsItem {
@@ -30,7 +30,8 @@ use crate::model::{
DashboardObservation, DashboardQuery, DiagnosticDimension, DiagnosticsObservation,
Granularity, OpsErrorPage, OpsErrorQuery, RequestMetricPoint, TimeRange,
UsageBillingProjection, UsageCalculatedBillingFact, UsageDetail, UsageFilter,
UsageInsightsObservation, UsagePage, UsageQuery, UsageSummaryObservation,
UsageInsightMetric, UsageInsightsObservation, UsagePage, UsageQuery,
UsageSummaryObservation,
},
provider_credentials::{
AuthorizationCommit, CredentialDetails, CredentialImportCommit, CredentialImportResult,
@@ -456,11 +457,22 @@ pub trait ObservabilityStore: Send + Sync {
) -> AdminStoreResult<Vec<RequestMetricPoint>>;
/// 同一读取快照中的汇总、趋势和已投影情景费用
/// 不需要并发时关闭 include_concurrency,返回空并发序列并跳过明细计算
async fn usage_insights_observation(
&self,
range: TimeRange,
filter: UsageFilter,
granularity: Granularity,
include_concurrency: bool,
) -> AdminStoreResult<UsageInsightsObservation>;
/// 只读取所选指标及其可见汇总依赖,未选择的字段不供业务层消费
async fn usage_metric_observation(
&self,
range: TimeRange,
filter: UsageFilter,
granularity: Granularity,
metric: UsageInsightMetric,
) -> AdminStoreResult<UsageInsightsObservation>;
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage>;
@@ -194,7 +194,8 @@ impl KeyUsageService for DefaultKeyUsageService {
self.observations.usage_insights_observation(
query.range,
filter,
Granularity::for_range(query.range)
Granularity::for_range(query.range),
false,
),
self.observations.usage_trend(
today,
@@ -19,11 +19,14 @@ use crate::{
DiagnosticDimension, DiagnosticsItem, DiagnosticsResult, Granularity, HealthStatus,
HealthTimeline, HealthTimelinePoint, OpsErrorPage, OpsErrorQuery, ProviderBillingInput,
RequestMetricPoint, RequestMetrics, TimeRange, Trend, TrendKind, TrendPoint,
TrendSummary, UsageBilling, UsageBillingProjection, UsageCalculatedBillingFact,
UsageConcurrencyPoint, UsageDetail, UsageFilter, UsageInsights, UsageInsightsCost,
UsageInsightsCostPoint, UsageInsightsHealth, UsageInsightsHealthPoint,
UsageInsightsLoad, UsageInsightsPerformance, UsageInsightsPerformancePoint,
UsageOverview, UsagePage, UsageQuery, UsageSummary,
TrendSummary, UsageBilling, UsageBillingProjection, UsageCacheEfficiency,
UsageCalculatedBillingFact, UsageConcurrencyPoint, UsageCostComparison,
UsageCostSummary, UsageDetail, UsageFilter, UsageInsight, UsageInsightData,
UsageInsightMetric, UsageInsights, UsageInsightsCost, UsageInsightsCostPoint,
UsageInsightsHealth, UsageInsightsHealthPoint, UsageInsightsLoad,
UsageInsightsObservation, UsageInsightsPerformance, UsageInsightsPerformancePoint,
UsageMetricPoint, UsageMetricSeries, UsageOverview, UsagePage, UsageQuery,
UsageSummary, UsageThroughputPercentiles,
},
provider_credentials::ProviderQuotaRequest,
},
@@ -65,6 +68,12 @@ pub trait ObservabilityService: Send + Sync {
range: TimeRange,
filter: UsageFilter,
) -> Result<UsageInsights, AdminError>;
async fn usage_metric(
&self,
range: TimeRange,
filter: UsageFilter,
metric: UsageInsightMetric,
) -> Result<UsageInsight, AdminError>;
async fn diagnostics(
&self,
range: TimeRange,
@@ -234,6 +243,7 @@ impl ObservabilityService for DefaultObservabilityService {
range,
UsageFilter::default(),
Granularity::for_range(range),
false,
)
.await
.map_err(|error| map_store_error(error, "dashboard trend"))?;
@@ -291,21 +301,25 @@ impl ObservabilityService for DefaultObservabilityService {
let granularity = Granularity::for_range(range);
let observation = self
.store
.usage_insights_observation(range, filter, granularity)
.usage_insights_observation(range, filter, granularity, true)
.await
.map_err(|error| map_store_error(error, "usage insights"))?;
let mut standard_costs = UsageCostScenarios::default();
for point in observation.costs {
if let Some(amount) = point.standard_amount {
standard_costs.standard.add(point.bucket_start, amount)?;
}
if let Some(amount) = point.no_cache_amount {
standard_costs.no_cache.add(point.bucket_start, amount)?;
}
}
let overview = observation.overview;
let trend = observation.trend;
build_usage_insights(overview, trend, standard_costs, observation.concurrency)
insights_from_observation(observation)
}
async fn usage_metric(
&self,
range: TimeRange,
filter: UsageFilter,
metric: UsageInsightMetric,
) -> Result<UsageInsight, AdminError> {
let observation = self
.store
.usage_metric_observation(range, filter, Granularity::for_range(range), metric)
.await
.map_err(|error| map_store_error(error, "usage metric"))?;
let insights = insights_from_observation(observation)?;
Ok(select_usage_metric(insights, metric))
}
async fn diagnostics(
@@ -382,6 +396,148 @@ impl ObservabilityService for DefaultObservabilityService {
}
}
fn insights_from_observation(
observation: UsageInsightsObservation,
) -> Result<UsageInsights, AdminError> {
let mut standard_costs = UsageCostScenarios::default();
for point in observation.costs {
if let Some(amount) = point.standard_amount {
standard_costs.standard.add(point.bucket_start, amount)?;
}
if let Some(amount) = point.no_cache_amount {
standard_costs.no_cache.add(point.bucket_start, amount)?;
}
}
let overview = observation.overview;
let trend = observation.trend;
build_usage_insights(overview, trend, standard_costs, observation.concurrency)
}
fn select_usage_metric(insights: UsageInsights, metric: UsageInsightMetric) -> UsageInsight {
let activity = insights
.health
.points
.iter()
.map(|point| (point.bucket_start, point.total_requests))
.collect::<BTreeMap<_, _>>();
let data = match metric {
UsageInsightMetric::Health => UsageInsightData::Health(insights.health),
UsageInsightMetric::Latency => UsageInsightData::Latency(metric_series(
insights.performance.latency_percentiles,
insights
.performance
.points
.into_iter()
.map(|point| (point.bucket_start, point.latency_percentiles)),
&activity,
)),
UsageInsightMetric::FirstToken => UsageInsightData::FirstToken(metric_series(
insights.performance.first_token_latency_percentiles,
insights
.performance
.points
.into_iter()
.map(|point| (point.bucket_start, point.first_token_latency_percentiles)),
&activity,
)),
UsageInsightMetric::Throughput => UsageInsightData::Throughput(metric_series(
UsageThroughputPercentiles {
p10: insights.performance.output_throughput_p10,
p50: insights.performance.output_throughput_p50,
p90: insights.performance.output_throughput_p90,
},
insights.performance.points.into_iter().map(|point| {
(
point.bucket_start,
UsageThroughputPercentiles {
p10: point.output_throughput_p10,
p50: point.output_throughput_p50,
p90: point.output_throughput_p90,
},
)
}),
&activity,
)),
UsageInsightMetric::Load => UsageInsightData::Load(metric_series(
insights.performance.load,
insights
.performance
.points
.into_iter()
.map(|point| (point.bucket_start, point.load)),
&activity,
)),
UsageInsightMetric::Cost => UsageInsightData::Cost(metric_series(
cost_summary(&insights.cost),
insights.cost.points.into_iter().map(|point| {
(
point.bucket_start,
UsageCostComparison {
estimated_cost: point.estimated_cost,
no_cache_cost: point.no_cache_cost,
},
)
}),
&activity,
)),
UsageInsightMetric::Savings => UsageInsightData::Savings(metric_series(
cost_summary(&insights.cost),
insights
.cost
.points
.into_iter()
.map(|point| (point.bucket_start, point.cache_savings)),
&activity,
)),
UsageInsightMetric::Cache => UsageInsightData::Cache(metric_series(
cost_summary(&insights.cost),
insights.cost.points.into_iter().map(|point| {
(
point.bucket_start,
UsageCacheEfficiency {
input_tokens: point.input_tokens,
cached_token_rate: point.cached_token_rate,
cache_hit_request_rate: point.cache_hit_request_rate,
},
)
}),
&activity,
)),
};
UsageInsight {
analytics: insights.analytics,
granularity: insights.granularity,
data,
}
}
fn cost_summary(cost: &UsageInsightsCost) -> UsageCostSummary {
UsageCostSummary {
estimated_cost: cost.estimated_cost.clone(),
cache_savings: cost.cache_savings.clone(),
tier_premium: cost.tier_premium.clone(),
cost_per_successful_request: cost.cost_per_successful_request.clone(),
coverage: cost.coverage.clone(),
}
}
fn metric_series<S, P>(
summary: S,
points: impl Iterator<Item = (DateTime<Utc>, P)>,
activity: &BTreeMap<DateTime<Utc>, u64>,
) -> UsageMetricSeries<S, P> {
UsageMetricSeries {
summary,
points: points
.map(|(bucket_start, value)| UsageMetricPoint {
bucket_start,
request_count: activity.get(&bucket_start).copied().unwrap_or_default(),
value,
})
.collect(),
}
}
fn build_usage_insights(
overview: UsageOverview,
trend: Vec<RequestMetricPoint>,
@@ -178,7 +178,8 @@ pub(crate) struct CachedObservabilityStore {
inner: Arc<dyn ObservabilityStore>,
dashboard: Cache<DashboardQuery, DashboardObservation>,
summaries: Cache<UsageKey, UsageSummaryObservation>,
insights: Cache<(UsageKey, Granularity), UsageInsightsObservation>,
insights: Cache<(UsageKey, Granularity, bool), UsageInsightsObservation>,
metrics: Cache<(UsageKey, Granularity, UsageInsightMetric), UsageInsightsObservation>,
diagnostics: Cache<(UsageKey, DiagnosticDimension, u16), DiagnosticsObservation>,
}
impl CachedObservabilityStore {
@@ -188,6 +189,7 @@ impl CachedObservabilityStore {
dashboard: Cache::default(),
summaries: Cache::default(),
insights: Cache::default(),
metrics: Cache::default(),
diagnostics: Cache::default(),
}
}
@@ -237,19 +239,44 @@ impl ObservabilityStore for CachedObservabilityStore {
range: TimeRange,
filter: UsageFilter,
granularity: Granularity,
include_concurrency: bool,
) -> AdminStoreResult<UsageInsightsObservation> {
self.insights
.get(
((range, filter.clone()), granularity),
((range, filter.clone()), granularity, include_concurrency),
self.inner.as_ref(),
TTL,
|| {
self.inner
.usage_insights_observation(range, filter, granularity)
self.inner.usage_insights_observation(
range,
filter,
granularity,
include_concurrency,
)
},
)
.await
}
async fn usage_metric_observation(
&self,
range: TimeRange,
filter: UsageFilter,
granularity: Granularity,
metric: UsageInsightMetric,
) -> AdminStoreResult<UsageInsightsObservation> {
self.metrics
.get(
((range, filter.clone()), granularity, metric),
self.inner.as_ref(),
TTL,
|| {
self.inner
.usage_metric_observation(range, filter, granularity, metric)
},
)
.await
}
async fn usage_summary(
&self,
range: TimeRange,
@@ -805,10 +805,27 @@ impl ObservabilityStore for UnavailableStore {
_: TimeRange,
_: UsageFilter,
_: Granularity,
_include_concurrency: bool,
) -> AdminStoreResult<gateway_admin::model::observability::UsageInsightsObservation> {
Err(unavailable("usage insights"))
}
async fn usage_metric_observation(
&self,
range: gateway_admin::model::observability::TimeRange,
filter: gateway_admin::model::observability::UsageFilter,
granularity: gateway_admin::model::observability::Granularity,
metric: gateway_admin::model::observability::UsageInsightMetric,
) -> AdminStoreResult<gateway_admin::model::observability::UsageInsightsObservation> {
self.usage_insights_observation(
range,
filter,
granularity,
metric == gateway_admin::model::observability::UsageInsightMetric::Load,
)
.await
}
async fn list_usage_records(&self, _: UsageQuery) -> AdminStoreResult<UsagePage> {
Err(unavailable("usage records"))
}
@@ -1046,6 +1046,50 @@ async fn dashboard_should_preserve_provider_plan_names() {
}
}
#[tokio::test]
async fn dashboard_and_usage_insights_keep_concurrency_cache_entries_separate() {
for dashboard_first in [true, false] {
let now = Utc::now();
let range = observation_range(now);
let store = Arc::new(FixtureObservabilityStore::new(range));
store
.concurrency
.lock()
.unwrap()
.push(UsageConcurrencyPoint {
bucket_start: range.start,
average_concurrency: 2.0,
peak_concurrency: 3,
});
let services = observability_services(store.clone()).await;
for dashboard in [
dashboard_first,
!dashboard_first,
dashboard_first,
!dashboard_first,
] {
if dashboard {
services
.observability()
.dashboard_trend(range, TrendKind::Usage)
.await
.unwrap();
} else {
let result = services
.observability()
.usage_insights(range, UsageFilter::default())
.await
.unwrap();
assert_eq!(result.performance.load.peak_concurrency, 3);
}
}
assert_eq!(
*store.concurrency_selections.lock().unwrap(),
vec![!dashboard_first, dashboard_first]
);
}
}
struct FixtureObservabilityStore {
yesterday: Mutex<RequestMetrics>,
trend: Mutex<Vec<RequestMetricPoint>>,
@@ -1053,6 +1097,8 @@ struct FixtureObservabilityStore {
projected_costs: Mutex<Vec<UsageProjectedCostPoint>>,
concurrency: Mutex<Vec<UsageConcurrencyPoint>>,
insights_fail: AtomicBool,
concurrency_selections: Mutex<Vec<bool>>,
metric_selections: Mutex<Vec<gateway_admin::model::observability::UsageInsightMetric>>,
summary_fail: AtomicBool,
diagnostics: Mutex<DiagnosticsObservation>,
runtime_slots: Mutex<Option<DashboardRuntimeSlots>>,
@@ -1081,6 +1127,8 @@ impl FixtureObservabilityStore {
projected_costs: Mutex::new(Vec::new()),
concurrency: Mutex::new(Vec::new()),
insights_fail: AtomicBool::new(false),
concurrency_selections: Mutex::new(Vec::new()),
metric_selections: Mutex::new(Vec::new()),
summary_fail: AtomicBool::new(false),
diagnostics: Mutex::new(DiagnosticsObservation::default()),
runtime_slots: Mutex::new(None),
@@ -1197,8 +1245,13 @@ impl ObservabilityStore for FixtureObservabilityStore {
_: TimeRange,
_: UsageFilter,
granularity: Granularity,
include_concurrency: bool,
) -> AdminStoreResult<UsageInsightsObservation> {
self.usage_granularities.lock().unwrap().push(granularity);
self.concurrency_selections
.lock()
.unwrap()
.push(include_concurrency);
if self.insights_fail.load(Ordering::SeqCst) {
return Err(super::unavailable("projection read failed"));
}
@@ -1210,10 +1263,31 @@ impl ObservabilityStore for FixtureObservabilityStore {
.lock()
.expect("projected costs")
.clone(),
concurrency: self.concurrency.lock().expect("concurrency").clone(),
concurrency: if include_concurrency {
self.concurrency.lock().expect("concurrency").clone()
} else {
Vec::new()
},
})
}
async fn usage_metric_observation(
&self,
range: gateway_admin::model::observability::TimeRange,
filter: gateway_admin::model::observability::UsageFilter,
granularity: gateway_admin::model::observability::Granularity,
metric: gateway_admin::model::observability::UsageInsightMetric,
) -> AdminStoreResult<gateway_admin::model::observability::UsageInsightsObservation> {
self.metric_selections.lock().unwrap().push(metric);
self.usage_insights_observation(
range,
filter,
granularity,
metric == gateway_admin::model::observability::UsageInsightMetric::Load,
)
.await
}
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage> {
let items = self.usage_records.lock().expect("usage records").clone();
let total = u64::try_from(items.len()).unwrap_or(u64::MAX);
@@ -1741,3 +1815,48 @@ async fn insights_cache_should_reject_deleted_building_and_old_generation_times(
);
}
}
#[tokio::test]
async fn usage_metric_cache_isolates_selections_and_filters() {
use gateway_admin::model::observability::UsageInsightMetric as Metric;
let range = observation_range(Utc::now());
let store = Arc::new(FixtureObservabilityStore::new(range));
let services = observability_services(store.clone()).await;
let metrics = [
Metric::Health,
Metric::Latency,
Metric::FirstToken,
Metric::Throughput,
Metric::Load,
Metric::Cost,
Metric::Savings,
Metric::Cache,
];
for _ in 0..2 {
for metric in metrics {
services
.observability()
.usage_metric(range, UsageFilter::default(), metric)
.await
.unwrap();
}
}
assert_eq!(*store.metric_selections.lock().unwrap(), metrics);
assert_eq!(
*store.concurrency_selections.lock().unwrap(),
vec![false, false, false, false, true, false, false, false]
);
services
.observability()
.usage_metric(
range,
UsageFilter {
provider_kind: Some("openai".into()),
..Default::default()
},
Metric::Latency,
)
.await
.unwrap();
assert_eq!(store.metric_selections.lock().unwrap().len(), 9);
}
@@ -1062,22 +1062,6 @@ pub(crate) fn usage_insights_view(
insights: domain::UsageInsights,
time: crate::time::TimePresenter,
) -> UsageInsightsOverviewView {
let health_points = insights
.health
.points
.iter()
.map(|point| OverviewHealthPointView {
bucket: point.bucket_start,
label: time.label(point.bucket_start, "%m-%d %H:%M"),
total_requests: point.total_requests,
success_requests: point.success_requests,
failed_requests: point.failed_requests,
cancelled_requests: point.cancelled_requests,
incomplete_requests: point.incomplete_requests,
caller_error_requests: point.caller_error_requests,
error_rate: point.error_rate,
})
.collect();
let performance_points = insights
.performance
.points
@@ -1128,19 +1112,7 @@ pub(crate) fn usage_insights_view(
UsageInsightsOverviewView {
analytics: analytics_view(insights.analytics, time),
granularity: granularity_name(insights.granularity).to_owned(),
health: OverviewHealthView {
total_requests: insights.health.total_requests,
success_requests: insights.health.success_requests,
failed_requests: insights.health.failed_requests,
cancelled_requests: insights.health.cancelled_requests,
incomplete_requests: insights.health.incomplete_requests,
caller_error_requests: insights.health.caller_error_requests,
success_rate: insights.health.success_rate,
completion_rate: insights.health.completion_rate,
request_change_rate: None,
success_rate_change: None,
points: health_points,
},
health: overview_health_view(&insights.health, time),
performance: OverviewPerformanceView {
latency_p50_ms: insights
.performance
@@ -1236,6 +1208,116 @@ pub(crate) fn usage_insights_view(
}
}
pub(crate) fn usage_metric_view(
insight: domain::UsageInsight,
time: crate::time::TimePresenter,
) -> UsageMetricView {
use domain::UsageInsightData;
let data = match insight.data {
UsageInsightData::Health(health) => {
UsageMetricDataView::Health(overview_health_view(&health, time))
}
UsageInsightData::Latency(series) => UsageMetricDataView::Latency(metric_series_view(
series,
time,
metric_latency_view,
metric_latency_view,
)),
UsageInsightData::FirstToken(series) => UsageMetricDataView::FirstToken(
metric_series_view(series, time, metric_latency_view, metric_latency_view),
),
UsageInsightData::Throughput(series) => UsageMetricDataView::Throughput(
metric_series_view(series, time, metric_throughput_view, metric_throughput_view),
),
UsageInsightData::Load(series) => UsageMetricDataView::Load(metric_series_view(
series,
time,
|value| overview_load_view(&value),
|value| overview_load_view(&value),
)),
UsageInsightData::Cost(series) => UsageMetricDataView::Cost(metric_series_view(
series,
time,
metric_cost_summary_view,
|value| UsageCostComparisonView {
estimated_cost: value.estimated_cost.as_ref().map(ToString::to_string),
no_cache_cost: value.no_cache_cost.as_ref().map(ToString::to_string),
},
)),
UsageInsightData::Savings(series) => UsageMetricDataView::Savings(metric_series_view(
series,
time,
metric_cost_summary_view,
|value| value.as_ref().map(ToString::to_string),
)),
UsageInsightData::Cache(series) => UsageMetricDataView::Cache(metric_series_view(
series,
time,
metric_cost_summary_view,
|value| UsageCacheView {
input_tokens: value.input_tokens,
cached_token_rate: value.cached_token_rate,
cache_hit_request_rate: value.cache_hit_request_rate,
},
)),
};
UsageMetricView {
analytics: analytics_view(insight.analytics, time),
granularity: granularity_name(insight.granularity).to_owned(),
data,
}
}
fn metric_series_view<S, P, SV, PV>(
series: domain::UsageMetricSeries<S, P>,
time: crate::time::TimePresenter,
summary: impl FnOnce(S) -> SV,
point: impl Fn(P) -> PV,
) -> UsageMetricSeriesView<SV, PV> {
UsageMetricSeriesView {
summary: summary(series.summary),
points: series
.points
.into_iter()
.map(|value| UsageMetricPointView {
bucket: value.bucket_start,
label: time.label(value.bucket_start, "%m-%d %H:%M"),
request_count: value.request_count,
value: point(value.value),
})
.collect(),
}
}
fn metric_latency_view(value: domain::LatencyPercentiles) -> UsageLatencyView {
UsageLatencyView {
p50_ms: value.p50_ms.map(|value| value.as_f64()),
p95_ms: value.p95_ms.map(|value| value.as_f64()),
p99_ms: value.p99_ms.map(|value| value.as_f64()),
}
}
fn metric_throughput_view(value: domain::UsageThroughputPercentiles) -> UsageThroughputView {
UsageThroughputView {
p10: value.p10,
p50: value.p50,
p90: value.p90,
}
}
fn metric_cost_summary_view(value: domain::UsageCostSummary) -> UsageCostSummaryView {
UsageCostSummaryView {
estimated_cost: value.estimated_cost.as_ref().map(ToString::to_string),
cache_savings: value.cache_savings.as_ref().map(ToString::to_string),
tier_premium: value.tier_premium.as_ref().map(ToString::to_string),
cost_per_successful_request: value
.cost_per_successful_request
.as_ref()
.map(ToString::to_string),
coverage: cost_coverage_view(&value.coverage),
}
}
pub(crate) fn granularity_name(granularity: domain::Granularity) -> &'static str {
match granularity {
domain::Granularity::FifteenMinutes => "15m",
@@ -1322,3 +1404,37 @@ pub(crate) fn analytics_view(
histogram_version: metadata.histogram_version,
}
}
fn overview_health_view(
health: &domain::UsageInsightsHealth,
time: crate::time::TimePresenter,
) -> OverviewHealthView {
let health_points = health
.points
.iter()
.map(|point| OverviewHealthPointView {
bucket: point.bucket_start,
label: time.label(point.bucket_start, "%m-%d %H:%M"),
total_requests: point.total_requests,
success_requests: point.success_requests,
failed_requests: point.failed_requests,
cancelled_requests: point.cancelled_requests,
incomplete_requests: point.incomplete_requests,
caller_error_requests: point.caller_error_requests,
error_rate: point.error_rate,
})
.collect();
OverviewHealthView {
total_requests: health.total_requests,
success_requests: health.success_requests,
failed_requests: health.failed_requests,
cancelled_requests: health.cancelled_requests,
incomplete_requests: health.incomplete_requests,
caller_error_requests: health.caller_error_requests,
success_rate: health.success_rate,
completion_rate: health.completion_rate,
request_change_rate: None,
success_rate_change: None,
points: health_points,
}
}
@@ -53,6 +53,71 @@ pub struct UsageAnalyticsQuery {
pub end_time: Option<String>,
}
/// 卡片内的指标选择与时间、Provider 范围
#[derive(Clone, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct UsageMetricQuery<T> {
pub metric: T,
pub provider: Option<String>,
pub period: Option<String>,
pub as_of: Option<i64>,
pub start_time: Option<String>,
pub end_time: Option<String>,
}
impl<T> UsageMetricQuery<T> {
pub fn into_parts(self) -> (UsageAnalyticsQuery, T) {
(
UsageAnalyticsQuery {
provider: self.provider,
period: self.period,
as_of: self.as_of,
start_time: self.start_time,
end_time: self.end_time,
},
self.metric,
)
}
}
#[derive(Clone, Copy, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum UsagePerformanceMetricKind {
Latency,
FirstToken,
Throughput,
Load,
}
impl From<UsagePerformanceMetricKind> for domain::UsageInsightMetric {
fn from(value: UsagePerformanceMetricKind) -> Self {
match value {
UsagePerformanceMetricKind::Latency => Self::Latency,
UsagePerformanceMetricKind::FirstToken => Self::FirstToken,
UsagePerformanceMetricKind::Throughput => Self::Throughput,
UsagePerformanceMetricKind::Load => Self::Load,
}
}
}
#[derive(Clone, Copy, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum UsageCostMetricKind {
Cost,
Savings,
Cache,
}
impl From<UsageCostMetricKind> for domain::UsageInsightMetric {
fn from(value: UsageCostMetricKind) -> Self {
match value {
UsageCostMetricKind::Cost => Self::Cost,
UsageCostMetricKind::Savings => Self::Savings,
UsageCostMetricKind::Cache => Self::Cache,
}
}
}
impl UsageQuery {
/// 校验 Element Plus 风格的页码分页字段
pub fn validate_pagination(&self) -> Result<(u32, u16), WireValidationError> {
@@ -24,6 +24,12 @@ where
"/api/admin/usage/insights/overview",
get(usage_insights_overview::<S>),
)
.route("/api/admin/usage/insights/health", get(usage_health::<S>))
.route(
"/api/admin/usage/insights/performance",
get(usage_performance::<S>),
)
.route("/api/admin/usage/insights/cost", get(usage_cost::<S>))
.route(
"/api/admin/usage/insights/diagnostics",
get(usage_insights_diagnostics::<S>),
@@ -207,6 +213,74 @@ where
))
}
pub(crate) async fn usage_health<S>(
_auth: AdminAuth,
State(state): State<S>,
AdminQuery(query): AdminQuery<UsageAnalyticsQuery>,
) -> Result<impl IntoResponse, AdminError>
where
S: SessionState + Send + Sync,
{
usage_metric_response(state, query, domain::UsageInsightMetric::Health).await
}
pub(crate) async fn usage_performance<S>(
_auth: AdminAuth,
State(state): State<S>,
AdminQuery(query): AdminQuery<UsageMetricQuery<UsagePerformanceMetricKind>>,
) -> Result<impl IntoResponse, AdminError>
where
S: SessionState + Send + Sync,
{
let (range, metric) = query.into_parts();
usage_metric_response(state, range, metric.into()).await
}
pub(crate) async fn usage_cost<S>(
_auth: AdminAuth,
State(state): State<S>,
AdminQuery(query): AdminQuery<UsageMetricQuery<UsageCostMetricKind>>,
) -> Result<impl IntoResponse, AdminError>
where
S: SessionState + Send + Sync,
{
let (range, metric) = query.into_parts();
usage_metric_response(state, range, metric.into()).await
}
async fn usage_metric_response<S>(
state: S,
query: UsageAnalyticsQuery,
metric: domain::UsageInsightMetric,
) -> Result<impl IntoResponse, AdminError>
where
S: SessionState + Send + Sync,
{
let time = crate::time::TimePresenter::new(state.admin_services().timezone());
let range = usage_range(
query.start_time.as_deref(),
query.end_time.as_deref(),
query.period.as_deref(),
query.as_of,
state.admin_services().timezone(),
)
.map_err(map_wire_error)?;
let filter = domain::UsageFilter {
provider_kind: non_empty(query.provider),
..Default::default()
};
let result = state
.admin_services()
.observability()
.usage_metric(range, filter, metric)
.await
.map_err(map_service_error)?;
Ok(AdminResponse::new(
StatusCode::OK,
AdminEnvelope::ok(usage_metric_view(result, time)),
))
}
pub(crate) async fn usage_insights_diagnostics<S>(
_auth: AdminAuth,
State(state): State<S>,
@@ -759,6 +759,86 @@ pub struct UsageInsightsOverviewView {
pub providers: Vec<ProviderOverviewView>,
}
/// 单项洞察响应
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageMetricView {
pub analytics: AnalyticsView,
pub granularity: String,
#[serde(flatten)]
pub data: UsageMetricDataView,
}
/// 指标标签与响应字段一一对应,未查询的数据不会出现在响应中
#[derive(Debug, Serialize)]
#[serde(tag = "metric", content = "data", rename_all = "camelCase")]
pub enum UsageMetricDataView {
Health(OverviewHealthView),
Latency(UsageMetricSeriesView<UsageLatencyView>),
FirstToken(UsageMetricSeriesView<UsageLatencyView>),
Throughput(UsageMetricSeriesView<UsageThroughputView>),
Load(UsageMetricSeriesView<OverviewLoadView>),
Cost(UsageMetricSeriesView<UsageCostSummaryView, UsageCostComparisonView>),
Savings(UsageMetricSeriesView<UsageCostSummaryView, Option<String>>),
Cache(UsageMetricSeriesView<UsageCostSummaryView, UsageCacheView>),
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageMetricSeriesView<S, P = S> {
pub summary: S,
pub points: Vec<UsageMetricPointView<P>>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageMetricPointView<P> {
pub bucket: DateTime<Utc>,
pub label: String,
pub request_count: u64,
pub value: P,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageLatencyView {
pub p50_ms: Option<f64>,
pub p95_ms: Option<f64>,
pub p99_ms: Option<f64>,
}
#[derive(Debug, Serialize)]
pub struct UsageThroughputView {
pub p10: Option<u64>,
pub p50: Option<u64>,
pub p90: Option<u64>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageCostSummaryView {
pub estimated_cost: Option<String>,
pub cache_savings: Option<String>,
pub tier_premium: Option<String>,
pub cost_per_successful_request: Option<String>,
pub coverage: CostCoverageView,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageCostComparisonView {
pub estimated_cost: Option<String>,
pub no_cache_cost: Option<String>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageCacheView {
pub input_tokens: u64,
pub cached_token_rate: f64,
pub cache_hit_request_rate: Option<f64>,
}
/// 诊断聚合项目
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
+24 -1
View File
@@ -693,6 +693,7 @@ pub(super) struct MemoryObservations {
pub trend: Option<Vec<RequestMetricPoint>>,
pub concurrency: Vec<gateway_admin::model::observability::UsageConcurrencyPoint>,
pub summaries: Vec<(TimeRange, UsageFilter)>,
pub concurrency_selections: Vec<bool>,
pub trends: Vec<(TimeRange, UsageFilter)>,
pub records: Vec<UsageQuery>,
pub errors: Vec<OpsErrorQuery>,
@@ -1491,8 +1492,10 @@ impl ObservabilityStore for UnusedStore {
range: TimeRange,
filter: UsageFilter,
_: Granularity,
include_concurrency: bool,
) -> AdminStoreResult<gateway_admin::model::observability::UsageInsightsObservation> {
let mut data = self.observations.lock().expect("observations");
data.concurrency_selections.push(include_concurrency);
data.summaries.push((range, filter.clone()));
data.trends.push((range, filter));
Ok(
@@ -1507,11 +1510,31 @@ impl ObservabilityStore for UnusedStore {
.clone()
.ok_or_else(|| unavailable("usage trend"))?,
costs: Vec::new(),
concurrency: data.concurrency.clone(),
concurrency: if include_concurrency {
data.concurrency.clone()
} else {
Vec::new()
},
},
)
}
async fn usage_metric_observation(
&self,
range: gateway_admin::model::observability::TimeRange,
filter: gateway_admin::model::observability::UsageFilter,
granularity: gateway_admin::model::observability::Granularity,
metric: gateway_admin::model::observability::UsageInsightMetric,
) -> AdminStoreResult<gateway_admin::model::observability::UsageInsightsObservation> {
self.usage_insights_observation(
range,
filter,
granularity,
metric == gateway_admin::model::observability::UsageInsightMetric::Load,
)
.await
}
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage> {
self.observations
.lock()
@@ -1382,6 +1382,157 @@ async fn usage_insights_maps_load_summary_and_points_to_camel_case() {
);
}
#[tokio::test]
async fn usage_card_routes_return_only_the_selected_metric_and_validate_options() {
use crate::admin::{AdminTestFixture, AdminTestState};
use axum::{
body::{Body, to_bytes},
http::{Request, StatusCode, header},
};
use chrono::Duration;
use gateway_admin::model::observability::{
AttemptMetrics, CurrencyCost, 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 metrics = RequestMetrics {
request_count: 10,
success_count: 8,
failure_count: 2,
latency_percentiles: LatencyPercentiles {
p95_ms: Some(PercentileMilliseconds::new(43.5).unwrap()),
..Default::default()
},
first_token_latency_percentiles: LatencyPercentiles {
p95_ms: Some(PercentileMilliseconds::new(17.25).unwrap()),
..Default::default()
},
output_throughput_p50: Some(22),
..Default::default()
};
{
let mut observations = fixture.observations.lock().unwrap();
observations.summary = Some(UsageOverview {
analytics: Default::default(),
range,
requests: metrics.clone(),
attempts: AttemptMetrics {
costs: vec![CurrencyCost {
currency: "USD".into(),
amount: "12.5".parse().unwrap(),
}],
..Default::default()
},
providers: Vec::new(),
});
observations.trend = Some(vec![RequestMetricPoint {
bucket_start: range.start,
granularity: Granularity::FifteenMinutes,
metrics,
cost_coverage: Default::default(),
costs: Vec::new(),
}]);
observations.concurrency = vec![UsageConcurrencyPoint {
bucket_start: range.start,
average_concurrency: 1.25,
peak_concurrency: 3,
}];
}
let router = observability::router::<AdminTestState>().with_state(fixture.state());
for (path, metric, field, expected) in [
("health", "health", "totalRequests", json!(10)),
(
"performance?metric=latency",
"latency",
"p95Ms",
json!(43.5),
),
(
"performance?metric=firstToken",
"firstToken",
"p95Ms",
json!(17.25),
),
(
"performance?metric=throughput",
"throughput",
"p50",
json!(22),
),
(
"performance?metric=load",
"load",
"peakConcurrency",
json!(3),
),
("cost?metric=cost", "cost", "estimatedCost", json!("12.5")),
(
"cost?metric=savings",
"savings",
"estimatedCost",
json!("12.5"),
),
("cost?metric=cache", "cache", "estimatedCost", json!("12.5")),
] {
let response = router
.clone()
.oneshot(
Request::builder()
.uri(format!("/api/admin/usage/insights/{path}"))
.header(header::COOKIE, "cpr_session=valid-session")
.header("x-request-id", "req_usage_card_metric")
.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, "{path}: {value}");
let result = &value["data"];
assert_eq!(result["metric"], metric);
assert_eq!(result.as_object().unwrap().len(), 4);
let data = &result["data"];
if metric == "health" {
assert_eq!(data[field], expected);
} else {
assert_eq!(data["summary"][field], expected, "{path}");
assert_eq!(data["points"][0]["requestCount"], 10);
assert_eq!(data.as_object().unwrap().len(), 2);
}
}
for path in [
"performance",
"cost",
"performance?metric=cost",
"cost?metric=load",
"health?metric=load",
"performance?metric=unknown",
"cost?metric=cache&status=error",
] {
let response = router
.clone()
.oneshot(
Request::builder()
.uri(format!("/api/admin/usage/insights/{path}"))
.header(header::COOKIE, "cpr_session=valid-session")
.header("x-request-id", "req_usage_card_metric")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{path}");
}
}
#[tokio::test]
async fn observability_queries_reject_group_filters_and_analytics_search() {
use crate::admin::{AdminTestFixture, AdminTestState};
@@ -218,6 +218,7 @@ async fn overview_scopes_every_query_and_projects_only_key_visible_fields() {
assert!(!data.to_string().contains("private-sentinel"));
let observations = fixture.observations.lock().unwrap();
assert_eq!(observations.summaries.len(), 1);
assert_eq!(observations.concurrency_selections, vec![false]);
assert_eq!(observations.trends.len(), 2);
for (_, filter) in observations
.summaries
@@ -287,6 +287,7 @@ impl PgObservabilityRepository {
range: ObservabilityRange,
filter: UsageRecordFilter,
granularity: admin_observability::Granularity,
include_concurrency: bool,
) -> StoreResult<admin_observability::UsageInsightsObservation> {
self.query_budget
.run(
@@ -295,7 +296,13 @@ impl PgObservabilityRepository {
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let (mut overview, trend, costs, concurrency) = snapshot
.insights(range, &filter, granularity, self.timezone)
.insights(
range,
&filter,
granularity,
self.timezone,
include_concurrency,
)
.await?;
overview.analytics = snapshot.metadata();
snapshot.finish().await?;
@@ -310,6 +317,30 @@ impl PgObservabilityRepository {
.await
}
pub async fn usage_metric_observation(
&self,
range: ObservabilityRange,
filter: UsageRecordFilter,
granularity: admin_observability::Granularity,
metric: admin_observability::UsageInsightMetric,
) -> StoreResult<admin_observability::UsageInsightsObservation> {
self.query_budget
.run(
&self.pool,
"load usage metric snapshot",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let mut observation = snapshot
.metric(range, &filter, granularity, self.timezone, metric)
.await?;
observation.overview.analytics = snapshot.metadata();
snapshot.finish().await?;
Ok(observation)
},
)
.await
}
pub async fn provider_account_usage(
&self,
query: ProviderAccountUsageQuery,
@@ -498,12 +529,32 @@ impl AdminObservabilityStore for PgAdminObservabilityStore {
range: admin_observability::TimeRange,
filter: admin_observability::UsageFilter,
granularity: admin_observability::Granularity,
include_concurrency: bool,
) -> AdminStoreResult<admin_observability::UsageInsightsObservation> {
self.repository
.usage_insights_observation(
store_range(range).map_err(observability_error)?,
store_usage_filter(filter),
granularity,
include_concurrency,
)
.await
.map_err(observability_error)
}
async fn usage_metric_observation(
&self,
range: admin_observability::TimeRange,
filter: admin_observability::UsageFilter,
granularity: admin_observability::Granularity,
metric: admin_observability::UsageInsightMetric,
) -> AdminStoreResult<admin_observability::UsageInsightsObservation> {
self.repository
.usage_metric_observation(
store_range(range).map_err(observability_error)?,
store_usage_filter(filter),
granularity,
metric,
)
.await
.map_err(observability_error)
@@ -335,6 +335,7 @@ impl<'a> AnalyticsSnapshot<'a> {
filter: &UsageRecordFilter,
granularity: ObservationGranularity,
timezone: gateway_core::time::DeploymentTimeZone,
include_concurrency: bool,
) -> StoreResult<(
UsageOverview,
Vec<RequestMetricPoint>,
@@ -345,19 +346,13 @@ impl<'a> AnalyticsSnapshot<'a> {
.load(range, filter, None, Some(granularity), timezone, true)
.await?;
let trend = metric_points(range, granularity, timezone, &states, true)?;
let costs = states
.iter()
.map(|((bucket_start, _), state)| {
Ok(admin_observability::UsageProjectedCostPoint {
bucket_start: *bucket_start,
standard_amount: state.projected_cost("standard_usd")?,
no_cache_amount: state.projected_cost("no_cache_usd")?,
})
})
.collect::<StoreResult<Vec<_>>>()?;
let concurrency = self
.concurrency(range, filter, granularity, timezone)
.await?;
let costs = projected_cost_points(&states)?;
let concurrency = if include_concurrency {
self.concurrency(range, filter, granularity, timezone)
.await?
} else {
Vec::new()
};
// 趋势已包含可合并的完整指标和情景费用,概览直接复用同一份状态
// 精确分位数不可跨桶合并,明细回退时仍按完整区间单独计算概览
let overview = if states.values().any(MetricState::has_exact_percentiles) {
@@ -369,6 +364,107 @@ impl<'a> AnalyticsSnapshot<'a> {
Ok((overview, trend, costs, concurrency))
}
pub async fn metric(
&mut self,
range: ObservabilityRange,
filter: &UsageRecordFilter,
granularity: ObservationGranularity,
timezone: gateway_core::time::DeploymentTimeZone,
metric: admin_observability::UsageInsightMetric,
) -> StoreResult<admin_observability::UsageInsightsObservation> {
use admin_observability::UsageInsightMetric;
let distribution_names: &[&str] = match metric {
UsageInsightMetric::Latency => &["latency_ms"],
UsageInsightMetric::FirstToken => &["first_token_ms"],
UsageInsightMetric::Throughput => &["throughput"],
UsageInsightMetric::Load => &["admission_decision_ms", "account_selection_wait_ms"],
_ => &[],
};
let costs = matches!(
metric,
UsageInsightMetric::Cost | UsageInsightMetric::Savings | UsageInsightMetric::Cache
);
let counters: &[&str] = match metric {
UsageInsightMetric::Health => &[
"request_count",
"success_count",
"failure_count",
"cancelled_count",
"incomplete_count",
"caller_error_count",
],
// 成本视图均显示费用、节省、溢价、单次成功费用与覆盖率汇总
UsageInsightMetric::Cost | UsageInsightMetric::Savings => &[
"request_count",
"success_count",
"provider_reported_count",
"calculated_count",
"unavailable_count",
],
UsageInsightMetric::Cache => &[
"request_count",
"success_count",
"input_tokens",
"cached_input_tokens",
"cache_hit_request_count",
"cache_eligible_request_count",
"provider_reported_count",
"calculated_count",
"unavailable_count",
],
_ => &["request_count"],
};
let selection = ReadSelection {
distributions: if distribution_names.is_empty() {
DistributionRead::None
} else {
DistributionRead::Full
},
costs,
keys: None,
counters: Some(counters),
distribution_names: Some(distribution_names),
};
let states = self
.load_selected(range, filter, None, Some(granularity), timezone, selection)
.await?;
let trend = metric_points(range, granularity, timezone, &states, costs)?;
let projected_costs = if costs {
projected_cost_points(&states)?
} else {
Vec::new()
};
// 明细筛选的精确分位数仍需按完整范围读取,不能合并各桶分位数
let total = if states.values().any(MetricState::has_exact_percentiles) {
merge_states(
self.load_selected(range, filter, None, None, timezone, selection)
.await?
.into_values(),
)?
} else {
merge_states(states.into_values())?
};
let concurrency = if metric == UsageInsightMetric::Load {
self.concurrency(range, filter, granularity, timezone)
.await?
} else {
Vec::new()
};
let overview = UsageOverview {
range,
requests: total.request_metrics()?,
attempts: total.attempts()?,
providers: Vec::new(),
analytics: self.metadata(),
};
Ok(admin_observability::UsageInsightsObservation {
overview: admin_usage_overview(overview),
trend: trend.into_iter().map(admin_request_metric_point).collect(),
costs: projected_costs,
concurrency,
})
}
async fn concurrency(
&mut self,
range: ObservabilityRange,
@@ -1072,6 +1168,21 @@ pub(super) fn push_summary_buckets(
}
}
fn projected_cost_points(
states: &BTreeMap<(DateTime<Utc>, String), MetricState>,
) -> StoreResult<Vec<admin_observability::UsageProjectedCostPoint>> {
states
.iter()
.map(|((bucket_start, _), state)| {
Ok(admin_observability::UsageProjectedCostPoint {
bucket_start: *bucket_start,
standard_amount: state.projected_cost("standard_usd")?,
no_cache_amount: state.projected_cost("no_cache_usd")?,
})
})
.collect()
}
fn metric_points(
range: ObservabilityRange,
granularity: ObservationGranularity,
@@ -144,6 +144,7 @@ async fn usage_summary_preserves_counts_latency_samples_and_costs_across_read_pa
range,
filter,
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.unwrap()
@@ -233,6 +234,7 @@ async fn usage_summary_preserves_counts_latency_samples_and_costs_across_read_pa
full,
exact_filter,
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.unwrap();
@@ -302,6 +304,7 @@ async fn provider_counts_match_for_rollups_boundaries_and_filtered_facts() {
range,
filter.clone(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -411,6 +414,7 @@ async fn insights_share_bucket_states_without_merging_exact_percentiles() {
range,
filter.clone(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -431,7 +435,7 @@ async fn insights_share_bucket_states_without_merging_exact_percentiles() {
gateway_admin::model::observability::Granularity::Hour,
] {
let observation = repo
.usage_insights_observation(range, filter.clone(), granularity)
.usage_insights_observation(range, filter.clone(), granularity, true)
.await
.unwrap();
assert_eq!(observation.overview.requests, expected.requests);
@@ -1110,6 +1114,7 @@ async fn persisted_histograms_merge_long_tails_and_distinguish_empty_from_zero()
range,
Default::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -1122,6 +1127,7 @@ async fn persisted_histograms_merge_long_tails_and_distinguish_empty_from_zero()
..Default::default()
},
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -1156,6 +1162,7 @@ async fn persisted_histograms_merge_long_tails_and_distinguish_empty_from_zero()
ObservabilityRange::new(start, start + TimeDelta::minutes(10)).unwrap(),
Default::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -1170,6 +1177,7 @@ async fn persisted_histograms_merge_long_tails_and_distinguish_empty_from_zero()
.unwrap(),
Default::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -1775,9 +1783,37 @@ async fn insights_concurrency_counts_logical_request_overlap_once_per_bucket() {
..Default::default()
},
granularity,
true,
)
.await
.unwrap();
let without_concurrency = 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,
false,
)
.await
.unwrap();
assert!(without_concurrency.concurrency.is_empty());
assert_eq!(
without_concurrency.overview.requests,
insights.overview.requests
);
assert_eq!(
without_concurrency.overview.attempts,
insights.overview.attempts
);
assert_eq!(without_concurrency.trend, insights.trend);
assert_eq!(without_concurrency.costs, insights.costs);
assert_eq!(insights.concurrency.len(), expected.len());
for (point, (average, peak)) in insights.concurrency.iter().zip(expected) {
assert!(
@@ -1823,6 +1859,7 @@ async fn insights_concurrency_weights_calendar_days_across_dst() {
..Default::default()
},
gateway_admin::model::observability::Granularity::Day,
true,
)
.await
.unwrap();
@@ -1874,6 +1911,7 @@ async fn sql_edge_histograms_preserve_repeated_samples_and_power_of_two_boundari
range,
Default::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.unwrap()
@@ -1885,6 +1923,7 @@ async fn sql_edge_histograms_preserve_repeated_samples_and_power_of_two_boundari
range,
Default::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.unwrap()
@@ -2036,6 +2036,7 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates
zero_attempt_range(&request),
UsageRecordFilter::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -2055,6 +2056,7 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates
zero_attempt_range(&request),
UsageRecordFilter::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -2083,6 +2085,7 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates
..Default::default()
},
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -2112,6 +2115,7 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates
zero_attempt_range(&request),
filter.clone(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -2142,6 +2146,7 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates
zero_attempt_range(&request),
filter,
admin_observability::Granularity::FifteenMinutes,
true,
)
.await
.expect("billing projection");
@@ -0,0 +1,164 @@
//! 单项洞察保留统计口径,未选择的分布和并发不进入结果
use super::*;
use admin_observability::{UsageInsightMetric as Metric, UsageInsightsObservation};
#[tokio::test]
async fn usage_metrics_preserve_selected_values_for_aggregate_and_exact_reads() {
let Some(database) = TestDatabase::create("usage_metrics").await else {
return;
};
let now = Utc::now();
seed_observability_facts(&database.pool, now).await.unwrap();
seed_calculated_billing_facts(&database.pool, now)
.await
.unwrap();
super::super::drain_analytics(&database.pool).await;
let store = admin_observability_store(&database.pool);
let range =
admin_observability::TimeRange::new(now - TimeDelta::hours(2), now + TimeDelta::hours(1))
.unwrap();
let granularity = admin_observability::Granularity::FifteenMinutes;
for filter in [
admin_observability::UsageFilter::default(),
admin_observability::UsageFilter {
model: Some("public-model".to_owned()),
..Default::default()
},
] {
let full = store
.usage_insights_observation(range, filter.clone(), granularity, true)
.await
.unwrap();
assert!(full.overview.requests.request_count > 0);
for metric in [
Metric::Health,
Metric::Latency,
Metric::FirstToken,
Metric::Throughput,
Metric::Load,
Metric::Cost,
Metric::Savings,
Metric::Cache,
] {
let selected = store
.usage_metric_observation(range, filter.clone(), granularity, metric)
.await
.unwrap();
assert_eq!(
selected.overview.requests.request_count, full.overview.requests.request_count,
"{metric:?}"
);
assert_eq!(selected.trend.len(), full.trend.len());
assert_selected_metrics(&selected.overview.requests, &full.overview.requests, metric);
for (actual, expected) in selected.trend.iter().zip(&full.trend) {
assert_eq!(actual.bucket_start, expected.bucket_start);
assert_eq!(actual.metrics.request_count, expected.metrics.request_count);
assert_selected_metrics(&actual.metrics, &expected.metrics, metric);
}
assert_costs(&selected, &full, metric);
if metric == Metric::Load {
assert_eq!(selected.concurrency, full.concurrency);
} else {
assert!(selected.concurrency.is_empty(), "{metric:?}");
}
}
}
database.close().await;
}
fn assert_selected_metrics(
actual: &admin_observability::RequestMetrics,
expected: &admin_observability::RequestMetrics,
metric: Metric,
) {
match metric {
Metric::Health => {
assert_eq!(actual.success_count, expected.success_count);
assert_eq!(actual.failure_count, expected.failure_count);
assert_eq!(actual.cancelled_count, expected.cancelled_count);
assert_eq!(actual.incomplete_count, expected.incomplete_count);
assert_eq!(actual.caller_error_count, expected.caller_error_count);
}
Metric::Latency => assert_eq!(actual.latency_percentiles, expected.latency_percentiles),
Metric::FirstToken => assert_eq!(
actual.first_token_latency_percentiles,
expected.first_token_latency_percentiles
),
Metric::Throughput => assert_eq!(
(
actual.output_throughput_p10,
actual.output_throughput_p50,
actual.output_throughput_p90
),
(
expected.output_throughput_p10,
expected.output_throughput_p50,
expected.output_throughput_p90
)
),
Metric::Load => {
assert_eq!(
actual.admission_decision_percentiles,
expected.admission_decision_percentiles
);
assert_eq!(
actual.account_selection_wait_percentiles,
expected.account_selection_wait_percentiles
);
}
Metric::Cost | Metric::Savings => {
assert_eq!(actual.success_count, expected.success_count);
}
Metric::Cache => {
assert_eq!(actual.success_count, expected.success_count);
assert_eq!(actual.input_tokens, expected.input_tokens);
assert_eq!(actual.cached_input_tokens, expected.cached_input_tokens);
assert_eq!(
actual.cache_hit_request_count,
expected.cache_hit_request_count
);
assert_eq!(
actual.cache_eligible_request_count,
expected.cache_eligible_request_count
);
}
}
if metric != Metric::Latency {
assert_eq!(actual.latency_percentiles, Default::default());
}
if metric != Metric::FirstToken {
assert_eq!(actual.first_token_latency_percentiles, Default::default());
}
if metric != Metric::Throughput {
assert_eq!(actual.output_throughput_p50, None);
}
if metric != Metric::Load {
assert_eq!(actual.admission_decision_percentiles, Default::default());
}
}
fn assert_costs(
actual: &UsageInsightsObservation,
expected: &UsageInsightsObservation,
metric: Metric,
) {
if matches!(metric, Metric::Cost | Metric::Savings | Metric::Cache) {
assert_eq!(
actual.overview.attempts.costs,
expected.overview.attempts.costs
);
assert_eq!(
actual.overview.attempts.cost_coverage,
expected.overview.attempts.cost_coverage
);
assert_eq!(actual.costs, expected.costs);
for (actual, expected) in actual.trend.iter().zip(&expected.trend) {
assert_eq!(actual.costs, expected.costs);
assert_eq!(actual.cost_coverage, expected.cost_coverage);
}
} else {
assert!(actual.overview.attempts.costs.is_empty());
assert!(actual.costs.is_empty());
}
}
@@ -1,5 +1,6 @@
//! 观测存储测试入口,以及查询范围、过滤与端口合同测试
mod metric;
mod projection;
use std::{
@@ -157,6 +158,7 @@ async fn output_throughput_uses_official_response_duration_without_local_fallbac
range,
UsageRecordFilter::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -1172,6 +1174,7 @@ async fn recovered_continuation_failure_should_be_visible_in_ops_but_hidden_from
range,
UsageRecordFilter::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -1538,6 +1541,7 @@ async fn calculated_usage_billing_facts_keep_only_completed_calculated_costs() {
range,
UsageRecordFilter::default(),
admin_observability::Granularity::FifteenMinutes,
true,
)
.await
.expect("projected costs");
@@ -1556,6 +1560,7 @@ async fn calculated_usage_billing_facts_keep_only_completed_calculated_costs() {
admin_observability::TimeRange::new(range.start, range.end).unwrap(),
admin_observability::UsageFilter::default(),
admin_observability::Granularity::FifteenMinutes,
true,
)
.await
.expect("admin projected costs");
@@ -1844,6 +1849,7 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det
range,
admin_observability::UsageFilter::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -2332,6 +2338,7 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact
range,
UsageRecordFilter::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -2359,6 +2366,7 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact
..UsageRecordFilter::default()
},
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
@@ -2694,6 +2702,7 @@ async fn calendar_trends_share_exact_day_boundaries_for_requests_costs_and_empty
range,
Default::default(),
admin_observability::Granularity::Day,
true,
)
.await
.expect("calendar cost buckets");
@@ -3004,7 +3013,7 @@ async fn observability_store_executes_caller_selection_without_replacing_policy(
.unwrap();
super::drain_analytics(&database.pool).await;
let facts = store
.usage_insights_observation(range, UsageFilter::default(), granularity)
.usage_insights_observation(range, UsageFilter::default(), granularity, true)
.await
.unwrap();
let facts = facts
@@ -534,6 +534,7 @@ async fn unassigned_correction_changes_totals_without_inventing_dimension_member
range,
NativeFilter::default(),
gateway_admin::model::observability::Granularity::Hour,
true,
)
.await
.map(|snapshot| snapshot.overview)
+20 -3
View File
@@ -2164,7 +2164,10 @@ errorCode, errorMessage, startedAt, completedAt, expiresAt, createdAt, updatedAt
| `GET` | `/api/admin/usage/records` | 请求记录分页列表 |
| `GET` | `/api/admin/usage/records/detail` | 按 `id` 查询请求详情 |
| `GET` | `/api/admin/usage/records/summary` | 当前筛选条件的请求汇总 |
| `GET` | `/api/admin/usage/insights/overview` | 用量、成本与成功率洞察 |
| `GET` | `/api/admin/usage/insights/overview` | 完整用量洞察 |
| `GET` | `/api/admin/usage/insights/health` | 请求健康 |
| `GET` | `/api/admin/usage/insights/performance` | 响应性能;`metric=latency\|firstToken\|throughput\|load` |
| `GET` | `/api/admin/usage/insights/cost` | 成本效率;`metric=cost\|savings\|cache` |
| `GET` | `/api/admin/usage/insights/diagnostics` | 按维度聚合诊断 |
| `GET` | `/api/admin/operations/errors` | 运维错误分页列表 |
| `GET` | `/api/admin/diagnostics/export` | 按 `requestId` 或 `errorId` 下载诊断 ZIP 附件 |
@@ -2186,7 +2189,7 @@ errorCode, errorMessage, startedAt, completedAt, expiresAt, createdAt, updatedAt
`points[]` 提供趋势图与首页卡片使用的请求数、Tokens、缓存比例、错误数、成功率、
`latencyValue`、首 Token P95、总延迟 P95 和输出吞吐 P50,以及时间标签与展示值
请求并发与内部调度等待通过用量洞察的 `performance.load` 查询
请求并发与内部调度等待通过 `/api/admin/usage/insights/performance?metric=load` 查询
### Dashboard 客户端身份
@@ -2223,7 +2226,7 @@ Dashboard 默认 `period=today`,用量、诊断和错误查询默认 `period=7
Dashboard 返回原始 `asOf` 及 `asOfDisplay`,作为当前数据的查询锚点
请求明细列表接受 `currentPage`、`pageSize`、时间范围、`provider`、`clientApiKeyName`、`accountName`、`model` 和 `status`。
汇总和洞察只接受时间范围与 `provider`;诊断另接受 `dimension`、`model` 和 `statusCode`。
汇总和洞察接受时间范围与 `provider`;性能和成本另要求 `metric`,诊断另接受 `dimension`、`model` 和 `statusCode`。
各接口拒绝未知参数,明细筛选不改变顶部统计范围
请求列表的 `status` 可取 `success`、`error`、`all`,默认 `success`;未知值返回 `400`。
@@ -2268,6 +2271,20 @@ Dashboard 返回原始 `asOf` 及 `asOfDisplay`,作为当前数据的查询锚
运维错误列表的 `search` 使用字面量包含匹配,`%`、`_` 和 `\` 按普通字符处理,不支持密钥值或可见前缀搜索。
请求详情通过 `/api/admin/usage/records/detail?id=...` 查询
### 卡片洞察查询
请求健康、响应性能、成本效率与热点诊断分别查询自己的路由;性能和成本按选中的 `metric` 查询,诊断按 `dimension` 查询。
各卡片独立加载和显示错误,切换选项取消该卡片的旧请求;时间范围与平台筛选共同作用于所有卡片。
`overview` 返回完整洞察,管理端卡片使用分项接口
健康、性能与成本接口返回 `analytics`、`granularity`、`metric` 和 `data`。
健康的 `metric` 固定为 `health`,`data` 含健康摘要和趋势;性能与成本的 `data` 含 `summary` 和 `points`,
每个点包含 `bucket`、`label`、`requestCount` 和所选指标的 `value`。缺失样本保留 `null`,不填充未查询指标
性能摘要与趋势分别返回总耗时分位数、首 Token 分位数、吞吐分位数或负载指标;只有 `metric=load` 读取请求并发。
成本的三个选项共用费用、节省、溢价、每次成功请求费用和覆盖率摘要,趋势分别返回实际与无缓存估算费用、缓存节省或缓存比例。
缺少 `metric`、选择其他卡片的指标或传入未知指标均返回 `400`
### 记录范围与统计口径
OpenAI 推理强度优先取 `input` 中最后一条 `type: "configuration_update"` 项的 `reasoning.effort`,
+2
View File
@@ -1010,6 +1010,8 @@ Store 从请求源事实维护窄事实及十五分钟、小时统计状态,
维护锁竞争造成的空轮不重置批量,未提交事务与未确认队列项由后续周期重试。
用量洞察的请求并发由 Store 在同一读取快照中按请求开始、完成时间聚合,运行中请求计至查询范围末尾。
Admin 按卡片与所选指标向 Store 请求计数、金额或分布,缓存按范围、筛选和指标隔离结果。
性能卡片只有负载选项读取请求并发;Key 用量与 Dashboard 趋势不请求并发计算,完整洞察的缓存仍按并发选择隔离。
每个时间桶返回按持续时间加权的平均并发和峰值,Admin 按各桶的有效时长合并全区间平均值,并取全区间峰值。
该趋势使用请求记录,不读取 Redis 准入租约,也不参与容量裁决;筛选与统计口径见[用量统计](api.md#记录范围与统计口径)
+58
View File
@@ -327,6 +327,37 @@ export interface UsageInsightsOverviewResponse {
cost: UsageOverviewCost
}
export type UsagePerformanceMetric = 'latency' | 'firstToken' | 'throughput' | 'load'
export type UsageCostMetric = 'cost' | 'savings' | 'cache'
export interface UsageMetricSeries<S, P = S> {
summary: S
points: { bucket: string, label: string, requestCount: number, value: P }[]
}
interface UsageLatencyPercentiles {
p50Ms: number | null
p95Ms: number | null
p99Ms: number | null
}
type UsageCostSummary = Pick<UsageOverviewCost, 'estimatedCost' | 'cacheSavings' | 'tierPremium' | 'costPerSuccessfulRequest' | 'coverage'>
interface UsageMetricData {
health: UsageOverviewHealth
latency: UsageMetricSeries<UsageLatencyPercentiles>
firstToken: UsageMetricSeries<UsageLatencyPercentiles>
throughput: UsageMetricSeries<{ p10: number | null, p50: number | null, p90: number | null }>
load: UsageMetricSeries<UsageOverviewLoad>
cost: UsageMetricSeries<UsageCostSummary, { estimatedCost: string | null, noCacheCost: string | null }>
savings: UsageMetricSeries<UsageCostSummary, string | null>
cache: UsageMetricSeries<UsageCostSummary, { inputTokens: number, cachedTokenRate: number, cacheHitRequestRate: number | null }>
}
export type UsageMetricResponse<M extends keyof UsageMetricData> = {
[K in M]: { analytics: AnalyticsMetadata, granularity: string, metric: K, data: UsageMetricData[K] }
}[M]
export interface UsageDiagnosticItem {
key: string
name: string
@@ -427,6 +458,33 @@ export function getUsageRecordInsightsDiagnostics(data: UsageDiagnosticsQuery, o
})
}
export function getUsageHealth(data: UsageRangeQuery, options: RequestOptions = {}) {
return request<UsageMetricResponse<'health'>>({
url: '/api/admin/usage/insights/health',
method: 'GET',
params: data,
...options,
})
}
export function getUsagePerformance<M extends UsagePerformanceMetric>(data: UsageRangeQuery & { metric: M }, options: RequestOptions = {}) {
return request<UsageMetricResponse<M>>({
url: '/api/admin/usage/insights/performance',
method: 'GET',
params: data,
...options,
})
}
export function getUsageCost<M extends UsageCostMetric>(data: UsageRangeQuery & { metric: M }, options: RequestOptions = {}) {
return request<UsageMetricResponse<M>>({
url: '/api/admin/usage/insights/cost',
method: 'GET',
params: data,
...options,
})
}
export function getDiagnosticsDownloadUrl(data: DiagnosticsDownloadParams) {
return `${API_BASE_URL}/api/admin/diagnostics/export?${new URLSearchParams(data)}`
}
@@ -1,14 +1,13 @@
<script setup lang="ts">
import type { EChartsOption, LineSeriesOption } from 'echarts'
import type { getUsageRecordInsightsOverview } from '@/api'
import type { UsageCostMetric, UsageMetricResponse } from '@/api'
import { ZCard, ZEmpty, ZSegmented } from '@codex-proxy/ui'
import { computed, shallowRef } from 'vue'
import { computed } from 'vue'
import { useI18n } from 'vue-i18n'
import BaseChart from '@/components/charts/BaseChart.vue'
import {
requestActivityByBucket,
zeroInactiveValues,
} from '@/components/charts/timeSeriesGap'
import { chartTooltipContent, tooltipIndex, tooltipRows } from '@/components/charts/tooltip'
@@ -27,8 +26,8 @@ import {
const props = withDefaults(
defineProps<{
cost: Cost
activity: Activity
result: UsageMetricResponse<UsageCostMetric> | null
error?: string
loading?: boolean
}>(),
{
@@ -38,22 +37,27 @@ const props = withDefaults(
const { t } = useI18n({ useScope: 'global' })
type Cost = Awaited<ReturnType<typeof getUsageRecordInsightsOverview>>['cost']
type Activity = Awaited<ReturnType<typeof getUsageRecordInsightsOverview>>['health']['points']
type UsageChartPalette = ReturnType<typeof useChartPalette>['palette']['value']
const activeView = shallowRef('cost')
const activeView = defineModel<UsageCostMetric>({ required: true })
const { palette } = useChartPalette()
const points = computed(() => props.cost.points)
const requestActivity = computed(() => requestActivityByBucket(
points.value.map(point => point.bucket),
props.activity,
))
const hasNoCacheCost = computed(() =>
points.value.some(point => point.noCacheCost != null),
)
const cost = computed(() => props.result?.data.summary ?? null)
const points = computed(() => props.result?.data.points ?? [])
const costPoints = computed(() => props.result?.metric === 'cost'
? props.result.data.points
: [])
const savingsPoints = computed(() => props.result?.metric === 'savings'
? props.result.data.points
: [])
const cachePoints = computed(() => props.result?.metric === 'cache'
? props.result.data.points
: [])
const requestActivity = computed(() => points.value.map(point => point.requestCount > 0))
const hasNoCacheCost = computed(() => costPoints.value.some(point => point.value.noCacheCost != null))
const coverageRate = computed(() => {
const { known, partial, unknown } = props.cost.coverage
if (!cost.value)
return null
const { known, partial, unknown } = cost.value.coverage
const total = known + partial + unknown
return total > 0
? (known + partial) / total
@@ -76,16 +80,13 @@ const viewOptions = computed(() => ([
]))
const hasData = computed(() => {
if (props.loading || points.value.length === 0)
if (props.loading || props.result?.metric !== activeView.value)
return false
if (activeView.value === 'cost') {
return points.value.some(point => point.estimatedCost != null || point.noCacheCost != null)
}
if (activeView.value === 'cost')
return costPoints.value.some(point => point.value.estimatedCost != null || point.value.noCacheCost != null)
if (activeView.value === 'savings')
return points.value.some(point => point.cacheSavings != null)
return points.value.some(
point => point.inputTokens > 0 || point.cachedTokenRate > 0 || (point.cacheHitRequestRate ?? 0) > 0,
)
return savingsPoints.value.some(point => point.value != null)
return cachePoints.value.some(({ value }) => value.inputTokens > 0 || value.cachedTokenRate > 0 || (value.cacheHitRequestRate ?? 0) > 0)
})
const chartOption = computed<EChartsOption>(() => {
@@ -152,13 +153,12 @@ function axisFormatter() {
}
function chartSeries(theme: UsageChartPalette): LineSeriesOption[] {
const chartPoints = points.value
if (activeView.value === 'savings') {
return [
...usageGapAwareLineSeries(
t('usage.cost.cacheSavings'),
zeroInactiveValues(
chartPoints.map(point => decimalDisplayNumber(point.cacheSavings)),
savingsPoints.value.map(point => decimalDisplayNumber(point.value)),
requestActivity.value,
),
theme.success,
@@ -171,12 +171,12 @@ function chartSeries(theme: UsageChartPalette): LineSeriesOption[] {
return [
...usageGapAwareLineSeries(
t('usage.cost.cacheTokenProportion'),
chartPoints.map(point => point.cachedTokenRate),
cachePoints.value.map(point => point.value.cachedTokenRate),
theme.normal,
),
...usageGapAwareLineSeries(
t('usage.cost.hitRequestRate'),
chartPoints.map(point => point.cacheHitRequestRate),
cachePoints.value.map(point => point.value.cacheHitRequestRate),
theme.success,
),
]
@@ -186,7 +186,7 @@ function chartSeries(theme: UsageChartPalette): LineSeriesOption[] {
...usageGapAwareLineSeries(
t('usage.cost.actualCost'),
zeroInactiveValues(
chartPoints.map(point => decimalDisplayNumber(point.estimatedCost)),
costPoints.value.map(point => decimalDisplayNumber(point.value.estimatedCost)),
requestActivity.value,
),
theme.success,
@@ -198,7 +198,7 @@ function chartSeries(theme: UsageChartPalette): LineSeriesOption[] {
...usageGapAwareLineSeries(
t('usage.cost.withoutCacheEstimate'),
zeroInactiveValues(
chartPoints.map(point => decimalDisplayNumber(point.noCacheCost)),
costPoints.value.map(point => decimalDisplayNumber(point.value.noCacheCost)),
requestActivity.value,
),
theme.textMuted,
@@ -220,7 +220,7 @@ function formatTooltip(params: unknown) {
return chartTooltipContent(point.label, [
usageTooltipItem(
t('usage.cost.cacheSavings'),
formatUsd(costAmount(point.cacheSavings, pointIndex)),
formatUsd(costAmount(savingsPoints.value[pointIndex]?.value ?? null, pointIndex)),
theme.success,
),
])
@@ -228,22 +228,22 @@ function formatTooltip(params: unknown) {
if (activeView.value === 'cache') {
return chartTooltipContent(point.label, [
usageTooltipItem(t('usage.cost.cacheTokenProportion'), formatPercent(point.cachedTokenRate), theme.normal),
usageTooltipItem(t('usage.cost.hitRequestRate'), formatPercent(point.cacheHitRequestRate), theme.success),
usageTooltipItem(t('usage.cost.cacheTokenProportion'), formatPercent(cachePoints.value[pointIndex]?.value.cachedTokenRate ?? null), theme.normal),
usageTooltipItem(t('usage.cost.hitRequestRate'), formatPercent(cachePoints.value[pointIndex]?.value.cacheHitRequestRate ?? null), theme.success),
])
}
const lines = [
usageTooltipItem(
t('usage.cost.actualCost'),
formatUsd(costAmount(point.estimatedCost, pointIndex)),
formatUsd(costAmount(costPoints.value[pointIndex]?.value.estimatedCost ?? null, pointIndex)),
theme.success,
),
]
if (point.noCacheCost != null) {
if (costPoints.value[pointIndex]?.value.noCacheCost != null) {
lines.push(usageTooltipItem(
t('usage.cost.withoutCacheEstimate'),
formatUsd(costAmount(point.noCacheCost, pointIndex)),
formatUsd(costAmount(costPoints.value[pointIndex]?.value.noCacheCost ?? null, pointIndex)),
theme.textMuted,
))
}
@@ -266,13 +266,13 @@ function costAmount(value: string | null, pointIndex: number) {
>
<template #actions>
<div class="flex justify-end">
<ZSegmented v-model="activeView" :aria-label="t('usage.cost.costView')" :options="viewOptions" :disabled="loading" />
<ZSegmented v-model="activeView" :aria-label="t('usage.cost.costView')" :options="viewOptions" />
</div>
</template>
<template #body>
<div class="grid min-h-66 gap-3">
<div v-if="hasData" class="grid grid-cols-2 gap-x-1 gap-y-3 rounded-xl bg-cp-fill-quaternary/45 p-2 @min-[24rem]/cost:grid-cols-3 @min-[40rem]/cost:grid-cols-5">
<div v-if="hasData && cost" class="grid grid-cols-2 gap-x-1 gap-y-3 rounded-xl bg-cp-fill-quaternary/45 p-2 @min-[24rem]/cost:grid-cols-3 @min-[40rem]/cost:grid-cols-5">
<div class="grid min-w-0 gap-1 px-1.5">
<span class="text-[10px] font-bold text-cp-text-quaternary" :title="t('usage.billing.totalCost')">{{ t('usage.cost.actualCost') }}</span>
<strong class="wrap-anywhere font-mono text-cp-sm font-heavy tabular-nums text-cp-text" :title="formatUsd(cost.estimatedCost)">
@@ -309,7 +309,7 @@ function costAmount(value: string | null, pointIndex: number) {
v-else
size="small"
surface="none"
:title="loading ? t('usage.cost.loading') : t('usage.cost.empty')"
:title="error || (loading ? t('usage.cost.loading') : t('usage.cost.empty'))"
class="h-52.5 place-content-center"
/>
</div>
@@ -12,7 +12,8 @@ import { formatLocalizedCompactNumber as formatCompactNumber, formatDuration, fo
const props = withDefaults(
defineProps<{
diagnostics: Diagnostics
diagnostics: Diagnostics | null
error?: string
loading?: boolean
}>(),
{
@@ -53,7 +54,7 @@ const dimensionOptions = computed(() => ([
},
]))
const resultDimension = computed(() => props.diagnostics.dimension || dimension.value)
const resultDimension = computed(() => props.diagnostics?.dimension || dimension.value)
const resultDimensionLabel = computed(
() => dimensionOptions.value.find(option => option.value === resultDimension.value)?.label ?? t('usage.diagnostics.dimensions'),
)
@@ -116,7 +117,7 @@ type DiagnosticDisplayItem = Diagnostics['items'][number] & {
}
const displayItems = computed<DiagnosticDisplayItem[]>(() =>
props.diagnostics.items.map(item => ({
(props.diagnostics?.items ?? []).map(item => ({
...item,
nameDisplay: diagnosticNameDisplay(item.name),
})),
@@ -155,7 +156,6 @@ function diagnosticNameDisplay(name: string) {
v-model="dimension"
:aria-label="t('usage.diagnostics.diagnosticDimension')"
:options="dimensionOptions"
:disabled="loading"
/>
</div>
</template>
@@ -263,7 +263,7 @@ function diagnosticNameDisplay(name: string) {
v-else
size="small"
surface="none"
:title="loading ? t('usage.diagnostics.loading') : t('usage.diagnostics.empty')"
:title="error || (loading ? t('usage.diagnostics.loading') : t('usage.diagnostics.empty'))"
class="h-full min-h-40 place-content-center"
/>
</template>
@@ -26,7 +26,8 @@ import {
const props = withDefaults(
defineProps<{
health: Health
health: Health | null
error?: string
granularity: string
loading?: boolean
}>(),
@@ -43,7 +44,7 @@ type HealthPoint = Health['points'][number]
use([BarChart])
const { palette } = useChartPalette()
const points = computed<HealthPoint[]>(() => props.health.points)
const points = computed<HealthPoint[]>(() => props.health?.points ?? [])
const hasData = computed(
() => !props.loading && points.value.some(point => requestCount(point) > 0),
@@ -186,7 +187,7 @@ function formatTooltip(params: unknown) {
>
<template #body>
<div class="grid min-h-66" :class="hasData ? 'gap-3' : 'h-full'">
<div v-if="hasData" class="grid grid-cols-3 gap-2 rounded-xl bg-cp-fill-quaternary/45 p-2">
<div v-if="hasData && health" class="grid grid-cols-3 gap-2 rounded-xl bg-cp-fill-quaternary/45 p-2">
<div class="grid gap-1 px-2">
<span class="text-[10px] font-bold text-cp-text-quaternary">{{ t('usage.health.serviceSuccessRate') }}</span>
<strong class="font-mono text-cp-base font-heavy tabular-nums text-cp-green-text">
@@ -211,7 +212,7 @@ function formatTooltip(params: unknown) {
v-else
size="small"
surface="none"
:title="loading ? t('usage.health.loading') : t('usage.health.empty')"
:title="error || (loading ? t('usage.health.loading') : t('usage.health.empty'))"
class="h-full place-content-center"
/>
</div>
@@ -1,33 +1,23 @@
<script setup lang="ts">
import type {
getUsageRecordInsightsDiagnostics,
getUsageRecordInsightsOverview,
} from '@/api'
import type { UsageQueryState } from '../composables/useUsageInsights'
import type { UsageCostMetric, UsageDiagnosticsResponse, UsageMetricResponse, UsagePerformanceMetric } from '@/api'
import { useI18n } from 'vue-i18n'
import UsageCostCard from './UsageCostCard.vue'
import UsageDiagnosticCard from './UsageDiagnosticCard.vue'
import UsageHealthCard from './UsageHealthCard.vue'
import UsagePerformanceCard from './UsagePerformanceCard.vue'
withDefaults(
defineProps<{
overview: Omit<Awaited<ReturnType<typeof getUsageRecordInsightsOverview>>, 'analytics'>
diagnostics: Omit<Awaited<ReturnType<typeof getUsageRecordInsightsDiagnostics>>, 'analytics'>
loading?: boolean
}>(),
{
loading: false,
},
)
defineProps<{
health: UsageQueryState<UsageMetricResponse<'health'>>
diagnostics: UsageQueryState<UsageDiagnosticsResponse>
performance: UsageQueryState<UsageMetricResponse<UsagePerformanceMetric>>
cost: UsageQueryState<UsageMetricResponse<UsageCostMetric>>
}>()
const { t } = useI18n({ useScope: 'global' })
const diagnosticDimension = defineModel('diagnosticDimension', {
type: String,
default: 'model',
})
const diagnosticDimension = defineModel<string>('diagnosticDimension', { required: true })
const performanceMetric = defineModel<UsagePerformanceMetric>('performanceMetric', { required: true })
const costMetric = defineModel<UsageCostMetric>('costMetric', { required: true })
</script>
<template>
@@ -36,27 +26,28 @@ const diagnosticDimension = defineModel('diagnosticDimension', {
:aria-label="t('usage.insights.useStatisticalObservations')"
>
<UsageHealthCard
:health="overview.health"
:granularity="overview.granularity"
:loading="loading"
:health="health.data?.data ?? null"
:granularity="health.data?.granularity ?? ''"
:loading="health.loading"
:error="health.error"
/>
<UsageDiagnosticCard
v-model:dimension="diagnosticDimension"
:diagnostics="diagnostics"
:loading="loading"
:diagnostics="diagnostics.data"
:loading="diagnostics.loading"
:error="diagnostics.error"
/>
<UsagePerformanceCard
:performance="overview.performance"
:activity="overview.health.points"
:loading="loading"
v-model="performanceMetric"
:performance="performance.data"
:loading="performance.loading"
:error="performance.error"
/>
<UsageCostCard
:cost="overview.cost"
:activity="overview.health.points"
:loading="loading"
v-model="costMetric"
:result="cost.data"
:loading="cost.loading"
:error="cost.error"
/>
</section>
</template>
@@ -1,13 +1,13 @@
<script setup lang="ts">
import type { EChartsOption } from 'echarts'
import type { getUsageRecordInsightsOverview } from '@/api'
import type { UsageMetricResponse, UsageOverviewLoad, UsagePerformanceMetric } from '@/api'
import { ZCard, ZEmpty, ZSegmented } from '@codex-proxy/ui'
import { computed, shallowRef } from 'vue'
import { computed } from 'vue'
import { useI18n } from 'vue-i18n'
import BaseChart from '@/components/charts/BaseChart.vue'
import { requestActivityByBucket, zeroInactiveValues } from '@/components/charts/timeSeriesGap'
import { zeroInactiveValues } from '@/components/charts/timeSeriesGap'
import { chartTooltipContent, tooltipIndex, tooltipRows } from '@/components/charts/tooltip'
import { useChartPalette } from '@/composables/useChartPalette'
import { formatLocalizedCompactNumber as formatCompactNumber, formatDuration } from '@/utils/format'
@@ -24,8 +24,8 @@ import {
const props = withDefaults(
defineProps<{
performance: Performance
activity: Activity
performance: UsageMetricResponse<UsagePerformanceMetric> | null
error?: string
loading?: boolean
}>(),
{
@@ -35,33 +35,25 @@ const props = withDefaults(
const { t } = useI18n({ useScope: 'global' })
type Performance = Awaited<ReturnType<typeof getUsageRecordInsightsOverview>>['performance']
type PerformancePoint = Performance['points'][number]
type PerformanceMetric = Exclude<keyof PerformancePoint, 'bucket' | 'label' | 'load'>
interface PerformanceSeries {
field: PerformanceMetric | keyof PerformancePoint['load']
value: (point: Omit<Performance, 'points'>) => number | null
field: string
summary: number | null
values: (number | null)[]
label: string
color: string
yAxisIndex: number
format: (value: number | null) => string
}
type Activity = Awaited<ReturnType<typeof getUsageRecordInsightsOverview>>['health']['points']
const activeView = shallowRef('total')
const activeView = defineModel<UsagePerformanceMetric>({ required: true })
const { palette } = useChartPalette()
const performancePoints = computed<PerformancePoint[]>(() => props.performance.points)
const requestActivity = computed(() => requestActivityByBucket(
performancePoints.value.map(point => point.bucket),
props.activity,
))
const performancePoints = computed(() => props.performance?.data.points ?? [])
const requestActivity = computed(() => performancePoints.value.map(point => point.requestCount > 0))
const viewOptions = computed(() => ([
{
label: t('usage.latency.totalLatency'),
value: 'total',
value: 'latency',
},
{
label: t('usage.ttft'),
@@ -78,12 +70,14 @@ const viewOptions = computed(() => ([
]))
const seriesDefinitions = computed<PerformanceSeries[]>(() => {
const response = props.performance
if (!response || response.metric !== activeView.value)
return []
const theme = palette.value
if (activeView.value === 'load') {
return [
if (response.metric === 'load') {
const fields: { field: keyof UsageOverviewLoad, label: string, color: string, yAxisIndex: number, format: (value: number | null) => string }[] = [
{
field: 'averageConcurrency',
value: point => point.load.averageConcurrency,
label: t('usage.performance.averageConcurrency'),
color: theme.info,
yAxisIndex: 0,
@@ -91,7 +85,6 @@ const seriesDefinitions = computed<PerformanceSeries[]>(() => {
},
{
field: 'peakConcurrency',
value: point => point.load.peakConcurrency,
label: t('usage.performance.peakConcurrency'),
color: theme.warning,
yAxisIndex: 0,
@@ -99,7 +92,6 @@ const seriesDefinitions = computed<PerformanceSeries[]>(() => {
},
{
field: 'admissionDecisionP95Ms',
value: point => point.load.admissionDecisionP95Ms,
label: t('usage.performance.admissionP95'),
color: theme.success,
yAxisIndex: 1,
@@ -107,52 +99,56 @@ const seriesDefinitions = computed<PerformanceSeries[]>(() => {
},
{
field: 'accountSelectionWaitP95Ms',
value: point => point.load.accountSelectionWaitP95Ms,
label: t('usage.performance.accountSelectionP95'),
color: theme.reasoning,
yAxisIndex: 1,
format: formatDuration,
},
]
return fields.map(item => ({
...item,
summary: response.data.summary[item.field],
values: response.data.points.map(point => point.value[item.field]),
}))
}
const throughput = activeView.value === 'throughput'
const fields: PerformanceMetric[] = throughput
? ['outputThroughputP10', 'outputThroughputP50', 'outputThroughputP90']
: activeView.value === 'firstToken'
? ['firstTokenP50Ms', 'firstTokenP95Ms', 'firstTokenP99Ms']
: ['latencyP50Ms', 'latencyP95Ms', 'latencyP99Ms']
const labels = throughput
? ['P10', 'P50', 'P90']
: ['P50', 'P95', 'P99']
const colors = [theme.info, theme.warning, theme.danger]
return fields.map((field, index) => ({
if (response.metric === 'throughput') {
return (['p10', 'p50', 'p90'] as const).map((field, index) => ({
field,
summary: response.data.summary[field],
values: response.data.points.map(point => point.value[field]),
label: field.toUpperCase(),
color: colors[index]!,
yAxisIndex: 0,
format: formatThroughput,
}))
}
return (['p50Ms', 'p95Ms', 'p99Ms'] as const).map((field, index) => ({
field,
value: point => point[field],
label: labels[index]!,
summary: response.data.summary[field],
values: response.data.points.map(point => point.value[field]),
label: ['P50', 'P95', 'P99'][index]!,
color: colors[index]!,
yAxisIndex: 0,
format: throughput ? formatThroughput : formatDuration,
format: formatDuration,
}))
})
const selectedSeries = computed(() => seriesDefinitions.value.map((series) => {
const values = performancePoints.value.map(series.value)
return {
...series,
// 并发包含之前时间桶开始的请求,不按本桶新请求数补零
values: series.field === 'averageConcurrency' || series.field === 'peakConcurrency'
? values
: zeroInactiveValues(values, requestActivity.value),
}
}))
const selectedSeries = computed(() => seriesDefinitions.value.map(series => ({
...series,
// 并发包含之前时间桶开始的请求,不按本桶新请求数补零
values: series.field === 'averageConcurrency' || series.field === 'peakConcurrency'
? series.values
: zeroInactiveValues(series.values, requestActivity.value),
})))
const summaryMetrics = computed(() => seriesDefinitions.value.map(series => ({
label: series.label,
value: series.format(series.value(props.performance)),
value: series.format(series.summary),
})))
const hasData = computed(() => !props.loading
&& seriesDefinitions.value.some(series => performancePoints.value.some(point => series.value(point) != null)))
&& seriesDefinitions.value.some(series => series.values.some(value => value != null)))
const chartOption = computed<EChartsOption>(() => {
const theme = palette.value
@@ -241,7 +237,7 @@ function formatThroughput(value: number | null) {
>
<template #actions>
<div class="flex justify-end">
<ZSegmented v-model="activeView" :aria-label="t('usage.performance.performanceIndicators')" :options="viewOptions" :disabled="loading" />
<ZSegmented v-model="activeView" :aria-label="t('usage.performance.performanceIndicators')" :options="viewOptions" />
</div>
</template>
@@ -265,7 +261,7 @@ function formatThroughput(value: number | null) {
size="small"
surface="none"
:title="
loading
error || (loading
? t('usage.performance.loading')
: activeView === 'load'
? t('usage.performance.noLoad')
@@ -273,7 +269,7 @@ function formatThroughput(value: number | null) {
? t('usage.performance.noTtftSamples')
: activeView === 'throughput'
? t('usage.performance.noThroughput')
: t('usage.performance.noLatency')
: t('usage.performance.noLatency'))
"
class="h-52.5 place-content-center"
/>
@@ -8,13 +8,15 @@ import { computed } from 'vue'
import { useI18n } from 'vue-i18n'
const props = defineProps<{
summary: Omit<Awaited<ReturnType<typeof getUsageRecordSummary>>, 'analytics'>
summary: Omit<Awaited<ReturnType<typeof getUsageRecordSummary>>, 'analytics'> | null
loading?: boolean
error?: string
}>()
const { t } = useI18n({ useScope: 'global' })
const averageLatency = computed(() => {
const raw = props.summary.averageLatencyMsValue
const raw = props.summary?.averageLatencyMsValue
const value = raw == null || !Number.isFinite(raw) || raw < 0
? 0
: raw
@@ -47,7 +49,7 @@ const items = computed(() => [
key: 'requests',
label: t('usage.successfulRequest'),
icon: Activity,
value: props.summary.totalRequests,
value: props.summary?.totalRequests,
detail: t('usage.withinFilterRange'),
tone: 'bg-cp-blue-container text-cp-blue-on-container',
},
@@ -55,15 +57,15 @@ const items = computed(() => [
key: 'tokens',
label: t('usage.tokens.totalTokens'),
icon: FileText,
value: props.summary.totalTokens,
detail: t('usage.inputOutput', { inputTokens: props.summary.inputTokens, outputTokens: props.summary.outputTokens }),
value: props.summary?.totalTokens,
detail: t('usage.inputOutput', { inputTokens: props.summary?.inputTokens, outputTokens: props.summary?.outputTokens }),
tone: 'bg-cp-green-container text-cp-green-on-container',
},
{
key: 'cached',
label: t('usage.cachedTokens'),
icon: Database,
value: props.summary.cachedInputTokens,
value: props.summary?.cachedInputTokens,
detail: t('usage.cacheReadHit'),
tone: 'bg-cp-orange-container text-cp-orange-on-container',
},
@@ -96,10 +98,10 @@ const items = computed(() => [
{{ item.label }}
</span>
<strong class="my-2 block truncate font-mono text-[24px] leading-none font-extrabold text-cp-text">
{{ item.value }}<span v-if="item.unit" class="ml-0.5 font-sans text-cp font-semibold text-cp-text-secondary">{{ item.unit }}</span>
{{ summary ? item.value : '—' }}<span v-if="summary && item.unit" class="ml-0.5 font-sans text-cp font-semibold text-cp-text-secondary">{{ item.unit }}</span>
</strong>
<span class="block truncate text-cp-sm leading-none font-emphasis text-cp-text-quaternary">
{{ item.detail }}
{{ error || (loading ? t('common.status.loading') : item.detail) }}
</span>
</div>
</ZCard>
@@ -0,0 +1,75 @@
import type { UsageCostMetric, UsagePerformanceMetric } from '@/api'
import type { RequestOptions } from '@/api/request'
import { computed, shallowRef, watch } from 'vue'
import { getUsageCost, getUsageHealth, getUsagePerformance, getUsageRecordInsightsDiagnostics, getUsageRecordSummary } from '@/api'
import { useRequestState } from '@/composables/useRequestState'
type UsageRange = Parameters<typeof getUsageRecordSummary>[0]
export interface UsageQueryState<T> {
data: T | null
loading: boolean
error: string
}
export function useUsageInsights(scopedParams: () => UsageRange) {
const performanceMetric = shallowRef<UsagePerformanceMetric>('latency')
const costMetric = shallowRef<UsageCostMetric>('cost')
const diagnosticDimension = shallowRef('model')
const summary = useUsageQuery(getUsageRecordSummary)
const health = useUsageQuery(getUsageHealth)
const performance = useUsageQuery((params, options) => getUsagePerformance({ ...params, metric: performanceMetric.value }, options))
const cost = useUsageQuery((params, options) => getUsageCost({ ...params, metric: costMetric.value }, options))
const diagnostics = useUsageQuery((params, options) => getUsageRecordInsightsDiagnostics({ ...params, dimension: diagnosticDimension.value }, options))
watch(performanceMetric, () => void performance.load(scopedParams()))
watch(costMetric, () => void cost.load(scopedParams()))
watch(diagnosticDimension, () => void diagnostics.load(scopedParams()))
function loadUsageAnalytics(params: UsageRange = scopedParams()) {
// 各查询收到结果即发布,慢指标不会阻塞其他卡片
return Promise.all([
summary,
health,
performance,
cost,
diagnostics,
].map(query => query.load(params)))
}
return {
summary: summary.state,
health: health.state,
performance: performance.state,
cost: cost.state,
diagnostics: diagnostics.state,
performanceMetric,
costMetric,
diagnosticDimension,
loadUsageAnalytics,
}
}
function useUsageQuery<T>(fetch: (params: UsageRange, options: RequestOptions) => Promise<T>) {
const data = shallowRef<T | null>(null)
const request = useRequestState()
const state = computed<UsageQueryState<T>>(() => ({ data: data.value, loading: request.loading.value, error: request.error.value }))
async function load(params: UsageRange) {
const id = request.start()
data.value = null
try {
const result = await fetch(params, { signal: request.signal })
if (request.isCurrent(id))
data.value = result
}
catch (cause) {
request.fail(id, cause)
}
finally {
request.finish(id)
}
}
return { state, load }
}
@@ -1,17 +1,12 @@
import type { Ref } from 'vue'
import type { UsageTimeRangeParams } from './useUsageTimeRange'
import type { UsageRecordStatus } from '@/api'
import type { AnalyticsMetadata } from '@/api/modules/analytics'
import { computed, onMounted, onScopeDispose, shallowRef, watch } from 'vue'
import {
getUsageRecordInsightsDiagnostics,
getUsageRecordInsightsOverview,
getUsageRecords,
getUsageRecordSummary,
} from '@/api'
import { computed, onMounted, shallowRef, watch } from 'vue'
import { getUsageRecords } from '@/api'
import { useStablePagedQuery } from '@/composables/useStablePagedQuery'
import { withMinimumDuration } from '@/utils/operation'
import { useUsageInsights } from './useUsageInsights'
interface UseUsageRecordsTableOptions {
timeRangeParams: Readonly<Ref<UsageTimeRangeParams>>
@@ -21,16 +16,12 @@ interface UseUsageRecordsTableOptions {
type UsageLoadScope = 'all' | 'table'
export type UsageRecordFilterField = 'key' | 'account' | 'model'
type LoadingAnalytics<T> = Omit<T, 'analytics'> & { analytics: AnalyticsMetadata | null }
interface UsageLoadOptions {
scope?: UsageLoadScope
}
export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
const analyticsLoading = shallowRef(true)
const summary = shallowRef(emptySummary())
const insights = shallowRef(emptyInsights())
const filterField = shallowRef<UsageRecordFilterField>('key')
const filterValue = shallowRef('')
const filterParams = computed(() => {
@@ -58,17 +49,13 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
})
const { currentPage, pageSize, loading, error, items: records, total: totalRecords } = table
const refreshingList = shallowRef(false)
const diagnosticDimension = shallowRef('model')
let analyticsRequestId = 0
let diagnosticRequestId = 0
let analyticsController: AbortController | undefined
let diagnosticController: AbortController | undefined
const scopedParams = () => ({
...options.timeRangeParams.value,
...(providerQuery.value
? { provider: providerQuery.value }
: {}),
})
const { loadUsageAnalytics, ...insights } = useUsageInsights(scopedParams)
const usagePagination = computed(() => ({
currentPage: currentPage.value,
pageSize: pageSize.value,
@@ -115,72 +102,6 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
])
}
async function loadUsageAnalytics(globalParams: ReturnType<typeof scopedParams>) {
const requestId = ++analyticsRequestId
const diagnosticsId = ++diagnosticRequestId
analyticsController?.abort()
diagnosticController?.abort()
analyticsController = new AbortController()
const requestOptions = { signal: analyticsController.signal }
const dimension = diagnosticDimension.value
analyticsLoading.value = true
summary.value = emptySummary()
insights.value = emptyInsights()
try {
const [nextSummary, overview, diagnostics] = await Promise.all([
getUsageRecordSummary(globalParams, requestOptions),
getUsageRecordInsightsOverview(globalParams, requestOptions),
getUsageRecordInsightsDiagnostics({
...globalParams,
dimension,
}, requestOptions),
])
if (requestId !== analyticsRequestId)
return
summary.value = nextSummary
insights.value = {
overview,
diagnostics:
diagnosticsId === diagnosticRequestId && dimension === diagnosticDimension.value
? diagnostics
: insights.value.diagnostics,
}
}
catch {
if (requestId === analyticsRequestId) {
summary.value = emptySummary()
insights.value = emptyInsights()
}
}
finally {
if (requestId === analyticsRequestId) {
analyticsLoading.value = false
}
}
}
async function loadDiagnostics() {
const requestId = ++diagnosticRequestId
diagnosticController?.abort()
diagnosticController = new AbortController()
const dimension = diagnosticDimension.value
const params = scopedParams()
try {
const diagnostics = await getUsageRecordInsightsDiagnostics({
...params,
dimension,
}, { signal: diagnosticController.signal })
if (requestId !== diagnosticRequestId || dimension !== diagnosticDimension.value)
return
insights.value = {
...insights.value,
diagnostics,
}
}
catch {}
}
async function refreshUsageRecords() {
if (refreshingList.value || loading.value)
return
@@ -219,10 +140,6 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
loadUsageRecords()
})
watch(diagnosticDimension, () => {
void loadDiagnostics()
})
watch(providerQuery, () => {
void loadUsageRecords()
})
@@ -243,13 +160,6 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
onCleanup(() => clearTimeout(timer))
})
onScopeDispose(() => {
analyticsRequestId += 1
diagnosticRequestId += 1
analyticsController?.abort()
diagnosticController?.abort()
})
return {
currentPage,
pageSize,
@@ -259,108 +169,12 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
usagePagination,
loading,
error,
analyticsLoading,
records,
summary,
insights,
...insights,
refreshingList,
diagnosticDimension,
loadUsageRecords,
refreshUsageRecords,
handlePageChange,
handlePageSizeChange,
}
}
function emptySummary() {
const summary: LoadingAnalytics<Awaited<ReturnType<typeof getUsageRecordSummary>>> = {
analytics: null,
totalRequests: '0',
inputTokens: '0',
outputTokens: '0',
cachedInputTokens: '0',
cacheWriteInputTokens: '0',
totalTokens: '0',
averageLatencyMs: '0 ms',
averageLatencyMsValue: null,
}
return summary
}
function emptyInsights() {
return {
overview: emptyOverview(),
diagnostics: emptyDiagnostics(),
}
}
function emptyOverview() {
const overview: LoadingAnalytics<Awaited<ReturnType<typeof getUsageRecordInsightsOverview>>> = {
analytics: null,
granularity: '',
health: {
totalRequests: 0,
successRequests: 0,
failedRequests: 0,
cancelledRequests: 0,
incompleteRequests: 0,
callerErrorRequests: 0,
successRate: 0,
completionRate: 0,
requestChangeRate: null,
successRateChange: null,
points: [],
},
performance: {
latencyP50Ms: null,
latencyP95Ms: null,
latencyP99Ms: null,
firstTokenP50Ms: null,
firstTokenP95Ms: null,
firstTokenP99Ms: null,
outputThroughputP10: null,
outputThroughputP50: null,
outputThroughputP90: null,
load: {
averageConcurrency: 0,
peakConcurrency: 0,
admissionDecisionP95Ms: null,
accountSelectionWaitP95Ms: null,
},
points: [],
},
cost: {
estimatedCost: null,
standardCost: null,
noCacheCost: null,
cacheSavings: null,
tierPremium: null,
costPerRequest: null,
costPerSuccessfulRequest: null,
tokensPerRequest: 0,
cachedTokenRate: 0,
cacheHitRequestRate: 0,
inputTokens: 0,
outputTokens: 0,
cachedInputTokens: 0,
totalTokens: 0,
points: [],
coverage: {
known: 0,
partial: 0,
unknown: 0,
notBillable: 0,
},
},
}
return overview
}
function emptyDiagnostics() {
const diagnostics: LoadingAnalytics<Awaited<ReturnType<typeof getUsageRecordInsightsDiagnostics>>> = {
analytics: null,
dimension: 'model',
items: [],
}
return diagnostics
}
+13 -6
View File
@@ -48,10 +48,14 @@ const {
providerQuery,
usagePagination,
loading,
analyticsLoading,
records,
summary,
insights,
health,
performance,
cost,
diagnostics,
performanceMetric,
costMetric,
refreshingList,
diagnosticDimension,
loadUsageRecords,
@@ -99,12 +103,15 @@ watch(timeRange, () => {
</template>
</ZPageHeader>
<UsageSummaryCards :summary="summary" />
<UsageSummaryCards :summary="summary.data" :loading="summary.loading" :error="summary.error" />
<UsageInsightsGrid
v-model:diagnostic-dimension="diagnosticDimension"
:overview="insights.overview"
:diagnostics="insights.diagnostics"
:loading="analyticsLoading"
v-model:performance-metric="performanceMetric"
v-model:cost-metric="costMetric"
:health="health"
:performance="performance"
:cost="cost"
:diagnostics="diagnostics"
/>
<ZCard