From 7e3ad7cb23124a48368d41012829ca9ef248078e Mon Sep 17 00:00:00 2001 From: xboard Date: Tue, 24 Mar 2026 23:12:55 +0800 Subject: [PATCH] feat: nlog, config hot-reload, and concurrency fixes --- .gitignore | 1 + cmd/xboard-node/main.go | 181 ++++++++++++++++--------- go.mod | 4 +- go.sum | 2 + internal/cert/cert.go | 24 ++-- internal/config/config.go | 27 ++-- internal/config/watcher.go | 117 ++++++++++++++++ internal/kernel/customcfg.go | 6 +- internal/kernel/geodata/geodata.go | 9 +- internal/kernel/singbox/config.go | 6 +- internal/kernel/singbox/conntracker.go | 32 +++-- internal/kernel/singbox/singbox.go | 22 +-- internal/kernel/xray/config.go | 6 +- internal/kernel/xray/dispatcher.go | 7 +- internal/kernel/xray/xray.go | 45 +++--- internal/limiter/speedtracker.go | 80 ++++++----- internal/monitor/monitor.go | 4 +- internal/nlog/nlog.go | 92 ++++++++++--- internal/panel/client.go | 6 +- internal/panel/ws.go | 42 +++--- internal/service/service.go | 124 +++++++++++------ internal/tracker/tracker.go | 16 +-- 22 files changed, 572 insertions(+), 281 deletions(-) create mode 100644 internal/config/watcher.go diff --git a/.gitignore b/.gitignore index 84ac9c9..584e3ea 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,4 @@ Thumbs.db # Go vendor/ config.json +benchmark/ diff --git a/cmd/xboard-node/main.go b/cmd/xboard-node/main.go index 3aebcf2..113a7f5 100644 --- a/cmd/xboard-node/main.go +++ b/cmd/xboard-node/main.go @@ -4,7 +4,6 @@ import ( "context" "flag" "fmt" - "log/slog" "net" "net/http" "os" @@ -46,36 +45,20 @@ func main() { // Apply runtime memory tuning before anything else allocates. applyRuntimeConfig(cfg.Runtime) - nodes := cfg.ExpandNodes() - nlog.Core().Info(fmt.Sprintf("xboard-node %s starting, %d nodes", version, len(nodes))) + runWithReload(cfg, *configPath) +} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - sigCh := make(chan os.Signal, 1) - signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) - - go func() { - sig := <-sigCh - nlog.Core().Info(fmt.Sprintf("received %v, shutting down...", sig)) - cancel() - - // Force-exit on second signal or after timeout. - select { - case sig = <-sigCh: - slog.Warn("received second signal, forcing exit", "signal", sig) - os.Exit(1) - case <-time.After(15 * time.Second): - slog.Error("shutdown timed out after 15s, forcing exit") - os.Exit(2) +// runWithReload restarts all node services when the config file changes. +func runWithReload(initialCfg *config.Config, configPath string) { + var healthSrv *http.Server + var healthPort int + startHealth := func(port int) { + if port <= 0 { + return } - }() - - // Optional health-check endpoint for container orchestrators. - if cfg.HealthPort > 0 { - ln, err := net.Listen("tcp", fmt.Sprintf(":%d", cfg.HealthPort)) + ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) if err != nil { - slog.Error("failed to start health check listener", "port", cfg.HealthPort, "error", err) + nlog.Core().Error("failed to start health check listener", "port", port, "error", err) os.Exit(1) } mux := http.NewServeMux() @@ -84,44 +67,118 @@ func main() { w.WriteHeader(http.StatusOK) w.Write([]byte(`{"status":"ok"}`)) }) - srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second} + healthSrv = &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second} + healthPort = port go func() { - nlog.Core().Debug(fmt.Sprintf("health check listening on :%d", cfg.HealthPort)) - if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed { - slog.Warn("health check server stopped", "error", err) - } - }() - go func() { - <-ctx.Done() - srv.Close() - }() - } - - // Run one service per node. If any node fails, cancel all others. - errCh := make(chan error, len(nodes)) - var wg sync.WaitGroup - for _, nodeCfg := range nodes { - nodeCfg := nodeCfg // capture - wg.Add(1) - go func() { - defer wg.Done() - svc := service.New(nodeCfg) - if err := svc.Run(ctx); err != nil { - slog.Error("node service exited with error", - "node_id", nodeCfg.Panel.NodeID, "error", err) - errCh <- err - cancel() // bring down all other nodes + nlog.Core().Debug(fmt.Sprintf("health check listening on :%d", port)) + if err := healthSrv.Serve(ln); err != nil && err != http.ErrServerClosed { + nlog.Core().Warn("health check server stopped", "error", err) } }() } - wg.Wait() - close(errCh) + startHealth(initialCfg.HealthPort) + defer func() { + if healthSrv != nil { + healthSrv.Close() + } + }() - if err := <-errCh; err != nil { - os.Exit(1) + for cfg := initialCfg; ; { + ctx, cancel := context.WithCancel(context.Background()) + + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + + reloadCh := make(chan *config.Config, 1) + + watcher, err := config.WatchConfig(ctx, configPath, func(newCfg *config.Config) { + select { + case reloadCh <- newCfg: + default: + } + }) + if err != nil { + nlog.Core().Warn("config watcher unavailable, hot-reload disabled", "error", err) + } + + go func() { + select { + case sig := <-sigCh: + nlog.Core().Info(fmt.Sprintf("received %v, shutting down...", sig)) + cancel() + + select { + case sig = <-sigCh: + nlog.Core().Warn("received second signal, forcing exit", "signal", sig) + os.Exit(1) + case <-time.After(15 * time.Second): + nlog.Core().Error("shutdown timed out after 15s, forcing exit") + os.Exit(2) + } + case <-ctx.Done(): + } + }() + + if cfg.HealthPort != healthPort { + if healthSrv != nil { + healthSrv.Close() + healthSrv = nil + } + startHealth(cfg.HealthPort) + } + + nodes := cfg.ExpandNodes() + nlog.Core().Info(fmt.Sprintf("xboard-node %s starting, %d nodes", version, len(nodes))) + + errCh := make(chan error, len(nodes)) + var wg sync.WaitGroup + for _, nodeCfg := range nodes { + nodeCfg := nodeCfg + wg.Add(1) + go func() { + defer wg.Done() + svc := service.New(nodeCfg) + if err := svc.Run(ctx); err != nil { + nlog.Core().Error("node service exited with error", + "node_id", nodeCfg.Panel.NodeID, "error", err) + errCh <- err + cancel() + } + }() + } + + doneCh := make(chan struct{}) + go func() { wg.Wait(); close(doneCh) }() + + var newCfg *config.Config + select { + case newCfg = <-reloadCh: + nlog.Core().Info("config changed, restarting all services...") + cancel() + <-doneCh + case <-doneCh: + } + + signal.Stop(sigCh) + if watcher != nil { + watcher.Stop() + } + + if newCfg == nil { + close(errCh) + if err := <-errCh; err != nil { + os.Exit(1) + } + nlog.Core().Info("stopped") + return + } + + config.InitLogger(newCfg.Log) + applyRuntimeConfig(newCfg.Runtime) + cfg = newCfg + nlog.Core().Info("reload complete, services restarting with new config") } - nlog.Core().Info("stopped") } // applyRuntimeConfig wires up Go runtime memory limits from the config file. @@ -133,17 +190,17 @@ func applyRuntimeConfig(rt config.RuntimeConfig) { // GOGC if rt.GoGCPercent > 0 { prev := debug.SetGCPercent(rt.GoGCPercent) - slog.Info("runtime: GOGC set", "gogc", rt.GoGCPercent, "prev", prev) + nlog.Core().Info("runtime: GOGC set", "gogc", rt.GoGCPercent, "prev", prev) } // GOMEMLIMIT — parse human-readable size string (e.g. "30MiB") if rt.GoMemLimit != "" { limit, err := parseMemLimit(rt.GoMemLimit) if err != nil { - slog.Warn("runtime: invalid gomemlimit, ignoring", "value", rt.GoMemLimit, "error", err) + nlog.Core().Warn("runtime: invalid gomemlimit, ignoring", "value", rt.GoMemLimit, "error", err) } else { prev := debug.SetMemoryLimit(limit) - slog.Info("runtime: GOMEMLIMIT set", + nlog.Core().Info("runtime: GOMEMLIMIT set", "limit", rt.GoMemLimit, "bytes", limit, "prev_bytes", prev, diff --git a/go.mod b/go.mod index 6955ad2..03712c6 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/sagernet/sing-box v1.13.2 github.com/shirou/gopsutil/v4 v4.26.2 github.com/xtls/xray-core v1.260206.0 + golang.org/x/term v0.40.0 golang.org/x/time v0.12.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -36,7 +37,7 @@ require ( github.com/ebitengine/purego v0.10.0 // indirect github.com/enfein/mieru/v3 v3.29.0 // indirect github.com/florianl/go-nfqueue/v2 v2.0.2 // indirect - github.com/fsnotify/fsnotify v1.7.0 // indirect + github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/fxamacker/cbor/v2 v2.7.0 // indirect github.com/gaissmai/bart v0.18.0 // indirect github.com/ghodss/yaml v1.0.1-0.20220118164431-d8423dcdf344 // indirect @@ -161,7 +162,6 @@ require ( golang.org/x/oauth2 v0.34.0 // indirect golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.41.0 // indirect - golang.org/x/term v0.40.0 // indirect golang.org/x/text v0.34.0 // indirect golang.org/x/tools v0.42.0 // indirect golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect diff --git a/go.sum b/go.sum index 77d0091..6718f1d 100644 --- a/go.sum +++ b/go.sum @@ -53,6 +53,8 @@ github.com/florianl/go-nfqueue/v2 v2.0.2 h1:FL5lQTeetgpCvac1TRwSfgaXUn0YSO7WzGvW github.com/florianl/go-nfqueue/v2 v2.0.2/go.mod h1:VA09+iPOT43OMoCKNfXHyzujQUty2xmzyCRkBOlmabc= github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA= github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM= +github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= +github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/fxamacker/cbor/v2 v2.7.0 h1:iM5WgngdRBanHcxugY4JySA0nk1wZorNOpTgCMedv5E= github.com/fxamacker/cbor/v2 v2.7.0/go.mod h1:pxXPTn3joSm21Gbwsv0w9OSA2y1HFR9qXEeXQVeNoDQ= github.com/gaissmai/bart v0.18.0 h1:jQLBT/RduJu0pv/tLwXE+xKPgtWJejbxuXAR+wLJafo= diff --git a/internal/cert/cert.go b/internal/cert/cert.go index 0e67e82..e449e92 100644 --- a/internal/cert/cert.go +++ b/internal/cert/cert.go @@ -9,7 +9,6 @@ import ( "crypto/x509/pkix" "encoding/pem" "fmt" - "log/slog" "math/big" "net" "os" @@ -23,6 +22,7 @@ import ( "github.com/libdns/cloudflare" "github.com/cedar2025/xboard-node/internal/config" + "github.com/cedar2025/xboard-node/internal/nlog" ) // Manager handles TLS certificate lifecycle. @@ -171,7 +171,7 @@ func (m *Manager) startFile() error { if _, err := os.Stat(m.cfg.KeyFile); err != nil { return fmt.Errorf("key file: %w", err) } - slog.Debug("using manual TLS certificates", "cert", m.certFile, "key", m.keyFile) + nlog.Core().Debug("using manual TLS certificates", "cert", m.certFile, "key", m.keyFile) return nil } @@ -180,7 +180,7 @@ func (m *Manager) startFile() error { func (m *Manager) startSelfSigned() error { // If files already exist and are valid, skip regeneration. if fileExists(m.certFile) && fileExists(m.keyFile) { - slog.Debug("self-signed cert exists, reusing", "cert", m.certFile) + nlog.Core().Debug("self-signed cert exists, reusing", "cert", m.certFile) return nil } @@ -238,7 +238,7 @@ func (m *Manager) startSelfSigned() error { return fmt.Errorf("write key: %w", err) } - slog.Info("self-signed certificate generated", + nlog.Core().Info("self-signed certificate generated", "domain", domain, "cert", m.certFile, "valid_years", 10) return nil } @@ -260,7 +260,7 @@ func (m *Manager) startContent() error { if err := atomicWriteFile(m.keyFile, []byte(m.cfg.KeyContent), 0o600); err != nil { return fmt.Errorf("write key content: %w", err) } - slog.Info("TLS certificate written from panel content", "cert", m.certFile) + nlog.Core().Info("TLS certificate written from panel content", "cert", m.certFile) return nil } @@ -270,7 +270,7 @@ func (m *Manager) startACME(ctx context.Context, dnsSolver *certmagic.DNS01Solve // Guard: certmagic's background goroutines must only be started once. // A process restart is required to change ACME settings at runtime. if m.acmeStarted { - slog.Info("cert: ACME already active, skipping re-init (restart to change ACME settings)") + nlog.Core().Info("cert: ACME already active, skipping re-init (restart to change ACME settings)") return nil } @@ -389,32 +389,32 @@ func (m *Manager) syncCertFiles(ctx context.Context, storage certmagic.Storage, certKey, _ := data["certificate_path"].(string) keyKey, _ := data["private_key_path"].(string) if certKey == "" || keyKey == "" { - slog.Warn("cert_obtained event missing storage paths", "data", data) + nlog.Core().Warn("cert_obtained event missing storage paths", "data", data) return } certPEM, err := storage.Load(ctx, certKey) if err != nil { - slog.Error("failed to load cert from storage", "error", err) + nlog.Core().Error("failed to load cert from storage", "error", err) return } keyPEM, err := storage.Load(ctx, keyKey) if err != nil { - slog.Error("failed to load key from storage", "error", err) + nlog.Core().Error("failed to load key from storage", "error", err) return } if err := atomicWriteFile(m.certFile, certPEM, 0o644); err != nil { - slog.Error("failed to write cert file", "path", m.certFile, "error", err) + nlog.Core().Error("failed to write cert file", "path", m.certFile, "error", err) return } if err := atomicWriteFile(m.keyFile, keyPEM, 0o600); err != nil { - slog.Error("failed to write key file", "path", m.keyFile, "error", err) + nlog.Core().Error("failed to write key file", "path", m.keyFile, "error", err) return } renewal, _ := data["renewal"].(bool) - slog.Info("TLS certificate synced", "domain", m.cfg.Domain, "renewal", renewal, + nlog.Core().Info("TLS certificate synced", "domain", m.cfg.Domain, "renewal", renewal, "cert", m.certFile) m.renewed.Store(true) } diff --git a/internal/config/config.go b/internal/config/config.go index 1859a50..67d0225 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -8,6 +8,8 @@ import ( "path/filepath" "strconv" + "github.com/cedar2025/xboard-node/internal/nlog" + "golang.org/x/term" "gopkg.in/yaml.v3" ) @@ -331,40 +333,47 @@ func (c *Config) ExpandNodes() []*Config { } func InitLogger(cfg LogConfig) { - var level slog.Level + var minLevel slog.Level switch cfg.Level { case "debug": - level = slog.LevelDebug + minLevel = slog.LevelDebug case "warn": - level = slog.LevelWarn + minLevel = slog.LevelWarn case "error": - level = slog.LevelError + minLevel = slog.LevelError default: - level = slog.LevelInfo + minLevel = slog.LevelInfo } var w io.Writer + useColor := false switch cfg.Output { case "stdout", "": w = os.Stdout + useColor = term.IsTerminal(int(os.Stdout.Fd())) case "stderr": w = os.Stderr + useColor = term.IsTerminal(int(os.Stderr.Fd())) default: dir := filepath.Dir(cfg.Output) if err := os.MkdirAll(dir, 0o755); err != nil { - slog.Error("failed to create log dir, falling back to stdout", "error", err) + fmt.Fprintf(os.Stderr, "failed to create log dir, falling back to stdout: %v\n", err) w = os.Stdout + useColor = term.IsTerminal(int(os.Stdout.Fd())) } else { f, err := os.OpenFile(cfg.Output, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) if err != nil { - slog.Error("failed to open log file, falling back to stdout", "error", err) + fmt.Fprintf(os.Stderr, "failed to open log file, falling back to stdout: %v\n", err) w = os.Stdout + useColor = term.IsTerminal(int(os.Stdout.Fd())) } else { w = f + useColor = false } } } - handler := slog.NewTextHandler(w, &slog.HandlerOptions{Level: level}) - slog.SetDefault(slog.New(handler)) + nlog.Init(w, minLevel, useColor) + // Application logging goes through nlog; silence slog.Default for stray library use. + slog.SetDefault(slog.New(slog.DiscardHandler)) } diff --git a/internal/config/watcher.go b/internal/config/watcher.go new file mode 100644 index 0000000..b6a06f4 --- /dev/null +++ b/internal/config/watcher.go @@ -0,0 +1,117 @@ +package config + +import ( + "context" + "path/filepath" + "sync" + "sync/atomic" + "time" + + "github.com/cedar2025/xboard-node/internal/nlog" + "github.com/fsnotify/fsnotify" +) + +// Watcher reloads the config file on change (debounced) and calls onChange. +type Watcher struct { + path string + debounce time.Duration + onChange func(*Config) + watcher *fsnotify.Watcher + + stopOnce sync.Once + stopCh chan struct{} + stopped atomic.Bool +} + +// WatchConfig watches path; invalid reloads are logged and ignored. +// Stop via Stop() or cancel ctx. +func WatchConfig(ctx context.Context, path string, onChange func(*Config)) (*Watcher, error) { + absPath, err := filepath.Abs(path) + if err != nil { + return nil, err + } + + fsw, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + + // Watch parent dir so rename/atomic saves are visible. + dir := filepath.Dir(absPath) + if err := fsw.Add(dir); err != nil { + fsw.Close() + return nil, err + } + + w := &Watcher{ + path: absPath, + debounce: 1 * time.Second, + onChange: onChange, + watcher: fsw, + stopCh: make(chan struct{}), + } + + go w.loop(ctx) + nlog.Core().Info("config watcher started", "path", absPath) + return w, nil +} + +func (w *Watcher) loop(ctx context.Context) { + var timer *time.Timer + defer func() { + if timer != nil { + timer.Stop() + } + w.watcher.Close() + }() + + for { + select { + case <-ctx.Done(): + return + case <-w.stopCh: + return + + case ev, ok := <-w.watcher.Events: + if !ok { + return + } + if filepath.Clean(ev.Name) != w.path { + continue + } + if !ev.Has(fsnotify.Write) && !ev.Has(fsnotify.Create) { + continue + } + + if timer == nil { + timer = time.AfterFunc(w.debounce, w.reload) + } else { + timer.Reset(w.debounce) + } + + case err, ok := <-w.watcher.Errors: + if !ok { + return + } + nlog.Core().Warn("config watcher error", "error", err) + } + } +} + +func (w *Watcher) reload() { + if w.stopped.Load() { + return + } + cfg, err := Load(w.path) + if err != nil { + nlog.Core().Error("config reload failed, keeping current config", "error", err) + return + } + nlog.Core().Info("config reloaded successfully") + w.onChange(cfg) +} + +func (w *Watcher) Stop() { + w.stopped.Store(true) + w.stopOnce.Do(func() { close(w.stopCh) }) +} diff --git a/internal/kernel/customcfg.go b/internal/kernel/customcfg.go index 8826060..2d1854f 100644 --- a/internal/kernel/customcfg.go +++ b/internal/kernel/customcfg.go @@ -3,10 +3,10 @@ package kernel import ( "encoding/json" "fmt" - "log/slog" "os" "strings" + "github.com/cedar2025/xboard-node/internal/nlog" "gopkg.in/yaml.v3" ) @@ -20,7 +20,7 @@ func LoadCustomConfig(path string) (map[string]any, error) { data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { - slog.Warn("custom config file not found, skipping", "path", path) + nlog.Core().Warn("custom config file not found, skipping", "path", path) return nil, nil } return nil, fmt.Errorf("read custom config %s: %w", path, err) @@ -43,7 +43,7 @@ func LoadCustomConfig(path string) (map[string]any, error) { } } - slog.Info("loaded custom config", "path", path, "keys", mapKeys(result)) + nlog.Core().Info("loaded custom config", "path", path, "keys", mapKeys(result)) return result, nil } diff --git a/internal/kernel/geodata/geodata.go b/internal/kernel/geodata/geodata.go index 31130d5..f9ad3b1 100644 --- a/internal/kernel/geodata/geodata.go +++ b/internal/kernel/geodata/geodata.go @@ -3,11 +3,12 @@ package geodata import ( "fmt" "io" - "log/slog" "net/http" "os" "path/filepath" "time" + + "github.com/cedar2025/xboard-node/internal/nlog" ) const ( @@ -49,7 +50,7 @@ func Ensure(dir string, needGeoIP, needGeoSite bool, kernelType string) error { var firstErr error for _, f := range files { if err := ensureFile(dir, f.name, f.url); err != nil { - slog.Warn("geo database auto-download failed", "file", f.name, "error", err) + nlog.Core().Warn("geo database auto-download failed", "file", f.name, "error", err) if firstErr == nil { firstErr = err } @@ -63,12 +64,12 @@ func ensureFile(dir, name, url string) error { if _, err := os.Stat(dst); err == nil { return nil } - slog.Info("geo database missing, downloading automatically", "file", name, "url", url) + nlog.Core().Info("geo database missing, downloading automatically", "file", name, "url", url) if err := atomicDownload(dst, url); err != nil { return err } if fi, err := os.Stat(dst); err == nil { - slog.Info("geo database ready", "file", name, "size_kb", fi.Size()/1024) + nlog.Core().Info("geo database ready", "file", name, "size_kb", fi.Size()/1024) } return nil } diff --git a/internal/kernel/singbox/config.go b/internal/kernel/singbox/config.go index b5b176d..cd62b86 100644 --- a/internal/kernel/singbox/config.go +++ b/internal/kernel/singbox/config.go @@ -3,7 +3,6 @@ package singbox import ( "encoding/base64" "fmt" - "log/slog" "net" "path/filepath" "strconv" @@ -11,6 +10,7 @@ import ( "github.com/cedar2025/xboard-node/internal/config" "github.com/cedar2025/xboard-node/internal/kernel" + "github.com/cedar2025/xboard-node/internal/nlog" "github.com/cedar2025/xboard-node/internal/panel" "github.com/go-viper/mapstructure/v2" ) @@ -251,7 +251,7 @@ func buildRoutes(panelRoutes []panel.RouteRule, custom []map[string]any) M { func mergeCustomSingbox(cfg M, kcfg config.KernelConfig) { custom, err := kernel.LoadCustomConfig(kcfg.CustomConfig) if err != nil { - slog.Error("failed to load custom sing-box config", "error", err) + nlog.Core().Error("failed to load custom sing-box config", "error", err) return } if custom == nil { @@ -427,7 +427,7 @@ func buildShadowsocks(base M, nc *panel.NodeConfig, users []panel.User) M { base["users"] = userList if nc.Plugin != "" { - slog.Warn("sing-box shadowsocks inbound does not support plugin, ignoring", "plugin", nc.Plugin) + nlog.Core().Warn("sing-box shadowsocks inbound does not support plugin, ignoring", "plugin", nc.Plugin) } return base diff --git a/internal/kernel/singbox/conntracker.go b/internal/kernel/singbox/conntracker.go index d7eaec6..edfdcfb 100644 --- a/internal/kernel/singbox/conntracker.go +++ b/internal/kernel/singbox/conntracker.go @@ -3,7 +3,6 @@ package singbox import ( "context" "io" - "log/slog" "net" "sort" "sync" @@ -15,6 +14,8 @@ import ( singM "github.com/sagernet/sing/common/metadata" N "github.com/sagernet/sing/common/network" "golang.org/x/time/rate" + + "github.com/cedar2025/xboard-node/internal/nlog" ) const rateLimitThreshold = 64 * 1024 @@ -171,7 +172,7 @@ func (t *ConnTracker) RoutedConnection( if dlf := t.deviceLimitFunc.Load(); dlf != nil { if limit, hasLimit := (*dlf)(uuid); hasLimit { if t.checkDeviceGate(us, sourceIP, limit) { - slog.Info("singbox: device limit gate-keep, rejecting connection", + nlog.Core().Info("singbox: device limit gate-keep, rejecting connection", "user", uuid, "ip", sourceIP, "limit", limit) conn.Close() return conn @@ -197,17 +198,17 @@ func (t *ConnTracker) RoutedConnection( } return &trackedConn{ - Conn: conn, - tracker: t, - us: us, - connID: connID, + Conn: conn, + tracker: t, + us: us, + connID: connID, sourceIP: sourceIP, - limiter: lim, - ctx: ctx, + limiter: lim, + ctx: ctx, } } -// RoutedPacketConnection wraps a UDP PacketConn with per-user byte counting. +// RoutedPacketConnection wraps UDP with per-user counting (UDP not in connMap). func (t *ConnTracker) RoutedPacketConnection( ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, @@ -221,6 +222,17 @@ func (t *ConnTracker) RoutedPacketConnection( us := t.users[uid] t.usersMu.RUnlock() + if dlf := t.deviceLimitFunc.Load(); dlf != nil { + if limit, hasLimit := (*dlf)(uuid); hasLimit { + if t.checkDeviceGate(us, sourceIP, limit) { + nlog.Core().Info("singbox: device limit gate-keep, rejecting UDP connection", + "user", uuid, "ip", sourceIP, "limit", limit) + conn.Close() + return conn + } + } + } + if us != nil { us.addConn(sourceIP) } @@ -372,7 +384,7 @@ func (t *ConnTracker) removeConnRef(connID string) { type trackedConn struct { net.Conn tracker *ConnTracker - us *userStats // per-user stats (upload/download atomics + IP tracking) + us *userStats // per-user stats (upload/download atomics + IP tracking) connID string sourceIP string limiter *rate.Limiter diff --git a/internal/kernel/singbox/singbox.go b/internal/kernel/singbox/singbox.go index cc2762f..b20d1e5 100644 --- a/internal/kernel/singbox/singbox.go +++ b/internal/kernel/singbox/singbox.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "log/slog" "sync" "time" @@ -20,6 +19,7 @@ import ( "github.com/cedar2025/xboard-node/internal/config" "github.com/cedar2025/xboard-node/internal/kernel" + "github.com/cedar2025/xboard-node/internal/nlog" "github.com/cedar2025/xboard-node/internal/panel" ) @@ -88,7 +88,7 @@ func (s *SingBox) Start(nodeConfig *panel.NodeConfig, users []panel.User, certFi return fmt.Errorf("marshal config: %w", err) } - slog.Debug("sing-box config generated", "len", len(data)) + nlog.Core().Debug("sing-box config generated", "len", len(data)) ctx, cancel := context.WithCancel(context.Background()) ctx = include.Context(ctx) @@ -147,7 +147,7 @@ func (s *SingBox) Start(nodeConfig *panel.NodeConfig, users []panel.User, certFi go recycleOldBox(oldBox, oldCancel, oldCtx, oldTracker) } - slog.Debug("sing-box started", "users", len(users)) + nlog.Core().Debug("sing-box started", "users", len(users)) return nil } @@ -176,7 +176,7 @@ func recycleOldBox(oldBox *box.Box, oldCancel context.CancelFunc, oldCtx context if oldCancel != nil { oldCancel() } - slog.Debug("sing-box: old instance recycled") + nlog.Core().Debug("sing-box: old instance recycled") } // Reload hot-swaps the inbound users and routing rules without restarting the box. @@ -213,9 +213,9 @@ func (s *SingBox) Reload(nodeConfig *panel.NodeConfig, users []panel.User, certF // Update routing rules if err := router.UpdateRules(opts.Route.Rules, opts.Route.RuleSet); err != nil { - slog.Debug("routing reload failed", "error", err) + nlog.Core().Debug("routing reload failed", "error", err) } else { - slog.Debug("sing-box routing reloaded") + nlog.Core().Debug("sing-box routing reloaded") } nopFactory := singLog.NewNOPFactory() @@ -270,7 +270,7 @@ func (s *SingBox) Reload(nodeConfig *panel.NodeConfig, users []panel.User, certF if err == nil { continue } - slog.Warn("incremental update failed, falling back to recreate", "tag", tag, "error", err) + nlog.Core().Warn("incremental update failed, falling back to recreate", "tag", tag, "error", err) } } @@ -292,7 +292,7 @@ func (s *SingBox) Reload(nodeConfig *panel.NodeConfig, users []panel.User, certF s.connTracker.SetUserMap(buildUserMap(users)) } - slog.Debug("sing-box reloaded", "users", len(users)) + nlog.Core().Debug("sing-box reloaded", "users", len(users)) s.users = users s.nodeConfig = nodeConfig s.certFile = certFile @@ -559,7 +559,7 @@ func (s *SingBox) reloadInboundsLocked(users []panel.User) error { if err == nil { continue } - slog.Warn("incremental update failed, recreating inbound", "tag", tag, "error", err) + nlog.Core().Warn("incremental update failed, recreating inbound", "tag", tag, "error", err) } _ = im.Remove(tag) @@ -573,7 +573,7 @@ func (s *SingBox) reloadInboundsLocked(users []panel.User) error { s.connTracker.SetUserMap(buildUserMap(users)) } - slog.Debug("sing-box users hot-swapped", "users", len(users)) + nlog.Core().Debug("sing-box users hot-swapped", "users", len(users)) return nil } @@ -594,7 +594,7 @@ func (s *SingBox) CloseConnection(_ context.Context, connID string) error { return fmt.Errorf("not running") } if !ct.CloseByID(connID) { - slog.Debug("CloseConnection: connection not found (already closed?)", "id", connID) + nlog.Core().Debug("CloseConnection: connection not found (already closed?)", "id", connID) } return nil } diff --git a/internal/kernel/xray/config.go b/internal/kernel/xray/config.go index 0487e29..0f695e8 100644 --- a/internal/kernel/xray/config.go +++ b/internal/kernel/xray/config.go @@ -3,11 +3,11 @@ package xray import ( "encoding/base64" "fmt" - "log/slog" "strings" "github.com/cedar2025/xboard-node/internal/config" "github.com/cedar2025/xboard-node/internal/kernel" + "github.com/cedar2025/xboard-node/internal/nlog" "github.com/cedar2025/xboard-node/internal/panel" ) @@ -70,7 +70,7 @@ func buildConfig(kcfg config.KernelConfig, nc *panel.NodeConfig, users []panel.U if inbound != nil { cfg["inbounds"] = []M{inbound} } else { - slog.Warn("xray: unsupported protocol, no inbound configured — node will not accept connections", + nlog.Core().Warn("xray: unsupported protocol, no inbound configured — node will not accept connections", "protocol", nc.Protocol, "supported", "vmess, vless, trojan, shadowsocks, socks, http") } @@ -116,7 +116,7 @@ func mergeRouteList(a, b []map[string]any) []map[string]any { func mergeCustomXray(cfg M, kcfg config.KernelConfig) { custom, err := kernel.LoadCustomConfig(kcfg.CustomConfig) if err != nil { - slog.Error("failed to load custom xray config", "error", err) + nlog.Core().Error("failed to load custom xray config", "error", err) return } if custom == nil { diff --git a/internal/kernel/xray/dispatcher.go b/internal/kernel/xray/dispatcher.go index 58b8497..c24ac8b 100644 --- a/internal/kernel/xray/dispatcher.go +++ b/internal/kernel/xray/dispatcher.go @@ -3,7 +3,6 @@ package xray import ( "context" "errors" - "log/slog" "reflect" "sort" "sync" @@ -18,6 +17,8 @@ import ( "github.com/xtls/xray-core/features/routing" "github.com/xtls/xray-core/transport" "golang.org/x/time/rate" + + "github.com/cedar2025/xboard-node/internal/nlog" ) // Access xray's internal config creator registry so we can replace the @@ -54,7 +55,7 @@ func limitDispatcherFactory(ctx context.Context, config interface{}) (interface{ limitedIPs: make(map[string]map[string]int), } globalLimitDispatcher.Store(ld) - slog.Debug("xray: limit dispatcher installed") + nlog.Core().Debug("xray: limit dispatcher installed") return ld, nil } @@ -158,7 +159,7 @@ func (d *LimitDispatcher) identifyAndCheck(ctx context.Context, dest net.Destina isTCP = dest.Network == net.Network_TCP if d.checkDeviceLimit(email, sourceIP, isTCP) { - slog.Info("xray: device limit exceeded", "email", email, "ip", sourceIP) + nlog.Core().Info("xray: device limit exceeded", "email", email, "ip", sourceIP) return "", "", false, errors.New("device limit exceeded for " + email) } return email, sourceIP, isTCP, nil diff --git a/internal/kernel/xray/xray.go b/internal/kernel/xray/xray.go index e82da73..47724f7 100644 --- a/internal/kernel/xray/xray.go +++ b/internal/kernel/xray/xray.go @@ -6,7 +6,6 @@ import ( "crypto/sha256" "encoding/json" "fmt" - "log/slog" "os" "strings" "sync" @@ -32,6 +31,7 @@ import ( "github.com/cedar2025/xboard-node/internal/config" "github.com/cedar2025/xboard-node/internal/kernel" "github.com/cedar2025/xboard-node/internal/kernel/geodata" + "github.com/cedar2025/xboard-node/internal/nlog" "github.com/cedar2025/xboard-node/internal/panel" ) @@ -153,7 +153,7 @@ func (x *Xray) Start(nodeConfig *panel.NodeConfig, users []panel.User, certFile, x.updateDispatcherLimits(users) - slog.Info("xray started", + nlog.Core().Info("xray started", "users", len(users), "protocol", nodeConfig.Protocol, ) @@ -174,12 +174,12 @@ func (x *Xray) Reload(nodeConfig *panel.NodeConfig, users []panel.User, certFile if same { x.users = users x.mu.Unlock() - slog.Debug("xray: limits updated, kernel configuration unchanged") + nlog.Core().Debug("xray: limits updated, kernel configuration unchanged") return nil } x.mu.Unlock() - slog.Info("xray: configuration changed, performing full restart") + nlog.Core().Info("xray: configuration changed, performing full restart") return x.Start(nodeConfig, users, certFile, keyFile) } @@ -220,8 +220,10 @@ func (x *Xray) GetUserTraffic(_ context.Context) (traffic map[int][2]int64, aliv return traffic, aliveIPs, connCount, nil } - // Fallback: read xray's native stats counters + // Fallback path: aggregateStats needs x.mu. + x.mu.Lock() traffic, err = x.aggregateStats() + x.mu.Unlock() return traffic, nil, 0, err } @@ -289,7 +291,7 @@ func (x *Xray) AddUsers(users []panel.User) (int, error) { // Protocol doesn't support UserManager → full restart nc, cf, kf := x.nodeConfig, x.certFile, x.keyFile x.mu.Unlock() - slog.Debug("xray: AddUsers fallback to restart", "reason", err) + nlog.Core().Debug("xray: AddUsers fallback to restart", "reason", err) if err := x.Start(nc, merged, cf, kf); err != nil { return 0, err } @@ -305,11 +307,11 @@ func (x *Xray) AddUsers(users []panel.User) (int, error) { for _, u := range toAdd { mu, err := toMemoryUser(proto, nc, u) if err != nil { - slog.Warn("xray: skip user, cannot build account", "user", u.ID, "error", err) + nlog.Core().Warn("xray: skip user, cannot build account", "user", u.ID, "error", err) continue } if err := um.AddUser(ctx, mu); err != nil { - slog.Warn("xray: AddUser failed", "user", u.ID, "error", err) + nlog.Core().Warn("xray: AddUser failed", "user", u.ID, "error", err) continue } added++ @@ -321,7 +323,7 @@ func (x *Xray) AddUsers(users []panel.User) (int, error) { x.mu.Unlock() x.updateDispatcherLimits(merged) - slog.Info("xray: users added via UserManager", "added", added, "total", len(merged)) + nlog.Core().Info("xray: users added via UserManager", "added", added, "total", len(merged)) return added, nil } @@ -361,7 +363,7 @@ func (x *Xray) RemoveUsers(users []panel.User) (int, error) { if err != nil { nc, cf, kf := x.nodeConfig, x.certFile, x.keyFile x.mu.Unlock() - slog.Debug("xray: RemoveUsers fallback to restart", "reason", err) + nlog.Core().Debug("xray: RemoveUsers fallback to restart", "reason", err) if err := x.Start(nc, kept, cf, kf); err != nil { return 0, err } @@ -374,7 +376,7 @@ func (x *Xray) RemoveUsers(users []panel.User) (int, error) { for _, u := range users { email := userEmail(u.ID) if err := um.RemoveUser(ctx, email); err != nil { - slog.Debug("xray: RemoveUser skipped", "user", u.ID, "error", err) + nlog.Core().Debug("xray: RemoveUser skipped", "user", u.ID, "error", err) continue } actualRemoved++ @@ -385,7 +387,7 @@ func (x *Xray) RemoveUsers(users []panel.User) (int, error) { x.mu.Unlock() x.updateDispatcherLimits(kept) - slog.Info("xray: users removed via UserManager", "removed", actualRemoved, "total", len(kept)) + nlog.Core().Info("xray: users removed via UserManager", "removed", actualRemoved, "total", len(kept)) return actualRemoved, nil } @@ -406,7 +408,7 @@ func (x *Xray) UpdateUsers(users []panel.User) (added, removed int, err error) { x.users = users x.mu.Unlock() x.updateDispatcherLimits(users) - slog.Info("xray: limits updated, kernel unchanged") + nlog.Core().Info("xray: limits updated, kernel unchanged") return 0, 0, nil } @@ -415,7 +417,7 @@ func (x *Xray) UpdateUsers(users []panel.User) (added, removed int, err error) { // Protocol doesn't support UserManager → full restart nc, cf, kf := x.nodeConfig, x.certFile, x.keyFile x.mu.Unlock() - slog.Debug("xray: UpdateUsers fallback to restart", "reason", umErr) + nlog.Core().Debug("xray: UpdateUsers fallback to restart", "reason", umErr) if err = x.Start(nc, users, cf, kf); err != nil { return 0, 0, err } @@ -432,17 +434,17 @@ func (x *Xray) UpdateUsers(users []panel.User) (added, removed int, err error) { for _, u := range toRemove { email := userEmail(u.ID) if err := um.RemoveUser(ctx, email); err != nil { - slog.Debug("xray: RemoveUser skipped in UpdateUsers", "user", u.ID, "error", err) + nlog.Core().Debug("xray: RemoveUser skipped in UpdateUsers", "user", u.ID, "error", err) } } for _, u := range toAdd { mu, err := toMemoryUser(proto, nc, u) if err != nil { - slog.Warn("xray: skip user in UpdateUsers", "user", u.ID, "error", err) + nlog.Core().Warn("xray: skip user in UpdateUsers", "user", u.ID, "error", err) continue } if err := um.AddUser(ctx, mu); err != nil { - slog.Warn("xray: AddUser failed in UpdateUsers", "user", u.ID, "error", err) + nlog.Core().Warn("xray: AddUser failed in UpdateUsers", "user", u.ID, "error", err) } } @@ -451,7 +453,7 @@ func (x *Xray) UpdateUsers(users []panel.User) (added, removed int, err error) { x.mu.Unlock() x.updateDispatcherLimits(users) - slog.Info("xray: users updated via UserManager", "added", added, "removed", removed, "total", len(users)) + nlog.Core().Info("xray: users updated via UserManager", "added", added, "removed", removed, "total", len(users)) return } @@ -597,7 +599,7 @@ func (x *Xray) ensureGeoData(nc *panel.NodeConfig) { } dir := x.cfg.GeoDataDir if err := geodata.Ensure(dir, needIP, needSite, "xray"); err != nil { - slog.Warn("geo database unavailable", "error", err) + nlog.Core().Warn("geo database unavailable", "error", err) } os.Setenv("XRAY_LOCATION_ASSET", dir) } @@ -609,7 +611,7 @@ func marshalConfig(cfg config.KernelConfig, nc *panel.NodeConfig, users []panel. if err != nil { return nil, fmt.Errorf("marshal config: %w", err) } - slog.Debug("xray config generated", "len", len(data)) + nlog.Core().Debug("xray config generated", "len", len(data)) return data, nil } @@ -625,6 +627,7 @@ func startWithTimeout(inst *xrayCore.Instance, timeout time.Duration) error { } return nil case <-time.After(timeout): + go inst.Close() return fmt.Errorf("start xray: timeout after %v", timeout) } } @@ -648,7 +651,7 @@ func closeOld(inst *xrayCore.Instance, ld *LimitDispatcher) { if ld != nil { ld.ResetConns() } - slog.Debug("xray: old instance recycled") + nlog.Core().Debug("xray: old instance recycled") }() } diff --git a/internal/limiter/speedtracker.go b/internal/limiter/speedtracker.go index ebc1c27..78f2e2d 100644 --- a/internal/limiter/speedtracker.go +++ b/internal/limiter/speedtracker.go @@ -54,54 +54,50 @@ func (t *SpeedTracker) UpdateBuckets() { } t.limiter.mu.RUnlock() - t.mu.Lock() + func() { + t.mu.Lock() + defer t.mu.Unlock() - newUUIDMap := make(map[string]int, len(currentUsers)) - activeIDs := make(map[int]struct{}, len(currentUsers)) + newUUIDMap := make(map[string]int, len(currentUsers)) + activeIDs := make(map[int]struct{}, len(currentUsers)) - for _, user := range currentUsers { - activeIDs[user.ID] = struct{}{} - if user.UUID != "" { - newUUIDMap[user.UUID] = user.ID + for _, user := range currentUsers { + activeIDs[user.ID] = struct{}{} + if user.UUID != "" { + newUUIDMap[user.UUID] = user.ID + } + + if user.SpeedLimit <= 0 { + delete(t.buckets, user.ID) + continue + } + + bytesPerSec := int(user.SpeedLimit) * 1_000_000 / 8 + burst := bytesPerSec + if burst < 64*1024 { + burst = 64 * 1024 + } + if cap4s := bytesPerSec * 4; cap4s > 64*1024 && burst > cap4s { + burst = cap4s + } + + if existing, ok := t.buckets[user.ID]; ok { + existing.SetLimit(rate.Limit(bytesPerSec)) + existing.SetBurst(burst) + } else { + t.buckets[user.ID] = rate.NewLimiter(rate.Limit(bytesPerSec), burst) + } } - if user.SpeedLimit <= 0 { - delete(t.buckets, user.ID) - continue + for id := range t.buckets { + if _, ok := activeIDs[id]; !ok { + delete(t.buckets, id) + } } - bytesPerSec := int(user.SpeedLimit) * 1_000_000 / 8 - burst := bytesPerSec - if burst < 64*1024 { - burst = 64 * 1024 - } - if cap4s := bytesPerSec * 4; cap4s > 64*1024 && burst > cap4s { - burst = cap4s - } - - if existing, ok := t.buckets[user.ID]; ok { - existing.SetLimit(rate.Limit(bytesPerSec)) - existing.SetBurst(burst) - } else { - t.buckets[user.ID] = rate.NewLimiter(rate.Limit(bytesPerSec), burst) - } - } - - // Remove limiters for users who are no longer in the list - for id := range t.buckets { - if _, ok := activeIDs[id]; !ok { - delete(t.buckets, id) - } - } - - t.uuidMap = newUUIDMap - t.hasLimits.Store(len(t.buckets) > 0) - - // IMPORTANT: logFunc must run after Unlock. The callback installed in - // service.startKernel calls LimitedUserCount(), which needs RLock on t.mu. - // Calling it while this goroutine holds Lock causes a self-deadlock on - // sync.RWMutex (writer blocked forever waiting for itself to release). - t.mu.Unlock() + t.uuidMap = newUUIDMap + t.hasLimits.Store(len(t.buckets) > 0) + }() if t.logFunc != nil { t.logFunc("buckets updated") diff --git a/internal/monitor/monitor.go b/internal/monitor/monitor.go index 13dde56..2f99f6a 100644 --- a/internal/monitor/monitor.go +++ b/internal/monitor/monitor.go @@ -1,10 +1,10 @@ package monitor import ( - "log/slog" "runtime" "time" + "github.com/cedar2025/xboard-node/internal/nlog" "github.com/shirou/gopsutil/v4/cpu" "github.com/shirou/gopsutil/v4/disk" "github.com/shirou/gopsutil/v4/load" @@ -50,7 +50,7 @@ func Collect() Status { if cpuPercent, err := cpu.Percent(0, false); err == nil && len(cpuPercent) > 0 { s.CPU = cpuPercent[0] } else if err != nil { - slog.Debug("failed to get CPU usage", "error", err) + nlog.Core().Debug("failed to get CPU usage", "error", err) } // Per-core CPU usage (best-effort; safe if it fails). diff --git a/internal/nlog/nlog.go b/internal/nlog/nlog.go index 2ec6e5e..defa15c 100644 --- a/internal/nlog/nlog.go +++ b/internal/nlog/nlog.go @@ -3,6 +3,7 @@ package nlog import ( "context" "fmt" + "io" "log/slog" "os" "strings" @@ -33,8 +34,44 @@ var ( mu sync.RWMutex defaultNode string nodeLoggers = make(map[string]*NodeLog) + + logMu sync.RWMutex + logWriter io.Writer = os.Stdout + logMin = slog.LevelInfo + logColor = true ) +// Init configures process-wide log output (called from config.InitLogger). +// w must be non-nil. minLevel drops messages below that severity. +// color enables ANSI coloring when true (typically TTY + stdout/stderr). +func Init(w io.Writer, minLevel slog.Level, color bool) { + if w == nil { + w = os.Stdout + } + logMu.Lock() + defer logMu.Unlock() + logWriter = w + logMin = minLevel + logColor = color +} + +// formatMsg appends alternating key-value pairs like slog: ("k", v, "k2", v2). +func formatMsg(msg string, args []any) string { + if len(args) == 0 { + return msg + } + var b strings.Builder + b.WriteString(msg) + for i := 0; i < len(args); i += 2 { + if i+1 < len(args) { + fmt.Fprintf(&b, " %v=%v", args[i], args[i+1]) + } else { + fmt.Fprintf(&b, " %v", args[i]) + } + } + return b.String() +} + // SetDefault sets the default node prefix for core-level logs. func SetDefault(prefix string) { mu.Lock() @@ -128,48 +165,61 @@ func (nl *NodeLog) Error(msg string, args ...any) { logWithColor(slog.LevelError, nl.prefix, msg, args...) } -// logWithColor outputs a colored log line directly to stdout. -// Format: HH:MM:SS.mmm LEVEL [prefix] message +// logWithColor writes one line to the configured writer (see Init). +// Format: HH:MM:SS.mmm LEVEL [prefix] message [key=value ...] func logWithColor(level slog.Level, prefix, msg string, args ...any) { - // Time - now := time.Now().Format("15:04:05.000") + logMu.RLock() + out := logWriter + min := logMin + color := logColor + logMu.RUnlock() - // Level with color - var levelStr, levelColor string + if level < min { + return + } + + now := time.Now().Format("15:04:05.000") + fullMsg := formatMsg(msg, args) + + var levelStr string switch level { case slog.LevelDebug: levelStr = "DEBUG" - levelColor = ColorGreen case slog.LevelInfo: levelStr = "INFO " - levelColor = ColorWhite case slog.LevelWarn: levelStr = "WARN " - levelColor = ColorYellow case slog.LevelError: levelStr = "ERROR" - levelColor = ColorRed default: levelStr = "?????" + } + + if !color { + fmt.Fprintf(out, "%s %s [%s] %s\n", now, strings.TrimSpace(levelStr), prefix, fullMsg) + return + } + + var levelColor string + switch level { + case slog.LevelDebug: + levelColor = ColorGreen + case slog.LevelInfo: + levelColor = ColorWhite + case slog.LevelWarn: + levelColor = ColorYellow + case slog.LevelError: + levelColor = ColorRed + default: levelColor = ColorWhite } - // Prefix with color prefixColor := ColorCyan if prefix == "core" { prefixColor = ColorBlue } - // Format message - var fullMsg string - if len(args) == 0 { - fullMsg = msg - } else { - fullMsg = fmt.Sprintf("%s %v", msg, args) - } - - // Output colored line - fmt.Fprintf(os.Stdout, "%s%s%s %s%s%s %s[%s]%s %s\n", + fmt.Fprintf(out, "%s%s%s %s%s%s %s[%s]%s %s\n", ColorGray, now, ColorReset, levelColor, levelStr, ColorReset, prefixColor, prefix, ColorReset, diff --git a/internal/panel/client.go b/internal/panel/client.go index b4e2a16..3f5fda0 100644 --- a/internal/panel/client.go +++ b/internal/panel/client.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "io" - "log/slog" "net/http" "net/url" "reflect" @@ -16,6 +15,7 @@ import ( "time" "github.com/cedar2025/xboard-node/internal/config" + "github.com/cedar2025/xboard-node/internal/nlog" "github.com/go-viper/mapstructure/v2" ) @@ -345,7 +345,7 @@ func (c *Client) doRequest(method, path string, body []byte, ifNoneMatch string) req.Header.Set("If-None-Match", ifNoneMatch) } - slog.Debug("panel request", "method", method, "path", path) + nlog.Core().Debug("panel request", "method", method, "path", path) resp, err := c.httpClient.Do(req) if err != nil { @@ -359,7 +359,7 @@ func (c *Client) doRequest(method, path string, body []byte, ifNoneMatch string) c.apiFailure.Add(1) } - slog.Debug("panel response", "method", method, "path", path, "status", resp.StatusCode) + nlog.Core().Debug("panel response", "method", method, "path", path, "status", resp.StatusCode) return resp, nil } diff --git a/internal/panel/ws.go b/internal/panel/ws.go index c31381b..8b5a489 100644 --- a/internal/panel/ws.go +++ b/internal/panel/ws.go @@ -4,13 +4,13 @@ import ( "context" "encoding/json" "fmt" - "log/slog" "math/rand" "net/url" "strconv" "sync/atomic" "time" + "github.com/cedar2025/xboard-node/internal/nlog" "github.com/gorilla/websocket" ) @@ -108,7 +108,7 @@ func (w *WSClient) Run(ctx context.Context) { } if err != nil { - slog.Warn("ws disconnected", "error", err) + nlog.Core().Warn("ws disconnected", "error", err) if !wasConnected { w.notifyStatus(false) } @@ -153,7 +153,7 @@ func (w *WSClient) connect(ctx context.Context) error { q.Set("node_id", strconv.Itoa(w.nodeID)) u.RawQuery = q.Encode() - slog.Debug("ws connecting", "url", u.Redacted()) + nlog.Core().Debug("ws connecting", "url", u.Host) dialer := websocket.Dialer{ HandshakeTimeout: 15 * time.Second, @@ -176,7 +176,9 @@ func (w *WSClient) connect(ctx context.Context) error { var errData struct { Message string `json:"message"` } - json.Unmarshal(firstMsg.Data, &errData) + if err := json.Unmarshal(firstMsg.Data, &errData); err != nil { + return fmt.Errorf("auth failed (unable to parse error: %v)", err) + } return fmt.Errorf("auth failed: %s", errData.Message) } @@ -213,6 +215,8 @@ func (w *WSClient) connect(ctx context.Context) error { case msgCh <- msg: case <-ctx.Done(): return + default: + nlog.Core().Warn("ws: message channel full, dropping message", "event", msg.Event) } } }() @@ -242,7 +246,7 @@ func (w *WSClient) connect(ctx context.Context) error { select { case writeCh <- wsMessage{Event: "pong"}: default: - slog.Warn("ws write channel full, skipping pong") + nlog.Core().Warn("ws write channel full, skipping pong") } } @@ -258,7 +262,7 @@ func (w *WSClient) connect(ctx context.Context) error { select { case writeCh <- msg: default: - slog.Warn("ws write channel full, skipping status push (network slow?)") + nlog.Core().Warn("ws write channel full, skipping status push (network slow?)") } } @@ -276,10 +280,10 @@ func (w *WSClient) handleMessage(msg wsMessage) { switch msg.Event { case "ping": // Server ping — handled by the pong timer reset above - slog.Debug("ws received ping") + nlog.Core().Debug("ws received ping") case "auth.success": - slog.Debug("ws auth confirmed") + nlog.Core().Debug("ws auth confirmed") case WSEventSyncConfig: w.handleDataEvent(msg) @@ -291,7 +295,7 @@ func (w *WSClient) handleMessage(msg wsMessage) { w.handleDataEvent(msg) default: - slog.Debug("ws unknown event", "event", msg.Event) + nlog.Core().Debug("ws unknown event", "event", msg.Event) } } @@ -310,44 +314,44 @@ func (w *WSClient) handleDataEvent(msg wsMessage) { switch msg.Event { case WSEventSyncConfig: - slog.Debug("ws sync config event received") + nlog.Core().Debug("ws sync config event received") var p syncConfigPayload if err := decodeData(msg.Data, &p); err != nil { - slog.Warn("ws: cannot decode config payload", "error", err) + nlog.Core().Warn("ws: cannot decode config payload", "error", err) return } if p.Config.Protocol == "" { - slog.Warn("ws: config payload missing protocol") + nlog.Core().Warn("ws: config payload missing protocol") return } event.Config = &p.Config case WSEventSyncUsers: - slog.Debug("ws sync users event received") + nlog.Core().Debug("ws sync users event received") var p syncUsersPayload if err := decodeData(msg.Data, &p); err != nil { - slog.Warn("ws: cannot decode users payload", "error", err) + nlog.Core().Warn("ws: cannot decode users payload", "error", err) return } if len(p.Users) == 0 { - slog.Warn("ws: users payload empty") + nlog.Core().Warn("ws: users payload empty") return } event.Users = p.Users case WSEventSyncUserDelta: - slog.Debug("ws sync user delta event received") + nlog.Core().Debug("ws sync user delta event received") var p syncUserDeltaPayload if err := decodeData(msg.Data, &p); err != nil { - slog.Warn("ws: cannot decode user delta payload", "error", err) + nlog.Core().Warn("ws: cannot decode user delta payload", "error", err) return } if p.Action == "" { - slog.Warn("ws: user delta payload missing action") + nlog.Core().Warn("ws: user delta payload missing action") return } if len(p.Users) == 0 { - slog.Warn("ws: user delta payload has no users") + nlog.Core().Warn("ws: user delta payload has no users") return } event.DeltaAction = p.Action diff --git a/internal/service/service.go b/internal/service/service.go index 7aba54d..3dcd084 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -7,9 +7,9 @@ import ( "encoding/json" "fmt" "io" - "log/slog" "math" "sort" + "sync" "sync/atomic" "time" @@ -66,6 +66,9 @@ type Service struct { wsStatusCh chan panel.WSStatusChange // receives WS connect/disconnect notifications wsCancel context.CancelFunc // cancels the WS client goroutine wsDisconnectAt time.Time // when WS last disconnected (zero if connected) + + // metricsMu: lastUsers, lastConfig, wsClient, wsDisconnectAt (buildMetrics vs main loop). + metricsMu sync.RWMutex } // pullResult carries the outcome of an async pullViaAPI back to the main goroutine. @@ -79,10 +82,13 @@ type pullResult struct { // apiBackoff implements simple exponential backoff for API failures. type apiBackoff struct { + mu sync.Mutex skipRemaining int } func (b *apiBackoff) shouldSkip() bool { + b.mu.Lock() + defer b.mu.Unlock() if b.skipRemaining > 0 { b.skipRemaining-- return true @@ -90,8 +96,15 @@ func (b *apiBackoff) shouldSkip() bool { return false } -func (b *apiBackoff) onSuccess() { b.skipRemaining = 0 } +func (b *apiBackoff) onSuccess() { + b.mu.Lock() + b.skipRemaining = 0 + b.mu.Unlock() +} + func (b *apiBackoff) onFailure() { + b.mu.Lock() + defer b.mu.Unlock() if b.skipRemaining <= 0 { b.skipRemaining = 1 } else if b.skipRemaining < 8 { @@ -110,7 +123,7 @@ func New(cfg *config.Config) *Service { case "xray": k = xray.New(cfg.Kernel) default: - slog.Error("unsupported kernel type, defaulting to sing-box", "type", cfg.Kernel.Type) + nlog.Core().Error("unsupported kernel type, defaulting to sing-box", "type", cfg.Kernel.Type) k = singbox.New(cfg.Kernel) } @@ -276,14 +289,14 @@ func (s *Service) initialSetup(ctx context.Context) error { s.limiter.UpdateUsers(users) s.speedTracker.UpdateBuckets() - slog.Info("handshake complete (V1 fallback)", + nlog.Core().Info("handshake complete (V1 fallback)", "protocol", nodeConfig.Protocol, "port", nodeConfig.ServerPort, "users", len(users), ) if len(users) == 0 { - slog.Warn("no users, kernel will not start until users are available") + nlog.Core().Warn("no users, kernel will not start until users are available") return nil } @@ -310,7 +323,7 @@ func (s *Service) applyRemoteOverrides(ctx context.Context, nc *panel.NodeConfig // Dynamic Log Level (Kernel) if nc.KernelLogLevel != "" && nc.KernelLogLevel != s.cfg.Kernel.LogLevel { - slog.Info("cert: kernel log level override", "old", s.cfg.Kernel.LogLevel, "new", nc.KernelLogLevel) + nlog.Core().Info("cert: kernel log level override", "old", s.cfg.Kernel.LogLevel, "new", nc.KernelLogLevel) s.cfg.Kernel.LogLevel = nc.KernelLogLevel } @@ -321,7 +334,7 @@ func (s *Service) applyRemoteOverrides(ctx context.Context, nc *panel.NodeConfig // Legacy fields (deprecated: prefer cert_config) if nc.AutoTLS != s.cfg.Cert.AutoTLS { - slog.Info("cert: auto_tls policy changed (deprecated field)", "new", nc.AutoTLS) + nlog.Core().Info("cert: auto_tls policy changed (deprecated field)", "new", nc.AutoTLS) s.cfg.Cert.AutoTLS = nc.AutoTLS } if nc.Domain != "" && nc.Domain != s.cfg.Cert.Domain { @@ -351,7 +364,7 @@ func (s *Service) applyPanelCert(ctx context.Context, pc *panel.CertConfig) bool changed, err := s.cert.Reconfigure(ctx, newCfg) if err != nil { - slog.Error("failed to apply panel cert config", "mode", pc.CertMode, "error", err) + nlog.Core().Error("failed to apply panel cert config", "mode", pc.CertMode, "error", err) return false } s.cfg.Cert = newCfg @@ -386,7 +399,7 @@ func (s *Service) newWSClient(wsURL string) *panel.WSClient { select { case s.wsEvents <- event: default: - slog.Warn("ws event channel full, dropping event", "type", event.Type) + nlog.Core().Warn("ws event channel full, dropping event", "type", event.Type) } }, func(status panel.WSStatusChange) { @@ -410,7 +423,9 @@ func (s *Service) newWSClient(wsURL string) *panel.WSClient { // - On reconnect: clear disconnect timestamp, REST poll to catch missed events. func (s *Service) handleWSStatus(ctx context.Context, status panel.WSStatusChange) { if status.Connected { + s.metricsMu.Lock() s.wsDisconnectAt = time.Time{} + s.metricsMu.Unlock() // Use nodeLog if available, otherwise core if s.nodeLog != nil { s.nodeLog.Info("ws connected") @@ -421,9 +436,11 @@ func (s *Service) handleWSStatus(ctx context.Context, status panel.WSStatusChang // any updates during the disconnection window. s.pullViaAPIAsync(ctx) } else { + s.metricsMu.Lock() if s.wsDisconnectAt.IsZero() { s.wsDisconnectAt = time.Now() } + s.metricsMu.Unlock() if s.nodeLog != nil { s.nodeLog.Info("ws disconnected") } else { @@ -447,10 +464,10 @@ func (s *Service) wsDiscovery(ctx context.Context) { if s.wsClient == nil { needsCheck = true - slog.Debug("ws discovery: no WS client, checking if panel enabled WS") + nlog.Core().Debug("ws discovery: no WS client, checking if panel enabled WS") } else if !s.wsDisconnectAt.IsZero() && time.Since(s.wsDisconnectAt) > 10*time.Minute { needsCheck = true - slog.Debug("ws discovery: WS disconnected for >10min, re-checking WS config") + nlog.Core().Debug("ws discovery: WS disconnected for >10min, re-checking WS config") } if !needsCheck { @@ -459,7 +476,7 @@ func (s *Service) wsDiscovery(ctx context.Context) { hs, err := s.panel.Handshake() if err != nil { - slog.Debug("ws discovery: handshake failed", "error", err) + nlog.Core().Debug("ws discovery: handshake failed", "error", err) return } @@ -468,19 +485,24 @@ func (s *Service) wsDiscovery(ctx context.Context) { if hs.WebSocket.Enabled && hs.WebSocket.WSURL != "" { if s.wsClient == nil { - slog.Info("ws discovery: panel has WS enabled, creating WS client") - s.wsClient = s.newWSClient(hs.WebSocket.WSURL) + nlog.Core().Info("ws discovery: panel has WS enabled, creating WS client") + wsc := s.newWSClient(hs.WebSocket.WSURL) + s.metricsMu.Lock() + s.wsClient = wsc s.wsDisconnectAt = time.Time{} + s.metricsMu.Unlock() s.startWSClient(ctx) } } else if s.wsClient != nil { - slog.Info("ws discovery: panel disabled WS, switching to REST-only") + nlog.Core().Info("ws discovery: panel disabled WS, switching to REST-only") if s.wsCancel != nil { s.wsCancel() } + s.metricsMu.Lock() s.wsClient = nil - s.wsCancel = nil s.wsDisconnectAt = time.Time{} + s.metricsMu.Unlock() + s.wsCancel = nil } } @@ -500,7 +522,9 @@ func (s *Service) handleWSEvent(ctx context.Context, event panel.WSEvent) { s.nodeLog = nlog.ForNode(event.Config.Protocol, event.Config.ServerPort) } s.nodeLog.Info(fmt.Sprintf("config updated, %d users", len(event.Users))) + s.metricsMu.Lock() s.lastConfig = event.Config + s.metricsMu.Unlock() s.lastConfigHash = newConfigHash s.applyRemoteOverrides(ctx, event.Config) s.applyChanges(ctx, true, false) @@ -536,12 +560,12 @@ func (s *Service) handleWSEvent(ctx context.Context, event panel.WSEvent) { // goroutine and sends the result to pullResults for the main goroutine to apply. func (s *Service) pullViaAPIAsync(ctx context.Context) { if !s.pullActive.CompareAndSwap(false, true) { - slog.Debug("pull already in progress, skipping") + nlog.Core().Debug("pull already in progress, skipping") return } if s.pullBackoff.shouldSkip() { - slog.Debug("skipping pull due to backoff") + nlog.Core().Debug("skipping pull due to backoff") s.pullActive.Store(false) return } @@ -555,14 +579,14 @@ func (s *Service) pullViaAPIAsync(ctx context.Context) { config, err := s.panel.GetConfig() if err != nil { - slog.Error("poll config failed", "error", err) + nlog.Core().Error("poll config failed", "error", err) s.pullBackoff.onFailure() return } users, err := s.panel.GetUsers() if err != nil { - slog.Error("poll users failed", "error", err) + nlog.Core().Error("poll users failed", "error", err) s.pullBackoff.onFailure() return } @@ -609,7 +633,9 @@ func (s *Service) applyPullResult(ctx context.Context, result pullResult) { s.nodeLog = nlog.ForNode(result.config.Protocol, result.config.ServerPort) } s.nodeLog.Info(fmt.Sprintf("config updated, %d users", len(s.lastUsers))) + s.metricsMu.Lock() s.lastConfig = result.config + s.metricsMu.Unlock() s.lastConfigHash = result.configHash if s.applyRemoteOverrides(ctx, result.config) { configChanged = true @@ -641,7 +667,11 @@ func (s *Service) updateUserState(users []panel.User) { s.limiter.UpdateUsers(users) s.speedTracker.UpdateBuckets() + + s.metricsMu.Lock() s.lastUsers = users + s.metricsMu.Unlock() + s.lastUserHash = computeUserHash(users) } @@ -649,7 +679,7 @@ func (s *Service) updateUserState(users []panel.User) { // records the successfully applied state. Returns false on error. func (s *Service) startKernel(nc *panel.NodeConfig, users []panel.User) bool { if err := s.kernel.Start(nc, users, s.cert.CertFile(), s.cert.KeyFile()); err != nil { - slog.Error("failed to start kernel", "error", err) + nlog.Core().Error("failed to start kernel", "error", err) return false } @@ -685,10 +715,6 @@ func (s *Service) ensureRunning() bool { // applyUserUpdate replaces the full user set and hot-swaps the kernel. // Called from WS sync.users and REST polling. func (s *Service) applyUserUpdate(ctx context.Context, users []panel.User, newHash string) { - // Update limiter/speed state first so even if kernel ops fail, - // the in-memory limits are already correct. - s.updateUserState(users) - if !s.ensureRunning() { return } @@ -699,23 +725,22 @@ func (s *Service) applyUserUpdate(ctx context.Context, users []panel.User, newHa s.startKernel(s.lastConfig, users) return } + s.updateUserState(users) if s.nodeLog != nil && (added > 0 || removed > 0) { s.nodeLog.Info(fmt.Sprintf("users updated: +%d -%d", added, removed)) } } // applyUserDelta applies an incremental user change (add or remove) directly -// via the kernel's atomic user API. +// via the kernel's atomic user API. Kernel updates run before updateUserState. func (s *Service) applyUserDelta(ctx context.Context, action string, deltaUsers []panel.User) { switch action { case "add": // Defensive check for empty or nil deltaUsers if deltaUsers == nil || len(deltaUsers) == 0 { - nlog.Core().Warn("add delta with empty users, skipping") return } merged := mergeUsers(s.lastUsers, deltaUsers) - s.updateUserState(merged) if !s.ensureRunning() { return @@ -723,22 +748,23 @@ func (s *Service) applyUserDelta(ctx context.Context, action string, deltaUsers added, err := s.kernel.AddUsers(deltaUsers) if err != nil { - nlog.Core().Warn(fmt.Sprintf("AddUsers failed: %v", err)) + nlog.Core().Warn(fmt.Sprintf("AddUsers failed: %v, falling back to UpdateUsers", err)) if _, _, err := s.kernel.UpdateUsers(merged); err != nil { nlog.Core().Error(fmt.Sprintf("UpdateUsers fallback failed: %v", err)) + return } - } else if s.nodeLog != nil && added > 0 { + } + s.updateUserState(merged) + if s.nodeLog != nil && added > 0 { s.nodeLog.Info(fmt.Sprintf("users added: +%d", added)) } case "remove": // Defensive check for empty or nil deltaUsers if deltaUsers == nil || len(deltaUsers) == 0 { - nlog.Core().Warn("remove delta with empty users, skipping") return } filtered := subtractUsers(s.lastUsers, deltaUsers) - s.updateUserState(filtered) if !s.kernel.IsRunning() { return @@ -746,11 +772,14 @@ func (s *Service) applyUserDelta(ctx context.Context, action string, deltaUsers removed, err := s.kernel.RemoveUsers(deltaUsers) if err != nil { - nlog.Core().Warn(fmt.Sprintf("RemoveUsers failed: %v", err)) + nlog.Core().Warn(fmt.Sprintf("RemoveUsers failed: %v, falling back to UpdateUsers", err)) if _, _, err := s.kernel.UpdateUsers(filtered); err != nil { nlog.Core().Error(fmt.Sprintf("UpdateUsers fallback failed: %v", err)) + return } - } else if s.nodeLog != nil && removed > 0 { + } + s.updateUserState(filtered) + if s.nodeLog != nil && removed > 0 { s.nodeLog.Info(fmt.Sprintf("users removed: -%d", removed)) } @@ -786,6 +815,12 @@ func mergeUsers(base, delta []panel.User) []panel.User { // subtractUsers returns base with all users in delta removed. func subtractUsers(base, delta []panel.User) []panel.User { + if base == nil { + return nil + } + if delta == nil || len(delta) == 0 { + return base + } removeSet := make(map[int]struct{}, len(delta)) for _, u := range delta { removeSet[u.ID] = struct{}{} @@ -839,7 +874,7 @@ func (s *Service) trackAndEnforce(ctx context.Context) { traffic, aliveIPs, connCount, err := s.kernel.GetUserTraffic(ctx) if err != nil { - slog.Debug("get user traffic failed", "error", err) + nlog.Core().Debug("get user traffic failed", "error", err) return } @@ -859,12 +894,12 @@ func (s *Service) trackAndEnforce(ctx context.Context) { // loop is never blocked by slow HTTP. Only one push runs at a time. func (s *Service) pushReportAsync() { if !s.pushActive.CompareAndSwap(false, true) { - slog.Debug("push already in progress, skipping") + nlog.Core().Debug("push already in progress, skipping") return } if s.pushBackoff.shouldSkip() { - slog.Debug("skipping report due to backoff") + nlog.Core().Debug("skipping report due to backoff") s.pushActive.Store(false) return } @@ -888,7 +923,7 @@ func (s *Service) pushReportAsync() { [2]uint64{status.DiskTotal, status.DiskUsed}, metrics, ); err != nil { - slog.Error("failed to push report", "error", err) + nlog.Core().Error("failed to push report", "error", err) if len(traffic) > 0 { s.tracker.RestoreTraffic(traffic) } @@ -921,7 +956,7 @@ func (s *Service) pushReportSync() { [2]uint64{status.DiskTotal, status.DiskUsed}, metrics, ); err != nil { - slog.Error("failed to push final report", "error", err) + nlog.Core().Error("failed to push final report", "error", err) } } @@ -929,6 +964,11 @@ func (s *Service) pushReportSync() { // This includes active connections, per-core CPU, GC stats, API call stats, // WebSocket status, and limiter hit counts. func (s *Service) buildMetrics(status monitor.Status) map[string]interface{} { + s.metricsMu.RLock() + lastUsers := s.lastUsers + wsClient := s.wsClient + s.metricsMu.RUnlock() + m := make(map[string]interface{}) online := s.tracker.CurrentOnline() @@ -939,7 +979,7 @@ func (s *Service) buildMetrics(status monitor.Status) map[string]interface{} { m["active_connections"] = s.tracker.ActiveConnections() m["total_connections"] = s.tracker.TotalConnections() m["active_users"] = len(online) - m["total_users"] = len(s.lastUsers) + m["total_users"] = len(lastUsers) // Speed m["inbound_speed"] = s.tracker.InboundSpeed() @@ -976,8 +1016,8 @@ func (s *Service) buildMetrics(status monitor.Status) map[string]interface{} { } // WebSocket status. - wsEnabled := s.wsClient != nil - wsConnected := wsEnabled && s.wsClient.IsConnected() + wsEnabled := wsClient != nil + wsConnected := wsEnabled && wsClient.IsConnected() m["ws"] = map[string]interface{}{ "enabled": wsEnabled, "connected": wsConnected, diff --git a/internal/tracker/tracker.go b/internal/tracker/tracker.go index 6694678..0cd50b6 100644 --- a/internal/tracker/tracker.go +++ b/internal/tracker/tracker.go @@ -203,17 +203,15 @@ func (t *Tracker) RestoreAliveIPs(data map[int][]string) { if ips == nil { ips = make([]string, 0, len(ipList)) } + // Use map for O(n) dedup instead of O(n²) linear search + existMap := make(map[string]struct{}, len(ips)+len(ipList)) + for _, existing := range ips { + existMap[existing] = struct{}{} + } for _, ip := range ipList { - // Dedup check - found := false - for _, existing := range ips { - if existing == ip { - found = true - break - } - } - if !found { + if _, exists := existMap[ip]; !exists { ips = append(ips, ip) + existMap[ip] = struct{}{} } } t.aliveIPsBuf[uid] = ips