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

Expose ranking metadata in request trace

Browse Source
This commit is contained in:
fawney19
2026-04-27 17:16:30 +08:00
parent 3e0293ac28
commit 17f09fc8c1
5 changed files with 191 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
apps/aether-gateway/src/handlers/admin/observability/monitoring/tests/trace.rs
View File

@@ -358,6 +358,66 @@ async fn admin_monitoring_trace_request_enriches_proxy_timing_from_usage_audit()
);
}
#[tokio::test]
async fn admin_monitoring_trace_request_exposes_structured_ranking_metadata() {
let mut candidate = sample_candidate(
"cand-used",
"request-1",
0,
RequestCandidateStatus::Success,
Some(101),
Some(33),
Some(200),
);
candidate.extra_data = Some(json!({
"ranking_mode": "CacheAffinity",
"priority_mode": "Provider",
"ranking_index": 0,
"priority_slot": 7,
"promoted_by": "cached_affinity",
"demoted_by": "cross_format",
"client_api_format": "openai:responses"
}));
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![candidate]));
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider()],
vec![sample_endpoint()],
vec![sample_key()],
));
let state = AppState::new()
.expect("state should build")
.with_decision_trace_data_readers_for_tests(request_candidates, provider_catalog);
let context = request_context(http::Method::GET, "/api/admin/monitoring/trace/request-1");
let response = local_monitoring_response(&state, &context)
.await
.expect("handler should not error")
.expect("route should be handled locally");
assert_eq!(response.status(), http::StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body should read");
let payload: serde_json::Value = serde_json::from_slice(&body).expect("json body should parse");
assert_eq!(
payload["candidates"][0]["ranking"],
json!({
"mode": "CacheAffinity",
"priority_mode": "Provider",
"index": 0,
"priority_slot": 7,
"promoted_by": "cached_affinity",
"demoted_by": "cross_format"
})
);
assert_eq!(
payload["candidates"][0]["extra_data"]["ranking_mode"],
json!("CacheAffinity")
);
}
#[tokio::test]
async fn admin_monitoring_trace_provider_stats_returns_local_payload() {
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![

61
crates/aether-admin/src/observability/monitoring.rs
View File

@@ -365,6 +365,7 @@ pub fn build_admin_monitoring_trace_request_candidate_payload_with_key_accounts(
"error_message": candidate.error_message,
"latency_ms": candidate.latency_ms,
"concurrent_requests": candidate.concurrent_requests,
"ranking": build_admin_monitoring_trace_candidate_ranking(candidate.extra_data.as_ref()),
"extra_data": build_admin_monitoring_trace_candidate_extra_data(candidate.extra_data.as_ref(), usage),
"created_at": unix_ms_to_rfc3339(candidate.created_at_unix_ms),
"started_at": candidate.started_at_unix_ms.and_then(unix_ms_to_rfc3339),
@@ -417,6 +418,41 @@ fn admin_monitoring_candidate_status_rank(status: RequestCandidateStatus) -> u8
}
}
fn build_admin_monitoring_trace_candidate_ranking(existing: Option<&Value>) -> Value {
let Some(object) = existing.and_then(Value::as_object) else {
return Value::Null;
};
let mut ranking = serde_json::Map::new();
if let Some(ranking_mode) = json_string_field(object, "ranking_mode") {
ranking.insert("mode".to_string(), Value::String(ranking_mode));
}
if let Some(priority_mode) = json_string_field(object, "priority_mode") {
ranking.insert("priority_mode".to_string(), Value::String(priority_mode));
}
if let Some(ranking_index) = json_u64_field(object, "ranking_index") {
ranking.insert("index".to_string(), Value::Number(ranking_index.into()));
}
if let Some(priority_slot) = json_i64_field(object, "priority_slot") {
ranking.insert(
"priority_slot".to_string(),
Value::Number(priority_slot.into()),
);
}
if let Some(promoted_by) = json_string_field(object, "promoted_by") {
ranking.insert("promoted_by".to_string(), Value::String(promoted_by));
}
if let Some(demoted_by) = json_string_field(object, "demoted_by") {
ranking.insert("demoted_by".to_string(), Value::String(demoted_by));
}
if ranking.is_empty() {
Value::Null
} else {
Value::Object(ranking)
}
}
fn build_admin_monitoring_trace_candidate_extra_data(
existing: Option<&Value>,
usage: Option<&StoredRequestUsageAudit>,
@@ -461,6 +497,31 @@ fn build_admin_monitoring_trace_candidate_extra_data(
}
}
fn json_string_field(object: &serde_json::Map<String, Value>, key: &str) -> Option<String> {
object
.get(key)
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn json_u64_field(object: &serde_json::Map<String, Value>, key: &str) -> Option<u64> {
match object.get(key)? {
Value::Number(value) => value.as_u64(),
Value::String(value) => value.trim().parse::<u64>().ok(),
_ => None,
}
}
fn json_i64_field(object: &serde_json::Map<String, Value>, key: &str) -> Option<i64> {
match object.get(key)? {
Value::Number(value) => value.as_i64().or_else(|| value.as_u64()?.try_into().ok()),
Value::String(value) => value.trim().parse::<i64>().ok(),
_ => None,
}
}
fn parse_admin_monitoring_usage_proxy_timing(usage: &StoredRequestUsageAudit) -> Option<Value> {
admin_monitoring_header_value(usage.response_headers.as_ref(), "x-proxy-timing")
.or_else(|| {

10
frontend/src/api/requestTrace.ts
View File

@@ -1,5 +1,14 @@
import apiClient from './client'
export interface CandidateRankingMetadata {
mode?: string
priority_mode?: string
index?: number
priority_slot?: number
promoted_by?: string
demoted_by?: string
}
export interface CandidateRecord {
id: string
request_id: string
@@ -36,6 +45,7 @@ export interface CandidateRecord {
error_message?: string
latency_ms?: number
concurrent_requests?: number
ranking?: CandidateRankingMetadata | null
extra_data?: Record<string, unknown>
created_at: string
started_at?: string

15
frontend/src/features/usage/components/HorizontalRequestTimeline.vue
View File

@@ -1342,15 +1342,16 @@ const currentAttemptRankingInfo = computed<{
} | null>(() => {
const attempt = currentAttempt.value
if (!attempt) return null
const ranking = extractObject(attempt.ranking)
const extra = extractObject(attempt.extra_data)
if (!extra) return null
if (!ranking && !extra) return null
const rankingMode = normalizeMetadataText(extra.ranking_mode)
const priorityMode = normalizeMetadataText(extra.priority_mode)
const promotedBy = normalizeMetadataText(extra.promoted_by)
const demotedBy = normalizeMetadataText(extra.demoted_by)
const rankingIndex = normalizePriorityNumber(extra.ranking_index)
const prioritySlot = normalizePriorityNumber(extra.priority_slot)
const rankingMode = normalizeMetadataText(ranking?.mode ?? extra?.ranking_mode)
const priorityMode = normalizeMetadataText(ranking?.priority_mode ?? extra?.priority_mode)
const promotedBy = normalizeMetadataText(ranking?.promoted_by ?? extra?.promoted_by)
const demotedBy = normalizeMetadataText(ranking?.demoted_by ?? extra?.demoted_by)
const rankingIndex = normalizePriorityNumber(ranking?.index ?? extra?.ranking_index)
const prioritySlot = normalizePriorityNumber(ranking?.priority_slot ?? extra?.priority_slot)
if (!rankingMode && !priorityMode && !promotedBy && !demotedBy && rankingIndex === null && prioritySlot === null) {
return null
}

52
frontend/src/mocks/handler.ts
View File

@@ -2461,6 +2461,20 @@ registerDynamicRoute('GET', '/api/admin/monitoring/trace/:requestId', async (_co
status: 'skipped',
skip_reason: ['并发限制已满', '健康分数过低', '倍率不匹配'][i % 3],
is_cached: false,
ranking: {
mode: record.cache_read_input_tokens > 0 ? 'CacheAffinity' : 'FixedOrder',
priority_mode: 'Provider',
index: i,
priority_slot: i + 1,
demoted_by: i > 0 ? 'cross_format' : undefined
},
extra_data: {
ranking_mode: record.cache_read_input_tokens > 0 ? 'CacheAffinity' : 'FixedOrder',
priority_mode: 'Provider',
ranking_index: i,
priority_slot: i + 1,
demoted_by: i > 0 ? 'cross_format' : undefined
},
latency_ms: 10 + Math.floor(Math.random() * 20),
created_at: skipStarted.toISOString(),
started_at: skipStarted.toISOString(),
@@ -2492,6 +2506,20 @@ registerDynamicRoute('GET', '/api/admin/monitoring/trace/:requestId', async (_co
status: 'success',
is_cached: record.cache_read_input_tokens > 0,
status_code: 200,
ranking: {
mode: record.cache_read_input_tokens > 0 ? 'CacheAffinity' : 'FixedOrder',
priority_mode: 'Provider',
index: skipCount,
priority_slot: record.cache_read_input_tokens > 0 ? 7 : skipCount + 1,
promoted_by: record.cache_read_input_tokens > 0 ? 'cached_affinity' : undefined
},
extra_data: {
ranking_mode: record.cache_read_input_tokens > 0 ? 'CacheAffinity' : 'FixedOrder',
priority_mode: 'Provider',
ranking_index: skipCount,
priority_slot: record.cache_read_input_tokens > 0 ? 7 : skipCount + 1,
promoted_by: record.cache_read_input_tokens > 0 ? 'cached_affinity' : undefined
},
latency_ms: baseLatency,
created_at: successStarted.toISOString(),
started_at: successStarted.toISOString(),
@@ -2524,6 +2552,18 @@ registerDynamicRoute('GET', '/api/admin/monitoring/trace/:requestId', async (_co
status_code: record.status_code,
error_type: ['rate_limit_error', 'api_error', 'timeout_error'][i % 3],
error_message: record.error_message || 'Request failed',
ranking: {
mode: 'FixedOrder',
priority_mode: 'Provider',
index: i,
priority_slot: i + 1
},
extra_data: {
ranking_mode: 'FixedOrder',
priority_mode: 'Provider',
ranking_index: i,
priority_slot: i + 1
},
latency_ms: attemptLatency,
created_at: attemptStarted.toISOString(),
started_at: attemptStarted.toISOString(),
@@ -2549,6 +2589,18 @@ registerDynamicRoute('GET', '/api/admin/monitoring/trace/:requestId', async (_co
required_capabilities: {},
status: 'streaming',
is_cached: false,
ranking: {
mode: 'FixedOrder',
priority_mode: 'Provider',
index: 0,
priority_slot: 1
},
extra_data: {
ranking_mode: 'FixedOrder',
priority_mode: 'Provider',
ranking_index: 0,
priority_slot: 1
},
latency_ms: undefined,
created_at: now.toISOString(),
started_at: now.toISOString(),