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

feat(proxy): surface node resource diagnostics

Browse Source
This commit is contained in:
fawney19
2026-05-11 02:59:18 +08:00
parent cc4512fbbb
commit b813498e40
12 changed files with 579 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
apps/aether-proxy/src/app.rs
View File

@@ -158,6 +158,7 @@ pub async fn run(mut config: Config, servers: Vec<ServerEntry>) -> anyhow::Resul
// Build a profile-keyed Hyper client pool for tunnel upstream requests.
let upstream_client_pool =
upstream_client::UpstreamClientPool::new(Arc::clone(&config), Arc::clone(&dns_cache));
let resource_monitor = Arc::new(hardware::RuntimeResourceMonitor::new());
// Register with each Aether server and build per-server contexts.
// Wrapped in Arc<Mutex> so retry_failed_registrations can append later.
@@ -218,6 +219,7 @@ pub async fn run(mut config: Config, servers: Vec<ServerEntry>) -> anyhow::Resul
dns_cache,
upstream_client_pool,
tunnel_tls_config,
resource_monitor,
stream_gate: None,
distributed_stream_gate: None,
};
@@ -947,6 +949,7 @@ mod tests {
dns_cache,
upstream_client_pool,
tunnel_tls_config: Arc::new(crate::tunnel::client::build_tls_config()),
resource_monitor: Arc::new(crate::hardware::RuntimeResourceMonitor::new()),
stream_gate: None,
distributed_stream_gate: None,
})

116
apps/aether-proxy/src/hardware.rs
View File

@@ -1,5 +1,8 @@
use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};
use serde::Serialize;
use sysinfo::System;
use sysinfo::{get_current_pid, Pid, ProcessesToUpdate, System};
use tracing::info;
/// Hardware information collected at startup.
@@ -17,6 +20,102 @@ pub struct HardwareInfo {
pub estimated_max_concurrency: u64,
}
/// Runtime resource usage sampled during heartbeat reporting.
#[derive(Debug, Clone, Serialize)]
pub struct RuntimeResourceSnapshot {
pub sampled_at_unix_secs: u64,
pub system_cpu_usage_percent: f64,
pub process_cpu_usage_percent: f64,
pub memory_total_bytes: u64,
pub memory_used_bytes: u64,
pub memory_available_bytes: u64,
pub memory_used_percent: f64,
pub process_memory_bytes: u64,
pub process_virtual_memory_bytes: u64,
pub process_memory_percent: f64,
pub load_average_1m: f64,
pub load_average_5m: f64,
pub load_average_15m: f64,
pub system_uptime_secs: u64,
pub process_uptime_secs: Option<u64>,
}
/// Small, reusable sysinfo monitor. Keeping it alive between samples makes CPU
/// usage deltas meaningful without re-enumerating the whole machine every time.
pub struct RuntimeResourceMonitor {
system: Mutex<System>,
current_pid: Option<Pid>,
}
impl RuntimeResourceMonitor {
pub fn new() -> Self {
let mut system = System::new_all();
let current_pid = get_current_pid().ok();
if let Some(pid) = current_pid {
system.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
}
system.refresh_cpu_usage();
system.refresh_memory();
Self {
system: Mutex::new(system),
current_pid,
}
}
pub fn snapshot(&self) -> RuntimeResourceSnapshot {
let mut system = match self.system.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
system.refresh_cpu_usage();
system.refresh_memory();
if let Some(pid) = self.current_pid {
system.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
}
let memory_total_bytes = system.total_memory();
let memory_used_bytes = system.used_memory();
let memory_available_bytes = system.available_memory();
let (
process_cpu_usage_percent,
process_memory_bytes,
process_virtual_memory_bytes,
process_uptime_secs,
) = self
.current_pid
.and_then(|pid| system.process(pid))
.map(|process| {
(
process.cpu_usage() as f64,
process.memory(),
process.virtual_memory(),
Some(process.run_time()),
)
})
.unwrap_or((0.0, 0, 0, None));
let load = System::load_average();
RuntimeResourceSnapshot {
sampled_at_unix_secs: current_unix_secs(),
system_cpu_usage_percent: system.global_cpu_usage() as f64,
process_cpu_usage_percent,
memory_total_bytes,
memory_used_bytes,
memory_available_bytes,
memory_used_percent: ratio_percent(memory_used_bytes, memory_total_bytes),
process_memory_bytes,
process_virtual_memory_bytes,
process_memory_percent: ratio_percent(process_memory_bytes, memory_total_bytes),
load_average_1m: load.one,
load_average_5m: load.five,
load_average_15m: load.fifteen,
system_uptime_secs: System::uptime(),
process_uptime_secs,
}
}
}
/// Collect hardware information and estimate max concurrency.
///
/// Should be called once at startup -- hardware does not change at runtime.
@@ -77,3 +176,18 @@ fn get_fd_limit() -> u64 {
// Fallback for non-unix or error
1024
}
fn ratio_percent(value: u64, total: u64) -> f64 {
if total == 0 {
0.0
} else {
value as f64 * 100.0 / total as f64
}
}
fn current_unix_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or(0)
}

107
apps/aether-proxy/src/state.rs
View File

@@ -10,6 +10,7 @@ use aether_runtime::{AdmissionPermit, ConcurrencyError, ConcurrencyGate, Concurr
use aether_runtime_state::{RuntimeSemaphore, RuntimeSemaphoreError, RuntimeSemaphoreSnapshot};
use crate::config::Config;
use crate::hardware::RuntimeResourceMonitor;
use crate::registration::client::AetherClient;
use crate::runtime::SharedDynamicConfig;
use crate::target_filter::DnsCache;
@@ -24,6 +25,8 @@ pub struct AppState {
pub upstream_client_pool: UpstreamClientPool,
/// Shared TLS config for tunnel WebSocket connections (avoids re-parsing root CAs on each reconnect).
pub tunnel_tls_config: Arc<rustls::ClientConfig>,
/// Runtime CPU/memory monitor sampled by heartbeat payloads.
pub resource_monitor: Arc<RuntimeResourceMonitor>,
/// Optional per-process stream admission gate.
pub stream_gate: Option<Arc<ConcurrencyGate>>,
/// Optional cross-instance stream admission gate.
@@ -96,6 +99,10 @@ pub struct TunnelErrorEvent {
pub timestamp_unix_secs: u64,
pub category: String,
pub message: String,
pub severity: String,
pub component: String,
pub summary: String,
pub operator_action: String,
}
#[derive(Debug, Clone, Copy, Default)]
@@ -229,11 +236,18 @@ impl TunnelMetrics {
pub fn record_error(&self, category: &str, message: &str) {
self.error_events_total.fetch_add(1, Ordering::Release);
let category = normalize_error_field(category, TUNNEL_ERROR_CATEGORY_MAX_CHARS, "unknown");
let message = normalize_error_field(message, TUNNEL_ERROR_MESSAGE_MAX_CHARS, "n/a");
let diagnostic = classify_tunnel_error(category.as_str(), message.as_str());
let event = TunnelErrorEvent {
timestamp_unix_secs: now_unix_secs(),
category: normalize_error_field(category, TUNNEL_ERROR_CATEGORY_MAX_CHARS, "unknown"),
message: normalize_error_field(message, TUNNEL_ERROR_MESSAGE_MAX_CHARS, "n/a"),
category,
message,
severity: diagnostic.severity.to_string(),
component: diagnostic.component.to_string(),
summary: diagnostic.summary.to_string(),
operator_action: diagnostic.operator_action.to_string(),
};
let mut recent_errors = match self.recent_errors.lock() {
@@ -302,6 +316,95 @@ fn normalize_error_field(value: &str, max_chars: usize, fallback: &str) -> Strin
normalized.chars().take(max_chars).collect()
}
struct TunnelErrorDiagnostic {
severity: &'static str,
component: &'static str,
summary: &'static str,
operator_action: &'static str,
}
fn classify_tunnel_error(category: &str, _message: &str) -> TunnelErrorDiagnostic {
match category {
"stale_timeout" => TunnelErrorDiagnostic {
severity: "warning",
component: "tunnel_read",
summary: "No inbound tunnel frames before stale timeout",
operator_action:
"Check gateway or reverse-proxy idle timeouts, packet loss, and WebSocket ping/pong reachability. Increase AETHER_PROXY_TUNNEL_STALE_TIMEOUT_MS if the network is high-latency.",
},
"ws_write_error" => TunnelErrorDiagnostic {
severity: "error",
component: "tunnel_write",
summary: "WebSocket write failed because the peer closed or reset the connection",
operator_action:
"Check gateway restarts, load balancer resets, NAT/firewall connection tracking, and whether the proxy is reconnecting successfully.",
},
"ws_ping_error" => TunnelErrorDiagnostic {
severity: "error",
component: "tunnel_write",
summary: "WebSocket keepalive ping could not be sent",
operator_action:
"Check whether the peer closed the socket or an intermediary is dropping idle WebSocket connections.",
},
"ws_read_error" => TunnelErrorDiagnostic {
severity: "error",
component: "tunnel_read",
summary: "WebSocket read failed",
operator_action:
"Check gateway logs and network stability around the same timestamp; compare with reconnect and heartbeat ACK counters.",
},
"tunnel_connect_error" => TunnelErrorDiagnostic {
severity: "critical",
component: "tunnel_connect",
summary: "Tunnel connection attempt failed",
operator_action:
"Check Aether URL reachability, DNS, TLS, management token validity, and any configured AETHER_PROXY_AETHER_PROXY_URL.",
},
"frame_decode_error" => TunnelErrorDiagnostic {
severity: "error",
component: "tunnel_protocol",
summary: "Received tunnel frame could not be decoded",
operator_action:
"Check proxy and gateway version compatibility and whether traffic is being modified by an intermediary.",
},
"stream_dispatch_timeout" => TunnelErrorDiagnostic {
severity: "warning",
component: "stream_dispatch",
summary: "Request body frame could not be delivered to its stream handler in time",
operator_action:
"Check proxy CPU, memory, stream concurrency saturation, and slow upstream provider requests.",
},
"heartbeat_ack_empty" | "heartbeat_ack_parse" => TunnelErrorDiagnostic {
severity: "warning",
component: "heartbeat",
summary: "Heartbeat ACK from gateway was missing or invalid",
operator_action:
"Check gateway heartbeat handler logs and proxy/gateway version compatibility.",
},
"writer_task_panic" | "writer_task_cancelled" => TunnelErrorDiagnostic {
severity: "error",
component: "tunnel_writer",
summary: "Tunnel writer task exited unexpectedly",
operator_action:
"Check proxy logs for the preceding write or ping error and confirm the tunnel reconnect loop is active.",
},
"dispatcher_error" => TunnelErrorDiagnostic {
severity: "error",
component: "tunnel_dispatcher",
summary: "Tunnel dispatcher exited with an error",
operator_action:
"Check the proxied request stream and gateway tunnel logs around the same timestamp.",
},
_ => TunnelErrorDiagnostic {
severity: "info",
component: "tunnel",
summary: "Tunnel reported an unclassified error",
operator_action:
"Inspect the raw message and compare it with proxy, gateway, and network logs at the same time.",
},
}
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum ProxyAdmissionError {
#[error("proxy stream admission saturated at {limit} for gate {gate}")]

78
apps/aether-proxy/src/tunnel/heartbeat.rs
View File

@@ -240,6 +240,7 @@ async fn build_heartbeat_payload(
let node_id = server.node_id.read().unwrap().clone();
let tunnel_snapshot = server.tunnel_metrics.snapshot();
let recent_errors = server.tunnel_metrics.recent_errors(8);
let resource_usage = state.resource_monitor.snapshot();
let avg_latency_ms = if snapshot.requests > 0 {
Some(snapshot.latency_ns as f64 / snapshot.requests as f64 / 1_000_000.0)
@@ -291,6 +292,7 @@ async fn build_heartbeat_payload(
"proxy_metadata": {
"version": CURRENT_VERSION,
"admission": admission,
"resource_usage": resource_usage,
"tunnel_metrics": {
"connect_attempts": tunnel_snapshot.connect_attempts,
"connect_successes": tunnel_snapshot.connect_successes,
@@ -419,13 +421,13 @@ mod tests {
use arc_swap::ArcSwap;
use clap::Parser;
use super::{handle_ack, AckDecision};
use super::{build_heartbeat_payload, handle_ack, AckDecision, HeartbeatSnapshot};
use crate::registration::client::AetherClient;
use crate::runtime::DynamicConfig;
use crate::state::{ProxyMetrics, ServerContext, TunnelMetrics};
use crate::state::{AppState, ProxyMetrics, ServerContext, TunnelMetrics};
fn sample_server() -> Arc<ServerContext> {
let config = Arc::new(crate::config::Config::parse_from([
fn sample_config() -> Arc<crate::config::Config> {
Arc::new(crate::config::Config::parse_from([
"aether-proxy",
"--aether-url",
"https://example.com",
@@ -433,7 +435,11 @@ mod tests {
"ae_test",
"--node-name",
"proxy-test",
]));
]))
}
fn sample_server() -> Arc<ServerContext> {
let config = sample_config();
Arc::new(ServerContext {
server_label: "heartbeat-test".to_string(),
aether_url: config.aether_url.clone(),
@@ -452,6 +458,24 @@ mod tests {
})
}
fn sample_state(config: Arc<crate::config::Config>) -> AppState {
let dns_cache = Arc::new(crate::target_filter::DnsCache::new(
std::time::Duration::from_secs(config.dns_cache_ttl_secs),
config.dns_cache_capacity,
));
AppState {
config: Arc::clone(&config),
dns_cache: Arc::clone(&dns_cache),
upstream_client_pool: crate::upstream_client::UpstreamClientPool::new(
config, dns_cache,
),
tunnel_tls_config: Arc::new(crate::tunnel::client::build_tls_config()),
resource_monitor: Arc::new(crate::hardware::RuntimeResourceMonitor::new()),
stream_gate: None,
distributed_stream_gate: None,
}
}
#[test]
fn heartbeat_ack_requires_heartbeat_id() {
let server = sample_server();
@@ -481,4 +505,48 @@ mod tests {
));
assert_eq!(server.dynamic.load().heartbeat_interval, 9);
}
#[tokio::test]
async fn heartbeat_payload_reports_resource_usage_and_tunnel_error_diagnostics() {
let config = sample_config();
let server = sample_server();
server
.tunnel_metrics
.record_error("ws_write_error", "IO error: Connection reset by peer");
let state = sample_state(config);
let payload = build_heartbeat_payload(
&state,
&server,
"session-1",
42,
HeartbeatSnapshot::default(),
)
.await;
let payload: serde_json::Value =
serde_json::from_slice(&payload).expect("heartbeat payload should be JSON");
let resource_usage = payload
.pointer("/proxy_metadata/resource_usage")
.and_then(serde_json::Value::as_object)
.expect("resource usage should be reported");
assert!(resource_usage.contains_key("system_cpu_usage_percent"));
assert!(resource_usage.contains_key("process_memory_bytes"));
let recent_error = payload
.pointer("/proxy_metadata/recent_tunnel_errors/0")
.and_then(serde_json::Value::as_object)
.expect("recent tunnel error should be reported");
assert_eq!(
recent_error
.get("component")
.and_then(serde_json::Value::as_str),
Some("tunnel_write")
);
assert_eq!(
recent_error
.get("severity")
.and_then(serde_json::Value::as_str),
Some("error")
);
}
}

1
apps/aether-proxy/src/tunnel/mod.rs
View File

@@ -462,6 +462,7 @@ mod tests {
dns_cache,
upstream_client_pool,
tunnel_tls_config: Arc::new(crate::tunnel::client::build_tls_config()),
resource_monitor: Arc::new(crate::hardware::RuntimeResourceMonitor::new()),
stream_gate: None,
distributed_stream_gate: None,
})

2
apps/aether-proxy/src/tunnel/stream_handler.rs
View File

@@ -2284,6 +2284,7 @@ mod tests {
dns_cache,
upstream_client_pool,
tunnel_tls_config: Arc::new(build_tls_config()),
resource_monitor: Arc::new(crate::hardware::RuntimeResourceMonitor::new()),
stream_gate,
distributed_stream_gate,
})
@@ -2314,6 +2315,7 @@ mod tests {
dns_cache,
upstream_client_pool,
tunnel_tls_config: Arc::new(build_tls_config()),
resource_monitor: Arc::new(crate::hardware::RuntimeResourceMonitor::new()),
stream_gate: None,
distributed_stream_gate: None,
})