From 5847705e356b0a79411f5eb75b39ed01caf1094b Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 21:38:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=E4=BF=AE=20Snapshot=20?= =?UTF-8?q?=E4=BA=A4=E5=87=BA=E6=B4=BB=20map=20=E5=AF=BC=E8=87=B4=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E8=A2=AB=20fatal=20=E6=89=93=E6=AD=BB=EF=BC=88?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=20map=20=E8=BF=AD=E4=BB=A3+=E5=86=99?= =?UTF-8?q?=E5=85=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 现场：cy.log:560，2026-10-03 21:00:28.403，栈顶 encoding/json.mapEncoder.encode ← writeJSON(rest.go:383) ← adminObservabilityMetrics(observability.go:89)， 不可 recover，网关 recover 中间件接不住 ⇒ 单二进制进程整体退出（端口 10942 零监听、 所有后台服务含 backthink 心跳一起没；比第一次心跳 tick 早 32.6 秒）。 根因：MetricsCollector.Snapshot 在锁内只新造了 StatusCounts，却把 collector 自己的 活 map m.perMethod / m.perPath 直接交出去；锁外（handler 的 writeJSON）才开始遍历， 而每个请求的 logging 中间件（middleware.go:79）在别的 goroutine 里 Record 写这两个 map ⇒ 写有锁、迭代没锁。面板并行轮询十几个端点，/admin/observability/metrics 被打了 29 次 且常见同毫秒两个请求，必踩。 修法：Snapshot 改为交出副本（新增 copyCounts / copyStatusCounts，都在锁内遍历）， 并把「必须在锁外可安全编码」的原因写进文档注释。对外读侧语义不变（仍是全量累计）。 测试（新增 observability_metrics_test.go，242 行）：① 快照与 collector 双向隔离（确定性） ② 并发「写 + 编码快照」（-race 下复现生产竞态）③ 真 HTTP + 中间件 + 并行轮询 /configmetric 端点（= 21:00:28 现场）。 验证：go build/vet 绿；-race ./internal/services/gateway/... ok；全量 go test ./... 58 包 ok / 0 FAIL； linux/amd64 + darwin/arm64 交叉编译绿。 变异验证：把 Snapshot 改回交活 map → 确定性测试 FAIL（"snapshot shares the collector's perPath map"） + -race 报出与生产同一处的 DATA RACE（Record 写在 observability_metrics.go:64）。 --- .../services/gateway/observability_metrics.go | 38 ++- .../gateway/observability_metrics_test.go | 242 ++++++++++++++++++ 2 files changed, 273 insertions(+), 7 deletions(-) create mode 100644 internal/services/gateway/observability_metrics_test.go diff --git a/internal/services/gateway/observability_metrics.go b/internal/services/gateway/observability_metrics.go index 100d3a5..0a560de 100644 --- a/internal/services/gateway/observability_metrics.go +++ b/internal/services/gateway/observability_metrics.go @@ -100,13 +100,16 @@ func (m *MetricsCollector) prune(now time.Time) { // Snapshot returns a point-in-time copy of the metrics. qps counts the requests // started within metricsQPSSpan of now. +// +// Every map in the result is a fresh copy, never the collector's own: the caller +// serializes the value AFTER this lock is released (the admin handler JSON-encodes +// it), so returning the live maps would let a concurrent Record mutate a map while +// the encoder iterates it — the Go runtime answers that with an unrecoverable +// fatal "concurrent map iteration and map write" that takes the whole process +// down (the 2026-10-03 21:00:28 crash on GET /api/v1/admin/observability/metrics). func (m *MetricsCollector) Snapshot(now time.Time) AdminMetrics { m.mu.Lock() defer m.mu.Unlock() - statuses := make(map[string]int64, len(m.statuses)) - for k, v := range m.statuses { - statuses[strconv.Itoa(k)] = v - } var qps int64 for _, t := range m.recent { if now.Sub(t) <= metricsQPSSpan { @@ -122,10 +125,31 @@ func (m *MetricsCollector) Snapshot(now time.Time) AdminMetrics { QPS: float64(qps), AvgLatencyMs: float64(avg), MaxLatencyMs: m.maxDur, - PerMethod: m.perMethod, - PerPath: m.perPath, - StatusCounts: statuses, + PerMethod: copyCounts(m.perMethod), + PerPath: copyCounts(m.perPath), + StatusCounts: copyStatusCounts(m.statuses), LLMCalls: m.llm, SubagentEvents: m.subagent, } } + +// copyCounts returns a detached copy of a string-keyed counter map, so the +// snapshot can be encoded after the collector's lock is released. The caller must +// hold m.mu while copying (the copy iterates the live map). +func copyCounts(src map[string]int64) map[string]int64 { + out := make(map[string]int64, len(src)) + for k, v := range src { + out[k] = v + } + return out +} + +// copyStatusCounts renders the int-keyed status counter as the string-keyed wire +// map the AdminMetrics JSON shape declares. The caller must hold m.mu while copying. +func copyStatusCounts(src map[int]int64) map[string]int64 { + out := make(map[string]int64, len(src)) + for k, v := range src { + out[strconv.Itoa(k)] = v + } + return out +} diff --git a/internal/services/gateway/observability_metrics_test.go b/internal/services/gateway/observability_metrics_test.go new file mode 100644 index 0000000..08056c4 --- /dev/null +++ b/internal/services/gateway/observability_metrics_test.go @@ -0,0 +1,242 @@ +package gateway + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" +) + +// These tests pin the contract that broke in production (2026-10-03 21:00:28): +// MetricsCollector.Snapshot must hand back DETACHED counter maps, because the +// admin handler JSON-encodes the snapshot after the collector's lock is released. +// Returning the collector's own maps let a concurrent Record write a map while the +// encoder iterated it → unrecoverable +// "fatal error: concurrent map iteration and map write" → the whole backend died, +// and with it every admin page (the reported「全链路流程功能无效」). + +// TestMetricsSnapshotDetachesMaps is the deterministic regression: a value read out +// of a snapshot must not change when the collector is counted afterwards, and +// writing into a snapshot must not reach the collector. +func TestMetricsSnapshotDetachesMaps(t *testing.T) { + t.Parallel() + m := newMetricsCollector() + m.Record(http.MethodGet, "/api/v1/a", http.StatusOK, 1) + + snap := m.Snapshot(time.Now()) + if got := snap.PerPath["/api/v1/a"]; got != 1 { + t.Fatalf("PerPath[/api/v1/a] = %d, want 1", got) + } + if got := snap.PerMethod[http.MethodGet]; got != 1 { + t.Fatalf("PerMethod[GET] = %d, want 1", got) + } + + // A count recorded after the snapshot must not appear in the older snapshot. + m.Record(http.MethodPost, "/api/v1/b", http.StatusTeapot, 1) + if _, ok := snap.PerPath["/api/v1/b"]; ok { + t.Fatalf("snapshot shares the collector's perPath map: later path leaked into it (%v)", snap.PerPath) + } + if _, ok := snap.PerMethod[http.MethodPost]; ok { + t.Fatalf("snapshot shares the collector's perMethod map: later method leaked into it (%v)", snap.PerMethod) + } + if _, ok := snap.StatusCounts["418"]; ok { + t.Fatalf("snapshot shares the collector's status map: later status leaked into it (%v)", snap.StatusCounts) + } + + assertSnapshotWriteIsolation(t, m, snap) + assertCollectorNotMutatedBySnapshot(t, m) +} + +// assertSnapshotWriteIsolation writes into the snapshot's maps and checks the +// collector is untouched afterwards. +func assertSnapshotWriteIsolation(t *testing.T, m *MetricsCollector, snap AdminMetrics) { + t.Helper() + snap.PerPath["/api/v1/evil"] = 999 + snap.PerMethod["DELETE"] = 999 + snap.StatusCounts["500"] = 999 + + after := m.Snapshot(time.Now()) + if _, ok := after.PerPath["/api/v1/evil"]; ok { + t.Fatalf("collector's perPath is the snapshot's map: %v", after.PerPath) + } + if _, ok := after.PerMethod["DELETE"]; ok { + t.Fatalf("collector's perMethod is the snapshot's map: %v", after.PerMethod) + } + if _, ok := after.StatusCounts["500"]; ok { + t.Fatalf("collector's status map is the snapshot's map: %v", after.StatusCounts) + } + if after.TotalRequests != 2 { + t.Fatalf("TotalRequests = %d, want 2", after.TotalRequests) + } + if after.PerPath["/api/v1/b"] != 1 { + t.Fatalf("PerPath[/api/v1/b] = %d, want 1 (real count lost)", after.PerPath["/api/v1/b"]) + } +} + +// assertCollectorNotMutatedBySnapshot re-reads a counter the collector recorded +// before the isolation writes to prove no counter value was disturbed. +func assertCollectorNotMutatedBySnapshot(t *testing.T, m *MetricsCollector) { + t.Helper() + final := m.Snapshot(time.Now()) + if final.PerPath["/api/v1/a"] != 1 { + t.Fatalf("PerPath[/api/v1/a] = %d, want 1", final.PerPath["/api/v1/a"]) + } + if final.PerMethod[http.MethodGet] != 1 { + t.Fatalf("PerMethod[GET] = %d, want 1", final.PerMethod[http.MethodGet]) + } +} + +// TestMetricsSnapshotEncodedConcurrently runs the exact interleaving that crashed +// production: one goroutine counts requests (the logging middleware does this on +// every request) while another encodes snapshots (the admin handler does this +// outside the lock). Under -race a shared live map is reported as a data race; on +// an unpatched build it can also abort the process outright. +func TestMetricsSnapshotEncodedConcurrently(t *testing.T) { + t.Parallel() + m := newMetricsCollector() + stop := make(chan struct{}) + var wg sync.WaitGroup + + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + m.Record(http.MethodGet, "/api/v1/x", http.StatusOK, 1) + } + }() + + deadline := time.Now().Add(300 * time.Millisecond) + for time.Now().Before(deadline) { + if _, err := json.Marshal(m.Snapshot(time.Now())); err != nil { + close(stop) + wg.Wait() + t.Fatalf("marshal snapshot: %v", err) + } + } + close(stop) + wg.Wait() +} + +// TestAdminMetricsEndpointConcurrentPoll drives the real HTTP path — the admin +// panel polls every module in parallel, so one request can be encoding the metrics +// JSON while another request's logging middleware records into the collector. This +// is the production crash reproduced end-to-end through the middleware + handler. +func TestAdminMetricsEndpointConcurrentPoll(t *testing.T) { + env := newEnv(t, nil, false) + tok := env.token(t, auth.RoleOwner) + metricsURL := env.ts.URL + "/api/v1/admin/observability/metrics" + healthURL := env.ts.URL + "/api/v1/health" + + // Both jobs run concurrently: the panel's metrics reads race the load requests + // that keep recording into the collector. + for _, err := range runConcurrent( + func() error { return driveRequests(healthURL, "", 4, 25) }, + func() error { return driveRequests(metricsURL, tok, 6, 25) }, + ) { + if err != nil { + t.Fatalf("concurrent metrics poll: %v", err) + } + } + + resp, body := doReq(t, http.MethodGet, metricsURL, nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var payload struct { + Data struct { + TotalRequests int64 `json:"totalRequests"` + PerPath map[string]int64 `json:"perPath"` + } `json:"data"` + } + if err := json.Unmarshal(body, &payload); err != nil { + t.Fatalf("decode metrics: %v (body=%s)", err, body) + } + if payload.Data.TotalRequests == 0 { + t.Fatal("totalRequests = 0, want the load above to be counted") + } + if payload.Data.PerPath["/api/v1/health"] == 0 { + t.Fatalf("perPath[/api/v1/health] = 0, want the health load counted (%v)", payload.Data.PerPath) + } +} + +// runConcurrent runs every job in its own goroutine and returns the non-nil errors. +func runConcurrent(jobs ...func() error) []error { + var wg sync.WaitGroup + errs := make([]error, len(jobs)) + for i, job := range jobs { + wg.Add(1) + go func(i int, job func() error) { + defer wg.Done() + errs[i] = job() + }(i, job) + } + wg.Wait() + return errs +} + +// driveRequests issues n GETs of url, workers of them in parallel, using token as +// a bearer credential when non-empty. It returns the first failure (status or I/O). +func driveRequests(url, token string, workers, n int) error { + var wg sync.WaitGroup + fail := make(chan error, workers) + + for w := 0; w < workers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < n; i++ { + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + select { + case fail <- err: + default: + } + return + } + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + select { + case fail <- err: + default: + } + return + } + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil { + select { + case fail <- err: + default: + } + return + } + if resp.StatusCode != http.StatusOK { + select { + case fail <- fmt.Errorf("GET %s status = %d (body=%s)", url, resp.StatusCode, body): + default: + } + return + } + } + }() + } + wg.Wait() + close(fail) + for err := range fail { + return err + } + return nil +}