package gateway import ( "sync" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // defaultTraceBufCap 是环形缓冲在逐出最旧 span 之前持有的已 // 完成 span 数。编译期常量。 const defaultTraceBufCap = 512 // AdminTraceSpan 是一个已追踪 span 的驼峰式线上结构。它把 // 前端与 trace 包蛇形命名的 SpanRecord 解耦。 type AdminTraceSpan struct { ID string `json:"id"` ParentID string `json:"parentId,omitempty"` Name string `json:"name"` Kind string `json:"kind"` Status string `json:"status"` StartedAt string `json:"startedAt"` EndedAt string `json:"endedAt,omitempty"` DurationNs int64 `json:"durationNs"` Props map[string]any `json:"props,omitempty"` Events []trace.Event `json:"events,omitempty"` } // RingBuffer 是已完成的 trace span 的并发安全有界缓冲。缓冲 // 满时逐出最旧 span,因此管理面板总能看到 // 最近的工作而不会无界增长。 type RingBuffer struct { mu sync.RWMutex 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 时用默认值)。 func newRingBuffer(cap int) *RingBuffer { if cap <= 0 { cap = defaultTraceBufCap } 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.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 { if n <= 0 { n = defaultObsLimit } if n > maxObsLimit { n = maxObsLimit } r.mu.RLock() defer r.mu.RUnlock() count := r.next if r.full { count = r.cap } if count > n { count = n } 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].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 } // traceSink 把 *RingBuffer 适配为 trace.Writer,使已完成的 span 被 // 追加进网关的 trace 缓冲。 type traceSink struct{ b *RingBuffer } // WriteSpan 把 rec 追加进环形缓冲。 func (s traceSink) WriteSpan(rec trace.SpanRecord) error { if s.b == nil { return nil } s.b.Append(rec) return nil } // MultiTraceWriter 把已完成的 span 扇出到多个 trace.Writer,使一个 // tracer 能一趟同时记录到网关的环形缓冲与一个 JSONL 文件。nil 的 // Writer 被跳过。它是主会话用来把 loop tracer 的输出引入 // 网关而不丢掉 stderr 的组合根辅助件。 type MultiTraceWriter struct { // Writers 是 span 被写入的汇聚点。nil 的 Writer 被跳过。 Writers []trace.Writer } // WriteSpan 把 rec 写入每一个已配置的 Writer。 func (m MultiTraceWriter) WriteSpan(rec trace.SpanRecord) error { for _, w := range m.Writers { if w == nil { continue } if err := w.WriteSpan(rec); err != nil { return err } } return nil } // adminSpanOf 把 trace.SpanRecord 映射为驼峰式管理线上结构。 // // 🔴 读侧脱敏(M42 #3,M38 #1 的读侧放大面):入参先过 `trace.SanitizeRecord` // —— 与写侧(stderr 的 JSONL,trace.JSONRecorder)**同一份投影**,不另写一套。 // 环形缓冲(ObserveSpan → traceBuf)持的是**原始**记录,因此 REST 尾部 // (GET /api/v1/admin/observability/trace)与 observability_trace 推送帧这两条 // 出图路径都必须在这里收口,否则面板就是「凭据与正文的最后一面」。 // // ⚠️ 这是**有意的行为变更**(admin 面有意改变响应内容):`props` 从原始值变成 // 脱敏投影 —— 白名单(trace 包 safePropKeys)内的键逐值打码+截断,表外键折成 // `omitted` 里的 Digest(留键名、留体积指纹、去正文),事件里的正文键同样折成 // Digest。输出**字段名与帧形状不变**(id/parentId/name/kind/status/startedAt/ // endedAt/durationNs/props/events 一个不改),客户端因此**不得依赖未声明的 // props**(即不得假定某个表外键一定带原文)—— 契约口径本就是「props 只含已 // 声明键」。 // // 排障字段一律保住:status / 时间 / durationNs / name / kind 都不是内容面, // 原样透出;`omitted` 的存在还让「这次调用有多大输出」仍可回答。 func adminSpanOf(rec trace.SpanRecord) AdminTraceSpan { rec = trace.SanitizeRecord(rec) sp := AdminTraceSpan{ ID: rec.ID, ParentID: rec.ParentID, Name: rec.Name, Kind: string(rec.Kind), Status: string(rec.Status), StartedAt: rec.StartedAt.UTC().Format(time.RFC3339), DurationNs: rec.Duration.Nanoseconds(), Props: rec.Props, Events: rec.Events, } if !rec.EndedAt.IsZero() { sp.EndedAt = rec.EndedAt.UTC().Format(time.RFC3339) } return sp }