Files
Cyrene_v2/internal/services/gateway/observability.go
T
AskaEth 074ffc4ebf
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled
docs(gateway): gateway 剩余文件注释改中文(逐行 1:1,代码零字节改动)
主人 2026-10-03 拍板「注释一律中文」;gateway 至此整包注释中文化完成(本会话共 3 批:
已推库 16 个文件 + 本批剩余文件)。
校验:机械校验(非注释改动行)= 空;剥注释后代码文本逐行比对全等(不一致文件数 0);
全树 go test ./... 58 包 ok / 0 FAIL;交叉编译绿。
2026-10-03 22:28:25 +08:00

133 lines
4.0 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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
// logBuf 是管理日志尾部端点背后的 logger 日志环形缓冲。
// 组合根在网关存在后经 SetLogBuffer 注入它;在那之前
// 该端点返回 501,而不是空但虚假的尾部。
logBuf atomic.Pointer[logger.RingBuffer]
}
// newObservability 构建仪表,未注入存储时回退到内存审计
// 存储。
func newObservability(deps Deps) *observability {
obs := &observability{
metrics: newMetricsCollector(),
traceBuf: newRingBuffer(defaultTraceBufCap),
}
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 返回当前网关指标(管理读)。
func (s *Server) adminObservabilityMetrics(w http.ResponseWriter, r *http.Request) {
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(管理读)。
func (s *Server) adminObservabilityTrace(w http.ResponseWriter, r *http.Request) {
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
}