package gateway import ( "net/http" "strconv" "sync/atomic" "time" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // trace/audit 端点共享的默认列表上限。编译期常量, // 因此运行路径中不出现魔法数字(项目规范 §10)。 const ( defaultObsLimit = 50 maxObsLimit = 500 ) // observability 打包网关的实时管理仪表:请求 // 指标计数器、trace span 环形缓冲、审计存储与 // 进程级日志环形缓冲。它在 New 中构建,且对并发 // 使用安全。 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,而不是空但虚假的尾部。 logBuf atomic.Pointer[logger.RingBuffer] } // newObservability 构建仪表,未注入存储时回退到内存审计 // 存储。 func newObservability(deps Deps) *observability { obs := &observability{ metrics: newMetricsCollector(), traceBuf: newRingBuffer(defaultTraceBufCap), watches: newObsWatches(), } if deps.AuditStore != nil { obs.audit = deps.AuditStore } else { obs.audit = newMemoryAuditStore() } return obs } // ObserveSpan 把一个已完成的 trace span 送入网关的环形缓冲与 // LLM/子代理指标计数器。它是网关的 trace 注入点: // 组合根把 loop tracer 接到这里(例如通过 TraceSink),使 // 管理面板能查询 span 而网关无需触碰 wire.go。 func (s *Server) ObserveSpan(rec trace.SpanRecord) { if s.obs == nil { return } s.obs.traceBuf.Append(rec) s.obs.metrics.Observe(rec) } // TraceSink 返回一个 trace.Writer,把已完成的 span 追加进网关的 // trace 环形缓冲。在组合根把它与既有的 JSONRecorder 结合 // (例如 trace.New(MultiTraceWriter{...})),使 span 既落 stderr 又进 // 管理面板。 func (s *Server) TraceSink() trace.Writer { if s.obs == nil { return nil } return traceSink{s.obs.traceBuf} } // recordMetric 为可观测面板计一次请求。 func (s *Server) recordMetric(method, path string, status int, durMs int64) { if s.obs == nil { return } s.obs.metrics.Record(method, path, status, durMs) } // 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{}, }) return } writeJSON(w, http.StatusOK, s.obs.metrics.Snapshot(time.Now())) } // 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 } spans := s.obs.traceBuf.Snapshot(parseLimit(r)) out := make([]AdminTraceSpan, 0, len(spans)) for _, sp := range spans { out = append(out, adminSpanOf(sp)) } writeJSON(w, http.StatusOK, out) } // adminObservabilityAudit 返回最近的审计条目(管理读)。 func (s *Server) adminObservabilityAudit(w http.ResponseWriter, r *http.Request) { if s.obs == nil || s.obs.audit == nil { writeJSON(w, http.StatusOK, []AuditRecord{}) return } records, err := s.obs.audit.List(r.Context(), parseLimit(r)) if err != nil { s.log.Error("list audit", "err", err) writeError(w, apperr.Wrap(apperr.ErrInternal, "audit_failed", "failed to list audit records")) return } writeJSON(w, http.StatusOK, records) } // parseLimit 解析 ?limit= 查询参数,夹取到 [1, maxObsLimit], // 缺失或非法时选用 defaultObsLimit。 func parseLimit(r *http.Request) int { n, err := strconv.Atoi(r.URL.Query().Get("limit")) if err != nil || n <= 0 { return defaultObsLimit } if n > maxObsLimit { return maxObsLimit } return n }