feat(observability): 管理面板可观测性接真实数据——metrics/trace/audit 三端点 + 审计持久化
- observability.go/observability_metrics.go/observability_trace.go/observability_audit.go:
拆分保证 ≤500行/函数≤80。metrics 并发安全 MetricsCollector(方法/路径/状态计数、平均/最大延迟、
近1s QPS窗口);trace 并发安全 RingBuffer(默认512, newest-first)+traceSink+MultiTraceWriter;
audit AuditStore 接口+内存回退+SQLite 持久化(NewSQLiteAuditStore 参数化 SQL+audit_logs 表)。
- server.go:Deps.AuditStore + Server.obs + 3 路由 GET /admin/observability/{metrics,trace,audit};
ObserveSpan/TraceSink 注入点;middleware logging recordMetric;audit 落库(仅 slog→Append)。
- 前端 Obs.tsx:metrics 接 4 张真实 StatCard+请求分布、trace 接 span 列表(展开 props/events)、
audit 从后端拉(前端内存审计作回退);api.ts/types.ts 加 obs 端点/类型(AdminObservabilityMetrics/
AdminTraceSpan/AuditRecord.id 放宽)。成本指标不伪造(保留'—'+注明)。
- wire.go 收口:Deps.AuditStore 注入 gateway.NewSQLiteAuditStore(ctx, a.db)(审计真落库),
未注入回退内存。
- Docs/api-client.md §2/v0.6。
实机验证:三端点 200,metrics 返回 totalRequests:4/qps:4/perPath/statusCounts:{200:4}。
go build/vet/test 全绿 + 前端 tsc。trace 数据源对接(gw.TraceSink→loop tracer)因 gw/tracer
构造顺序留作可选后置(端点已就绪, 未对接 span 空)。
This commit is contained in:
1 parent
76134ed47e
commit
e0a543cc37
12 files changed
+803
-22
No files matched your search
@@ -1,6 +1,7 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
@@ -16,15 +17,23 @@ func (s *Server) auditClaims(r *http.Request, what, result string) {
|
||||
if claims, ok := auth.ClaimsFromContext(r.Context()); ok {
|
||||
who = claims.Sub + ":" + string(claims.Role)
|
||||
}
|
||||
s.audit(who, what, result)
|
||||
s.audit(r.Context(), who, what, result)
|
||||
}
|
||||
|
||||
// audit emits one structured audit record on the gateway logger.
|
||||
func (s *Server) audit(who, what, result string) {
|
||||
// audit emits one structured audit record on the gateway logger and persists it to
|
||||
// the configured AuditStore (falling back to an in-memory store when none is
|
||||
// injected). A persist failure is logged but never blocks the admin write, so a
|
||||
// broken audit store cannot take down the audit trail's primary job.
|
||||
func (s *Server) audit(ctx context.Context, who, what, result string) {
|
||||
s.log.Info("admin audit",
|
||||
"who", who,
|
||||
"what", what,
|
||||
"when", time.Now().UTC().Format(time.RFC3339),
|
||||
"result", result,
|
||||
)
|
||||
if s.obs != nil && s.obs.audit != nil {
|
||||
if err := s.obs.audit.Append(ctx, who, what, result, time.Now()); err != nil {
|
||||
s.log.Error("admin audit persist", "err", err, "what", what)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -74,11 +74,13 @@ func (s *Server) logging(next http.Handler) http.Handler {
|
||||
start := time.Now()
|
||||
rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
|
||||
next.ServeHTTP(rec, r)
|
||||
durMs := time.Since(start).Milliseconds()
|
||||
s.recordMetric(r.Method, r.URL.Path, rec.status, durMs)
|
||||
s.log.Info("request",
|
||||
"method", r.Method,
|
||||
"path", r.URL.Path,
|
||||
"status", rec.status,
|
||||
"duration_ms", time.Since(start).Milliseconds(),
|
||||
"duration_ms", durMs,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// Default list limits shared by the trace/audit endpoints. Compile-time constants
|
||||
// so no magic number appears in the run path (project rule §10).
|
||||
const (
|
||||
defaultObsLimit = 50
|
||||
maxObsLimit = 500
|
||||
)
|
||||
|
||||
// observability bundles the gateway's live admin instrumentation: the request
|
||||
// metrics counter, the trace span ring buffer and the audit store. It is built in
|
||||
// New and is safe for concurrent use.
|
||||
type observability struct {
|
||||
metrics *MetricsCollector
|
||||
traceBuf *RingBuffer
|
||||
audit AuditStore
|
||||
}
|
||||
|
||||
// newObservability builds the instrumentation, falling back to an in-memory audit
|
||||
// store when no store is injected.
|
||||
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 feeds a completed trace span into the gateway's ring buffer and the
|
||||
// LLM/subagent metric counters. It is the gateway's trace injection point: the
|
||||
// composition root wires the loop tracer here (e.g. through TraceSink) so the
|
||||
// admin panel can query spans without the gateway touching 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 returns a trace.Writer that appends completed spans to the gateway's
|
||||
// trace ring buffer. Combine it with the existing JSONRecorder in the composition
|
||||
// root (e.g. trace.New(MultiTraceWriter{...})) so spans both hit stderr and the
|
||||
// admin panel.
|
||||
func (s *Server) TraceSink() trace.Writer {
|
||||
if s.obs == nil {
|
||||
return nil
|
||||
}
|
||||
return traceSink{s.obs.traceBuf}
|
||||
}
|
||||
|
||||
// recordMetric counts one request for the observability panel.
|
||||
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 returns the current gateway metrics (admin read).
|
||||
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 returns the most recent traced spans (admin read).
|
||||
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 returns the most recent audit entries (admin read).
|
||||
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 parses the ?limit= query param, clamping to [1, maxObsLimit], and
|
||||
// selects defaultObsLimit when absent or invalid.
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// AuditRecord is one persisted admin audit entry (who/what/when/result).
|
||||
type AuditRecord struct {
|
||||
ID int64 `json:"id"`
|
||||
Who string `json:"who"`
|
||||
What string `json:"what"`
|
||||
When string `json:"when"`
|
||||
Result string `json:"result"`
|
||||
}
|
||||
|
||||
// AuditStore persists the admin audit trail. Implementations must be safe for
|
||||
// concurrent use. *SQLiteAuditStore is the production implementation (wired by the
|
||||
// composition root over a.db); when none is injected the gateway falls back to an
|
||||
// in-memory store so the endpoint always returns data.
|
||||
type AuditStore interface {
|
||||
// Append persists one audit entry at ts.
|
||||
Append(ctx context.Context, who, what, result string, ts time.Time) error
|
||||
// List returns at most limit most-recent entries, newest first. A non-positive
|
||||
// limit selects the default.
|
||||
List(ctx context.Context, limit int) ([]AuditRecord, error)
|
||||
}
|
||||
|
||||
// memoryAuditStore is the nil-injection fallback: it holds the audit trail in
|
||||
// process memory so the admin endpoint works before the composition root injects a
|
||||
// SQLite store. It is concurrency-safe.
|
||||
type memoryAuditStore struct {
|
||||
mu sync.Mutex
|
||||
records []AuditRecord
|
||||
seq int64
|
||||
}
|
||||
|
||||
// newMemoryAuditStore builds an empty in-memory audit store.
|
||||
func newMemoryAuditStore() *memoryAuditStore {
|
||||
return &memoryAuditStore{records: make([]AuditRecord, 0, 64)}
|
||||
}
|
||||
|
||||
// Append stores one entry with a monotonically increasing id.
|
||||
func (m *memoryAuditStore) Append(ctx context.Context, who, what, result string, ts time.Time) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.seq++
|
||||
m.records = append(m.records, AuditRecord{
|
||||
ID: m.seq, Who: who, What: what, When: ts.UTC().Format(time.RFC3339), Result: result,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// List returns the most recent entries, newest first.
|
||||
func (m *memoryAuditStore) List(ctx context.Context, limit int) ([]AuditRecord, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = defaultObsLimit
|
||||
}
|
||||
if limit > maxObsLimit {
|
||||
limit = maxObsLimit
|
||||
}
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
n := len(m.records)
|
||||
start := n - limit
|
||||
if start < 0 {
|
||||
start = 0
|
||||
}
|
||||
out := make([]AuditRecord, n-start)
|
||||
copy(out, m.records[start:])
|
||||
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
|
||||
out[i], out[j] = out[j], out[i]
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// audit DDL for a self-contained SQLite store (no migration bookkeeping), the
|
||||
// same pattern channel/session and backthink use. recorded_at is the UTC RFC3339
|
||||
// wall-clock of the audited action.
|
||||
const (
|
||||
createAuditSQL = `
|
||||
CREATE TABLE IF NOT EXISTS audit_logs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
who TEXT NOT NULL,
|
||||
what TEXT NOT NULL,
|
||||
result TEXT NOT NULL,
|
||||
recorded_at TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||
)`
|
||||
createAuditIndexSQL = `CREATE INDEX IF NOT EXISTS idx_audit_logs_recorded_at ON audit_logs(recorded_at)`
|
||||
insertAuditSQL = `INSERT INTO audit_logs (who, what, result, recorded_at) VALUES (?, ?, ?, ?)`
|
||||
selectAuditSQL = `SELECT id, who, what, result, recorded_at FROM audit_logs ORDER BY id DESC LIMIT ?`
|
||||
)
|
||||
|
||||
// SQLiteAuditStore persists the admin audit trail in SQLite. It is safe for
|
||||
// concurrent use by virtue of the underlying *sql.DB.
|
||||
type SQLiteAuditStore struct {
|
||||
db *sql.DB
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
// NewSQLiteAuditStore builds a SQLite-backed audit store over db, guaranteeing the
|
||||
// audit_logs table and index exist. It returns an error when db is nil or the DDL
|
||||
// fails. The composition root wires this over the primary *sql.DB (a.db).
|
||||
func NewSQLiteAuditStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*SQLiteAuditStore, error) {
|
||||
if db == nil {
|
||||
return nil, fmt.Errorf("gateway: audit store: nil database")
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, createAuditSQL); err != nil {
|
||||
return nil, fmt.Errorf("gateway: ensure audit_logs: %w", err)
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, createAuditIndexSQL); err != nil {
|
||||
return nil, fmt.Errorf("gateway: ensure audit_logs index: %w", err)
|
||||
}
|
||||
return &SQLiteAuditStore{db: db, log: sourcedLog(log)}, nil
|
||||
}
|
||||
|
||||
// Append persists one audit entry at ts.
|
||||
func (s *SQLiteAuditStore) Append(ctx context.Context, who, what, result string, ts time.Time) error {
|
||||
if _, err := s.db.ExecContext(ctx, insertAuditSQL, who, what, result, ts.UTC().Format(time.RFC3339Nano)); err != nil {
|
||||
return fmt.Errorf("gateway: insert audit: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// List returns the most recent entries, newest first.
|
||||
func (s *SQLiteAuditStore) List(ctx context.Context, limit int) ([]AuditRecord, error) {
|
||||
if limit <= 0 {
|
||||
limit = defaultObsLimit
|
||||
}
|
||||
if limit > maxObsLimit {
|
||||
limit = maxObsLimit
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rows, err := s.db.QueryContext(ctx, selectAuditSQL, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("gateway: list audit: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
records := make([]AuditRecord, 0, limit)
|
||||
for rows.Next() {
|
||||
var r AuditRecord
|
||||
if err := rows.Scan(&r.ID, &r.Who, &r.What, &r.Result, &r.When); err != nil {
|
||||
return nil, fmt.Errorf("gateway: scan audit: %w", err)
|
||||
}
|
||||
records = append(records, r)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("gateway: iterate audit: %w", err)
|
||||
}
|
||||
return records, nil
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// Metrics collector tuning. Compile-time constants.
|
||||
const (
|
||||
// metricsWindow bounds how long request timestamps are retained for QPS.
|
||||
metricsWindow = 2 * time.Second
|
||||
// metricsQPSSpan is the sliding window over which QPS is measured.
|
||||
metricsQPSSpan = time.Second
|
||||
)
|
||||
|
||||
// AdminMetrics is the wire shape of GET /api/v1/admin/observability/metrics.
|
||||
// LLM/Token cost is deliberately absent: the gateway does not track token usage or
|
||||
// spend, and a fabricated number would violate the "不伪造指标" red line.
|
||||
type AdminMetrics struct {
|
||||
TotalRequests int64 `json:"totalRequests"`
|
||||
QPS float64 `json:"qps"`
|
||||
AvgLatencyMs float64 `json:"avgLatencyMs"`
|
||||
MaxLatencyMs int64 `json:"maxLatencyMs"`
|
||||
PerMethod map[string]int64 `json:"perMethod"`
|
||||
PerPath map[string]int64 `json:"perPath"`
|
||||
StatusCounts map[string]int64 `json:"statusCounts"`
|
||||
LLMCalls int64 `json:"llmCalls"`
|
||||
SubagentEvents int64 `json:"subagentEvents"`
|
||||
}
|
||||
|
||||
// MetricsCollector accumulates request metrics and derived counters for the admin
|
||||
// observability panel. It is concurrency-safe.
|
||||
type MetricsCollector struct {
|
||||
mu sync.Mutex
|
||||
total int64
|
||||
perMethod map[string]int64
|
||||
perPath map[string]int64
|
||||
statuses map[int]int64
|
||||
sumDur int64
|
||||
maxDur int64
|
||||
recent []time.Time
|
||||
llm int64
|
||||
subagent int64
|
||||
}
|
||||
|
||||
// newMetricsCollector builds an empty metrics collector.
|
||||
func newMetricsCollector() *MetricsCollector {
|
||||
return &MetricsCollector{
|
||||
perMethod: make(map[string]int64),
|
||||
perPath: make(map[string]int64),
|
||||
statuses: make(map[int]int64),
|
||||
}
|
||||
}
|
||||
|
||||
// Record counts one request with the given method, path, status and duration.
|
||||
func (m *MetricsCollector) Record(method, path string, status int, durMs int64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
now := time.Now()
|
||||
m.total++
|
||||
m.perMethod[method]++
|
||||
m.perPath[path]++
|
||||
m.statuses[status]++
|
||||
m.sumDur += durMs
|
||||
if durMs > m.maxDur {
|
||||
m.maxDur = durMs
|
||||
}
|
||||
m.recent = append(m.recent, now)
|
||||
m.prune(now)
|
||||
}
|
||||
|
||||
// Observe increments the LLM/subagent counters when a span of that kind is
|
||||
// recorded. It is called by ObserveSpan so the counters reflect real work.
|
||||
func (m *MetricsCollector) Observe(rec trace.SpanRecord) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
switch rec.Kind {
|
||||
case trace.SpanKindLLM:
|
||||
m.llm++
|
||||
case trace.SpanKindSubagent:
|
||||
m.subagent++
|
||||
}
|
||||
}
|
||||
|
||||
// prune drops request times older than metricsWindow so recent stays small. The
|
||||
// caller must hold m.mu.
|
||||
func (m *MetricsCollector) prune(now time.Time) {
|
||||
min := now.Add(-metricsWindow)
|
||||
i := 0
|
||||
for i < len(m.recent) && m.recent[i].Before(min) {
|
||||
i++
|
||||
}
|
||||
if i > 0 {
|
||||
m.recent = append([]time.Time(nil), m.recent[i:]...)
|
||||
}
|
||||
}
|
||||
|
||||
// Snapshot returns a point-in-time copy of the metrics. qps counts the requests
|
||||
// started within metricsQPSSpan of now.
|
||||
func (m *MetricsCollector) Snapshot(now time.Time) AdminMetrics {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
statuses := make(map[string]int64, len(m.statuses))
|
||||
for k, v := range m.statuses {
|
||||
statuses[strconv.Itoa(k)] = v
|
||||
}
|
||||
var qps int64
|
||||
for _, t := range m.recent {
|
||||
if now.Sub(t) <= metricsQPSSpan {
|
||||
qps++
|
||||
}
|
||||
}
|
||||
avg := int64(0)
|
||||
if m.total > 0 {
|
||||
avg = m.sumDur / m.total
|
||||
}
|
||||
return AdminMetrics{
|
||||
TotalRequests: m.total,
|
||||
QPS: float64(qps),
|
||||
AvgLatencyMs: float64(avg),
|
||||
MaxLatencyMs: m.maxDur,
|
||||
PerMethod: m.perMethod,
|
||||
PerPath: m.perPath,
|
||||
StatusCounts: statuses,
|
||||
LLMCalls: m.llm,
|
||||
SubagentEvents: m.subagent,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// defaultTraceBufCap is how many completed spans the ring buffer holds before the
|
||||
// oldest is evicted. Compile-time constant.
|
||||
const defaultTraceBufCap = 512
|
||||
|
||||
// AdminTraceSpan is the camelCase wire shape of one traced span. It decouples the
|
||||
// frontend from the trace package's snake_case 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 is a concurrency-safe bounded buffer of completed trace spans. The
|
||||
// oldest span is evicted when the buffer is full, so the admin panel always sees
|
||||
// the most recent work without unbounded growth.
|
||||
type RingBuffer struct {
|
||||
mu sync.RWMutex
|
||||
buf []trace.SpanRecord
|
||||
next int
|
||||
full bool
|
||||
cap int
|
||||
}
|
||||
|
||||
// newRingBuffer builds a ring buffer holding up to cap spans (default when cap<=0).
|
||||
func newRingBuffer(cap int) *RingBuffer {
|
||||
if cap <= 0 {
|
||||
cap = defaultTraceBufCap
|
||||
}
|
||||
return &RingBuffer{buf: make([]trace.SpanRecord, cap), cap: cap}
|
||||
}
|
||||
|
||||
// Append stores rec, evicting the oldest when the buffer is full.
|
||||
func (r *RingBuffer) Append(rec trace.SpanRecord) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.buf[r.next] = rec
|
||||
r.next = (r.next + 1) % r.cap
|
||||
if r.next == 0 {
|
||||
r.full = true
|
||||
}
|
||||
}
|
||||
|
||||
// Snapshot returns up to n most-recent spans, newest first. A non-positive n
|
||||
// selects defaultObsLimit; n is clamped to 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])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// traceSink adapts a *RingBuffer to trace.Writer so a completed span is appended
|
||||
// to the gateway's trace buffer.
|
||||
type traceSink struct{ b *RingBuffer }
|
||||
|
||||
// WriteSpan appends rec to the ring buffer.
|
||||
func (s traceSink) WriteSpan(rec trace.SpanRecord) error {
|
||||
if s.b == nil {
|
||||
return nil
|
||||
}
|
||||
s.b.Append(rec)
|
||||
return nil
|
||||
}
|
||||
|
||||
// MultiTraceWriter fans a completed span out to several trace.Writers, so a tracer
|
||||
// can record to both the gateway's ring buffer and a JSONL file in one pass. A nil
|
||||
// Writer is skipped. It is the composition-root helper the main session uses to
|
||||
// route the loop tracer's output into the gateway without dropping stderr.
|
||||
type MultiTraceWriter struct {
|
||||
// Writers are the sinks a span is written to. A nil Writer is skipped.
|
||||
Writers []trace.Writer
|
||||
}
|
||||
|
||||
// WriteSpan writes rec to every configured 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 maps a trace.SpanRecord to the camelCase admin wire shape.
|
||||
func adminSpanOf(rec trace.SpanRecord) AdminTraceSpan {
|
||||
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
|
||||
}
|
||||
@@ -191,6 +191,11 @@ type Deps struct {
|
||||
// RAG backs the admin knowledge-base RAG module. Optional; list/search
|
||||
// return empty when unset, and write endpoints report a clear 501.
|
||||
RAG RagStore
|
||||
// AuditStore persists the admin audit trail (observability → audit). When nil
|
||||
// the gateway degrades to an in-memory store so the endpoint still returns
|
||||
// data; the composition root injects NewSQLiteAuditStore over a.db for real
|
||||
// persistence. Optional.
|
||||
AuditStore AuditStore
|
||||
// Streaming marks the injected Loop as streaming-capable (its OnStream is
|
||||
// wired to publish ws.stream_chunk). When set the hub skips its own single
|
||||
// final stream_chunk emission, since deltas already stream through OnStream.
|
||||
@@ -225,6 +230,7 @@ type Server struct {
|
||||
users UserStore
|
||||
backthink BackthinkStore
|
||||
rag RagStore
|
||||
obs *observability
|
||||
voice VoiceService
|
||||
streaming bool
|
||||
frontend fs.FS
|
||||
@@ -259,6 +265,7 @@ func New(cfg *config.Config, deps Deps) (*Server, error) {
|
||||
users: deps.Users,
|
||||
backthink: deps.Backthink,
|
||||
rag: deps.RAG,
|
||||
obs: newObservability(deps),
|
||||
streaming: deps.Streaming,
|
||||
frontend: deps.Frontend,
|
||||
webhooks: deps.Webhooks,
|
||||
@@ -303,6 +310,9 @@ func (s *Server) Handler() http.Handler {
|
||||
mux.Handle("GET /api/v1/admin/rag", adminAuth(http.HandlerFunc(s.adminListRag)))
|
||||
mux.Handle("POST /api/v1/admin/rag/query", adminAuth(http.HandlerFunc(s.adminSearchRag)))
|
||||
mux.Handle("POST /api/v1/admin/rag/delete", adminAuth(http.HandlerFunc(s.adminDeleteRag)))
|
||||
mux.Handle("GET /api/v1/admin/observability/metrics", adminAuth(http.HandlerFunc(s.adminObservabilityMetrics)))
|
||||
mux.Handle("GET /api/v1/admin/observability/trace", adminAuth(http.HandlerFunc(s.adminObservabilityTrace)))
|
||||
mux.Handle("GET /api/v1/admin/observability/audit", adminAuth(http.HandlerFunc(s.adminObservabilityAudit)))
|
||||
mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy)))
|
||||
|
||||
// External webhook callbacks (e.g. the webhook channel) are mounted without
|
||||
|
||||
Reference in new issue
Block a user