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 +}