fix(observability): separate live status from history range

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-08-13 00:26:10 +08:00
co-authored by Codex
parent 6cad2bb6a1
commit 986a128d15
8 changed files with 75 additions and 15 deletions
@@ -218,7 +218,12 @@ pub trait ClientKeyStore: Send + Sync {
/// 用量、趋势、诊断与运维错误的只读能力。
#[async_trait]
pub trait ObservabilityStore: Send + Sync {
async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult<DashboardObservation>;
/// 返回历史统计区间和指定观测时刻下的实时账号状态。
async fn dashboard_summary(
&self,
range: TimeRange,
observed_at: DateTime<Utc>,
) -> AdminStoreResult<DashboardObservation>;
/// 返回 Dashboard 可选的实时槽位事实。
///
@@ -98,10 +98,11 @@ impl ObservabilityService for DefaultObservabilityService {
range: TimeRange,
kind: TrendKind,
) -> Result<DashboardResult, AdminError> {
let observed_at = Utc::now();
let (mut observation, settings, runtime_slots) = futures::try_join!(
self.store.dashboard_summary(range),
self.store.dashboard_summary(range, observed_at),
self.settings.load_runtime_settings(),
self.store.dashboard_runtime_slots(range.end),
self.store.dashboard_runtime_slots(observed_at),
)
.map_err(|error| map_store_error(error, "dashboard"))?;
self.enrich_billing(&mut observation.recent_requests);
@@ -14,6 +14,7 @@ use std::{
};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures::future::BoxFuture;
use gateway_admin::{
AdminConfig, AdminServices, InitialAdminPassword,
@@ -391,7 +392,11 @@ impl ClientKeyStore for UnavailableStore {
#[async_trait]
impl ObservabilityStore for UnavailableStore {
async fn dashboard_summary(&self, _: TimeRange) -> AdminStoreResult<DashboardObservation> {
async fn dashboard_summary(
&self,
_: TimeRange,
_: DateTime<Utc>,
) -> AdminStoreResult<DashboardObservation> {
Err(unavailable("dashboard"))
}
@@ -200,6 +200,25 @@ async fn dashboard_summary_should_project_rebuildable_runtime_slots() {
assert_eq!(capacity.available_slots, Some(1));
}
#[tokio::test]
async fn dashboard_summary_should_share_one_current_observation_time_across_runtime_facts() {
let historical_end = Utc::now() - Duration::hours(1);
let store = Arc::new(FixtureObservabilityStore::new(observation_range(
historical_end,
)));
let services = observability_services(store.clone()).await;
services
.observability()
.dashboard_summary(observation_range(historical_end), TrendKind::Usage)
.await
.expect("dashboard summary");
let (summary_observed_at, slots_observed_at) = store.observed_times();
assert_eq!(summary_observed_at, slots_observed_at);
assert!(summary_observed_at.is_some_and(|value| value > historical_end));
}
#[tokio::test]
async fn observability_services_should_calculate_usage_insights_and_diagnostic_shares() {
let now = Utc::now();
@@ -394,6 +413,8 @@ struct FixtureObservabilityStore {
calculated_billing_facts: Mutex<Vec<UsageCalculatedBillingFact>>,
diagnostics: Mutex<Vec<DiagnosticObservation>>,
runtime_slots: Mutex<Option<DashboardRuntimeSlots>>,
summary_observed_at: Mutex<Option<DateTime<Utc>>>,
slots_observed_at: Mutex<Option<DateTime<Utc>>>,
usage_records: Mutex<Vec<UsageRecord>>,
}
@@ -410,6 +431,8 @@ impl FixtureObservabilityStore {
calculated_billing_facts: Mutex::new(Vec::new()),
diagnostics: Mutex::new(Vec::new()),
runtime_slots: Mutex::new(None),
summary_observed_at: Mutex::new(None),
slots_observed_at: Mutex::new(None),
usage_records: Mutex::new(Vec::new()),
}
}
@@ -437,6 +460,16 @@ impl FixtureObservabilityStore {
*self.runtime_slots.lock().expect("runtime slots") = runtime_slots;
}
fn observed_times(&self) -> (Option<DateTime<Utc>>, Option<DateTime<Utc>>) {
(
*self
.summary_observed_at
.lock()
.expect("summary observed at"),
*self.slots_observed_at.lock().expect("slots observed at"),
)
}
fn replace_usage_records(&self, records: Vec<UsageRecord>) {
*self.usage_records.lock().expect("usage records") = records;
}
@@ -444,7 +477,15 @@ impl FixtureObservabilityStore {
#[async_trait]
impl ObservabilityStore for FixtureObservabilityStore {
async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult<DashboardObservation> {
async fn dashboard_summary(
&self,
range: TimeRange,
observed_at: DateTime<Utc>,
) -> AdminStoreResult<DashboardObservation> {
*self
.summary_observed_at
.lock()
.expect("summary observed at") = Some(observed_at);
Ok(DashboardObservation {
range,
requests: RequestMetrics::default(),
@@ -459,8 +500,9 @@ impl ObservabilityStore for FixtureObservabilityStore {
async fn dashboard_runtime_slots(
&self,
_: DateTime<Utc>,
observed_at: DateTime<Utc>,
) -> AdminStoreResult<Option<DashboardRuntimeSlots>> {
*self.slots_observed_at.lock().expect("slots observed at") = Some(observed_at);
Ok(*self.runtime_slots.lock().expect("runtime slots"))
}
@@ -7,7 +7,7 @@ use std::{
};
use async_trait::async_trait;
use chrono::{Duration, Utc};
use chrono::{DateTime, Duration, Utc};
use futures::future::BoxFuture;
use gateway_admin::{
AdminConfig, AdminServices, InitialAdminPassword,
@@ -528,7 +528,11 @@ impl AccountStore for UnusedStore {
#[async_trait]
impl ObservabilityStore for UnusedStore {
async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult<DashboardObservation> {
async fn dashboard_summary(
&self,
range: TimeRange,
_: DateTime<Utc>,
) -> AdminStoreResult<DashboardObservation> {
*self
.dashboard_summary_range
.lock()
@@ -98,6 +98,7 @@ impl ObservabilityRepository for PgObservabilityRepository {
async fn dashboard_summary(
&self,
range: ObservabilityRange,
observed_at: DateTime<Utc>,
) -> StoreResult<DashboardObservation> {
let filter = UsageRecordFilter::default();
let account_usage_range = ObservabilityRange::new(
@@ -123,7 +124,7 @@ impl ObservabilityRepository for PgObservabilityRepository {
attempt_metrics(&self.pool, range, &filter),
dashboard_totals(&self.pool),
)?;
let (provider_accounts, _) = self.account_status_snapshot(range.end).await?;
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),
@@ -215,10 +216,11 @@ impl AdminObservabilityStore for PgAdminObservabilityStore {
async fn dashboard_summary(
&self,
range: admin_observability::TimeRange,
observed_at: DateTime<Utc>,
) -> AdminStoreResult<admin_observability::DashboardObservation> {
let observation = self
.repository
.dashboard_summary(store_range(range)?)
.dashboard_summary(store_range(range)?, observed_at)
.await
.map_err(observability_error)?;
admin_dashboard_observation(observation)
@@ -683,6 +683,7 @@ pub trait ObservabilityRepository: Send + Sync {
async fn dashboard_summary(
&self,
range: ObservabilityRange,
observed_at: DateTime<Utc>,
) -> StoreResult<DashboardObservation>;
async fn dashboard_trend(
&self,
@@ -166,7 +166,7 @@ async fn dashboard_account_metrics_should_partition_account_statuses() {
.expect("dashboard range");
let metrics = repository
.dashboard_summary(range)
.dashboard_summary(range, now)
.await
.expect("dashboard summary")
.provider_accounts;
@@ -252,7 +252,7 @@ async fn dashboard_account_metrics_with_cooldowns_should_only_reclassify_eligibl
.expect("dashboard range");
let metrics = repository
.dashboard_summary(range)
.dashboard_summary(range, now)
.await
.expect("dashboard summary")
.provider_accounts;
@@ -416,7 +416,7 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det
let store = PgAdminObservabilityStore::new(database.pool.clone(), None, None);
let dashboard = store
.dashboard_summary(range)
.dashboard_summary(range, now)
.await
.expect("admin dashboard summary");
assert_eq!(dashboard.range, range);
@@ -714,7 +714,7 @@ async fn dashboard_summary_totals_include_history_outside_selected_range() {
let repository = PgObservabilityRepository::new(database.pool.clone(), None);
let dashboard = repository
.dashboard_summary(range)
.dashboard_summary(range, now)
.await
.expect("dashboard summary");
assert_eq!(dashboard.requests.request_count, 2);
@@ -753,7 +753,7 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact
let repository = PgObservabilityRepository::new(database.pool.clone(), None);
let dashboard = repository
.dashboard_summary(range)
.dashboard_summary(range, now)
.await
.expect("dashboard summary");
assert_eq!(dashboard.requests.request_count, 3);