scripts: consolidate Agent deployment and remove Probe packages
Use Agent for native, container, systemd and OpenWrt installations and signed updates. Remove retired Probe deployment templates and smoke scripts.
This commit is contained in:
1 parent
ad5efc62ff
commit
016dbc2bbf
30 files changed
+96
-1206
No files matched your search
+8
-6
@@ -1,12 +1,13 @@
|
||||
#!/usr/bin/env bash
|
||||
# 打包:scripts/package.sh <agent|probe> <rid> → backend/artifacts/packages/moshiro-<role>-<version>-<rid>.tar.gz(附 .sha256)
|
||||
# 打包:scripts/package.sh <agent> <rid> → backend/artifacts/packages/moshiro-<role>-<version>-<rid>.tar.gz(附 .sha256)
|
||||
# 内容:节点二进制(agent 还带 fleet-enroll)、systemd 单元、OpenWrt procd 脚本、install.sh。
|
||||
# 发布产物要先用 scripts/publish.sh 生成(agent 需要 agent 和 fleet-enroll 两个)。
|
||||
set -euo pipefail
|
||||
role="${1:?usage: package.sh <agent|probe> <rid>}"
|
||||
rid="${2:?usage: package.sh <agent|probe> <rid>}"
|
||||
role="${1:?usage: package.sh <agent> <rid>}"
|
||||
rid="${2:?usage: package.sh <agent> <rid>}"
|
||||
[[ "$role" == agent ]] || { echo "only agent packages are supported" >&2; exit 2; }
|
||||
root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
version="$(sed -n 's#.*<Version>\(.*\)</Version>.*#\1#p' "$root/Directory.Build.props" | head -1)"
|
||||
version="$(sed -n 's#.*<Version>\(.*\)</Version>.*#\1#p' "$root/backend/Directory.Build.props" | head -1)"
|
||||
name="moshiro-$role-$version-$rid"
|
||||
stage="$root/backend/artifacts/packages/stage/$name"
|
||||
quic_runtime="$root/backend/artifacts/packages/quic-runtime/$rid"
|
||||
@@ -27,10 +28,11 @@ fi
|
||||
cp "$root/backend/artifacts/publish/$role/$rid/moshiro-$role" "$stage/"
|
||||
cp "$root/packaging/install.sh" "$stage/"
|
||||
cp "$root/packaging/systemd/moshiro-$role.service" "$stage/systemd/"
|
||||
cp "$root/$role/README.md" "$stage/README.md"
|
||||
cp "$root/backend/$role/README.md" "$stage/README.md"
|
||||
if [ "$role" = agent ]; then
|
||||
cp "$root/backend/artifacts/publish/fleet-enroll/$rid/fleet-enroll" "$stage/"
|
||||
else
|
||||
fi
|
||||
if [[ "$rid" == linux-* ]]; then
|
||||
mkdir -p "$stage/openwrt"
|
||||
cp "$root/packaging/openwrt/"* "$stage/openwrt/"
|
||||
fi
|
||||
|
||||
+3
-4
@@ -1,16 +1,15 @@
|
||||
#!/usr/bin/env bash
|
||||
# 发布一个节点程序:scripts/publish.sh <agent|probe|backend|fleet-enroll> <rid>
|
||||
# 发布一个节点程序:scripts/publish.sh <agent|backend|fleet-enroll> <rid>
|
||||
# 输出到 backend/artifacts/publish/<project>/<rid>/。linux-musl-x64(Alpine)和 linux-x64(Ubuntu noble,glibc)在 NativeAOT SDK 容器里构建(任何主机都可以,包括 Windows),
|
||||
# 其他 rid 用本机的 dotnet(NativeAOT 不能交叉编译到别的操作系统,所以 win-x64 要在 Windows 上、linux-x64 要在 glibc 的 Linux 上构建)。
|
||||
set -euo pipefail
|
||||
|
||||
project="${1:?usage: publish.sh <agent|probe|backend|fleet-enroll> <rid>}"
|
||||
rid="${2:?usage: publish.sh <agent|probe|backend|fleet-enroll> <rid>}"
|
||||
project="${1:?usage: publish.sh <agent|backend|fleet-enroll> <rid>}"
|
||||
rid="${2:?usage: publish.sh <agent|backend|fleet-enroll> <rid>}"
|
||||
root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
|
||||
case "$project" in
|
||||
agent) csproj="backend/agent/src/MoShiro.Agent" ;;
|
||||
probe) csproj="backend/probe/src/MoShiro.Probe" ;;
|
||||
backend) csproj="backend/src/MoShiro.Backend" ;;
|
||||
fleet-enroll) csproj="backend/fleet/tools/fleet-enroll" ;;
|
||||
*) echo "unknown project: $project" >&2; exit 2 ;;
|
||||
|
||||
@@ -1,86 +0,0 @@
|
||||
#!/bin/sh
|
||||
# 在 Linux 容器里用发布出来的 NativeAOT 二进制跑一遍完整链路:后端 → 注册 Agent → Agent → Probe → 数据落库。
|
||||
#
|
||||
# RID=linux-musl-x64 IMAGE=alpine:3.22 (默认)
|
||||
# RID=linux-x64 IMAGE=debian:bookworm-slim
|
||||
# MODE=install 用打包产物里的 install.sh 安装(agent 注册、probe 身份、以 moshiro 用户运行),而不是直接运行二进制
|
||||
#
|
||||
# 用法(在仓库根目录;Windows 下先 export MSYS_NO_PATHCONV=1;需先运行 scripts/package.sh):
|
||||
# docker run --rm -e RID -e MODE -v "$PWD/backend/artifacts:/art:ro" -v "$PWD/scripts/smoke-linux.sh:/smoke.sh:ro" $IMAGE sh /smoke.sh [秒数]
|
||||
set -eu
|
||||
seconds="${1:-60}"
|
||||
rid="${RID:-linux-musl-x64}"
|
||||
mode="${MODE:-direct}"
|
||||
pub=/art/publish
|
||||
|
||||
if command -v apk >/dev/null 2>&1; then
|
||||
apk add --no-cache curl jq libssl3 libstdc++ >/dev/null
|
||||
else
|
||||
apt-get update >/dev/null && apt-get install -y --no-install-recommends curl jq ca-certificates libssl3 >/dev/null
|
||||
fi
|
||||
|
||||
work=/tmp/smoke
|
||||
mkdir -p "$work/data" "$work/probe"
|
||||
base=http://127.0.0.1:18080
|
||||
|
||||
export ASPNETCORE_URLS="$base" MoShiro__DataDirectory="$work/data" MoShiro__InitialAdminUser=admin \
|
||||
MoShiro__InitialAdminPassword="correct horse battery" MoShiro__SecureCookies=false
|
||||
(cd "$pub/backend/$rid" && ./moshiro-backend >"$work/backend.log" 2>&1 &)
|
||||
for i in $(seq 1 60); do curl -fs "$base/healthz" >/dev/null 2>&1 && break; sleep 1; done
|
||||
curl -fs "$base/healthz" >/dev/null || { cat "$work/backend.log"; echo "backend did not start"; exit 1; }
|
||||
echo "backend up"
|
||||
|
||||
login=$(curl -si -X POST "$base/api/admin/auth/login" -H 'Content-Type: application/json' -d '{"username":"admin","password":"correct horse battery"}')
|
||||
cookie=$(echo "$login" | grep -i '^set-cookie' | head -1 | sed 's/^[^:]*: //; s/;.*//' | tr -d '\r')
|
||||
csrf=$(echo "$login" | tail -1 | jq -r .csrf_token)
|
||||
token=$(curl -fs -X POST "$base/api/admin/api-tokens" -H "Cookie: $cookie" -H "X-CSRF-Token: $csrf" -H 'Content-Type: application/json' -d '{"name":"smoke","scope":"admin"}' | jq -r .plaintext)
|
||||
auth="Authorization: Bearer $token"
|
||||
join=$(curl -fs -X POST "$base/api/admin/join-tokens" -H "$auth" -H 'Content-Type: application/json' -d '{"auto_approve":true,"max_uses":1,"expires_in_hours":1}' | jq -r .plaintext)
|
||||
probe_join=$(curl -fs -X POST "$base/api/admin/join-tokens" -H "$auth" -H 'Content-Type: application/json' -d '{"role":"probe","max_uses":1,"expires_in_hours":1}' | jq -r .plaintext)
|
||||
|
||||
if [ "$mode" = install ]; then
|
||||
# 安装到 /usr/lib/moshiro 和 /etc/moshiro,以 moshiro 用户运行(没有 systemd:自己启动)
|
||||
mkdir -p /tmp/pkg && for role in agent probe; do tar -C /tmp/pkg -xzf /art/packages/moshiro-$role-*-$rid.tar.gz; done
|
||||
(cd /tmp/pkg/moshiro-agent-* && ./install.sh agent --server "$base" --join "$join" --name smoke-agent --no-start)
|
||||
ls -l /usr/lib/moshiro /etc/moshiro
|
||||
run_as() { su -s /bin/sh moshiro -c "exec $*"; }
|
||||
run_as /usr/lib/moshiro/moshiro-agent --config /etc/moshiro/agent.json >"$work/agent.log" 2>&1 &
|
||||
agent_pid=$!
|
||||
sleep 3
|
||||
(cd /tmp/pkg/moshiro-probe-* && ./install.sh probe --server "$base" --join "$probe_join" --name smoke-probe --no-start)
|
||||
run_as /usr/lib/moshiro/moshiro-probe --config /etc/moshiro/probe.json >"$work/probe.log" 2>&1 &
|
||||
probe_pid=$!
|
||||
else
|
||||
for role in agent probe; do
|
||||
package_dir="$work/$role-bin"
|
||||
mkdir -p "$package_dir"
|
||||
set -- /art/packages/moshiro-"$role"-*-"$rid".tar.gz
|
||||
[ -f "$1" ] || { echo "missing QUIC-bundled package: $1 (run scripts/package.sh first)" >&2; exit 1; }
|
||||
tar -xzf "$1" -C "$package_dir"
|
||||
done
|
||||
mkdir -p "$work/agent-id"
|
||||
"$work/agent-bin/fleet-enroll" --server "$base" --join "$join" --name smoke-agent --identity-dir "$work/agent-id"
|
||||
printf '{"fleet":{"server":"%s","identity_dir":"%s"},"listen":{"bind":"127.0.0.1"}}' "$base" "$work/agent-id" >"$work/agent.json"
|
||||
"$work/agent-bin/moshiro-agent" --config "$work/agent.json" >"$work/agent.log" 2>&1 &
|
||||
agent_pid=$!
|
||||
sleep 3
|
||||
mkdir -p "$work/probe-id"
|
||||
"$work/probe-bin/fleet-enroll" --server "$base" --join "$probe_join" --name smoke-probe --identity-dir "$work/probe-id"
|
||||
printf '{"fleet":{"server":"%s","identity_dir":"%s"}}' "$base" "$work/probe-id" >"$work/probe/probe.json"
|
||||
"$work/probe-bin/moshiro-probe" --config "$work/probe/probe.json" >"$work/probe.log" 2>&1 &
|
||||
probe_pid=$!
|
||||
fi
|
||||
|
||||
rss() { awk '/VmRSS/ {printf "%.1f", $2/1024}' "/proc/$1/status"; }
|
||||
i=0
|
||||
while [ "$i" -lt "$seconds" ]; do
|
||||
sleep 10; i=$((i + 10))
|
||||
echo "t+${i}s probe=$(rss "$probe_pid")MB agent=$(rss "$agent_pid")MB"
|
||||
done
|
||||
|
||||
echo "agents: $(curl -fs "$base/api/admin/agents" -H "$auth" | jq -c '[.items[] | {name, status, probe_connected}]')"
|
||||
echo "chunks: $(curl -fs "$base/api/admin/raw/chunks" -H "$auth" | jq -c '[.items[].kind] | group_by(.) | map({(.[0]): length}) | add')"
|
||||
agent_id=$(curl -fs "$base/api/admin/agents" -H "$auth" | jq -r '.items[0].id')
|
||||
echo "tcp summary: $(curl -fs "$base/api/public/lines/$agent_id/series?range=5m&step=1s&metrics=tcp.rtt.p50,tcp.rtt.p999,tcp.rtt.max,tcp.rtt.tail,tcp.slow,tcp.retrans,tcp.srtt.p95" -H "$auth" | jq -c '.summary.metrics')"
|
||||
echo "--- probe.log"; tail -4 "$work/probe.log"
|
||||
echo "--- agent.log"; tail -4 "$work/agent.log"
|
||||
@@ -1,31 +0,0 @@
|
||||
# 在带 systemd 的 Debian 容器里用 install.sh 安装并以 systemd 单元运行(验证沙箱指令与 NativeAOT 兼容)。
|
||||
# docker run -d --name sysd --privileged --cgroupns=host -v /sys/fs/cgroup:/sys/fs/cgroup:rw -v "$PWD/backend/artifacts:/art:ro" jrei/systemd-debian:12
|
||||
# docker cp scripts/smoke-systemd.sh sysd:/s.sh && docker exec sysd bash /s.sh
|
||||
set -eu
|
||||
apt-get update >/dev/null && apt-get install -y --no-install-recommends curl jq ca-certificates libssl3 >/dev/null
|
||||
base=http://127.0.0.1:18080
|
||||
mkdir -p /tmp/data
|
||||
cd /art/publish/backend/linux-x64
|
||||
ASPNETCORE_URLS=$base MoShiro__DataDirectory=/tmp/data MoShiro__InitialAdminUser=admin MoShiro__InitialAdminPassword="correct horse battery" MoShiro__SecureCookies=false nohup ./moshiro-backend >/tmp/backend.log 2>&1 &
|
||||
for i in $(seq 1 60); do curl -fs $base/healthz >/dev/null 2>&1 && break; sleep 1; done
|
||||
login=$(curl -si -X POST $base/api/admin/auth/login -H 'Content-Type: application/json' -d '{"username":"admin","password":"correct horse battery"}')
|
||||
cookie=$(echo "$login" | grep -i '^set-cookie' | head -1 | sed 's/^[^:]*: //; s/;.*//' | tr -d '\r')
|
||||
csrf=$(echo "$login" | tail -1 | jq -r .csrf_token)
|
||||
token=$(curl -fs -X POST $base/api/admin/api-tokens -H "Cookie: $cookie" -H "X-CSRF-Token: $csrf" -H 'Content-Type: application/json' -d '{"name":"t","scope":"admin"}' | jq -r .plaintext)
|
||||
auth="Authorization: Bearer $token"
|
||||
join=$(curl -fs -X POST $base/api/admin/join-tokens -H "$auth" -H 'Content-Type: application/json' -d '{"auto_approve":true,"max_uses":1,"expires_in_hours":1}' | jq -r .plaintext)
|
||||
mkdir -p /tmp/pkg; for r in agent probe; do tar -C /tmp/pkg -xzf /art/packages/moshiro-$r-*-linux-x64.tar.gz; done
|
||||
(cd /tmp/pkg/moshiro-agent-* && ./install.sh agent --server $base --join "$join" --name sysd-agent)
|
||||
sleep 3
|
||||
probe_join=$(curl -fs -X POST $base/api/admin/join-tokens -H "$auth" -H 'Content-Type: application/json' -d '{"role":"probe","max_uses":1,"expires_in_hours":1}' | jq -r .plaintext)
|
||||
(cd /tmp/pkg/moshiro-probe-* && ./install.sh probe --server $base --join "$probe_join" --name sysd-probe)
|
||||
sleep 25
|
||||
systemctl is-active moshiro-agent moshiro-probe
|
||||
echo "agents: $(curl -fs $base/api/admin/agents -H "$auth" | jq -c '[.items[] | {name, status, probe_connected}]')"
|
||||
echo "chunks: $(curl -fs $base/api/admin/raw/chunks -H "$auth" | jq -c '[.items[].kind] | length')"
|
||||
systemctl show moshiro-agent -p MemoryCurrent -p NRestarts | tr '\n' ' '; systemctl show moshiro-probe -p MemoryCurrent -p NRestarts | tr '\n' ' '; echo
|
||||
ps -o user,rss,comm -C moshiro-agent,moshiro-probe
|
||||
journalctl -u moshiro-agent -u moshiro-probe --no-pager | tail -8
|
||||
echo "--- sandbox check: agent cannot write /etc/moshiro"; nsenter -t $(systemctl show moshiro-agent -p MainPID --value) -m sh -c 'touch /etc/moshiro/x 2>&1; ls /proc/self/status >/dev/null' 2>&1 | head -2
|
||||
grep -E "NoNewPrivs|CapEff|Seccomp:" /proc/$(systemctl show moshiro-agent -p MainPID --value)/status
|
||||
grep -E "NoNewPrivs|CapEff|Seccomp:" /proc/$(systemctl show moshiro-probe -p MainPID --value)/status
|
||||
@@ -1,77 +0,0 @@
|
||||
"""跨机器的真实链路验证:Windows 上的 Probe(NativeAOT exe)→ WSL2 里的 Agent 和 Backend(真实的虚拟网络、不同的单调时钟原点)。
|
||||
|
||||
前提:WSL 里有 Debian(root)、tc(netem 注入丢包/延迟/黑洞用);已经用 scripts/publish.sh 发布了 linux-x64 的 agent、backend、fleet-enroll,
|
||||
并且 Windows 上发布了 probe(dotnet publish backend/probe/src/MoShiro.Probe -c Release -r win-x64 -p:PublishAot=true -o backend/artifacts/aot-probe)。
|
||||
把 linux-x64 的产物复制到 WSL 的 /root/mon/,在 WSL 里用 setsid -f nohup 启动 moshiro-backend(监听 0.0.0.0:29181),再运行本脚本。
|
||||
故障注入见 docs/ops/deployment.md 第 3 节。
|
||||
"""
|
||||
import json, os, subprocess, sys, time, urllib.request, urllib.error
|
||||
|
||||
# 要连的是本机的虚拟网络(WSL),不能走环境变量里的 HTTP(S)_PROXY
|
||||
OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||||
|
||||
ROOT = r"C:\Users\JustMe\Desktop\moshiro-proj"
|
||||
PROBE_EXE = os.path.join(ROOT, "backend", "artifacts", "aot-probe", "moshiro-probe.exe")
|
||||
WORK = r"C:\Users\JustMe\AppData\Local\Temp\realnet"
|
||||
PORT = 29181
|
||||
|
||||
|
||||
def wsl(cmd, check=True):
|
||||
r = subprocess.run(["wsl", "-d", "Debian", "-u", "root", "-e", "bash", "-c", cmd], capture_output=True, text=True)
|
||||
if check and r.returncode != 0:
|
||||
print("WSL FAILED:", cmd, r.stdout, r.stderr)
|
||||
raise SystemExit(1)
|
||||
return r.stdout.strip()
|
||||
|
||||
|
||||
WSLIP = wsl("ip -4 addr show eth0 | grep -oP '(?<=inet )[0-9.]+'")
|
||||
BASE = f"http://{WSLIP}:{PORT}"
|
||||
|
||||
|
||||
def req(method, path, body=None, headers=None, ctype="application/json"):
|
||||
data = None if body is None else json.dumps(body).encode()
|
||||
h = {"Content-Type": ctype}
|
||||
h.update(headers or {})
|
||||
r = urllib.request.Request(BASE + path, data=data, method=method, headers=h)
|
||||
try:
|
||||
with OPENER.open(r, timeout=30) as resp:
|
||||
text = resp.read().decode()
|
||||
return resp, (json.loads(text) if text else None)
|
||||
except urllib.error.HTTPError as e:
|
||||
print("HTTP", e.code, e.read().decode())
|
||||
raise
|
||||
|
||||
|
||||
def main():
|
||||
os.makedirs(WORK, exist_ok=True)
|
||||
print("wsl ip", WSLIP)
|
||||
resp, login = req("POST", "/api/admin/auth/login", {"username": "admin", "password": os.environ.get("MOSHIRO_ADMIN_PASSWORD", "correct horse battery")})
|
||||
cookie = resp.headers["Set-Cookie"].split(";")[0]
|
||||
_, tok = req("POST", "/api/admin/api-tokens", {"name": "net", "scope": "admin"}, {"Cookie": cookie, "X-CSRF-Token": login["csrf_token"]})
|
||||
auth = {"Authorization": "Bearer " + tok["plaintext"]}
|
||||
json.dump({"base": BASE, "token": tok["plaintext"], "wslip": WSLIP}, open(WORK + r"\state.json", "w"))
|
||||
_, join = req("POST", "/api/admin/join-tokens", {"auto_approve": True, "max_uses": 1, "expires_in_hours": 1}, auth)
|
||||
|
||||
wsl("pkill moshiro-agent; cd /root/mon && rm -rf id agent.json")
|
||||
print(wsl(f"cd /root/mon && ./fleet-enroll --server http://127.0.0.1:{PORT} --join {join['plaintext']} --name wsl-agent --identity-dir /root/mon/id"))
|
||||
cfg = {"fleet": {"server": f"http://127.0.0.1:{PORT}", "identity_dir": "/root/mon/id"}, "listen": {"bind": "0.0.0.0"}}
|
||||
wsl("cat > /root/mon/agent.json <<'EOF'\n" + json.dumps(cfg) + "\nEOF")
|
||||
wsl("cd /root/mon && (setsid nohup ./moshiro-agent --config /root/mon/agent.json > agent.log 2>&1 < /dev/null &)")
|
||||
time.sleep(3)
|
||||
_, agents = req("GET", "/api/admin/agents", None, auth)
|
||||
agent = agents["items"][0]
|
||||
print("agent", agent["id"], agent["status"], "online", agent.get("online"))
|
||||
# the backend only sees 127.0.0.1 as the agent's address: the panel needs the address the probe can reach
|
||||
req("PATCH", "/api/admin/agents/" + agent["id"], {"public_host": WSLIP}, auth, "application/merge-patch+json")
|
||||
_, ident = req("POST", "/api/admin/probes", {"name": "home"}, auth)
|
||||
print("bootstrap", ident["bootstrap"])
|
||||
os.makedirs(WORK + r"\probe", exist_ok=True)
|
||||
json.dump(ident, open(WORK + r"\probe\probe.json", "w"))
|
||||
json.dump({"base": BASE, "token": tok["plaintext"], "wslip": WSLIP, "agent_id": agent["id"]}, open(WORK + r"\state.json", "w"))
|
||||
log = open(WORK + r"\probe.log", "w")
|
||||
p = subprocess.Popen([PROBE_EXE, "--config", WORK + r"\probe\probe.json"], stdout=log, stderr=subprocess.STDOUT)
|
||||
open(WORK + r"\probe.pid", "w").write(str(p.pid))
|
||||
print("probe pid", p.pid)
|
||||
|
||||
|
||||
main()
|
||||
@@ -1,613 +0,0 @@
|
||||
"""
|
||||
稳定性实测:真实的初始化、正常关闭、崩溃关闭、链路故障与恢复。
|
||||
|
||||
拓扑:Windows 上的 Probe(NativeAOT exe)→ WSL2 里的 Agent → 同一个 WSL 里的 Backend。
|
||||
Probe↔Agent 走真实的虚拟网络(eth0),Agent↔Backend 走 WSL 里的回环(故障用 tc 注入在 lo 上)。
|
||||
|
||||
前提:WSL 里有 Debian(root)和 tc;已发布 linux-x64 的 agent、backend、fleet-enroll(scripts/publish.sh),
|
||||
以及 Windows 的 probe(dotnet publish backend/probe/src/MoShiro.Probe -c Release -r win-x64 -p:PublishAot=true -o backend/artifacts/aot-probe)。
|
||||
|
||||
用法:python scripts/stability_wsl.py [场景名 ...] (不带参数运行全部)
|
||||
场景:init graceful crash probe_link backend_link silent soak
|
||||
"""
|
||||
import ctypes
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
# 要连的是本机的虚拟网络(WSL),不能走环境变量里的 HTTP(S)_PROXY
|
||||
OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
PROBE_EXE = os.path.join(ROOT, "backend", "artifacts", "aot-probe", "moshiro-probe.exe")
|
||||
PUB = "/mnt/" + ROOT[0].lower() + ROOT[2:].replace("\\", "/") + "/backend/artifacts/publish"
|
||||
WORK = os.path.join(os.environ.get("TEMP", "."), "stability")
|
||||
PORT = 29181
|
||||
ADMIN_PASSWORD = os.environ.get("MOSHIRO_ADMIN_PASSWORD", "correct horse battery") # 测试用后端的初始管理员密码
|
||||
RESULTS = []
|
||||
HIDDEN = subprocess.CREATE_NO_WINDOW # 所有子进程都不弹窗
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ WSL / process helpers
|
||||
|
||||
def wsl(cmd, check=True, timeout=120):
|
||||
r = subprocess.run(["wsl", "-d", "Debian", "-u", "root", "-e", "bash", "-c", cmd], capture_output=True, text=True, timeout=timeout, creationflags=HIDDEN)
|
||||
if check and r.returncode != 0:
|
||||
raise RuntimeError(f"wsl failed ({r.returncode}): {cmd}\n{r.stdout}\n{r.stderr}")
|
||||
return r.stdout.strip()
|
||||
|
||||
|
||||
WSLIP = wsl("ip -4 addr show eth0 | grep -oP '(?<=inet )[0-9.]+'")
|
||||
BASE = f"http://{WSLIP}:{PORT}"
|
||||
TOKEN = ""
|
||||
|
||||
LAUNCHER = r"""#!/bin/sh
|
||||
# start.sh <name> <command...>: run detached, append to <name>.log, record the exit code in <name>.exit
|
||||
name=$1; shift
|
||||
rm -f /root/mon/$name.exit
|
||||
setsid -f sh -c 'cd /root/mon; "$@" >> /root/mon/'$name'.log 2>&1; echo $? > /root/mon/'$name'.exit' sh "$@" < /dev/null
|
||||
"""
|
||||
|
||||
|
||||
def now():
|
||||
return time.time()
|
||||
|
||||
|
||||
def check(name, ok, detail=""):
|
||||
RESULTS.append((name, bool(ok), detail))
|
||||
print(f" [{'PASS' if ok else 'FAIL'}] {name}" + (f" - {detail}" if detail else ""), flush=True)
|
||||
return ok
|
||||
|
||||
|
||||
def wait_until(cond, timeout, step=0.5):
|
||||
start = now()
|
||||
while now() - start < timeout:
|
||||
try:
|
||||
if cond():
|
||||
return now() - start
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(step)
|
||||
return None
|
||||
|
||||
|
||||
def api(method, path, body=None, ctype="application/json", timeout=15):
|
||||
headers = {"Content-Type": ctype}
|
||||
if TOKEN:
|
||||
headers["Authorization"] = "Bearer " + TOKEN
|
||||
data = None if body is None else json.dumps(body).encode()
|
||||
req = urllib.request.Request(BASE + path, data=data, method=method, headers=headers)
|
||||
with OPENER.open(req, timeout=timeout) as resp:
|
||||
text = resp.read().decode()
|
||||
return json.loads(text) if text else None
|
||||
|
||||
|
||||
def backend_up():
|
||||
try:
|
||||
OPENER.open(BASE + "/healthz", timeout=2)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ components
|
||||
|
||||
def pids(name):
|
||||
out = wsl(f"pgrep -x {name} || true")
|
||||
return [int(x) for x in out.split()]
|
||||
|
||||
|
||||
def rss_mb(name):
|
||||
p = pids(name)
|
||||
if not p:
|
||||
return None
|
||||
return int(wsl(f"ps -o rss= -p {p[0]}")) / 1024
|
||||
|
||||
|
||||
def start_backend():
|
||||
wsl(f"cd /root/mon && /root/mon/start.sh backend env ASPNETCORE_URLS=http://0.0.0.0:{PORT} MoShiro__DataDirectory=/root/mon/data "
|
||||
f"MoShiro__InitialAdminUser=admin 'MoShiro__InitialAdminPassword={ADMIN_PASSWORD}' MoShiro__SecureCookies=false ./moshiro-backend")
|
||||
return wait_until(backend_up, 30)
|
||||
|
||||
|
||||
def start_agent():
|
||||
wsl("/root/mon/start.sh agent ./moshiro-agent --config /root/mon/agent.json")
|
||||
return wait_until(lambda: bool(pids("moshiro-agent")), 10)
|
||||
|
||||
|
||||
def stop_proc(name, sig="TERM", timeout=30):
|
||||
"""Send a signal and wait for the process to exit; returns (seconds, exit code or None)."""
|
||||
start = now()
|
||||
wsl(f"pkill -{sig} -x {name} || true")
|
||||
gone = wait_until(lambda: not pids(name), timeout, 0.2)
|
||||
code = wsl(f"cat /root/mon/{name.replace('moshiro-', '')}.exit 2>/dev/null || true")
|
||||
return (now() - start if gone is not None else None), (int(code) if code.strip().lstrip('-').isdigit() else None)
|
||||
|
||||
|
||||
def send_ctrl_break(pid):
|
||||
"""
|
||||
Ctrl+Break to a process that owns its own (hidden) console: attach to that console, broadcast the event, detach.
|
||||
Ctrl+Break rather than Ctrl+C: a console process can inherit "ignore Ctrl+C" from its parent (it does under most job runners),
|
||||
Ctrl+Break cannot be ignored that way. .NET reports it as SIGQUIT.
|
||||
"""
|
||||
from ctypes import wintypes
|
||||
k = ctypes.WinDLL("kernel32", use_last_error=True)
|
||||
handler = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.DWORD)(lambda event: True) # swallow the event in this process
|
||||
k.FreeConsole()
|
||||
if not k.AttachConsole(pid):
|
||||
raise OSError(f"AttachConsole({pid}) failed: {ctypes.get_last_error()}")
|
||||
k.SetConsoleCtrlHandler(handler, True)
|
||||
k.GenerateConsoleCtrlEvent(1, 0)
|
||||
time.sleep(0.3)
|
||||
k.FreeConsole()
|
||||
k.SetConsoleCtrlHandler(handler, False)
|
||||
|
||||
|
||||
class Probe:
|
||||
def __init__(self):
|
||||
self.proc = None
|
||||
self.log = os.path.join(WORK, "probe.log")
|
||||
|
||||
def start(self):
|
||||
os.makedirs(WORK, exist_ok=True)
|
||||
handle = open(self.log, "a")
|
||||
# 自己的隐藏控制台:这样才能从外面给它发 Ctrl+Break
|
||||
flags = subprocess.CREATE_NO_WINDOW
|
||||
self.proc = subprocess.Popen([PROBE_EXE, "--config", os.path.join(WORK, "probe", "probe.json")], stdout=handle, stderr=subprocess.STDOUT, creationflags=flags)
|
||||
return self.proc.pid
|
||||
|
||||
def alive(self):
|
||||
return self.proc is not None and self.proc.poll() is None
|
||||
|
||||
def stop(self, mode="graceful", timeout=20):
|
||||
if not self.alive():
|
||||
return 0.0, self.proc.poll() if self.proc else None
|
||||
start = now()
|
||||
if mode == "graceful":
|
||||
send_ctrl_break(self.proc.pid)
|
||||
else:
|
||||
self.proc.kill()
|
||||
try:
|
||||
code = self.proc.wait(timeout)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.proc.kill()
|
||||
return None, None
|
||||
return now() - start, code
|
||||
|
||||
def suspend(self):
|
||||
"""Freeze the process (its TCP connections stay open and go silent, like a hung machine)."""
|
||||
handle = ctypes.windll.kernel32.OpenProcess(0x0800, False, self.proc.pid) # PROCESS_SUSPEND_RESUME
|
||||
ctypes.windll.ntdll.NtSuspendProcess(handle)
|
||||
ctypes.windll.kernel32.CloseHandle(handle)
|
||||
|
||||
def resume(self):
|
||||
handle = ctypes.windll.kernel32.OpenProcess(0x0800, False, self.proc.pid)
|
||||
ctypes.windll.ntdll.NtResumeProcess(handle)
|
||||
ctypes.windll.kernel32.CloseHandle(handle)
|
||||
|
||||
def working_set_mb(self):
|
||||
if not self.alive():
|
||||
return None
|
||||
out = subprocess.run(["powershell", "-NoProfile", "-Command", f"(Get-Process -Id {self.proc.pid}).WorkingSet64"], capture_output=True, text=True, creationflags=HIDDEN).stdout.strip()
|
||||
return int(out) / 1048576 if out else None
|
||||
|
||||
def log_text(self):
|
||||
return open(self.log, encoding="utf-8", errors="replace").read() if os.path.exists(self.log) else ""
|
||||
|
||||
|
||||
PROBE = Probe()
|
||||
AGENT_ID = ""
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ environment setup
|
||||
|
||||
def setup(fresh=True):
|
||||
"""Clean state, copy the artifacts, start backend/agent, create the probe identity. The probe is not started."""
|
||||
global TOKEN, AGENT_ID
|
||||
PROBE.stop("kill")
|
||||
wsl("pkill -9 -x moshiro-agent; pkill -9 -x moshiro-backend; tc qdisc del dev eth0 ingress 2>/dev/null; tc qdisc del dev eth0 root 2>/dev/null; tc qdisc del dev lo root 2>/dev/null; tc qdisc del dev lo ingress 2>/dev/null; true")
|
||||
if fresh:
|
||||
wsl(f"rm -rf /root/mon && mkdir -p /root/mon && cp {PUB}/backend/linux-x64/* {PUB}/agent/linux-x64/* {PUB}/fleet-enroll/linux-x64/* /root/mon/ && chmod +x /root/mon/moshiro-* /root/mon/fleet-enroll")
|
||||
wsl("cat > /root/mon/start.sh <<'EOF'\n" + LAUNCHER + "EOF\nchmod +x /root/mon/start.sh")
|
||||
if os.path.isdir(WORK):
|
||||
subprocess.run(["cmd", "/c", "rmdir", "/s", "/q", WORK], capture_output=True, creationflags=HIDDEN)
|
||||
os.makedirs(os.path.join(WORK, "probe"), exist_ok=True)
|
||||
assert start_backend() is not None, "backend did not start"
|
||||
if fresh:
|
||||
req = urllib.request.Request(BASE + "/api/admin/auth/login", data=json.dumps({"username": "admin", "password": ADMIN_PASSWORD}).encode(), headers={"Content-Type": "application/json"}, method="POST")
|
||||
with OPENER.open(req, timeout=15) as resp:
|
||||
cookie = resp.headers["Set-Cookie"].split(";")[0]
|
||||
csrf = json.loads(resp.read())["csrf_token"]
|
||||
req = urllib.request.Request(BASE + "/api/admin/api-tokens", data=json.dumps({"name": "stab", "scope": "admin"}).encode(), headers={"Content-Type": "application/json", "Cookie": cookie, "X-CSRF-Token": csrf}, method="POST")
|
||||
TOKEN = json.loads(OPENER.open(req, timeout=15).read())["plaintext"]
|
||||
join = api("POST", "/api/admin/join-tokens", {"auto_approve": True, "max_uses": 5, "expires_in_hours": 1})["plaintext"]
|
||||
wsl(f"cd /root/mon && ./fleet-enroll --server http://127.0.0.1:{PORT} --join {join} --name wsl-agent --identity-dir /root/mon/id")
|
||||
cfg = {"fleet": {"server": f"http://127.0.0.1:{PORT}", "identity_dir": "/root/mon/id"}, "listen": {"bind": "0.0.0.0"}}
|
||||
wsl("cat > /root/mon/agent.json <<'EOF'\n" + json.dumps(cfg) + "\nEOF")
|
||||
assert start_agent() is not None
|
||||
assert wait_until(lambda: api("GET", "/api/admin/agents")["items"][0]["online"], 20) is not None, "agent did not come online"
|
||||
agent = api("GET", "/api/admin/agents")["items"][0]
|
||||
AGENT_ID = agent["id"]
|
||||
if fresh:
|
||||
api("PATCH", "/api/admin/agents/" + AGENT_ID, {"public_host": WSLIP}, "application/merge-patch+json")
|
||||
ident = api("POST", "/api/admin/probes", {"name": "home"})
|
||||
json.dump(ident, open(os.path.join(WORK, "probe", "probe.json"), "w"))
|
||||
PROBE.start()
|
||||
assert wait_until(agent_view_probe_connected, 30) is not None, "probe did not connect"
|
||||
|
||||
|
||||
def agent_view():
|
||||
return api("GET", "/api/admin/agents/" + AGENT_ID)
|
||||
|
||||
|
||||
def agent_view_probe_connected():
|
||||
return agent_view()["probe_connected"]
|
||||
|
||||
|
||||
def agent_online():
|
||||
return agent_view()["online"]
|
||||
|
||||
|
||||
def metrics(rng="2m", names="loss.total,loss.up,loss.down,rtt.p50,tcp.rtt.p50,tcp.rtt.max"):
|
||||
s = api("GET", f"/api/public/lines/{AGENT_ID}/series?range={rng}&step=1s&metrics={names}")
|
||||
return s["summary"]["metrics"], s["summary"]["samples"]
|
||||
|
||||
|
||||
def gaps(minutes=3):
|
||||
"""Buckets (10 s) in the last N minutes, after the first data, whose udp sample count is below half of the expected 100."""
|
||||
s = api("GET", f"/api/public/lines/{AGENT_ID}/series?range={minutes}m&step=10s&metrics=n.udp_echo,n.tcp_echo")
|
||||
udp = s["values"]["n.udp_echo"]
|
||||
first = next((i for i, v in enumerate(udp) if v is not None and v >= 50), len(udp)) # buckets before the first data are not gaps
|
||||
return [i for i, v in enumerate(udp[:-1]) if i > first and (v is None or v < 50)], len(udp)
|
||||
|
||||
|
||||
def stored_chunks():
|
||||
return len(api("GET", f"/api/admin/raw/chunks?agent_id={AGENT_ID}&limit=1000")["items"])
|
||||
|
||||
|
||||
def healthy():
|
||||
try:
|
||||
return agent_online() and agent_view_probe_connected()
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def ensure_healthy(timeout=60):
|
||||
if not backend_up():
|
||||
start_backend()
|
||||
if not pids("moshiro-agent"):
|
||||
start_agent()
|
||||
if not PROBE.alive():
|
||||
PROBE.start()
|
||||
return wait_until(healthy, timeout)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ scenarios
|
||||
|
||||
def scenario_init():
|
||||
print("== init: cold starts in every order, bad configuration")
|
||||
# 1. everything down, then start in the opposite order: probe -> agent -> backend (each must retry until its peer is up)
|
||||
PROBE.stop("kill")
|
||||
wsl("pkill -9 -x moshiro-agent; pkill -9 -x moshiro-backend; true")
|
||||
time.sleep(1)
|
||||
PROBE.start()
|
||||
time.sleep(3)
|
||||
start_agent()
|
||||
time.sleep(3)
|
||||
t0 = now()
|
||||
start_backend()
|
||||
took = wait_until(healthy, 90)
|
||||
check("probe -> agent -> backend start order converges", took is not None, f"{took:.1f}s after the backend came up" if took else "never became healthy")
|
||||
chunks_flow = wait_until(lambda: stored_chunks() > 0, 40)
|
||||
check("data flows after the cold start", chunks_flow is not None)
|
||||
|
||||
# 2. backend first, then agent, then probe (the normal order) after a full stop
|
||||
PROBE.stop("kill")
|
||||
wsl("pkill -9 -x moshiro-agent; pkill -9 -x moshiro-backend; true")
|
||||
start_backend()
|
||||
start_agent()
|
||||
PROBE.start()
|
||||
took = wait_until(healthy, 60)
|
||||
check("backend -> agent -> probe start order converges", took is not None, f"{took:.1f}s" if took else "")
|
||||
|
||||
# 3. the probe starts when its agent list is only a stale cache (agents.json wrong): bootstrap is still tried
|
||||
# 4. bad configuration: clear errors and a non-zero exit code, no crash dump
|
||||
bad = os.path.join(WORK, "bad.json")
|
||||
open(bad, "w").write("{ not json")
|
||||
r = subprocess.run([PROBE_EXE, "--config", bad], capture_output=True, text=True, timeout=20, creationflags=HIDDEN)
|
||||
check("probe: malformed config exits 1 with a readable error", r.returncode == 1 and "startup failed" in r.stderr + r.stdout, (r.stderr + r.stdout).strip()[:120])
|
||||
r = subprocess.run([PROBE_EXE, "--config", os.path.join(WORK, "missing.json")], capture_output=True, text=True, timeout=20, creationflags=HIDDEN)
|
||||
check("probe: missing config exits 1 with the file name", r.returncode == 1 and "missing.json" in r.stderr + r.stdout, (r.stderr + r.stdout).strip()[:120])
|
||||
out = wsl("cd /root/mon && echo '{ not json' > bad.json && ./moshiro-agent --config bad.json 2>&1; echo exit=$?", check=False)
|
||||
check("agent: malformed config exits 1", "exit=1" in out and "startup failed" in out, out.strip()[-120:])
|
||||
out = wsl("cd /root/mon && printf '{\"fleet\":{\"server\":\"http://127.0.0.1:1\",\"identity_dir\":\"/nonexistent\"}}' > noid.json && ./moshiro-agent --config noid.json 2>&1; echo exit=$?", check=False)
|
||||
check("agent: no identity says to run fleet-enroll first", "exit=1" in out and "fleet-enroll" in out, out.strip()[-140:])
|
||||
out = wsl(f"cd /root/mon && ./fleet-enroll --server http://127.0.0.1:{PORT} --join bogus --name x --identity-dir /tmp/enroll-x 2>&1; echo exit=$?", check=False)
|
||||
check("fleet-enroll: a bad join token fails clearly (non-zero exit)", "exit=0" not in out, out.strip()[-140:])
|
||||
out = wsl(f"cd /root/mon && ./fleet-enroll --server http://127.0.0.1:{PORT} --join bogus --name x --identity-dir /root/mon/id 2>&1; echo exit=$?", check=False)
|
||||
check("fleet-enroll: refuses to overwrite an existing identity", "exit=0" not in out and "already enrolled" in out, out.strip()[-140:])
|
||||
ensure_healthy()
|
||||
|
||||
|
||||
def scenario_graceful():
|
||||
print("== graceful: SIGTERM / Ctrl+Break of each component, then restart")
|
||||
ensure_healthy()
|
||||
# probe (Ctrl+Break in its own console)
|
||||
secs, code = PROBE.stop("graceful")
|
||||
check("probe exits on Ctrl+Break within 5 s with code 0", secs is not None and secs < 5 and code == 0, f"{secs and round(secs, 2)}s code={code}")
|
||||
seen = wait_until(lambda: not agent_view_probe_connected(), 10)
|
||||
check("backend sees the probe disconnect", seen is not None, f"{seen and round(seen, 1)}s")
|
||||
PROBE.start()
|
||||
check("probe restarts and reconnects", wait_until(healthy, 30) is not None)
|
||||
|
||||
# agent
|
||||
secs, code = stop_proc("moshiro-agent", "TERM")
|
||||
log = wsl("tail -5 /root/mon/agent.log")
|
||||
check("agent exits on SIGTERM within 5 s with code 0", secs is not None and secs < 5 and code == 0, f"{secs and round(secs, 2)}s code={code}")
|
||||
check("agent logs a clean stop (no crash trace)", "stopping" in log and "Unhandled" not in log, log.splitlines()[-1][:100] if log else "")
|
||||
seen = wait_until(lambda: not agent_online(), 10)
|
||||
check("backend sees the agent go offline", seen is not None, f"{seen and round(seen, 1)}s")
|
||||
start_agent()
|
||||
check("agent restarts, reconnects to backend and probe", wait_until(healthy, 40) is not None)
|
||||
|
||||
# backend
|
||||
secs, code = stop_proc("moshiro-backend", "TERM", 30)
|
||||
log = wsl("tail -5 /root/mon/backend.log")
|
||||
check("backend exits on SIGTERM within 15 s with code 0", secs is not None and secs < 15 and code == 0, f"{secs and round(secs, 2)}s code={code}")
|
||||
check("backend log has no unhandled exception", "Unhandled" not in log and "Exception" not in log, log.replace(chr(10), ' | ')[-160:])
|
||||
time.sleep(5)
|
||||
start_backend()
|
||||
check("backend restarts, agent and probe converge again", wait_until(healthy, 60) is not None)
|
||||
ensure_healthy()
|
||||
|
||||
|
||||
def scenario_crash():
|
||||
print("== crash: kill -9 / taskkill /F of each component, data durability")
|
||||
ensure_healthy()
|
||||
time.sleep(25)
|
||||
before = stored_chunks()
|
||||
|
||||
# backend crash: the agent keeps its outbox; after restart nothing acknowledged is lost
|
||||
t = now()
|
||||
wsl("pkill -9 -x moshiro-backend")
|
||||
time.sleep(25) # agent outbox accumulates (2-3 chunk rounds)
|
||||
start_backend()
|
||||
back = wait_until(healthy, 60)
|
||||
check("backend kill -9: agent reconnects and everything converges", back is not None, f"{back and round(back, 1)}s after restart")
|
||||
time.sleep(20)
|
||||
after = stored_chunks()
|
||||
g, n = gaps(3)
|
||||
check("backend kill -9: chunks buffered by the agent are delivered after the restart (no gap)", not g, f"gaps in {n} buckets: {g}; chunks {before}->{after}")
|
||||
|
||||
# agent crash
|
||||
wsl("pkill -9 -x moshiro-agent")
|
||||
time.sleep(5)
|
||||
start_agent()
|
||||
check("agent kill -9: recovers", wait_until(healthy, 40) is not None)
|
||||
|
||||
# probe crash
|
||||
PROBE.stop("kill")
|
||||
time.sleep(5)
|
||||
PROBE.start()
|
||||
check("probe kill: recovers", wait_until(healthy, 40) is not None)
|
||||
time.sleep(25)
|
||||
m, samples = metrics("30s")
|
||||
check("after all three crashes the line reports correct data (no phantom loss)", (m.get("loss.up") or 0) < 1 and (m.get("loss.down") or 0) < 1, f"loss.up={m.get('loss.up')} loss.down={m.get('loss.down')}")
|
||||
|
||||
# crash in the middle of an on-demand test
|
||||
test = api("POST", f"/api/admin/lines/{AGENT_ID}/tests", {"plan": "hy2_ladder", "params": {"size": 1250}})
|
||||
time.sleep(12)
|
||||
PROBE.stop("kill")
|
||||
state = wait_until(lambda: api("GET", "/api/public/tests/" + test["id"])["status"] in ("failed", "aborted"), 30)
|
||||
final = api("GET", "/api/public/tests/" + test["id"])
|
||||
check("probe crash mid-test: the test ends (failed/aborted), it does not hang", state is not None, f"status={final['status']} error={final.get('error_code')}")
|
||||
PROBE.start()
|
||||
wait_until(healthy, 40)
|
||||
test2 = api("POST", f"/api/admin/lines/{AGENT_ID}/tests", {"plan": "hy2_ladder", "params": {"size": 1250}})
|
||||
done = wait_until(lambda: api("GET", "/api/public/tests/" + test2["id"])["status"] in ("succeeded", "failed", "aborted"), 90, 2)
|
||||
check("a new test runs to completion after the crash (the agent is not stuck busy)", done is not None and api("GET", "/api/public/tests/" + test2["id"])["status"] == "succeeded", api("GET", "/api/public/tests/" + test2["id"])["status"])
|
||||
ensure_healthy()
|
||||
|
||||
|
||||
def netem(device, spec, direction="egress", ports=(7443, 7444)):
|
||||
"""Impair only the data/control ports of the agent (so the rest of WSL is unaffected)."""
|
||||
clear_netem(device)
|
||||
if direction == "egress":
|
||||
wsl(f"tc qdisc add dev {device} root handle 1: prio && tc qdisc add dev {device} parent 1:3 handle 30: netem {spec}")
|
||||
for p in ports:
|
||||
wsl(f"tc filter add dev {device} protocol ip parent 1:0 prio 3 u32 match ip sport {p} 0xffff flowid 1:3")
|
||||
else:
|
||||
wsl(f"tc qdisc add dev {device} handle ffff: ingress")
|
||||
for p in ports:
|
||||
action = "action drop" if spec == "drop" else ""
|
||||
wsl(f"tc filter add dev {device} parent ffff: protocol ip u32 match ip dport {p} 0xffff {action}")
|
||||
|
||||
|
||||
def clear_netem(device):
|
||||
wsl(f"tc qdisc del dev {device} root 2>/dev/null; tc qdisc del dev {device} ingress 2>/dev/null; true")
|
||||
|
||||
|
||||
def scenario_probe_link():
|
||||
print("== probe<->agent link: flapping, long blackhole (half-open), reconnect storms")
|
||||
ensure_healthy()
|
||||
time.sleep(20)
|
||||
mem0 = (PROBE.working_set_mb(), rss_mb("moshiro-agent"))
|
||||
|
||||
# 1. flapping: 3 s blackholes every 10 s, 8 times
|
||||
for _ in range(8):
|
||||
netem("eth0", "drop", "ingress")
|
||||
time.sleep(3)
|
||||
clear_netem("eth0")
|
||||
time.sleep(7)
|
||||
check("flapping link: probe still connected afterwards", wait_until(healthy, 30) is not None)
|
||||
|
||||
# 2. a 70 s blackhole: both ends must notice or survive, and recover once the path returns
|
||||
t = now()
|
||||
netem("eth0", "drop", "ingress")
|
||||
time.sleep(70)
|
||||
clear_netem("eth0")
|
||||
back = wait_until(healthy, 120)
|
||||
check("70 s blackhole: converges back to healthy", back is not None, f"{back and round(back, 1)}s after the path returned")
|
||||
time.sleep(35)
|
||||
m, _ = metrics("30s")
|
||||
check("after the blackhole the line is measured correctly again", (m.get("loss.total") or 0) < 2, f"loss.total={m.get('loss.total')}")
|
||||
|
||||
# 3. reconnect storm: kill the agent 12 times in a row
|
||||
for _ in range(12):
|
||||
wsl("pkill -9 -x moshiro-agent; true")
|
||||
time.sleep(0.5)
|
||||
start_agent()
|
||||
time.sleep(2)
|
||||
check("12 agent restarts in a row: converges", wait_until(healthy, 60) is not None)
|
||||
time.sleep(30)
|
||||
g, n = gaps(1)
|
||||
check("... and the data is flowing again", not g, f"gaps: {g}")
|
||||
|
||||
# 4. restart storm of the probe
|
||||
for _ in range(8):
|
||||
PROBE.stop("kill")
|
||||
time.sleep(0.3)
|
||||
PROBE.start()
|
||||
time.sleep(2)
|
||||
check("8 probe restarts in a row: converges", wait_until(healthy, 60) is not None)
|
||||
time.sleep(40)
|
||||
mem1 = (PROBE.working_set_mb(), rss_mb("moshiro-agent"))
|
||||
check("memory stays bounded after the storms", mem1[0] < mem0[0] + 15 and mem1[1] < mem0[1] + 15, f"probe {mem0[0]:.1f}->{mem1[0]:.1f} MB, agent {mem0[1]:.1f}->{mem1[1]:.1f} MB")
|
||||
ensure_healthy()
|
||||
|
||||
|
||||
def scenario_backend_link():
|
||||
print("== agent<->backend link: outage buffering, replay, duplicates")
|
||||
ensure_healthy()
|
||||
time.sleep(30)
|
||||
# an outage on loopback (tc on lo: the agent<->backend link) shorter than the buffering capacity
|
||||
netem("lo", "drop", "ingress", ports=(PORT,))
|
||||
t = now()
|
||||
time.sleep(45)
|
||||
clear_netem("lo")
|
||||
check("45 s agent<->backend outage: agent stays up and reconnects", wait_until(agent_online, 90) is not None, f"{now() - t:.0f}s")
|
||||
time.sleep(35)
|
||||
g, n = gaps(4)
|
||||
check("data produced during the outage is replayed from the agent's outbox (no gap)", not g, f"gaps in {n} buckets: {g}")
|
||||
|
||||
# duplicates: chunk ids are idempotent. The count of stored chunks grows by the expected 2 per 10 s, not more
|
||||
c0 = stored_chunks()
|
||||
time.sleep(30)
|
||||
c1 = stored_chunks()
|
||||
check("steady state: ~2 chunks / 10 s (udp + tcp), no duplicates", 4 <= c1 - c0 <= 8, f"{c1 - c0} chunks in 30 s")
|
||||
|
||||
# the backend is stopped for a while (graceful): the agent keeps collecting, delivers after
|
||||
stop_proc("moshiro-backend", "TERM", 30)
|
||||
time.sleep(40)
|
||||
start_backend()
|
||||
check("backend down 40 s then up: everything converges", wait_until(healthy, 90) is not None)
|
||||
time.sleep(35)
|
||||
g, n = gaps(4)
|
||||
check("... with no gap in the data", not g, f"gaps: {g}")
|
||||
ensure_healthy()
|
||||
|
||||
|
||||
def scenario_silent():
|
||||
print("== silent death: a frozen process keeps its sockets open and says nothing (hung machine, lost power)")
|
||||
ensure_healthy()
|
||||
time.sleep(10)
|
||||
|
||||
# a frozen probe: the agent must notice and tell the backend, instead of keeping a dead session forever
|
||||
PROBE.suspend()
|
||||
t = now()
|
||||
seen = wait_until(lambda: not agent_view_probe_connected(), 150, 1)
|
||||
check("frozen probe: the backend learns the probe is gone", seen is not None, f"{seen and round(seen)}s" if seen else "still shown as connected after 150 s")
|
||||
PROBE.resume()
|
||||
check("... and when the probe thaws it comes back", wait_until(healthy, 90) is not None)
|
||||
|
||||
# a frozen agent: the probe must drop the session and keep retrying; the backend must mark the agent offline
|
||||
wsl("pkill -STOP -x moshiro-agent")
|
||||
seen_off = wait_until(lambda: not agent_online(), 150, 1)
|
||||
check("frozen agent: the backend marks it offline", seen_off is not None, f"{seen_off and round(seen_off)}s" if seen_off else "still online after 150 s")
|
||||
wsl("pkill -CONT -x moshiro-agent")
|
||||
check("... and when the agent thaws everything converges", wait_until(healthy, 90) is not None)
|
||||
|
||||
# a frozen backend: the agent must notice (WebSocket ping timeout) and reconnect once the backend thaws
|
||||
wsl("pkill -STOP -x moshiro-backend")
|
||||
time.sleep(70)
|
||||
wsl("pkill -CONT -x moshiro-backend")
|
||||
check("frozen backend (70 s): converges after it thaws", wait_until(healthy, 90) is not None)
|
||||
time.sleep(30)
|
||||
g, n = gaps(3)
|
||||
check("... and the data produced while it was frozen is delivered", not g, f"gaps: {g}")
|
||||
ensure_healthy()
|
||||
|
||||
|
||||
def scenario_soak():
|
||||
print("== soak: 8 minutes with random faults; memory, convergence and data continuity")
|
||||
ensure_healthy()
|
||||
time.sleep(20)
|
||||
mem0 = (PROBE.working_set_mb(), rss_mb("moshiro-agent"), rss_mb("moshiro-backend"))
|
||||
import random
|
||||
rnd = random.Random(7)
|
||||
end = now() + 8 * 60
|
||||
faults = 0
|
||||
while now() < end:
|
||||
fault = rnd.choice(["blackhole", "loss", "agent_kill", "probe_kill", "none", "none"])
|
||||
faults += fault != "none"
|
||||
if fault == "blackhole":
|
||||
netem("eth0", "drop", "ingress")
|
||||
time.sleep(rnd.uniform(1, 6))
|
||||
clear_netem("eth0")
|
||||
elif fault == "loss":
|
||||
netem("eth0", "delay 20ms 5ms loss 5%", "egress")
|
||||
time.sleep(rnd.uniform(5, 12))
|
||||
clear_netem("eth0")
|
||||
elif fault == "agent_kill":
|
||||
wsl("pkill -9 -x moshiro-agent; true")
|
||||
time.sleep(1)
|
||||
start_agent()
|
||||
elif fault == "probe_kill":
|
||||
PROBE.stop("kill")
|
||||
time.sleep(0.5)
|
||||
PROBE.start()
|
||||
time.sleep(rnd.uniform(8, 20))
|
||||
clear_netem("eth0")
|
||||
check(f"after {faults} random faults everything converges", ensure_healthy(90) is not None)
|
||||
time.sleep(45)
|
||||
mem1 = (PROBE.working_set_mb(), rss_mb("moshiro-agent"), rss_mb("moshiro-backend"))
|
||||
check("memory bounded", mem1[0] < mem0[0] + 15 and mem1[1] < mem0[1] + 20 and mem1[2] < mem0[2] + 40, f"probe {mem0[0]:.0f}->{mem1[0]:.0f}, agent {mem0[1]:.0f}->{mem1[1]:.0f}, backend {mem0[2]:.0f}->{mem1[2]:.0f} MB")
|
||||
m, _ = metrics("30s")
|
||||
check("the final minute is measured correctly", (m.get("loss.up") or 0) < 1 and (m.get("loss.down") or 0) < 1, f"{m.get('loss.up')} {m.get('loss.down')}")
|
||||
|
||||
|
||||
SCENARIOS = {
|
||||
"init": scenario_init,
|
||||
"graceful": scenario_graceful,
|
||||
"crash": scenario_crash,
|
||||
"probe_link": scenario_probe_link,
|
||||
"backend_link": scenario_backend_link,
|
||||
"silent": scenario_silent,
|
||||
"soak": scenario_soak,
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
names = sys.argv[1:] or list(SCENARIOS)
|
||||
print(f"wsl ip {WSLIP}")
|
||||
setup(fresh=True)
|
||||
for name in names:
|
||||
try:
|
||||
SCENARIOS[name]()
|
||||
except Exception as ex:
|
||||
check(f"{name} ran to completion", False, repr(ex)[:300])
|
||||
PROBE.stop("kill")
|
||||
wsl("pkill -9 -x moshiro-agent; pkill -9 -x moshiro-backend; tc qdisc del dev eth0 root 2>/dev/null; tc qdisc del dev eth0 ingress 2>/dev/null; tc qdisc del dev lo ingress 2>/dev/null; true")
|
||||
failed = [r for r in RESULTS if not r[1]]
|
||||
print(f"\n{len(RESULTS) - len(failed)}/{len(RESULTS)} checks passed")
|
||||
for name, _, detail in failed:
|
||||
print(f"FAILED: {name} - {detail}")
|
||||
sys.exit(1 if failed else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -44,11 +44,11 @@ elif name == 'jsonfilter':
|
||||
args = sys.argv[1:]
|
||||
data = json.loads(pathlib.Path(args[args.index('-i') + 1]).read_text() if '-i' in args else sys.stdin.read())
|
||||
query = args[args.index('-e') + 1]
|
||||
value = data['moshiro-probe']['instances']['default']['pid'] if query.startswith('@[') else data.get(query[2:])
|
||||
value = data['moshiro-agent']['instances']['default']['pid'] if query.startswith('@[') else data.get(query[2:])
|
||||
if value is None: sys.exit(1)
|
||||
print(value)
|
||||
elif name == 'ubus':
|
||||
print(json.dumps({'moshiro-probe': {'instances': {'default': {'pid': 1234}}}}))
|
||||
print(json.dumps({'moshiro-agent': {'instances': {'default': {'pid': 1234}}}}))
|
||||
elif name == 'service':
|
||||
with (base / 'restarts').open('a') as f: f.write('restart\n')
|
||||
if os.environ.get('HEALTH') != 'good': sys.exit(0)
|
||||
@@ -93,7 +93,7 @@ class UpdaterTests(unittest.TestCase):
|
||||
|
||||
def publish(self, *, seq=100, version="1.0.1-dev.1", members=None, changes=None):
|
||||
payload = io.BytesIO()
|
||||
members = members or {name: b"executable\n" for name in ("moshiro-probe", "fleet-enroll", "openwrt.sh", "probe.init", "updater.init")}
|
||||
members = members or {name: b"executable\n" for name in ("moshiro-agent", "fleet-enroll", "openwrt.sh", "agent.init", "updater.init")}
|
||||
with tarfile.open(fileobj=payload, mode="w:gz") as archive:
|
||||
for name, content in members.items():
|
||||
entry = tarfile.TarInfo(name)
|
||||
@@ -101,11 +101,11 @@ class UpdaterTests(unittest.TestCase):
|
||||
archive.addfile(entry, io.BytesIO(content))
|
||||
data = payload.getvalue()
|
||||
self.hash = hashlib.sha256(data).hexdigest()
|
||||
package = f"probe-linux-musl-arm64-{self.hash}.tar.gz"
|
||||
package = f"agent-linux-musl-arm64-{self.hash}.tar.gz"
|
||||
version_dir = self.base / "feed" / version
|
||||
version_dir.mkdir(exist_ok=True)
|
||||
(version_dir / package).write_bytes(data)
|
||||
self.manifest = dict(schema=1, component="probe", channel="dev", rid="linux-musl-arm64", version=version,
|
||||
self.manifest = dict(schema=1, component="agent", channel="dev", rid="linux-musl-arm64", version=version,
|
||||
package=package, sha256=self.hash, size=len(data), seq=seq)
|
||||
self.manifest.update(changes or {})
|
||||
self.sign()
|
||||
@@ -144,7 +144,7 @@ class UpdaterTests(unittest.TestCase):
|
||||
def test_ci_publisher_produces_a_signed_package_the_updater_can_install(self):
|
||||
binaries = self.base / "binaries"
|
||||
binaries.mkdir()
|
||||
for name in ("moshiro-probe", "fleet-enroll"):
|
||||
for name in ("moshiro-agent", "fleet-enroll"):
|
||||
(binaries / name).write_bytes(b"native executable fixture\n")
|
||||
env = dict(self.env, CI_COMMIT_TAG="", CI_PIPELINE_EVENT="push", CI_COMMIT_BRANCH="main", CI_COMMIT_MESSAGE="[CI] fixture", GITEA_BASE="https://test.invalid",
|
||||
GITEA_TOKEN="test-token", RELEASE_SIGNING_KEY=self.key.read_text())
|
||||
|
||||
Reference in new issue
Block a user