feat(observability): 日志聚合后端 — logger环形缓冲 + GET /admin/observability/logs
- internal/logger/ring.go: Entry{time,level,source,msg} + RingBuffer(2000,满覆盖最旧,不阻塞)+ringHandler(slog旁路,Enabled委托内层,sourceOf提取source=再回落Scoped); Options.Ring非nil才包
- gateway/observability_logs.go: GET /admin/observability/logs?level=&limit= (writeAuth owner/admin) → {data:[{time(RFC3339 UTC),level,source,msg}]} 最新在前; level阈值过滤(warn=warn+error,已标注取舍); limit缺省200上限1000; 未知level400/未注入501
- routes.go注册 + observability.go SetLogBuffer(atomic.Pointer) + cmd/cyrene main.go/wireGatewayLogBuffer接线(logger先于gateway)
- 控制台/文件输出零变化; 记录只源自已产出日志,不额外打印敏感
- Docs/api-client v0.35; 测试(ring 10 + observability_logs 8 + cmd端到端2); build/vet/test绿(56)+win/darwin交叉编译
This commit is contained in:
1 parent
79589b5c9f
commit
025dd1796b
11 files changed
+850
-6
No files matched your search
+9
-2
@@ -69,6 +69,7 @@
|
||||
| 15 | `/api/v1/admin/observability/metrics` | GET | ✅ | 网关请求指标(总数/QPS/延迟/方法·路径·状态计数/LLM·子代理计数) |
|
||||
| 16 | `/api/v1/admin/observability/trace` | GET | ✅ | 最近 trace span(环形缓冲,默认 512 条;**loop/工具/编排的真实 span**) |
|
||||
| 17 | `/api/v1/admin/observability/audit` | GET | ✅ | 审计记录(who/what/when/result,SQLite 持久化) |
|
||||
| 17b | `/api/v1/admin/observability/logs` | GET | ✅ | **后端日志 tail**(logger 环形缓冲,默认 2000 条;`?level=`/`?limit=`,见下方「可观测性契约」) |
|
||||
| 17a | `/api/v1/admin/notifications` | GET | ✅ | **统一通知 feed**(审计 + 主动推送聚合为只读 feed,design/10 §3 #16;见下方「通知聚合契约」) |
|
||||
| 18 | `/api/v1/admin/memories` | POST | ✅(写) | 手动添加记忆(owner-only,审计);body `{personaId?,content,scope?}`,无 store 回 `501` |
|
||||
| 19 | `/api/v1/admin/rag` | POST | ✅(写) | 导入知识文档(owner-only,审计);body `{name,desc?,content}`,分块后落项目索引可检索;无 store 回 `501` |
|
||||
@@ -77,7 +78,7 @@
|
||||
| 22 | `/api/v1/admin/system/backup` | POST | ✅(写) | 备份数据(owner-only,审计);用 SQLite `VACUUM INTO` 快照(含已提交 WAL 内容)到 `<dataDir>/backup-<ts>.db` → `{path,size}` |
|
||||
| 23 | `/api/v1/admin/system/restore` | POST | ✅(写) | 恢复数据(owner-only,审计);body `{path}` → 校验路径在 dataDir 内→**先备份当前**→替换 DB 文件→`{restored:true,backup,note:"需重启后生效"}`(运行进程持有 DB 句柄,热替换需重启可见) |
|
||||
|
||||
> **M10 前端契约补充(管理面板)**:`GET /api/v1/admin/users` · `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` · `GET /api/v1/admin/rag` · `POST /api/v1/admin/rag/query` · `GET /api/v1/admin/observability/metrics` · `GET /api/v1/admin/observability/trace` · `GET /api/v1/admin/observability/audit` · `GET /api/v1/admin/notifications` 已实现,返回真实数据;`GET /api/v1/admin/logs` 仍🕐未实现(前端按契约封装、后端 404 时显示"待实现"壳,不报错)。**写操作已补齐**(见下方「管理面板写操作契约」):人格/API key/设备控制/自动化/LLM 测试/渠道连接与识别测试/记忆检索/会话导出/插件卸载**真实持久化或真实生效**;**系统设置安全与存储类**(吊销 API key / 轮换 JWT 密钥 / 备份 / 恢复)也已**真实实现**(见下方「系统设置契约」);仍为后置 M10:**新增用户**、**IoT 场景与规则**、**主人档案保存**、**编辑环境变量/重启服务/检查更新**——后端如实返回 `501 not_implemented`,不伪造成功。(**知识库删除已实现**:`POST /admin/rag/delete` 真实删除,见下方写操作契约;**插件安装已实现**:v1 为 MCP server 热安装。)
|
||||
> **M10 前端契约补充(管理面板)**:`GET /api/v1/admin/users` · `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` · `GET /api/v1/admin/rag` · `POST /api/v1/admin/rag/query` · `GET /api/v1/admin/observability/metrics` · `GET /api/v1/admin/observability/trace` · `GET /api/v1/admin/observability/audit` · `GET /api/v1/admin/observability/logs` · `GET /api/v1/admin/notifications` 已实现,返回真实数据(**日志 tail 见下方「可观测性契约」**;未注册 `GET /api/v1/admin/logs`)。**写操作已补齐**(见下方「管理面板写操作契约」):人格/API key/设备控制/自动化/LLM 测试/渠道连接与识别测试/记忆检索/会话导出/插件卸载**真实持久化或真实生效**;**系统设置安全与存储类**(吊销 API key / 轮换 JWT 密钥 / 备份 / 恢复)也已**真实实现**(见下方「系统设置契约」);仍为后置 M10:**新增用户**、**IoT 场景与规则**、**主人档案保存**、**编辑环境变量/重启服务/检查更新**——后端如实返回 `501 not_implemented`,不伪造成功。(**知识库删除已实现**:`POST /admin/rag/delete` 真实删除,见下方写操作契约;**插件安装已实现**:v1 为 MCP server 热安装。)
|
||||
|
||||
**管理面板写操作契约**(`POST/PUT/DELETE /api/v1/admin/*`,均 **owner-only**,读为 owner/admin;写操作经 audit 落 `audit_logs`;成功 `{data}` / 失败 `{error:{code,message}}`):
|
||||
- `POST /admin/personas` body `{id,name,role,systemPrompt?,characterRef?,modes?/mode?,memoryNamespace?}` → `201 {data:{id,name,character,isDefault,mode,memoryNamespace}}`(注册进 persona registry + 持久化到 `configs/personas/<id>.yaml`);id 已存在回 `409 duplicate`。
|
||||
@@ -122,10 +123,15 @@
|
||||
- `POST /admin/rag/query` body `{"query":"..."}` → `{"data":{"query":"...","hits":[{docId,docName,chapter?,text,score,semantic}]}}`。检索走项目索引关键词+行上下文;配置 pgvector/向量增强时并入语义/向量命中(`semantic:true`)。
|
||||
- `POST /admin/rag/delete` body `{"id":"..."}` → `200 {"data":{"id":"...","deleted":true}}`。**真实删除**:从项目索引删除该文档条目并**级联删除其全部章节**(单事务),删除后 `GET /admin/rag` 不再出现、检索不再命中。`id` 缺失/空白回 `400`;文档不存在回 `404 not_found`;未注入 `RagStore` 回 `501`(**owner-only + 审计 `rag.delete`**)。
|
||||
|
||||
**可观测性契约**(`GET /api/v1/admin/observability/*`,均需 owner/admin,`?limit=N` 缺省 50、上限 500):
|
||||
**可观测性契约**(`GET /api/v1/admin/observability/*`,均需 owner/admin;`metrics`/`trace`/`audit` 的 `?limit=N` 缺省 50、上限 500):
|
||||
- `GET /metrics` → `{"data":{"totalRequests","qps","avgLatencyMs","maxLatencyMs","perMethod{}","perPath{}","statusCounts{}","llmCalls","subagentEvents"}}`。`qps` 为近 1 秒窗口计数;`llmCalls`/`subagentEvents` 由实际观测到的 trace span 按 `kind` 累计(Tracer 未注入时为 0)。**Token/成本后端不统计、不返回**(不伪造指标)。
|
||||
- `GET /trace?limit=100` → `{"data":[{id,parentId?,name,kind,status,startedAt,endedAt?,durationNs,props?,events?]}`。`kind` ∈ `llm_call|tool_call|orchestration|state_transition|subagent|memory|guardrail`;`durationNs` 为纳秒;`events` 为 `[{time,type,props?}]`。
|
||||
- `GET /audit?limit=100` → `{"data":[{id,who,what,when,result}]}`。`id` 为整数自增;`when` 为 RFC3339;来源为 SQLite `audit_logs` 表(未注入 AuditStore 时退化为进程内存)。
|
||||
- `GET /logs?level=&limit=` → `{"data":[{time,level,source,msg}]}`(**后端日志 tail**,v0.35)。
|
||||
- **来源**:logger 进程内**有界环形缓冲**(容量 2000,满则丢弃最旧),只存 logger 已产生的记录,**不额外打印、不重渲染、不落库**;控制台/文件输出完全不受影响。`source` 取自 `logger.WithSource` 的 `source=` 属性(`logger.Scoped` 的记录回退为 `module/sub`),无来源记 `""`。
|
||||
- **排序**:`time` **倒序**(最新在前)。`time` 为 **RFC3339(UTC,秒精度)**,与 `trace.startedAt` / `audit.when` 一致;`level` ∈ `debug|info|warn|error`;`msg` 为记录消息。
|
||||
- **查询参数**:`level` ∈ `debug|info|warn|error`(大小写不敏感)为**最低等级过滤**(如 `level=warn` 返回 warn + error),**空/缺省 = 全部**,未知值回 `400 invalid_level`;`limit` 缺省 `200`、上限 `1000`(小于等于 0 或非法值回落缺省)。日志量超过 `limit` 时按时间倒序截断(先按 level 过滤再取 `limit`,缓冲全量 2000 条内扫描)。
|
||||
- **诚实边界**:组合根未注入日志环形缓冲(非二进制主链路启动)时回 **`501 not_implemented`**,不返回伪造/空的成功 tail。日志内容为 logger 已产出内容,**敏感信息按既定规范本就不落日志**(本端点不新增任何打印)。
|
||||
|
||||
**通知聚合契约**(`GET /api/v1/admin/notifications`,**owner/admin 读**;design/10 §3 #16):
|
||||
|
||||
@@ -247,6 +253,7 @@
|
||||
- 语音:上行 `voice_input`(base64 音频或已转写 text),下行 `voice_output`(合成音频 base64 + text)。
|
||||
|
||||
## 5. 变更记录
|
||||
- `v0.35`(**可观测性收尾:后端日志 tail 端点**,对客户端**新增一个 admin 只读端点**):`GET /api/v1/admin/observability/logs?level=&limit=`(owner/admin)→ `200 {data:[{time,level,source,msg}]}`,**最新在前**。数据源为 logger 新增的**有界环形缓冲**(`internal/logger`,容量 2000,满则丢最旧、并发安全、不阻塞、不改动既有控制台/文件输出),组合根在构建 logger 时挂上、在网关就绪后经 `SetLogBuffer` 注入;缓冲未注入回 `501 not_implemented`(不伪造空 tail)。`time` 为 **RFC3339 UTC(秒精度)**,与 `trace.startedAt`/`audit.when` 一致;`level ∈ debug|info|warn|error`(大小写不敏感)为**最低等级过滤**,空 = 全部,未知值 `400 invalid_level`;`limit` 缺省 `200`、上限 `1000`。`source` 取 `logger.WithSource` 的 `source=`(`Scoped` 记录回落 `module/sub`),无来源为空串。详见 §2「可观测性契约」。
|
||||
- `v0.34`(**D3 收尾:消息编辑 / 删除的 WS 实时广播**,对客户端**新增两个下行事件、无 REST 契约变更**):`PATCH /api/v1/messages/{id}` 成功后广播 **`message_edited`**(`{type, sessionId, data:<更新后的完整消息DTO>}`,`thinking` 已剥离)、`DELETE /api/v1/messages/{id}` 成功后广播 **`message_deleted`**(`{type, sessionId, data:{id}}`);两者复用 `message_persisted` 的同一 hub 广播路径(**全连接广播 + 客户端按 `sessionId` 过滤**,hub 无按会话订阅),客户端**按 `id` 原位改 `content` / 按 `id` 移除**,不再需要「改/删后重拉」的绕行。`message_deleted` 的 `sessionId` 由后端在删除前读取消息所属会话填入(`HistoryStore` 新增 `GetMessage`;读不到时客户端按全局唯一的 `id` 兜底)。`404`/`400`/`501` 等失败路径**不广播**。Web 端已消费(`lib/ws.ts` + `store/useAppStore` 新增 `updateMessageById`/`removeMessageById`)。详见 §3.8。
|
||||
- `v0.33`(**A2 记忆异步提取 / A4 可观测性 Tracing 接线 / A6 通知页后端**):① **A6 新增 `GET /api/v1/admin/notifications`**(owner/admin 读,**全新端点**):把**现有源**(审计 `audit_logs` + backthink 主动推送记录)聚合为**统一只读通知 feed**,`type`/`before`/`limit` 过滤与游标见 §2「通知聚合契约」;**不新建表**,未注入的源该类型为空、不伪造;未读游标后置。② **A4 trace 端点由恒空变真实数据**(**无契约变更**,`GET /admin/observability/trace` 与 `metrics` 的 `llmCalls`/`subagentEvents` 从此反映 loop/工具/编排的真实 span):此前 `TraceSink` 从未接线、环形缓冲无人喂;现在组合根把 loop 产生的 span 扇出到网关观测缓冲。响应字段/形状不变。③ **A2 LLM 异步记忆提取接线**(**对客户端无变更**):loop 回合结束后改由 LLM 提取器**后台异步**抽取记忆(不阻塞回复、失败仅记日志),**群聊/陌生人仍只读不写**;无可用模型凭据时回落确定性 `ManualExtractor`(`记住 X` 路径不变)。
|
||||
- `v0.32`(**A1 情绪系统接线:情绪/好感度真跑起来**):后端把既有的 `persona/emotion` 状态机接进 loop 回合边界——每轮结束按该轮用户消息的情绪信号移动**该角色自己**的情绪状态机(沿配置转移表走一步)并调整好感度(按角色 `affection.delta`);无信号轮次衰减一步趋向 `calm`。① **消息 `mood` 不再恒空**:assistant 消息落库带该轮情绪,`message_persisted` / `history` 回放 / `GET /sessions/{id}/messages` 的 `mood` 为真实值(user 消息不带)。② **新增下行 `mood` 事件**(`{data:{mood,personaId,affection}}`,**仅情绪变化时**广播,见 §3.7),前端顶栏「心情 · x」与消息气泡接真数据。③ **隔离**:群聊/陌生人(`MemWritable=false`)情绪只读不写——不改情绪/好感度、不发事件、该轮 `mood` 空。④ **持久性**:情绪/好感度为进程内按角色状态,重启回 `initialState`(未落库;design/01 未要求)。**触发词表与「每轮一步」为 design/01 未规定处的最小实现,待主人确认**。**契约新增(向后兼容):`mood` 事件 + 消息 `mood` 变真实值。**
|
||||
|
||||
+5
-1
@@ -70,7 +70,8 @@ func run() error {
|
||||
if cfg.Env == "production" {
|
||||
format = logger.FormatJSON
|
||||
}
|
||||
log := logger.New(logger.Options{Level: cfg.LogLevel, Format: string(format)})
|
||||
logRing := logger.NewRingBuffer(logger.DefaultRingCapacity)
|
||||
log := logger.New(logger.Options{Level: cfg.LogLevel, Format: string(format), Ring: logRing})
|
||||
log = logger.WithSource(log, "main")
|
||||
|
||||
ctx, stop := platform.SignalContext()
|
||||
@@ -80,6 +81,9 @@ func run() error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Observability log tail: the logger's ring buffer is built above (before the
|
||||
// gateway), so hand it to the gateway now that the gateway exists.
|
||||
wireGatewayLogBuffer(a, logRing)
|
||||
|
||||
if err := a.automation.Start(ctx); err != nil {
|
||||
return fmt.Errorf("start automations: %w", err)
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"os"
|
||||
"sync"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway"
|
||||
)
|
||||
@@ -81,6 +82,20 @@ func (f *traceFanout) WriteSpan(rec trace.SpanRecord) error {
|
||||
return firstErr
|
||||
}
|
||||
|
||||
// wireGatewayLogBuffer hands the composition root's log ring buffer to the
|
||||
// gateway, so GET /api/v1/admin/observability/logs serves the recent backend log
|
||||
// tail. The logger is built before the gateway (its records must be captured
|
||||
// from the first log line), so the buffer is injected here, the first moment the
|
||||
// gateway exists — like wireGatewayTraceSink does for spans. It is a no-op on a
|
||||
// boot without a gateway.
|
||||
func wireGatewayLogBuffer(a *app, buf *logger.RingBuffer) {
|
||||
if a == nil || a.gw == nil || buf == nil {
|
||||
return
|
||||
}
|
||||
a.gw.SetLogBuffer(buf)
|
||||
a.log.Info("observability log buffer wired", "capacity", buf.Len())
|
||||
}
|
||||
|
||||
// gatewaySpanSink adapts *gateway.Server to trace.Writer through its ObserveSpan
|
||||
// entry point, so an emitted span lands in the admin panel's trace ring buffer
|
||||
// AND in the LLM/subagent metric counters (ObserveSpan appends to the buffer and
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway"
|
||||
)
|
||||
@@ -142,6 +143,80 @@ func TestGatewaySpanSink_FeedsTraceEndpointAndMetrics(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestWireGatewayLogBuffer_Guards proves the boot helper tolerates a missing
|
||||
// gateway or buffer, so a degraded boot cannot panic on the log-tail path.
|
||||
func TestWireGatewayLogBuffer_Guards(t *testing.T) {
|
||||
t.Parallel()
|
||||
wireGatewayLogBuffer(nil, nil)
|
||||
wireGatewayLogBuffer(&app{log: testLog()}, logger.NewRingBuffer(4))
|
||||
wireGatewayLogBuffer(&app{log: testLog()}, nil)
|
||||
|
||||
jwt := auth.New(config.AuthConfig{JWTSecret: obsTestSecret})
|
||||
gw, err := gateway.New(&config.Config{}, gateway.Deps{Log: testLog(), Auth: jwt})
|
||||
if err != nil {
|
||||
t.Fatalf("gateway.New: %v", err)
|
||||
}
|
||||
t.Cleanup(gw.Close)
|
||||
wireGatewayLogBuffer(&app{log: testLog(), gw: gw}, nil)
|
||||
}
|
||||
|
||||
// TestWireGatewayLogBuffer_FeedsLogEndpoint is the log-tail acceptance: the
|
||||
// process logger's ring buffer is handed to the gateway by the same helper run()
|
||||
// calls, and the records the app logged surface newest-first on
|
||||
// GET /api/v1/admin/observability/logs.
|
||||
func TestWireGatewayLogBuffer_FeedsLogEndpoint(t *testing.T) {
|
||||
t.Parallel()
|
||||
jwt := auth.New(config.AuthConfig{JWTSecret: obsTestSecret})
|
||||
gw, err := gateway.New(&config.Config{}, gateway.Deps{Log: testLog(), Auth: jwt})
|
||||
if err != nil {
|
||||
t.Fatalf("gateway.New: %v", err)
|
||||
}
|
||||
t.Cleanup(gw.Close)
|
||||
ts := httptest.NewServer(gw.Handler())
|
||||
t.Cleanup(ts.Close)
|
||||
|
||||
// Exactly the boot path: run() builds the logger with the ring buffer before
|
||||
// newApp and hands it to the gateway afterwards.
|
||||
ring := logger.NewRingBuffer(8)
|
||||
log := logger.WithSource(logger.New(logger.Options{
|
||||
Level: "debug", Format: "text", Out: io.Discard, Ring: ring,
|
||||
}), "main")
|
||||
log.Info("boot ok")
|
||||
log.Warn("disk filling")
|
||||
|
||||
a := &app{log: testLog(), gw: gw}
|
||||
wireGatewayLogBuffer(a, ring)
|
||||
|
||||
tok, err := jwt.Sign("acct", auth.RoleOwner, time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("sign token: %v", err)
|
||||
}
|
||||
var body struct {
|
||||
Data []struct {
|
||||
Time string `json:"time"`
|
||||
Level string `json:"level"`
|
||||
Source string `json:"source"`
|
||||
Msg string `json:"msg"`
|
||||
} `json:"data"`
|
||||
}
|
||||
status, raw := obsGet(t, ts.URL+"/api/v1/admin/observability/logs", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("logs status = %d, want 200 (body=%s)", status, raw)
|
||||
}
|
||||
if err := json.Unmarshal(raw, &body); err != nil {
|
||||
t.Fatalf("decode logs: %v (body=%s)", err, raw)
|
||||
}
|
||||
if len(body.Data) != 2 {
|
||||
t.Fatalf("logs = %+v, want the two records logged above", body.Data)
|
||||
}
|
||||
if body.Data[0].Msg != "disk filling" || body.Data[0].Level != "warn" || body.Data[0].Source != "main" {
|
||||
t.Fatalf("logs[0] = %+v, want the newest record (warn/disk filling/main)", body.Data[0])
|
||||
}
|
||||
if body.Data[1].Msg != "boot ok" || body.Data[1].Level != "info" {
|
||||
t.Fatalf("logs[1] = %+v, want the older info record", body.Data[1])
|
||||
}
|
||||
}
|
||||
|
||||
// obsGet performs an authenticated owner GET and returns the status plus the raw
|
||||
// body.
|
||||
func obsGet(t *testing.T, url, token string) (int, []byte) {
|
||||
|
||||
@@ -31,6 +31,10 @@ type Options struct {
|
||||
Format string
|
||||
// Out is the destination writer. Defaults to os.Stderr when nil.
|
||||
Out io.Writer
|
||||
// Ring, when non-nil, additionally captures every emitted record into a
|
||||
// bounded ring buffer served by the admin observability log tail. The
|
||||
// console/file output through Out is unchanged.
|
||||
Ring *RingBuffer
|
||||
}
|
||||
|
||||
// New builds a slog.Logger from opts.
|
||||
@@ -49,7 +53,7 @@ func New(opts Options) *slog.Logger {
|
||||
handler = slog.NewTextHandler(out, handlerOpts)
|
||||
}
|
||||
|
||||
return slog.New(handler)
|
||||
return slog.New(newRingHandler(handler, opts.Ring))
|
||||
}
|
||||
|
||||
// ParseLevel maps a string level name to a slog.Level. Unknown or empty values
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
package logger
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DefaultRingCapacity is how many recent log records the observability ring
|
||||
// buffer retains before the oldest is evicted. Compile-time constant.
|
||||
const DefaultRingCapacity = 2000
|
||||
|
||||
// Entry is one normalized log record as the admin observability log tail serves
|
||||
// it: {time, level, source, msg}. Time is the record's own timestamp; Source is
|
||||
// the provenance tag ("module/subfeature") when the record carried one.
|
||||
type Entry struct {
|
||||
Time time.Time
|
||||
Level string
|
||||
Source string
|
||||
Msg string
|
||||
}
|
||||
|
||||
// RingBuffer is a concurrency-safe bounded buffer of recent log entries. When
|
||||
// full the oldest entry is overwritten, so it never grows past its capacity and
|
||||
// Append never blocks or fails. It is the storage behind
|
||||
// GET /api/v1/admin/observability/logs.
|
||||
type RingBuffer struct {
|
||||
mu sync.RWMutex
|
||||
buf []Entry
|
||||
next int
|
||||
full bool
|
||||
cap int
|
||||
}
|
||||
|
||||
// NewRingBuffer builds a ring buffer holding up to capacity entries, falling
|
||||
// back to DefaultRingCapacity when capacity is not positive.
|
||||
func NewRingBuffer(capacity int) *RingBuffer {
|
||||
if capacity <= 0 {
|
||||
capacity = DefaultRingCapacity
|
||||
}
|
||||
return &RingBuffer{buf: make([]Entry, capacity), cap: capacity}
|
||||
}
|
||||
|
||||
// Append stores e, overwriting the oldest entry when the buffer is full.
|
||||
func (r *RingBuffer) Append(e Entry) {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.buf[r.next] = e
|
||||
r.next = (r.next + 1) % r.cap
|
||||
if r.next == 0 {
|
||||
r.full = true
|
||||
}
|
||||
}
|
||||
|
||||
// Len returns how many entries are currently retained (at most the capacity).
|
||||
func (r *RingBuffer) Len() int {
|
||||
if r == nil {
|
||||
return 0
|
||||
}
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
if r.full {
|
||||
return r.cap
|
||||
}
|
||||
return r.next
|
||||
}
|
||||
|
||||
// Snapshot returns up to limit most-recent entries, newest first. A non-positive
|
||||
// limit returns every retained entry.
|
||||
func (r *RingBuffer) Snapshot(limit int) []Entry {
|
||||
if r == nil {
|
||||
return nil
|
||||
}
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
count := r.next
|
||||
if r.full {
|
||||
count = r.cap
|
||||
}
|
||||
if limit > 0 && limit < count {
|
||||
count = limit
|
||||
}
|
||||
out := make([]Entry, 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
|
||||
}
|
||||
|
||||
// ringHandler wraps another slog.Handler: every record the inner handler emits
|
||||
// is also normalized into buf. The inner handler's own output (console/file) is
|
||||
// untouched, so installing the ring changes nothing downstream.
|
||||
type ringHandler struct {
|
||||
inner slog.Handler
|
||||
buf *RingBuffer
|
||||
attrs []slog.Attr
|
||||
}
|
||||
|
||||
// newRingHandler wraps inner so emitted records are captured into buf. A nil
|
||||
// buffer returns inner unchanged.
|
||||
func newRingHandler(inner slog.Handler, buf *RingBuffer) slog.Handler {
|
||||
if buf == nil {
|
||||
return inner
|
||||
}
|
||||
return &ringHandler{inner: inner, buf: buf}
|
||||
}
|
||||
|
||||
// Enabled delegates to the inner handler: the ring captures exactly the records
|
||||
// the configured level lets through, never more.
|
||||
func (h *ringHandler) Enabled(ctx context.Context, level slog.Level) bool {
|
||||
return h.inner.Enabled(ctx, level)
|
||||
}
|
||||
|
||||
// Handle captures rec into the ring buffer and forwards it to the inner handler.
|
||||
func (h *ringHandler) Handle(ctx context.Context, rec slog.Record) error {
|
||||
h.buf.Append(Entry{
|
||||
Time: rec.Time,
|
||||
Level: levelName(rec.Level),
|
||||
Source: h.sourceOf(rec),
|
||||
Msg: rec.Message,
|
||||
})
|
||||
return h.inner.Handle(ctx, rec)
|
||||
}
|
||||
|
||||
// WithAttrs keeps the captured attributes in sync with the child logger, so a
|
||||
// source tagged with logger.WithSource is visible to Handle.
|
||||
func (h *ringHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
|
||||
if len(attrs) == 0 {
|
||||
return h
|
||||
}
|
||||
return &ringHandler{
|
||||
inner: h.inner.WithAttrs(attrs),
|
||||
buf: h.buf,
|
||||
attrs: append(append([]slog.Attr{}, h.attrs...), attrs...),
|
||||
}
|
||||
}
|
||||
|
||||
// WithGroup keeps the group nesting in sync with the child logger.
|
||||
func (h *ringHandler) WithGroup(name string) slog.Handler {
|
||||
if name == "" {
|
||||
return h
|
||||
}
|
||||
return &ringHandler{inner: h.inner.WithGroup(name), buf: h.buf, attrs: h.attrs}
|
||||
}
|
||||
|
||||
// sourceOf resolves a record's provenance: the "source" attribute tagged by
|
||||
// logger.WithSource when present, else the "module" (+"/"+"sub") pair tagged by
|
||||
// logger.Scoped, else "".
|
||||
func (h *ringHandler) sourceOf(rec slog.Record) string {
|
||||
attrs := make([]slog.Attr, 0, len(h.attrs)+rec.NumAttrs())
|
||||
attrs = append(attrs, h.attrs...)
|
||||
rec.Attrs(func(a slog.Attr) bool {
|
||||
attrs = append(attrs, a)
|
||||
return true
|
||||
})
|
||||
if v, ok := attrValue(attrs, "source"); ok {
|
||||
return v
|
||||
}
|
||||
module, ok := attrValue(attrs, "module")
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
if sub, ok := attrValue(attrs, "sub"); ok && sub != "" {
|
||||
return module + "/" + sub
|
||||
}
|
||||
return module
|
||||
}
|
||||
|
||||
// attrValue returns the value of key in attrs, recursing into groups.
|
||||
func attrValue(attrs []slog.Attr, key string) (string, bool) {
|
||||
for _, a := range attrs {
|
||||
if a.Value.Kind() == slog.KindGroup {
|
||||
if v, ok := attrValue(a.Value.Group(), key); ok {
|
||||
return v, true
|
||||
}
|
||||
continue
|
||||
}
|
||||
if a.Key == key {
|
||||
return a.Value.String(), true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// levelName maps a slog level to the wire name used by the log tail.
|
||||
func levelName(l slog.Level) string {
|
||||
switch {
|
||||
case l >= slog.LevelError:
|
||||
return "error"
|
||||
case l >= slog.LevelWarn:
|
||||
return "warn"
|
||||
case l >= slog.LevelInfo:
|
||||
return "info"
|
||||
default:
|
||||
return "debug"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,184 @@
|
||||
package logger
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestRingBuffer_BoundedDropsOldest(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ring := NewRingBuffer(3)
|
||||
for i := 0; i < 5; i++ {
|
||||
ring.Append(Entry{Time: time.Unix(int64(i), 0), Level: "info", Msg: string(rune('a' + i))})
|
||||
}
|
||||
|
||||
if got := ring.Len(); got != 3 {
|
||||
t.Fatalf("Len() = %d, want 3 (bounded)", got)
|
||||
}
|
||||
got := ring.Snapshot(0)
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("Snapshot len = %d, want 3", len(got))
|
||||
}
|
||||
want := []string{"e", "d", "c"} // newest first, a/b evicted
|
||||
for i, msg := range want {
|
||||
if got[i].Msg != msg {
|
||||
t.Fatalf("Snapshot[%d].Msg = %q, want %q (got %v)", i, got[i].Msg, msg, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRingBuffer_SnapshotLimit(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ring := NewRingBuffer(5)
|
||||
for i := 0; i < 5; i++ {
|
||||
ring.Append(Entry{Level: "info", Msg: string(rune('a' + i))})
|
||||
}
|
||||
|
||||
if got := ring.Snapshot(2); len(got) != 2 || got[0].Msg != "e" || got[1].Msg != "d" {
|
||||
t.Fatalf("Snapshot(2) = %v, want newest two [e d]", got)
|
||||
}
|
||||
if got := ring.Snapshot(-1); len(got) != 5 {
|
||||
t.Fatalf("Snapshot(-1) len = %d, want all 5", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRingBuffer_DefaultCapacityOnNonPositive(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ring := NewRingBuffer(0)
|
||||
if got := cap(ring.buf); got != DefaultRingCapacity {
|
||||
t.Fatalf("capacity = %d, want default %d", got, DefaultRingCapacity)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNew_RingCapturesSourceLevelAndMsg(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var out bytes.Buffer
|
||||
ring := NewRingBuffer(8)
|
||||
log := New(Options{Level: "debug", Format: "text", Out: &out, Ring: ring})
|
||||
WithSource(log, "runtime/loop").Warn("agent stalled", "step", 3)
|
||||
|
||||
entries := ring.Snapshot(0)
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("captured %d entries, want 1", len(entries))
|
||||
}
|
||||
e := entries[0]
|
||||
if e.Level != "warn" {
|
||||
t.Fatalf("level = %q, want warn", e.Level)
|
||||
}
|
||||
if e.Source != "runtime/loop" {
|
||||
t.Fatalf("source = %q, want runtime/loop", e.Source)
|
||||
}
|
||||
if e.Msg != "agent stalled" {
|
||||
t.Fatalf("msg = %q, want %q", e.Msg, "agent stalled")
|
||||
}
|
||||
if e.Time.IsZero() {
|
||||
t.Fatal("time is zero, want the record timestamp")
|
||||
}
|
||||
// The wrapped output is unchanged: the ring is a pure side channel.
|
||||
if !strings.Contains(out.String(), "level=WARN") || !strings.Contains(out.String(), "msg=\"agent stalled\"") {
|
||||
t.Fatalf("inner output changed, got %q", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestNew_RingExtractsScopedSource(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ring := NewRingBuffer(8)
|
||||
log := New(Options{Level: "debug", Format: "text", Out: io.Discard, Ring: ring})
|
||||
Scoped(log, "channel", "adapter").Info("connected")
|
||||
|
||||
entries := ring.Snapshot(0)
|
||||
if len(entries) != 1 || entries[0].Source != "channel/adapter" {
|
||||
t.Fatalf("entries = %v, want one with source channel/adapter", entries)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNew_RingCapturesLevelsNewestFirst(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ring := NewRingBuffer(8)
|
||||
log := New(Options{Level: "debug", Format: "text", Out: io.Discard, Ring: ring})
|
||||
log.Debug("d")
|
||||
log.Info("i")
|
||||
log.Warn("w")
|
||||
log.Error("e")
|
||||
|
||||
entries := ring.Snapshot(0)
|
||||
if len(entries) != 4 {
|
||||
t.Fatalf("captured %d entries, want 4", len(entries))
|
||||
}
|
||||
wantLevels := []string{"error", "warn", "info", "debug"}
|
||||
wantMsgs := []string{"e", "w", "i", "d"}
|
||||
for i := range wantLevels {
|
||||
if entries[i].Level != wantLevels[i] || entries[i].Msg != wantMsgs[i] {
|
||||
t.Fatalf("entries[%d] = %+v, want level=%s msg=%s", i, entries[i], wantLevels[i], wantMsgs[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNew_RingHonorsConfiguredLevel(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ring := NewRingBuffer(8)
|
||||
log := New(Options{Level: "warn", Format: "text", Out: io.Discard, Ring: ring})
|
||||
log.Info("dropped")
|
||||
log.Warn("kept")
|
||||
|
||||
entries := ring.Snapshot(0)
|
||||
if len(entries) != 1 || entries[0].Msg != "kept" {
|
||||
t.Fatalf("entries = %v, want only the warn record", entries)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNew_NilRingIsTransparent(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var out bytes.Buffer
|
||||
log := New(Options{Level: "debug", Format: "text", Out: &out})
|
||||
log.Info("hello")
|
||||
if !strings.Contains(out.String(), "msg=hello") {
|
||||
t.Fatalf("logger without a ring stopped emitting, got %q", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestRingBuffer_NilReceiverIsSafe(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var ring *RingBuffer
|
||||
ring.Append(Entry{Msg: "x"})
|
||||
if got := ring.Snapshot(0); got != nil {
|
||||
t.Fatalf("Snapshot on nil = %v, want nil", got)
|
||||
}
|
||||
if got := ring.Len(); got != 0 {
|
||||
t.Fatalf("Len on nil = %d, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLevelName(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
in slog.Level
|
||||
want string
|
||||
}{
|
||||
{slog.LevelDebug, "debug"},
|
||||
{slog.LevelInfo, "info"},
|
||||
{slog.LevelWarn, "warn"},
|
||||
{slog.LevelError, "error"},
|
||||
{slog.LevelError + 4, "error"},
|
||||
{slog.LevelDebug - 4, "debug"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
if got := levelName(tt.in); got != tt.want {
|
||||
t.Fatalf("levelName(%v) = %q, want %q", tt.in, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3,9 +3,11 @@ 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"
|
||||
)
|
||||
|
||||
@@ -17,12 +19,17 @@ const (
|
||||
)
|
||||
|
||||
// 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.
|
||||
// metrics counter, the trace span ring buffer, the audit store and the
|
||||
// process-wide log ring buffer. It is built in New and is safe for concurrent
|
||||
// use.
|
||||
type observability struct {
|
||||
metrics *MetricsCollector
|
||||
traceBuf *RingBuffer
|
||||
audit AuditStore
|
||||
// logBuf is the logger's log ring buffer behind the admin log-tail endpoint.
|
||||
// The composition root injects it via SetLogBuffer once the gateway exists;
|
||||
// until then the endpoint answers 501 rather than an empty-but-fake tail.
|
||||
logBuf atomic.Pointer[logger.RingBuffer]
|
||||
}
|
||||
|
||||
// newObservability builds the instrumentation, falling back to an in-memory audit
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
)
|
||||
|
||||
// Log-tail endpoint tuning. Compile-time constants (project rule §10).
|
||||
const (
|
||||
defaultLogLimit = 200
|
||||
maxLogLimit = 1000
|
||||
)
|
||||
|
||||
// AdminLogEntry is the wire shape of one entry in
|
||||
// GET /api/v1/admin/observability/logs: {time,level,source,msg}. `time` is
|
||||
// RFC3339 UTC (matching trace/audit); `level` ∈ debug|info|warn|error; `source`
|
||||
// is the record's provenance tag, empty when the record carried none.
|
||||
type AdminLogEntry struct {
|
||||
Time string `json:"time"`
|
||||
Level string `json:"level"`
|
||||
Source string `json:"source"`
|
||||
Msg string `json:"msg"`
|
||||
}
|
||||
|
||||
// SetLogBuffer injects the process-wide log ring buffer behind
|
||||
// GET /api/v1/admin/observability/logs. The composition root calls it once the
|
||||
// gateway exists (the logger is built before the gateway); until then the
|
||||
// endpoint answers 501. It is safe to call concurrently.
|
||||
func (s *Server) SetLogBuffer(buf *logger.RingBuffer) {
|
||||
if s == nil || s.obs == nil {
|
||||
return
|
||||
}
|
||||
s.obs.logBuf.Store(buf)
|
||||
}
|
||||
|
||||
// logBuffer returns the injected log ring buffer, or nil when none was injected.
|
||||
func (s *Server) logBuffer() *logger.RingBuffer {
|
||||
if s == nil || s.obs == nil {
|
||||
return nil
|
||||
}
|
||||
return s.obs.logBuf.Load()
|
||||
}
|
||||
|
||||
// adminObservabilityLogs returns the most recent backend log records, newest
|
||||
// first, optionally filtered to a minimum level (admin read). It is the backend
|
||||
// behind the panel's log tail: the records come from the logger's own ring
|
||||
// buffer and nothing is re-rendered or fabricated.
|
||||
func (s *Server) adminObservabilityLogs(w http.ResponseWriter, r *http.Request) {
|
||||
minLevel, filtered, ok := parseLogLevel(r.URL.Query().Get("level"))
|
||||
if !ok {
|
||||
writeError(w, &apperr.AppError{
|
||||
Code: "invalid_level",
|
||||
Message: "level 只能为 debug/info/warn/error 或留空",
|
||||
HTTPCode: http.StatusBadRequest,
|
||||
})
|
||||
return
|
||||
}
|
||||
buf := s.logBuffer()
|
||||
if buf == nil {
|
||||
writeError(w, &apperr.AppError{
|
||||
Code: "not_implemented",
|
||||
Message: "日志环形缓冲未注入(组合根未挂载 logger.RingBuffer)",
|
||||
HTTPCode: http.StatusNotImplemented,
|
||||
})
|
||||
return
|
||||
}
|
||||
limit := parseLogLimit(r)
|
||||
entries := buf.Snapshot(0)
|
||||
out := make([]AdminLogEntry, 0, min(len(entries), limit))
|
||||
for _, e := range entries {
|
||||
if filtered && logger.ParseLevel(e.Level) < minLevel {
|
||||
continue
|
||||
}
|
||||
out = append(out, adminLogOf(e))
|
||||
if len(out) == limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
// adminLogOf maps a captured log entry to the camelCase admin wire shape.
|
||||
func adminLogOf(e logger.Entry) AdminLogEntry {
|
||||
return AdminLogEntry{
|
||||
Time: e.Time.UTC().Format(time.RFC3339),
|
||||
Level: e.Level,
|
||||
Source: e.Source,
|
||||
Msg: e.Msg,
|
||||
}
|
||||
}
|
||||
|
||||
// parseLogLevel resolves the ?level= filter. An empty value disables filtering
|
||||
// (filtered=false); debug/info/warn/error select that level and above; anything
|
||||
// else is rejected with ok=false so the handler answers 400.
|
||||
func parseLogLevel(raw string) (min slog.Level, filtered, ok bool) {
|
||||
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||||
case "":
|
||||
return 0, false, true
|
||||
case "debug":
|
||||
return slog.LevelDebug, true, true
|
||||
case "info":
|
||||
return slog.LevelInfo, true, true
|
||||
case "warn":
|
||||
return slog.LevelWarn, true, true
|
||||
case "error":
|
||||
return slog.LevelError, true, true
|
||||
default:
|
||||
return 0, false, false
|
||||
}
|
||||
}
|
||||
|
||||
// parseLogLimit parses the ?limit= query param, clamping to [1, maxLogLimit] and
|
||||
// selecting defaultLogLimit when absent or invalid.
|
||||
func parseLogLimit(r *http.Request) int {
|
||||
n, err := strconv.Atoi(r.URL.Query().Get("limit"))
|
||||
if err != nil || n <= 0 {
|
||||
return defaultLogLimit
|
||||
}
|
||||
if n > maxLogLimit {
|
||||
return maxLogLimit
|
||||
}
|
||||
return n
|
||||
}
|
||||
@@ -0,0 +1,214 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
)
|
||||
|
||||
// logEnvelope decodes the {data:[...]} envelope of the log-tail endpoint.
|
||||
type logEnvelope struct {
|
||||
Data []AdminLogEntry `json:"data"`
|
||||
}
|
||||
|
||||
// logsServer builds a gateway with an injected log ring buffer already fed by a
|
||||
// real logger, so the endpoint is exercised against the production capture path.
|
||||
func logsServer(t *testing.T) (*Server, string, string) {
|
||||
t.Helper()
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
|
||||
ring := logger.NewRingBuffer(16)
|
||||
log := logger.New(logger.Options{Level: "debug", Format: "text", Out: io.Discard, Ring: ring})
|
||||
log.Info("boot ok")
|
||||
logger.WithSource(log, "gateway").Warn("slow request")
|
||||
logger.WithSource(log, "runtime/loop").Error("loop crashed")
|
||||
srv.SetLogBuffer(ring)
|
||||
return srv, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
}
|
||||
|
||||
func TestAdminObservabilityLogs_NewestFirstWithFields(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, base, tok := logsServer(t)
|
||||
|
||||
resp, body := doReq(t, http.MethodGet, base+"/api/v1/admin/observability/logs", nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var env logEnvelope
|
||||
if err := json.Unmarshal(body, &env); err != nil {
|
||||
t.Fatalf("decode: %v (body=%s)", err, body)
|
||||
}
|
||||
want := []AdminLogEntry{
|
||||
{Level: "error", Source: "runtime/loop", Msg: "loop crashed"},
|
||||
{Level: "warn", Source: "gateway", Msg: "slow request"},
|
||||
{Level: "info", Source: "", Msg: "boot ok"},
|
||||
}
|
||||
if len(env.Data) != len(want) {
|
||||
t.Fatalf("got %d entries, want %d (%+v)", len(env.Data), len(want), env.Data)
|
||||
}
|
||||
for i, w := range want {
|
||||
got := env.Data[i]
|
||||
if got.Level != w.Level || got.Source != w.Source || got.Msg != w.Msg {
|
||||
t.Fatalf("data[%d] = %+v, want level=%s source=%q msg=%q", i, got, w.Level, w.Source, w.Msg)
|
||||
}
|
||||
if _, err := time.Parse(time.RFC3339, got.Time); err != nil {
|
||||
t.Fatalf("data[%d].Time = %q, want RFC3339: %v", i, got.Time, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminObservabilityLogs_LevelFilter(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, base, tok := logsServer(t)
|
||||
|
||||
cases := []struct {
|
||||
level string
|
||||
want []string
|
||||
}{
|
||||
{"debug", []string{"error", "warn", "info"}},
|
||||
{"info", []string{"error", "warn", "info"}},
|
||||
{"warn", []string{"error", "warn"}},
|
||||
{"error", []string{"error"}},
|
||||
{"ERROR", []string{"error"}}, // case-insensitive
|
||||
{"", []string{"error", "warn", "info"}},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
tc := tc
|
||||
t.Run(tc.level, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
resp, body := doReq(t, http.MethodGet, base+"/api/v1/admin/observability/logs?level="+tc.level, nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var env logEnvelope
|
||||
if err := json.Unmarshal(body, &env); err != nil {
|
||||
t.Fatalf("decode: %v (body=%s)", err, body)
|
||||
}
|
||||
if len(env.Data) != len(tc.want) {
|
||||
t.Fatalf("level=%q got %d entries, want %d (%+v)", tc.level, len(env.Data), len(tc.want), env.Data)
|
||||
}
|
||||
for i, lvl := range tc.want {
|
||||
if env.Data[i].Level != lvl {
|
||||
t.Fatalf("level=%q data[%d].Level = %q, want %q", tc.level, i, env.Data[i].Level, lvl)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminObservabilityLogs_UnknownLevelIs400(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, base, tok := logsServer(t)
|
||||
|
||||
for _, bad := range []string{"trace", "warning", "verbose"} {
|
||||
resp, body := doReq(t, http.MethodGet, base+"/api/v1/admin/observability/logs?level="+bad, nil, tok)
|
||||
if resp.StatusCode != http.StatusBadRequest {
|
||||
t.Fatalf("level=%q status = %d, want 400 (body=%s)", bad, resp.StatusCode, body)
|
||||
}
|
||||
wantErrorShape(t, body)
|
||||
if !strings.Contains(string(body), "invalid_level") {
|
||||
t.Fatalf("level=%q body = %s, want invalid_level code", bad, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminObservabilityLogs_LimitClamped(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, base, tok := logsServer(t)
|
||||
|
||||
resp, body := doReq(t, http.MethodGet, base+"/api/v1/admin/observability/logs?limit=2", nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var env logEnvelope
|
||||
if err := json.Unmarshal(body, &env); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if len(env.Data) != 2 || env.Data[0].Msg != "loop crashed" || env.Data[1].Msg != "slow request" {
|
||||
t.Fatalf("limit=2 data = %+v, want the two newest", env.Data)
|
||||
}
|
||||
|
||||
// An invalid/absent limit falls back to the default (all 3 here < 200).
|
||||
resp, body = doReq(t, http.MethodGet, base+"/api/v1/admin/observability/logs?limit=abc", nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("bad limit status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var env2 logEnvelope
|
||||
if err := json.Unmarshal(body, &env2); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if len(env2.Data) != 3 {
|
||||
t.Fatalf("bad limit returned %d entries, want the default (3)", len(env2.Data))
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseLogLimit_ClampBounds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
mk := func(raw string) int {
|
||||
r, err := http.NewRequest(http.MethodGet, "/?limit="+raw, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("new request: %v", err)
|
||||
}
|
||||
return parseLogLimit(r)
|
||||
}
|
||||
if got := mk("99999"); got != maxLogLimit {
|
||||
t.Fatalf("parseLogLimit(99999) = %d, want max %d", got, maxLogLimit)
|
||||
}
|
||||
if got := mk("0"); got != defaultLogLimit {
|
||||
t.Fatalf("parseLogLimit(0) = %d, want default %d", got, defaultLogLimit)
|
||||
}
|
||||
if got := mk(""); got != defaultLogLimit {
|
||||
t.Fatalf("parseLogLimit(empty) = %d, want default %d", got, defaultLogLimit)
|
||||
}
|
||||
if got := mk("7"); got != 7 {
|
||||
t.Fatalf("parseLogLimit(7) = %d, want 7", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminObservabilityLogs_NoBufferIs501(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/observability/logs", nil, tok)
|
||||
if resp.StatusCode != http.StatusNotImplemented {
|
||||
t.Fatalf("status = %d, want 501 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
wantErrorShape(t, body)
|
||||
if !strings.Contains(string(body), "not_implemented") {
|
||||
t.Fatalf("body = %s, want not_implemented code", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminObservabilityLogs_EmptyBufferIs200Empty(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
|
||||
srv.SetLogBuffer(logger.NewRingBuffer(4))
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/observability/logs", nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
if !strings.Contains(string(body), `"data":[]`) {
|
||||
t.Fatalf("body = %s, want an empty data list", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminObservabilityLogs_ViewerForbidden(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
|
||||
srv.SetLogBuffer(logger.NewRingBuffer(4))
|
||||
tok := signAny(t, jwt, "acct", auth.RoleViewer)
|
||||
|
||||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/observability/logs", nil, tok)
|
||||
if resp.StatusCode != http.StatusForbidden {
|
||||
t.Fatalf("viewer status = %d, want 403 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
}
|
||||
@@ -121,6 +121,9 @@ func (s *Server) registerAdminRoutes(
|
||||
mux.Handle("GET /api/v1/admin/observability/metrics", writeAuth(http.HandlerFunc(s.adminObservabilityMetrics)))
|
||||
mux.Handle("GET /api/v1/admin/observability/trace", writeAuth(http.HandlerFunc(s.adminObservabilityTrace)))
|
||||
mux.Handle("GET /api/v1/admin/observability/audit", writeAuth(http.HandlerFunc(s.adminObservabilityAudit)))
|
||||
// Backend log tail: the logger's own bounded ring buffer, injected by the
|
||||
// composition root (501 until then). Newest first, `?level=`/`?limit=` filtered.
|
||||
mux.Handle("GET /api/v1/admin/observability/logs", writeAuth(http.HandlerFunc(s.adminObservabilityLogs)))
|
||||
// Unified notification feed (design/10 §3 #16): a read-only projection of the
|
||||
// audit trail + backthink pushes, so the panel's notification page needs no
|
||||
// new persisted model.
|
||||
|
||||
Reference in new issue
Block a user