From 025dd1796ba08db26cbe13067621714eb4d7a2bf Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 15:22:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(observability):=20=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E8=81=9A=E5=90=88=E5=90=8E=E7=AB=AF=20=E2=80=94=20logger?= =?UTF-8?q?=E7=8E=AF=E5=BD=A2=E7=BC=93=E5=86=B2=20+=20GET=20/admin/observa?= =?UTF-8?q?bility/logs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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交叉编译 --- Docs/api-client.md | 11 +- cmd/cyrene/main.go | 6 +- cmd/cyrene/observability.go | 15 ++ cmd/cyrene/observability_test.go | 75 ++++++ internal/logger/logger.go | 6 +- internal/logger/ring.go | 202 +++++++++++++++++ internal/logger/ring_test.go | 184 +++++++++++++++ internal/services/gateway/observability.go | 11 +- .../services/gateway/observability_logs.go | 129 +++++++++++ .../gateway/observability_logs_test.go | 214 ++++++++++++++++++ internal/services/gateway/routes.go | 3 + 11 files changed, 850 insertions(+), 6 deletions(-) create mode 100644 internal/logger/ring.go create mode 100644 internal/logger/ring_test.go create mode 100644 internal/services/gateway/observability_logs.go create mode 100644 internal/services/gateway/observability_logs_test.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 4350f05..e098a86 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -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 内容）到 `/backup-.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/.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` 变真实值。** diff --git a/cmd/cyrene/main.go b/cmd/cyrene/main.go index 568ee65..90121a7 100644 --- a/cmd/cyrene/main.go +++ b/cmd/cyrene/main.go @@ -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) diff --git a/cmd/cyrene/observability.go b/cmd/cyrene/observability.go index 06c06bc..9c03dbe 100644 --- a/cmd/cyrene/observability.go +++ b/cmd/cyrene/observability.go @@ -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 diff --git a/cmd/cyrene/observability_test.go b/cmd/cyrene/observability_test.go index bcbee0c..a4aac02 100644 --- a/cmd/cyrene/observability_test.go +++ b/cmd/cyrene/observability_test.go @@ -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) { diff --git a/internal/logger/logger.go b/internal/logger/logger.go index d737c7c..e7745fa 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -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 diff --git a/internal/logger/ring.go b/internal/logger/ring.go new file mode 100644 index 0000000..24a9aa5 --- /dev/null +++ b/internal/logger/ring.go @@ -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" + } +} diff --git a/internal/logger/ring_test.go b/internal/logger/ring_test.go new file mode 100644 index 0000000..434a595 --- /dev/null +++ b/internal/logger/ring_test.go @@ -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) + } + } +} diff --git a/internal/services/gateway/observability.go b/internal/services/gateway/observability.go index c7993c2..f7d1a2c 100644 --- a/internal/services/gateway/observability.go +++ b/internal/services/gateway/observability.go @@ -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 diff --git a/internal/services/gateway/observability_logs.go b/internal/services/gateway/observability_logs.go new file mode 100644 index 0000000..73b4ebf --- /dev/null +++ b/internal/services/gateway/observability_logs.go @@ -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 +} diff --git a/internal/services/gateway/observability_logs_test.go b/internal/services/gateway/observability_logs_test.go new file mode 100644 index 0000000..9e30089 --- /dev/null +++ b/internal/services/gateway/observability_logs_test.go @@ -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) + } +} diff --git a/internal/services/gateway/routes.go b/internal/services/gateway/routes.go index 2cffefd..590f039 100644 --- a/internal/services/gateway/routes.go +++ b/internal/services/gateway/routes.go @@ -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.