commit bed0dbf076692b6ab17937f5f2d55ca94ac5f719 Author: weichen Date: Tue Sep 29 13:51:16 2026 +0000 Add a structured logger and collector so runs can be traced across processes. Sync and async sinks, redaction, and a queryable HTTP service keep concurrent logs tied to a trace id. Co-authored-by: Cursor diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..a084d4c --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,20 @@ +name: ci +on: + push: + branches: [main] + pull_request: + branches: [main] +jobs: + vet-and-test: + runs-on: nightly + steps: + - uses: actions/checkout@v4 + - run: go vet ./... && go test -count=1 ./... + docker-checkout-hostgateway: + runs-on: ubuntu-latest + options: --add-host gitea:host-gateway + env: + GITHUB_SERVER_URL: http://gitea:3000 + steps: + - uses: actions/checkout@v4 + - run: test -f go.mod && echo "docker-mode checkout via host-gateway OK" diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..31d2457 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +*.test +/bin/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..ef664ed --- /dev/null +++ b/README.md @@ -0,0 +1,83 @@ +# Logging-System + +跨系统共用的结构化日志库，以及可独立部署的采集服务。目标是把一次请求的运行日志和错误码对齐到同一个 `trace_id` 上，方便事后追溯。 + +本仓库是该模块的源码位置。Go 模块路径为 `logging-system`。只依赖标准库，不依赖错误码模块：只要错误实现了 `ErrorCode()` 和 `ErrorFields()`，错误日志会自动带上码和字段。 + +## 分层 + +| 包 | 职责 | 依赖边界 | +|---|---|---| +| logsys | 级别、条目、Logger、上下文追踪、全局默认日志器 | 只依赖标准库 | +| sink | console / 滚动文件 / 异步队列 / HTTP 批量 / 多路分发 | 只依赖 logsys | +| redact | 写前脱敏（JWT、sk-key、Bearer、api_key、password，可追加正则） | 只依赖标准库 | +| config | 默认值 ← JSON ← `LOGSYS_*` 环境变量 | 组装 logsys 与 sink | +| slogadapter | 与 `log/slog` 双向转发 | 只依赖 logsys | +| server | 采集服务，默认 `:8092` | 组合根之一 | +| cli | `logctl` | 组合根之一 | +| conformance | 并发不丢、异步排空、脱敏、多路一致 | 只用公开 API | + +## 同步与异步 + +每个出口自己保证并发安全。 + +| 模式 | 行为 | +|---|---| +| 同步 | 调用方 goroutine 在互斥锁内写完（console、file，或 `async: false` 的 HTTP） | +| 异步 | 有界队列加单写者。队列满时丢弃**最新**一条并计数，不阻塞业务 goroutine。`Sync` / `Close` 排空队列，关闭有超时 | + +HTTP 出口默认异步攒批（条数或时间间隔）。文件可按大小滚动并保留 N 份历史。 + +模块级别可以比全局更细，例如全局 `info`、`engine` 为 `debug`、`server` 为 `warn`。出口还可以再设自己的最低级别。 + +## 配置 + +完整示例见 [docs/example.json](docs/example.json)。 + +环境变量覆盖文件：`LOGSYS_CONFIG`、`LOGSYS_LEVEL`、`LOGSYS_SERVICE`、`LOGSYS_REDACT`（on/off）。 + +```go +cfg, err := config.Load("log.json") +log, err := config.Build(cfg) +defer log.Close() +log.Module("engine").WithContext(ctx).Error("provider down", err, "step", id) +``` + +`log.Error` 的 `err` 若带有错误码，条目里会出现 `code` 和该错误的字段。追踪标识用 `logsys.WithTrace` 放进 context，或由调用方写入 `X-Trace-Id`。 + +## 采集 API + +| 方法/路径 | 说明 | +|---|---| +| `GET /v1/health` | 健康检查 | +| `POST /v1/logs` | 正文 `{"entries":[...]}`。配置了 token 时要求 `Authorization: Bearer` 或 `X-Log-Token` | +| `GET /v1/logs` | `trace_id`、`level`（精确）、`service`、`q`、`cursor`、`limit` | +| `GET /v1/logs/tail` | SSE。`cursor` 之后先回放再实时推送；慢消费者丢帧 | + +写入时再脱敏一次，避免客户端绕过进程内红线。 + +## CLI + +```bash +go build -o bin/logctl . +./bin/logctl serve --addr :8092 --data logs/collector.jsonl --token dev +./bin/logctl query --addr http://127.0.0.1:8092 --trace trace-abc-1 --token dev +./bin/logctl tail --addr http://127.0.0.1:8092 --token dev +``` + +## 开发 + +```bash +go vet ./... +go test -count=1 ./... +``` + +## 依赖本仓库 + +``` +require logging-system v0.0.0 + +replace logging-system => ../Logging-System +``` + +远程：`ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Logging-System.git` diff --git a/cli/cli.go b/cli/cli.go new file mode 100644 index 0000000..7ca88e5 --- /dev/null +++ b/cli/cli.go @@ -0,0 +1,174 @@ +// Package cli 实现 logctl：serve / tail / query。 +package cli + +import ( + "bufio" + "bytes" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + + "logging-system/server" +) + +// Main 执行命令行并返回退出码。 +func Main(args []string) int { return Run(args, os.Stdout, os.Stderr) } + +// Run 执行命令。 +func Run(args []string, stdout, stderr io.Writer) int { + if len(args) == 0 { + fmt.Fprintln(stderr, usage) + return 2 + } + switch args[0] { + case "serve": + return cmdServe(args[1:], stderr) + case "query": + return cmdQuery(args[1:], stdout, stderr) + case "tail": + return cmdTail(args[1:], stdout, stderr) + case "-h", "--help", "help": + fmt.Fprintln(stdout, usage) + return 0 + default: + fmt.Fprintf(stderr, "未知命令 %q\n%s\n", args[0], usage) + return 2 + } +} + +const usage = `用法: + logctl serve [--addr :8092] [--data 文件] [--token 密钥] + logctl query --addr URL [--trace ID] [--level 级别] [--service 名] [--q 关键词] [--token 密钥] + logctl tail --addr URL [--trace ID] [--token 密钥]` + +func cmdServe(args []string, stderr io.Writer) int { + fs := flag.NewFlagSet("serve", flag.ContinueOnError) + fs.SetOutput(stderr) + addr := fs.String("addr", server.DefaultAddr, "监听地址") + data := fs.String("data", "", "JSONL 落盘路径，空则只放内存") + token := fs.String("token", "", "可选写入令牌") + if err := fs.Parse(args); err != nil { + return 2 + } + srv, err := server.New(*data, *token, 0) + if err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + defer srv.Close() + fmt.Fprintf(stderr, "logctl serve %s\n", *addr) + if err := srv.ListenAndServe(*addr); err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + return 0 +} + +func cmdQuery(args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("query", flag.ContinueOnError) + fs.SetOutput(stderr) + addr := fs.String("addr", "http://127.0.0.1:8092", "采集器地址") + traceID := fs.String("trace", "", "追踪标识") + level := fs.String("level", "", "级别（精确匹配）") + service := fs.String("service", "", "服务名") + q := fs.String("q", "", "关键词") + token := fs.String("token", "", "令牌") + if err := fs.Parse(args); err != nil { + return 2 + } + u, err := url.Parse(strings.TrimRight(*addr, "/") + "/v1/logs") + if err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + params := u.Query() + set := func(k, v string) { + if v != "" { + params.Set(k, v) + } + } + set("trace_id", *traceID) + set("level", *level) + set("service", *service) + set("q", *q) + u.RawQuery = params.Encode() + req, err := http.NewRequest(http.MethodGet, u.String(), nil) + if err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + if *token != "" { + req.Header.Set("Authorization", "Bearer "+*token) + } + res, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + defer res.Body.Close() + body, _ := io.ReadAll(res.Body) + if res.StatusCode != http.StatusOK { + fmt.Fprintf(stderr, "query %s: %s\n", res.Status, body) + return 1 + } + var pretty bytes.Buffer + if err := json.Indent(&pretty, body, "", " "); err != nil { + stdout.Write(body) + return 0 + } + fmt.Fprintln(stdout, pretty.String()) + return 0 +} + +func cmdTail(args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("tail", flag.ContinueOnError) + fs.SetOutput(stderr) + addr := fs.String("addr", "http://127.0.0.1:8092", "采集器地址") + traceID := fs.String("trace", "", "只打印该追踪标识") + token := fs.String("token", "", "令牌") + if err := fs.Parse(args); err != nil { + return 2 + } + u := strings.TrimRight(*addr, "/") + "/v1/logs/tail" + req, err := http.NewRequest(http.MethodGet, u, nil) + if err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + if *token != "" { + req.Header.Set("Authorization", "Bearer "+*token) + } + res, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + body, _ := io.ReadAll(res.Body) + fmt.Fprintf(stderr, "tail %s: %s\n", res.Status, body) + return 1 + } + sc := bufio.NewScanner(res.Body) + for sc.Scan() { + line := sc.Text() + if !strings.HasPrefix(line, "data: ") { + continue + } + payload := strings.TrimPrefix(line, "data: ") + if *traceID != "" && !strings.Contains(payload, *traceID) { + continue + } + fmt.Fprintln(stdout, payload) + } + if err := sc.Err(); err != nil { + fmt.Fprintln(stderr, err) + return 1 + } + return 0 +} diff --git a/cli/cli_test.go b/cli/cli_test.go new file mode 100644 index 0000000..9c262f4 --- /dev/null +++ b/cli/cli_test.go @@ -0,0 +1,33 @@ +package cli + +import ( + "bytes" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "logging-system/server" +) + +func TestQueryCommand(t *testing.T) { + srv, err := server.New("", "", 100) + if err != nil { + t.Fatal(err) + } + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + res, err := http.Post(ts.URL+"/v1/logs", "application/json", strings.NewReader(`{"entries":[{"level":"info","msg":"hello-cli","service":"demo"}]}`)) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + var out, errBuf bytes.Buffer + code := Run([]string{"query", "--addr", ts.URL, "--service", "demo"}, &out, &errBuf) + if code != 0 { + t.Fatal(errBuf.String()) + } + if !strings.Contains(out.String(), "hello-cli") { + t.Fatalf("query output: %s", out.String()) + } +} diff --git a/config/config.go b/config/config.go new file mode 100644 index 0000000..684f665 --- /dev/null +++ b/config/config.go @@ -0,0 +1,216 @@ +// Package config 加载日志配置：默认值，再被 JSON 文件覆盖，再被环境变量覆盖。 +package config + +import ( + "encoding/json" + "fmt" + "os" + "strings" + "time" + + "logging-system/logsys" + "logging-system/redact" + "logging-system/sink" +) + +// Config 是日志系统的全部可配置项。 +type Config struct { + Service string `json:"service"` + Level string `json:"level"` + Modules map[string]string `json:"modules"` + Sinks []SinkConfig `json:"sinks"` + Redaction RedactionConfig `json:"redaction"` + Trace TraceConfig `json:"trace"` +} + +// SinkConfig 描述一个出口。Async 为空时：http 默认异步，其余默认同步。 +type SinkConfig struct { + Type string `json:"type"` + Format string `json:"format"` + Level string `json:"level"` + Path string `json:"path"` + URL string `json:"url"` + Token string `json:"token"` + Async *bool `json:"async"` + Queue int `json:"queue"` + Rotation *RotationConfig `json:"rotation"` + Batch *BatchConfig `json:"batch"` +} + +// RotationConfig 控制文件滚动。 +type RotationConfig struct { + MaxBytes int64 `json:"max_bytes"` + Keep int `json:"keep"` +} + +// BatchConfig 控制 HTTP 批量发送。 +type BatchConfig struct { + MaxEntries int `json:"max_entries"` + FlushIntervalMs int `json:"flush_interval_ms"` +} + +// RedactionConfig 控制写前脱敏。 +type RedactionConfig struct { + Enabled bool `json:"enabled"` + ExtraPatterns []string `json:"extra_patterns"` +} + +// TraceConfig 控制追踪标识。生成动作由调用方的 HTTP 中间件执行。 +type TraceConfig struct { + Generate bool `json:"generate"` + Header string `json:"header"` +} + +// Defaults 返回可用的默认配置：info 级别、同步 stderr、开启脱敏。 +func Defaults() Config { + return Config{ + Level: "info", + Redaction: RedactionConfig{Enabled: true}, + Trace: TraceConfig{Generate: true, Header: "X-Trace-Id"}, + Sinks: []SinkConfig{{ + Type: "console", + Format: "human", + Level: "info", + }}, + } +} + +// Load 读取 JSON 文件并应用环境变量。path 为空时只使用默认值与环境变量。 +// 若 path 为空且设置了 LOGSYS_CONFIG，则读取该路径。 +func Load(path string) (Config, error) { + cfg := Defaults() + if path == "" { + path = os.Getenv("LOGSYS_CONFIG") + } + if path != "" { + data, err := os.ReadFile(path) + if err != nil { + return Config{}, err + } + if err := json.Unmarshal(data, &cfg); err != nil { + return Config{}, fmt.Errorf("config: %w", err) + } + if cfg.Level == "" { + cfg.Level = "info" + } + if cfg.Trace.Header == "" { + cfg.Trace.Header = "X-Trace-Id" + } + } + applyEnv(&cfg) + return cfg, nil +} + +func applyEnv(cfg *Config) { + if v := strings.TrimSpace(os.Getenv("LOGSYS_LEVEL")); v != "" { + cfg.Level = v + } + if v := strings.TrimSpace(os.Getenv("LOGSYS_SERVICE")); v != "" { + cfg.Service = v + } + switch strings.ToLower(strings.TrimSpace(os.Getenv("LOGSYS_REDACT"))) { + case "1", "true", "yes", "on": + cfg.Redaction.Enabled = true + case "0", "false", "no", "off": + cfg.Redaction.Enabled = false + } +} + +// Build 按配置组装日志器。未知出口类型或非法级别会失败。 +func Build(cfg Config) (*logsys.Logger, error) { + level, err := logsys.ParseLevel(cfg.Level) + if err != nil { + return nil, err + } + modules := make(map[string]logsys.Level, len(cfg.Modules)) + for name, raw := range cfg.Modules { + lv, err := logsys.ParseLevel(raw) + if err != nil { + return nil, fmt.Errorf("config: module %s: %w", name, err) + } + modules[name] = lv + } + var redactFn func(string) string + if cfg.Redaction.Enabled { + rd, err := redact.New(cfg.Redaction.ExtraPatterns) + if err != nil { + return nil, err + } + redactFn = rd.Apply + } + sinks := make([]logsys.Sink, 0, len(cfg.Sinks)) + for i, sc := range cfg.Sinks { + built, err := buildSink(sc) + if err != nil { + return nil, fmt.Errorf("config: sinks[%d]: %w", i, err) + } + sinks = append(sinks, built) + } + return logsys.New(logsys.Options{ + Service: cfg.Service, + Level: level, + ModuleLevels: modules, + Sinks: sinks, + Redact: redactFn, + }), nil +} + +func buildSink(sc SinkConfig) (logsys.Sink, error) { + async := false + if sc.Type == "http" && sc.Async == nil { + async = true + } + if sc.Async != nil { + async = *sc.Async + } + var inner logsys.Sink + switch sc.Type { + case "console", "": + format := sc.Format + if format == "" { + format = "human" + } + inner = &sink.Console{Format: format} + case "file": + format := sc.Format + if format == "" { + format = "json" + } + f := &sink.File{Path: sc.Path, Format: format} + if sc.Rotation != nil { + f.MaxBytes = sc.Rotation.MaxBytes + f.Keep = sc.Rotation.Keep + } + inner = f + case "http": + h := &sink.HTTPBatch{URL: sc.URL, Token: sc.Token, Async: async} + if sc.Batch != nil { + h.MaxEntries = sc.Batch.MaxEntries + if sc.Batch.FlushIntervalMs > 0 { + h.FlushInterval = ms(sc.Batch.FlushIntervalMs) + } + } + if !async { + h.Async = false + } + inner = h + async = false // 批量器自己有后台循环，不再套一层队列 + default: + return nil, fmt.Errorf("unknown type %q", sc.Type) + } + if sc.Level != "" { + lv, err := logsys.ParseLevel(sc.Level) + if err != nil { + return nil, err + } + inner = sink.MinLevel{Min: lv, Next: inner} + } + if async { + wrapped := &sink.Async{Inner: inner, Queue: sc.Queue} + wrapped.Start() + inner = wrapped + } + return inner, nil +} + +func ms(v int) time.Duration { return time.Duration(v) * time.Millisecond } diff --git a/config/config_test.go b/config/config_test.go new file mode 100644 index 0000000..6f59cad --- /dev/null +++ b/config/config_test.go @@ -0,0 +1,82 @@ +package config + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "logging-system/logsys" + "logging-system/sink" +) + +func TestLoadEnvOverridesFile(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "log.json") + raw := []byte(`{ + "service": "from-file", + "level": "debug", + "modules": {"engine": "trace"}, + "redaction": {"enabled": true}, + "sinks": [{"type": "console", "format": "json", "level": "warn"}] + }`) + if err := os.WriteFile(path, raw, 0o644); err != nil { + t.Fatal(err) + } + t.Setenv("LOGSYS_LEVEL", "error") + t.Setenv("LOGSYS_SERVICE", "from-env") + t.Setenv("LOGSYS_REDACT", "off") + cfg, err := Load(path) + if err != nil { + t.Fatal(err) + } + if cfg.Level != "error" || cfg.Service != "from-env" || cfg.Redaction.Enabled { + t.Fatalf("%+v", cfg) + } + if cfg.Modules["engine"] != "trace" || cfg.Trace.Header != "X-Trace-Id" { + t.Fatalf("%+v", cfg) + } + log, err := Build(cfg) + if err != nil { + t.Fatal(err) + } + mem := &sink.Memory{} + log2 := logsys.New(logsys.Options{Service: cfg.Service, Level: logsys.LevelError, Sinks: []logsys.Sink{mem}}) + log2.Info("skip") + log2.Error("keep", nil) + if len(mem.Snapshot()) != 1 { + t.Fatal(mem.Snapshot()) + } + if err := log.Close(); err != nil { + t.Fatal(err) + } +} + +func TestBuildFileSink(t *testing.T) { + dir := t.TempDir() + cfg := Defaults() + cfg.Sinks = []SinkConfig{{ + Type: "file", + Path: filepath.Join(dir, "app.jsonl"), + Format: "json", + Rotation: &RotationConfig{MaxBytes: 1024, Keep: 2}, + }} + log, err := Build(cfg) + if err != nil { + t.Fatal(err) + } + log.Info("hello", "token", "sk-abcdefghijklmnopqrstuvwxyz") + if err := log.Sync(); err != nil { + t.Fatal(err) + } + if err := log.Close(); err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(filepath.Join(dir, "app.jsonl")) + if err != nil { + t.Fatal(err) + } + if string(data) == "" || strings.Contains(string(data), "sk-abcdefghijklmnopqrstuvwxyz") { + t.Fatalf("redaction failed: %s", data) + } +} diff --git a/conformance/conformance_test.go b/conformance/conformance_test.go new file mode 100644 index 0000000..202320d --- /dev/null +++ b/conformance/conformance_test.go @@ -0,0 +1,110 @@ +package conformance + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "logging-system/logsys" + "logging-system/redact" + "logging-system/sink" +) + +func TestSyncConcurrentNoLoss(t *testing.T) { + mem := &sink.Memory{} + log := logsys.New(logsys.Options{Level: logsys.LevelInfo, Sinks: []logsys.Sink{mem}}) + const goroutines = 100 + const per = 1000 + var wg sync.WaitGroup + wg.Add(goroutines) + for g := 0; g < goroutines; g++ { + go func() { + defer wg.Done() + for i := 0; i < per; i++ { + log.Info("sync") + } + }() + } + wg.Wait() + if err := log.Sync(); err != nil { + t.Fatal(err) + } + if got := len(mem.Snapshot()); got != goroutines*per { + t.Fatalf("got %d want %d", got, goroutines*per) + } +} + +func TestAsyncCloseDrains(t *testing.T) { + mem := &sink.Memory{} + async := &sink.Async{Inner: mem, Queue: 4096} + log := logsys.New(logsys.Options{Level: logsys.LevelInfo, Sinks: []logsys.Sink{async}}) + const n = 500 + for i := 0; i < n; i++ { + log.Info("async") + } + if err := log.Close(); err != nil { + t.Fatal(err) + } + if got := len(mem.Snapshot()); got != n { + t.Fatalf("drained %d want %d dropped %d", got, n, async.Dropped()) + } + if async.Dropped() != 0 { + t.Fatalf("dropped %d", async.Dropped()) + } +} + +func TestRedactDoesNotLeak(t *testing.T) { + rd, err := redact.New(nil) + if err != nil { + t.Fatal(err) + } + mem := &sink.Memory{} + log := logsys.New(logsys.Options{ + Level: logsys.LevelInfo, + Sinks: []logsys.Sink{mem}, + Redact: rd.Apply, + }) + log.Error("call", nil, "header", "Bearer abcdefghijklmnop", "key", "sk-abcdefghijklmnopqrstuvwxyz") + raw := mem.Snapshot()[0] + blob := raw.Message + raw.Fields["header"].(string) + raw.Fields["key"].(string) + for _, secret := range []string{"abcdefghijklmnop", "sk-abcdefghijklmnopqrstuvwxyz"} { + if stringsContains(blob, secret) { + t.Fatalf("leaked %s in %+v", secret, raw.Fields) + } + } +} + +func TestFanoutConsistent(t *testing.T) { + a, b := &sink.Memory{}, &sink.Memory{} + log := logsys.New(logsys.Options{Level: logsys.LevelInfo, Sinks: []logsys.Sink{sink.Fanout{Sinks: []logsys.Sink{a, b}}}}) + var seq atomic.Int64 + var wg sync.WaitGroup + wg.Add(20) + for i := 0; i < 20; i++ { + go func() { + defer wg.Done() + for j := 0; j < 50; j++ { + log.Info("fan", "n", seq.Add(1)) + } + }() + } + wg.Wait() + if len(a.Snapshot()) != 1000 || len(b.Snapshot()) != 1000 { + t.Fatalf("a=%d b=%d", len(a.Snapshot()), len(b.Snapshot())) + } + _ = time.Second +} + +func stringsContains(s, sub string) bool { + return len(s) >= len(sub) && (s == sub || len(sub) > 0 && index(s, sub)) +} + +func index(s, sub string) bool { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return true + } + } + return false +} diff --git a/docs/example.json b/docs/example.json new file mode 100644 index 0000000..c81617a --- /dev/null +++ b/docs/example.json @@ -0,0 +1,22 @@ +{ + "service": "agent-squad", + "level": "info", + "modules": { "engine": "debug", "server": "warn" }, + "sinks": [ + { "type": "console", "format": "human", "level": "info" }, + { + "type": "file", + "path": "logs/app.jsonl", + "format": "json", + "rotation": { "max_bytes": 104857600, "keep": 10 } + }, + { + "type": "http", + "url": "http://127.0.0.1:8092/v1/logs", + "async": true, + "batch": { "max_entries": 200, "flush_interval_ms": 2000 } + } + ], + "redaction": { "enabled": true, "extra_patterns": [] }, + "trace": { "generate": true, "header": "X-Trace-Id" } +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..ec04037 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module logging-system + +go 1.27 diff --git a/logsys/context.go b/logsys/context.go new file mode 100644 index 0000000..b2a6011 --- /dev/null +++ b/logsys/context.go @@ -0,0 +1,39 @@ +package logsys + +import "context" + +type traceKey struct{} + +// WithTrace 把追踪标识放进上下文。空标识不写入。 +func WithTrace(ctx context.Context, id string) context.Context { + if ctx == nil { + ctx = context.Background() + } + if id == "" { + return ctx + } + return context.WithValue(ctx, traceKey{}, id) +} + +// TraceFrom 读取追踪标识。不存在时返回空字符串。 +func TraceFrom(ctx context.Context) string { + if ctx == nil { + return "" + } + id, _ := ctx.Value(traceKey{}).(string) + return id +} + +// WithContext 返回带有上下文追踪标识的子 logger。上下文没有标识时原样返回副本。 +func (l *Logger) WithContext(ctx context.Context) *Logger { + if l == nil { + return Nop() + } + id := TraceFrom(ctx) + if id == "" { + return l.clone() + } + n := l.clone() + n.traceID = id + return n +} diff --git a/logsys/entry.go b/logsys/entry.go new file mode 100644 index 0000000..af6ae48 --- /dev/null +++ b/logsys/entry.go @@ -0,0 +1,31 @@ +package logsys + +import "time" + +// Entry 是一条已结构化的日志。JSON 字段名跨进程稳定。 +type Entry struct { + Time time.Time `json:"ts"` + Level string `json:"level"` + Service string `json:"service,omitempty"` + Module string `json:"module,omitempty"` + Message string `json:"msg"` + TraceID string `json:"trace_id,omitempty"` + Error string `json:"error,omitempty"` + Code string `json:"code,omitempty"` + Fields map[string]any `json:"fields,omitempty"` +} + +// Sink 是日志出口。实现必须允许多个 goroutine 同时 Write。 +// Sync 把缓冲刷出；Close 之后不得再 Write。 +type Sink interface { + Write(e Entry) error + Sync() error + Close() error +} + +// codedError 是错误码模块的鸭子接口。实现这两个方法即可被展开，无需互相引用。 +type codedError interface { + error + ErrorCode() string + ErrorFields() map[string]any +} diff --git a/logsys/format.go b/logsys/format.go new file mode 100644 index 0000000..86ef2d4 --- /dev/null +++ b/logsys/format.go @@ -0,0 +1,61 @@ +package logsys + +import ( + "bytes" + "encoding/json" + "fmt" + "sort" + "strings" + "time" +) + +// FormatJSON 把条目编码为一行 JSON（含换行）。 +func FormatJSON(e Entry) ([]byte, error) { + buf := &bytes.Buffer{} + enc := json.NewEncoder(buf) + enc.SetEscapeHTML(false) + if err := enc.Encode(e); err != nil { + return nil, err + } + return buf.Bytes(), nil +} + +// FormatHuman 把条目格式化成单行文本（含换行）。 +func FormatHuman(e Entry) string { + ts := e.Time.UTC().Format(time.RFC3339Nano) + var b strings.Builder + fmt.Fprintf(&b, "%s %s", ts, strings.ToUpper(e.Level)) + if e.Service != "" || e.Module != "" { + b.WriteByte(' ') + b.WriteString(e.Service) + if e.Module != "" { + if e.Service != "" { + b.WriteByte('/') + } + b.WriteString(e.Module) + } + } + b.WriteByte(' ') + b.WriteString(e.Message) + if e.Code != "" { + fmt.Fprintf(&b, " code=%s", e.Code) + } + if e.TraceID != "" { + fmt.Fprintf(&b, " trace=%s", e.TraceID) + } + if e.Error != "" { + fmt.Fprintf(&b, " error=%q", e.Error) + } + if len(e.Fields) > 0 { + keys := make([]string, 0, len(e.Fields)) + for k := range e.Fields { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + fmt.Fprintf(&b, " %s=%v", k, e.Fields[k]) + } + } + b.WriteByte('\n') + return b.String() +} diff --git a/logsys/level.go b/logsys/level.go new file mode 100644 index 0000000..a68ab7c --- /dev/null +++ b/logsys/level.go @@ -0,0 +1,65 @@ +// Package logsys 是进程内结构化日志核心。 +// +// 它不绑定具体出口：实现 Sink 即可接到控制台、文件或远端采集器。 +// 错误码通过鸭子接口识别（ErrorCode / ErrorFields），本包不导入错误码模块。 +package logsys + +import ( + "fmt" + "strings" +) + +// Level 是日志级别。数值越大越严重。 +type Level int + +const ( + LevelTrace Level = iota + LevelDebug + LevelInfo + LevelWarn + LevelError + LevelFatal +) + +// ParseLevel 解析级别文本。 +func ParseLevel(s string) (Level, error) { + switch strings.ToLower(strings.TrimSpace(s)) { + case "trace": + return LevelTrace, nil + case "debug": + return LevelDebug, nil + case "info", "": + if strings.TrimSpace(s) == "" { + return 0, fmt.Errorf("logsys: empty level") + } + return LevelInfo, nil + case "warn", "warning": + return LevelWarn, nil + case "error": + return LevelError, nil + case "fatal": + return LevelFatal, nil + default: + return 0, fmt.Errorf("logsys: unknown level %q", s) + } +} + +// String 返回级别名。 +func (l Level) String() string { + switch l { + case LevelTrace: + return "trace" + case LevelDebug: + return "debug" + case LevelInfo: + return "info" + case LevelWarn: + return "warn" + case LevelError: + return "error" + case LevelFatal: + return "fatal" + default: + return "unknown" + } +} diff --git a/logsys/logger.go b/logsys/logger.go new file mode 100644 index 0000000..5a7ea08 --- /dev/null +++ b/logsys/logger.go @@ -0,0 +1,342 @@ +package logsys + +import ( + "fmt" + "sync" + "time" +) + +// Options 构造一个 Logger。构造完成后不要修改切片和映射。 +type Options struct { + Service string + Module string + Level Level + ModuleLevels map[string]Level + Sinks []Sink + Redact func(string) string + Fields map[string]any + Now func() time.Time +} + +// Logger 是并发安全的日志器。With / Module 返回副本，不改变原对象。 +type Logger struct { + service string + module string + min Level + moduleLevels map[string]Level + fields map[string]any + traceID string + sinks []Sink + redact func(string) string + now func() time.Time + nop bool +} + +// New 创建日志器。没有 sink 时写入被丢弃（与 Nop 相同，但保留级别与字段）。 +func New(opt Options) *Logger { + now := opt.Now + if now == nil { + now = func() time.Time { return time.Now().UTC() } + } + levels := make(map[string]Level, len(opt.ModuleLevels)) + for k, v := range opt.ModuleLevels { + levels[k] = v + } + return &Logger{ + service: opt.Service, + module: opt.Module, + min: opt.Level, + moduleLevels: levels, + fields: cloneFields(opt.Fields), + sinks: append([]Sink(nil), opt.Sinks...), + redact: opt.Redact, + now: now, + } +} + +// Nop 丢弃全部日志。 +func Nop() *Logger { return &Logger{nop: true, now: time.Now} } + +// Module 返回绑定模块名的子 logger，用于匹配模块级别。 +func (l *Logger) Module(name string) *Logger { + if l == nil { + return Nop() + } + n := l.clone() + n.module = name + return n +} + +// With 追加字段。键必须是非空字符串，且与值成对；否则 panic。 +func (l *Logger) With(kv ...any) *Logger { + if l == nil { + return Nop() + } + n := l.clone() + n.fields = mergeFields(n.fields, fieldsFrom(kv)) + return n +} + +// WithTraceID 固定本 logger 的追踪标识。 +func (l *Logger) WithTraceID(id string) *Logger { + if l == nil { + return Nop() + } + n := l.clone() + n.traceID = id + return n +} + +// Enabled 报告该级别是否会写出去。 +func (l *Logger) Enabled(level Level) bool { + if l == nil || l.nop { + return false + } + return l.enabled(level) +} + +// Trace 记录 trace 级别日志。 +func (l *Logger) Trace(msg string, kv ...any) { l.log(LevelTrace, msg, nil, kv...) } + +// Debug 记录 debug 级别日志。 +func (l *Logger) Debug(msg string, kv ...any) { l.log(LevelDebug, msg, nil, kv...) } + +// Info 记录 info 级别日志。 +func (l *Logger) Info(msg string, kv ...any) { l.log(LevelInfo, msg, nil, kv...) } + +// Warn 记录 warn 级别日志。 +func (l *Logger) Warn(msg string, kv ...any) { l.log(LevelWarn, msg, nil, kv...) } + +// Error 记录 error 级别日志。err 可为 nil；若实现 ErrorCode/ErrorFields，会展开进条目。 +func (l *Logger) Error(msg string, err error, kv ...any) { l.log(LevelError, msg, err, kv...) } + +// Fatal 记录 fatal 级别日志，不退出进程。 +func (l *Logger) Fatal(msg string, err error, kv ...any) { l.log(LevelFatal, msg, err, kv...) } + +// Sync 刷出全部 sink。 +func (l *Logger) Sync() error { + if l == nil || l.nop { + return nil + } + var errs []error + for _, s := range l.sinks { + if s == nil { + continue + } + if err := s.Sync(); err != nil { + errs = append(errs, err) + } + } + return joinErrs(errs) +} + +// Close 关闭全部 sink。 +func (l *Logger) Close() error { + if l == nil || l.nop { + return nil + } + var errs []error + for _, s := range l.sinks { + if s == nil { + continue + } + if err := s.Close(); err != nil { + errs = append(errs, err) + } + } + return joinErrs(errs) +} + +func (l *Logger) log(level Level, msg string, err error, kv ...any) { + extra := fieldsFrom(kv) + if l == nil || l.nop || !l.enabled(level) { + return + } + fields := mergeFields(l.fields, extra) + entry := Entry{ + Time: l.now(), + Level: level.String(), + Service: l.service, + Module: l.module, + Message: msg, + TraceID: l.traceID, + Fields: fields, + } + applyError(&entry, err) + if l.redact != nil { + entry = redactEntry(entry, l.redact) + } + for _, s := range l.sinks { + if s == nil { + continue + } + _ = s.Write(entry) + } +} + +func (l *Logger) enabled(level Level) bool { + min := l.min + if l.module != "" { + if ml, ok := l.moduleLevels[l.module]; ok { + min = ml + } + } + return level >= min +} + +func (l *Logger) clone() *Logger { + n := *l + n.fields = cloneFields(l.fields) + n.moduleLevels = l.moduleLevels + n.sinks = l.sinks + return &n +} + +func applyError(entry *Entry, err error) { + if err == nil { + return + } + entry.Error = err.Error() + if coded, ok := err.(codedError); ok { + entry.Code = coded.ErrorCode() + for k, v := range coded.ErrorFields() { + if entry.Fields == nil { + entry.Fields = map[string]any{} + } + entry.Fields[k] = v + } + } +} + +func redactEntry(entry Entry, redact func(string) string) Entry { + entry.Message = redact(entry.Message) + entry.Error = redact(entry.Error) + entry.Fields = redactFields(entry.Fields, redact) + return entry +} + +func redactFields(in map[string]any, redact func(string) string) map[string]any { + if len(in) == 0 { + return nil + } + out := make(map[string]any, len(in)) + for k, v := range in { + out[redact(k)] = redactValue(v, redact) + } + return out +} + +func redactValue(v any, redact func(string) string) any { + switch t := v.(type) { + case string: + return redact(t) + case error: + if t == nil { + return nil + } + return redact(t.Error()) + case map[string]any: + return redactFields(t, redact) + default: + return v + } +} + +func fieldsFrom(kv []any) map[string]any { + if len(kv) == 0 { + return nil + } + if len(kv)%2 != 0 { + panic("logsys: odd number of key-value arguments") + } + out := make(map[string]any, len(kv)/2) + for i := 0; i < len(kv); i += 2 { + key, ok := kv[i].(string) + if !ok || key == "" { + panic(fmt.Sprintf("logsys: field key must be a non-empty string, got %T", kv[i])) + } + out[key] = kv[i+1] + } + return out +} + +func cloneFields(in map[string]any) map[string]any { + if len(in) == 0 { + return nil + } + out := make(map[string]any, len(in)) + for k, v := range in { + out[k] = v + } + return out +} + +func mergeFields(base, extra map[string]any) map[string]any { + if len(base) == 0 && len(extra) == 0 { + return nil + } + out := cloneFields(base) + if out == nil { + out = map[string]any{} + } + for k, v := range extra { + out[k] = v + } + return out +} + +func joinErrs(errs []error) error { + if len(errs) == 0 { + return nil + } + if len(errs) == 1 { + return errs[0] + } + return &multiErr{errs: errs} +} + +type multiErr struct{ errs []error } + +func (m *multiErr) Error() string { + var b bytesTrim + for i, err := range m.errs { + if i > 0 { + b.WriteString("; ") + } + b.WriteString(err.Error()) + } + return b.String() +} + +func (m *multiErr) Unwrap() []error { return m.errs } + +// bytesTrim 避免为拼接错误再引入 bytes 与 strings 的混用。 +type bytesTrim struct{ b []byte } + +func (b *bytesTrim) WriteString(s string) { b.b = append(b.b, s...) } +func (b *bytesTrim) String() string { return string(b.b) } + +var ( + defaultMu sync.RWMutex + defaultL = Nop() +) + +// SetDefault 设置进程默认日志器。nil 等价于 Nop。 +func SetDefault(l *Logger) { + if l == nil { + l = Nop() + } + defaultMu.Lock() + defaultL = l + defaultMu.Unlock() +} + +// Default 返回进程默认日志器。未设置时为 Nop。 +func Default() *Logger { + defaultMu.RLock() + defer defaultMu.RUnlock() + return defaultL +} + +// Sync 刷出默认日志器。 +func Sync() error { return Default().Sync() } diff --git a/logsys/logger_test.go b/logsys/logger_test.go new file mode 100644 index 0000000..9d5f118 --- /dev/null +++ b/logsys/logger_test.go @@ -0,0 +1,134 @@ +package logsys + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" +) + +type memSink struct { + mu sync.Mutex + entries []Entry +} + +func (m *memSink) Write(e Entry) error { + m.mu.Lock() + defer m.mu.Unlock() + m.entries = append(m.entries, e) + return nil +} +func (m *memSink) Sync() error { return nil } +func (m *memSink) Close() error { return nil } + +func (m *memSink) list() []Entry { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]Entry, len(m.entries)) + copy(out, m.entries) + return out +} + +type stubErr struct{} + +func (stubErr) Error() string { return "dial failed" } +func (stubErr) ErrorCode() string { return "agent.provider_unavailable" } +func (stubErr) ErrorFields() map[string]any { + return map[string]any{"model": "glm-4", "reason": "模型不可达"} +} + +func TestLevelFilterModuleAndTrace(t *testing.T) { + mem := &memSink{} + fixed := time.Date(2026, 9, 29, 12, 0, 0, 0, time.UTC) + log := New(Options{ + Service: "agent-squad", + Level: LevelInfo, + ModuleLevels: map[string]Level{ + "engine": LevelDebug, + "server": LevelWarn, + }, + Sinks: []Sink{mem}, + Now: func() time.Time { return fixed }, + }) + log.Debug("hidden") + log.Info("visible") + log.Module("engine").Debug("engine-debug") + log.Module("server").Info("server-info") + log.Module("server").Warn("server-warn") + ctx := WithTrace(context.Background(), "trace-1") + log.WithContext(ctx).With("step", "plan").Info("traced") + + got := mem.list() + if len(got) != 4 { + t.Fatalf("len %d %+v", len(got), got) + } + if got[0].Message != "visible" || got[0].Time != fixed || got[0].Service != "agent-squad" { + t.Fatalf("first %+v", got[0]) + } + if got[1].Module != "engine" || got[1].Message != "engine-debug" { + t.Fatalf("engine %+v", got[1]) + } + if got[2].Message != "server-warn" { + t.Fatalf("server %+v", got[2]) + } + if got[3].TraceID != "trace-1" || got[3].Fields["step"] != "plan" { + t.Fatalf("trace %+v", got[3]) + } +} + +func TestErrorDuckTypingAndRedact(t *testing.T) { + mem := &memSink{} + log := New(Options{ + Level: LevelInfo, + Sinks: []Sink{mem}, + Redact: func(s string) string { return strings.ReplaceAll(s, "glm-4", "***") }, + }) + log.Error("provider down", stubErr{}, "attempt", 2) + got := mem.list() + if len(got) != 1 { + t.Fatal(len(got)) + } + e := got[0] + if e.Code != "agent.provider_unavailable" || e.Level != "error" { + t.Fatalf("%+v", e) + } + if e.Fields["model"] != "***" || e.Fields["reason"] != "模型不可达" || e.Fields["attempt"] != 2 { + t.Fatalf("fields %+v", e.Fields) + } + if !strings.Contains(e.Error, "dial failed") { + t.Fatal(e.Error) + } + plain := errors.New("boom") + log.Error("plain", plain) + if mem.list()[1].Code != "" || mem.list()[1].Error != "boom" { + t.Fatalf("%+v", mem.list()[1]) + } +} + +func TestDefaultLogger(t *testing.T) { + prev := Default() + t.Cleanup(func() { SetDefault(prev) }) + mem := &memSink{} + SetDefault(New(Options{Level: LevelInfo, Sinks: []Sink{mem}})) + Default().Info("from-default") + if err := Sync(); err != nil { + t.Fatal(err) + } + if len(mem.list()) != 1 { + t.Fatal(mem.list()) + } +} + +func TestHumanAndJSON(t *testing.T) { + e := Entry{Time: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), Level: "info", Service: "svc", Message: "hello", Fields: map[string]any{"b": 1, "a": "x"}} + line := FormatHuman(e) + if !strings.Contains(line, "INFO svc hello") || !strings.Contains(line, "a=x") { + t.Fatal(line) + } + raw, err := FormatJSON(e) + if err != nil || !strings.Contains(string(raw), `"msg":"hello"`) { + t.Fatalf("%s %v", raw, err) + } +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..fdb72d8 --- /dev/null +++ b/main.go @@ -0,0 +1,11 @@ +package main + +import ( + "os" + + "logging-system/cli" +) + +func main() { + os.Exit(cli.Main(os.Args[1:])) +} diff --git a/redact/redact.go b/redact/redact.go new file mode 100644 index 0000000..b982561 --- /dev/null +++ b/redact/redact.go @@ -0,0 +1,56 @@ +// Package redact 在日志落盘前去掉常见秘密形状。 +// 规则来自 agent-squad 的事件脱敏，并允许追加自定义正则。 +package redact + +import ( + "fmt" + "regexp" + "strings" +) + +// Redactor 替换秘密子串。宁可多脱，不漏已识别的形状。 +type Redactor struct { + extra []*regexp.Regexp +} + +var ( + reJWT = regexp.MustCompile(`eyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}`) + reSKKey = regexp.MustCompile(`(?:sk|rk|pk)-[A-Za-z0-9_\-]{12,}`) + reBearer = regexp.MustCompile(`(?i)bearer\s+[A-Za-z0-9._\-]{12,}`) + reAPIKey = regexp.MustCompile(`(?i)(api[_-]?key["':=\s]+)[A-Za-z0-9_\-\.]{12,}`) + rePass = regexp.MustCompile(`(?i)(password["':=\s]+)[^\s"',}]+`) +) + +// New 编译额外正则。任一非法则失败，不返回半成品。 +func New(extra []string) (*Redactor, error) { + compiled := make([]*regexp.Regexp, 0, len(extra)) + for _, pattern := range extra { + if strings.TrimSpace(pattern) == "" { + continue + } + re, err := regexp.Compile(pattern) + if err != nil { + return nil, fmt.Errorf("redact: pattern %q: %w", pattern, err) + } + compiled = append(compiled, re) + } + return &Redactor{extra: compiled}, nil +} + +// Apply 脱敏一段文本。 +func (r *Redactor) Apply(s string) string { + if r == nil || s == "" { + return s + } + s = reJWT.ReplaceAllString(s, "***") + s = reSKKey.ReplaceAllString(s, "***") + s = reBearer.ReplaceAllString(s, "Bearer ***") + s = reAPIKey.ReplaceAllString(s, "${1}***") + if strings.Contains(strings.ToLower(s), "password") { + s = rePass.ReplaceAllString(s, "${1}***") + } + for _, re := range r.extra { + s = re.ReplaceAllString(s, "***") + } + return s +} diff --git a/redact/redact_test.go b/redact/redact_test.go new file mode 100644 index 0000000..a02f73b --- /dev/null +++ b/redact/redact_test.go @@ -0,0 +1,36 @@ +package redact + +import ( + "strings" + "testing" +) + +func TestBuiltinShapes(t *testing.T) { + r, err := New(nil) + if err != nil { + t.Fatal(err) + } + in := "token=eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.signatureX sk-abcdefghijklmnopqrstuvwxyz Bearer abcdefghijklmnop api_key=abcdefghijklmnop password=hunter2" + out := r.Apply(in) + for _, secret := range []string{"eyJhbGciOiJIUzI1NiJ9", "sk-abcdefghijklmnopqrstuvwxyz", "abcdefghijklmnop", "hunter2"} { + if strings.Contains(out, secret) { + t.Fatalf("leaked %s in %s", secret, out) + } + } + if !strings.Contains(out, "Bearer ***") || !strings.Contains(out, "password=***") { + t.Fatalf("markers missing: %s", out) + } +} + +func TestExtraPattern(t *testing.T) { + r, err := New([]string{`session-[0-9]+`}) + if err != nil { + t.Fatal(err) + } + if got := r.Apply("session-42"); got != "***" { + t.Fatal(got) + } + if _, err := New([]string{"("}); err == nil { + t.Fatal("expected bad pattern") + } +} diff --git a/server/server.go b/server/server.go new file mode 100644 index 0000000..dd93673 --- /dev/null +++ b/server/server.go @@ -0,0 +1,187 @@ +// Package server 是日志采集服务：批量写入、条件查询、SSE 尾随。默认监听 :8092。 +package server + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" + + "logging-system/logsys" + "logging-system/redact" +) + +// DefaultAddr 是 logcollector 的默认监听地址。 +const DefaultAddr = ":8092" + +// TraceHeader 与错误码系统使用同一个追踪头。 +const TraceHeader = "X-Trace-Id" + +// Server 提供日志采集 HTTP API。 +type Server struct { + Store *Store + Token string +} + +// New 创建服务。redact 为 nil 时仍做内置脱敏。 +func New(path, token string, limit int) (*Server, error) { + rd, err := redact.New(nil) + if err != nil { + return nil, err + } + st, err := OpenStore(path, limit, rd) + if err != nil { + return nil, err + } + return &Server{Store: st, Token: token}, nil +} + +// Handler 返回路由。 +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /v1/health", s.handleHealth) + mux.HandleFunc("POST /v1/logs", s.handleIngest) + mux.HandleFunc("GET /v1/logs", s.handleQuery) + mux.HandleFunc("GET /v1/logs/tail", s.handleTail) + return mux +} + +// ListenAndServe 在 addr 上服务。 +func (s *Server) ListenAndServe(addr string) error { + if strings.TrimSpace(addr) == "" { + addr = DefaultAddr + } + return http.ListenAndServe(addr, s.Handler()) +} + +// Close 关闭存储。 +func (s *Server) Close() error { + if s.Store == nil { + return nil + } + return s.Store.Close() +} + +func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) +} + +func (s *Server) handleIngest(w http.ResponseWriter, r *http.Request) { + if !s.authorized(r) { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 8<<20)) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "body too large or unreadable"}) + return + } + var req struct { + Entries []logsys.Entry `json:"entries"` + } + if err := json.Unmarshal(body, &req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"}) + return + } + traceID := sanitizeTrace(r.Header.Get(TraceHeader)) + for i := range req.Entries { + if req.Entries[i].TraceID == "" && traceID != "" { + req.Entries[i].TraceID = traceID + } + } + recs := s.Store.Append(req.Entries) + writeJSON(w, http.StatusOK, map[string]any{"accepted": len(recs)}) +} + +func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { + if !s.authorized(r) { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + q := r.URL.Query() + cursor, _ := strconv.ParseInt(q.Get("cursor"), 10, 64) + limit, _ := strconv.Atoi(q.Get("limit")) + recs, next := s.Store.Query(Query{ + TraceID: q.Get("trace_id"), + Level: q.Get("level"), + Service: q.Get("service"), + Text: q.Get("q"), + Cursor: cursor, + Limit: limit, + }) + if recs == nil { + recs = []Record{} + } + writeJSON(w, http.StatusOK, map[string]any{"entries": recs, "next_cursor": next}) +} + +func (s *Server) handleTail(w http.ResponseWriter, r *http.Request) { + if !s.authorized(r) { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + flusher, ok := w.(http.Flusher) + if !ok { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "streaming unsupported"}) + return + } + cursor, _ := strconv.ParseInt(r.URL.Query().Get("cursor"), 10, 64) + ch, cancel := s.Store.Subscribe(cursor) + defer cancel() + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.WriteHeader(http.StatusOK) + for { + select { + case <-r.Context().Done(): + return + case rec, ok := <-ch: + if !ok { + return + } + data, err := json.Marshal(rec) + if err != nil { + continue + } + fmt.Fprintf(w, "data: %s\n\n", data) + flusher.Flush() + } + } +} + +func (s *Server) authorized(r *http.Request) bool { + if s.Token == "" { + return true + } + if r.Header.Get("X-Log-Token") == s.Token { + return true + } + auth := r.Header.Get("Authorization") + return auth == "Bearer "+s.Token +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(code) + enc := json.NewEncoder(w) + enc.SetEscapeHTML(false) + _ = enc.Encode(v) +} + +func sanitizeTrace(s string) string { + s = strings.TrimSpace(s) + if len(s) < 8 || len(s) > 128 { + return "" + } + for _, r := range s { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_', r == '.', r == ':': + continue + default: + return "" + } + } + return s +} diff --git a/server/server_test.go b/server/server_test.go new file mode 100644 index 0000000..9d62ed2 --- /dev/null +++ b/server/server_test.go @@ -0,0 +1,98 @@ +package server + +import ( + "bufio" + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestIngestQueryTail(t *testing.T) { + srv, err := New(t.TempDir()+"/logs.jsonl", "tok", 100) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = srv.Close() }) + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + body := []byte(`{"entries":[{"ts":"2026-09-29T00:00:00Z","level":"error","service":"agent-squad","msg":"provider down","trace_id":"trace-abc-1","code":"agent.provider_unavailable","fields":{"api_key":"sk-abcdefghijklmnopqrstuvwxyz"}}]}`) + req, err := http.NewRequest(http.MethodPost, ts.URL+"/v1/logs", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer tok") + res, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if res.StatusCode != http.StatusOK { + t.Fatalf("ingest %d", res.StatusCode) + } + + unauth, err := http.Get(ts.URL + "/v1/logs") + if err != nil { + t.Fatal(err) + } + unauth.Body.Close() + if unauth.StatusCode != http.StatusUnauthorized { + t.Fatalf("auth %d", unauth.StatusCode) + } + + req, err = http.NewRequest(http.MethodGet, ts.URL+"/v1/logs?trace_id=trace-abc-1&level=error", nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("X-Log-Token", "tok") + res, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + var payload struct { + Entries []Record `json:"entries"` + Next int64 `json:"next_cursor"` + } + if err := json.NewDecoder(res.Body).Decode(&payload); err != nil { + t.Fatal(err) + } + if len(payload.Entries) != 1 || payload.Entries[0].Code != "agent.provider_unavailable" { + t.Fatalf("%+v", payload.Entries) + } + raw, _ := json.Marshal(payload.Entries[0]) + if strings.Contains(string(raw), "sk-abcdefghijklmnopqrstuvwxyz") { + t.Fatalf("secret leaked: %s", raw) + } + + req, err = http.NewRequest(http.MethodGet, ts.URL+"/v1/logs/tail?cursor=0", nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("X-Log-Token", "tok") + tailRes, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer tailRes.Body.Close() + reader := bufio.NewReader(tailRes.Body) + line, err := reader.ReadString('\n') + if err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(line, "data: ") || !strings.Contains(line, "provider down") { + t.Fatalf("sse %q", line) + } + + health, err := http.Get(ts.URL + "/v1/health") + if err != nil { + t.Fatal(err) + } + health.Body.Close() + if health.StatusCode != http.StatusOK { + t.Fatal(health.StatusCode) + } +} diff --git a/server/store.go b/server/store.go new file mode 100644 index 0000000..871f992 --- /dev/null +++ b/server/store.go @@ -0,0 +1,245 @@ +package server + +import ( + "bufio" + "encoding/json" + "os" + "sync" + + "logging-system/logsys" + "logging-system/redact" +) + +// Record 是带序号的日志，序号即查询游标。 +type Record struct { + Seq int64 `json:"seq"` + logsys.Entry +} + +// Store 是线程安全的内存日志表，可选追加到 JSONL 文件。 +type Store struct { + mu sync.Mutex + records []Record + nextSeq int64 + limit int + path string + file *os.File + redact *redact.Redactor + subs map[int]chan Record + nextSub int +} + +// OpenStore 打开存储。path 为空则只放内存。limit <= 0 时保留最近 100000 条。 +func OpenStore(path string, limit int, redactor *redact.Redactor) (*Store, error) { + if limit <= 0 { + limit = 100000 + } + s := &Store{ + limit: limit, + path: path, + redact: redactor, + subs: map[int]chan Record{}, + } + if path == "" { + return s, nil + } + if err := s.load(); err != nil { + return nil, err + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return nil, err + } + s.file = f + return s, nil +} + +func (s *Store) load() error { + f, err := os.Open(s.path) + if os.IsNotExist(err) { + return nil + } + if err != nil { + return err + } + defer f.Close() + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 2<<20) + for sc.Scan() { + var rec Record + if err := json.Unmarshal(sc.Bytes(), &rec); err != nil { + continue + } + s.records = append(s.records, rec) + if rec.Seq > s.nextSeq { + s.nextSeq = rec.Seq + } + } + if len(s.records) > s.limit { + s.records = s.records[len(s.records)-s.limit:] + } + return sc.Err() +} + +// Append 写入一批日志并通知订阅者。慢订阅者丢帧，不阻塞写入。 +func (s *Store) Append(entries []logsys.Entry) []Record { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]Record, 0, len(entries)) + for _, e := range entries { + if s.redact != nil { + e.Message = s.redact.Apply(e.Message) + e.Error = s.redact.Apply(e.Error) + e.Fields = redactFields(e.Fields, s.redact) + } + s.nextSeq++ + rec := Record{Seq: s.nextSeq, Entry: e} + s.records = append(s.records, rec) + if s.file != nil { + if data, err := json.Marshal(rec); err == nil { + _, _ = s.file.Write(append(data, '\n')) + } + } + out = append(out, rec) + for _, ch := range s.subs { + select { + case ch <- rec: + default: + } + } + } + if len(s.records) > s.limit { + s.records = s.records[len(s.records)-s.limit:] + } + return out +} + +// Query 按条件过滤。cursor 为已见过的最大序号，只返回更大的序号。 +func (s *Store) Query(q Query) (records []Record, next int64) { + s.mu.Lock() + defer s.mu.Unlock() + if q.Limit <= 0 { + q.Limit = 100 + } + if q.Limit > 1000 { + q.Limit = 1000 + } + for _, rec := range s.records { + if rec.Seq <= q.Cursor { + continue + } + if q.TraceID != "" && rec.TraceID != q.TraceID { + continue + } + if q.Level != "" && rec.Level != q.Level { + continue + } + if q.Service != "" && rec.Service != q.Service { + continue + } + if q.Text != "" && !contains(rec, q.Text) { + continue + } + records = append(records, rec) + next = rec.Seq + if len(records) >= q.Limit { + break + } + } + if next == 0 { + next = q.Cursor + } + return records, next +} + +// Query 是检索条件。 +type Query struct { + TraceID string + Level string + Service string + Text string + Cursor int64 + Limit int +} + +func contains(rec Record, text string) bool { + if substr(rec.Message, text) || substr(rec.Error, text) || substr(rec.Code, text) || substr(rec.TraceID, text) { + return true + } + for k, v := range rec.Fields { + if substr(k, text) || substr(toString(v), text) { + return true + } + } + return false +} + +func substr(s, sub string) bool { + return len(sub) > 0 && len(s) >= len(sub) && (s == sub || indexOf(s, sub) >= 0) +} + +func indexOf(s, sub string) int { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return i + } + } + return -1 +} + +func toString(v any) string { + s, _ := v.(string) + return s +} + +// Subscribe 从 after 序号之后回放，并继续接收新日志。 +func (s *Store) Subscribe(after int64) (<-chan Record, func()) { + s.mu.Lock() + defer s.mu.Unlock() + var replay []Record + for _, rec := range s.records { + if rec.Seq > after { + replay = append(replay, rec) + } + } + buf := len(replay) + 256 + ch := make(chan Record, buf) + for _, rec := range replay { + ch <- rec + } + id := s.nextSub + s.nextSub++ + s.subs[id] = ch + return ch, func() { + s.mu.Lock() + delete(s.subs, id) + s.mu.Unlock() + } +} + +// Close 关闭落盘文件。 +func (s *Store) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.file == nil { + return nil + } + err := s.file.Close() + s.file = nil + return err +} + +func redactFields(in map[string]any, r *redact.Redactor) map[string]any { + if len(in) == 0 { + return in + } + out := make(map[string]any, len(in)) + for k, v := range in { + if s, ok := v.(string); ok { + out[k] = r.Apply(s) + continue + } + out[k] = v + } + return out +} diff --git a/sink/async.go b/sink/async.go new file mode 100644 index 0000000..08048b3 --- /dev/null +++ b/sink/async.go @@ -0,0 +1,140 @@ +package sink + +import ( + "errors" + "sync" + "sync/atomic" + "time" + + "logging-system/logsys" +) + +// ErrClosed 表示出口已经关闭。 +var ErrClosed = errors.New("sink: closed") + +// Async 用有界队列把写日志从调用方 goroutine 挪走。 +// 队列满时丢弃最新一条并计数（不阻塞调用方）。Close 会排空队列并限时等待。 +type Async struct { + Inner logsys.Sink + Queue int + FlushTimeout time.Duration + + ch chan logsys.Entry + syncCh chan chan struct{} + stop chan struct{} + done chan struct{} + once sync.Once + dropped atomic.Int64 + accepted atomic.Int64 + closeOnce sync.Once + closeErr error +} + +// Start 启动后台写者。重复调用无效果。 +func (a *Async) Start() { + a.once.Do(func() { + if a.Inner == nil { + a.Inner = &Memory{} + } + if a.Queue < 1 { + a.Queue = 1024 + } + if a.FlushTimeout <= 0 { + a.FlushTimeout = 5 * time.Second + } + a.ch = make(chan logsys.Entry, a.Queue) + a.syncCh = make(chan chan struct{}) + a.stop = make(chan struct{}) + a.done = make(chan struct{}) + go a.loop() + }) +} + +// Write 非阻塞入队。队列满时丢弃本条。 +func (a *Async) Write(e logsys.Entry) error { + a.Start() + select { + case <-a.stop: + return ErrClosed + default: + } + select { + case a.ch <- e: + a.accepted.Add(1) + return nil + case <-a.stop: + return ErrClosed + default: + a.dropped.Add(1) + return nil + } +} + +// Dropped 返回因队列满被丢弃的条数。 +func (a *Async) Dropped() int64 { return a.dropped.Load() } + +// Accepted 返回成功入队的条数。 +func (a *Async) Accepted() int64 { return a.accepted.Load() } + +// Sync 等待已入队的日志写完并刷出底层出口。 +func (a *Async) Sync() error { + a.Start() + req := make(chan struct{}) + select { + case <-a.stop: + return ErrClosed + case a.syncCh <- req: + case <-time.After(a.FlushTimeout): + return errors.New("sink: async sync enqueue timeout") + } + select { + case <-req: + return nil + case <-time.After(a.FlushTimeout): + return errors.New("sink: async sync timeout") + } +} + +// Close 通知写者排空队列、刷盘并关闭底层出口。重复调用返回同一次结果。 +func (a *Async) Close() error { + a.Start() + a.closeOnce.Do(func() { + close(a.stop) + select { + case <-a.done: + a.closeErr = a.Inner.Close() + case <-time.After(a.FlushTimeout): + a.closeErr = errors.New("sink: async close timeout") + } + }) + return a.closeErr +} + +func (a *Async) loop() { + defer close(a.done) + for { + select { + case <-a.stop: + a.drain() + _ = a.Inner.Sync() + return + case e := <-a.ch: + _ = a.Inner.Write(e) + case req := <-a.syncCh: + a.drain() + _ = a.Inner.Sync() + close(req) + } + } +} + +func (a *Async) drain() { + for { + select { + case e := <-a.ch: + _ = a.Inner.Write(e) + default: + return + } + } +} diff --git a/sink/console.go b/sink/console.go new file mode 100644 index 0000000..0524645 --- /dev/null +++ b/sink/console.go @@ -0,0 +1,56 @@ +package sink + +import ( + "io" + "os" + "sync" + + "logging-system/logsys" +) + +// Console 把日志写到 io.Writer。Format 为 human 或 json，空值按 human。 +type Console struct { + mu sync.Mutex + Out io.Writer + Format string +} + +// Write 写入一行。 +func (c *Console) Write(e logsys.Entry) error { + if c.Out == nil { + c.Out = os.Stderr + } + line, err := formatLine(c.Format, e, true) + if err != nil { + return err + } + c.mu.Lock() + defer c.mu.Unlock() + _, err = c.Out.Write(line) + return err +} + +// Sync 无独立缓冲。 +func (c *Console) Sync() error { return nil } + +// Close 不关闭底层 Writer（它通常是 stderr）。 +func (c *Console) Close() error { return nil } + +func formatLine(format string, e logsys.Entry, humanDefault bool) ([]byte, error) { + if format == "" && humanDefault { + format = "human" + } + if format == "" || format == "json" { + return logsys.FormatJSON(e) + } + if format == "human" { + return []byte(logsys.FormatHuman(e)), nil + } + return nil, errFormat(format) +} + +type formatError string + +func errFormat(f string) error { return formatError(f) } + +func (f formatError) Error() string { return "sink: unknown format " + string(f) } diff --git a/sink/fanout.go b/sink/fanout.go new file mode 100644 index 0000000..269555a --- /dev/null +++ b/sink/fanout.go @@ -0,0 +1,91 @@ +package sink + +import ( + "errors" + + "logging-system/logsys" +) + +// Fanout 把同一条日志写到多个出口。单个出口失败不影响其他出口。 +type Fanout struct { + Sinks []logsys.Sink +} + +// Write 逐个写入。 +func (f Fanout) Write(e logsys.Entry) error { + var errs []error + for _, s := range f.Sinks { + if s == nil { + continue + } + if err := s.Write(e); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// Sync 刷出全部出口。 +func (f Fanout) Sync() error { + var errs []error + for _, s := range f.Sinks { + if s == nil { + continue + } + if err := s.Sync(); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// Close 关闭全部出口。 +func (f Fanout) Close() error { + var errs []error + for _, s := range f.Sinks { + if s == nil { + continue + } + if err := s.Close(); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// MinLevel 丢弃低于阈值的日志。 +type MinLevel struct { + Min logsys.Level + Next logsys.Sink +} + +// Write 在级别足够时转发。 +func (m MinLevel) Write(e logsys.Entry) error { + lv, err := logsys.ParseLevel(e.Level) + if err != nil { + return err + } + if lv < m.Min { + return nil + } + if m.Next == nil { + return nil + } + return m.Next.Write(e) +} + +// Sync 转发。 +func (m MinLevel) Sync() error { + if m.Next == nil { + return nil + } + return m.Next.Sync() +} + +// Close 转发。 +func (m MinLevel) Close() error { + if m.Next == nil { + return nil + } + return m.Next.Close() +} diff --git a/sink/file.go b/sink/file.go new file mode 100644 index 0000000..74b3e45 --- /dev/null +++ b/sink/file.go @@ -0,0 +1,116 @@ +package sink + +import ( + "fmt" + "os" + "path/filepath" + "sync" + + "logging-system/logsys" +) + +// File 以互斥锁保护的方式追加日志，可选按大小滚动。 +// MaxBytes 为 0 时不滚动。Keep 是保留的历史文件份数（不含当前文件）。 +type File struct { + mu sync.Mutex + Path string + Format string + MaxBytes int64 + Keep int + + f *os.File + size int64 +} + +// Write 追加一条日志，必要时先滚动。 +func (f *File) Write(e logsys.Entry) error { + line, err := formatLine(f.Format, e, false) + if err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + if err := f.open(); err != nil { + return err + } + if f.MaxBytes > 0 && f.size > 0 && f.size+int64(len(line)) > f.MaxBytes { + if err := f.rotate(); err != nil { + return err + } + } + n, err := f.f.Write(line) + f.size += int64(n) + return err +} + +// Sync 把文件缓冲刷到磁盘。 +func (f *File) Sync() error { + f.mu.Lock() + defer f.mu.Unlock() + if f.f == nil { + return nil + } + return f.f.Sync() +} + +// Close 关闭文件句柄。 +func (f *File) Close() error { + f.mu.Lock() + defer f.mu.Unlock() + if f.f == nil { + return nil + } + err := f.f.Close() + f.f = nil + return err +} + +func (f *File) open() error { + if f.f != nil { + return nil + } + if f.Path == "" { + return fmt.Errorf("sink: file path is required") + } + if err := os.MkdirAll(filepath.Dir(f.Path), 0o755); err != nil { + return err + } + fp, err := os.OpenFile(f.Path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return err + } + info, err := fp.Stat() + if err != nil { + _ = fp.Close() + return err + } + f.f = fp + f.size = info.Size() + return nil +} + +func (f *File) rotate() error { + if f.f != nil { + if err := f.f.Close(); err != nil { + return err + } + f.f = nil + } + keep := f.Keep + if keep < 1 { + keep = 1 + } + _ = os.Remove(fmt.Sprintf("%s.%d", f.Path, keep)) + for i := keep - 1; i >= 1; i-- { + src := fmt.Sprintf("%s.%d", f.Path, i) + dst := fmt.Sprintf("%s.%d", f.Path, i+1) + if _, err := os.Stat(src); err == nil { + _ = os.Rename(src, dst) + } + } + if err := os.Rename(f.Path, f.Path+".1"); err != nil && !os.IsNotExist(err) { + return err + } + f.size = 0 + return f.open() +} diff --git a/sink/httpbatch.go b/sink/httpbatch.go new file mode 100644 index 0000000..71812e1 --- /dev/null +++ b/sink/httpbatch.go @@ -0,0 +1,186 @@ +package sink + +import ( + "bytes" + "encoding/json" + "errors" + "net/http" + "sync" + "sync/atomic" + "time" + + "logging-system/logsys" +) + +// HTTPBatch 把日志攒批 POST 到采集器。Async 为 false 时在 Write 里同步发送。 +type HTTPBatch struct { + URL string + Token string + MaxEntries int + FlushInterval time.Duration + MaxRetries int + Async bool + Client *http.Client + FlushTimeout time.Duration + + mu sync.Mutex + buf []logsys.Entry + stop chan struct{} + done chan struct{} + started bool + dropped atomic.Int64 + once sync.Once +} + +// Start 启动异步刷盘。同步模式不必调用。 +func (h *HTTPBatch) Start() { + if !h.Async { + h.defaults() + return + } + h.once.Do(func() { + h.defaults() + h.stop = make(chan struct{}) + h.done = make(chan struct{}) + h.started = true + go h.loop() + }) +} + +func (h *HTTPBatch) defaults() { + if h.MaxEntries < 1 { + h.MaxEntries = 200 + } + if h.FlushInterval <= 0 { + h.FlushInterval = 2 * time.Second + } + if h.MaxRetries < 1 { + h.MaxRetries = 3 + } + if h.Client == nil { + h.Client = &http.Client{Timeout: 5 * time.Second} + } + if h.FlushTimeout <= 0 { + h.FlushTimeout = 5 * time.Second + } +} + +// Write 追加一条日志。同步模式立即发送；异步模式达到批量上限时触发发送。 +func (h *HTTPBatch) Write(e logsys.Entry) error { + h.Start() + if !h.Async { + return h.post([]logsys.Entry{e}) + } + h.mu.Lock() + h.buf = append(h.buf, e) + full := len(h.buf) >= h.MaxEntries + var batch []logsys.Entry + if full { + batch = h.take() + } + h.mu.Unlock() + if full { + if err := h.post(batch); err != nil { + h.dropped.Add(int64(len(batch))) + return err + } + } + return nil +} + +// Dropped 返回发送失败后丢弃的条数。 +func (h *HTTPBatch) Dropped() int64 { return h.dropped.Load() } + +// Sync 立刻送出缓冲。 +func (h *HTTPBatch) Sync() error { + h.Start() + h.mu.Lock() + batch := h.take() + h.mu.Unlock() + if len(batch) == 0 { + return nil + } + if err := h.post(batch); err != nil { + h.dropped.Add(int64(len(batch))) + return err + } + return nil +} + +// Close 停止后台循环并送出剩余日志。 +func (h *HTTPBatch) Close() error { + h.Start() + if h.Async && h.started { + select { + case <-h.stop: + default: + close(h.stop) + } + select { + case <-h.done: + case <-time.After(h.FlushTimeout): + return errors.New("sink: http batch close timeout") + } + } + return h.Sync() +} + +func (h *HTTPBatch) loop() { + defer close(h.done) + ticker := time.NewTicker(h.FlushInterval) + defer ticker.Stop() + for { + select { + case <-h.stop: + return + case <-ticker.C: + _ = h.Sync() + } + } +} + +func (h *HTTPBatch) take() []logsys.Entry { + if len(h.buf) == 0 { + return nil + } + out := h.buf + h.buf = nil + return out +} + +func (h *HTTPBatch) post(batch []logsys.Entry) error { + if len(batch) == 0 { + return nil + } + if h.URL == "" { + return errors.New("sink: http url is required") + } + body, err := json.Marshal(map[string]any{"entries": batch}) + if err != nil { + return err + } + var last error + for attempt := 0; attempt < h.MaxRetries; attempt++ { + req, err := http.NewRequest(http.MethodPost, h.URL, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if h.Token != "" { + req.Header.Set("Authorization", "Bearer "+h.Token) + } + res, err := h.Client.Do(req) + if err == nil { + _ = res.Body.Close() + if res.StatusCode >= 200 && res.StatusCode < 300 { + return nil + } + err = errors.New(res.Status) + } + last = err + if attempt+1 < h.MaxRetries { + time.Sleep(time.Duration(attempt+1) * 10 * time.Millisecond) + } + } + return last +} diff --git a/sink/memory.go b/sink/memory.go new file mode 100644 index 0000000..12b546f --- /dev/null +++ b/sink/memory.go @@ -0,0 +1,36 @@ +package sink + +import ( + "sync" + + "logging-system/logsys" +) + +// Memory 把日志留在内存里，供测试和进程内订阅前的缓冲。 +type Memory struct { + mu sync.Mutex + Entries []logsys.Entry +} + +// Write 追加一条日志。 +func (m *Memory) Write(e logsys.Entry) error { + m.mu.Lock() + defer m.mu.Unlock() + m.Entries = append(m.Entries, e) + return nil +} + +// Sync 无缓冲可刷。 +func (m *Memory) Sync() error { return nil } + +// Close 不释放已记录的日志。 +func (m *Memory) Close() error { return nil } + +// Snapshot 返回当前副本。 +func (m *Memory) Snapshot() []logsys.Entry { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]logsys.Entry, len(m.Entries)) + copy(out, m.Entries) + return out +} diff --git a/sink/sink_test.go b/sink/sink_test.go new file mode 100644 index 0000000..fcd8586 --- /dev/null +++ b/sink/sink_test.go @@ -0,0 +1,211 @@ +package sink + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "logging-system/logsys" +) + +func entry(i int) logsys.Entry { + return logsys.Entry{Time: time.Unix(0, 0).UTC(), Level: "info", Message: "m", Fields: map[string]any{"i": i}} +} + +func TestFileRotateAndConsole(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "app.jsonl") + f := &File{Path: path, Format: "json", MaxBytes: 80, Keep: 2} + for i := 0; i < 5; i++ { + if err := f.Write(entry(i)); err != nil { + t.Fatal(err) + } + } + if err := f.Sync(); err != nil { + t.Fatal(err) + } + if err := f.Close(); err != nil { + t.Fatal(err) + } + matches, err := filepath.Glob(filepath.Join(dir, "app.jsonl*")) + if err != nil { + t.Fatal(err) + } + if len(matches) < 2 { + t.Fatalf("rotated files: %v", matches) + } + var lines int + for _, name := range matches { + b, err := os.ReadFile(name) + if err != nil { + t.Fatal(err) + } + for _, c := range b { + if c == '\n' { + lines++ + } + } + } + if lines != 3 { + t.Fatalf("kept lines %d, want current+2 backups, files %v", lines, matches) + } + + var buf safeBuf + c := &Console{Out: &buf, Format: "human"} + if err := c.Write(logsys.Entry{Time: time.Unix(0, 0).UTC(), Level: "info", Message: "hi"}); err != nil { + t.Fatal(err) + } + if buf.String() == "" { + t.Fatal("empty console") + } +} + +type safeBuf struct { + mu sync.Mutex + b []byte +} + +func (s *safeBuf) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.b = append(s.b, p...) + return len(p), nil +} +func (s *safeBuf) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return string(s.b) +} + +func TestAsyncDropAndDrain(t *testing.T) { + hold := &lockSink{entered: make(chan struct{})} + hold.mu.Lock() + a := &Async{Inner: hold, Queue: 2, FlushTimeout: 2 * time.Second} + go func() { _ = a.Write(entry(0)) }() + select { + case <-hold.entered: + case <-time.After(2 * time.Second): + t.Fatal("worker did not enter write") + } + for i := 0; i < a.Queue; i++ { + if err := a.Write(entry(i + 1)); err != nil { + t.Fatal(err) + } + } + const extra = 15 + for i := 0; i < extra; i++ { + if err := a.Write(entry(100 + i)); err != nil { + t.Fatal(err) + } + } + if a.Dropped() != extra { + t.Fatalf("dropped %d", a.Dropped()) + } + hold.mu.Unlock() + if err := a.Close(); err != nil { + t.Fatal(err) + } + if hold.written.Load() != int64(1+a.Queue) { + t.Fatalf("written %d accepted %d", hold.written.Load(), a.Accepted()) + } +} + +type lockSink struct { + mu sync.Mutex + entered chan struct{} + written atomic.Int64 +} + +func (s *lockSink) Write(e logsys.Entry) error { + select { + case <-s.entered: + default: + close(s.entered) + } + s.mu.Lock() + s.mu.Unlock() + s.written.Add(1) + return nil +} +func (s *lockSink) Sync() error { return nil } +func (s *lockSink) Close() error { return nil } + +func TestAsyncSyncDrains(t *testing.T) { + mem := &Memory{} + a := &Async{Inner: mem, Queue: 64} + for i := 0; i < 30; i++ { + if err := a.Write(entry(i)); err != nil { + t.Fatal(err) + } + } + if err := a.Sync(); err != nil { + t.Fatal(err) + } + if len(mem.Snapshot()) != 30 { + t.Fatalf("synced %d", len(mem.Snapshot())) + } + if err := a.Close(); err != nil { + t.Fatal(err) + } +} + +func TestFanoutAndHTTPBatch(t *testing.T) { + a, b := &Memory{}, &Memory{} + f := Fanout{Sinks: []logsys.Sink{ + MinLevel{Min: logsys.LevelWarn, Next: a}, + b, + }} + _ = f.Write(logsys.Entry{Level: "info", Message: "low"}) + _ = f.Write(logsys.Entry{Level: "error", Message: "high"}) + if len(a.Snapshot()) != 1 || len(b.Snapshot()) != 2 { + t.Fatalf("a=%d b=%d", len(a.Snapshot()), len(b.Snapshot())) + } + + var got atomic.Int32 + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.Header.Get("Authorization") != "Bearer secret" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + if calls.Load() == 1 { + http.Error(w, "busy", http.StatusServiceUnavailable) + return + } + var body struct { + Entries []logsys.Entry `json:"entries"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + return + } + got.Add(int32(len(body.Entries))) + })) + defer srv.Close() + h := &HTTPBatch{ + URL: srv.URL, Token: "secret", Async: true, + MaxEntries: 2, FlushInterval: time.Hour, MaxRetries: 3, + } + if err := h.Write(entry(1)); err != nil { + t.Fatal(err) + } + if err := h.Write(entry(2)); err != nil { + t.Fatal(err) + } + if err := h.Write(entry(3)); err != nil { + t.Fatal(err) + } + if err := h.Close(); err != nil { + t.Fatal(err) + } + if got.Load() != 3 { + t.Fatalf("posted %d dropped %d", got.Load(), h.Dropped()) + } +} diff --git a/slogadapter/slogadapter.go b/slogadapter/slogadapter.go new file mode 100644 index 0000000..4f98c92 --- /dev/null +++ b/slogadapter/slogadapter.go @@ -0,0 +1,159 @@ +// Package slogadapter 在 logsys 与标准库 log/slog 之间双向转发。 +package slogadapter + +import ( + "context" + "log/slog" + + "logging-system/logsys" +) + +// Handler 把 slog 记录写入 logsys。 +type Handler struct { + logger *logsys.Logger + group string + attrs []slog.Attr +} + +// NewHandler 包装一个 logsys 日志器。 +func NewHandler(l *logsys.Logger) *Handler { + if l == nil { + l = logsys.Nop() + } + return &Handler{logger: l} +} + +// Slog 返回写入 logsys 的 *slog.Logger。 +func Slog(l *logsys.Logger) *slog.Logger { return slog.New(NewHandler(l)) } + +// Enabled 按 logsys 的级别过滤。 +func (h *Handler) Enabled(_ context.Context, level slog.Level) bool { + return h.logger.Enabled(fromSlogLevel(level)) +} + +// Handle 转发一条记录。 +func (h *Handler) Handle(ctx context.Context, rec slog.Record) error { + kv := make([]any, 0, (len(h.attrs)+rec.NumAttrs())*2) + for _, attr := range h.attrs { + kv = append(kv, h.name(attr.Key), attr.Value.Any()) + } + rec.Attrs(func(attr slog.Attr) bool { + kv = append(kv, h.name(attr.Key), attr.Value.Any()) + return true + }) + l := h.logger.WithContext(ctx) + switch { + case rec.Level >= slog.LevelError+4: + l.Fatal(rec.Message, nil, kv...) + case rec.Level >= slog.LevelError: + l.Error(rec.Message, nil, kv...) + case rec.Level >= slog.LevelWarn: + l.Warn(rec.Message, kv...) + case rec.Level >= slog.LevelInfo: + l.Info(rec.Message, kv...) + case rec.Level >= slog.LevelDebug: + l.Debug(rec.Message, kv...) + default: + l.Trace(rec.Message, kv...) + } + return nil +} + +// WithAttrs 追加属性。 +func (h *Handler) WithAttrs(attrs []slog.Attr) slog.Handler { + n := *h + n.attrs = append(append([]slog.Attr{}, h.attrs...), attrs...) + return &n +} + +// WithGroup 追加分组前缀。 +func (h *Handler) WithGroup(name string) slog.Handler { + n := *h + if name == "" { + return &n + } + if n.group != "" { + n.group += "." + } + n.group += name + return &n +} + +func (h *Handler) name(key string) string { + if h.group == "" { + return key + } + return h.group + "." + key +} + +// Sink 把 logsys 条目写进 slog.Handler。 +type Sink struct { + Handler slog.Handler +} + +// Write 转发一条日志。 +func (s Sink) Write(e logsys.Entry) error { + if s.Handler == nil { + return nil + } + rec := slog.NewRecord(e.Time, toSlogLevel(e.Level), e.Message, 0) + if e.Service != "" { + rec.AddAttrs(slog.String("service", e.Service)) + } + if e.Module != "" { + rec.AddAttrs(slog.String("module", e.Module)) + } + if e.TraceID != "" { + rec.AddAttrs(slog.String("trace_id", e.TraceID)) + } + if e.Code != "" { + rec.AddAttrs(slog.String("code", e.Code)) + } + if e.Error != "" { + rec.AddAttrs(slog.String("error", e.Error)) + } + for k, v := range e.Fields { + rec.AddAttrs(slog.Any(k, v)) + } + return s.Handler.Handle(context.Background(), rec) +} + +// Sync 无独立缓冲。 +func (s Sink) Sync() error { return nil } + +// Close 不关闭 slog.Handler。 +func (s Sink) Close() error { return nil } + +func fromSlogLevel(level slog.Level) logsys.Level { + switch { + case level >= slog.LevelError+4: + return logsys.LevelFatal + case level >= slog.LevelError: + return logsys.LevelError + case level >= slog.LevelWarn: + return logsys.LevelWarn + case level >= slog.LevelInfo: + return logsys.LevelInfo + case level >= slog.LevelDebug: + return logsys.LevelDebug + default: + return logsys.LevelTrace + } +} + +func toSlogLevel(level string) slog.Level { + switch level { + case "trace": + return slog.LevelDebug - 4 + case "debug": + return slog.LevelDebug + case "warn": + return slog.LevelWarn + case "error": + return slog.LevelError + case "fatal": + return slog.LevelError + 4 + default: + return slog.LevelInfo + } +} diff --git a/slogadapter/slogadapter_test.go b/slogadapter/slogadapter_test.go new file mode 100644 index 0000000..27ec7cc --- /dev/null +++ b/slogadapter/slogadapter_test.go @@ -0,0 +1,33 @@ +package slogadapter + +import ( + "bytes" + "log/slog" + "strings" + "testing" + + "logging-system/logsys" + "logging-system/sink" +) + +func TestSlogToLogsys(t *testing.T) { + mem := &sink.Memory{} + base := logsys.New(logsys.Options{Level: logsys.LevelInfo, Sinks: []logsys.Sink{mem}}) + logger := Slog(base).With("component", "api") + logger.Info("hello", "n", 1) + logger.Debug("hidden") + got := mem.Snapshot() + if len(got) != 1 || got[0].Message != "hello" || got[0].Fields["component"] != "api" || got[0].Fields["n"] != int64(1) && got[0].Fields["n"] != 1 { + t.Fatalf("%+v", got) + } +} + +func TestLogsysToSlog(t *testing.T) { + var buf bytes.Buffer + h := slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}) + log := logsys.New(logsys.Options{Level: logsys.LevelInfo, Sinks: []logsys.Sink{Sink{Handler: h}}}) + log.WithTraceID("trace-9").Error("boom", nil, "code_hint", "x") + if !strings.Contains(buf.String(), "boom") || !strings.Contains(buf.String(), "trace-9") { + t.Fatalf("slog output: %s", buf.String()) + } +}