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

Improve runtime miss usage diagnostics

Browse Source
This commit is contained in:
fawney19
2026-04-26 01:44:54 +08:00
parent 1d3ea3232d
commit baa0ddd787
6 changed files with 238 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
apps/aether-gateway/src/execution_runtime/transport.rs
View File

@@ -22,10 +22,6 @@ use serde_json::json;
use serde_json::Value;
use thiserror::Error;
use crate::constants::{
EXECUTION_RUNTIME_LOOP_GUARD_HEADER, EXECUTION_RUNTIME_LOOP_GUARD_VALUE,
EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN,
};
#[cfg(test)]
use crate::execution_runtime::remote_compat::execute_sync_plan_via_remote_execution_runtime;
use crate::frontdoor_loop_guard::{
@@ -298,7 +294,6 @@ pub(crate) async fn execute_stream_plan_via_local_tunnel(
plan.content_encoding.as_deref(),
plan.body.body_bytes_b64.is_some(),
)?;
let headers = append_execution_loop_guard_header(headers);
let started_at = Instant::now();
let response = state
.tunnel
@@ -350,8 +345,6 @@ async fn execute_sync_plan_via_local_tunnel(
plan.content_encoding.as_deref(),
plan.body.body_bytes_b64.is_some(),
)?;
let headers = append_execution_loop_guard_header(headers);
let started_at = Instant::now();
let mut response = state
.tunnel
@@ -444,7 +437,6 @@ async fn send_request(
plan.content_encoding.as_deref(),
plan.body.body_bytes_b64.is_some(),
)?;
let headers = append_execution_loop_guard_header(headers);
let total_timeout = plan
.timeouts
.as_ref()
@@ -480,34 +472,6 @@ async fn send_request(
})
}
fn append_execution_loop_guard_header(mut headers: HeaderMap) -> HeaderMap {
headers.insert(
HeaderName::from_static(EXECUTION_RUNTIME_LOOP_GUARD_HEADER),
HeaderValue::from_static(EXECUTION_RUNTIME_LOOP_GUARD_VALUE),
);
let via_name = HeaderName::from_static("via");
let via_value = headers
.get(&via_name)
.and_then(|value| value.to_str().ok())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| {
if value
.to_ascii_lowercase()
.contains(EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN)
{
value.to_string()
} else {
format!("{value}, 1.1 {EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN}")
}
})
.unwrap_or_else(|| format!("1.1 {EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN}"));
if let Ok(value) = HeaderValue::from_str(via_value.as_str()) {
headers.insert(via_name, value);
}
headers
}
async fn send_via_tunnel_relay(
plan: &ExecutionPlan,
method: reqwest::Method,
@@ -972,6 +936,7 @@ mod tests {
use axum::body::Bytes;
use axum::extract::ws::Message;
use axum::extract::Path;
use axum::http::HeaderMap as AxumHeaderMap;
use axum::routing::post;
use axum::{Json, Router};
use serde_json::json;
@@ -980,6 +945,9 @@ mod tests {
use super::{
build_client, execute_sync_plan, DirectSyncExecutionRuntime, ExecutionTransportControls,
};
use crate::constants::{
EXECUTION_RUNTIME_LOOP_GUARD_HEADER, EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN,
};
use crate::frontdoor_loop_guard::{
frontdoor_self_loop_public_ai_path, gateway_frontdoor_self_loop_guard_error_with_port,
gateway_frontdoor_self_loop_guard_matches_with_port,
@@ -1087,7 +1055,21 @@ mod tests {
let addr = listener.local_addr().expect("local addr should resolve");
let app = Router::new().route(
"/chat",
post(|| async {
post(|headers: AxumHeaderMap| async move {
assert!(
!headers.contains_key(EXECUTION_RUNTIME_LOOP_GUARD_HEADER),
"plain upstream requests must not leak internal execution loop guard headers"
);
assert!(
!headers
.get_all("via")
.iter()
.filter_map(|value| value.to_str().ok())
.any(|value| value
.to_ascii_lowercase()
.contains(EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN)),
"plain upstream requests must not leak internal execution runtime Via markers"
);
(
axum::http::StatusCode::TOO_MANY_REQUESTS,
Json(json!({"error": {"message": "slow down"}})),
@@ -1152,6 +1134,22 @@ mod tests {
assert_eq!(node_id, "node-1");
assert_eq!(meta["method"], "POST");
assert_eq!(meta["url"], "https://example.com/chat");
let headers = meta["headers"]
.as_object()
.expect("relay meta headers should be an object");
assert!(
!headers.contains_key(EXECUTION_RUNTIME_LOOP_GUARD_HEADER),
"tunnel relay metadata must not leak internal execution loop guard headers"
);
let via = headers
.get("via")
.and_then(|value| value.as_str())
.unwrap_or_default();
assert!(
!via.to_ascii_lowercase()
.contains(EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN),
"tunnel relay metadata must not leak internal execution runtime Via markers"
);
let request_json: serde_json::Value =
serde_json::from_slice(&request_body).expect("request body should be json");
assert_eq!(request_json["model"], "gpt-4.1");

72
apps/aether-gateway/src/executor/outcome.rs
View File

@@ -12,7 +12,9 @@ use aether_usage_runtime::{
build_usage_event_data_seed, UsageEvent, UsageEventData, UsageEventType,
};
use axum::body::Body;
use axum::http::{self, Response};
use axum::body::Bytes;
use axum::http::{self, HeaderMap, Response};
use base64::Engine as _;
use serde_json::{json, Map, Value};
use tracing::warn;
@@ -293,6 +295,8 @@ pub(crate) async fn record_failed_usage_for_runtime_miss_request(
decision: Option<&GatewayControlDecision>,
diagnostic: Option<&LocalExecutionRuntimeMissDiagnostic>,
context: &LocalExecutionRuntimeMissContext,
request_headers: &HeaderMap,
request_body: Option<&Bytes>,
) {
if !state.usage_runtime.is_enabled() {
return;
@@ -311,7 +315,6 @@ pub(crate) async fn record_failed_usage_for_runtime_miss_request(
let provider_name = selected_candidate
.and_then(|value| value.provider_name.clone())
.or_else(|| selected_candidate.and_then(|value| value.candidate.provider_id.clone()))
.or_else(|| trimmed_non_empty(decision.and_then(|value| value.route_family.as_deref())))
.unwrap_or_else(|| "unknown".to_string());
let model = trimmed_non_empty(diagnostic.and_then(|value| value.requested_model.as_deref()))
.or_else(|| selected_candidate.and_then(|value| value.global_model_name.clone()))
@@ -391,6 +394,8 @@ pub(crate) async fn record_failed_usage_for_runtime_miss_request(
error_message: Some(local_execution_runtime_miss_detail.to_string()),
error_category: error_category_for_failed_status(status_code),
response_time_ms: Some(started_at.elapsed().as_millis() as u64),
request_headers: Some(runtime_miss_original_headers_json(request_headers)),
request_body: runtime_miss_original_request_body_json(request_headers, request_body),
response_headers: Some(json_header_map()),
response_body: Some(client_body.clone()),
client_response_headers: Some(Value::Object(client_headers)),
@@ -550,6 +555,69 @@ fn json_header_map() -> Value {
)]))
}
fn runtime_miss_original_headers_json(headers: &HeaderMap) -> Value {
let mut headers = crate::headers::collect_control_headers(headers);
for (name, value) in headers.iter_mut() {
if runtime_miss_sensitive_header(name) {
*value = runtime_miss_mask_header_value(value);
}
}
serde_json::to_value(headers).unwrap_or_else(|_| json!({}))
}
fn runtime_miss_original_request_body_json(
headers: &HeaderMap,
body: Option<&Bytes>,
) -> Option<Value> {
let body = body?;
if crate::headers::is_json_request(headers) {
if body.is_empty() {
return Some(json!({}));
}
return serde_json::from_slice::<Value>(body.as_ref()).ok();
}
(!body.is_empty()).then(|| {
json!({
"body_bytes_b64": base64::engine::general_purpose::STANDARD.encode(body.as_ref())
})
})
}
fn runtime_miss_sensitive_header(name: &str) -> bool {
const SENSITIVE_HEADERS: &[&str] = &[
"authorization",
"x-api-key",
"api-key",
"x-goog-api-key",
"cookie",
"proxy-authorization",
];
SENSITIVE_HEADERS
.iter()
.any(|candidate| name.eq_ignore_ascii_case(candidate))
}
fn runtime_miss_mask_header_value(value: &str) -> String {
let value = value.trim();
let char_count = value.chars().count();
if char_count <= 8 {
return "****".to_string();
}
let prefix: String = value.chars().take(4).collect();
let suffix: String = value
.chars()
.rev()
.take(4)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect();
format!("{prefix}****{suffix}")
}
async fn load_runtime_miss_candidate_contexts(
state: &AppState,
request_id: &str,

17
apps/aether-gateway/src/handlers/proxy/mod.rs
View File

@@ -1358,6 +1358,8 @@ pub(crate) async fn proxy_request(
control_decision,
local_execution_runtime_miss_diagnostic.as_ref(),
&local_execution_runtime_miss_context,
&parts.headers,
Some(buffered_body),
)
.await;
}
@@ -1365,8 +1367,9 @@ pub(crate) async fn proxy_request(
&trace_id,
control_decision,
http::StatusCode::SERVICE_UNAVAILABLE,
beautify_local_execution_client_error_message(
local_execution_runtime_miss_client_message(
local_execution_runtime_miss_detail.as_str(),
local_execution_runtime_miss_diagnostic.as_ref(),
)
.as_str(),
)?;
@@ -1435,6 +1438,17 @@ fn local_execution_runtime_miss_detail(
local_execution_runtime_miss_route_detail(decision).map(ToOwned::to_owned)
}
fn local_execution_runtime_miss_client_message(
detail: &str,
diagnostic: Option<&LocalExecutionRuntimeMissDiagnostic>,
) -> String {
if diagnostic.is_some_and(|diagnostic| diagnostic.reason.as_str() == "candidate_list_empty") {
beautify_local_execution_client_error_message(detail)
} else {
detail.to_string()
}
}
fn local_execution_runtime_miss_diagnostic_detail(
decision: Option<&GatewayControlDecision>,
diagnostic: Option<&LocalExecutionRuntimeMissDiagnostic>,
@@ -1608,6 +1622,7 @@ fn local_execution_runtime_miss_skip_reason_label(reason: &str) -> &str {
"mapped_model_missing" => "模型映射缺失",
"provider_inactive" => "提供商未启用",
"provider_request_body_missing" => "无法构建上游请求体",
"provider_request_body_build_failed" => "上游请求体转换失败",
"transport_api_format_mismatch" => "传输层 API 格式不匹配",
"transport_api_format_unsupported" => "传输层不支持该 API 格式",
"transport_auth_unavailable" => "上游认证信息不可用",

43
apps/aether-gateway/src/tests/usage/local.rs
View File

@@ -972,7 +972,7 @@ async fn gateway_records_failed_usage_for_claude_runtime_miss_without_execution_
stored_usage.user_id.as_deref(),
Some("user-claude-runtime-miss-usage-1")
);
assert_eq!(stored_usage.provider_name, "claude");
assert_eq!(stored_usage.provider_name, "unknown");
assert_eq!(stored_usage.model, "claude-sonnet-4-5");
assert_eq!(stored_usage.api_format.as_deref(), Some("claude:chat"));
assert_eq!(
@@ -1593,7 +1593,7 @@ async fn gateway_records_failed_usage_when_all_local_claude_cli_candidates_are_s
stored_usage.user_id.as_deref(),
Some("user-claude-cli-usage-local-miss-1")
);
assert_eq!(stored_usage.provider_name, "claude");
assert_eq!(stored_usage.provider_name, "unknown");
assert_eq!(stored_usage.model, "gpt-5.4");
assert_eq!(stored_usage.api_format.as_deref(), Some("claude:cli"));
assert_eq!(
@@ -1618,6 +1618,31 @@ async fn gateway_records_failed_usage_when_all_local_claude_cli_candidates_are_s
"没有可用提供商支持模型 gpt-5.4 的同步请求。请检查模型映射、端点启用状态和 API Key 权限（原因代码: candidate_list_empty）"
)
);
assert_eq!(
stored_usage
.request_headers
.as_ref()
.and_then(|value| value.get("authorization"))
.and_then(|value| value.as_str()),
Some("Bear****miss")
);
assert_eq!(
stored_usage
.request_headers
.as_ref()
.and_then(|value| value.get("content-type"))
.and_then(|value| value.as_str()),
Some("application/json")
);
assert_eq!(
stored_usage
.request_body
.as_ref()
.and_then(|value| value.get("model"))
.and_then(|value| value.as_str()),
Some("gpt-5.4")
);
assert!(stored_usage.provider_request_body.is_none());
assert_eq!(
stored_usage
.request_metadata
@@ -1863,7 +1888,19 @@ async fn gateway_keeps_failed_usage_request_capture_lightweight_for_large_local_
)
.await;
assert_eq!(stored_usage.status, "failed");
assert!(stored_usage.request_body.is_none());
assert_eq!(
stored_usage.request_body_state,
Some(UsageBodyCaptureState::Inline)
);
assert_eq!(
stored_usage
.request_body
.as_ref()
.and_then(|value| value.get("model"))
.and_then(|value| value.as_str()),
Some("gpt-5.4")
);
assert!(stored_usage.provider_request_body.is_none());
assert_eq!(
stored_usage
.request_metadata