AskaEth
0e1f163ef9
- 新增三个下行帧（纯新增，既有帧名/形状零改动）：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 需新增「可观测实时推送」一节
285 lines
10 KiB
Go
285 lines
10 KiB
Go
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)
|
||
}
|
||
}
|