From b39397ded03bd8dc6b6aeb18598adcac5c020aa8 Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 15 Mar 2026 10:18:55 +0800 Subject: [PATCH] feat(monitor): integrate monitor service start and stop in load and unload processes - Added functionality to start the monitor service during the loading process, ensuring that watchers are registered. - Implemented the stopping of the monitor service before unloading other services to allow for event handling. - Updated middleware to utilize access logging for improved request tracking. Made-with: Cursor --- engine/load.go | 12 + monitor/README.md | 153 +++++++++++ monitor/logger.go | 60 +++++ monitor/service.go | 250 ++++++++++++++++++ monitor/service_test.go | 507 +++++++++++++++++++++++++++++++++++++ monitor/types.go | 55 ++++ service/log/access.go | 96 +++++++ service/log/access_test.go | 214 ++++++++++++++++ service/middleware.go | 3 +- service/service.go | 3 + 10 files changed, 1352 insertions(+), 1 deletion(-) create mode 100644 monitor/README.md create mode 100644 monitor/logger.go create mode 100644 monitor/service.go create mode 100644 monitor/service_test.go create mode 100644 monitor/types.go create mode 100644 service/log/access.go create mode 100644 service/log/access_test.go diff --git a/engine/load.go b/engine/load.go index ac02803b..646887dd 100644 --- a/engine/load.go +++ b/engine/load.go @@ -34,6 +34,7 @@ import ( "github.com/yaoapp/yao/messenger" "github.com/yaoapp/yao/moapi" "github.com/yaoapp/yao/model" + "github.com/yaoapp/yao/monitor" "github.com/yaoapp/yao/openapi" "github.com/yaoapp/yao/pack" "github.com/yaoapp/yao/pipe" @@ -242,6 +243,14 @@ func Load(cfg config.Config, options LoadOption, progressCallback ...func(string warnings = append(warnings, Warning{Widget: "Event", Error: err}) } + // Start Monitor Service (watchers registered via init()) + err = loadStep("Monitor", func() error { + return monitor.Start(context.Background()) + }, callback) + if err != nil { + warnings = append(warnings, Warning{Widget: "Monitor", Error: err}) + } + // Load Uploaders err = loadStep("Uploader", func() error { return attachment.Load(cfg) @@ -455,6 +464,9 @@ func Unload() (err error) { } } + // Stop Monitor Service (before event, so watchers can still push events) + monitor.Stop() + // Stop Event Service (before runtime, so in-flight handlers can still use V8) event.Stop(context.Background()) diff --git a/monitor/README.md b/monitor/README.md new file mode 100644 index 00000000..5908b71a --- /dev/null +++ b/monitor/README.md @@ -0,0 +1,153 @@ +# Yao Monitor + +A process-level inspection service for Yao. Monitor schedules periodic health checks (watchers) and records anomalies. It knows nothing about business logic — the business layer defines what to check, how to judge, and what action to take. + +## Quick Start + +### 1. Implement a Watcher + +```go +package sandbox + +import ( + "context" + "time" + + "github.com/yaoapp/yao/monitor" +) + +type sandboxWatcher struct{} + +func (w *sandboxWatcher) Name() string { return "sandbox" } +func (w *sandboxWatcher) Interval() time.Duration { return 30 * time.Second } + +func (w *sandboxWatcher) Check(ctx context.Context) []monitor.Alert { + // Inspect containers, compare states, detect idle timeouts, etc. + // Return an empty slice if everything is normal. + return nil +} +``` + +### 2. Register via init() + +```go +func init() { + monitor.Register(&sandboxWatcher{}) +} +``` + +Registration happens before `monitor.Start()` is called. The watcher will be picked up automatically when the engine boots. + +### 3. That's It + +The engine calls `monitor.Start()` / `monitor.Stop()` during load/unload. Your watcher's `Check()` will be called at the interval you specified, in its own goroutine. + +## Alert Levels + +| Level | Constant | Use Case | +|-------|----------|----------| +| Trace | `monitor.Trace` | Heartbeat, periodic status sync, routine checks | +| Info | `monitor.Info` | Notable events: state changes, service registrations | +| Warn | `monitor.Warn` | Needs attention: idle timeout, degraded state | +| Error | `monitor.Error` | Needs immediate action: crash, unreachable | + +Which level to use is entirely up to the business watcher — the monitor just records what it's told. + +All alert levels are delivered to subscribers via `Subscribe()`. + +### Log Level by Mode + +The minimum level written to `monitor.log` depends on Yao's run mode (`YAO_ENV`): + +| Mode | Min Level | Effect | +|------|-----------|--------| +| `production` | Info | Trace alerts are **not** written to the log file | +| `development` | Trace | **Everything** is written | + +This keeps production logs lean while giving full visibility during development. + +## Alert Actions + +An alert can carry an `Action` — a function the monitor executes synchronously within the tick: + +```go +monitor.Alert{ + Level: monitor.Warn, + Target: "box:abc123", + Message: "idle timeout exceeded, stopping", + Action: func(ctx context.Context) { box.Stop(ctx) }, +} +``` + +- Actions run synchronously in the watcher's goroutine. +- A panicking action is recovered and logged; subsequent alerts in the same tick continue. +- A long-running action blocks the next tick of *this* watcher only, not others. + +## API + +```go +// Register a watcher (call before Start, typically in init). +monitor.Register(w Watcher) + +// Start the monitor (called by engine). +monitor.Start(ctx context.Context) error + +// Stop the monitor (called by engine). +monitor.Stop() error + +// Subscribe to alert notifications. Returns a subscription ID. +// Non-blocking: full channels are skipped. +monitor.Subscribe(ch chan<- *monitor.Alert) string + +// Unsubscribe by ID. +monitor.Unsubscribe(id string) + +// Health returns runtime status of the monitor and all watchers. +monitor.Health() HealthStatus +``` + +## Health Check + +```go +status := monitor.Health() +// status.Running — is the monitor running? +// status.Watchers — per-watcher stats: +// .Name — watcher name +// .Interval — check frequency +// .LastTick — when the last tick completed +// .LastAlerts — alert count from the last tick +// .TotalTicks — total ticks since start +// .Panics — total panics caught +``` + +A watcher is considered healthy if `LastTick` is within `Interval × 3` of the current time. + +## Logging + +Monitor writes to `logs/monitor.log` (independent from `application.log`): + +- **Lifecycle events** (Info): monitor started/stopped, watcher started/stopped +- **Warn/Error alerts**: always written with watcher name, target, and message +- **Info alerts**: written in both production and development +- **Trace alerts**: written only in development mode (skipped in production) +- **Panics**: always written at Error level + +Log rotation uses lumberjack (50 MB, 3 backups, 7 days). Format follows `YAO_LOG_MODE` (TEXT or JSON). + +## Panic Safety + +- If `Check()` panics, the watcher recovers and continues on the next tick. +- If `Action()` panics, the watcher recovers and processes remaining alerts. +- Panic counts are tracked in `Health().Watchers[].Panics`. + +## File Structure + +``` +monitor/ +├── DESIGN.md — Architecture and design decisions +├── README.md — This file +├── types.go — Level, Alert, Watcher interface +├── logger.go — Independent slog.Logger → monitor.log +├── service.go — Register, Start, Stop, Subscribe, Health +└── service_test.go +``` diff --git a/monitor/logger.go b/monitor/logger.go new file mode 100644 index 00000000..07c126e0 --- /dev/null +++ b/monitor/logger.go @@ -0,0 +1,60 @@ +package monitor + +import ( + "log/slog" + "os" + "path/filepath" + + "gopkg.in/natefinch/lumberjack.v2" +) + +const slogLevelTrace = slog.Level(-8) + +var logger *slog.Logger + +// initLogger creates the monitor logger. +// appMode controls the minimum log level: +// - "production" → Info (Trace alerts are not written) +// - "development" → Trace (everything is written) +func initLogger(root string, logMode string, appMode string) { + logDir := filepath.Join(root, "logs") + if _, err := os.Stat(logDir); os.IsNotExist(err) { + os.MkdirAll(logDir, 0755) + } + + w := &lumberjack.Logger{ + Filename: filepath.Join(logDir, "monitor.log"), + MaxSize: 50, + MaxBackups: 3, + MaxAge: 7, + LocalTime: true, + } + + minLevel := slog.LevelInfo + if appMode == "development" { + minLevel = slogLevelTrace + } + + opts := &slog.HandlerOptions{Level: minLevel} + var handler slog.Handler + if logMode == "JSON" { + handler = slog.NewJSONHandler(w, opts) + } else { + handler = slog.NewTextHandler(w, opts) + } + + logger = slog.New(handler) +} + +func levelToSlog(l Level) slog.Level { + switch l { + case Trace: + return slogLevelTrace + case Warn: + return slog.LevelWarn + case Error: + return slog.LevelError + default: + return slog.LevelInfo + } +} diff --git a/monitor/service.go b/monitor/service.go new file mode 100644 index 00000000..0243031c --- /dev/null +++ b/monitor/service.go @@ -0,0 +1,250 @@ +package monitor + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/yaoapp/yao/config" +) + +var svc = &monitorService{ + watchers: make(map[string]*watcherEntry), + subs: make(map[string]chan<- *Alert), +} + +type watcherEntry struct { + watcher Watcher + cancel context.CancelFunc + lastTick atomic.Int64 // unix timestamp of last tick completion + lastAlerts atomic.Int64 // alert count from last tick + totalTicks atomic.Int64 // total ticks since start + panics atomic.Int64 // total panics caught +} + +type monitorService struct { + mu sync.Mutex + watchers map[string]*watcherEntry + subs map[string]chan<- *Alert + subSeq int + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + started bool +} + +// Register adds a watcher. Call before Start (typically in init). +// Registering a watcher with the same name replaces the previous one. +func Register(w Watcher) { + svc.mu.Lock() + defer svc.mu.Unlock() + + name := w.Name() + if old, ok := svc.watchers[name]; ok && old.cancel != nil { + old.cancel() + } + svc.watchers[name] = &watcherEntry{watcher: w} + + if svc.started { + svc.startWatcher(svc.watchers[name]) + } +} + +// Start initializes the logger and launches a goroutine per registered watcher. +func Start(ctx context.Context) error { + svc.mu.Lock() + defer svc.mu.Unlock() + + if svc.started { + return fmt.Errorf("monitor: already started") + } + + initLogger(config.Conf.Root, config.Conf.LogMode, config.Conf.Mode) + + svc.ctx, svc.cancel = context.WithCancel(ctx) + for _, entry := range svc.watchers { + svc.startWatcher(entry) + } + svc.started = true + + if logger != nil { + logger.Info("monitor started", "watchers", len(svc.watchers)) + } + return nil +} + +// Stop cancels all watcher goroutines and waits for them to finish. +func Stop() error { + svc.mu.Lock() + if !svc.started { + svc.mu.Unlock() + return nil + } + svc.cancel() + svc.started = false + svc.mu.Unlock() + + svc.wg.Wait() + + if logger != nil { + logger.Info("monitor stopped") + } + return nil +} + +// Subscribe registers a channel to receive alert notifications. +// Returns a subscription ID for unsubscribing. +// Non-blocking: if the channel is full, alerts are dropped for that subscriber. +func Subscribe(ch chan<- *Alert) string { + svc.mu.Lock() + defer svc.mu.Unlock() + + svc.subSeq++ + id := fmt.Sprintf("sub-%d", svc.subSeq) + svc.subs[id] = ch + return id +} + +// Unsubscribe removes a subscription by ID. +func Unsubscribe(id string) { + svc.mu.Lock() + defer svc.mu.Unlock() + delete(svc.subs, id) +} + +// WatcherHealth describes the runtime status of a single watcher. +type WatcherHealth struct { + Name string `json:"name"` + Interval time.Duration `json:"interval"` + LastTick time.Time `json:"last_tick"` // zero if never ticked + LastAlerts int64 `json:"last_alerts"` // alert count from most recent tick + TotalTicks int64 `json:"total_ticks"` + Panics int64 `json:"panics"` +} + +// HealthStatus describes the overall monitor health. +type HealthStatus struct { + Running bool `json:"running"` + Watchers []WatcherHealth `json:"watchers"` +} + +// Health returns the current health status of the monitor service. +func Health() HealthStatus { + svc.mu.Lock() + defer svc.mu.Unlock() + + status := HealthStatus{Running: svc.started} + for _, entry := range svc.watchers { + wh := WatcherHealth{ + Name: entry.watcher.Name(), + Interval: entry.watcher.Interval(), + LastAlerts: entry.lastAlerts.Load(), + TotalTicks: entry.totalTicks.Load(), + Panics: entry.panics.Load(), + } + if ts := entry.lastTick.Load(); ts > 0 { + wh.LastTick = time.Unix(ts, 0) + } + status.Watchers = append(status.Watchers, wh) + } + return status +} + +func (s *monitorService) startWatcher(entry *watcherEntry) { + ctx, cancel := context.WithCancel(s.ctx) + entry.cancel = cancel + s.wg.Add(1) + go s.runLoop(ctx, entry) +} + +func (s *monitorService) runLoop(ctx context.Context, entry *watcherEntry) { + defer s.wg.Done() + + w := entry.watcher + name := w.Name() + interval := w.Interval() + + if logger != nil { + logger.Info("watcher started", "watcher", name, "interval", interval) + } + + // Run first check immediately, then on ticker. + s.tick(ctx, entry) + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + s.tick(ctx, entry) + case <-ctx.Done(): + if logger != nil { + logger.Info("watcher stopped", "watcher", name) + } + return + } + } +} + +func (s *monitorService) tick(ctx context.Context, entry *watcherEntry) { + name := entry.watcher.Name() + + defer func() { + if r := recover(); r != nil { + entry.panics.Add(1) + if logger != nil { + logger.Error("watcher panic", "watcher", name, "recover", fmt.Sprintf("%v", r)) + } + } + entry.totalTicks.Add(1) + entry.lastTick.Store(time.Now().Unix()) + }() + + alerts := entry.watcher.Check(ctx) + entry.lastAlerts.Store(int64(len(alerts))) + + for i := range alerts { + a := &alerts[i] + a.Watcher = name + + // Log level filtering is handled by slog handler: + // production → Info and above (Trace skipped) + // development → Trace and above (everything) + if logger != nil { + logger.Log(ctx, levelToSlog(a.Level), a.Message, + "watcher", name, "target", a.Target) + } + + if a.Action != nil { + s.execAction(ctx, name, a) + } + + s.notify(a) + } +} + +func (s *monitorService) execAction(ctx context.Context, watcherName string, a *Alert) { + defer func() { + if r := recover(); r != nil { + if logger != nil { + logger.Error("action panic", "watcher", watcherName, "target", a.Target, "recover", fmt.Sprintf("%v", r)) + } + } + }() + a.Action(ctx) +} + +func (s *monitorService) notify(a *Alert) { + s.mu.Lock() + defer s.mu.Unlock() + + for _, ch := range s.subs { + select { + case ch <- a: + default: + } + } +} diff --git a/monitor/service_test.go b/monitor/service_test.go new file mode 100644 index 00000000..1aab3b06 --- /dev/null +++ b/monitor/service_test.go @@ -0,0 +1,507 @@ +package monitor + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" +) + +// resetService resets the global service for test isolation. +func resetService() { + svc.mu.Lock() + defer svc.mu.Unlock() + + if svc.started && svc.cancel != nil { + svc.cancel() + svc.wg.Wait() + } + + svc.watchers = make(map[string]*watcherEntry) + svc.subs = make(map[string]chan<- *Alert) + svc.subSeq = 0 + svc.started = false + svc.ctx = nil + svc.cancel = nil + + // Use a discard logger for tests + logger = nil +} + +// testWatcher is a simple watcher for testing. +type testWatcher struct { + name string + interval time.Duration + checkFn func(ctx context.Context) []Alert +} + +func (w *testWatcher) Name() string { return w.name } +func (w *testWatcher) Interval() time.Duration { return w.interval } +func (w *testWatcher) Check(ctx context.Context) []Alert { + if w.checkFn != nil { + return w.checkFn(ctx) + } + return nil +} + +func TestRegisterAndStart(t *testing.T) { + resetService() + defer resetService() + + var count atomic.Int32 + Register(&testWatcher{ + name: "test-basic", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + count.Add(1) + return nil + }, + }) + + err := Start(context.Background()) + if err != nil { + t.Fatalf("Start: %v", err) + } + + // Wait for a few ticks (first immediate + ticker) + time.Sleep(200 * time.Millisecond) + + if err := Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + + c := count.Load() + if c < 2 { + t.Errorf("expected at least 2 checks (immediate + ticker), got %d", c) + } +} + +func TestDoubleStartError(t *testing.T) { + resetService() + defer resetService() + + Register(&testWatcher{name: "dummy", interval: time.Second}) + + if err := Start(context.Background()); err != nil { + t.Fatal(err) + } + + if err := Start(context.Background()); err == nil { + t.Error("expected error on double Start") + } + + Stop() +} + +func TestStopWithoutStart(t *testing.T) { + resetService() + if err := Stop(); err != nil { + t.Errorf("Stop without Start should not error: %v", err) + } +} + +func TestAlertWatcherName(t *testing.T) { + resetService() + defer resetService() + + var got string + var mu sync.Mutex + + Register(&testWatcher{ + name: "namer", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + mu.Lock() + defer mu.Unlock() + if got != "" { + return nil + } + return []Alert{{ + Level: Info, + Target: "test:1", + Message: "hello", + }} + }, + }) + + ch := make(chan *Alert, 8) + subID := Subscribe(ch) + defer Unsubscribe(subID) + + Start(context.Background()) + defer Stop() + + select { + case a := <-ch: + if a.Watcher != "namer" { + t.Errorf("expected Watcher='namer', got %q", a.Watcher) + } + mu.Lock() + got = a.Watcher + mu.Unlock() + case <-time.After(time.Second): + t.Fatal("timeout waiting for alert") + } +} + +func TestAlertAction(t *testing.T) { + resetService() + defer resetService() + + var acted atomic.Bool + + Register(&testWatcher{ + name: "actor", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + if acted.Load() { + return nil + } + return []Alert{{ + Level: Warn, + Target: "test:action", + Message: "do something", + Action: func(ctx context.Context) { + acted.Store(true) + }, + }} + }, + }) + + Start(context.Background()) + defer Stop() + + time.Sleep(200 * time.Millisecond) + + if !acted.Load() { + t.Error("action was not executed") + } +} + +func TestPanicRecovery_Check(t *testing.T) { + resetService() + defer resetService() + + var count atomic.Int32 + + Register(&testWatcher{ + name: "panicker", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + n := count.Add(1) + if n == 1 { + panic("boom") + } + return nil + }, + }) + + Start(context.Background()) + time.Sleep(200 * time.Millisecond) + Stop() + + c := count.Load() + if c < 2 { + t.Errorf("expected watcher to continue after panic, got %d checks", c) + } +} + +func TestPanicRecovery_Action(t *testing.T) { + resetService() + defer resetService() + + var postPanic atomic.Bool + + Register(&testWatcher{ + name: "action-panicker", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + return []Alert{ + { + Level: Error, + Target: "test:panic-action", + Message: "will panic", + Action: func(ctx context.Context) { + if !postPanic.Load() { + panic("action boom") + } + }, + }, + } + }, + }) + + Start(context.Background()) + time.Sleep(150 * time.Millisecond) + postPanic.Store(true) + time.Sleep(100 * time.Millisecond) + Stop() +} + +func TestSubscribeUnsubscribe(t *testing.T) { + resetService() + defer resetService() + + ch := make(chan *Alert, 16) + id := Subscribe(ch) + + Register(&testWatcher{ + name: "sub-test", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + return []Alert{{Level: Info, Target: "t", Message: "msg"}} + }, + }) + + Start(context.Background()) + time.Sleep(100 * time.Millisecond) + Unsubscribe(id) + time.Sleep(100 * time.Millisecond) + Stop() + + // Drain and count + close(ch) + count := 0 + for range ch { + count++ + } + if count == 0 { + t.Error("expected at least one alert before unsubscribe") + } +} + +func TestSubscribeFullChanDrops(t *testing.T) { + resetService() + defer resetService() + + ch := make(chan *Alert, 1) // tiny buffer + Subscribe(ch) + + Register(&testWatcher{ + name: "flood", + interval: 10 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + return []Alert{{Level: Info, Target: "t", Message: "flood"}} + }, + }) + + Start(context.Background()) + time.Sleep(200 * time.Millisecond) + Stop() + // No deadlock = pass +} + +func TestRegisterOverwrite(t *testing.T) { + resetService() + defer resetService() + + var first, second atomic.Int32 + + Register(&testWatcher{ + name: "dup", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + first.Add(1) + return nil + }, + }) + + Register(&testWatcher{ + name: "dup", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + second.Add(1) + return nil + }, + }) + + Start(context.Background()) + time.Sleep(200 * time.Millisecond) + Stop() + + if first.Load() > 0 { + t.Error("first watcher should have been replaced") + } + if second.Load() == 0 { + t.Error("second watcher should have run") + } +} + +func TestRegisterAfterStart(t *testing.T) { + resetService() + defer resetService() + + Start(context.Background()) + defer Stop() + + var count atomic.Int32 + Register(&testWatcher{ + name: "late", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + count.Add(1) + return nil + }, + }) + + time.Sleep(200 * time.Millisecond) + + if count.Load() == 0 { + t.Error("watcher registered after Start should still run") + } +} + +func TestLevelString(t *testing.T) { + tests := []struct { + level Level + want string + }{ + {Trace, "trace"}, + {Info, "info"}, + {Warn, "warn"}, + {Error, "error"}, + {Level(99), "unknown"}, + } + for _, tt := range tests { + got := tt.level.String() + if got != tt.want { + t.Errorf("Level(%d).String() = %q, want %q", tt.level, got, tt.want) + } + } +} + +func TestContextCancelledDuringCheck(t *testing.T) { + resetService() + defer resetService() + + var checkDone atomic.Bool + + Register(&testWatcher{ + name: "slow", + interval: time.Hour, // only immediate check runs + checkFn: func(ctx context.Context) []Alert { + select { + case <-ctx.Done(): + checkDone.Store(true) + case <-time.After(2 * time.Second): + } + return nil + }, + }) + + Start(context.Background()) + + // Give the immediate check time to start + time.Sleep(50 * time.Millisecond) + + // Stop should cancel the context + done := make(chan struct{}) + go func() { + Stop() + close(done) + }() + + select { + case <-done: + if !checkDone.Load() { + // The check might have not started yet, that's ok + fmt.Println("note: check may not have started before stop") + } + case <-time.After(3 * time.Second): + t.Fatal("Stop timed out — watcher may not respect context cancellation") + } +} + +func TestHealth_NotStarted(t *testing.T) { + resetService() + defer resetService() + + Register(&testWatcher{name: "idle", interval: time.Second}) + + h := Health() + if h.Running { + t.Error("expected Running=false before Start") + } + if len(h.Watchers) != 1 { + t.Errorf("expected 1 watcher, got %d", len(h.Watchers)) + } + if h.Watchers[0].TotalTicks != 0 { + t.Errorf("expected 0 ticks before Start, got %d", h.Watchers[0].TotalTicks) + } +} + +func TestHealth_Running(t *testing.T) { + resetService() + defer resetService() + + Register(&testWatcher{ + name: "healthy", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + return []Alert{{Level: Info, Target: "t:1", Message: "ok"}} + }, + }) + + Start(context.Background()) + time.Sleep(200 * time.Millisecond) + + h := Health() + if !h.Running { + t.Error("expected Running=true") + } + if len(h.Watchers) != 1 { + t.Fatalf("expected 1 watcher, got %d", len(h.Watchers)) + } + + wh := h.Watchers[0] + if wh.Name != "healthy" { + t.Errorf("expected name 'healthy', got %q", wh.Name) + } + if wh.TotalTicks < 2 { + t.Errorf("expected at least 2 ticks, got %d", wh.TotalTicks) + } + if wh.LastTick.IsZero() { + t.Error("expected non-zero LastTick") + } + if wh.LastAlerts != 1 { + t.Errorf("expected 1 alert per tick, got %d", wh.LastAlerts) + } + if wh.Panics != 0 { + t.Errorf("expected 0 panics, got %d", wh.Panics) + } + + Stop() +} + +func TestHealth_PanicCount(t *testing.T) { + resetService() + defer resetService() + + var n atomic.Int32 + Register(&testWatcher{ + name: "crasher", + interval: 50 * time.Millisecond, + checkFn: func(ctx context.Context) []Alert { + if n.Add(1) <= 2 { + panic("crash") + } + return nil + }, + }) + + Start(context.Background()) + time.Sleep(250 * time.Millisecond) + + h := Health() + wh := h.Watchers[0] + if wh.Panics < 2 { + t.Errorf("expected at least 2 panics, got %d", wh.Panics) + } + if wh.TotalTicks <= wh.Panics { + t.Errorf("expected some successful ticks after panics: total=%d, panics=%d", wh.TotalTicks, wh.Panics) + } + + Stop() +} diff --git a/monitor/types.go b/monitor/types.go new file mode 100644 index 00000000..faa62f00 --- /dev/null +++ b/monitor/types.go @@ -0,0 +1,55 @@ +package monitor + +import ( + "context" + "time" +) + +// Level represents alert severity. +type Level int + +const ( + Trace Level = iota // Heartbeat, periodic status sync — not logged + Info // Notable events: state changes, registrations + Warn // Needs attention: idle timeout, degraded state + Error // Needs immediate action: crash, unreachable +) + +func (l Level) String() string { + switch l { + case Trace: + return "trace" + case Info: + return "info" + case Warn: + return "warn" + case Error: + return "error" + default: + return "unknown" + } +} + +// Alert represents a single finding from a watcher check. +type Alert struct { + Watcher string // Source watcher name (set by monitor) + Level Level // Severity + Target string // Target identifier, e.g. "box:abc123", "robot:member-456" + Message string // Human-readable description + Action func(ctx context.Context) // Business-layer action; nil means notification only +} + +// Watcher is the interface that business modules implement and register +// with the monitor service. +type Watcher interface { + // Name returns a globally unique watcher name, used for logging and dedup. + Name() string + + // Interval returns the check frequency. + Interval() time.Duration + + // Check performs a single inspection and returns any alerts found. + // An empty slice means everything is normal. + // ctx is cancelled when the monitor stops. + Check(ctx context.Context) []Alert +} diff --git a/service/log/access.go b/service/log/access.go new file mode 100644 index 00000000..495424af --- /dev/null +++ b/service/log/access.go @@ -0,0 +1,96 @@ +package log + +import ( + "fmt" + "os" + "path/filepath" + "time" + + "github.com/gin-gonic/gin" + "gopkg.in/natefinch/lumberjack.v2" +) + +var ( + accessWriter *lumberjack.Logger + accessErrorWriter *lumberjack.Logger +) + +// InitAccessLog initializes access log and access-error log writers. +// Must be called before any HTTP request is served. +func InitAccessLog(root string) { + logDir := filepath.Join(root, "logs") + if _, err := os.Stat(logDir); os.IsNotExist(err) { + os.MkdirAll(logDir, 0755) + } + + accessWriter = &lumberjack.Logger{ + Filename: filepath.Join(logDir, "access.log"), + MaxSize: 100, + MaxBackups: 5, + MaxAge: 30, + LocalTime: true, + } + accessErrorWriter = &lumberjack.Logger{ + Filename: filepath.Join(logDir, "access-error.log"), + MaxSize: 50, + MaxBackups: 5, + MaxAge: 30, + LocalTime: true, + } +} + +// AccessLog returns a gin middleware that writes NGINX Combined Log Format +// to access.log (all requests) and access-error.log (4xx/5xx only). +func AccessLog() gin.HandlerFunc { + return func(c *gin.Context) { + c.Next() + + if accessWriter == nil { + return + } + + status := c.Writer.Status() + size := c.Writer.Size() + if size < 0 { + size = 0 + } + + line := fmt.Sprintf("%s - %s [%s] \"%s %s %s\" %d %d \"%s\" \"%s\"\n", + c.ClientIP(), + remoteUser(c), + time.Now().Format("02/Jan/2006:15:04:05 -0700"), + c.Request.Method, + c.Request.RequestURI, + c.Request.Proto, + status, + size, + dash(c.Request.Referer()), + dash(c.Request.UserAgent()), + ) + + accessWriter.Write([]byte(line)) + if status >= 400 { + accessErrorWriter.Write([]byte(line)) + } + } +} + +// remoteUser extracts a user identifier from the gin context. +// Tries __username (JWT), __user_id (OAuth), __sid (SUI session) in order. +func remoteUser(c *gin.Context) string { + for _, key := range []string{"__username", "__user_id", "__sid"} { + if v, ok := c.Get(key); ok { + if s, ok := v.(string); ok && s != "" { + return s + } + } + } + return "-" +} + +func dash(s string) string { + if s == "" { + return "-" + } + return s +} diff --git a/service/log/access_test.go b/service/log/access_test.go new file mode 100644 index 00000000..bc132728 --- /dev/null +++ b/service/log/access_test.go @@ -0,0 +1,214 @@ +package log + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "regexp" + "strings" + "testing" + + "github.com/gin-gonic/gin" +) + +func init() { + gin.SetMode(gin.TestMode) +} + +func setupTestLog(t *testing.T) (string, func()) { + t.Helper() + dir := t.TempDir() + InitAccessLog(dir) + return dir, func() { + if accessWriter != nil { + accessWriter.Close() + } + if accessErrorWriter != nil { + accessErrorWriter.Close() + } + accessWriter = nil + accessErrorWriter = nil + } +} + +// NGINX Combined: $remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent" +var nginxCombinedRe = regexp.MustCompile( + `^(\S+) - (\S+) \[\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4}\] "(\S+) (\S+) (\S+)" (\d{3}) (\d+) "(.*)" "(.*)"$`, +) + +func TestAccessLog_NginxFormat(t *testing.T) { + dir, cleanup := setupTestLog(t) + defer cleanup() + + router := gin.New() + router.Use(AccessLog()) + router.GET("/api/test", func(c *gin.Context) { + c.String(http.StatusOK, "ok") + }) + + req := httptest.NewRequest("GET", "/api/test", nil) + req.Header.Set("User-Agent", "TestAgent/1.0") + req.Header.Set("Referer", "https://example.com") + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != 200 { + t.Fatalf("expected 200, got %d", w.Code) + } + + data, err := os.ReadFile(filepath.Join(dir, "logs", "access.log")) + if err != nil { + t.Fatalf("read access.log: %v", err) + } + + line := strings.TrimSpace(string(data)) + if !nginxCombinedRe.MatchString(line) { + t.Errorf("access.log line does not match NGINX Combined format:\n%s", line) + } + + if !strings.Contains(line, `"GET /api/test HTTP/1.1"`) { + t.Errorf("expected request line in log, got: %s", line) + } + if !strings.Contains(line, `"https://example.com"`) { + t.Errorf("expected referer in log, got: %s", line) + } + if !strings.Contains(line, `"TestAgent/1.0"`) { + t.Errorf("expected user-agent in log, got: %s", line) + } + + // access-error.log should be empty for 200 + errData, err := os.ReadFile(filepath.Join(dir, "logs", "access-error.log")) + if err != nil && !os.IsNotExist(err) { + t.Fatalf("read access-error.log: %v", err) + } + if len(strings.TrimSpace(string(errData))) > 0 { + t.Errorf("access-error.log should be empty for 200, got: %s", string(errData)) + } +} + +func TestAccessLog_ErrorDoubleWrite(t *testing.T) { + dir, cleanup := setupTestLog(t) + defer cleanup() + + router := gin.New() + router.Use(AccessLog()) + router.GET("/api/fail", func(c *gin.Context) { + c.String(http.StatusInternalServerError, "error") + }) + router.GET("/api/notfound", func(c *gin.Context) { + c.String(http.StatusNotFound, "not found") + }) + + // 500 request + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/fail", nil)) + + // 404 request + w = httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/notfound", nil)) + + // 200 request (should NOT appear in error log) + router.GET("/api/ok", func(c *gin.Context) { + c.String(http.StatusOK, "ok") + }) + w = httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/ok", nil)) + + accessData, _ := os.ReadFile(filepath.Join(dir, "logs", "access.log")) + accessLines := nonEmptyLines(string(accessData)) + if len(accessLines) != 3 { + t.Fatalf("access.log: expected 3 lines, got %d:\n%s", len(accessLines), string(accessData)) + } + + errData, _ := os.ReadFile(filepath.Join(dir, "logs", "access-error.log")) + errLines := nonEmptyLines(string(errData)) + if len(errLines) != 2 { + t.Fatalf("access-error.log: expected 2 lines (500+404), got %d:\n%s", len(errLines), string(errData)) + } + + if !strings.Contains(errLines[0], "500") { + t.Errorf("first error line should contain 500: %s", errLines[0]) + } + if !strings.Contains(errLines[1], "404") { + t.Errorf("second error line should contain 404: %s", errLines[1]) + } +} + +func TestAccessLog_RemoteUser(t *testing.T) { + dir, cleanup := setupTestLog(t) + defer cleanup() + + router := gin.New() + router.Use(AccessLog()) + router.GET("/api/user", func(c *gin.Context) { + c.Set("__username", "alice") + c.String(http.StatusOK, "ok") + }) + router.GET("/api/userid", func(c *gin.Context) { + c.Set("__user_id", "uid-123") + c.String(http.StatusOK, "ok") + }) + router.GET("/api/anon", func(c *gin.Context) { + c.String(http.StatusOK, "ok") + }) + + for _, path := range []string{"/api/user", "/api/userid", "/api/anon"} { + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", path, nil)) + } + + data, _ := os.ReadFile(filepath.Join(dir, "logs", "access.log")) + lines := nonEmptyLines(string(data)) + if len(lines) != 3 { + t.Fatalf("expected 3 lines, got %d", len(lines)) + } + + // Note: AccessLog middleware runs c.Next() first, then reads context. + // The user keys are set inside the handler which runs during c.Next(), + // so they should be available when the log line is written. + if !strings.Contains(lines[0], " alice ") { + t.Errorf("line 1 should have user 'alice': %s", lines[0]) + } + if !strings.Contains(lines[1], " uid-123 ") { + t.Errorf("line 2 should have user 'uid-123': %s", lines[1]) + } + if !strings.Contains(lines[2], " - ") { + t.Errorf("line 3 should have '-' for anonymous: %s", lines[2]) + } +} + +func TestAccessLog_DashForEmpty(t *testing.T) { + dir, cleanup := setupTestLog(t) + defer cleanup() + + router := gin.New() + router.Use(AccessLog()) + router.GET("/api/test", func(c *gin.Context) { + c.String(http.StatusOK, "ok") + }) + + req := httptest.NewRequest("GET", "/api/test", nil) + // No Referer, no User-Agent + req.Header.Del("User-Agent") + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + data, _ := os.ReadFile(filepath.Join(dir, "logs", "access.log")) + line := strings.TrimSpace(string(data)) + + // Should end with "-" "-" for empty referer and user-agent + if !strings.HasSuffix(line, `"-" "-"`) { + t.Errorf("expected dash for empty referer/ua, got: %s", line) + } +} + +func nonEmptyLines(s string) []string { + var result []string + for _, line := range strings.Split(s, "\n") { + if strings.TrimSpace(line) != "" { + result = append(result, line) + } + } + return result +} diff --git a/service/middleware.go b/service/middleware.go index 6589c92c..aa57e031 100644 --- a/service/middleware.go +++ b/service/middleware.go @@ -11,13 +11,14 @@ import ( "github.com/gin-gonic/gin" "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/openapi" + servicelog "github.com/yaoapp/yao/service/log" "github.com/yaoapp/yao/share" "github.com/yaoapp/yao/sui/api" ) // Middlewares the middlewares var Middlewares = []gin.HandlerFunc{ - gin.Logger(), + servicelog.AccessLog(), withStaticFileServer, } diff --git a/service/service.go b/service/service.go index 74c2bee2..c84994f3 100644 --- a/service/service.go +++ b/service/service.go @@ -9,6 +9,7 @@ import ( "github.com/yaoapp/gou/server/http" "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/openapi" + servicelog "github.com/yaoapp/yao/service/log" "github.com/yaoapp/yao/share" ) @@ -148,6 +149,8 @@ func Restart(svc *Service, cfg config.Config) error { } func prepare() error { + servicelog.InitAccessLog(config.Conf.Root) + err := share.SessionStart() if err != nil { return err