From d3fef656edc71813ea0c9982eb939ec81f8788f4 Mon Sep 17 00:00:00 2001 From: Andrewkydev Date: Wed, 4 Mar 2026 22:21:59 +0300 Subject: [PATCH] chore: update console logging docs and pipeline behavior --- README.md | 44 +++++++++ api.go | 2 + examples/pipeline/main.go | 13 +++ logging/logging.go | 194 +++++++++++++++++++++++++++++++++++++- 4 files changed, 250 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 1555c7d..706f91b 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,16 @@ logFile, err := consolex.SetupDefaultSlog(consolex.LoggerConfig{ ArchiveDir: "logs", Level: slog.LevelDebug, Theme: consolex.NordTheme(), + Dedupe: consolex.DedupeConfig{ + Enabled: true, + Remap: []consolex.LevelRemapRule{ + { + From: "INFO", + To: "DEBUG", + Contains: []string{"conn write packet failed", "context canceled"}, + }, + }, + }, FieldProvider: consolex.StaticFieldProvider{ "proto_id": consolex.New().White().BgBlue().Bold(), "raddr": consolex.New().Black().BgYellow(), @@ -58,6 +68,40 @@ logFile, err := consolex.SetupDefaultSlog(consolex.LoggerConfig{ } ``` +## Dedup/Aggregation (`xN`) + +Enable duplicate line aggregation in `LoggerConfig.Dedupe`: + +```go +cfg := consolex.LoggerConfig{ + Level: slog.LevelDebug, + Dedupe: consolex.DedupeConfig{ + Enabled: true, + Window: time.Second, // default: 1s + }, +} +``` + +Repeated identical lines are collapsed and emitted once with: +- `repeat="xN"` + +Example: + +```text +time=... level=INFO msg="conn write packet failed" packet=*packet.UpdateBlock err="context canceled" repeat="x37" +``` + +You can also remap levels by rules: + +```go +Dedupe: consolex.DedupeConfig{ + Enabled: true, + Remap: []consolex.LevelRemapRule{ + {From: "INFO", To: "DEBUG", Contains: []string{"context canceled"}}, + }, +}, +``` + ## Chalk-like style API ```go diff --git a/api.go b/api.go index 91d274c..b00d4fe 100644 --- a/api.go +++ b/api.go @@ -19,6 +19,8 @@ func NordTheme() Theme { return style.NordTheme() } func SunsetTheme() Theme { return style.SunsetTheme() } type LoggerConfig = logging.LoggerConfig +type DedupeConfig = logging.DedupeConfig +type LevelRemapRule = logging.LevelRemapRule type Profile = logging.Profile type LogRecord = logging.LogRecord type RecordField = logging.RecordField diff --git a/examples/pipeline/main.go b/examples/pipeline/main.go index a3a2ce1..9068a8a 100644 --- a/examples/pipeline/main.go +++ b/examples/pipeline/main.go @@ -17,6 +17,16 @@ func main() { Level: slog.LevelDebug, Theme: consolex.DefaultTheme(), Profile: profile, + Dedupe: consolex.DedupeConfig{ + Enabled: true, + Remap: []consolex.LevelRemapRule{ + { + From: "INFO", + To: "DEBUG", + Contains: []string{"conn write packet failed", "context canceled"}, + }, + }, + }, Processors: []consolex.Processor{ consolex.ProcessorFunc(func(rec *consolex.LogRecord) { // Example: force showing key for world pointer fields in compact layout. @@ -44,4 +54,7 @@ func main() { defer logFile.Close() slog.Info("world register", "name", "hub_snow", "ptr", "0xc004ba0ea0") + for i := 0; i < 10; i++ { + slog.Info("conn write packet failed", "packet", "*packet.UpdateBlock", "err", "context canceled") + } } diff --git a/logging/logging.go b/logging/logging.go index 789e0ae..b9be2ea 100644 --- a/logging/logging.go +++ b/logging/logging.go @@ -9,6 +9,7 @@ import ( "log/slog" "os" "path/filepath" + "strconv" "strings" "sync" "time" @@ -347,6 +348,20 @@ type LoggerConfig struct { FieldTransform FieldTransformer Processors []Processor Renderer Renderer + Dedupe DedupeConfig +} + +type DedupeConfig struct { + Enabled bool + Window time.Duration + KeyFunc func(rec *LogRecord) string + Remap []LevelRemapRule +} + +type LevelRemapRule struct { + From string + To string + Contains []string } func SetupDefaultSlog(cfg LoggerConfig) (*os.File, error) { @@ -380,9 +395,19 @@ func SetupDefaultSlog(cfg LoggerConfig) (*os.File, error) { return nil, fmt.Errorf("open %s: %w", logPath, err) } - consoleOut := NewColorizingWriter(os.Stdout) - consoleHandler := slog.NewTextHandler(consoleOut, &slog.HandlerOptions{Level: cfg.Level}) - fileHandler := slog.NewTextHandler(file, &slog.HandlerOptions{Level: cfg.Level}) + consoleSink := io.Writer(NewColorizingWriter(os.Stdout)) + fileSink := io.Writer(file) + if cfg.Dedupe.Enabled { + window := cfg.Dedupe.Window + if window <= 0 { + window = time.Second + } + consoleSink = NewAggregateLineWriter(consoleSink, window, cfg.Dedupe.KeyFunc, cfg.Dedupe.Remap) + fileSink = NewAggregateLineWriter(fileSink, window, cfg.Dedupe.KeyFunc, cfg.Dedupe.Remap) + } + + consoleHandler := slog.NewTextHandler(consoleSink, &slog.HandlerOptions{Level: cfg.Level}) + fileHandler := slog.NewTextHandler(fileSink, &slog.HandlerOptions{Level: cfg.Level}) slog.SetDefault(slog.New(fanoutHandler{handlers: []slog.Handler{consoleHandler, fileHandler}})) return file, nil } @@ -514,3 +539,166 @@ func ColorizeLogLine(line string) string { } return pl.Colorize(line) } + +type aggregateEntry struct { + key string + line string + count int +} + +type AggregateLineWriter struct { + dst io.Writer + window time.Duration + keyFn func(*LogRecord) string + remap []LevelRemapRule + + mu sync.Mutex + buf []byte + timer *time.Timer + cur *aggregateEntry +} + +func NewAggregateLineWriter(dst io.Writer, window time.Duration, keyFn func(*LogRecord) string, remap []LevelRemapRule) *AggregateLineWriter { + if window <= 0 { + window = time.Second + } + return &AggregateLineWriter{ + dst: dst, + window: window, + keyFn: keyFn, + remap: remap, + } +} + +func (w *AggregateLineWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + + w.buf = append(w.buf, p...) + for { + i := bytes.IndexByte(w.buf, '\n') + if i < 0 { + break + } + line := string(w.buf[:i]) + w.buf = w.buf[i+1:] + w.ingestLocked(line) + } + return len(p), nil +} + +func (w *AggregateLineWriter) ingestLocked(line string) { + rec := ParseTextLogLine(line) + applyLevelRemap(rec, w.remap) + rendered := renderTextRecord(rec) + + key := rendered + if w.keyFn != nil { + key = w.keyFn(rec) + } + if key == "" { + key = rendered + } + + if w.cur != nil && w.cur.key == key { + w.cur.count++ + w.resetTimerLocked() + return + } + + _ = w.flushLocked() + w.cur = &aggregateEntry{ + key: key, + line: rendered, + count: 1, + } + w.resetTimerLocked() +} + +func (w *AggregateLineWriter) resetTimerLocked() { + if w.timer == nil { + w.timer = time.AfterFunc(w.window, w.onTimer) + return + } + w.timer.Reset(w.window) +} + +func (w *AggregateLineWriter) onTimer() { + w.mu.Lock() + defer w.mu.Unlock() + _ = w.flushLocked() +} + +func (w *AggregateLineWriter) flushLocked() error { + if w.cur == nil { + return nil + } + line := w.cur.line + if w.cur.count > 1 { + line = line + " repeat=" + strconv.Quote("x"+strconv.Itoa(w.cur.count)) + } + w.cur = nil + _, err := io.WriteString(w.dst, line+"\n") + return err +} + +func applyLevelRemap(rec *LogRecord, rules []LevelRemapRule) { + if rec == nil || len(rules) == 0 { + return + } + for _, rule := range rules { + if !matchesLevelRule(rec, rule) { + continue + } + to := strings.TrimSpace(rule.To) + if to != "" { + rec.Level = strings.ToUpper(to) + } + return + } +} + +func matchesLevelRule(rec *LogRecord, rule LevelRemapRule) bool { + from := strings.TrimSpace(rule.From) + if from != "" && !strings.EqualFold(strings.TrimSpace(rec.Level), from) { + return false + } + if len(rule.Contains) == 0 { + return true + } + hay := strings.ToLower(rec.Raw) + for _, needle := range rule.Contains { + n := strings.ToLower(strings.TrimSpace(needle)) + if n == "" { + continue + } + if !strings.Contains(hay, n) { + return false + } + } + return true +} + +func renderTextRecord(rec *LogRecord) string { + if rec == nil { + return "" + } + parts := make([]string, 0, 3+len(rec.Fields)) + if rec.Time != "" { + parts = append(parts, "time="+rec.Time) + } + if rec.Level != "" { + parts = append(parts, "level="+strings.ToUpper(rec.Level)) + } + if rec.Message != "" { + parts = append(parts, "msg="+rec.Message) + } + for _, f := range rec.Fields { + value := f.Value + if f.ValueOut != "" { + value = f.ValueOut + } + parts = append(parts, f.Key+"="+value) + } + return strings.Join(parts, " ") +}