fix: bound HTTP2 soak resources and cleanup

2026-09-23 01:23:20 +08:00
parent 1a534c7976
commit d94df4fc54
7 changed files with 279 additions and 23 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,7 +5,8 @@
 - 新增独立 `transport/http2` 实验 Runner：支持 h2c/TLS h2、连接上限、全局请求并发闸门、GOAWAY 排空、RST_STREAM 单流取消和生命周期统计；不改变 native HTTP/1.1 默认路径。
 - HTTP/2 实验 Runner 增加标准 1 MiB 头部上限、默认 8 MiB 请求体上限、128 路同连接多路复用、2 MiB 慢读取流控验收、协议错误、RST_STREAM 和 GOAWAY 帧计数；已声明超限请求在进入业务 handler 前返回 413。
 - 新增可配置时长的 HTTP/2 TLS soak：默认 3 秒，Linux 验收可通过 `GO_OMNI_HTTP2_SOAK_DURATION=30m` 覆盖，混合验证 128 路 stream、慢 handler、请求体、Keep-Alive 和退出回收。
- 新增 `scripts/verifyHTTP2.py`，固定在 Go 1.25 Linux 容器中运行 HTTP/2 TLS race 长测。
+- HTTP/2 soak 每 15 秒输出请求/失败数、Go 堆、Linux RSS、goroutine、FD 和协议统计；校验响应协议、内容和独立请求 ID，不再忽略压测期限附近已发出请求的失败。
+- 新增 `scripts/verifyHTTP2.py`，固定在 Go 1.25 Linux 容器中运行 HTTP/2 TLS 长测；默认使用普通模式，`--race` 默认运行 60 秒。容器默认限制为 2 核、2 GiB 内存、256 个进程/线程，带总时限、中断清理和日志大小限制。
 - native TLS 明确限制 ALPN 为 `http/1.1`：缺省自动宣告该协议，误配置 `h2` 等未实现协议时在启动阶段报错。
 - native `Server.Stats()` 新增连接和请求过载拒绝计数；新增可信代理配置，访问日志只在 socket 对端属于配置网段时解析 `X-Forwarded-For`。
 - 新增 transport-neutral `metrics` 包，支持请求数、错误数、in-flight、累计耗时和状态分布，并可直接注册 `/metrics` 接口。
--- a/README.md
+++ b/README.md
@@ -228,7 +228,9 @@ python3 scripts/verifyBusiness.py --version v0.6.0-preview

 标准库与 native 的并发、延迟和长连接内存对照见 [PERFORMANCE.md](PERFORMANCE.md)。`python3 scripts/verifyPerformance.py` 运行检查与 benchmark；增加 `--load --output /tmp/goOmniPerf.json` 可测量独立服务进程并保存报告。

-HTTP/2 仍是显式 opt-in 实验适配器；Linux TLS 长测使用 `python3 scripts/verifyHTTP2.py --duration 30m`，不会改变 native 的 HTTP/1.1 默认路径。
+HTTP/2 仍是显式 opt-in 实验适配器，不会改变 native 的 HTTP/1.1 默认路径。Linux TLS 长测使用 `python3 scripts/verifyHTTP2.py --duration 30m`，默认使用普通 Go 运行模式；`--race` 默认改为 60 秒，用于并发检查，其额外内存开销不能当成生产运行基线。
+
+该脚本默认把测试容器限制为 2 核、2 GiB 内存（禁用 swap）、256 个进程/线程，整个命令最多运行 40 分钟，中断和超时会清理本次容器。这些限制可用 `--cpus`、`--memory-mb`、`--pids-limit`、`--wall-timeout` 调整；共享服务器应先用 `--duration 60s` 短测，再逐步延长。每 15 秒输出的堆、RSS 和 FD 统计包含同一进程内的客户端与服务端；当前验证范围见 [Linux HTTP/2 短测记录](performance/http2SmokeLinux.json)。

 在框架根目录运行：

--- a/ROADMAP.md
+++ b/ROADMAP.md
@@ -51,7 +51,8 @@
 - native `Stats()` 已记录连接和请求过载拒绝数；访问日志已支持可信代理网段配置，默认拒绝伪造的 `X-Forwarded-For`。
 - native TLS 已明确只宣告 HTTP/1.1 ALPN；HTTP/2 暂由边缘代理终止或使用 `transport/nethttp`，已整理独立协议设计、流控和压测门槛，见 [native HTTP/2 进入条件](transport/native/HTTP2.md)，完成后再接入 native。
 - 已加入独立 `transport/http2` 实验 Runner，验证 h2c/TLS h2、连接上限、全局请求并发拒绝和恢复，以及 GOAWAY 排空、RST_STREAM 单流取消和服务关闭超时；它使用 `x/net/http2` 作为临时帧/流控实现，尚未达到 native 生产协议的流控和长测门槛。
- HTTP/2 实验已补齐头部/请求体上限、128 路同连接多路复用、慢读取流控、协议错误、RST_STREAM 与 GOAWAY 统计，并提供 `scripts/verifyHTTP2.py --duration 30m` 的 Linux TLS 长测入口；代理联调和长测结果仍是进入 native 前的门槛。
+- HTTP/2 实验已补齐头部/请求体上限、128 路同连接多路复用、慢读取流控、协议错误、RST_STREAM 与 GOAWAY 统计，并提供 `scripts/verifyHTTP2.py --duration 30m` 的 Linux TLS 长测入口；长测默认不启用 race，短时并发检查使用 `--race`（默认 60 秒）。测试容器默认限制为 2 核、2 GiB 内存、256 个进程/线程，并设总时限和退出清理；代理联调和长测结果仍是进入 native 前的门槛。
+- 新 Linux amd64 环境在上述硬上限内完成 HTTP/2 普通模式 2 分钟与 race 模式 60 秒短测，分别为 3,885,273 和 377,788 次请求，均为 0 失败；连接/请求归零、FD 回到基线。SIGTERM、SIGHUP 和总时限退出清理已实测。报告见 [HTTP/2 短测记录](performance/http2SmokeLinux.json)；旧的无资源上限 race 长测因 RSS 增长被停止，30 分钟 HTTP/2 验收仍未完成。
 - 已用 16 个并发 worker 完成 3,200 次建连/断连 race 验收；MaxConnections 满载拒绝、槽位释放和 TLS 连接上限均已覆盖。
 - 新增 `scripts/verifyLinux.py`，固定在 Go 1.25 Linux 容器内运行 race、vet、build 和可选 TLS/非 TLS perfcheck；perfcheck 记录堆、goroutine、连接和 Linux 文件描述符。
 - Linux/arm64 已完成 30 秒、128 并发、512 空闲连接的 TLS 与非 TLS 持续回归，报告见 `performance/linuxSoak.json` 和 `performance/linuxSoakTLS.json`，均为 0 错误且连接、goroutine、FD 回收正常。
--- a/performance/http2SmokeLinux.json
+++ b/performance/http2SmokeLinux.json
@@ -0,0 +1,84 @@
+{
+  "createdAt": "2026-09-22T17:22:44.674797+00:00",
+  "platform": "linux/amd64",
+  "image": "golang:1.25",
+  "imageId": "sha256:699337d620559a59b4a2bb298ad59611e535d2ee755a34cf2d2a98f37578dc80",
+  "sourceSha256": {
+    "scripts/verifyHTTP2.py": "2c9f4ca095e16e9b30b80b5d812b2644577222274e5a8041e0fb178e305197a4",
+    "transport/http2/soak_test.go": "9964a1670c8ffe30b27e96375e0ae72d00996b59beede748756013e8fd52eacf"
+  },
+  "containerLimits": {
+    "cpus": 2,
+    "memoryBytes": 2147483648,
+    "memoryAndSwapBytes": 2147483648,
+    "pids": 256
+  },
+  "gomaxprocs": 2,
+  "workers": 128,
+  "notes": [
+    "Bounded smoke validation, not completion of the 30-minute HTTP/2 acceptance gate.",
+    "TLS HTTP/2 over container loopback, with the client and server in one process; memory is not server-only.",
+    "One TCP connection multiplexes the 128 workers; each response checks HTTP/2, status, body and a unique request ID.",
+    "MiB uses 1024*1024 bytes. RSS is /proc/self/statm; heap is runtime.ReadMemStats. Samples every 15 seconds.",
+    "The after sample includes the one monitoring goroutine; it is stopped and joined before the test returns.",
+    "All seven pre-existing containers remained running, and no soak containers remained after cleanup.",
+    "The prior unbounded 30-minute race run was stopped after RSS growth; its result is not counted as a pass.",
+    "The short A/B comparison does not establish the complete cause of the old server incident."
+  ],
+  "results": [
+    {
+      "mode": "regular",
+      "duration": "2m",
+      "elapsedSeconds": 120.04,
+      "requests": 3885273,
+      "failures": 0,
+      "protocolErrors": 0,
+      "sampledPeakRssBytes": 22437888,
+      "before": {
+        "heapBytes": 168040,
+        "rssBytes": 8491008,
+        "goroutines": 2,
+        "fileDescriptors": 6
+      },
+      "after": {
+        "heapBytes": 3925848,
+        "rssBytes": 21364736,
+        "goroutines": 3,
+        "fileDescriptors": 6,
+        "activeConnections": 0,
+        "activeRequests": 0
+      },
+      "exitCode": 0
+    },
+    {
+      "mode": "race",
+      "duration": "60s",
+      "elapsedSeconds": 60.127,
+      "requests": 377788,
+      "failures": 0,
+      "protocolErrors": 0,
+      "sampledPeakRssBytes": 492707840,
+      "before": {
+        "heapBytes": 169256,
+        "rssBytes": 23732224,
+        "goroutines": 2,
+        "fileDescriptors": 6
+      },
+      "after": {
+        "heapBytes": 2722632,
+        "rssBytes": 487239680,
+        "goroutines": 3,
+        "fileDescriptors": 6,
+        "activeConnections": 0,
+        "activeRequests": 0
+      },
+      "exitCode": 0
+    }
+  ],
+  "cleanupValidation": {
+    "sigterm": "passed",
+    "sighup": "passed",
+    "wallTimeout": "passed"
+  },
+  "thirtyMinuteAcceptance": "pending"
+}
--- a/scripts/verifyHTTP2.py
+++ b/scripts/verifyHTTP2.py
@@ -2,42 +2,90 @@
 """Run the opt-in HTTP/2 TLS soak in a pinned Linux Go image."""

 import argparse
+import math
 from pathlib import Path
 import shutil
+import signal
 import subprocess
-import sys
+import uuid


 def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--image", default="golang:1.25")
    parser.add_argument("--goproxy", default="", help="optional Go module proxy passed to the container")
-    parser.add_argument("--duration", default="30m")
+    parser.add_argument("--duration", help="soak duration (default: 30m, or 60s with --race)")
    parser.add_argument("--timeout", default="35m", help="go test timeout (default: 35m)")
    parser.add_argument("--gomaxprocs", type=int, default=2)
+    parser.add_argument("--cpus", type=float, default=2, help="container CPU quota (default: 2)")
+    parser.add_argument("--memory-mb", type=int, default=2048, help="container memory cap, with no swap (default: 2048 MiB)")
+    parser.add_argument("--pids-limit", type=int, default=256, help="container process/thread cap (default: 256)")
+    parser.add_argument("--wall-timeout", type=int, default=2400, help="total time limit including build/download, in seconds (default: 2400)")
+    parser.add_argument(
+        "--race",
+        action="store_true",
+        help="enable the race detector for short concurrency checks; its memory overhead is not a production baseline",
+    )
    args = parser.parse_args()
+    if args.duration is None:
+        args.duration = "60s" if args.race else "30m"
    if not args.duration or not args.timeout or args.gomaxprocs < 1:
        parser.error("--duration and --timeout must be non-empty and --gomaxprocs must be positive")
+    if not math.isfinite(args.cpus) or args.cpus <= 0 or args.memory_mb < 128 or args.pids_limit < 16 or args.wall_timeout < 1:
+        parser.error("--cpus and --wall-timeout must be positive; --memory-mb >= 128 and --pids-limit >= 16 are required")

    docker = shutil.which("docker")
    if docker is None:
        raise SystemExit("docker is required for HTTP/2 Linux soak")
    root = Path(__file__).resolve().parent.parent
+    name = "go-omni-http2-soak-" + uuid.uuid4().hex[:12]
    command = [
-        docker, "run", "--rm", "--pull=missing",
-        "-v", f"{root}:/workspace", "-w", "/workspace",
+        docker, "run", "--rm", "--pull=missing", "--name", name,
+        "--cpus", str(args.cpus), "--memory", f"{args.memory_mb}m",
+        "--memory-swap", f"{args.memory_mb}m", "--pids-limit", str(args.pids_limit),
+        "--log-opt", "max-size=10m", "--log-opt", "max-file=1",
+        "-v", f"{root}:/workspace:ro", "-w", "/workspace",
        "-e", "GOWORK=off", "-e", f"GOMAXPROCS={args.gomaxprocs}",
        "-e", f"GO_OMNI_HTTP2_SOAK_DURATION={args.duration}",
    ]
    if args.goproxy:
        command += ["-e", f"GOPROXY={args.goproxy}"]
+    command += [args.image, "go", "test"]
+    if args.race:
+        command.append("-race")
    command += [
-        args.image, "go", "test", "-race", "-timeout", args.timeout, "./transport/http2",
-        "-run", "^TestTLSSoak$", "-count=1",
+        "-v", "-timeout", args.timeout, "./transport/http2", "-run", "^TestTLSSoak$", "-count=1",
    ]
    print(f"HTTP/2 Linux soak image: {args.image}", flush=True)
    print(f"HTTP/2 TLS soak duration: {args.duration}", flush=True)
-    result = subprocess.run(command, cwd=root)
+    print(f"HTTP/2 TLS soak mode: {'race' if args.race else 'regular'}", flush=True)
+    print(f"HTTP/2 soak container: {name}; limits: {args.cpus:g} CPUs, {args.memory_mb} MiB, {args.pids_limit} tasks", flush=True)
+
+    def interrupt(signum, frame):
+        raise KeyboardInterrupt
+
+    previous_handlers = {
+        sig: signal.signal(sig, interrupt) for sig in (signal.SIGTERM, signal.SIGHUP)
+    }
+    try:
+        result = subprocess.run(command, cwd=root, timeout=args.wall_timeout)
+    except subprocess.TimeoutExpired:
+        raise SystemExit("HTTP/2 Linux soak exceeded --wall-timeout; stopping its container")
+    except KeyboardInterrupt:
+        raise SystemExit("HTTP/2 Linux soak interrupted; stopping its container")
+    finally:
+        for sig, handler in previous_handlers.items():
+            signal.signal(sig, handler)
+        # Only this run's unique container is targeted, including on timeout or
+        # a disconnected SSH session. Do not prune other jobs.
+        try:
+            cleanup = subprocess.run(
+                [docker, "rm", "--force", name], capture_output=True, text=True, timeout=15,
+            )
+            if cleanup.returncode != 0 and "No such container" not in cleanup.stderr:
+                raise SystemExit(f"Could not clean up {name}: {cleanup.stderr.strip()}")
+        except subprocess.TimeoutExpired:
+            raise SystemExit(f"Cleanup timed out; inspect container {name} before another run")
    if result.returncode != 0:
        raise SystemExit(f"HTTP/2 Linux soak failed with exit code {result.returncode}")
    print("PASS HTTP/2 TLS soak", flush=True)
--- a/scripts/verifyHTTP2_test.py
+++ b/scripts/verifyHTTP2_test.py
@@ -0,0 +1,65 @@
+"""Regression checks for stress-container cleanup on failure and interruption."""
+
+import contextlib
+import importlib.util
+import io
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+
+
+spec = importlib.util.spec_from_file_location("http2Runner", Path(__file__).with_name("verifyHTTP2.py"))
+runner = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(runner)
+
+
+class CleanupTests(unittest.TestCase):
+    def testFailedRunsCleanUpOnlyTheirOwnContainer(self):
+        failures = (
+            subprocess.TimeoutExpired("docker", 1),
+            KeyboardInterrupt(),
+            subprocess.CompletedProcess([], 7),
+        )
+        for failure in failures:
+            with self.subTest(failure=type(failure).__name__):
+                calls = []
+                original_handlers = {sig: signal.getsignal(sig) for sig in (signal.SIGTERM, signal.SIGHUP)}
+
+                def execute(command, **kwargs):
+                    calls.append(command)
+                    if command[1] == "run":
+                        if isinstance(failure, BaseException):
+                            raise failure
+                        return failure
+                    return subprocess.CompletedProcess(command, 0, "", "")
+
+                with patch.object(sys, "argv", ["verifyHTTP2.py", "--duration", "1s"]), \
+                        patch.object(runner.shutil, "which", return_value="/fake/docker"), \
+                        patch.object(runner.subprocess, "run", side_effect=execute), \
+                        contextlib.redirect_stdout(io.StringIO()):
+                    with self.assertRaises(SystemExit) as raised:
+                        runner.main()
+                    self.assertNotEqual(raised.exception.code, 0)
+                name = calls[0][calls[0].index("--name") + 1]
+                self.assertEqual(calls[1:], [["/fake/docker", "rm", "--force", name]])
+                self.assertEqual(original_handlers, {sig: signal.getsignal(sig) for sig in original_handlers})
+
+    def testCleanupFailureIsNotReportedAsPass(self):
+        results = [
+            subprocess.CompletedProcess([], 0),
+            subprocess.CompletedProcess([], 1, "", "Cannot connect to the Docker daemon"),
+        ]
+        with patch.object(sys, "argv", ["verifyHTTP2.py", "--duration", "1s"]), \
+                patch.object(runner.shutil, "which", return_value="/fake/docker"), \
+                patch.object(runner.subprocess, "run", side_effect=results), \
+                contextlib.redirect_stdout(io.StringIO()) as output:
+            with self.assertRaisesRegex(SystemExit, "Could not clean up"):
+                runner.main()
+            self.assertNotIn("PASS", output.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()
--- a/transport/http2/soak_test.go
+++ b/transport/http2/soak_test.go
@@ -7,6 +7,9 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"runtime"
+	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -18,6 +21,7 @@ import (
 // TestTLSSoak is short by default so it remains part of normal package
 // verification. Linux acceptance runs set GO_OMNI_HTTP2_SOAK_DURATION=30m.
 func TestTLSSoak(t *testing.T) {
+	started := time.Now()
 	duration := 3 * time.Second
 	if value := os.Getenv("GO_OMNI_HTTP2_SOAK_DURATION"); value != "" {
 		parsed, err := time.ParseDuration(value)
@@ -26,6 +30,7 @@ func TestTLSSoak(t *testing.T) {
 		}
 		duration = parsed
 	}
+	logSoakResources(t, started, "before", 0, 0, Stats{})

 	certificate, roots := testCertificate(t)
 	server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
@@ -44,6 +49,7 @@ func TestTLSSoak(t *testing.T) {
 				return
 			}
 		}
+		writer.Header().Set("X-Soak-Request-Id", request.Header.Get("X-Soak-Request-Id"))
 		_, _ = io.WriteString(writer, request.URL.Path)
 	}))
 	server.MaxConnections = 8
@@ -73,13 +79,37 @@ func TestTLSSoak(t *testing.T) {
 	client := &http.Client{Transport: transport, Timeout: 3 * time.Second}

 	var requests, failures atomic.Uint64
+	var firstFailure string
+	var failureOnce sync.Once
+	recordFailure := func(message string) {
+		failures.Add(1)
+		failureOnce.Do(func() { firstFailure = message })
+	}
+	monitorStop, monitorDone := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(monitorDone)
+		ticker := time.NewTicker(15 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				logSoakResources(t, started, "running", requests.Load(), failures.Load(), server.Stats())
+			case <-monitorStop:
+				return
+			}
+		}
+	}()
+	defer func() {
+		close(monitorStop)
+		<-monitorDone
+	}()
 	deadline := time.Now().Add(duration)
 	var workers sync.WaitGroup
 	for worker := 0; worker < 128; worker++ {
 		workers.Add(1)
 		go func(worker int) {
 			defer workers.Done()
-			for time.Now().Before(deadline) {
+			for sequence := uint64(0); time.Now().Before(deadline); sequence++ {
 				path := "/health"
 				method := http.MethodGet
 				var body io.Reader
@@ -95,24 +125,26 @@ func TestTLSSoak(t *testing.T) {
 				request, err := http.NewRequestWithContext(requestContext, method, "https://"+listener.Addr().String()+path, body)
 				if err != nil {
 					cancelRequest()
-					failures.Add(1)
+					recordFailure(err.Error())
 					continue
 				}
+				requestID := strconv.Itoa(worker) + ":" + strconv.FormatUint(sequence, 10)
+				request.Header.Set("X-Soak-Request-Id", requestID)
 				response, err := client.Do(request)
 				if err != nil {
 					cancelRequest()
-					if time.Now().Before(deadline) {
-						failures.Add(1)
-					}
+					recordFailure(err.Error())
 					continue
 				}
-				_, readErr := io.Copy(io.Discard, response.Body)
+				responseBody, readErr := io.ReadAll(io.LimitReader(response.Body, 1024))
 				response.Body.Close()
 				cancelRequest()
-				if readErr != nil || response.StatusCode != http.StatusOK {
-					if time.Now().Before(deadline) {
-						failures.Add(1)
+				if readErr != nil {
+					recordFailure(readErr.Error())
+					continue
 				}
+				if response.StatusCode != http.StatusOK || response.ProtoMajor != 2 || string(responseBody) != path || response.Header.Get("X-Soak-Request-Id") != requestID {
+					recordFailure("unexpected status, protocol, body or request ID: " + response.Status)
 					continue
 				}
 				requests.Add(1)
@@ -131,14 +163,37 @@ func TestTLSSoak(t *testing.T) {
 	if err := <-done; err != nil {
 		t.Fatalf("TLS soak serve: %v", err)
 	}
+	logSoakResources(t, started, "after", requests.Load(), failures.Load(), server.Stats())
 	if failures.Load() != 0 || requests.Load() == 0 {
-		t.Fatalf("TLS soak requests=%d failures=%d stats=%+v", requests.Load(), failures.Load(), server.Stats())
+		t.Fatalf("TLS soak requests=%d failures=%d firstFailure=%q stats=%+v", requests.Load(), failures.Load(), firstFailure, server.Stats())
 	}
-	if stats := server.Stats(); stats.ActiveConnections != 0 || stats.ActiveRequests != 0 {
-		t.Fatalf("TLS soak did not reclaim resources: %+v", stats)
+	if stats := server.Stats(); stats.ActiveConnections != 0 || stats.ActiveRequests != 0 || stats.ProtocolErrors != 0 || stats.RejectedRequests != 0 {
+		t.Fatalf("TLS soak protocol errors, unexpected rejection or unreclaimed resources: %+v", stats)
 	}
 }

+// These measurements cover the client and server in the same test process.
+// RSS includes race runtime allocations that do not appear in Go heap stats.
+func logSoakResources(t *testing.T, started time.Time, phase string, requests, failures uint64, stats Stats) {
+	t.Helper()
+	var memory runtime.MemStats
+	runtime.ReadMemStats(&memory)
+	rssBytes, fileDescriptors := int64(-1), -1
+	if data, err := os.ReadFile("/proc/self/statm"); err == nil {
+		if fields := strings.Fields(string(data)); len(fields) >= 2 {
+			if pages, err := strconv.ParseInt(fields[1], 10, 64); err == nil {
+				rssBytes = pages * int64(os.Getpagesize())
+			}
+		}
+	}
+	if entries, err := os.ReadDir("/proc/self/fd"); err == nil {
+		fileDescriptors = len(entries)
+	}
+	t.Logf("phase=%s elapsed=%s requests=%d failures=%d heapBytes=%d rssBytes=%d goroutines=%d fileDescriptors=%d stats=%+v",
+		phase, time.Since(started).Round(time.Millisecond), requests, failures,
+		memory.HeapAlloc, rssBytes, runtime.NumGoroutine(), fileDescriptors, stats)
+}
+
 type repeatingReader struct {
 	remaining int
 }