mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: improve egress and job diagnostics (#969)

Browse Source 
Co-authored-by: Codex <codex@noreply.local>
This commit is contained in:
Lyon
2026-06-26 12:51:58 +08:00
committed by GitHub
parent d1c189e498
commit d3d542fbd3
10 changed files with 375 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/backend-core/src/egress_tcp.rs
+127 -6
View File
@@ -1,3 +1,5 @@
// SPEC: PJ2026-01060509 出站诊断 draft-2026-06-26-p8-egress-job-friction.
// Backend-core side of provider egress TCP bridge with stage logs.
use std::sync::Arc;
use std::time::Duration;
@@ -43,6 +45,23 @@ fn send_egress_close(provider: &Arc<ProviderConnection>, connection_id: &str, er
let _ = provider.sender.send(Message::Text(message.to_string()));
}
fn egress_failure_kind(error: &str) -> &'static str {
let normalized = error.to_ascii_lowercase();
if normalized.contains("connect timeout") {
"connect-timeout"
} else if normalized.contains("timed out") || normalized.contains("timeout") {
"timeout"
} else if normalized.contains("invalid") {
"invalid-target"
} else if normalized.contains("refused") {
"connect-refused"
} else if normalized.contains("unreachable") {
"connect-unreachable"
} else {
"connect-failed"
}
}
pub async fn handle_egress_tcp_open(
state: &Arc<AppState>,
provider: Arc<ProviderConnection>,
@@ -70,10 +89,42 @@ pub async fn handle_egress_tcp_open(
|| provider_id.is_empty()
|| connection_id.is_empty()
{
state.log(
"warn",
"egress_tcp_open_rejected",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"targetHost": host.as_str(),
"targetPort": port,
"failureKind": "invalid-target"
})),
);
send_egress_close(&provider, &connection_id, Some("invalid egress target"));
return Ok(());
}
state.log(
"info",
"egress_tcp_open_requested",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"targetHost": host.as_str(),
"targetPort": port
})),
);
close_egress_tcp_connection(state, &provider_id, &connection_id, None).await;
state.log(
"info",
"egress_tcp_connect_start",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"targetHost": host.as_str(),
"targetPort": port,
"timeoutMs": 15_000
})),
);
let stream = match tokio::time::timeout(
Duration::from_secs(15),
TcpStream::connect((host.as_str(), port as u16)),
@@ -82,10 +133,35 @@ pub async fn handle_egress_tcp_open(
{
Ok(Ok(stream)) => stream,
Ok(Err(error)) => {
send_egress_close(&provider, &connection_id, Some(&error.to_string()));
let error_text = error.to_string();
state.log(
"warn",
"egress_tcp_connect_failed",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"targetHost": host.as_str(),
"targetPort": port,
"failureKind": egress_failure_kind(&error_text),
"error": error_text.as_str()
})),
);
send_egress_close(&provider, &connection_id, Some(&error_text));
return Ok(());
}
Err(_) => {
state.log(
"warn",
"egress_tcp_connect_timeout",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"targetHost": host.as_str(),
"targetPort": port,
"failureKind": "connect-timeout",
"timeoutMs": 15_000
})),
);
send_egress_close(
&provider,
&connection_id,
@@ -108,6 +184,16 @@ pub async fn handle_egress_tcp_open(
let _ = provider.sender.send(Message::Text(
json!({ "type": "egress_tcp_opened", "connectionId": connection_id }).to_string(),
));
state.log(
"info",
"egress_tcp_connected",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"targetHost": host.as_str(),
"targetPort": port
})),
);
tokio::spawn({
let state = state.clone();
@@ -125,7 +211,18 @@ pub async fn handle_egress_tcp_open(
let _ = provider.sender.send(Message::Text(json!({ "type": "egress_tcp_data", "connectionId": connection_id, "data": data, "encoding": "base64" }).to_string()));
}
Err(error) => {
send_egress_close(&provider, &connection_id, Some(&error.to_string()));
let error_text = error.to_string();
state.log(
"warn",
"egress_tcp_read_failed",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"failureKind": "read-failed",
"error": error_text.as_str()
})),
);
send_egress_close(&provider, &connection_id, Some(&error_text));
break;
}
}
@@ -144,7 +241,17 @@ pub async fn handle_egress_tcp_open(
let connection_id = connection_id.clone();
async move {
while let Some(data) = rx.recv().await {
if writer.write_all(&data).await.is_err() {
if let Err(error) = writer.write_all(&data).await {
state.log(
"warn",
"egress_tcp_write_failed",
Some(json!({
"providerId": provider_id.as_str(),
"connectionId": connection_id.as_str(),
"failureKind": "write-failed",
"error": error.to_string()
})),
);
break;
}
}
@@ -204,13 +311,27 @@ pub async fn close_egress_tcp_connection(
state: &Arc<AppState>,
provider_id: &str,
connection_id: &str,
_error: Option<&str>,
error: Option<&str>,
) {
state
let removed = state
.active_egress_tcp
.lock()
.await
.remove(&egress_tcp_key(provider_id, connection_id));
.remove(&egress_tcp_key(provider_id, connection_id))
.is_some();
if removed || error.is_some() {
state.log(
if error.is_some() { "warn" } else { "info" },
"egress_tcp_connection_closed",
Some(json!({
"providerId": provider_id,
"connectionId": connection_id,
"removed": removed,
"failureKind": error.map(egress_failure_kind).unwrap_or("normal-close"),
"error": error.unwrap_or("")
})),
);
}
}
pub async fn close_egress_tcp_connections_for_provider(state: &Arc<AppState>, provider_id: &str) {
src/components/provider-gateway/src/egress-proxy.ts
+64 -9
View File
@@ -1,3 +1,5 @@
// SPEC: PJ2026-01060509 出站诊断 draft-2026-06-26-p8-egress-job-friction.
// Provider-side egress TCP proxy with lifecycle logs for tunnel diagnosis.
import { createServer, type Server, type Socket } from "node:net";
import type {
CoreEgressTcpCloseMessage,
@@ -33,6 +35,8 @@ interface Tunnel {
closed: boolean;
createdAt: number;
lastActivityAt: number;
targetHost: string;
targetPort: number;
openTimer: ReturnType<typeof setTimeout> | null;
idleTimer: ReturnType<typeof setTimeout> | null;
}
@@ -116,6 +120,18 @@ function proxyUrlFor(host: string, port: number): string {
return `http://${host}:${port}`;
}
function egressFailureKind(error: string | undefined): string {
if (error === undefined || error.length === 0) return "normal-close";
const normalized = error.toLowerCase();
if (normalized.includes("open timeout")) return "open-timeout";
if (normalized.includes("idle timeout")) return "idle-timeout";
if (normalized.includes("pending buffer")) return "pending-buffer-exceeded";
if (normalized.includes("not connected")) return "core-channel-not-connected";
if (normalized.includes("timeout")) return "timeout";
if (normalized.includes("closed")) return "closed";
return "remote-or-socket-error";
}
const tunnelOpenTimeoutMs = 15_000;
const tunnelIdleTimeoutMs = 600_000;
const maxPendingBytes = 4 * 1024 * 1024;
@@ -163,14 +179,19 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
tunnel.pendingBytes = 0;
if (!tunnel.client.destroyed) tunnel.client.destroy();
if (notifyCore) send({ type: "egress_tcp_close", providerId: options.providerId, connectionId: tunnel.id, at: nowIso() });
if (error !== undefined) {
options.logger("warn", "egress_proxy_tunnel_closed", {
connectionId: tunnel.id,
opened: tunnel.opened,
ageMs: Date.now() - tunnel.createdAt,
error,
});
}
const event = {
providerId: options.providerId,
connectionId: tunnel.id,
method: tunnel.method,
targetHost: tunnel.targetHost,
targetPort: tunnel.targetPort,
opened: tunnel.opened,
ageMs: Date.now() - tunnel.createdAt,
pendingBytes: tunnel.pendingBytes,
failureKind: egressFailureKind(error),
...(error === undefined ? {} : { error }),
};
options.logger(error === undefined ? "info" : "warn", "egress_proxy_tunnel_closed", event);
};
const closeTunnel = (id: string, error?: string): void => {
@@ -206,6 +227,15 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
tunnel.openTimer = null;
}
refreshTunnelIdle(tunnel);
options.logger("info", "egress_proxy_tunnel_opened", {
providerId: options.providerId,
connectionId: tunnel.id,
method: tunnel.method,
targetHost: tunnel.targetHost,
targetPort: tunnel.targetPort,
ageMs: Date.now() - tunnel.createdAt,
pendingBytes: tunnel.pendingBytes,
});
if (tunnel.method === "CONNECT") {
tunnel.client.write("HTTP/1.1 200 Connection Established\r\nProxy-Agent: UniDesk-ProviderGateway\r\n\r\n");
}
@@ -226,7 +256,14 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
destroyTunnel(tunnel, false, message.error);
}
if (message.error !== undefined && message.error.length > 0) {
options.logger("warn", "egress_proxy_remote_close", { connectionId: message.connectionId, error: message.error });
options.logger("warn", "egress_proxy_remote_close", {
providerId: options.providerId,
connectionId: message.connectionId,
targetHost: tunnel?.targetHost ?? null,
targetPort: tunnel?.targetPort ?? null,
failureKind: egressFailureKind(message.error),
error: message.error,
});
}
return true;
}
@@ -295,10 +332,20 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
closed: false,
createdAt,
lastActivityAt: createdAt,
targetHost: target.host,
targetPort: target.port,
openTimer: null,
idleTimer: null,
};
tunnels.set(id, tunnel);
options.logger("info", "egress_proxy_tunnel_open_request", {
providerId: options.providerId,
connectionId: id,
method: parsed.method,
targetHost: target.host,
targetPort: target.port,
firstPayloadBytes: firstPayload === null ? 0 : firstPayload.byteLength,
});
client.on("data", (nextChunk) => {
const nextBuffer = Buffer.isBuffer(nextChunk) ? nextChunk : Buffer.from(nextChunk);
if (!tunnel.opened) {
@@ -315,6 +362,14 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
tunnels.delete(id);
tunnel.closed = true;
clearTunnelTimers(tunnel);
options.logger("warn", "egress_proxy_tunnel_open_not_sent", {
providerId: options.providerId,
connectionId: id,
method: parsed.method,
targetHost: target.host,
targetPort: target.port,
failureKind: "core-channel-not-connected",
});
fail("503 Service Unavailable", "provider-gateway core channel is not connected\n");
return;
}