diff --git a/README.md b/README.md index 797983b..21cb050 100644 --- a/README.md +++ b/README.md @@ -97,7 +97,7 @@ Docker 镜像中的默认路径为： | 环境变量 | 默认值 | 说明 | | --- | --- | --- | | `CONFIG_PATH` | `/app/config/config.json` | 主配置文件 | -| `STATS_PATH` | `/app/data/stats.json` | 月流量与状态持久化文件 | +| `STATS_PATH` | `/app/data/stats.json` | 月流量与状态持久化文件；每 60 秒写入，关键操作与正常退出时立即写入 | | `WEB_DIR` | `/app/web` | WebUI 静态文件目录 | | `HTTP_ADDR` | `:80` | WebUI 与 HTTP API 监听地址 | | `AGENT_ADDR` | `:35601` | 客户端 TCP 上报监听地址 | diff --git a/server/app.go b/server/app.go index b29dd54..cc8a130 100644 --- a/server/app.go +++ b/server/app.go @@ -55,21 +55,22 @@ type App struct { document ConfigDocument runtime RuntimeConfig - nodeMu sync.RWMutex - nodes map[string]*NodeState - connectionID atomic.Uint64 - generation atomic.Uint64 - agentRunning atomic.Bool - reloadWrites atomic.Int32 + nodeMu sync.RWMutex + nodes map[string]*NodeState + connectionID atomic.Uint64 + generation atomic.Uint64 + agentRunning atomic.Bool + reloadPending atomic.Bool certMu sync.RWMutex certs map[string]*CertState - statsWake chan struct{} persistMu sync.Mutex logger *log.Logger } +const statsFlushInterval = time.Minute + func NewApp(opts Options) (*App, error) { doc, runtime, err := readConfig(opts.ConfigPath) if err != nil { @@ -83,7 +84,6 @@ func NewApp(opts Options) (*App, error) { cancel: cancel, nodes: make(map[string]*NodeState), certs: make(map[string]*CertState), - statsWake: make(chan struct{}, 1), logger: log.New(os.Stdout, "serverstatus ", log.LstdFlags|log.Lmicroseconds), } app.applyValidatedConfig(doc, runtime, false) @@ -94,7 +94,6 @@ func NewApp(opts Options) (*App, error) { func (a *App) StartBackground() { go a.statsLoop() go a.sslLoop() - a.wakeStatsWriter() } func (a *App) Close() { @@ -216,8 +215,12 @@ func (a *App) applyValidatedConfig(doc ConfigDocument, runtime RuntimeConfig, di _ = conn.Close() } } - a.reloadWrites.Store(2) - a.wakeStatsWriter() + if disconnect { + a.reloadPending.Store(true) + if err := a.PersistStats(); err != nil { + a.logger.Printf("write stats after config change: %v", err) + } + } } func sameServerIdentity(left, right ServerConfig) bool { @@ -246,14 +249,13 @@ func (a *App) disconnectAll(reason string) { } func (a *App) statsLoop() { - ticker := time.NewTicker(time.Second) + ticker := time.NewTicker(statsFlushInterval) defer ticker.Stop() for { select { case <-a.ctx.Done(): return case <-ticker.C: - case <-a.statsWake: } if err := a.PersistStats(); err != nil { a.logger.Printf("write stats: %v", err) @@ -261,18 +263,15 @@ func (a *App) statsLoop() { } } -func (a *App) wakeStatsWriter() { - select { - case a.statsWake <- struct{}{}: - default: - } -} - func (a *App) SnapshotStats() map[string]any { - return a.snapshotStats(false) + result := a.snapshotStats() + if a.reloadPending.Swap(false) { + result["reload"] = true + } + return result } -func (a *App) snapshotStats(consumeReload bool) map[string]any { +func (a *App) snapshotStats() map[string]any { runtime := a.RuntimeSnapshot() now := time.Now() servers := make([]any, 0, len(runtime.Servers)) @@ -325,19 +324,13 @@ func (a *App) snapshotStats(consumeReload bool) map[string]any { "sslcerts": a.sslSnapshot(runtime.SSLCerts, now), "updated": strconv.FormatInt(now.Unix(), 10), } - if a.reloadWrites.Load() > 0 { - result["reload"] = true - if consumeReload { - a.reloadWrites.Add(-1) - } - } return result } func (a *App) PersistStats() error { a.persistMu.Lock() defer a.persistMu.Unlock() - return writeStatsFile(a.opts.StatsPath, a.snapshotStats(true)) + return writeStatsFile(a.opts.StatsPath, a.snapshotStats()) } func monthResetWindow(now time.Time, monthStart int) bool { @@ -452,7 +445,9 @@ func (a *App) ResetTraffic(username string) (map[string]any, *APIError) { node.LastNetworkIn, node.LastNetworkOut = networkIn, networkOut server := node.Config a.nodeMu.Unlock() - a.wakeStatsWriter() + if err := a.PersistStats(); err != nil { + a.logger.Printf("write stats after traffic reset for %q: %v", username, err) + } return map[string]any{ "server": server, "stats": map[string]any{ diff --git a/server/app_test.go b/server/app_test.go index 537c7d4..74d4f45 100644 --- a/server/app_test.go +++ b/server/app_test.go @@ -1,10 +1,34 @@ package main import ( + "encoding/json" "net" + "os" "testing" + "time" ) +func readPersistedServer(t *testing.T, app *App) map[string]any { + t.Helper() + data, err := os.ReadFile(app.opts.StatsPath) + if err != nil { + t.Fatal(err) + } + var stats map[string]any + if err := json.Unmarshal(data, &stats); err != nil { + t.Fatal(err) + } + servers, ok := stats["servers"].([]any) + if !ok || len(servers) != 1 { + t.Fatalf("unexpected persisted servers: %#v", stats["servers"]) + } + server, ok := servers[0].(map[string]any) + if !ok { + t.Fatalf("unexpected persisted server: %#v", servers[0]) + } + return server +} + func TestTrafficBaselinesResetIndependently(t *testing.T) { node := &NodeState{LastNetworkIn: 100, LastNetworkOut: 0} updateTrafficBaselines(node, 150, 500, false) @@ -48,3 +72,110 @@ func TestDisconnectPreservesOfflineDisplayMetadata(t *testing.T) { t.Fatalf("offline display metadata was discarded: %#v", serverStats) } } + +func TestStatsPersistenceUsesLiveMemoryBetweenFlushes(t *testing.T) { + if statsFlushInterval != time.Minute { + t.Fatalf("unexpected stats flush interval: %s", statsFlushInterval) + } + + app := newTestApp(t, minimalTestConfig()) + if err := app.PersistStats(); err != nil { + t.Fatal(err) + } + initialInfo, err := os.Stat(app.opts.StatsPath) + if err != nil { + t.Fatal(err) + } + + app.nodeMu.Lock() + node := app.nodes["s01"] + node.Connected = true + node.ConnectionID = 7 + app.nodeMu.Unlock() + if !app.updateAgent("s01", 7, AgentStats{CPU: 42, NetworkIn: 1000, NetworkOut: 2000}) { + t.Fatal("agent update was rejected") + } + + live := app.SnapshotStats()["servers"].([]any)[0].(map[string]any) + if live["cpu"] != 42 { + t.Fatalf("live snapshot was not updated: %#v", live) + } + beforeFlushInfo, err := os.Stat(app.opts.StatsPath) + if err != nil { + t.Fatal(err) + } + if !os.SameFile(initialInfo, beforeFlushInfo) { + t.Fatal("reading the live snapshot unexpectedly rewrote stats.json") + } + if _, exists := readPersistedServer(t, app)["cpu"]; exists { + t.Fatal("agent update reached disk before the next persistence run") + } + + if err := app.PersistStats(); err != nil { + t.Fatal(err) + } + afterFlushInfo, err := os.Stat(app.opts.StatsPath) + if err != nil { + t.Fatal(err) + } + if os.SameFile(beforeFlushInfo, afterFlushInfo) { + t.Fatal("updated stats were not persisted") + } + if cpu := readPersistedServer(t, app)["cpu"]; cpu != float64(42) { + t.Fatalf("unexpected persisted CPU value: %#v", cpu) + } + + if err := app.PersistStats(); err != nil { + t.Fatal(err) + } + cleanInfo, err := os.Stat(app.opts.StatsPath) + if err != nil { + t.Fatal(err) + } + if os.SameFile(afterFlushInfo, cleanInfo) { + t.Fatal("fixed-interval persistence unexpectedly skipped a write") + } +} + +func TestReloadNoticeIsIndependentFromPersistence(t *testing.T) { + app := newTestApp(t, minimalTestConfig()) + app.reloadPending.Store(true) + if err := app.PersistStats(); err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(app.opts.StatsPath) + if err != nil { + t.Fatal(err) + } + var persisted map[string]any + if err := json.Unmarshal(data, &persisted); err != nil { + t.Fatal(err) + } + if _, exists := persisted["reload"]; exists { + t.Fatal("reload notice must not be persisted to stats.json") + } + if reloaded, _ := app.SnapshotStats()["reload"].(bool); !reloaded { + t.Fatal("pending reload notice was not returned by the live endpoint") + } + if _, exists := app.SnapshotStats()["reload"]; exists { + t.Fatal("reload notice was returned more than once") + } +} + +func TestResetTrafficPersistsImmediately(t *testing.T) { + app := newTestApp(t, minimalTestConfig()) + app.nodeMu.Lock() + node := app.nodes["s01"] + node.Connected = true + node.HasUpdate = true + node.Stats = AgentStats{NetworkIn: 1234, NetworkOut: 5678} + app.nodeMu.Unlock() + + if _, apiErr := app.ResetTraffic("s01"); apiErr != nil { + t.Fatal(apiErr) + } + persisted := readPersistedServer(t, app) + if persisted["last_network_in"] != float64(1234) || persisted["last_network_out"] != float64(5678) { + t.Fatalf("traffic reset was not persisted immediately: %#v", persisted) + } +} diff --git a/server/ssl.go b/server/ssl.go index 90e925a..b45e0f7 100644 --- a/server/ssl.go +++ b/server/ssl.go @@ -98,7 +98,6 @@ func (a *App) executeSSLCheck(key string, config SSLCertConfig) { if err != nil { state.LastError = err.Error() a.certMu.Unlock() - a.wakeStatsWriter() return } state.ExpireTS = expireTS @@ -126,7 +125,6 @@ func (a *App) executeSSLCheck(key string, config SSLCertConfig) { } } a.certMu.Unlock() - a.wakeStatsWriter() for _, alert := range alerts { if err := a.sendCallback(config.Callback, alert.message, "ServerStatusSSL"); err != nil { diff --git a/server/tcp_server.go b/server/tcp_server.go index 9aef664..118d8f3 100644 --- a/server/tcp_server.go +++ b/server/tcp_server.go @@ -176,7 +176,6 @@ func (a *App) connectAgent(username, password string, conn net.Conn, family int) node.Pong = false node.Online4 = family == 4 node.Online6 = family == 6 - a.wakeStatsWriter() return id, append([]MonitorConfig(nil), a.runtime.Monitors...), nil } @@ -194,7 +193,6 @@ func (a *App) disconnectAgent(username string, conn net.Conn, connectionID uint6 node.HasUpdate = false node.Pong = false a.nodeMu.Unlock() - a.wakeStatsWriter() time.AfterFunc(25*time.Second, func() { if a.ctx.Err() != nil { return @@ -226,7 +224,6 @@ func (a *App) updateAgent(username string, connectionID uint64, update AgentStat node.HasUpdate = true node.LastUpdate = time.Now() a.nodeMu.Unlock() - a.wakeStatsWriter() a.evaluateWatchdogs(username, false) return true }