i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-01 17:00:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: improve admin performance analysis

Browse Source
This commit is contained in:
fawney19
2026-05-06 02:01:32 +08:00
parent f1358dd845
commit cf27773582
9 changed files with 833 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
crates/aether-admin/src/observability/stats.rs
View File

@@ -1110,9 +1110,13 @@ pub fn build_admin_stats_provider_performance_response(
"avg_first_byte_time_ms": rounded_option(row.avg_first_byte_time_ms, 2),
"avg_response_time_ms": rounded_option(row.avg_response_time_ms, 2),
"p90_response_time_ms": row.p90_response_time_ms,
"p99_response_time_ms": row.p99_response_time_ms,
"p90_first_byte_time_ms": row.p90_first_byte_time_ms,
"p99_first_byte_time_ms": row.p99_first_byte_time_ms,
"tps_sample_count": row.tps_sample_count,
"response_time_sample_count": row.response_time_sample_count,
"first_byte_sample_count": row.first_byte_sample_count,
"slow_request_count": row.slow_request_count,
})
})
.collect::<Vec<_>>();
@@ -1129,6 +1133,7 @@ pub fn build_admin_stats_provider_performance_response(
"avg_output_tps": rounded_option(row.avg_output_tps, 2),
"avg_first_byte_time_ms": rounded_option(row.avg_first_byte_time_ms, 2),
"avg_response_time_ms": rounded_option(row.avg_response_time_ms, 2),
"slow_request_count": row.slow_request_count,
"success_rate": success_rate(row.request_count, row.success_count),
})
})
@@ -1141,6 +1146,14 @@ pub fn build_admin_stats_provider_performance_response(
"avg_output_tps": rounded_option(summary.avg_output_tps, 2),
"avg_first_byte_time_ms": rounded_option(summary.avg_first_byte_time_ms, 2),
"avg_response_time_ms": rounded_option(summary.avg_response_time_ms, 2),
"p90_response_time_ms": summary.p90_response_time_ms,
"p99_response_time_ms": summary.p99_response_time_ms,
"p90_first_byte_time_ms": summary.p90_first_byte_time_ms,
"p99_first_byte_time_ms": summary.p99_first_byte_time_ms,
"tps_sample_count": summary.tps_sample_count,
"response_time_sample_count": summary.response_time_sample_count,
"first_byte_sample_count": summary.first_byte_sample_count,
"slow_request_count": summary.slow_request_count,
},
"providers": providers,
"timeline": timeline,

20
crates/aether-data-contracts/src/repository/usage/types.rs
View File

@@ -957,6 +957,13 @@ pub struct UsageProviderPerformanceQuery {
pub granularity: UsageTimeSeriesGranularity,
pub tz_offset_minutes: i32,
pub limit: usize,
pub provider_id: Option<String>,
pub model: Option<String>,
pub api_format: Option<String>,
pub endpoint_kind: Option<String>,
pub is_stream: Option<bool>,
pub has_format_conversion: Option<bool>,
pub slow_threshold_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Default, serde::Serialize, serde::Deserialize)]
@@ -966,6 +973,14 @@ pub struct StoredUsageProviderPerformanceSummary {
pub avg_output_tps: Option<f64>,
pub avg_first_byte_time_ms: Option<f64>,
pub avg_response_time_ms: Option<f64>,
pub p90_response_time_ms: Option<u64>,
pub p99_response_time_ms: Option<u64>,
pub p90_first_byte_time_ms: Option<u64>,
pub p99_first_byte_time_ms: Option<u64>,
pub tps_sample_count: u64,
pub response_time_sample_count: u64,
pub first_byte_sample_count: u64,
pub slow_request_count: u64,
}
#[derive(Debug, Clone, PartialEq, Default, serde::Serialize, serde::Deserialize)]
@@ -979,9 +994,13 @@ pub struct StoredUsageProviderPerformanceProviderRow {
pub avg_first_byte_time_ms: Option<f64>,
pub avg_response_time_ms: Option<f64>,
pub p90_response_time_ms: Option<u64>,
pub p99_response_time_ms: Option<u64>,
pub p90_first_byte_time_ms: Option<u64>,
pub p99_first_byte_time_ms: Option<u64>,
pub tps_sample_count: u64,
pub response_time_sample_count: u64,
pub first_byte_sample_count: u64,
pub slow_request_count: u64,
}
#[derive(Debug, Clone, PartialEq, Default, serde::Serialize, serde::Deserialize)]
@@ -995,6 +1014,7 @@ pub struct StoredUsageProviderPerformanceTimelineRow {
pub avg_output_tps: Option<f64>,
pub avg_first_byte_time_ms: Option<f64>,
pub avg_response_time_ms: Option<f64>,
pub slow_request_count: u64,
}
#[derive(Debug, Clone, PartialEq, Default, serde::Serialize, serde::Deserialize)]

70
crates/aether-data/src/repository/usage/memory.rs
View File

@@ -618,6 +618,36 @@ fn usage_matches_provider_performance_query(
{
return None;
}
if let Some(provider_id) = query.provider_id.as_deref() {
if item.provider_id.as_deref() != Some(provider_id) {
return None;
}
}
if let Some(model) = query.model.as_deref() {
if item.model != model {
return None;
}
}
if let Some(api_format) = query.api_format.as_deref() {
if item.api_format.as_deref() != Some(api_format) {
return None;
}
}
if let Some(endpoint_kind) = query.endpoint_kind.as_deref() {
if item.endpoint_kind.as_deref() != Some(endpoint_kind) {
return None;
}
}
if let Some(is_stream) = query.is_stream {
if item.is_stream != is_stream {
return None;
}
}
if let Some(has_format_conversion) = query.has_format_conversion {
if item.has_format_conversion != has_format_conversion {
return None;
}
}
usage_provider_performance_identity(item)
}
@@ -1703,12 +1733,19 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
response_time_sample_count: u64,
response_times: Vec<u64>,
first_byte_times: Vec<u64>,
slow_request_count: u64,
}
impl ProviderPerformanceBucket {
fn add(&mut self, item: &StoredRequestUsageAudit) {
fn add(&mut self, item: &StoredRequestUsageAudit, slow_threshold_ms: u64) {
self.request_count = self.request_count.saturating_add(1);
self.output_tokens = self.output_tokens.saturating_add(item.output_tokens);
if item
.response_time_ms
.is_some_and(|value| value >= slow_threshold_ms)
{
self.slow_request_count = self.slow_request_count.saturating_add(1);
}
if !usage_is_success(item) {
return;
}
@@ -1771,14 +1808,16 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
else {
continue;
};
summary_bucket.add(item);
summary_bucket.add(item, query.slow_threshold_ms);
let bucket = grouped.entry(provider_id).or_default();
if bucket.provider.is_empty() {
bucket.provider = provider;
}
bucket.add(item);
bucket.add(item, query.slow_threshold_ms);
}
let mut summary_response_times = summary_bucket.response_times.clone();
let mut summary_first_byte_times = summary_bucket.first_byte_times.clone();
let summary = StoredUsageProviderPerformanceSummary {
request_count: summary_bucket.request_count,
success_count: summary_bucket.success_count,
@@ -1794,14 +1833,25 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
summary_bucket.response_time_ms_sum,
summary_bucket.response_time_sample_count,
),
p90_response_time_ms: usage_percentile_cont(&mut summary_response_times, 0.9),
p99_response_time_ms: usage_percentile_cont(&mut summary_response_times, 0.99),
p90_first_byte_time_ms: usage_percentile_cont(&mut summary_first_byte_times, 0.9),
p99_first_byte_time_ms: usage_percentile_cont(&mut summary_first_byte_times, 0.99),
tps_sample_count: summary_bucket.tps_sample_count,
response_time_sample_count: summary_bucket.response_time_sample_count,
first_byte_sample_count: summary_bucket.first_byte_sample_count,
slow_request_count: summary_bucket.slow_request_count,
};
let mut providers = grouped
.into_iter()
.map(|(provider_id, mut bucket)| {
let p90_response_time_ms = usage_percentile_cont(&mut bucket.response_times, 0.9);
let p99_response_time_ms = usage_percentile_cont(&mut bucket.response_times, 0.99);
let p90_first_byte_time_ms =
usage_percentile_cont(&mut bucket.first_byte_times, 0.9);
let p99_first_byte_time_ms =
usage_percentile_cont(&mut bucket.first_byte_times, 0.99);
StoredUsageProviderPerformanceProviderRow {
provider_id,
provider: bucket.provider,
@@ -1821,9 +1871,13 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
bucket.response_time_sample_count,
),
p90_response_time_ms,
p99_response_time_ms,
p90_first_byte_time_ms,
p99_first_byte_time_ms,
tps_sample_count: bucket.tps_sample_count,
response_time_sample_count: bucket.response_time_sample_count,
first_byte_sample_count: bucket.first_byte_sample_count,
slow_request_count: bucket.slow_request_count,
}
})
.collect::<Vec<_>>();
@@ -1860,7 +1914,7 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
if bucket.provider.is_empty() {
bucket.provider = provider;
}
bucket.add(item);
bucket.add(item, query.slow_threshold_ms);
}
let timeline = timeline_grouped
@@ -1885,6 +1939,7 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
bucket.response_time_ms_sum,
bucket.response_time_sample_count,
),
slow_request_count: bucket.slow_request_count,
},
)
.collect();
@@ -4846,6 +4901,13 @@ mod tests {
granularity: UsageTimeSeriesGranularity::Hour,
tz_offset_minutes: 0,
limit: 1,
provider_id: None,
model: None,
api_format: None,
endpoint_kind: None,
is_stream: None,
has_format_conversion: None,
slow_threshold_ms: 10_000,
})
.await
.expect("provider performance should summarize");

198
crates/aether-data/src/repository/usage/postgres/mod.rs
View File

@@ -829,6 +829,38 @@ fn decode_usage_provider_performance_summary(
avg_response_time_ms: row
.try_get::<Option<f64>, _>("avg_response_time_ms")
.map_postgres_err()?,
p90_response_time_ms: row
.try_get::<Option<i64>, _>("p90_response_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
p99_response_time_ms: row
.try_get::<Option<i64>, _>("p99_response_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
p90_first_byte_time_ms: row
.try_get::<Option<i64>, _>("p90_first_byte_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
p99_first_byte_time_ms: row
.try_get::<Option<i64>, _>("p99_first_byte_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
tps_sample_count: row
.try_get::<i64, _>("tps_sample_count")
.map_postgres_err()?
.max(0) as u64,
response_time_sample_count: row
.try_get::<i64, _>("response_time_sample_count")
.map_postgres_err()?
.max(0) as u64,
first_byte_sample_count: row
.try_get::<i64, _>("first_byte_sample_count")
.map_postgres_err()?
.max(0) as u64,
slow_request_count: row
.try_get::<i64, _>("slow_request_count")
.map_postgres_err()?
.max(0) as u64,
})
}
@@ -863,18 +895,34 @@ fn decode_usage_provider_performance_provider_row(
.try_get::<Option<i64>, _>("p90_response_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
p99_response_time_ms: row
.try_get::<Option<i64>, _>("p99_response_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
p90_first_byte_time_ms: row
.try_get::<Option<i64>, _>("p90_first_byte_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
p99_first_byte_time_ms: row
.try_get::<Option<i64>, _>("p99_first_byte_time_ms")
.map_postgres_err()?
.map(|value| value.max(0) as u64),
tps_sample_count: row
.try_get::<i64, _>("tps_sample_count")
.map_postgres_err()?
.max(0) as u64,
response_time_sample_count: row
.try_get::<i64, _>("response_time_sample_count")
.map_postgres_err()?
.max(0) as u64,
first_byte_sample_count: row
.try_get::<i64, _>("first_byte_sample_count")
.map_postgres_err()?
.max(0) as u64,
slow_request_count: row
.try_get::<i64, _>("slow_request_count")
.map_postgres_err()?
.max(0) as u64,
})
}
@@ -906,9 +954,64 @@ fn decode_usage_provider_performance_timeline_row(
avg_response_time_ms: row
.try_get::<Option<f64>, _>("avg_response_time_ms")
.map_postgres_err()?,
slow_request_count: row
.try_get::<i64, _>("slow_request_count")
.map_postgres_err()?
.max(0) as u64,
})
}
fn push_usage_provider_performance_text_filter(
builder: &mut QueryBuilder<'_, Postgres>,
column: &'static str,
value: &Option<String>,
) {
let Some(value) = value
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
else {
return;
};
builder
.push(" AND NULLIF(BTRIM(COALESCE(")
.push(column)
.push(", '')), '') = ")
.push_bind(value.to_string());
}
fn push_usage_provider_performance_filters(
builder: &mut QueryBuilder<'_, Postgres>,
query: &UsageProviderPerformanceQuery,
) {
push_usage_provider_performance_text_filter(
builder,
r#""usage".provider_id"#,
&query.provider_id,
);
push_usage_provider_performance_text_filter(builder, r#""usage".model"#, &query.model);
push_usage_provider_performance_text_filter(
builder,
r#""usage".api_format"#,
&query.api_format,
);
push_usage_provider_performance_text_filter(
builder,
r#""usage".endpoint_kind"#,
&query.endpoint_kind,
);
if let Some(is_stream) = query.is_stream {
builder
.push(r#" AND "usage".is_stream = "#)
.push_bind(is_stream);
}
if let Some(has_format_conversion) = query.has_format_conversion {
builder
.push(r#" AND "usage".has_format_conversion = "#)
.push_bind(has_format_conversion);
}
}
fn decode_usage_time_series_bucket_row(
row: &PgRow,
) -> Result<StoredUsageTimeSeriesBucket, DataLayerError> {
@@ -4604,6 +4707,11 @@ WITH filtered_usage AS (
AND NULLIF(BTRIM(COALESCE("usage".provider_id, '')), '') IS NOT NULL
AND lower(BTRIM(COALESCE("usage".provider_id, ''))) NOT IN ('unknown', 'pending')
AND lower(BTRIM(COALESCE("usage".provider_name, ''))) NOT IN ('unknown', 'pending')
"#,
);
push_usage_provider_performance_filters(&mut builder, query);
builder.push(
r#"
)
SELECT
COUNT(*)::BIGINT AS request_count,
@@ -4628,7 +4736,49 @@ SELECT
AVG(first_byte_time_ms::DOUBLE PRECISION)
FILTER (WHERE success_flag = 1 AND has_first_byte_time) AS avg_first_byte_time_ms,
AVG(response_time_ms::DOUBLE PRECISION)
FILTER (WHERE success_flag = 1 AND has_response_time) AS avg_response_time_ms
FILTER (WHERE success_flag = 1 AND has_response_time) AS avg_response_time_ms,
CASE
WHEN COUNT(response_time_ms) FILTER (WHERE success_flag = 1 AND has_response_time) >= 10
THEN FLOOR(PERCENTILE_CONT(0.9) WITHIN GROUP (ORDER BY response_time_ms)
FILTER (WHERE success_flag = 1 AND has_response_time))::BIGINT
ELSE NULL
END AS p90_response_time_ms,
CASE
WHEN COUNT(response_time_ms) FILTER (WHERE success_flag = 1 AND has_response_time) >= 10
THEN FLOOR(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY response_time_ms)
FILTER (WHERE success_flag = 1 AND has_response_time))::BIGINT
ELSE NULL
END AS p99_response_time_ms,
CASE
WHEN COUNT(first_byte_time_ms) FILTER (WHERE success_flag = 1 AND has_first_byte_time) >= 10
THEN FLOOR(PERCENTILE_CONT(0.9) WITHIN GROUP (ORDER BY first_byte_time_ms)
FILTER (WHERE success_flag = 1 AND has_first_byte_time))::BIGINT
ELSE NULL
END AS p90_first_byte_time_ms,
CASE
WHEN COUNT(first_byte_time_ms) FILTER (WHERE success_flag = 1 AND has_first_byte_time) >= 10
THEN FLOOR(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY first_byte_time_ms)
FILTER (WHERE success_flag = 1 AND has_first_byte_time))::BIGINT
ELSE NULL
END AS p99_first_byte_time_ms,
COALESCE(SUM(CASE
WHEN success_flag = 1 AND response_time_ms > 0 AND output_tokens > 0
THEN 1
ELSE 0
END), 0)::BIGINT AS tps_sample_count,
(COUNT(response_time_ms) FILTER (WHERE success_flag = 1 AND has_response_time))::BIGINT
AS response_time_sample_count,
(COUNT(first_byte_time_ms) FILTER (WHERE success_flag = 1 AND has_first_byte_time))::BIGINT
AS first_byte_sample_count,
COALESCE(SUM(CASE
WHEN has_response_time AND response_time_ms >= "#,
);
builder.push_bind(query.slow_threshold_ms as i64);
builder.push(
r#"
THEN 1
ELSE 0
END), 0)::BIGINT AS slow_request_count
FROM filtered_usage
"#,
);
@@ -4677,6 +4827,11 @@ WITH filtered_usage AS (
AND NULLIF(BTRIM(COALESCE("usage".provider_id, '')), '') IS NOT NULL
AND lower(BTRIM(COALESCE("usage".provider_id, ''))) NOT IN ('unknown', 'pending')
AND lower(BTRIM(COALESCE("usage".provider_name, ''))) NOT IN ('unknown', 'pending')
"#,
);
push_usage_provider_performance_filters(&mut builder, query);
builder.push(
r#"
)
SELECT
provider_id,
@@ -4711,19 +4866,42 @@ SELECT
FILTER (WHERE success_flag = 1 AND has_response_time))::BIGINT
ELSE NULL
END AS p90_response_time_ms,
CASE
WHEN COUNT(response_time_ms) FILTER (WHERE success_flag = 1 AND has_response_time) >= 10
THEN FLOOR(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY response_time_ms)
FILTER (WHERE success_flag = 1 AND has_response_time))::BIGINT
ELSE NULL
END AS p99_response_time_ms,
CASE
WHEN COUNT(first_byte_time_ms) FILTER (WHERE success_flag = 1 AND has_first_byte_time) >= 10
THEN FLOOR(PERCENTILE_CONT(0.9) WITHIN GROUP (ORDER BY first_byte_time_ms)
FILTER (WHERE success_flag = 1 AND has_first_byte_time))::BIGINT
ELSE NULL
END AS p90_first_byte_time_ms,
CASE
WHEN COUNT(first_byte_time_ms) FILTER (WHERE success_flag = 1 AND has_first_byte_time) >= 10
THEN FLOOR(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY first_byte_time_ms)
FILTER (WHERE success_flag = 1 AND has_first_byte_time))::BIGINT
ELSE NULL
END AS p99_first_byte_time_ms,
COALESCE(SUM(CASE
WHEN success_flag = 1 AND response_time_ms > 0 AND output_tokens > 0
THEN 1
ELSE 0
END), 0)::BIGINT AS tps_sample_count,
(COUNT(response_time_ms) FILTER (WHERE success_flag = 1 AND has_response_time))::BIGINT
AS response_time_sample_count,
(COUNT(first_byte_time_ms) FILTER (WHERE success_flag = 1 AND has_first_byte_time))::BIGINT
AS first_byte_sample_count
AS first_byte_sample_count,
COALESCE(SUM(CASE
WHEN has_response_time AND response_time_ms >= "#,
);
builder.push_bind(query.slow_threshold_ms as i64);
builder.push(
r#"
THEN 1
ELSE 0
END), 0)::BIGINT AS slow_request_count
FROM filtered_usage
GROUP BY provider_id
ORDER BY request_count DESC, provider_id ASC
@@ -4792,6 +4970,11 @@ ORDER BY request_count DESC, provider_id ASC
AND NULLIF(BTRIM(COALESCE("usage".provider_id, '')), '') IS NOT NULL
AND lower(BTRIM(COALESCE("usage".provider_id, ''))) NOT IN ('unknown', 'pending')
AND lower(BTRIM(COALESCE("usage".provider_name, ''))) NOT IN ('unknown', 'pending')
"#,
);
push_usage_provider_performance_filters(&mut builder, query);
builder.push(
r#"
AND "usage".provider_id = ANY("#,
);
builder.push_bind(provider_ids.to_vec());
@@ -4825,7 +5008,16 @@ SELECT
AVG(first_byte_time_ms::DOUBLE PRECISION)
FILTER (WHERE success_flag = 1 AND has_first_byte_time) AS avg_first_byte_time_ms,
AVG(response_time_ms::DOUBLE PRECISION)
FILTER (WHERE success_flag = 1 AND has_response_time) AS avg_response_time_ms
FILTER (WHERE success_flag = 1 AND has_response_time) AS avg_response_time_ms,
COALESCE(SUM(CASE
WHEN has_response_time AND response_time_ms >= "#,
);
builder.push_bind(query.slow_threshold_ms as i64);
builder.push(
r#"
THEN 1
ELSE 0
END), 0)::BIGINT AS slow_request_count
FROM filtered_usage
GROUP BY date, provider_id
ORDER BY date ASC, provider_id ASC