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

fix: isolate provider egress tunnel diagnostics

Browse Source
This commit is contained in:
Codex
2026-06-28 16:15:41 +00:00
parent d225fd1a61
commit 4a636abb73
9 changed files with 368 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/provider-gateway/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@unidesk/provider-gateway",
"version": "0.2.30",
"version": "0.2.31",
"private": true,
"type": "module",
"scripts": {
src/components/provider-gateway/src/egress-proxy.ts
+124 -14
View File
@@ -20,15 +20,22 @@ interface ProviderEgressProxyOptions {
providerId: string;
listenHost: string;
listenPort: number;
openTimeoutMs: number;
idleTimeoutMs: number;
maxTunnelAgeMs: number;
staleTunnelIdleMs: number;
maxPendingBytes: number;
sendToCore: (message: EgressToCoreMessage) => boolean;
isCoreConnected: () => boolean;
logger: Logger;
onStatusChanged?: () => void;
}
interface Tunnel {
id: string;
client: Socket;
method: string;
targetPathSummary: string | null;
opened: boolean;
pending: Buffer[];
pendingBytes: number;
@@ -37,8 +44,28 @@ interface Tunnel {
lastActivityAt: number;
targetHost: string;
targetPort: number;
bytesFromClient: number;
bytesToClient: number;
openTimer: ReturnType<typeof setTimeout> | null;
idleTimer: ReturnType<typeof setTimeout> | null;
maxAgeTimer: ReturnType<typeof setTimeout> | null;
}
interface ClosedTunnelSummary {
[key: string]: JsonValue;
connectionId: string;
method: string;
targetHost: string;
targetPort: number;
targetPathSummary: string | null;
opened: boolean;
ageMs: number;
idleMs: number;
closedAt: string;
failureKind: string;
closeReason: string | null;
bytesFromClient: number;
bytesToClient: number;
}
export interface ProviderEgressProxyHandle {
@@ -77,7 +104,13 @@ function parseHttpProxyRequest(headerText: string): { method: string; target: st
return { method: match[1] ?? "", target: match[2] ?? "", version: match[3] ?? "HTTP/1.1" };
}
function httpRequestTarget(rawTarget: string): { host: string; port: number; rewrittenTarget: string } | null {
function summarizeHttpTargetPath(url: URL): string {
const queryKeys = Array.from(new Set(Array.from(url.searchParams.keys()).filter(Boolean))).sort();
if (queryKeys.length === 0) return url.pathname || "/";
return `${url.pathname || "/"}?${queryKeys.slice(0, 8).join(",")}${queryKeys.length > 8 ? ",..." : ""}`;
}
function httpRequestTarget(rawTarget: string): { host: string; port: number; rewrittenTarget: string; pathSummary: string } | null {
try {
const url = new URL(rawTarget);
if (url.protocol !== "http:") return null;
@@ -85,6 +118,7 @@ function httpRequestTarget(rawTarget: string): { host: string; port: number; rew
host: url.hostname,
port: url.port.length > 0 ? Number(url.port) : 80,
rewrittenTarget: `${url.pathname}${url.search}`,
pathSummary: summarizeHttpTargetPath(url),
};
} catch {
return null;
@@ -125,6 +159,7 @@ function egressFailureKind(error: string | undefined): string {
const normalized = error.toLowerCase();
if (normalized.includes("open timeout")) return "open-timeout";
if (normalized.includes("idle timeout")) return "idle-timeout";
if (normalized.includes("max tunnel age")) return "max-age-exceeded";
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";
@@ -132,13 +167,15 @@ function egressFailureKind(error: string | undefined): string {
return "remote-or-socket-error";
}
const tunnelOpenTimeoutMs = 15_000;
const tunnelIdleTimeoutMs = 600_000;
const maxPendingBytes = 4 * 1024 * 1024;
function tunnelState(tunnel: Tunnel, now: number, staleTunnelIdleMs: number): "pending" | "open" | "stale" {
if (!tunnel.opened) return "pending";
return now - tunnel.lastActivityAt >= staleTunnelIdleMs ? "stale" : "open";
}
export function startProviderEgressProxy(options: ProviderEgressProxyOptions): ProviderEgressProxyHandle {
const proxyUrl = proxyUrlFor(options.listenHost, options.listenPort);
const tunnels = new Map<string, Tunnel>();
const recentClosedTunnels: ClosedTunnelSummary[] = [];
let closed = false;
const send = (message: EgressToCoreMessage): boolean => options.sendToCore(message);
@@ -147,6 +184,26 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
const now = Date.now();
const tunnelList = Array.from(tunnels.values());
const ages = tunnelList.map((tunnel) => now - tunnel.createdAt);
const details = tunnelList
.sort((left, right) => left.createdAt - right.createdAt)
.slice(0, 12)
.map((tunnel) => {
const idleMs = now - tunnel.lastActivityAt;
return {
connectionId: tunnel.id,
method: tunnel.method,
targetHost: tunnel.targetHost,
targetPort: tunnel.targetPort,
targetPathSummary: tunnel.targetPathSummary,
state: tunnelState(tunnel, now, options.staleTunnelIdleMs),
opened: tunnel.opened,
ageMs: now - tunnel.createdAt,
idleMs,
pendingBytes: tunnel.pendingBytes,
bytesFromClient: tunnel.bytesFromClient,
bytesToClient: tunnel.bytesToClient,
};
});
return {
enabled: true,
providerId: options.providerId,
@@ -156,10 +213,18 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
listenPort: options.listenPort,
activeTunnels: tunnels.size,
pendingTunnels: tunnelList.filter((tunnel) => !tunnel.opened).length,
openedTunnels: tunnelList.filter((tunnel) => tunnel.opened).length,
staleTunnels: tunnelList.filter((tunnel) => tunnelState(tunnel, now, options.staleTunnelIdleMs) === "stale").length,
oldestTunnelAgeMs: ages.length > 0 ? Math.max(...ages) : 0,
openTimeoutMs: tunnelOpenTimeoutMs,
idleTimeoutMs: tunnelIdleTimeoutMs,
maxPendingBytes,
openTimeoutMs: options.openTimeoutMs,
idleTimeoutMs: options.idleTimeoutMs,
maxTunnelAgeMs: options.maxTunnelAgeMs,
staleTunnelIdleMs: options.staleTunnelIdleMs,
maxPendingBytes: options.maxPendingBytes,
activeTunnelDetails: details,
activeTunnelDetailLimit: 12,
activeTunnelDetailsTruncated: tunnelList.length > details.length,
recentClosedTunnels,
channel: "provider-gateway",
};
};
@@ -167,31 +232,59 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
const clearTunnelTimers = (tunnel: Tunnel): void => {
if (tunnel.openTimer !== null) clearTimeout(tunnel.openTimer);
if (tunnel.idleTimer !== null) clearTimeout(tunnel.idleTimer);
if (tunnel.maxAgeTimer !== null) clearTimeout(tunnel.maxAgeTimer);
tunnel.openTimer = null;
tunnel.idleTimer = null;
tunnel.maxAgeTimer = null;
};
const destroyTunnel = (tunnel: Tunnel, notifyCore: boolean, error?: string): void => {
tunnels.delete(tunnel.id);
tunnel.closed = true;
const closedAt = nowIso();
const ageMs = Date.now() - tunnel.createdAt;
const idleMs = Date.now() - tunnel.lastActivityAt;
const pendingBytes = tunnel.pendingBytes;
clearTunnelTimers(tunnel);
tunnel.pending.splice(0);
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() });
const failureKind = egressFailureKind(error);
recentClosedTunnels.unshift({
connectionId: tunnel.id,
method: tunnel.method,
targetHost: tunnel.targetHost,
targetPort: tunnel.targetPort,
targetPathSummary: tunnel.targetPathSummary,
opened: tunnel.opened,
ageMs,
idleMs,
closedAt,
failureKind,
closeReason: error ?? null,
bytesFromClient: tunnel.bytesFromClient,
bytesToClient: tunnel.bytesToClient,
});
recentClosedTunnels.splice(16);
const event = {
providerId: options.providerId,
connectionId: tunnel.id,
method: tunnel.method,
targetHost: tunnel.targetHost,
targetPort: tunnel.targetPort,
targetPathSummary: tunnel.targetPathSummary,
opened: tunnel.opened,
ageMs: Date.now() - tunnel.createdAt,
pendingBytes: tunnel.pendingBytes,
failureKind: egressFailureKind(error),
ageMs,
idleMs,
pendingBytes,
bytesFromClient: tunnel.bytesFromClient,
bytesToClient: tunnel.bytesToClient,
failureKind,
...(error === undefined ? {} : { error }),
};
options.logger(error === undefined ? "info" : "warn", "egress_proxy_tunnel_closed", event);
options.onStatusChanged?.();
};
const closeTunnel = (id: string, error?: string): void => {
@@ -204,15 +297,16 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
if (tunnel.closed) return;
tunnel.lastActivityAt = Date.now();
if (tunnel.idleTimer !== null) clearTimeout(tunnel.idleTimer);
tunnel.idleTimer = setTimeout(() => closeTunnel(tunnel.id, "egress proxy idle timeout"), tunnelIdleTimeoutMs);
tunnel.idleTimer = setTimeout(() => closeTunnel(tunnel.id, "egress proxy idle timeout"), options.idleTimeoutMs);
};
const queuePendingChunk = (tunnel: Tunnel, chunk: Buffer): boolean => {
if (tunnel.closed) return false;
tunnel.pending.push(chunk);
tunnel.pendingBytes += chunk.byteLength;
tunnel.bytesFromClient += chunk.byteLength;
refreshTunnelIdle(tunnel);
if (tunnel.pendingBytes <= maxPendingBytes) return true;
if (tunnel.pendingBytes <= options.maxPendingBytes) return true;
closeTunnel(tunnel.id, "egress proxy pending buffer exceeded");
return false;
};
@@ -233,6 +327,7 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
method: tunnel.method,
targetHost: tunnel.targetHost,
targetPort: tunnel.targetPort,
targetPathSummary: tunnel.targetPathSummary,
ageMs: Date.now() - tunnel.createdAt,
pendingBytes: tunnel.pendingBytes,
});
@@ -243,12 +338,15 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: tunnel.id, data: chunk.toString("base64"), encoding: "base64", at: nowIso() });
}
tunnel.pendingBytes = 0;
options.onStatusChanged?.();
return true;
}
if (message.type === "egress_tcp_data") {
if (tunnel === undefined || tunnel.client.destroyed) return true;
refreshTunnelIdle(tunnel);
tunnel.client.write(Buffer.from(message.data, message.encoding === "base64" ? "base64" : "utf8"));
const payload = Buffer.from(message.data, message.encoding === "base64" ? "base64" : "utf8");
tunnel.bytesToClient += payload.byteLength;
tunnel.client.write(payload);
return true;
}
if (message.type === "egress_tcp_close") {
@@ -307,6 +405,7 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
const id = safeConnectionId();
let firstPayload: Buffer | null = null;
let target: { host: string; port: number } | null = null;
let targetPathSummary: string | null = null;
if (parsed.method === "CONNECT") {
target = parseConnectTarget(parsed.target);
firstPayload = rest.length > 0 ? rest : null;
@@ -314,6 +413,7 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
const httpTarget = httpRequestTarget(parsed.target);
if (httpTarget !== null) {
target = { host: httpTarget.host, port: httpTarget.port };
targetPathSummary = httpTarget.pathSummary;
firstPayload = Buffer.concat([rewriteAbsoluteHttpRequest(head, httpTarget.rewrittenTarget), rest]);
}
}
@@ -326,6 +426,7 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
id,
client,
method: parsed.method,
targetPathSummary,
opened: false,
pending: [],
pendingBytes: 0,
@@ -334,8 +435,11 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
lastActivityAt: createdAt,
targetHost: target.host,
targetPort: target.port,
bytesFromClient: 0,
bytesToClient: 0,
openTimer: null,
idleTimer: null,
maxAgeTimer: null,
};
tunnels.set(id, tunnel);
options.logger("info", "egress_proxy_tunnel_open_request", {
@@ -344,8 +448,10 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
method: parsed.method,
targetHost: target.host,
targetPort: target.port,
targetPathSummary,
firstPayloadBytes: firstPayload === null ? 0 : firstPayload.byteLength,
});
options.onStatusChanged?.();
client.on("data", (nextChunk) => {
const nextBuffer = Buffer.isBuffer(nextChunk) ? nextChunk : Buffer.from(nextChunk);
if (!tunnel.opened) {
@@ -353,6 +459,7 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
return;
}
refreshTunnelIdle(tunnel);
tunnel.bytesFromClient += nextBuffer.byteLength;
send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: id, data: nextBuffer.toString("base64"), encoding: "base64", at: nowIso() });
});
client.on("close", () => closeTunnel(id));
@@ -368,12 +475,15 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
method: parsed.method,
targetHost: target.host,
targetPort: target.port,
targetPathSummary,
failureKind: "core-channel-not-connected",
});
fail("503 Service Unavailable", "provider-gateway core channel is not connected\n");
options.onStatusChanged?.();
return;
}
tunnel.openTimer = setTimeout(() => closeTunnel(id, "egress proxy open timeout"), tunnelOpenTimeoutMs);
tunnel.openTimer = setTimeout(() => closeTunnel(id, "egress proxy open timeout"), options.openTimeoutMs);
tunnel.maxAgeTimer = setTimeout(() => closeTunnel(id, "egress proxy max tunnel age exceeded"), options.maxTunnelAgeMs);
refreshTunnelIdle(tunnel);
if (firstPayload !== null) queuePendingChunk(tunnel, firstPayload);
});
src/components/provider-gateway/src/index.ts
+27
View File
@@ -54,6 +54,11 @@ interface RuntimeConfig {
egressProxyEnabled: boolean;
egressProxyListenHost: string;
egressProxyPort: number;
egressProxyOpenTimeoutMs: number;
egressProxyIdleTimeoutMs: number;
egressProxyMaxTunnelAgeMs: number;
egressProxyStaleTunnelIdleMs: number;
egressProxyMaxPendingBytes: number;
logFile: string;
}
@@ -534,6 +539,11 @@ function readConfig(): RuntimeConfig {
egressProxyEnabled: readBooleanEnv("PROVIDER_EGRESS_PROXY_ENABLED", true, "UNIDESK_PROVIDER_EGRESS_PROXY_ENABLED"),
egressProxyListenHost: readOptionalStringEnv("PROVIDER_EGRESS_PROXY_LISTEN_HOST", "UNIDESK_PROVIDER_EGRESS_PROXY_LISTEN_HOST") ?? "0.0.0.0",
egressProxyPort: readNumberEnv("PROVIDER_EGRESS_PROXY_PORT", 18789, "UNIDESK_PROVIDER_EGRESS_PROXY_PORT"),
egressProxyOpenTimeoutMs: Math.max(1_000, Math.min(120_000, Math.floor(readNumberEnv("PROVIDER_EGRESS_PROXY_OPEN_TIMEOUT_MS", 15_000, "UNIDESK_PROVIDER_EGRESS_PROXY_OPEN_TIMEOUT_MS")))),
egressProxyIdleTimeoutMs: Math.max(30_000, Math.min(3_600_000, Math.floor(readNumberEnv("PROVIDER_EGRESS_PROXY_IDLE_TIMEOUT_MS", 600_000, "UNIDESK_PROVIDER_EGRESS_PROXY_IDLE_TIMEOUT_MS")))),
egressProxyMaxTunnelAgeMs: Math.max(60_000, Math.min(14_400_000, Math.floor(readNumberEnv("PROVIDER_EGRESS_PROXY_MAX_TUNNEL_AGE_MS", 3_600_000, "UNIDESK_PROVIDER_EGRESS_PROXY_MAX_TUNNEL_AGE_MS")))),
egressProxyStaleTunnelIdleMs: Math.max(10_000, Math.min(600_000, Math.floor(readNumberEnv("PROVIDER_EGRESS_PROXY_STALE_TUNNEL_IDLE_MS", 120_000, "UNIDESK_PROVIDER_EGRESS_PROXY_STALE_TUNNEL_IDLE_MS")))),
egressProxyMaxPendingBytes: Math.max(64 * 1024, Math.min(64 * 1024 * 1024, Math.floor(readNumberEnv("PROVIDER_EGRESS_PROXY_MAX_PENDING_BYTES", 4 * 1024 * 1024, "UNIDESK_PROVIDER_EGRESS_PROXY_MAX_PENDING_BYTES")))),
logFile: readOptionalStringEnv("LOG_FILE") ?? `/var/log/unidesk/provider-gateway-${providerId}.jsonl`,
};
}
@@ -587,6 +597,17 @@ function currentLabels(): ProviderLabels {
providerGatewayEgressProxyPort: config.egressProxyPort,
providerGatewayEgressProxyConnected: egressProxyStatus === null ? false : egressProxyStatus.connected === true,
providerGatewayEgressProxyActiveTunnels: egressProxyStatus === null ? 0 : egressProxyStatus.activeTunnels ?? 0,
providerGatewayEgressProxyPendingTunnels: egressProxyStatus === null ? 0 : egressProxyStatus.pendingTunnels ?? 0,
providerGatewayEgressProxyOpenedTunnels: egressProxyStatus === null ? 0 : egressProxyStatus.openedTunnels ?? 0,
providerGatewayEgressProxyStaleTunnels: egressProxyStatus === null ? 0 : egressProxyStatus.staleTunnels ?? 0,
providerGatewayEgressProxyOldestTunnelAgeMs: egressProxyStatus === null ? 0 : egressProxyStatus.oldestTunnelAgeMs ?? 0,
providerGatewayEgressProxyOpenTimeoutMs: egressProxyStatus === null ? config.egressProxyOpenTimeoutMs : egressProxyStatus.openTimeoutMs ?? config.egressProxyOpenTimeoutMs,
providerGatewayEgressProxyIdleTimeoutMs: egressProxyStatus === null ? config.egressProxyIdleTimeoutMs : egressProxyStatus.idleTimeoutMs ?? config.egressProxyIdleTimeoutMs,
providerGatewayEgressProxyMaxTunnelAgeMs: egressProxyStatus === null ? config.egressProxyMaxTunnelAgeMs : egressProxyStatus.maxTunnelAgeMs ?? config.egressProxyMaxTunnelAgeMs,
providerGatewayEgressProxyStaleTunnelIdleMs: egressProxyStatus === null ? config.egressProxyStaleTunnelIdleMs : egressProxyStatus.staleTunnelIdleMs ?? config.egressProxyStaleTunnelIdleMs,
providerGatewayEgressProxyMaxPendingBytes: egressProxyStatus === null ? config.egressProxyMaxPendingBytes : egressProxyStatus.maxPendingBytes ?? config.egressProxyMaxPendingBytes,
providerGatewayEgressProxyActiveTunnelDetails: egressProxyStatus === null ? [] : egressProxyStatus.activeTunnelDetails ?? [],
providerGatewayEgressProxyRecentClosedTunnels: egressProxyStatus === null ? [] : egressProxyStatus.recentClosedTunnels ?? [],
providerGatewaySshDataTransport: "tcp-pool",
providerGatewaySshDataHost: config.providerDataHost,
providerGatewaySshDataPort: config.providerDataPort,
@@ -2706,9 +2727,15 @@ if (config.egressProxyEnabled) {
providerId: config.providerId,
listenHost: config.egressProxyListenHost,
listenPort: config.egressProxyPort,
openTimeoutMs: config.egressProxyOpenTimeoutMs,
idleTimeoutMs: config.egressProxyIdleTimeoutMs,
maxTunnelAgeMs: config.egressProxyMaxTunnelAgeMs,
staleTunnelIdleMs: config.egressProxyStaleTunnelIdleMs,
maxPendingBytes: config.egressProxyMaxPendingBytes,
isCoreConnected: () => socket?.readyState === WebSocket.OPEN,
sendToCore: (message) => sendJsonOk(message),
logger,
onStatusChanged: () => sendHeartbeat(),
});
}