feat(platform): 日志管理 - 统一日志出口并屏蔽查询与 Agent 流式事件日志

新增 internal/logging,slog、标准库 log、GORM 与 Gin 访问日志共用同一 Handler,由 CANVAS_LOG_LEVEL / CANVAS_LOG_FORMAT 控制。info 级别不再输出成功的查询类请求、成功 SQL、message_end 等 Agent 流式事件和常规画布自动保存;4xx/5xx、慢请求、慢 SQL、SQL 失败与 panic 仍会记录,SQL 不内联参数值。
This commit is contained in:
ddcat666 committed 2026-09-30 03:33:34 +08:00
1 parent b7ff958c2d
commit a0f025aa96
21 files changed
+602 -44

No files matched your search

+102
View File
@@ -0,0 +1,102 @@
package logging
import (
"context"
"errors"
"log/slog"
"time"
"gorm.io/gorm"
gormlogger "gorm.io/gorm/logger"
)
// GormLogger 把 GORM 日志接入统一 Handler:成功的 SQL 一律不输出,只记录失败与慢查询。
// SQL 以参数化形式输出(不内联参数值),避免密码哈希、token 等写进日志。
func GormLogger() gormlogger.Interface {
return gormLogger{level: gormlogger.Warn}
}
type gormLogger struct {
level gormlogger.LogLevel
}
func (l gormLogger) LogMode(level gormlogger.LogLevel) gormlogger.Interface {
l.level = level
return l
}
func (l gormLogger) Info(ctx context.Context, msg string, data ...any) {
if l.level >= gormlogger.Info {
slog.InfoContext(ctx, "gorm: "+msg, "data", data)
}
}
func (l gormLogger) Warn(ctx context.Context, msg string, data ...any) {
if l.level >= gormlogger.Warn {
slog.WarnContext(ctx, "gorm: "+msg, "data", data)
}
}
func (l gormLogger) Error(ctx context.Context, msg string, data ...any) {
if l.level >= gormlogger.Error {
slog.ErrorContext(ctx, "gorm: "+msg, "data", data)
}
}
func (l gormLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
if l.level <= gormlogger.Silent {
return
}
elapsed := time.Since(begin)
var level slog.Level
var msg string
switch {
case err != nil && isExpectedQueryError(err):
return
case err != nil:
if l.level < gormlogger.Error {
return
}
level, msg = slog.LevelWarn, "sql failed"
case slowSQLThreshold() > 0 && elapsed >= slowSQLThreshold():
if l.level < gormlogger.Warn {
return
}
level, msg = slog.LevelWarn, "sql slow"
default:
return
}
if ctx == nil {
ctx = context.Background()
}
if !slog.Default().Enabled(ctx, level) {
return
}
sql, rows := fc()
attrs := []slog.Attr{
slog.Int64("duration_ms", elapsed.Milliseconds()),
slog.Int64("rows", rows),
slog.String("sql", truncate(sql, 2000)),
}
if err != nil {
attrs = append(attrs, slog.String("error", err.Error()))
}
slog.LogAttrs(ctx, level, msg, attrs...)
}
// ParamsFilter 让 GORM 在生成日志用的 SQL 时丢弃参数值。
func (gormLogger) ParamsFilter(_ context.Context, sql string, _ ...any) (string, []any) {
return sql, nil
}
// 查不到记录、请求取消属于正常业务分支,由调用方决定是否需要记录。
func isExpectedQueryError(err error) bool {
return errors.Is(err, gorm.ErrRecordNotFound) || errors.Is(err, context.Canceled)
}
func truncate(value string, limit int) string {
if len(value) <= limit {
return value
}
return value[:limit] + "...(truncated)"
}
+111
View File
@@ -0,0 +1,111 @@
package logging
import (
"io"
"log/slog"
"net/http"
"runtime/debug"
"strings"
"time"
"github.com/gin-gonic/gin"
)
// quietWriteSuffixes 是语义上只读、但用 POST 发出的高频路由(心跳、状态查询、批量读取)。
// 与 GET 一样,成功时只在 debug 级别输出。
var quietWriteSuffixes = []string{
"/heartbeat",
"/query",
"/query-provider",
"/query-task",
"/checkout/refresh",
"/assets/batch",
"/announcements/read",
"/diagnostics/preview",
"/system-update/check",
}
// AccessLog 替代 gin.Logger:成功的查询类请求(GET/HEAD/OPTIONS 与 quietWriteSuffixes)降为 debug,
// 写请求记 info,4xx 记 warn,5xx 记 error;超过慢请求阈值的非流式请求无论方法都记 warn。
// redactPath 用于脱敏 URL 中的凭据段(例如分享 token),为 nil 时原样输出。
func AccessLog(redactPath func(string) string) gin.HandlerFunc {
return func(c *gin.Context) {
start := time.Now()
c.Next()
latency := time.Since(start)
status := c.Writer.Status()
route := c.FullPath()
level := accessLevel(c.Request.Method, route, status, latency, isStreamResponse(c))
ctx := c.Request.Context()
if !slog.Default().Enabled(ctx, level) {
return
}
path := c.Request.URL.Path
if redactPath != nil {
path = redactPath(path)
}
attrs := []slog.Attr{
slog.String("method", c.Request.Method),
slog.String("path", path),
slog.Int("status", status),
slog.Int64("latency_ms", latency.Milliseconds()),
slog.String("ip", c.ClientIP()),
}
if requestID := c.Writer.Header().Get("X-Request-ID"); requestID != "" {
attrs = append(attrs, slog.String("request_id", requestID))
}
if message := c.Errors.ByType(gin.ErrorTypePrivate).String(); message != "" {
attrs = append(attrs, slog.String("error", strings.TrimSpace(message)))
}
slog.LogAttrs(ctx, level, "http request", attrs...)
}
}
func accessLevel(method, route string, status int, latency time.Duration, stream bool) slog.Level {
switch {
case status >= http.StatusInternalServerError:
return slog.LevelError
case status >= http.StatusBadRequest:
return slog.LevelWarn
}
if threshold := slowRequestThreshold(); threshold > 0 && !stream && latency >= threshold {
return slog.LevelWarn
}
if isQueryRequest(method, route) {
return slog.LevelDebug
}
return slog.LevelInfo
}
func isQueryRequest(method, route string) bool {
switch method {
case http.MethodGet, http.MethodHead, http.MethodOptions:
return true
case http.MethodPost:
for _, suffix := range quietWriteSuffixes {
if strings.HasSuffix(route, suffix) {
return true
}
}
}
return false
}
// SSE 与分块下载天然是长连接,耗时不代表慢。
func isStreamResponse(c *gin.Context) bool {
return strings.Contains(c.Writer.Header().Get("Content-Type"), "text/event-stream")
}
// Recovery 替代 gin.Recovery:panic 与堆栈进入统一日志,而不是直接写带颜色码的 stderr。
func Recovery() gin.HandlerFunc {
return gin.CustomRecoveryWithWriter(io.Discard, func(c *gin.Context, recovered any) {
slog.ErrorContext(c.Request.Context(), "http panic recovered",
"method", c.Request.Method,
"route", c.FullPath(),
"request_id", c.Writer.Header().Get("X-Request-ID"),
"panic", recovered,
"stack", string(debug.Stack()),
)
c.AbortWithStatus(http.StatusInternalServerError)
})
}
+165
View File
@@ -0,0 +1,165 @@
// Package logging 是后端唯一的日志出口。
//
// slog、标准库 log、GORM 与 Gin 访问日志都经由同一个 slog Handler 输出,格式与级别由环境变量统一控制:
//
// CANVAS_LOG_LEVEL debug | info | warn | error(默认 info)
// CANVAS_LOG_FORMAT text | json(默认 text)
// CANVAS_LOG_SLOW_REQUEST API 慢请求阈值(默认 2s,0 关闭)
// CANVAS_LOG_SLOW_SQL 慢 SQL 阈值(默认 500ms,0 关闭)
//
// 约定:新代码直接使用 log/slog 并选择合适级别;查询、流式事件、轮询等高频路径只能用 Debug。
// 存量 log.Printf 基本都在失败分支,Setup 后统一按 WARN 桥接进同一个 Handler。
package logging
import (
"fmt"
"io"
"log/slog"
"os"
"strings"
"sync"
"time"
)
const (
EnvLevel = "CANVAS_LOG_LEVEL"
EnvFormat = "CANVAS_LOG_FORMAT"
EnvSlowRequest = "CANVAS_LOG_SLOW_REQUEST"
EnvSlowSQL = "CANVAS_LOG_SLOW_SQL"
defaultSlowRequest = 2 * time.Second
defaultSlowSQL = 500 * time.Millisecond
)
type Config struct {
Level slog.Level
JSON bool
SlowRequest time.Duration
SlowSQL time.Duration
Output io.Writer
}
var (
level = new(slog.LevelVar)
settingsMu sync.RWMutex
slowRequest = defaultSlowRequest
slowSQL = defaultSlowSQL
)
// ConfigFromEnv 读取日志环境变量;非法取值直接报错,避免线上悄悄退回到噪声更大的配置。
func ConfigFromEnv() (Config, error) {
config := Config{Level: slog.LevelInfo, SlowRequest: defaultSlowRequest, SlowSQL: defaultSlowSQL}
if raw := strings.TrimSpace(os.Getenv(EnvLevel)); raw != "" {
parsed, err := ParseLevel(raw)
if err != nil {
return Config{}, err
}
config.Level = parsed
}
switch strings.ToLower(strings.TrimSpace(os.Getenv(EnvFormat))) {
case "", "text":
case "json":
config.JSON = true
default:
return Config{}, fmt.Errorf("%s 只能是 text 或 json", EnvFormat)
}
var err error
if config.SlowRequest, err = envThreshold(EnvSlowRequest, defaultSlowRequest); err != nil {
return Config{}, err
}
if config.SlowSQL, err = envThreshold(EnvSlowSQL, defaultSlowSQL); err != nil {
return Config{}, err
}
return config, nil
}
func ParseLevel(raw string) (slog.Level, error) {
switch strings.ToLower(strings.TrimSpace(raw)) {
case "debug":
return slog.LevelDebug, nil
case "info":
return slog.LevelInfo, nil
case "warn", "warning":
return slog.LevelWarn, nil
case "error":
return slog.LevelError, nil
default:
return 0, fmt.Errorf("%s 只能是 debug、info、warn 或 error", EnvLevel)
}
}
func envThreshold(key string, fallback time.Duration) (time.Duration, error) {
raw := strings.TrimSpace(os.Getenv(key))
if raw == "" {
return fallback, nil
}
if raw == "0" {
return 0, nil
}
parsed, err := time.ParseDuration(raw)
if err != nil || parsed < 0 {
return 0, fmt.Errorf("%s 必须是非负时长,例如 2s;0 表示关闭", key)
}
return parsed, nil
}
// Setup 安装全局日志 Handler,必须在打开数据库、创建路由之前调用。
func Setup(config Config) {
output := config.Output
if output == nil {
output = os.Stderr
}
level.Set(config.Level)
settingsMu.Lock()
slowRequest, slowSQL = config.SlowRequest, config.SlowSQL
settingsMu.Unlock()
options := &slog.HandlerOptions{Level: level}
var handler slog.Handler
if config.JSON {
handler = slog.NewJSONHandler(output, options)
} else {
handler = slog.NewTextHandler(output, options)
}
slog.SetDefault(slog.New(handler))
// SetDefault 之后标准库 log 也写入同一个 Handler;存量调用基本都是失败分支,按 WARN 输出。
slog.SetLogLoggerLevel(slog.LevelWarn)
}
// SetLevel 允许运行时调整级别(例如临时打开 debug 排障)。
func SetLevel(value slog.Level) { level.Set(value) }
func DebugEnabled() bool { return level.Level() <= slog.LevelDebug }
func slowRequestThreshold() time.Duration {
settingsMu.RLock()
defer settingsMu.RUnlock()
return slowRequest
}
func slowSQLThreshold() time.Duration {
settingsMu.RLock()
defer settingsMu.RUnlock()
return slowSQL
}
var sampled sync.Map // key -> 上次放行时间
// Every 让同一个 key 在 interval 内只放行一次,用于读路径上会随请求重复出现的告警。
// key 必须来自有限集合(例如模型 ID),不能带请求级别的随机值,否则会无限增长。
func Every(key string, interval time.Duration) bool {
now := time.Now()
for {
previous, loaded := sampled.LoadOrStore(key, now)
if !loaded {
return true
}
if now.Sub(previous.(time.Time)) < interval {
return false
}
if sampled.CompareAndSwap(key, previous, now) {
return true
}
}
}
+134
View File
@@ -0,0 +1,134 @@
package logging
import (
"bytes"
"errors"
"log"
"log/slog"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"time"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
func setupBuffer(t *testing.T, level slog.Level) *bytes.Buffer {
t.Helper()
previous := slog.Default()
buffer := &bytes.Buffer{}
Setup(Config{Level: level, SlowRequest: time.Second, SlowSQL: 100 * time.Millisecond, Output: buffer})
t.Cleanup(func() {
slog.SetDefault(previous)
log.SetOutput(os.Stderr)
log.SetFlags(log.LstdFlags)
})
return buffer
}
func TestConfigFromEnv(t *testing.T) {
t.Setenv(EnvLevel, "WARN")
t.Setenv(EnvFormat, "json")
t.Setenv(EnvSlowRequest, "0")
t.Setenv(EnvSlowSQL, "250ms")
config, err := ConfigFromEnv()
if err != nil {
t.Fatal(err)
}
if config.Level != slog.LevelWarn || !config.JSON || config.SlowRequest != 0 || config.SlowSQL != 250*time.Millisecond {
t.Fatalf("unexpected config: %+v", config)
}
t.Setenv(EnvLevel, "verbose")
if _, err := ConfigFromEnv(); err == nil {
t.Fatal("invalid level must be rejected")
}
}
func TestAccessLogSkipsSuccessfulQueries(t *testing.T) {
gin.SetMode(gin.TestMode)
buffer := setupBuffer(t, slog.LevelInfo)
router := gin.New()
router.Use(AccessLog(func(path string) string { return strings.ReplaceAll(path, "secret", ":token") }))
router.GET("/api/items", func(c *gin.Context) { c.Status(http.StatusOK) })
router.GET("/api/broken", func(c *gin.Context) { c.Status(http.StatusInternalServerError) })
router.POST("/api/items", func(c *gin.Context) { c.Status(http.StatusCreated) })
router.POST("/api/runs/:id/heartbeat", func(c *gin.Context) { c.Status(http.StatusOK) })
router.GET("/api/share/:token", func(c *gin.Context) { c.Status(http.StatusNotFound) })
serve := func(method, path string) {
router.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(method, path, nil))
}
serve(http.MethodGet, "/api/items")
serve(http.MethodPost, "/api/runs/r1/heartbeat")
if buffer.Len() != 0 {
t.Fatalf("successful queries must not be logged at info level: %s", buffer.String())
}
serve(http.MethodPost, "/api/items")
serve(http.MethodGet, "/api/broken")
serve(http.MethodGet, "/api/share/secret")
output := buffer.String()
for _, want := range []string{"level=INFO", "status=201", "level=ERROR", "status=500", "level=WARN", "path=/api/share/:token"} {
if !strings.Contains(output, want) {
t.Fatalf("missing %q in %s", want, output)
}
}
if strings.Contains(output, "secret") {
t.Fatalf("redacted path leaked: %s", output)
}
}
func TestAccessLevelSlowRequest(t *testing.T) {
setupBuffer(t, slog.LevelInfo)
if got := accessLevel(http.MethodGet, "/api/items", 200, 2*time.Second, false); got != slog.LevelWarn {
t.Fatalf("slow query should warn, got %v", got)
}
if got := accessLevel(http.MethodGet, "/api/agent/runs/:id/events", 200, time.Hour, true); got != slog.LevelDebug {
t.Fatalf("long-lived streams are not slow requests, got %v", got)
}
}
func TestStdlogBridgesToWarn(t *testing.T) {
buffer := setupBuffer(t, slog.LevelWarn)
log.Printf("legacy failure: %v", errors.New("boom"))
if !strings.Contains(buffer.String(), "level=WARN") || !strings.Contains(buffer.String(), "legacy failure: boom") {
t.Fatalf("std log should reach the unified handler as WARN: %s", buffer.String())
}
slog.Info("info line")
if strings.Contains(buffer.String(), "info line") {
t.Fatal("info must be filtered at warn level")
}
}
func TestGormLoggerOnlyLogsFailuresAndSlowQueries(t *testing.T) {
buffer := setupBuffer(t, slog.LevelInfo)
logger := GormLogger()
sql := func() (string, int64) { return "SELECT * FROM users WHERE id = $1", 1 }
logger.Trace(t.Context(), time.Now(), sql, nil)
logger.Trace(t.Context(), time.Now(), sql, gorm.ErrRecordNotFound)
if buffer.Len() != 0 {
t.Fatalf("successful or not-found queries must not be logged: %s", buffer.String())
}
logger.Trace(t.Context(), time.Now().Add(-time.Second), sql, nil)
logger.Trace(t.Context(), time.Now(), sql, errors.New("deadlock"))
output := buffer.String()
if !strings.Contains(output, `msg="sql slow"`) || !strings.Contains(output, `msg="sql failed"`) || !strings.Contains(output, "deadlock") {
t.Fatalf("slow and failed SQL must be logged: %s", output)
}
filter, ok := logger.(gorm.ParamsFilter)
if !ok {
t.Fatal("GORM logger must implement ParamsFilter so SQL parameters stay out of logs")
}
if _, vars := filter.ParamsFilter(t.Context(), "SELECT 1", "secret"); vars != nil {
t.Fatal("SQL parameters must be dropped from logs")
}
}
func TestEvery(t *testing.T) {
key := "test-every-" + time.Now().String()
if !Every(key, time.Hour) || Every(key, time.Hour) {
t.Fatal("Every should allow the first call and suppress repeats within the interval")
}
}