feat: 补齐虚拟机验收并去掉 Docker 弱网占位

This commit is contained in:
Nixevol
2026-10-04 04:12:54 +08:00
parent 069039f4cb
commit fca23c310c
23 changed files with 291 additions and 622 deletions
+3 -20
View File
@@ -1,22 +1,5 @@
# 混沌 / 弱网测试辅助(Q 线)
# 崩溃续传
用 toxiproxy 官方镜像做延迟和断开;Linux 丢包用容器内 netem。Q3 相关 Docker 资源名带 `q3` 前缀,用完删除。
`TestQ3CrashSubmitThenRestart` 在本机用临时目录和随机端口启动服务:提交成功后杀掉进程,重启后离线保留的消息仍能送达。
## 启动 toxiproxy
```powershell
docker compose -p q3-chaos -f test/chaos/compose.yml up -d
docker compose -p q3-chaos -f test/chaos/compose.yml port toxiproxy 8474
```
清理:
```powershell
docker compose -p q3-chaos -f test/chaos/compose.yml down -v
```
集成测 `TestQ3ToxiproxyOfflineKeepDelivered` 会自动起停上述 compose。
## netem
见 [netem/README.md](./netem/README.md)。本机 Windows 上 Q3 将 netem 20% 丢包记为未测(见 `docs/DEVIATIONS.md` 测试交付 Q)。
弱网丢包不在这里用 Docker 或 toxiproxy 做。20% 丢包在 Linux 虚拟机里用 `tc netem` 测,不进入 `task check`。
-13
View File
@@ -1,13 +0,0 @@
# toxiproxy:延迟与断开。项目名用 -p q3-chaos,容器名带 q3 前缀。
# API 与代理端口映射到本机,具体宿主机端口用 `docker compose port` 查询,不要写死业务端口。
services:
toxiproxy:
image: ghcr.io/shopify/toxiproxy:2.12.0
container_name: q3-toxiproxy
ports:
- "8474"
- "8475"
- "8476"
extra_hosts:
- "host.docker.internal:host-gateway"
-192
View File
@@ -1,192 +0,0 @@
// Package chaos 提供基于 toxiproxy 的弱网故障注入辅助。
// 上游地址由调用方传入,不写死端口。
package chaos
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"time"
)
// Client 调用 toxiproxy HTTP API。
type Client struct {
BaseURL string
HTTPClient *http.Client
}
// NewClient 创建客户端。baseURL 形如 http://127.0.0.1:8474。
func NewClient(baseURL string) *Client {
return &Client{
BaseURL: strings.TrimRight(baseURL, "/"),
HTTPClient: &http.Client{
Timeout: 10 * time.Second,
},
}
}
// Proxy 描述一个 toxiproxy 代理。
type Proxy struct {
Name string `json:"name"`
Listen string `json:"listen"`
Upstream string `json:"upstream"`
Enabled bool `json:"enabled"`
}
// Toxic 描述一条故障规则。
type Toxic struct {
Name string `json:"name"`
Type string `json:"type"`
Stream string `json:"stream,omitempty"`
Toxicity float32 `json:"toxicity,omitempty"`
Attributes map[string]any `json:"attributes,omitempty"`
}
// CreateProxy 创建或覆盖同名代理。listen 可用 host:0 让 toxiproxy 分配端口。
func (c *Client) CreateProxy(name, listen, upstream string) (*Proxy, error) {
if name == "" {
return nil, fmt.Errorf("proxy name required")
}
if listen == "" {
return nil, fmt.Errorf("listen required")
}
if upstream == "" {
return nil, fmt.Errorf("upstream required")
}
body := Proxy{
Name: name,
Listen: listen,
Upstream: upstream,
Enabled: true,
}
var out Proxy
if err := c.doJSON(http.MethodPost, "/proxies", body, &out); err != nil {
return nil, err
}
return &out, nil
}
// DeleteProxy 删除代理;不存在时忽略。
func (c *Client) DeleteProxy(name string) error {
req, err := http.NewRequest(http.MethodDelete, c.BaseURL+"/proxies/"+name, nil)
if err != nil {
return err
}
resp, err := c.HTTPClient.Do(req)
if err != nil {
return err
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusNoContent || resp.StatusCode == http.StatusOK {
return nil
}
b, _ := io.ReadAll(resp.Body)
return fmt.Errorf("delete proxy: %s: %s", resp.Status, strings.TrimSpace(string(b)))
}
// GetProxy 读取代理(含实际 listen 地址)。
func (c *Client) GetProxy(name string) (*Proxy, error) {
var out Proxy
if err := c.doJSON(http.MethodGet, "/proxies/"+name, nil, &out); err != nil {
return nil, err
}
return &out, nil
}
// AddLatency 注入下行/上行延迟(毫秒)。stream 为空时默认 downstream。
func (c *Client) AddLatency(proxyName, toxicName string, latencyMs, jitterMs int, stream string) (*Toxic, error) {
if stream == "" {
stream = "downstream"
}
t := Toxic{
Name: toxicName,
Type: "latency",
Stream: stream,
Toxicity: 1,
Attributes: map[string]any{
"latency": latencyMs,
"jitter": jitterMs,
},
}
return c.addToxic(proxyName, t)
}
// AddResetPeer 在连接上注入 TCP RST(断开)。timeoutMs 为触发前等待。
func (c *Client) AddResetPeer(proxyName, toxicName string, timeoutMs int, stream string) (*Toxic, error) {
if stream == "" {
stream = "downstream"
}
t := Toxic{
Name: toxicName,
Type: "reset_peer",
Stream: stream,
Toxicity: 1,
Attributes: map[string]any{
"timeout": timeoutMs,
},
}
return c.addToxic(proxyName, t)
}
// RemoveToxic 删除一条 toxic。
func (c *Client) RemoveToxic(proxyName, toxicName string) error {
req, err := http.NewRequest(http.MethodDelete, c.BaseURL+"/proxies/"+proxyName+"/toxics/"+toxicName, nil)
if err != nil {
return err
}
resp, err := c.HTTPClient.Do(req)
if err != nil {
return err
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusNoContent || resp.StatusCode == http.StatusOK {
return nil
}
b, _ := io.ReadAll(resp.Body)
return fmt.Errorf("remove toxic: %s: %s", resp.Status, strings.TrimSpace(string(b)))
}
func (c *Client) addToxic(proxyName string, t Toxic) (*Toxic, error) {
var out Toxic
if err := c.doJSON(http.MethodPost, "/proxies/"+proxyName+"/toxics", t, &out); err != nil {
return nil, err
}
return &out, nil
}
func (c *Client) doJSON(method, path string, in any, out any) error {
var body io.Reader
if in != nil {
b, err := json.Marshal(in)
if err != nil {
return err
}
body = bytes.NewReader(b)
}
req, err := http.NewRequest(method, c.BaseURL+path, body)
if err != nil {
return err
}
if in != nil {
req.Header.Set("Content-Type", "application/json")
}
resp, err := c.HTTPClient.Do(req)
if err != nil {
return err
}
defer func() { _ = resp.Body.Close() }()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return err
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("%s %s: %s: %s", method, path, resp.Status, strings.TrimSpace(string(respBody)))
}
if out == nil || len(respBody) == 0 {
return nil
}
return json.Unmarshal(respBody, out)
}
-113
View File
@@ -1,113 +0,0 @@
package chaos
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestClientCreateLatencyAndReset(t *testing.T) {
t.Parallel()
toxics := map[string][]Toxic{}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.Method == http.MethodPost && r.URL.Path == "/proxies":
var p Proxy
if err := json.NewDecoder(r.Body).Decode(&p); err != nil {
t.Errorf("decode proxy: %v", err)
http.Error(w, err.Error(), 400)
return
}
if p.Listen == "0.0.0.0:0" {
p.Listen = "0.0.0.0:18080"
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(p)
case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/toxics"):
var toxic Toxic
if err := json.NewDecoder(r.Body).Decode(&toxic); err != nil {
http.Error(w, err.Error(), 400)
return
}
name := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/proxies/"), "/toxics")
toxics[name] = append(toxics[name], toxic)
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(toxic)
case r.Method == http.MethodDelete:
w.WriteHeader(http.StatusNoContent)
default:
http.NotFound(w, r)
}
}))
t.Cleanup(srv.Close)
c := NewClient(srv.URL)
p, err := c.CreateProxy("q-demo", "0.0.0.0:0", "127.0.0.1:19000")
if err != nil {
t.Fatalf("CreateProxy: %v", err)
}
if p.Upstream != "127.0.0.1:19000" {
t.Fatalf("upstream = %q", p.Upstream)
}
if p.Listen != "0.0.0.0:18080" {
t.Fatalf("listen = %q", p.Listen)
}
lat, err := c.AddLatency("q-demo", "lag", 200, 50, "")
if err != nil {
t.Fatalf("AddLatency: %v", err)
}
if lat.Type != "latency" {
t.Fatalf("type = %q", lat.Type)
}
if _, err := c.AddResetPeer("q-demo", "drop", 0, ""); err != nil {
t.Fatalf("AddResetPeer: %v", err)
}
if got := toxics["q-demo"]; len(got) != 2 {
t.Fatalf("toxics len = %d", len(got))
}
if got := toxics["q-demo"][0]; got.Type != "latency" {
t.Fatalf("first toxic type = %q", got.Type)
}
if err := c.RemoveToxic("q-demo", "lag"); err != nil {
t.Fatalf("RemoveToxic: %v", err)
}
if err := c.DeleteProxy("q-demo"); err != nil {
t.Fatalf("DeleteProxy: %v", err)
}
}
func TestClientRejectsEmptyFields(t *testing.T) {
t.Parallel()
c := NewClient("http://127.0.0.1:1")
if _, err := c.CreateProxy("", "0.0.0.0:0", "127.0.0.1:1"); err == nil {
t.Fatal("expected error for empty name")
}
if _, err := c.CreateProxy("x", "", "127.0.0.1:1"); err == nil {
t.Fatal("expected error for empty listen")
}
if _, err := c.CreateProxy("x", "0.0.0.0:0", ""); err == nil {
t.Fatal("expected error for empty upstream")
}
}
func TestCreateProxyErrorStatus(t *testing.T) {
t.Parallel()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusConflict)
_, _ = io.WriteString(w, "already exists")
}))
t.Cleanup(srv.Close)
c := NewClient(srv.URL)
if _, err := c.CreateProxy("q", "0.0.0.0:0", "127.0.0.1:9"); err == nil {
t.Fatal("expected error")
}
}
-31
View File
@@ -1,31 +0,0 @@
# Linux 容器内用 netem 做丢包(本机 Windows 不能直接跑 tc)
## 用法
对已经在同一网络里的目标容器网卡注入 20% 丢包(需要 `NET_ADMIN`):
```powershell
# 示例:起一个带 net-tools/iproute2 的旁路容器,对网卡 eth0 丢包
docker run --rm --name q-netem --cap-add=NET_ADMIN --network container:q-toxiproxy `
nicolaka/netshoot `
bash -lc "tc qdisc replace dev eth0 root netem loss 20%"
```
或把本目录脚本挂进去:
```powershell
docker run --rm --name q-netem --cap-add=NET_ADMIN --network container:<目标容器名> `
-v ${PWD}/test/chaos/netem:/scripts:ro `
nicolaka/netshoot `
bash /scripts/apply-loss.sh eth0 20
```
清除:
```powershell
docker run --rm --name q-netem-clear --cap-add=NET_ADMIN --network container:<目标容器名> `
nicolaka/netshoot `
bash -lc "tc qdisc del dev eth0 root 2>/dev/null || true"
```
用完删除容器(上面 `--rm` 已自动删)。不要把 NixMsg 业务端口写死进脚本。
-15
View File
@@ -1,15 +0,0 @@
#!/usr/bin/env bash
# 在 Linux 容器内执行:对指定网卡注入丢包。用法:apply-loss.sh <iface> <loss_percent>
set -euo pipefail
IFACE="${1:-eth0}"
LOSS="${2:-20}"
if ! command -v tc >/dev/null 2>&1; then
echo "tc not found; use an image with iproute2 (e.g. nicolaka/netshoot)" >&2
exit 1
fi
tc qdisc replace dev "$IFACE" root netem loss "${LOSS}%"
echo "netem: iface=${IFACE} loss=${LOSS}%"
tc qdisc show dev "$IFACE"
+2
View File
@@ -8,6 +8,8 @@ import (
"git.asio.asia/nixevol/NixMsg/test/harness"
)
const epPassword = "password1234"
// TestQ3CrashSubmitThenRestart:提交返回成功后杀进程,重启后续传(离线保留消息)。
func TestQ3CrashSubmitThenRestart(t *testing.T) {
srv, err := accept.StartManaged()
-180
View File
@@ -1,180 +0,0 @@
package chaos_test
import (
"net"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"time"
"git.asio.asia/nixevol/NixMsg/test/accept"
"git.asio.asia/nixevol/NixMsg/test/chaos"
"git.asio.asia/nixevol/NixMsg/test/harness"
)
const (
q3ComposeProject = "q3-chaos"
q3ProxyName = "q3-nixmsg"
epPassword = "password1234"
)
// TestQ3ToxiproxyOfflineKeepDelivered:延迟 + 断开后,离线保留期内消息最终送达。
func TestQ3ToxiproxyOfflineKeepDelivered(t *testing.T) {
if _, err := exec.LookPath("docker"); err != nil {
t.Skip("docker 不可用,跳过 toxiproxy 弱网")
}
composeFile := filepath.Join(moduleRoot(t), "test", "chaos", "compose.yml")
down := func() {
cmd := exec.Command("docker", "compose", "-p", q3ComposeProject, "-f", composeFile, "down", "-v", "--remove-orphans")
_ = cmd.Run()
}
down()
t.Cleanup(down)
up := exec.Command("docker", "compose", "-p", q3ComposeProject, "-f", composeFile, "up", "-d")
if out, err := up.CombinedOutput(); err != nil {
t.Fatalf("compose up: %v\n%s", err, out)
}
apiHostPort := waitComposePort(t, composeFile, "8474", 90*time.Second)
listenHostPort := waitComposePort(t, composeFile, "8475", 90*time.Second)
apiURL := "http://127.0.0.1:" + apiHostPort
waitHTTP(t, apiURL+"/version", 90*time.Second)
srv, err := harness.Start(harness.Options{})
if err != nil {
t.Fatal(err)
}
defer func() { _ = srv.Stop() }()
_, nixPort, err := net.SplitHostPort(srv.Addr)
if err != nil {
t.Fatal(err)
}
c := chaos.NewClient(apiURL)
_ = c.DeleteProxy(q3ProxyName)
if _, err := c.CreateProxy(q3ProxyName, "0.0.0.0:8475", "host.docker.internal:"+nixPort); err != nil {
t.Fatalf("CreateProxy: %v", err)
}
t.Cleanup(func() { _ = c.DeleteProxy(q3ProxyName) })
proxyBase := "http://127.0.0.1:" + listenHostPort
if _, err := c.AddLatency(q3ProxyName, "q3-lag", 150, 50, ""); err != nil {
t.Fatalf("AddLatency: %v", err)
}
ac := accept.AdminLogin(t, srv)
accept.CreateEndpoint(t, ac, "q3alice01", epPassword)
accept.CreateEndpoint(t, ac, "q3bob0001", epPassword)
alice := accept.MQTTLogin(t, proxyBase, "q3alice01", epPassword)
defer alice.Close()
sendOff := alice.Request(t, map[string]any{
"v": 1, "type": "send", "rid": "q3s1", "id": "q3-off-keep",
"to": map[string]any{"kind": "endpoint", "id": "q3bob0001"},
"body": map[string]any{"enc": "utf8", "data": "keep-me"},
"delay_ms": int64(0),
"offline": map[string]any{"keep": true, "ttl_seconds": int64(86400)},
})
if !sendOff.OK {
t.Fatalf("send offline keep: %+v", sendOff)
}
if _, err := c.AddResetPeer(q3ProxyName, "q3-rst", 0, ""); err != nil {
t.Fatalf("AddResetPeer: %v", err)
}
time.Sleep(400 * time.Millisecond)
_ = c.RemoveToxic(q3ProxyName, "q3-rst")
_ = c.RemoveToxic(q3ProxyName, "q3-lag")
bob := accept.MQTTLogin(t, proxyBase, "q3bob0001", epPassword)
defer bob.Close()
msg := bob.WaitType(t, "msg", 20*time.Second)
if msg["id"] != "q3-off-keep" {
t.Fatalf("want q3-off-keep got %v", msg)
}
bob.Request(t, map[string]any{"v": 1, "type": "ack", "rid": "q3a1", "from": "q3alice01", "id": "q3-off-keep"})
}
// TestQ3NetemUntested 记录本机 Windows 上 netem 20% 丢包未测原因。
func TestQ3NetemUntested(t *testing.T) {
t.Log("未测:Linux netem 20% 丢包。本机为 Windows;无法在宿主直接用 tc/netem。" +
"若只对 toxiproxy 容器网卡挂 netshoot,不等于 NixMsg 业务端到端丢包验收。" +
"本波用 toxiproxy 延迟/断开覆盖弱网;netem 需 Linux 宿主或把服务放进同网络 Linux 容器后再测。")
}
func moduleRoot(t *testing.T) string {
t.Helper()
dir, err := os.Getwd()
if err != nil {
t.Fatal(err)
}
for {
if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
return dir
}
parent := filepath.Dir(dir)
if parent == dir {
t.Fatal("go.mod not found")
}
dir = parent
}
}
func waitComposePort(t *testing.T, composeFile, containerPort string, timeout time.Duration) string {
t.Helper()
deadline := time.Now().Add(timeout)
var last string
for time.Now().Before(deadline) {
cmd := exec.Command("docker", "compose", "-p", q3ComposeProject, "-f", composeFile, "port", "toxiproxy", containerPort)
out, err := cmd.CombinedOutput()
last = strings.TrimSpace(string(out))
if err == nil && last != "" {
_, port, perr := net.SplitHostPort(normalizeComposeAddr(last))
if perr == nil && port != "" {
return port
}
}
time.Sleep(400 * time.Millisecond)
}
t.Fatalf("compose port %s timeout, last=%q", containerPort, last)
return ""
}
func normalizeComposeAddr(s string) string {
s = strings.TrimSpace(s)
if strings.HasPrefix(s, "[::]") {
return "127.0.0.1" + s[len("[::]"):]
}
if strings.HasPrefix(s, "0.0.0.0:") {
return "127.0.0.1" + s[len("0.0.0.0"):]
}
return s
}
func waitHTTP(t *testing.T, url string, timeout time.Duration) {
t.Helper()
client := &http.Client{Timeout: 2 * time.Second}
deadline := time.Now().Add(timeout)
var last error
for time.Now().Before(deadline) {
resp, err := client.Get(url)
if err == nil {
_ = resp.Body.Close()
if resp.StatusCode < 500 {
return
}
last = err
} else {
last = err
}
time.Sleep(400 * time.Millisecond)
}
t.Fatalf("wait HTTP %s: %v", url, last)
}