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(proxy): 支持远程推送升级与proxy元数据上报

Browse Source 
- aether-proxy 注册和心跳时上报 proxy_metadata（含版本号）
- 心跳 ACK 支持 upgrade_to 字段，proxy 收到后自动执行升级
- 重构 upgrade 逻辑，新增 perform_upgrade 用于远程触发的自动升级
- stream_handler 延迟统计改为仅记录连接建立延迟(DNS+TCP/TLS+TTFB)
- 后端新增 proxy_metadata 数据库字段和批量升级 API
- 远程配置支持下发 upgrade_to 版本指令
- 前端展示节点版本号，支持单节点和批量升级操作
This commit is contained in:
fawney19
2026-03-02 12:58:41 +08:00
parent f978888759
commit 68bae686da
14 changed files with 538 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
aether-proxy/src/registration/client.rs
View File

@@ -20,6 +20,8 @@ struct RegisterRequest {
hardware_info: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
estimated_max_concurrency: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
proxy_metadata: Option<serde_json::Value>,
tunnel_mode: bool,
}
@@ -107,6 +109,9 @@ impl AetherClient {
heartbeat_interval: config.heartbeat_interval,
hardware_info: hw.and_then(|h| serde_json::to_value(h).ok()),
estimated_max_concurrency: hw.map(|h| h.estimated_max_concurrency),
proxy_metadata: Some(serde_json::json!({
"version": env!("CARGO_PKG_VERSION"),
})),
tunnel_mode: true,
};

82
aether-proxy/src/setup/upgrade.rs
View File

@@ -281,8 +281,17 @@ fn atomic_replace(new_binary: &Path) -> anyhow::Result<PathBuf> {
// ── Public entry point ───────────────────────────────────────────────────────
/// `aether-proxy upgrade [version]` -- self-upgrade from GitHub releases.
pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
#[derive(Clone, Copy)]
enum RestartMode {
BestEffort,
Required,
}
async fn execute_upgrade(
version: Option<&str>,
require_root: bool,
restart_mode: RestartMode,
) -> anyhow::Result<()> {
// Resolve exe path once; reuse throughout the function
let current_exe = std::env::current_exe()?.canonicalize()?;
let exe_dir = current_exe
@@ -290,8 +299,12 @@ pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
.ok_or_else(|| anyhow::anyhow!("cannot determine binary directory"))?;
let temp_path = exe_dir.join(".aether-proxy.upgrade.tmp");
// Check write permission to binary directory
if !super::service::is_root() {
if require_root {
if !super::service::is_root() {
anyhow::bail!("automatic upgrade requires root privileges");
}
} else if !super::service::is_root() {
// Check write permission to binary directory for manual upgrade mode.
let test_path = exe_dir.join(".aether-proxy.write-test");
match std::fs::File::create(&test_path) {
Ok(_) => {
@@ -311,7 +324,7 @@ pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
eprintln!(" Current version: {}", CURRENT_VERSION);
let client = build_github_client()?;
let release = fetch_release(&client, version.as_deref()).await?;
let release = fetch_release(&client, version).await?;
let target_tag = &release.tag_name;
let target_semver = target_tag.strip_prefix("proxy-v").unwrap_or(target_tag);
@@ -341,26 +354,38 @@ pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
}
};
// Restart systemd service if running.
// Use best-effort: binary is already replaced, so a restart failure should
// not abort the whole upgrade -- the user can restart manually.
if super::service::is_service_active() {
if super::service::is_root() {
eprintln!(" Restarting systemd service...");
match super::service::run_cmd("systemctl", &["restart", "aether-proxy"]) {
Ok(()) => eprintln!(" Service restarted."),
Err(e) => {
eprintln!(" WARNING: failed to restart service: {}", e);
eprintln!(" Run manually: sudo systemctl restart aether-proxy");
match restart_mode {
RestartMode::BestEffort => {
// Restart systemd service if running.
// Use best-effort: binary is already replaced, so a restart failure should
// not abort the whole upgrade -- the user can restart manually.
if super::service::is_service_active() {
if super::service::is_root() {
eprintln!(" Restarting systemd service...");
match super::service::run_cmd("systemctl", &["restart", "aether-proxy"]) {
Ok(()) => eprintln!(" Service restarted."),
Err(e) => {
eprintln!(" WARNING: failed to restart service: {}", e);
eprintln!(" Run manually: sudo systemctl restart aether-proxy");
}
}
} else {
eprintln!(" Systemd service is active, but restart requires root.");
eprintln!(" Run: sudo systemctl restart aether-proxy");
eprintln!(" Skipping restart.");
}
} else {
eprintln!(" No active systemd service detected, skipping restart.");
}
} else {
eprintln!(" Systemd service is active, but restart requires root.");
eprintln!(" Run: sudo systemctl restart aether-proxy");
eprintln!(" Skipping restart.");
}
} else {
eprintln!(" No active systemd service detected, skipping restart.");
RestartMode::Required => {
if !super::service::is_root() {
anyhow::bail!("automatic upgrade requires root privileges");
}
eprintln!(" Restarting systemd service...");
super::service::run_cmd("systemctl", &["restart", "aether-proxy"])?;
eprintln!(" Service restarted.");
}
}
eprintln!();
@@ -371,3 +396,16 @@ pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
);
Ok(())
}
/// `aether-proxy upgrade [version]` -- self-upgrade from GitHub releases.
pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
execute_upgrade(version.as_deref(), false, RestartMode::BestEffort).await
}
/// Perform automatic upgrade to a specific version.
///
/// This path is designed for server-pushed upgrades in systemd/root scenarios:
/// it requires root and requires a successful `systemctl restart aether-proxy`.
pub async fn perform_upgrade(version: &str) -> anyhow::Result<()> {
execute_upgrade(Some(version), true, RestartMode::Required).await
}

8
aether-proxy/src/state.rs
View File

@@ -47,6 +47,8 @@ pub struct ServerContext {
/// Aggregate metrics for reporting to Aether.
pub struct ProxyMetrics {
pub total_requests: AtomicU64,
/// Cumulative connection-establishment latency in nanoseconds
/// (DNS + TCP/TLS + TTFB, excludes response body streaming).
pub total_latency_ns: AtomicU64,
pub failed_requests: AtomicU64,
pub dns_failures: AtomicU64,
@@ -64,8 +66,10 @@ impl ProxyMetrics {
}
}
pub fn record_request(&self, elapsed: Duration) {
let nanos = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX);
/// Record a completed request with its connection-establishment latency
/// (DNS + TCP/TLS + TTFB, excludes response body streaming).
pub fn record_request(&self, connect_elapsed: Duration) {
let nanos = u64::try_from(connect_elapsed.as_nanos()).unwrap_or(u64::MAX);
self.total_requests.fetch_add(1, Ordering::Release);
self.total_latency_ns.fetch_add(nanos, Ordering::Release);
}

88
aether-proxy/src/tunnel/heartbeat.rs
View File

@@ -1,6 +1,6 @@
//! Tunnel heartbeat: sends metrics over the tunnel, processes ACKs.
use std::sync::atomic::Ordering;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use std::time::SystemTime;
@@ -8,7 +8,7 @@ use std::time::UNIX_EPOCH;
use bytes::Bytes;
use tokio::sync::watch;
use tracing::{debug, warn};
use tracing::{debug, info, warn};
use crate::config::Config;
use crate::registration::client::RemoteConfig;
@@ -18,8 +18,15 @@ use crate::state::ServerContext;
use super::protocol::{Frame, MsgType};
use super::writer::FrameSender;
const CURRENT_VERSION: &str = env!("CARGO_PKG_VERSION");
static UPGRADE_IN_PROGRESS: AtomicBool = AtomicBool::new(false);
static NON_ROOT_UPGRADE_WARNED: AtomicBool = AtomicBool::new(false);
enum AckDecision {
Accept(Option<u64>),
Accept {
heartbeat_id: Option<u64>,
upgrade_to: Option<String>,
},
Ignore,
}
@@ -130,7 +137,10 @@ pub fn spawn(
}
Some(ack_payload) = ack_rx.recv() => {
match handle_ack(&server, &ack_payload) {
AckDecision::Accept(ack_id) => {
AckDecision::Accept {
heartbeat_id: ack_id,
upgrade_to,
} => {
if let Some((pending_id, _)) = pending {
match ack_id {
Some(id) if id == pending_id => {
@@ -144,6 +154,7 @@ pub fn spawn(
_ => {}
}
}
maybe_trigger_upgrade(upgrade_to);
}
AckDecision::Ignore => {}
}
@@ -229,6 +240,9 @@ fn build_heartbeat_payload(
"failed_requests": snapshot.failed,
"dns_failures": snapshot.dns_failures,
"stream_errors": snapshot.stream_errors,
"proxy_metadata": {
"version": CURRENT_VERSION,
},
});
Bytes::from(serde_json::to_vec(&payload).unwrap_or_default())
@@ -236,7 +250,10 @@ fn build_heartbeat_payload(
fn handle_ack(server: &ServerContext, payload: &[u8]) -> AckDecision {
if payload.is_empty() {
return AckDecision::Accept(None);
return AckDecision::Accept {
heartbeat_id: None,
upgrade_to: None,
};
}
#[derive(serde::Deserialize)]
@@ -247,6 +264,8 @@ fn handle_ack(server: &ServerContext, payload: &[u8]) -> AckDecision {
config_version: u64,
#[serde(default)]
heartbeat_id: Option<u64>,
#[serde(default)]
upgrade_to: Option<String>,
}
match serde_json::from_slice::<AckPayload>(payload) {
@@ -254,7 +273,10 @@ fn handle_ack(server: &ServerContext, payload: &[u8]) -> AckDecision {
if let Some(ref rc) = ack.remote_config {
runtime::apply_remote_config(&server.dynamic, rc, ack.config_version);
}
AckDecision::Accept(ack.heartbeat_id)
AckDecision::Accept {
heartbeat_id: ack.heartbeat_id,
upgrade_to: ack.upgrade_to.and_then(normalize_upgrade_target),
}
}
Err(e) => {
warn!(error = %e, "failed to parse heartbeat ACK");
@@ -262,3 +284,57 @@ fn handle_ack(server: &ServerContext, payload: &[u8]) -> AckDecision {
}
}
}
fn normalize_upgrade_target(raw: String) -> Option<String> {
let trimmed = raw.trim();
if trimmed.is_empty() {
return None;
}
let normalized = trimmed.strip_prefix("proxy-v").unwrap_or(trimmed);
if normalized == CURRENT_VERSION {
return None;
}
Some(normalized.to_string())
}
fn maybe_trigger_upgrade(version: Option<String>) {
let Some(target_version) = version else {
return;
};
if !crate::setup::service::is_root() {
if NON_ROOT_UPGRADE_WARNED
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
warn!(
target_version = %target_version,
"remote upgrade skipped: root privileges are required"
);
}
return;
}
if UPGRADE_IN_PROGRESS
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
debug!(target_version = %target_version, "upgrade already in progress, ignoring");
return;
}
tokio::spawn(async move {
info!(target_version = %target_version, "received remote upgrade instruction");
match crate::setup::upgrade::perform_upgrade(&target_version).await {
Ok(()) => {
info!(target_version = %target_version, "remote upgrade finished");
}
Err(e) => {
warn!(
target_version = %target_version,
error = %e,
"remote upgrade failed"
);
UPGRADE_IN_PROGRESS.store(false, Ordering::Release);
}
}
});
}

46
aether-proxy/src/tunnel/stream_handler.rs
View File

@@ -55,13 +55,15 @@ pub async fn handle_stream(
mut body_rx: mpsc::Receiver<Frame>,
frame_tx: FrameSender,
) {
let start = Instant::now();
server.active_connections.fetch_add(1, Ordering::Release);
handle_stream_inner(&state, &server, stream_id, meta, &mut body_rx, &frame_tx).await;
let connect_elapsed =
handle_stream_inner(&state, &server, stream_id, meta, &mut body_rx, &frame_tx).await;
server.active_connections.fetch_sub(1, Ordering::Release);
server.metrics.record_request(start.elapsed());
if let Some(d) = connect_elapsed {
server.metrics.record_request(d);
}
}
/// Send a frame to the writer with a timeout. Returns false if send failed.
@@ -80,6 +82,9 @@ async fn send_frame(tx: &FrameSender, frame: Frame) -> bool {
}
}
/// Returns the connection-establishment duration (DNS + TCP/TLS + TTFB) if the
/// upstream request succeeded, or `None` if the request never reached the
/// response-headers stage.
async fn handle_stream_inner(
state: &AppState,
server: &ServerContext,
@@ -87,7 +92,7 @@ async fn handle_stream_inner(
meta: RequestMeta,
body_rx: &mut mpsc::Receiver<Frame>,
frame_tx: &FrameSender,
) {
) -> Option<Duration> {
// Collect request body
let mut body_parts: Vec<Bytes> = Vec::new();
let mut body_done = false;
@@ -106,7 +111,7 @@ async fn handle_stream_inner(
&format!("gzip decompress failed: {e}"),
)
.await;
return;
return None;
}
};
if !payload.is_empty() {
@@ -120,11 +125,11 @@ async fn handle_stream_inner(
{
body_done = true;
if frame.msg_type == MsgType::StreamError {
return; // Client cancelled
return None; // Client cancelled
}
}
}
None => return, // Channel closed
None => return None, // Channel closed
}
}
@@ -146,7 +151,7 @@ async fn handle_stream_inner(
Ok(u) => u,
Err(e) => {
send_error(frame_tx, stream_id, &format!("invalid URL: {e}")).await;
return;
return None;
}
};
@@ -160,7 +165,7 @@ async fn handle_stream_inner(
&format!("unsupported URL scheme: {other}"),
)
.await;
return;
return None;
}
}
@@ -168,13 +173,13 @@ async fn handle_stream_inner(
Some(h) => h.to_string(),
None => {
send_error(frame_tx, stream_id, "missing host in URL").await;
return;
return None;
}
};
let port = target_url.port_or_known_default().unwrap_or(443);
// DNS + target validation (populates dns_cache for SafeDnsResolver)
let dns_start = Instant::now();
let connect_start = Instant::now();
{
let allowed_ports = Arc::clone(&server.dynamic.load().allowed_ports);
if let Err(e) =
@@ -182,10 +187,10 @@ async fn handle_stream_inner(
{
server.metrics.dns_failures.fetch_add(1, Ordering::Release);
send_error(frame_tx, stream_id, &format!("target blocked: {e}")).await;
return;
return None;
}
}
let dns_ms = dns_start.elapsed().as_millis() as u64;
let dns_ms = connect_start.elapsed().as_millis() as u64;
// Execute upstream request
let client = &state.reqwest_client;
@@ -231,10 +236,14 @@ async fn handle_stream_inner(
format!("upstream error: {e}")
};
send_error(frame_tx, stream_id, &msg).await;
return;
return None;
}
};
// Capture connection-establishment duration (DNS + TCP/TLS + TTFB)
// before proceeding to stream the response body.
let connect_elapsed = connect_start.elapsed();
// Send RESPONSE_HEADERS
let status = response.status().as_u16();
let ttfb_ms = upstream_start.elapsed().as_millis() as u64;
@@ -271,7 +280,7 @@ async fn handle_stream_inner(
)
.await
{
return;
return Some(connect_elapsed);
}
// Stream response body — relay upstream bytes through the tunnel.
@@ -291,7 +300,7 @@ async fn handle_stream_inner(
)
.await
{
return;
return Some(connect_elapsed);
}
} else {
// Split oversized chunks, compress each slice
@@ -306,7 +315,7 @@ async fn handle_stream_inner(
)
.await
{
return;
return Some(connect_elapsed);
}
offset = end;
}
@@ -316,7 +325,7 @@ async fn handle_stream_inner(
server.metrics.stream_errors.fetch_add(1, Ordering::Release);
warn!(stream_id, error = %e, "upstream body read error");
send_error(frame_tx, stream_id, &format!("body read error: {e}")).await;
return;
return Some(connect_elapsed);
}
}
}
@@ -334,6 +343,7 @@ async fn handle_stream_inner(
.await;
debug!(stream_id, status, "stream completed");
Some(connect_elapsed)
}
async fn send_error(tx: &FrameSender, stream_id: u32, msg: &str) {