fix(gateway): 修 Snapshot 交出活 map 导致进程被 fatal 打死（并发 map 迭代+写入）
现场：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）。
This commit is contained in:
AskaEth committed
1 parent
08b3a01752
commit
5847705e35
2 files changed
+273
-7
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in new issue
Block a user