mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix(accounts): calculate model success rates from request outcomes (#611)
Co-authored-by: Zyy <zyycn@users.noreply.github.com> Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -472,6 +472,80 @@ impl CostCoverage {
|
||||
}
|
||||
}
|
||||
|
||||
/// 请求健康计数与服务成功率,调用方错误和取消不计入有效样本
|
||||
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
|
||||
pub struct RequestHealth {
|
||||
pub success_requests: u64,
|
||||
pub failed_requests: u64,
|
||||
pub cancelled_requests: u64,
|
||||
pub incomplete_requests: u64,
|
||||
pub caller_error_requests: u64,
|
||||
}
|
||||
|
||||
impl RequestHealth {
|
||||
#[must_use]
|
||||
pub fn from_counts(
|
||||
success: u64,
|
||||
failed: u64,
|
||||
cancelled: u64,
|
||||
incomplete: u64,
|
||||
caller_error: u64,
|
||||
) -> Self {
|
||||
Self {
|
||||
success_requests: success,
|
||||
failed_requests: failed.saturating_sub(caller_error),
|
||||
cancelled_requests: cancelled,
|
||||
incomplete_requests: incomplete,
|
||||
caller_error_requests: caller_error,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn from_metrics(metrics: &RequestMetrics) -> Self {
|
||||
Self::from_counts(
|
||||
metrics.success_count,
|
||||
metrics.failure_count,
|
||||
metrics.cancelled_count,
|
||||
metrics.incomplete_count,
|
||||
metrics.caller_error_count,
|
||||
)
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn success_rate(self) -> Option<f64> {
|
||||
let total = self.eligible_requests();
|
||||
(total > 0).then(|| self.success_requests as f64 / total as f64 * 100.0)
|
||||
}
|
||||
|
||||
// 未完成保留交付终态,但上游失败、超时和进程中断都不能视为服务成功
|
||||
pub(crate) fn non_success_requests(self) -> u64 {
|
||||
self.failed_requests
|
||||
.saturating_add(self.incomplete_requests)
|
||||
}
|
||||
|
||||
pub(crate) fn eligible_requests(self) -> u64 {
|
||||
self.success_requests
|
||||
.saturating_add(self.non_success_requests())
|
||||
}
|
||||
|
||||
pub(crate) fn add_metrics(&mut self, metrics: &RequestMetrics) {
|
||||
self.add_window(Self::from_metrics(metrics));
|
||||
}
|
||||
|
||||
pub(crate) fn add_window(&mut self, other: Self) {
|
||||
self.success_requests = self.success_requests.saturating_add(other.success_requests);
|
||||
self.failed_requests = self.failed_requests.saturating_add(other.failed_requests);
|
||||
self.cancelled_requests = self
|
||||
.cancelled_requests
|
||||
.saturating_add(other.cancelled_requests);
|
||||
self.incomplete_requests = self
|
||||
.incomplete_requests
|
||||
.saturating_add(other.incomplete_requests);
|
||||
self.caller_error_requests = self
|
||||
.caller_error_requests
|
||||
.saturating_add(other.caller_error_requests);
|
||||
}
|
||||
}
|
||||
|
||||
/// 请求级聚合指标
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct RequestMetrics {
|
||||
@@ -622,7 +696,7 @@ pub struct AccountPoolMetrics {
|
||||
pub struct AccountModelUsage {
|
||||
pub model: String,
|
||||
pub request_count: u64,
|
||||
pub success_count: u64,
|
||||
pub health: RequestHealth,
|
||||
pub input_tokens: Option<u64>,
|
||||
pub output_tokens: Option<u64>,
|
||||
pub cached_input_tokens: Option<u64>,
|
||||
@@ -635,7 +709,7 @@ pub struct AccountModelUsage {
|
||||
pub total_tokens: Option<u64>,
|
||||
pub cost_coverage: CostCoverage,
|
||||
pub costs: Vec<CurrencyCost>,
|
||||
pub last_used_at: DateTime<Utc>,
|
||||
pub last_used_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
/// 账号在单小时窗口内的请求数
|
||||
|
||||
@@ -19,8 +19,8 @@ use crate::{
|
||||
DashboardPeriodMetrics, DashboardQuery, DashboardResult, DecimalAmount,
|
||||
DiagnosticDimension, DiagnosticsItem, DiagnosticsResult, Granularity, HealthStatus,
|
||||
HealthTimeline, HealthTimelinePoint, OpsErrorPage, OpsErrorQuery, ProviderBillingInput,
|
||||
RequestMetricPoint, RequestMetrics, TimeRange, Trend, TrendKind, TrendPoint,
|
||||
TrendSummary, UsageBilling, UsageBillingProjection, UsageCacheEfficiency,
|
||||
RequestHealth, RequestMetricPoint, RequestMetrics, TimeRange, Trend, TrendKind,
|
||||
TrendPoint, TrendSummary, UsageBilling, UsageBillingProjection, UsageCacheEfficiency,
|
||||
UsageCalculatedBillingFact, UsageConcurrencyPoint, UsageCostComparison,
|
||||
UsageCostSummary, UsageDetail, UsageFilter, UsageInsight, UsageInsightData,
|
||||
UsageInsightMetric, UsageInsights, UsageInsightsCost, UsageInsightsCostPoint,
|
||||
@@ -574,7 +574,7 @@ fn build_usage_insights(
|
||||
}
|
||||
|
||||
let requests = &overview.requests;
|
||||
let health_window = HealthWindow::from_metrics(requests);
|
||||
let health_window = RequestHealth::from_metrics(requests);
|
||||
let failed_requests = health_window.failed_requests;
|
||||
let health_requests = health_window.eligible_requests();
|
||||
let health = UsageInsightsHealth {
|
||||
@@ -594,7 +594,7 @@ fn build_usage_insights(
|
||||
points: trend
|
||||
.iter()
|
||||
.map(|point| {
|
||||
let health_window = HealthWindow::from_metrics(&point.metrics);
|
||||
let health_window = RequestHealth::from_metrics(&point.metrics);
|
||||
UsageInsightsHealthPoint {
|
||||
total_requests: point.metrics.request_count,
|
||||
bucket_start: point.bucket_start,
|
||||
@@ -1125,7 +1125,7 @@ pub(super) fn health_timeline_at(
|
||||
.map(|index| {
|
||||
(
|
||||
start + Duration::minutes(HEALTH_TIMELINE_SLOT_MINUTES * index),
|
||||
HealthWindow::default(),
|
||||
RequestHealth::default(),
|
||||
)
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
@@ -1151,7 +1151,7 @@ pub(super) fn health_timeline_at(
|
||||
let totals = buckets
|
||||
.iter()
|
||||
.filter(|(bucket_start, _)| *bucket_start <= current_slot)
|
||||
.fold(HealthWindow::default(), |mut totals, (_, bucket)| {
|
||||
.fold(RequestHealth::default(), |mut totals, (_, bucket)| {
|
||||
totals.add_window(*bucket);
|
||||
totals
|
||||
});
|
||||
@@ -1215,10 +1215,10 @@ fn trend(
|
||||
}
|
||||
}
|
||||
let average_latency_ms = average(metrics.latency_sum_ms, metrics.latency_count);
|
||||
let service_failure_count = service_failure_count(&metrics);
|
||||
let service_failure_count = RequestHealth::from_metrics(&metrics).failed_requests;
|
||||
let success_rate = ratio(
|
||||
metrics.success_count,
|
||||
HealthWindow::from_metrics(&metrics).eligible_requests(),
|
||||
RequestHealth::from_metrics(&metrics).eligible_requests(),
|
||||
);
|
||||
let cache_hit_request_rate = ratio(
|
||||
metrics.cache_hit_request_count,
|
||||
@@ -1267,7 +1267,7 @@ fn trend(
|
||||
}
|
||||
|
||||
fn trend_point(point: RequestMetricPoint) -> TrendPoint {
|
||||
let service_failure_count = service_failure_count(&point.metrics);
|
||||
let service_failure_count = RequestHealth::from_metrics(&point.metrics).failed_requests;
|
||||
TrendPoint {
|
||||
bucket_start: point.bucket_start,
|
||||
granularity: point.granularity,
|
||||
@@ -1287,7 +1287,7 @@ fn trend_point(point: RequestMetricPoint) -> TrendPoint {
|
||||
),
|
||||
success_rate: ratio(
|
||||
point.metrics.success_count,
|
||||
HealthWindow::from_metrics(&point.metrics).eligible_requests(),
|
||||
RequestHealth::from_metrics(&point.metrics).eligible_requests(),
|
||||
),
|
||||
metrics: point.metrics,
|
||||
cost_coverage: point.cost_coverage,
|
||||
@@ -1372,57 +1372,7 @@ fn rate_or_zero(numerator: u64, denominator: u64) -> f64 {
|
||||
ratio(numerator, denominator).unwrap_or(0.0)
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, Default)]
|
||||
struct HealthWindow {
|
||||
success_requests: u64,
|
||||
failed_requests: u64,
|
||||
cancelled_requests: u64,
|
||||
incomplete_requests: u64,
|
||||
caller_error_requests: u64,
|
||||
}
|
||||
|
||||
impl HealthWindow {
|
||||
fn from_metrics(metrics: &RequestMetrics) -> Self {
|
||||
Self {
|
||||
success_requests: metrics.success_count,
|
||||
failed_requests: service_failure_count(metrics),
|
||||
cancelled_requests: metrics.cancelled_count,
|
||||
incomplete_requests: metrics.incomplete_count,
|
||||
caller_error_requests: metrics.caller_error_count,
|
||||
}
|
||||
}
|
||||
|
||||
// 未完成保留交付终态,但上游失败、超时和进程中断都不能视为服务成功
|
||||
fn non_success_requests(self) -> u64 {
|
||||
self.failed_requests
|
||||
.saturating_add(self.incomplete_requests)
|
||||
}
|
||||
|
||||
fn eligible_requests(self) -> u64 {
|
||||
self.success_requests
|
||||
.saturating_add(self.non_success_requests())
|
||||
}
|
||||
|
||||
fn add_metrics(&mut self, metrics: &RequestMetrics) {
|
||||
self.add_window(Self::from_metrics(metrics));
|
||||
}
|
||||
|
||||
fn add_window(&mut self, other: Self) {
|
||||
self.success_requests = self.success_requests.saturating_add(other.success_requests);
|
||||
self.failed_requests = self.failed_requests.saturating_add(other.failed_requests);
|
||||
self.cancelled_requests = self
|
||||
.cancelled_requests
|
||||
.saturating_add(other.cancelled_requests);
|
||||
self.incomplete_requests = self
|
||||
.incomplete_requests
|
||||
.saturating_add(other.incomplete_requests);
|
||||
self.caller_error_requests = self
|
||||
.caller_error_requests
|
||||
.saturating_add(other.caller_error_requests);
|
||||
}
|
||||
}
|
||||
|
||||
fn health_status(bucket: HealthWindow, is_future: bool) -> HealthStatus {
|
||||
fn health_status(bucket: RequestHealth, is_future: bool) -> HealthStatus {
|
||||
let eligible_requests = bucket.eligible_requests();
|
||||
if is_future {
|
||||
HealthStatus::Future
|
||||
@@ -1443,14 +1393,6 @@ fn health_status(bucket: HealthWindow, is_future: bool) -> HealthStatus {
|
||||
}
|
||||
}
|
||||
|
||||
fn health_reliability(bucket: HealthWindow) -> Option<f64> {
|
||||
let eligible_requests = bucket.eligible_requests();
|
||||
(eligible_requests > 0)
|
||||
.then(|| bucket.success_requests as f64 / eligible_requests as f64 * 100.0)
|
||||
}
|
||||
|
||||
fn service_failure_count(metrics: &RequestMetrics) -> u64 {
|
||||
metrics
|
||||
.failure_count
|
||||
.saturating_sub(metrics.caller_error_count)
|
||||
fn health_reliability(bucket: RequestHealth) -> Option<f64> {
|
||||
bucket.success_rate()
|
||||
}
|
||||
|
||||
@@ -455,20 +455,14 @@ pub(super) fn account_model_usage_view(
|
||||
.costs
|
||||
.iter()
|
||||
.find(|cost| cost.currency.eq_ignore_ascii_case("USD"));
|
||||
let success_rate = usage.health.success_rate();
|
||||
ModelUsageView {
|
||||
model: usage.model,
|
||||
request_count: usage.request_count,
|
||||
request_count_display: format_number(usage.request_count),
|
||||
success_rate: (usage.request_count > 0)
|
||||
.then(|| usage.success_count as f64 * 100.0 / usage.request_count as f64),
|
||||
success_rate_display: if usage.request_count == 0 {
|
||||
"—".to_owned()
|
||||
} else {
|
||||
format!(
|
||||
"{:.1}%",
|
||||
usage.success_count as f64 * 100.0 / usage.request_count as f64
|
||||
)
|
||||
},
|
||||
success_rate,
|
||||
success_rate_display: success_rate
|
||||
.map_or_else(|| "—".to_owned(), |rate| format!("{rate:.1}%")),
|
||||
input_tokens: usage.input_tokens,
|
||||
input_tokens_display: display_optional_tokens(usage.input_tokens),
|
||||
output_tokens: usage.output_tokens,
|
||||
@@ -495,9 +489,11 @@ pub(super) fn account_model_usage_view(
|
||||
partial_cost_count: u64::from(cost_estimate_status == "partial"),
|
||||
unknown_cost_count: usage.cost_coverage.unavailable_count,
|
||||
costs: usage.costs.iter().map(account_currency_cost_view).collect(),
|
||||
last_used_at: time.rfc3339(&usage.last_used_at),
|
||||
last_used_at_full_display: Some(time.datetime(&usage.last_used_at)),
|
||||
last_used_at_display: time.relative(usage.last_used_at, now),
|
||||
last_used_at: usage.last_used_at.map(|value| time.rfc3339(&value)),
|
||||
last_used_at_full_display: usage.last_used_at.map(|value| time.datetime(&value)),
|
||||
last_used_at_display: usage
|
||||
.last_used_at
|
||||
.map_or_else(|| "—".to_owned(), |value| time.relative(value, now)),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -515,7 +515,7 @@ pub struct ModelUsageView {
|
||||
pub partial_cost_count: u64,
|
||||
pub unknown_cost_count: u64,
|
||||
pub costs: Vec<CurrencyCostView>,
|
||||
pub last_used_at: String,
|
||||
pub last_used_at: Option<String>,
|
||||
pub last_used_at_display: String,
|
||||
pub last_used_at_full_display: Option<String>,
|
||||
}
|
||||
|
||||
@@ -446,7 +446,8 @@ fn account_page_item(
|
||||
#[tokio::test]
|
||||
async fn account_usage_preserves_periods_cache_counts_and_missing_values() {
|
||||
use chrono::Utc;
|
||||
use gateway_admin::model::accounts::AccountUsage;
|
||||
use gateway_admin::model::accounts::{AccountModelUsage, AccountUsage};
|
||||
use gateway_admin::model::observability::RequestHealth;
|
||||
use gateway_admin::model::provider_credentials::{
|
||||
ProviderQuota, ProviderQuotaWindow, QuotaLocalUsageAttribution,
|
||||
};
|
||||
@@ -497,7 +498,35 @@ async fn account_usage_preserves_periods_cache_counts_and_missing_values() {
|
||||
bucket_start: now,
|
||||
request_count: 1,
|
||||
}],
|
||||
models: Vec::new(),
|
||||
models: [
|
||||
(
|
||||
"mixed",
|
||||
RequestHealth::from_counts(1, 2, 3, 1, 1),
|
||||
Some(now),
|
||||
),
|
||||
("failed", RequestHealth::from_counts(0, 1, 0, 0, 0), None),
|
||||
("no-sample", RequestHealth::from_counts(0, 1, 1, 0, 1), None),
|
||||
]
|
||||
.into_iter()
|
||||
.map(|(model, health, last_used_at)| AccountModelUsage {
|
||||
model: model.to_owned(),
|
||||
request_count: u64::from(last_used_at.is_some()),
|
||||
health,
|
||||
input_tokens: None,
|
||||
output_tokens: None,
|
||||
cached_input_tokens: None,
|
||||
cache_write_input_tokens: None,
|
||||
reasoning_output_tokens: None,
|
||||
image_input_tokens: None,
|
||||
image_output_tokens: None,
|
||||
image_request_count: 0,
|
||||
image_request_failed_count: 0,
|
||||
total_tokens: None,
|
||||
cost_coverage: Default::default(),
|
||||
costs: Vec::new(),
|
||||
last_used_at,
|
||||
})
|
||||
.collect(),
|
||||
};
|
||||
*fixture.account_usage.lock().unwrap() = vec![local_usage.clone()];
|
||||
*fixture.provider_quota.lock().unwrap() =
|
||||
@@ -545,6 +574,13 @@ async fn account_usage_preserves_periods_cache_counts_and_missing_values() {
|
||||
"general"
|
||||
}
|
||||
);
|
||||
assert_eq!(usage["models"][0]["successRateDisplay"], "33.3%");
|
||||
assert!((usage["models"][0]["successRate"].as_f64().unwrap() - 100.0 / 3.0).abs() < 0.001);
|
||||
assert_eq!(usage["models"][1]["successRate"], 0.0);
|
||||
assert_eq!(usage["models"][1]["successRateDisplay"], "0.0%");
|
||||
assert!(usage["models"][1]["lastUsedAt"].is_null());
|
||||
assert!(usage["models"][2]["successRate"].is_null());
|
||||
assert_eq!(usage["models"][2]["successRateDisplay"], "—");
|
||||
assert_eq!(usage["requestCount"], 1);
|
||||
assert_eq!(usage["totalTokens"], 120);
|
||||
if authentication_kind == "oauth" {
|
||||
|
||||
@@ -88,6 +88,11 @@ impl AnalyticsSnapshot<'_> {
|
||||
.account_fact_states("model_requests", range, account_ids, false, by_model)
|
||||
.await;
|
||||
}
|
||||
let retained_counter = if by_model {
|
||||
"request_count"
|
||||
} else {
|
||||
"completed_request_count"
|
||||
};
|
||||
let mut states = BTreeMap::new();
|
||||
// 排序与分组费用只需要账号总量,不额外展开模型维度
|
||||
let (collection, model) = if by_model {
|
||||
@@ -117,7 +122,7 @@ impl AnalyticsSnapshot<'_> {
|
||||
.push_bind(account_ids)
|
||||
.push("::text[])");
|
||||
push_summary_buckets(&mut query, &buckets, seconds);
|
||||
let counters = usage_counter_names();
|
||||
let counters = usage_counter_names(by_model);
|
||||
// 每行一次展开指标,避免对可能位于 TOAST 的同一 JSON 反复解码
|
||||
query.push(format!(
|
||||
"), totals as (select account_id,model,jsonb_build_object({}) as counts,max(last_used_at) as last_used_at from selected cross join lateral jsonb_to_record(counts) as metric({}) group by account_id,model),
|
||||
@@ -149,7 +154,7 @@ impl AnalyticsSnapshot<'_> {
|
||||
merge_account_states(&mut states, boundary)?;
|
||||
}
|
||||
if !self.has_pending_requests().await? {
|
||||
states.retain(|_, state| state.count("completed_request_count") > 0);
|
||||
states.retain(|_, state| state.count(retained_counter) > 0);
|
||||
return Ok(states);
|
||||
}
|
||||
let old = self
|
||||
@@ -192,7 +197,10 @@ impl AnalyticsSnapshot<'_> {
|
||||
.map_err(|_| invalid("invalid account cost difference"))?;
|
||||
state.costs.insert(currency.clone(), value.canonical());
|
||||
}
|
||||
repair.insert(key);
|
||||
// 失败终态没有用量极值,更新它们只需调整计数
|
||||
if previous.last_used_at().is_some() {
|
||||
repair.insert(key);
|
||||
}
|
||||
}
|
||||
merge_account_states(&mut states, current)?;
|
||||
// 撤销最大值不能用减法;只对发生源版本替换的账号模型做索引定位
|
||||
@@ -201,12 +209,12 @@ impl AnalyticsSnapshot<'_> {
|
||||
let state = states
|
||||
.get_mut(&(account.clone(), model.clone()))
|
||||
.ok_or_else(|| invalid("pending account state disappeared"))?;
|
||||
if state.count("completed_request_count") == 0 {
|
||||
if state.count(retained_counter) == 0 {
|
||||
*state = MetricState::default();
|
||||
continue;
|
||||
}
|
||||
let statement = format!(
|
||||
"select started_at from model_requests mr where provider_account_ref=$1 and ($2::text is null or coalesce(upstream_model_id,requested_model_id,'')=$2) and started_at >= $3 and started_at < $4 and {completed} order by started_at desc,id desc limit 1"
|
||||
"select started_at from model_requests mr where provider_account_ref=$1 and ($2::text is null or coalesce(upstream_model_id,requested_model_id,'')=$2) and started_at >= $3 and started_at < $4 and {completed} and mr.recovered_at is null order by started_at desc,id desc limit 1"
|
||||
);
|
||||
state.last_used_at = sqlx::query_scalar(sqlx::AssertSqlSafe(statement))
|
||||
.bind(&account)
|
||||
@@ -224,7 +232,7 @@ impl AnalyticsSnapshot<'_> {
|
||||
.collect::<Vec<_>>();
|
||||
for currency in zero_currencies {
|
||||
let statement = format!(
|
||||
"select exists(select 1 from model_requests mr where provider_account_ref=$1 and ($2::text is null or coalesce(upstream_model_id,requested_model_id,'')=$2) and started_at >= $3 and started_at < $4 and {completed} and cost_currency=$5 and cost_amount is not null)"
|
||||
"select exists(select 1 from model_requests mr where provider_account_ref=$1 and ($2::text is null or coalesce(upstream_model_id,requested_model_id,'')=$2) and started_at >= $3 and started_at < $4 and {completed} and mr.recovered_at is null and cost_currency=$5 and cost_amount is not null)"
|
||||
);
|
||||
let exists: bool = sqlx::query_scalar(sqlx::AssertSqlSafe(statement))
|
||||
.bind(&account)
|
||||
@@ -240,7 +248,7 @@ impl AnalyticsSnapshot<'_> {
|
||||
}
|
||||
}
|
||||
}
|
||||
states.retain(|_, state| state.count("completed_request_count") > 0);
|
||||
states.retain(|_, state| state.count(retained_counter) > 0);
|
||||
Ok(states)
|
||||
}
|
||||
|
||||
@@ -409,7 +417,10 @@ impl AnalyticsSnapshot<'_> {
|
||||
.push_bind(range.start)
|
||||
.push(" and mr.started_at < ")
|
||||
.push_bind(range.end);
|
||||
push_completed_usage_fact_filter(&mut query, "mr");
|
||||
if !by_model {
|
||||
push_completed_usage_fact_filter(&mut query, "mr");
|
||||
}
|
||||
push_unrecovered_request_filter(&mut query, "mr");
|
||||
query.push(" group by mr.provider_account_ref,mr.cost_currency");
|
||||
if by_model {
|
||||
query.push(",coalesce(mr.upstream_model_id,mr.requested_model_id,'')");
|
||||
@@ -465,7 +476,7 @@ fn merge_account_states(states: &mut AccountStates, incoming: AccountStates) ->
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn usage_counter_names() -> Vec<String> {
|
||||
fn usage_counter_names(by_model: bool) -> Vec<String> {
|
||||
let mut names = [
|
||||
"completed_request_count",
|
||||
"image_request_count",
|
||||
@@ -477,6 +488,19 @@ fn usage_counter_names() -> Vec<String> {
|
||||
]
|
||||
.map(str::to_owned)
|
||||
.to_vec();
|
||||
if by_model {
|
||||
names.extend(
|
||||
[
|
||||
"request_count",
|
||||
"success_count",
|
||||
"failure_count",
|
||||
"cancelled_count",
|
||||
"incomplete_count",
|
||||
"caller_error_count",
|
||||
]
|
||||
.map(str::to_owned),
|
||||
);
|
||||
}
|
||||
for name in super::metrics::TOKEN_METRICS {
|
||||
names.push(name.to_owned());
|
||||
names.push(format!("{name}_known_count"));
|
||||
@@ -491,25 +515,49 @@ fn usage_fact_aggregate_sql(table: &str, by_model: bool) -> String {
|
||||
"''::text"
|
||||
};
|
||||
let mut counters = vec![
|
||||
"'completed_request_count',count(*)".to_owned(),
|
||||
"'image_request_count',count(*) filter(where image_generation_succeeded is true)".to_owned(),
|
||||
"'image_request_failed_count',count(*) filter(where image_generation_succeeded is false)".to_owned(),
|
||||
"'window_total_tokens',coalesce(sum(coalesce(total_tokens,coalesce(input_tokens,0)+coalesce(output_tokens,0))),0)".to_owned(),
|
||||
"'completed_request_count',count(*) filter(where is_completed_usage)".to_owned(),
|
||||
"'image_request_count',count(*) filter(where is_completed_usage and image_generation_succeeded is true)".to_owned(),
|
||||
"'image_request_failed_count',count(*) filter(where is_completed_usage and image_generation_succeeded is false)".to_owned(),
|
||||
"'window_total_tokens',coalesce(sum(coalesce(total_tokens,coalesce(input_tokens,0)+coalesce(output_tokens,0))) filter(where is_completed_usage),0)".to_owned(),
|
||||
];
|
||||
if by_model {
|
||||
counters.push("'request_count',count(*)".to_owned());
|
||||
counters.push(
|
||||
"'caller_error_count',count(*) filter(where client_status_code between 400 and 499)"
|
||||
.to_owned(),
|
||||
);
|
||||
for (outcome, name) in [
|
||||
("succeeded", "success"),
|
||||
("failed", "failure"),
|
||||
("cancelled", "cancelled"),
|
||||
("incomplete", "incomplete"),
|
||||
] {
|
||||
counters.push(format!(
|
||||
"'{name}_count',count(*) filter(where outcome='{outcome}')"
|
||||
));
|
||||
}
|
||||
}
|
||||
for name in super::metrics::TOKEN_METRICS {
|
||||
counters.push(format!(
|
||||
"'{name}',coalesce(sum({name}),0),'{name}_known_count',count({name})"
|
||||
"'{name}',coalesce(sum({name}) filter(where is_completed_usage),0),'{name}_known_count',count({name}) filter(where is_completed_usage)"
|
||||
));
|
||||
}
|
||||
for source in ["provider_reported", "calculated", "unavailable"] {
|
||||
counters.push(format!(
|
||||
"'{source}_count',count(*) filter(where cost_source='{source}')"
|
||||
"'{source}_count',count(*) filter(where is_completed_usage and cost_source='{source}')"
|
||||
));
|
||||
}
|
||||
// 排序与分组费用已在 WHERE 限定完整交付,只在模型健康查询中按指标分别筛选
|
||||
let completed = if by_model {
|
||||
completed_usage_fact_predicate("mr")
|
||||
} else {
|
||||
"true".to_owned()
|
||||
};
|
||||
format!("select mr.provider_account_ref as account_id,{model} as model,
|
||||
jsonb_build_object({}) as counts,case when mr.cost_currency is not null and sum(mr.cost_amount) is not null
|
||||
then jsonb_build_object(mr.cost_currency,sum(mr.cost_amount)::text) else '{{}}'::jsonb end as costs,
|
||||
'{{}}'::jsonb as distributions,max(mr.started_at) as last_used_at from {table} mr",counters.join(","))
|
||||
jsonb_build_object({}) as counts,case when mr.cost_currency is not null and sum(mr.cost_amount) filter(where is_completed_usage) is not null
|
||||
then jsonb_build_object(mr.cost_currency,(sum(mr.cost_amount) filter(where is_completed_usage))::text) else '{{}}'::jsonb end as costs,
|
||||
'{{}}'::jsonb as distributions,max(mr.started_at) filter(where is_completed_usage) as last_used_at
|
||||
from (select mr.*,coalesce(({completed}),false) as is_completed_usage from {table} mr) mr",counters.join(","))
|
||||
}
|
||||
|
||||
pub(crate) async fn account_usage_windows(
|
||||
@@ -542,13 +590,11 @@ pub(crate) async fn account_usage_windows(
|
||||
continue;
|
||||
}
|
||||
total.merge(state)?;
|
||||
if !model.is_empty()
|
||||
&& let Some(last_used_at) = state.last_used_at()
|
||||
{
|
||||
if !model.is_empty() {
|
||||
models.push(AccountModelUsage {
|
||||
model: model.clone(),
|
||||
request_count: state.count("completed_request_count"),
|
||||
success_count: state.count("completed_request_count"),
|
||||
health: state.request_health(),
|
||||
input_tokens: state.optional_count("input_tokens"),
|
||||
output_tokens: state.optional_count("output_tokens"),
|
||||
cached_input_tokens: state.optional_count("cached_input_tokens"),
|
||||
@@ -561,7 +607,7 @@ pub(crate) async fn account_usage_windows(
|
||||
image_request_failed_count: state.count("image_request_failed_count"),
|
||||
cost_coverage: admin_cost_coverage(state.coverage()),
|
||||
costs: account_costs(state)?,
|
||||
last_used_at,
|
||||
last_used_at: state.last_used_at(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -320,6 +320,16 @@ impl MetricState {
|
||||
p99_ms: percentile(0.99)?,
|
||||
})
|
||||
}
|
||||
pub fn request_health(&self) -> admin_observability::RequestHealth {
|
||||
admin_observability::RequestHealth::from_counts(
|
||||
self.count("success_count"),
|
||||
self.count("failure_count"),
|
||||
self.count("cancelled_count"),
|
||||
self.count("incomplete_count"),
|
||||
self.count("caller_error_count"),
|
||||
)
|
||||
}
|
||||
|
||||
pub fn request_summary_metrics(&self) -> admin_observability::RequestSummaryMetrics {
|
||||
admin_observability::RequestSummaryMetrics {
|
||||
request_count: self.count("request_count"),
|
||||
|
||||
@@ -174,7 +174,7 @@ pub(crate) fn admin_account_model_usage(
|
||||
admin_observability::AccountModelUsage {
|
||||
model: usage.model,
|
||||
request_count: usage.request_count,
|
||||
success_count: usage.success_count,
|
||||
health: usage.health,
|
||||
input_tokens: usage.input_tokens,
|
||||
output_tokens: usage.output_tokens,
|
||||
cached_input_tokens: usage.cached_input_tokens,
|
||||
|
||||
@@ -252,7 +252,7 @@ pub struct ProviderAccountUsageObservation {
|
||||
pub struct ProviderAccountModelUsageObservation {
|
||||
pub model: String,
|
||||
pub request_count: u64,
|
||||
pub success_count: u64,
|
||||
pub health: admin_observability::RequestHealth,
|
||||
pub input_tokens: Option<u64>,
|
||||
pub output_tokens: Option<u64>,
|
||||
pub cached_input_tokens: Option<u64>,
|
||||
@@ -265,7 +265,7 @@ pub struct ProviderAccountModelUsageObservation {
|
||||
pub total_tokens: Option<u64>,
|
||||
pub cost_coverage: CostCoverage,
|
||||
pub costs: Vec<CurrencyCostTotal>,
|
||||
pub last_used_at: DateTime<Utc>,
|
||||
pub last_used_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
|
||||
@@ -42,14 +42,12 @@ pub(crate) async fn provider_account_usage(
|
||||
.await?
|
||||
{
|
||||
totals.entry(account.clone()).or_default().merge(&state)?;
|
||||
if !model.is_empty()
|
||||
&& let Some(last_used_at) = state.last_used_at()
|
||||
{
|
||||
if !model.is_empty() {
|
||||
models_by_account.entry(account).or_default().push(
|
||||
ProviderAccountModelUsageObservation {
|
||||
model,
|
||||
request_count: state.count("completed_request_count"),
|
||||
success_count: state.count("completed_request_count"),
|
||||
health: state.request_health(),
|
||||
input_tokens: state.optional_count("input_tokens"),
|
||||
output_tokens: state.optional_count("output_tokens"),
|
||||
cached_input_tokens: state.optional_count("cached_input_tokens"),
|
||||
@@ -62,7 +60,7 @@ pub(crate) async fn provider_account_usage(
|
||||
image_request_failed_count: state.count("image_request_failed_count"),
|
||||
cost_coverage: state.coverage(),
|
||||
costs: state.costs()?,
|
||||
last_used_at,
|
||||
last_used_at: state.last_used_at(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
@@ -148,3 +148,162 @@ async fn quota_windows_reconcile_moved_requests_and_pending_deletions_without_wa
|
||||
assert!(old.usage.costs.is_empty());
|
||||
database.close().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn model_success_rates_preserve_usage_across_source_summary_and_pending_reads() {
|
||||
let Some(database) = TestDatabase::create("account_model_health").await else {
|
||||
return;
|
||||
};
|
||||
PgProviderAccountRepository::new(database.pool.clone())
|
||||
.insert_provider_account(account("acct_health", "user-health"))
|
||||
.await
|
||||
.unwrap();
|
||||
let start = chrono::DateTime::from_timestamp(Utc::now().timestamp().div_euclid(3600) * 3600, 0)
|
||||
.unwrap()
|
||||
- TimeDelta::hours(3);
|
||||
for (index, (id, model, outcome, status)) in [
|
||||
("health_success", "mixed", "succeeded", 200),
|
||||
("health_retry", "mixed", "succeeded", 200),
|
||||
("health_failed", "mixed", "failed", 502),
|
||||
("health_incomplete", "mixed", "incomplete", 200),
|
||||
("health_cancelled", "mixed", "cancelled", 200),
|
||||
("health_caller", "mixed", "failed", 400),
|
||||
("health_recovered", "mixed", "failed", 502),
|
||||
("health_only_failed", "failed-only", "failed", 502),
|
||||
("health_only_cancelled", "no-sample", "cancelled", 200),
|
||||
("health_only_caller", "no-sample", "failed", 400),
|
||||
]
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
{
|
||||
seed_model_request(
|
||||
&database.pool,
|
||||
ModelRequestSeed {
|
||||
request_id: id,
|
||||
account_id: "acct_health",
|
||||
provider_kind: "openai",
|
||||
model,
|
||||
total_tokens: 11,
|
||||
cost_amount: "2.5",
|
||||
started_at: start + TimeDelta::minutes(5 + index as i64 * 5),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query("update model_requests set outcome=$2,client_status_code=$3,attempt_count=case when id='health_retry' then 3 else 1 end where id=$1")
|
||||
.bind(id).bind(outcome).bind(status).execute(&database.pool).await.unwrap();
|
||||
}
|
||||
sqlx::query("update model_requests set recovery_request_id='health_retry',recovered_at=completed_at,recovery_attempt_count=1,request_observation_json=jsonb_set(request_observation_json,'{recovery}',jsonb_build_object('retryDelayMs',0,'totalLatencyMs',1)) where id='health_recovered'")
|
||||
.execute(&database.pool).await.unwrap();
|
||||
let windows = [
|
||||
AccountUsageWindowQuery {
|
||||
account_id: "acct_health".to_owned(),
|
||||
key: "whole".to_owned(),
|
||||
range: TimeRange {
|
||||
start,
|
||||
end: start + TimeDelta::hours(2),
|
||||
},
|
||||
},
|
||||
AccountUsageWindowQuery {
|
||||
account_id: "acct_health".to_owned(),
|
||||
key: "boundary".to_owned(),
|
||||
range: TimeRange {
|
||||
start: start + TimeDelta::minutes(3),
|
||||
end: start + TimeDelta::minutes(117),
|
||||
},
|
||||
},
|
||||
];
|
||||
let store = admin_account_store(&database.pool);
|
||||
let source = store.load_account_usage_by_windows(&windows).await.unwrap();
|
||||
for result in &source {
|
||||
let usage = &result.usage;
|
||||
assert_eq!((usage.request_count, usage.total_tokens), (2, Some(22)));
|
||||
assert_eq!(usage.costs[0].amount.as_str(), "5");
|
||||
assert_eq!(usage.cost_coverage.provider_reported_count, 2);
|
||||
let mixed = usage
|
||||
.models
|
||||
.iter()
|
||||
.find(|model| model.model == "mixed")
|
||||
.unwrap();
|
||||
assert_eq!(mixed.health.success_rate(), Some(50.0));
|
||||
assert_eq!(mixed.health.success_requests, 2);
|
||||
assert_eq!(mixed.health.failed_requests, 1);
|
||||
assert_eq!(mixed.health.incomplete_requests, 1);
|
||||
let failed = usage
|
||||
.models
|
||||
.iter()
|
||||
.find(|model| model.model == "failed-only")
|
||||
.unwrap();
|
||||
assert_eq!(failed.health.success_rate(), Some(0.0));
|
||||
assert_eq!(failed.last_used_at, None);
|
||||
assert!(failed.costs.is_empty());
|
||||
let excluded = usage
|
||||
.models
|
||||
.iter()
|
||||
.find(|model| model.model == "no-sample")
|
||||
.unwrap();
|
||||
assert_eq!(excluded.health.success_rate(), None);
|
||||
}
|
||||
drain_analytics(&database.pool).await;
|
||||
assert_eq!(
|
||||
store.load_account_usage_by_windows(&windows).await.unwrap(),
|
||||
source
|
||||
);
|
||||
let rolling = store
|
||||
.load_account_usage(windows[0].range, &["acct_health".to_owned()])
|
||||
.await
|
||||
.unwrap();
|
||||
for model in &rolling[0].models {
|
||||
assert_eq!(
|
||||
model.health,
|
||||
source[0]
|
||||
.usage
|
||||
.models
|
||||
.iter()
|
||||
.find(|expected| expected.model == model.model)
|
||||
.unwrap()
|
||||
.health
|
||||
);
|
||||
}
|
||||
|
||||
// 待投影的成功变失败必须撤销费用与最近使用时间,失败变成功则补入当前源版本
|
||||
sqlx::query("update model_requests set outcome='failed',client_status_code=502 where id in ('health_success','health_retry')")
|
||||
.execute(&database.pool).await.unwrap();
|
||||
sqlx::query("update model_requests set outcome='succeeded',client_status_code=200 where id='health_only_failed'")
|
||||
.execute(&database.pool).await.unwrap();
|
||||
sqlx::query("delete from model_requests where id='health_failed'")
|
||||
.execute(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let pending = store.load_account_usage_by_windows(&windows).await.unwrap();
|
||||
for result in &pending {
|
||||
assert_eq!(
|
||||
(result.usage.request_count, result.usage.total_tokens),
|
||||
(1, Some(11))
|
||||
);
|
||||
assert_eq!(result.usage.costs[0].amount.as_str(), "2.5");
|
||||
let mixed = result
|
||||
.usage
|
||||
.models
|
||||
.iter()
|
||||
.find(|model| model.model == "mixed")
|
||||
.unwrap();
|
||||
assert_eq!(mixed.health.success_rate(), Some(0.0));
|
||||
assert_eq!(mixed.health.failed_requests, 2);
|
||||
assert_eq!(mixed.last_used_at, None);
|
||||
assert!(mixed.costs.is_empty());
|
||||
let recovered = result
|
||||
.usage
|
||||
.models
|
||||
.iter()
|
||||
.find(|model| model.model == "failed-only")
|
||||
.unwrap();
|
||||
assert_eq!(recovered.health.success_rate(), Some(100.0));
|
||||
}
|
||||
drain_analytics(&database.pool).await;
|
||||
assert_eq!(
|
||||
store.load_account_usage_by_windows(&windows).await.unwrap(),
|
||||
pending
|
||||
);
|
||||
database.close().await;
|
||||
}
|
||||
|
||||
@@ -1100,6 +1100,12 @@ config 返回 `{ name, plaintextKey }`,仅读取服务端会话绑定的当前
|
||||
|
||||
账号列表的 `search` 仅按字面量包含匹配名称和邮箱,不区分大小写;`%`、`_` 和 `\` 按普通字符处理
|
||||
|
||||
账号模型用量的 `successRate` 按同一账号、模型和用量窗口内的[服务成功率](#记录范围与统计口径)计算,
|
||||
范围为 0–100,无有效样本时为 `null`,`successRateDisplay` 显示 `—`
|
||||
|
||||
`requestCount`、Token、费用及 `lastUsedAt` 只统计完整交付的成功推理响应;
|
||||
仅有失败记录的模型也返回统计行,没有成功用量时 `lastUsedAt` 为 `null`
|
||||
|
||||
账号视图的 `capacity` 返回查询时的普通并发容量:`usedSlots` 不含独立审批池和排队请求,
|
||||
读取运行态失败时为 `null`。`totalSlots` 是按[并发与排队](#并发与排队)取值的本机上限,`null` 表示不限。上游独立执行其密钥限制
|
||||
|
||||
|
||||
@@ -92,7 +92,7 @@ export interface AccountModelUsage {
|
||||
partialCostCount: number
|
||||
unknownCostCount: number
|
||||
costs: AccountCurrencyCost[]
|
||||
lastUsedAt: string
|
||||
lastUsedAt: string | null
|
||||
lastUsedAtDisplay: string
|
||||
lastUsedAtFullDisplay: string | null
|
||||
}
|
||||
|
||||
@@ -44,6 +44,8 @@ function latestModelUsage(models: readonly AccountModelUsage[]) {
|
||||
let latestTimestamp = Number.NEGATIVE_INFINITY
|
||||
|
||||
for (const model of models) {
|
||||
if (!model.lastUsedAt)
|
||||
continue
|
||||
const timestamp = Date.parse(model.lastUsedAt)
|
||||
if (!Number.isFinite(timestamp) || timestamp <= latestTimestamp)
|
||||
continue
|
||||
|
||||
Reference in New Issue
Block a user