feat(gateway,admin-web): 可观测面从轮询改为 WS 帧推送(裁决 05.4.3 + 16.4.3-WS)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 新增三个下行帧(纯新增,既有帧名/形状零改动):observability_log / observability_trace /
  observability_metrics,各带 ≤1s 节拍、无新数据不发帧、单帧上限(日志 200 / span 100)
- 帧壳 {type,data} 不带 sessionId;**不进 frameReachesAll 白名单** ⇒ fail-closed 只发 owner+admin
  (变异验证两次:加进白名单 → viewer 用例必红)
- 推送开启条件=owner/admin 凭据「读过一次」对应 REST(引导数据仍由该次读取返回),
  连接全断即退订并停循环(无常驻 goroutine);既有 REST 端点路径/RBAC/形状完全不变
- 前端:删除 logs/metrics/trace 三处 useAutoReload 轮询;新增 observabilityStream(引导+去重合并+窗口)
  与 observabilityViews(三个面板 + WS 徽标),长列表统一走已落地的 VirtualList
- 复核:build/vet 绿 + gateway 全包测试绿(含真 bus→真 /ws/chat 三角色帧断言)+ `-race`;
  前端 tsc 0 错 + vite build 成功
- 待登记契约:api-client 需新增「可观测实时推送」一节
This commit is contained in:
AskaEth committed 2026-10-04 13:58:03 +08:00
1 parent 74b12d87ca
commit 0e1f163ef9
10 files changed
+1449 -312

No files matched your search

+11 -2
View File
@@ -26,6 +26,9 @@ type observability struct {
metrics *MetricsCollector
traceBuf *RingBuffer
audit AuditStore
// watches 是日志/指标/trace 推送的订阅登记(裁决 05.4.3 + 16.4.3-WS)。
// 没有订阅者时推送循环不运行,见 observability_push.go。
watches obsWatches
// logBuf 是管理日志尾部端点背后的 logger 日志环形缓冲。
// 组合根在网关存在后经 SetLogBuffer 注入它;在那之前
// 该端点返回 501,而不是空但虚假的尾部。
@@ -38,6 +41,7 @@ func newObservability(deps Deps) *observability {
obs := &observability{
metrics: newMetricsCollector(),
traceBuf: newRingBuffer(defaultTraceBufCap),
watches: newObsWatches(),
}
if deps.AuditStore != nil {
obs.audit = deps.AuditStore
@@ -78,8 +82,11 @@ func (s *Server) recordMetric(method, path string, status int, durMs int64) {
s.obs.metrics.Record(method, path, status, durMs)
}
// adminObservabilityMetrics 返回当前网关指标(管理读)。
// adminObservabilityMetrics 返回当前网关指标(管理读)。读取同时充当
// 实时推送的订阅握手(见 observability_push.go):此后指标由 WS 帧推来,
// 面板不再轮询。
func (s *Server) adminObservabilityMetrics(w http.ResponseWriter, r *http.Request) {
s.watchObservability(r)
if s.obs == nil {
writeJSON(w, http.StatusOK, AdminMetrics{
PerMethod: map[string]int64{}, PerPath: map[string]int64{}, StatusCounts: map[string]int64{},
@@ -89,8 +96,10 @@ func (s *Server) adminObservabilityMetrics(w http.ResponseWriter, r *http.Reques
writeJSON(w, http.StatusOK, s.obs.metrics.Snapshot(time.Now()))
}
// adminObservabilityTrace 返回最近的已追踪 span(管理读)。
// adminObservabilityTrace 返回最近的已追踪 span(管理读)。读取同时充当
// 实时推送的订阅握手(见 observability_push.go)。
func (s *Server) adminObservabilityTrace(w http.ResponseWriter, r *http.Request) {
s.watchObservability(r)
if s.obs == nil {
writeJSON(w, http.StatusOK, []AdminTraceSpan{})
return
@@ -51,7 +51,12 @@ func (s *Server) logBuffer() *logger.RingBuffer {
// 优先,可选按最低级别过滤(管理读)。它是面板日志
// 尾部背后的后端:记录来自 logger 自身的环形
// 缓冲,没有任何东西被重新渲染或编造。
//
// 读取同时充当实时推送的订阅握手(裁决 05.4.3 + 16.4.3-WS,见
// observability_push.go):此后新日志由 observability_log 帧推来,
// 面板不再轮询,分级过滤改在本地进行(帧是广播的,无法按连接分级)。
func (s *Server) adminObservabilityLogs(w http.ResponseWriter, r *http.Request) {
s.watchObservability(r)
minLevel, filtered, ok := parseLogLevel(r.URL.Query().Get("level"))
if !ok {
writeError(w, &apperr.AppError{
@@ -0,0 +1,284 @@
package gateway
import (
"bytes"
"context"
"encoding/json"
"net/http"
"sync"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
)
// 本文件是裁决 05.4.3 + 16.4.3-WS 的落点:日志 / 指标 / trace 从「前端轮询
// 管理 REST」改为「网关主动推 WS 帧」(design/10 §5 口径)。
//
// 两条边界分开,谁都不越界:
//
// - **谁能收到帧**:仍由 ws_frame_scope.go 的角色门决定。这三个帧类型没有
// 登记进 frameReachesAll,因此按 fail-closed 默认档只发 owner + admin;
// viewer 继续走既有 REST 只读档取数(design/11 §7.1)。
// - **什么时候发帧**:由本文件的订阅登记决定。没有订阅者时推送循环不启动,
// 网关不为可观测数据做任何周期性工作 —— 这正是「推送替代轮询」相对
// 「每个面板各自 1s 轮询」的收益所在。
//
// 订阅从哪来:面板挂载时的首次读取(GET /api/v1/admin/observability/{logs,
// metrics,trace})兼任订阅握手。上游订阅帧需要 ws_route.go 的上行分发开口,
// 那不在本次可动的文件面内;用「读过实时数据」作显式订阅动作语义等价,且天然
// 只由真正打开过面板的凭据触发。订阅以凭据身份(sub + role + iat,等价于那次
// 登录)为键,其连接全部断开后自动退订。
const (
// wsObservabilityLog / Trace / Metrics 是三个下行帧类型。帧壳与既有下行
// 帧一致(wsEvent{type,data}),不带 sessionId:它们不属任何会话。
wsObservabilityLog = "observability_log"
wsObservabilityTrace = "observability_trace"
wsObservabilityMetrics = "observability_metrics"
// 产生这三个帧的 bus 主题。ws_bridge.go 的 eventBridge 把主题映射成上面的
// 帧类型,因此推送走的是与其它实时事件完全相同的
// 「bus → relay → 角色门 → 连接」路径(不新开通道、不新开端口)。
obsLogSubject = "observability.log"
obsTraceSubject = "observability.trace"
obsMetricsSubject = "observability.metrics"
// observabilityPushInterval 是推送循环的节拍:每拍最多各发一帧
// (日志增量 / 新 span / 变化过的指标快照),只发真的有新数据的那类。
observabilityPushInterval = time.Second
// maxLogPushBatch / maxSpanPushBatch 界定单帧载荷大小。一拍的突发超过上限
// 时只发最新的那些 —— 更旧的条目本就落在面板保留窗口之外,与既有 REST 尾部
// (limit 上限)语义一致,因此不丢「看得见」的数据。
maxLogPushBatch = 200
maxSpanPushBatch = 100
)
// AdminLogPush 是 observability_log 帧的 data:一批日志条目,**最新在前**
// (与 GET /api/v1/admin/observability/logs 同序),面板据此前缀拼接。
type AdminLogPush struct {
Entries []AdminLogEntry `json:"entries"`
}
// AdminTracePush 是 observability_trace 帧的 data:一批新完成的 span,
// **最新在前**(与 GET /api/v1/admin/observability/trace 同序)。
type AdminTracePush struct {
Spans []AdminTraceSpan `json:"spans"`
}
// obsWatchKey 是「谁在订阅」的键:凭据身份,等价于那次登录。用 iat 而不用
// token 串,避免把凭据本身留在内存里。
type obsWatchKey struct {
sub string
role auth.Role
iat int64
}
// watchKeyOf 把连接/请求的凭据折叠成订阅键。
func watchKeyOf(claims *auth.Claims) obsWatchKey {
return obsWatchKey{sub: claims.Sub, role: claims.Role, iat: claims.IssuedAt}
}
// obsWatches 是订阅登记表。它与推送循环同生命周期:第一个订阅建立时启动
// 循环,最后一个订阅被剪掉时循环自行退出,因此订阅表空着的时候没有任何
// 常驻 goroutine,也没有任何周期性开销。
type obsWatches struct {
mu sync.Mutex
set map[obsWatchKey]struct{}
running bool
}
// newObsWatches 构建一个空的订阅登记表。
func newObsWatches() obsWatches {
return obsWatches{set: make(map[obsWatchKey]struct{})}
}
// add 登记一个订阅,必要时启动推送循环。没有 bus 就没有下行通道,
// 登记也不会产生任何推送,因此直接跳过(不启动循环)。
func (w *obsWatches) add(s *Server, key obsWatchKey) {
if s == nil || s.bus == nil {
return
}
w.mu.Lock()
w.set[key] = struct{}{}
start := !w.running
if start {
w.running = true
}
w.mu.Unlock()
if start {
go w.loop(s)
}
}
// loop 是推送循环:每拍先剪掉「凭据已不在线」的订阅,再按需发布三类数据。
// 返回即代表循环已退出(下一个订阅会重新启动它)。
//
// 锚点在循环启动时取定(见 newObsPushState):只发订阅之后新产生的数据,
// 历史由面板自己那次引导读取提供。
func (w *obsWatches) loop(s *Server) {
ticker := time.NewTicker(observabilityPushInterval)
defer ticker.Stop()
state := newObsPushState(s)
for range ticker.C {
if !w.prune(s) {
return
}
state.publish(s)
}
}
// prune 剪掉没有在线 owner/admin 连接的订阅,并报告循环是否还要继续。
// 「订阅表空了」与「退出」必须在同一把锁里判定:否则一个新订阅可能刚看到
// running=true(以为循环还活着),而循环已经走向退出,推送就此静默停摆。
func (w *obsWatches) prune(s *Server) bool {
w.mu.Lock()
defer w.mu.Unlock()
live := liveWatchKeys(s)
for k := range w.set {
if !live[k] {
delete(w.set, k)
}
}
if len(w.set) == 0 {
w.running = false
return false
}
return true
}
// liveWatchKeys 返回当前在线 owner/admin 连接的凭据身份集合。
func liveWatchKeys(s *Server) map[obsWatchKey]bool {
out := make(map[obsWatchKey]bool)
if s == nil || s.hub == nil {
return out
}
s.hub.mu.RLock()
defer s.hub.mu.RUnlock()
for c := range s.hub.conns {
if c.claims == nil || !ownerAdminClaims(c.claims) {
continue
}
out[watchKeyOf(c.claims)] = true
}
return out
}
// watchObservability 把一次可观测读取登记为订阅。只有 owner/admin 会登记:
// viewer 仍走纯 REST 只读档,不被推送(design/10 §5)。读取本身照常返回引导
// 数据,因此面板不需要轮询也能拿到首屏。
func (s *Server) watchObservability(r *http.Request) {
if s == nil || s.obs == nil || r == nil {
return
}
claims, ok := auth.ClaimsFromContext(r.Context())
if !ok || claims == nil || !ownerAdminClaims(claims) {
return
}
s.obs.watches.add(s, watchKeyOf(claims))
}
// obsPushState 是推送循环的游标:日志/span 已经发到哪里、上一次发出的指标
// 指纹是什么。它只被循环自己的 goroutine 触碰,因此不需要锁。
type obsPushState struct {
lastLogTime time.Time
lastSpanSeq uint64
lastMetrics []byte
}
// newObsPushState 取定本轮的推送锚点:日志取当前最新条目的时间、trace 取当前
// 写入序号、指标取当前快照的指纹。三者都表示「截至此刻的数据面板已经通过
// REST 引导读取看到过了」,因此循环不会把环形缓冲里的历史整段灌成巨型帧。
func newObsPushState(s *Server) *obsPushState {
st := &obsPushState{}
if buf := s.logBuffer(); buf != nil {
if entries := buf.Snapshot(1); len(entries) == 1 {
st.lastLogTime = entries[0].Time
}
}
if s.obs != nil {
st.lastSpanSeq = s.obs.traceBuf.Seq()
if raw, err := json.Marshal(s.obs.metrics.Snapshot(time.Now())); err == nil {
st.lastMetrics = raw
}
}
return st
}
// publish 发布本拍新增的数据:日志增量、新 span、变化过的指标快照。
// 三类互相独立 —— 没有新数据的那类不发帧(design/10 §5「推送只在有数据时
// 发一帧」),因此空转的服务端一帧都不发。
func (st *obsPushState) publish(s *Server) {
st.publishLogs(s)
st.publishSpans(s)
st.publishMetrics(s)
}
// publishLogs 把锚点之后的新日志条目发成一帧(最新在前)。日志环形缓冲未注入
// (网关没接组合根的日志尾部)时没有数据源,直接跳过。
func (st *obsPushState) publishLogs(s *Server) {
buf := s.logBuffer()
if buf == nil {
return
}
entries := buf.Snapshot(maxLogPushBatch)
fresh := make([]AdminLogEntry, 0, len(entries))
for _, e := range entries {
if !e.Time.After(st.lastLogTime) {
break // 最新在前:一旦遇到不新的条目,后面的只会更旧
}
fresh = append(fresh, adminLogOf(e))
}
if len(fresh) == 0 {
return
}
st.lastLogTime = entries[0].Time
s.publishObservability(obsLogSubject, AdminLogPush{Entries: fresh})
}
// publishSpans 把锚点之后新完成的 span 发成一帧(最新在前)。
func (st *obsPushState) publishSpans(s *Server) {
if s.obs == nil {
return
}
entries := s.obs.traceBuf.SnapshotSince(st.lastSpanSeq, maxSpanPushBatch)
if len(entries) == 0 {
return
}
spans := make([]AdminTraceSpan, 0, len(entries))
for _, e := range entries {
spans = append(spans, adminSpanOf(e.rec))
}
st.lastSpanSeq = entries[0].seq
s.publishObservability(obsTraceSubject, AdminTracePush{Spans: spans})
}
// publishMetrics 在当前指标快照与上次发出的不同时发一帧。用序列化结果做指纹:
// 快照的 map 副本会被编码,编码是确定性的(键排序),因此「没变」可以精确判定,
// 而 QPS 这类会随时间衰减的量自然在变化时照常推。
func (st *obsPushState) publishMetrics(s *Server) {
if s.obs == nil {
return
}
snap := s.obs.metrics.Snapshot(time.Now())
raw, err := json.Marshal(snap)
if err != nil {
return
}
if bytes.Equal(raw, st.lastMetrics) {
return
}
st.lastMetrics = raw
s.publishObservability(obsMetricsSubject, snap)
}
// publishObservability 把一个可观测事件发上 bus:它会经 ws_bridge.go 的主题
// 映射变成下行帧,并照常过一遍角色门(本包唯一的可见性决策点)。bus 缺席时
// 是 no-op。
func (s *Server) publishObservability(subject string, payload any) {
if s == nil || s.bus == nil {
return
}
if err := s.bus.Publish(context.Background(), subject, payload); err != nil {
s.log.Warn("publish observability frame", "subject", subject, "err", err)
}
}
@@ -0,0 +1,346 @@
package gateway
import (
"context"
"encoding/json"
"net/http"
"testing"
"time"
"github.com/gorilla/websocket"
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// 本文件锁住 05.4.3 + 16.4.3-WS(可观测面 WS 推送,替换轮询):
//
// - 帧壳与载荷:observability_log / observability_trace / observability_metrics
// 三个帧类型经 ws_bridge 的主题映射成形,data 与既有 REST 端点同形;
// - 角色门:三者都不是 design/11 §7.1 的状态机信号,因此 fail-closed,
// 只发 owner + admin,viewer 收不到(新增一帧忘记分类也不会泄露);
// - 推送时机:只有建立了订阅(面板读取过一次实时数据)才启动推送循环 ——
// 这是「推送替代轮询」不把成本转嫁给并发面板的那一半,也保证未订阅的
// 连接不会凭空收到周期性帧。
//
// 断言都走真路径:真 bus Publish(或真订阅 + 真推送循环)→ hub 的真订阅与中继
// → 真 /ws/chat 连接(真 JWT)。
// obsPushReadBudget 是等待推送帧的总预算。推送循环的节拍是
// observabilityPushInterval,一个节拍内应到齐全部三类帧。
const obsPushReadBudget = 6 * time.Second
// TestObservabilityFramesReachOwnerAndAdminOnly 在真实主题上发布三类可观测
// 载荷:owner/admin 收到全部三帧且载荷键集与登记的一致,viewer 只收到白名单
// 的哨兵帧。
func TestObservabilityFramesReachOwnerAndAdminOnly(t *testing.T) {
t.Parallel()
for _, typ := range []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics} {
if frameReachesAll(typ) {
t.Fatalf("%s reached every connection; an observability frame carries content and must stay owner/admin only", typ)
}
}
env := newEnv(t, nil, false)
conns := map[auth.Role]*websocket.Conn{}
for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin, auth.RoleViewer} {
conn := dialWS(t, env.ts.URL, env.token(t, role))
defer conn.Close()
conns[role] = conn
}
waitConnCount(t, env.srv, len(conns))
logs := AdminLogPush{Entries: []AdminLogEntry{
{Time: "2026-10-04T10:00:01Z", Level: "warn", Source: "gateway/observability", Msg: "推送测试"},
}}
spans := AdminTracePush{Spans: []AdminTraceSpan{
{ID: "span-push-1", Name: "llm.call", Kind: "llm_call", Status: "ok",
StartedAt: "2026-10-04T10:00:02Z", EndedAt: "2026-10-04T10:00:03Z", DurationNs: 1234},
}}
metrics := AdminMetrics{
TotalRequests: 7, QPS: 1, AvgLatencyMs: 5, MaxLatencyMs: 9,
PerMethod: map[string]int64{"GET": 7}, PerPath: map[string]int64{"/api/v1/x": 7},
StatusCounts: map[string]int64{"200": 7}, LLMCalls: 2, SubagentEvents: 1,
}
for _, pub := range []struct {
subject string
payload any
}{
{obsLogSubject, logs},
{obsTraceSubject, spans},
{obsMetricsSubject, metrics},
} {
if err := env.bus.Publish(context.Background(), pub.subject, pub.payload); err != nil {
t.Fatalf("publish %s: %v", pub.subject, err)
}
}
// 哨兵:一个白名单状态信号收尾,使「viewer 什么都没收到」不会与
// 「连接根本没工作」混淆(与既有帧作用域用例同一手法)。
if err := env.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil {
t.Fatalf("publish sentinel: %v", err)
}
for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} {
frames := drainFrames(t, conns[role])
for _, typ := range []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics} {
assertFrameCount(t, string(role), frames, typ, 1)
if frame := findFrame(t, string(role), frames, typ); frame.SessionID != "" {
t.Fatalf("%s %s frame sessionId = %q, want none: observability frames are not session-scoped",
role, typ, frame.SessionID)
}
}
if len(frames) != 4 {
t.Fatalf("%s received %d frames, want the three observability frames + the sentinel: %v",
role, len(frames), frameTypesOf(frames))
}
assertLogPushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityLog), logs)
assertTracePushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityTrace), spans)
assertMetricsPushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityMetrics), metrics)
}
viewerFrames := drainFrames(t, conns[auth.RoleViewer])
if len(viewerFrames) != 1 || viewerFrames[0].Type != scopeSentinel {
t.Fatalf("viewer received %v, want only the whitelisted %q: observability frames are owner/admin only (design/11 §7.1)",
frameTypesOf(viewerFrames), scopeSentinel)
}
}
// TestObservabilityPushStreamsToSubscribers 驱动真实的生产者路径:面板的引导
// 读取建立订阅 → 推送循环取锚点 → 之后新产生的日志 / 指标 / span 各发一帧,
// owner 收得到;同一次推送里 viewer 一帧都收不到(角色门)。
func TestObservabilityPushStreamsToSubscribers(t *testing.T) {
t.Parallel()
env := newEnv(t, nil, false)
ownerToken := env.token(t, auth.RoleOwner)
viewerToken := env.token(t, auth.RoleViewer)
owner := dialWS(t, env.ts.URL, ownerToken)
defer owner.Close()
viewer := dialWS(t, env.ts.URL, viewerToken)
defer viewer.Close()
waitConnCount(t, env.srv, 2)
ring := logger.NewRingBuffer(64)
env.srv.SetLogBuffer(ring)
// 面板挂载时的引导读取:既拿首屏,也建立订阅(watchObservability)。
resp, _ := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/observability/logs?limit=5", nil, ownerToken)
if resp.StatusCode != http.StatusOK {
t.Fatalf("bootstrap log read = %d, want 200: the read must keep serving the guide data", resp.StatusCode)
}
// 等循环起步(它在启动时取锚点:只发订阅之后的数据)。
time.Sleep(300 * time.Millisecond)
ring.Append(logger.Entry{Time: time.Now(), Level: "warn", Source: "gateway/push-test", Msg: "订阅后的新日志"})
env.srv.recordMetric(http.MethodGet, "/api/v1/admin/push-test", http.StatusOK, 4)
env.srv.ObserveSpan(trace.SpanRecord{
ID: "span-sub", Name: "llm.call", Kind: trace.SpanKindLLM, Status: trace.SpanStatusOK,
StartedAt: time.Now(), EndedAt: time.Now(), Duration: 3 * time.Millisecond,
})
wanted := []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics}
frames := readFramesUntil(t, owner, wanted, obsPushReadBudget)
for _, typ := range wanted {
assertFrameCount(t, "owner", frames, typ, 1)
}
logFrame := findFrame(t, "owner", frames, wsObservabilityLog)
entries := decodeFrameEntries(t, "owner", logFrame)
if len(entries) != 1 || entries[0].Msg != "订阅后的新日志" || entries[0].Source != "gateway/push-test" || entries[0].Level != "warn" {
t.Fatalf("pushed log entry = %+v, want the appended entry", entries)
}
var traceData struct {
Spans []AdminTraceSpan `json:"spans"`
}
unmarshalFrameData(t, "owner", findFrame(t, "owner", frames, wsObservabilityTrace), &traceData)
if len(traceData.Spans) != 1 || traceData.Spans[0].ID != "span-sub" || traceData.Spans[0].Kind != "llm_call" {
t.Fatalf("pushed spans = %+v, want the observed span", traceData.Spans)
}
var pushedMetrics AdminMetrics
unmarshalFrameData(t, "owner", findFrame(t, "owner", frames, wsObservabilityMetrics), &pushedMetrics)
if pushedMetrics.TotalRequests < 1 || pushedMetrics.LLMCalls < 1 {
t.Fatalf("pushed metrics = %+v, want the recorded request and the observed LLM span", pushedMetrics)
}
// 同一次推送里 viewer 依旧一帧都拿不到:门在 hub,与订阅无关。
viewerFrames := readFramesUntil(t, viewer, wanted, 400*time.Millisecond)
for _, typ := range wanted {
assertFrameCount(t, "viewer", viewerFrames, typ, 0)
}
}
// TestObservabilityPushRequiresSubscription 是推送时机的反证:连接上没有任何
// 人读取过实时数据 → 没有订阅 → 推送循环根本不启动,即使数据在连接建立后
// 立刻变化,也不会凭空出现周期性帧(这正是既有「精确帧数」用例得以保持的
// 前提,也是推送替代轮询不把成本转嫁给所有连接的那一半)。
func TestObservabilityPushRequiresSubscription(t *testing.T) {
t.Parallel()
env := newEnv(t, nil, false)
token := env.token(t, auth.RoleOwner)
conn := dialWS(t, env.ts.URL, token)
defer conn.Close()
waitConnCount(t, env.srv, 1)
ring := logger.NewRingBuffer(16)
env.srv.SetLogBuffer(ring)
ring.Append(logger.Entry{Time: time.Now(), Level: "info", Source: "gateway/no-sub", Msg: "无人订阅"})
env.srv.recordMetric(http.MethodGet, "/api/v1/admin/no-sub", http.StatusOK, 1)
env.srv.ObserveSpan(trace.SpanRecord{
ID: "span-no-sub", Name: "llm.call", Kind: trace.SpanKindLLM, Status: trace.SpanStatusOK,
StartedAt: time.Now(), EndedAt: time.Now(), Duration: time.Millisecond,
})
// 覆盖一个完整节拍:未订阅时这里必须一帧都没有。
time.Sleep(observabilityPushInterval + 200*time.Millisecond)
frames := drainFrames(t, conn)
if len(frames) != 0 {
t.Fatalf("an unsubscribed connection received %v; the push loop must not run without a subscription", frameTypesOf(frames))
}
}
// assertLogPushFrame 断言一帧 observability_log 的 data 形状(键集恰好
// {"entries"})与条目内容。
func assertLogPushFrame(t *testing.T, role string, frame subagentFrame, want AdminLogPush) {
t.Helper()
assertFrameDataKeys(t, role, frame, []string{"entries"})
entries := decodeFrameEntries(t, role, frame)
if len(entries) != len(want.Entries) {
t.Fatalf("%s log entries = %d, want %d", role, len(entries), len(want.Entries))
}
for i, e := range entries {
if e != want.Entries[i] {
t.Fatalf("%s log entry[%d] = %+v, want %+v", role, i, e, want.Entries[i])
}
}
}
// decodeFrameEntries 解出 observability_log 帧的条目,并锁住每条目的键集。
func decodeFrameEntries(t *testing.T, role string, frame subagentFrame) []AdminLogEntry {
t.Helper()
var data struct {
Entries []json.RawMessage `json:"entries"`
}
unmarshalFrameData(t, role, frame, &data)
out := make([]AdminLogEntry, 0, len(data.Entries))
for _, raw := range data.Entries {
assertRawObjectKeys(t, role+" log entry", raw, []string{"time", "level", "source", "msg"})
var e AdminLogEntry
if err := json.Unmarshal(raw, &e); err != nil {
t.Fatalf("%s log entry %s: %v", role, raw, err)
}
out = append(out, e)
}
return out
}
// assertTracePushFrame 断言一帧 observability_trace 的 data 形状(键集恰好
// {"spans"})与 span 内容。
func assertTracePushFrame(t *testing.T, role string, frame subagentFrame, want AdminTracePush) {
t.Helper()
var data struct {
Spans []json.RawMessage `json:"spans"`
}
assertFrameDataKeys(t, role, frame, []string{"spans"})
unmarshalFrameData(t, role, frame, &data)
if len(data.Spans) != len(want.Spans) {
t.Fatalf("%s spans = %d, want %d", role, len(data.Spans), len(want.Spans))
}
assertRawObjectKeys(t, role+" span", data.Spans[0],
[]string{"id", "name", "kind", "status", "startedAt", "endedAt", "durationNs"})
var sp AdminTraceSpan
if err := json.Unmarshal(data.Spans[0], &sp); err != nil {
t.Fatalf("%s span %s: %v", role, data.Spans[0], err)
}
if sp.ID != want.Spans[0].ID || sp.Name != want.Spans[0].Name || sp.Kind != want.Spans[0].Kind ||
sp.Status != want.Spans[0].Status || sp.DurationNs != want.Spans[0].DurationNs {
t.Fatalf("%s span = %+v, want %+v", role, sp, want.Spans[0])
}
}
// assertMetricsPushFrame 断言一帧 observability_metrics 的 data 就是 REST 端点
// 那份 AdminMetrics(键集逐个锁定),避免推送与轮询两条路给出不同形状。
func assertMetricsPushFrame(t *testing.T, role string, frame subagentFrame, want AdminMetrics) {
t.Helper()
wantKeys := []string{
"totalRequests", "qps", "avgLatencyMs", "maxLatencyMs", "perMethod",
"perPath", "statusCounts", "llmCalls", "subagentEvents",
}
assertFrameDataKeys(t, role, frame, wantKeys)
var got AdminMetrics
unmarshalFrameData(t, role, frame, &got)
if got.TotalRequests != want.TotalRequests || got.QPS != want.QPS ||
got.AvgLatencyMs != want.AvgLatencyMs || got.MaxLatencyMs != want.MaxLatencyMs ||
got.LLMCalls != want.LLMCalls || got.SubagentEvents != want.SubagentEvents ||
got.PerPath["/api/v1/x"] != 7 || got.StatusCounts["200"] != 7 {
t.Fatalf("%s metrics = %+v, want %+v (REST shape)", role, got, want)
}
}
// assertFrameDataKeys 断言帧的 data 是键集恰好为 want 的 JSON 对象。
func assertFrameDataKeys(t *testing.T, role string, frame subagentFrame, want []string) {
t.Helper()
assertRawObjectKeys(t, role+" frame data", frame.Data, want)
}
// assertRawObjectKeys 断言 raw 是键集恰好为 want 的 JSON 对象。
func assertRawObjectKeys(t *testing.T, what string, raw json.RawMessage, want []string) {
t.Helper()
var keys map[string]json.RawMessage
if err := json.Unmarshal(raw, &keys); err != nil {
t.Fatalf("%s is not a JSON object: %v (%s)", what, err, raw)
}
if len(keys) != len(want) {
t.Fatalf("%s keys = %v, want exactly %v (api-client 登记的形状)", what, keys, want)
}
for _, k := range want {
if _, ok := keys[k]; !ok {
t.Fatalf("%s lacks key %q: %v", what, k, keys)
}
}
}
// unmarshalFrameData 把帧的 data 解进 out。
func unmarshalFrameData(t *testing.T, role string, frame subagentFrame, out any) {
t.Helper()
if err := json.Unmarshal(frame.Data, out); err != nil {
t.Fatalf("%s frame data %s: %v", role, frame.Data, err)
}
}
// readFramesUntil 读连接直到 wanted 里的每个帧类型都至少出现一次,或预算耗尽。
// 它返回已收到的帧(超时也返回),使「帧没来」的失败信息里能看到到底来了什么。
func readFramesUntil(t *testing.T, conn *websocket.Conn, wanted []string, budget time.Duration) []subagentFrame {
t.Helper()
deadline := time.Now().Add(budget)
out := make([]subagentFrame, 0, len(wanted))
for !hasAllFrameTypes(out, wanted) && time.Now().Before(deadline) {
_ = conn.SetReadDeadline(deadline)
_, msg, err := conn.ReadMessage()
if err != nil {
return out
}
var frame subagentFrame
if uerr := json.Unmarshal(msg, &frame); uerr != nil {
t.Fatalf("decode frame %s: %v", msg, uerr)
}
out = append(out, frame)
}
return out
}
// hasAllFrameTypes 报告 frames 是否已经覆盖 wanted 里的每个类型。
func hasAllFrameTypes(frames []subagentFrame, wanted []string) bool {
for _, typ := range wanted {
found := false
for _, f := range frames {
if f.Type == typ {
found = true
break
}
}
if !found {
return false
}
}
return true
}
@@ -31,10 +31,20 @@ type AdminTraceSpan struct {
// 最近的工作而不会无界增长。
type RingBuffer struct {
mu sync.RWMutex
buf []trace.SpanRecord
buf []traceEntry
next int
full bool
cap int
// seq 是最后写入 span 的序号(单调递增,从 1 起)。
// 增量推送(observability_trace 帧)用它做游标,见 SnapshotSince。
seq uint64
}
// traceEntry 是环形缓冲内的一个存储格。序号随写入单调递增,因此
// 下游可以「只取比游标新的」而不是靠时间戳去猜。
type traceEntry struct {
seq uint64
rec trace.SpanRecord
}
// newRingBuffer 构建至多容纳 cap 个 span 的环形缓冲(cap<=0 时用默认值)。
@@ -42,20 +52,28 @@ func newRingBuffer(cap int) *RingBuffer {
if cap <= 0 {
cap = defaultTraceBufCap
}
return &RingBuffer{buf: make([]trace.SpanRecord, cap), cap: cap}
return &RingBuffer{buf: make([]traceEntry, cap), cap: cap}
}
// Append 存储 rec,缓冲满时逐出最旧的。
func (r *RingBuffer) Append(rec trace.SpanRecord) {
r.mu.Lock()
defer r.mu.Unlock()
r.buf[r.next] = rec
r.seq++
r.buf[r.next] = traceEntry{seq: r.seq, rec: rec}
r.next = (r.next + 1) % r.cap
if r.next == 0 {
r.full = true
}
}
// Seq 返回最后写入 span 的序号(还没写过任何 span 时为 0)。
func (r *RingBuffer) Seq() uint64 {
r.mu.RLock()
defer r.mu.RUnlock()
return r.seq
}
// Snapshot 返回至多 n 条最近 span,最新优先。非正的 n
// 选用 defaultObsLimit;n 被夹取到 maxObsLimit。
func (r *RingBuffer) Snapshot(n int) []trace.SpanRecord {
@@ -77,7 +95,32 @@ func (r *RingBuffer) Snapshot(n int) []trace.SpanRecord {
out := make([]trace.SpanRecord, 0, count)
for i := 0; i < count; i++ {
idx := (r.next - 1 - i + r.cap*2) % r.cap
out = append(out, r.buf[idx])
out = append(out, r.buf[idx].rec)
}
return out
}
// SnapshotSince 返回序号大于 since 的至多 n 条 span,最新优先(与 Snapshot
// 同序),供增量推送使用。n 非正时只用 defaultObsLimit 兜底;调用方按
// 「帧载荷有上限」的约束传入自己的上限。
func (r *RingBuffer) SnapshotSince(since uint64, n int) []traceEntry {
if n <= 0 {
n = defaultObsLimit
}
r.mu.RLock()
defer r.mu.RUnlock()
count := r.next
if r.full {
count = r.cap
}
out := make([]traceEntry, 0, min(count, n))
for i := 0; i < count && len(out) < n; i++ {
idx := (r.next - 1 - i + r.cap*2) % r.cap
e := r.buf[idx]
if e.seq <= since {
break
}
out = append(out, e)
}
return out
}
+9
View File
@@ -26,6 +26,12 @@ var eventBridge = map[string]string{
"automation.result": "automation_result",
"reminder.trigger": "reminder",
orchestration.SubagentResultSubject: wsSubagentResult,
// 可观测面(裁决 05.4.3 + 16.4.3-WS):日志/指标/trace 的推送帧。载荷形状
// 与既有 REST 端点同形(见 observability_push.go),因此面板换成消费帧后
// 不需要第二套解码。
obsLogSubject: wsObservabilityLog,
obsTraceSubject: wsObservabilityTrace,
obsMetricsSubject: wsObservabilityMetrics,
}
// bridgePatterns 是 hub 在 ws.> 之外额外订阅的 bus 模式,用于
@@ -36,6 +42,9 @@ var bridgePatterns = []string{
"automation.>",
"reminder.>",
"cyrene.agent.*.event",
// 可观测推送(observability.log/trace/metrics):日志/指标/trace 的增量
// 帧都走这个主题族,见 observability_push.go。
"observability.>",
// 后台子代理批次结果:它既不匹配 ws.>,也不是 cyrene.agent 的 4 段广播形态,
// 所以必须在这里登记订阅,否则后台结果永远上不了前台(只能看到进度帧)。
orchestration.SubagentResultSubject,