okooooooo/NodeFlare
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化

Browse Source
This commit is contained in:
imengying committed 2026-08-27 21:27:40 +08:00
1 parent e153d0ad10
commit 0cdcbe40f4
58 files changed
+5295 -2138
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/smoke-test.sh
+28 -97
View File
@@ -77,11 +77,7 @@ server_input='{"name":"Smoke Test Node","region":"JP","group_name":"Test","tags"
server_id=
latency_task_id=
alert_rule_id=
stress_report_file=
cleanup() {
if [ -n "$stress_report_file" ]; then
rm -f "$stress_report_file"
fi
if [ -n "$alert_rule_id" ]; then
monitor_curl --silent --show-error -H "Authorization: Bearer $admin_token" \
-X DELETE "$MONITOR_BASE_URL/api/admin/alert-rules/$alert_rule_id" >/dev/null || true
@@ -97,6 +93,18 @@ cleanup() {
}
trap cleanup EXIT
request -H "Authorization: Bearer $admin_token" \
"$MONITOR_BASE_URL/api/admin/telegram" | \
jq -e '.telegram == null or (.telegram.bot_token == "********" and (.telegram.chat_id | length > 0))' >/dev/null
invalid_telegram_status=$(monitor_curl --silent --output /dev/null --write-out '%{http_code}' \
-X PUT -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/json' \
--data '{"bot_token":"invalid","chat_id":"","message_thread_id":null,"template":"{{message}}"}' \
"$MONITOR_BASE_URL/api/admin/telegram")
[ "$invalid_telegram_status" = "400" ]
removed_channels_status=$(monitor_curl --silent --output /dev/null --write-out '%{http_code}' \
-H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/admin/notification-channels")
[ "$removed_channels_status" = "404" ]
latency_task_json=$(request -H "Authorization: Bearer $admin_token" \
-H 'Content-Type: application/json' \
--data '{"name":"Smoke TCP","task_type":"tcp","target":"example.com","port":443,"interval_seconds":60,"default_enabled":true,"server_ids":[]}' \
@@ -110,106 +118,29 @@ server_json=$(request -H "Authorization: Bearer $admin_token" \
server_id=$(printf '%s' "$server_json" | jq -er '.id')
agent_token=$(printf '%s' "$server_json" | jq -er '.agent_token')
MONITOR_BASE_URL="$MONITOR_BASE_URL" MONITOR_ADMIN_TOKEN="$admin_token" \
MONITOR_AGENT_TOKEN="$agent_token" node scripts/websocket-smoke.mjs
MONITOR_AGENT_TOKEN="$agent_token" MONITOR_SERVER_ID="$server_id" \
MONITOR_LATENCY_TASK_ID="$latency_task_id" \
node scripts/websocket-smoke.mjs
token_without_admin_status=$(monitor_curl --silent --output /dev/null --write-out '%{http_code}' \
"$MONITOR_BASE_URL/api/admin/servers/$server_id/token")
[ "$token_without_admin_status" = "401" ]
request -H "Authorization: Bearer $admin_token" \
"$MONITOR_BASE_URL/api/admin/servers/$server_id/token" | \
jq -e --arg token "$agent_token" '.agent_token == $token' >/dev/null
agent_version=$(sh scripts/resolve-version.sh)
sample=$(jq -nc --arg agent_version "$agent_version" '{
timestamp:((((now|floor) / 60 | floor) + 1) * 60 + 10),cpu:18.5,load1:0.42,load5:0.36,load15:0.31,
mem_used:2147483648,mem_total:4294967296,swap_used:0,swap_total:0,disk_used:21474836480,disk_total:53687091200,
net_in:4096,net_out:2048,net_rx_total:1073741824,net_tx_total:536870912,
uptime:86400,processes:90,tcp_connections:18,udp_connections:4,cpu_cores:2,
cpu_model:"Smoke CPU",os:"Debian 12",kernel:"6.1",arch:"x86_64",virtualization:"kvm",
gpu_usage:32.5,gpu_model:"NVIDIA T4",disk_read_bps:4194304,disk_write_bps:2097152,
disk_read_iops:120,disk_write_iops:48,disk_await_ms:1.4,disk_utilization:8.2,
disks:[{name:"/dev/vda1",mount_point:"/",used:21474836480,total:53687091200,read_bps:4194304,write_bps:2097152,read_iops:120,write_iops:48,await_ms:1.4,utilization:8.2}],
gpus:[{model:"NVIDIA T4",usage:32.5,memory_used:1073741824,memory_total:17179869184}],
agent_version:$agent_version,latency_results:[]
}')
report=$(jq -nc --argjson sample "$sample" '{samples:[$sample]}')
invalid_token_status=$(monitor_curl --silent --output /dev/null --write-out '%{http_code}' \
-H 'Authorization: Bearer invalid-agent-token' -H 'Content-Type: application/json' \
--data "$report" "$MONITOR_BASE_URL/api/agent/report")
[ "$invalid_token_status" = "401" ]
report_response=$(request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
-H 'CF-Connecting-IP: 8.8.8.8' \
--data "$report" "$MONITOR_BASE_URL/api/agent/report")
printf '%s' "$report_response" | jq -e \
--arg task_id "$latency_task_id" \
'.collect_interval == 5 and .agent_mirror == "https://mirror.example.com" and .auto_update == true and (.latency_tasks | any(.id == $task_id and .task_type == "tcp" and .target == "example.com" and .port == 443))' >/dev/null
request -H "Authorization: Bearer $agent_token" \
"$MONITOR_BASE_URL/api/agent/config" | jq -e \
--arg task_id "$latency_task_id" \
'.collect_interval == 5 and .agent_mirror == "https://mirror.example.com" and (.latency_tasks | any(.id == $task_id))' >/dev/null
legacy_report_status=$(monitor_curl --silent --output /dev/null --write-out '%{http_code}' \
-X POST -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data '{}' "$MONITOR_BASE_URL/api/agent/report")
[ "$legacy_report_status" = "404" ]
for legacy_path in /api/agent/config /api/agent/live; do
legacy_status=$(monitor_curl --silent --output /dev/null --write-out '%{http_code}' \
-H "Authorization: Bearer $agent_token" "$MONITOR_BASE_URL$legacy_path")
[ "$legacy_status" = "404" ]
done
request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/admin/servers" | \
jq -e --arg id "$server_id" '.servers | any(.id == $id and .last_ip == "8.8.8.8")' >/dev/null
latency_report=$(printf '%s' "$report" | jq --arg task_id "$latency_task_id" \
'.samples[0].timestamp += 1 |
.samples[0].latency_results=[{
task_id:$task_id,
timestamp:.samples[0].timestamp,
latency_ms:28.4,
packet_loss:25
}]')
request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data "$latency_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null
latency_report_second=$(printf '%s' "$report" | jq --arg task_id "$latency_task_id" \
'.samples[0].timestamp += 2 |
.samples[0].latency_results=[{
task_id:$task_id,
timestamp:.samples[0].timestamp,
latency_ms:48.4,
packet_loss:75
}]')
request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data "$latency_report_second" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null
# A whole Agent batch must collapse into one traffic-cycle write while preserving
# counter resets. Replaying the same batch must not count any bytes twice.
traffic_report=$(printf '%s' "$report" | jq '.samples = [
(.samples[0] | .timestamp += 3 | .net_rx_total=2147483648 | .net_tx_total=1073741824),
(.samples[0] | .timestamp += 4 | .net_rx_total=268435456 | .net_tx_total=134217728),
(.samples[0] | .timestamp += 5 | .net_rx_total=536870912 | .net_tx_total=268435456)
]')
request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data "$traffic_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null
request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data "$traffic_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null
request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/bootstrap" | jq -e --arg id "$server_id" --arg task_id "$latency_task_id" '.servers | any(.id == $id and .cpu == 18.5 and .gpu_usage == 32.5 and .disk_await_ms == 1.4 and (.gpus | length) == 1 and (.disks | length) == 1 and .disk_used == 21474836480 and .traffic_limit == 107374182400 and .net_rx_total == 2684354560 and .net_tx_total == 1342177280 and .price == 9.9 and (has("last_ip") | not) and (.latency | any(.task_id == $task_id and .latency_ms == 48.4 and .packet_loss == 75)))' >/dev/null
# Changing the reset day starts a new traffic cycle. Only growth after the first
# sample in that cycle counts toward usage.
request -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/json' -X PATCH \
--data "$(printf '%s' "$server_input" | jq '.reset_day=2')" \
"$MONITOR_BASE_URL/api/admin/servers/$server_id" >/dev/null
cycle_report=$(printf '%s' "$report" | jq '.samples = [
(.samples[0] | .timestamp += 6 | .net_rx_total=805306368 | .net_tx_total=536870912),
(.samples[0] | .timestamp += 7 | .net_rx_total=1073741824 | .net_tx_total=805306368)
]')
request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data "$cycle_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null
request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/bootstrap" | \
jq -e --arg id "$server_id" '.servers | any(.id == $id and .net_rx_total == 268435456 and .net_tx_total == 268435456)' >/dev/null
# Exercise the protocol maximum so batch aggregation and request-size limits
# stay covered when an Agent flushes a full pending queue.
stress_report_file=$(mktemp "${TMPDIR:-/tmp}/nodeflare-stress.XXXXXX")
printf '%s' "$report" | jq '(now | floor) as $start | .samples = [
range(0; 720) as $index |
(.samples[0] |
.timestamp=($start + $index + 10) |
.net_rx_total=(1073741824 + (($index + 1) * 4096)) |
.net_tx_total=(805306368 + (($index + 1) * 2048)))
]' >"$stress_report_file"
request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data-binary "@$stress_report_file" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null
request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/history/$server_id?hours=1" | jq -e '.points | length >= 1 and any(.gpu_usage == 32.5)' >/dev/null
history_cache_header=$(monitor_curl --silent --dump-header - --output /dev/null \
-H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/history/$server_id?hours=1" | \
@@ -223,10 +154,10 @@ request -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/js
--data '{"name":"Smoke TCP","task_type":"tcp","target":"example.com","port":443,"interval_seconds":60,"default_enabled":true,"server_ids":[]}' \
"$MONITOR_BASE_URL/api/admin/latency-tasks/$latency_task_id" >/dev/null
request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/latency/$server_id?hours=1" | jq -e --arg task_id "$latency_task_id" '(.tasks | all(.id != $task_id)) and (.points | all(.task_id != $task_id))' >/dev/null
stale_latency_response=$(request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \
--data "$latency_report" "$MONITOR_BASE_URL/api/agent/report")
printf '%s' "$stale_latency_response" | jq -e --arg task_id "$latency_task_id" \
'.latency_tasks | all(.id != $task_id)' >/dev/null
MONITOR_BASE_URL="$MONITOR_BASE_URL" MONITOR_ADMIN_TOKEN="$admin_token" \
MONITOR_AGENT_TOKEN="$agent_token" MONITOR_SERVER_ID="$server_id" \
MONITOR_LATENCY_TASK_ID="$latency_task_id" MONITOR_EXPECT_TASK_ASSIGNED=0 \
MONITOR_CONFIG_ONLY=1 node scripts/websocket-smoke.mjs
alert_rule_json=$(request -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/json' \
--data "$(jq -nc --arg server_id "$server_id" '{name:"Smoke CPU",metric:"cpu",threshold:80,duration_minutes:5,aggregation:"average",enabled:true,server_ids:[$server_id]}')" \
scripts/websocket-smoke.mjs
+246 -23
View File
@@ -3,9 +3,15 @@ import WebSocket from "ws";
const baseUrl = process.env.MONITOR_BASE_URL;
const adminToken = process.env.MONITOR_ADMIN_TOKEN;
const agentToken = process.env.MONITOR_AGENT_TOKEN;
const serverId = process.env.MONITOR_SERVER_ID;
const latencyTaskId = process.env.MONITOR_LATENCY_TASK_ID;
const expectTaskAssigned = process.env.MONITOR_EXPECT_TASK_ASSIGNED !== "0";
const configOnly = process.env.MONITOR_CONFIG_ONLY === "1";
if (!baseUrl || !adminToken || !agentToken) {
throw new Error("MONITOR_BASE_URL, MONITOR_ADMIN_TOKEN and MONITOR_AGENT_TOKEN are required");
if (!baseUrl || !adminToken || !agentToken || !serverId || !latencyTaskId) {
throw new Error(
"MONITOR_BASE_URL, MONITOR_ADMIN_TOKEN, MONITOR_AGENT_TOKEN, MONITOR_SERVER_ID and MONITOR_LATENCY_TASK_ID are required",
);
}
function websocketUrl(path) {
@@ -14,10 +20,10 @@ function websocketUrl(path) {
return url.toString();
}
function openSocket(path, token) {
function openSocket(path, token, headers = {}) {
return new Promise((resolve, reject) => {
const socket = new WebSocket(websocketUrl(path), {
headers: { Authorization: `Bearer ${token}` },
headers: { Authorization: `Bearer ${token}`, ...headers },
});
const timer = setTimeout(() => {
socket.terminate();
@@ -38,6 +44,36 @@ function openSocket(path, token) {
});
}
function expectSocketStatus(path, token, statusCode) {
return new Promise((resolve, reject) => {
const socket = new WebSocket(websocketUrl(path), {
headers: { Authorization: `Bearer ${token}` },
});
const timer = setTimeout(() => {
socket.terminate();
reject(new Error(`${path} did not reject the WebSocket handshake`));
}, 5_000);
socket.once("open", () => {
clearTimeout(timer);
socket.terminate();
reject(new Error(`${path} unexpectedly accepted the WebSocket handshake`));
});
socket.once("unexpected-response", (_request, response) => {
clearTimeout(timer);
response.resume();
if (response.statusCode === statusCode) {
resolve();
} else {
reject(new Error(`${path} returned HTTP ${response.statusCode}, expected ${statusCode}`));
}
});
socket.once("error", (error) => {
clearTimeout(timer);
reject(error);
});
});
}
function waitForMessage(socket, expected) {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => reject(new Error(`WebSocket did not receive ${expected}`)), 5_000);
@@ -54,29 +90,216 @@ function waitForMessage(socket, expected) {
});
}
function closeSocket(socket) {
function waitForJsonMessage(socket, expected, predicate) {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
socket.terminate();
reject(new Error("WebSocket close timed out"));
}, 5_000);
socket.once("close", () => {
clearTimeout(timer);
resolve();
});
socket.once("error", (error) => {
const cleanup = () => {
clearTimeout(timer);
socket.off("message", onMessage);
socket.off("close", onClose);
socket.off("error", onError);
};
const fail = (error) => {
cleanup();
reject(error);
});
socket.close();
};
const onMessage = (data) => {
let message;
try {
message = JSON.parse(data.toString());
} catch {
return;
}
if (!predicate(message)) return;
cleanup();
resolve(message);
};
const onClose = () => fail(new Error(`WebSocket closed before receiving ${expected}`));
const onError = (error) => fail(error);
const timer = setTimeout(
() => fail(new Error(`WebSocket did not receive ${expected}`)),
5_000,
);
socket.on("message", onMessage);
socket.once("close", onClose);
socket.once("error", onError);
});
}
const dashboard = await openSocket("/api/ws", adminToken);
const pong = waitForMessage(dashboard, "pong");
dashboard.send("ping");
await pong;
await closeSocket(dashboard);
function closeSocket(socket) {
return new Promise((resolve, reject) => {
if (socket.readyState === WebSocket.CLOSED) {
resolve();
return;
}
const cleanup = () => {
clearTimeout(timer);
socket.off("close", onClose);
socket.off("error", onError);
};
const onClose = () => {
cleanup();
resolve();
};
const onError = (error) => {
cleanup();
reject(error);
};
const timer = setTimeout(() => {
cleanup();
socket.terminate();
reject(new Error("WebSocket close timed out"));
}, 5_000);
socket.once("close", onClose);
socket.once("error", onError);
if (socket.readyState === WebSocket.OPEN) socket.close();
});
}
const agent = await openSocket("/api/agent/live", agentToken);
await closeSocket(agent);
await expectSocketStatus("/api/agent/ws", "invalid-agent-token", 401);
const dashboard = configOnly ? null : await openSocket("/api/ws", adminToken);
let agent;
try {
if (dashboard) {
const pong = waitForMessage(dashboard, "pong");
dashboard.send("ping");
await pong;
}
agent = await openSocket("/api/agent/ws", agentToken, { "CF-Connecting-IP": "8.8.8.8" });
const config = await waitForJsonMessage(
agent,
"Agent config",
(message) => message.type === "config" && message.config,
);
const hasTask = config.config.latency_tasks?.some(
(task) =>
task.id === latencyTaskId &&
task.task_type === "tcp" &&
task.target === "example.com" &&
task.port === 443,
);
if (
config.config.collect_interval !== 5 ||
config.config.agent_mirror !== "https://mirror.example.com" ||
config.config.auto_update !== true ||
hasTask !== expectTaskAssigned
) {
throw new Error(`Invalid Agent config: ${JSON.stringify(config)}`);
}
if (!configOnly) {
const timestamp = Math.floor(Date.now() / 1_000);
const baseSample = {
timestamp,
cpu: 18.5,
load1: 0.42,
load5: 0.36,
load15: 0.31,
mem_used: 2147483648,
mem_total: 4294967296,
swap_used: 0,
swap_total: 0,
disk_used: 21474836480,
disk_total: 53687091200,
net_in: 4096,
net_out: 2048,
net_rx_total: 1073741824,
net_tx_total: 536870912,
uptime: 86400,
processes: 90,
tcp_connections: 18,
udp_connections: 4,
cpu_cores: 2,
cpu_model: "Smoke CPU",
os: "Debian 12",
kernel: "6.1",
arch: "x86_64",
virtualization: "kvm",
gpu_usage: 32.5,
gpu_model: "NVIDIA T4",
agent_version: "smoke",
disk_read_bps: 4194304,
disk_write_bps: 2097152,
disk_read_iops: 120,
disk_write_iops: 48,
disk_await_ms: 1.4,
disk_utilization: 8.2,
disks: [
{
name: "/dev/vda1",
mount_point: "/",
used: 21474836480,
total: 53687091200,
read_bps: 4194304,
write_bps: 2097152,
read_iops: 120,
write_iops: 48,
await_ms: 1.4,
utilization: 8.2,
},
],
gpus: [
{ model: "NVIDIA T4", usage: 32.5, memory_used: 1073741824, memory_total: 17179869184 },
],
latency_results: [],
};
const sample = (offset, overrides = {}) => ({
...structuredClone(baseSample),
timestamp: timestamp + offset,
...overrides,
});
const samples = [
sample(0),
sample(1, {
latency_results: [
{ task_id: latencyTaskId, timestamp: timestamp + 1, latency_ms: 28.4, packet_loss: 25 },
],
}),
sample(2, {
latency_results: [
{ task_id: latencyTaskId, timestamp: timestamp + 2, latency_ms: 48.4, packet_loss: 75 },
],
}),
sample(3, { net_rx_total: 2147483648, net_tx_total: 1073741824 }),
sample(4, { net_rx_total: 268435456, net_tx_total: 134217728 }),
sample(5, { net_rx_total: 536870912, net_tx_total: 268435456 }),
];
const latestTimestamp = timestamp + 5;
const ackPromise = waitForJsonMessage(
agent,
"Agent metric ACK",
(message) => message.type === "ack" && message.realtimeHint === false,
);
const updatePromise = waitForJsonMessage(
dashboard,
"dashboard batchUpdate",
(message) =>
message.type === "batchUpdate" &&
message.updates?.some(
(update) =>
update.serverId === serverId &&
update.samples?.some(
(entry) => entry.ts === latestTimestamp && entry.data?.cpu === 18.5,
),
),
);
agent.send(JSON.stringify({ type: "update", batchId: `smoke-${timestamp}`, samples }));
const [ack] = await Promise.all([ackPromise, updatePromise]);
if (
ack.ts <= 0 ||
ack.persisted !== true ||
ack.persistenceError !== false ||
!Number.isInteger(ack.persistedThroughTs) ||
ack.persistedThroughTs < latestTimestamp ||
ack.nextD1WriteAfterMs !== 120_000 ||
ack.nextWssReportAfterMs !== 5_000
) {
throw new Error(`Invalid Agent metric ACK: ${JSON.stringify(ack)}`);
}
}
} finally {
await Promise.all([
agent ? closeSocket(agent) : Promise.resolve(),
dashboard ? closeSocket(dashboard) : Promise.resolve(),
]);
}