perf(observability): narrow usage and ops read models

Separate Usage list and detail reads, use keyset paging with first-page totals, push Ops predicates into source branches, and merge repeated account and diagnostics aggregates. Apply the shared per-query PostgreSQL budget, coalesce identical Dashboard windows, and preserve the existing IP, User-Agent, and latency detail surface.

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