From 3e9154def2380ac60132593ba6bbeef94d280b64 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 22 Jul 2026 19:33:32 -0700 Subject: [PATCH] telemetry: persist server state across restarts (#10399) * telemetry: persist server state across restarts The telemetry server kept the instance map and Prometheus gauges only in process memory, so every deploy or restart reset all collected metrics until clusters re-reported over the next 24h. Snapshot the instance map to a JSON state file (atomic tmp+rename) on a debounced interval and on SIGTERM, and restore it on startup, preserving received_at so the cleanup and active-cluster windows stay correct. Defaults to data/telemetry-state.json, which the deployed systemd unit's WorkingDirectory already provides; -state-file='' disables. * telemetry: keep instances for 90 days by default With state now persisted across restarts, a longer retention default is meaningful; raise -max-age from 30 to 90 days so per-cluster data survives long enough for quarterly views. --- telemetry/DEPLOYMENT.md | 2 +- telemetry/README.md | 6 +- telemetry/server/main.go | 48 +++++++++- telemetry/server/storage/persistence.go | 93 ++++++++++++++++++++ telemetry/server/storage/persistence_test.go | 75 ++++++++++++++++ telemetry/server/storage/prometheus.go | 34 ++++--- 6 files changed, 238 insertions(+), 20 deletions(-) create mode 100644 telemetry/server/storage/persistence.go create mode 100644 telemetry/server/storage/persistence_test.go diff --git a/telemetry/DEPLOYMENT.md b/telemetry/DEPLOYMENT.md index a1dd54907..5b9a3aa62 100644 --- a/telemetry/DEPLOYMENT.md +++ b/telemetry/DEPLOYMENT.md @@ -215,7 +215,7 @@ After setup, the remote server will have: ├── logs/ │ ├── telemetry.log # Application logs │ └── telemetry.error.log # Error logs -├── data/ # Data directory (if needed) +├── data/ # State snapshots (telemetry-state.json), restored on restart ├── grafana-dashboard.json # Grafana dashboard configuration └── prometheus.yml # Prometheus configuration ``` diff --git a/telemetry/README.md b/telemetry/README.md index 6f1e4797b..8de08f911 100644 --- a/telemetry/README.md +++ b/telemetry/README.md @@ -121,10 +121,12 @@ weed master -telemetry=true -telemetry.url=http://localhost:8080/api/collect -port=8080 # Server port -dashboard=true # Enable built-in dashboard -cleanup=24h # Cleanup interval --max-age=720h # Maximum data retention (30 days) +-max-age=2160h # Maximum data retention (90 days) +-state-file=data/telemetry-state.json # Persist state across restarts (empty to disable) +-state-save=1h # How often to save changed state # Example -./telemetry-server -port=8080 -dashboard=true -cleanup=24h -max-age=720h +./telemetry-server -port=8080 -dashboard=true -cleanup=24h -max-age=2160h ``` ## Prometheus Metrics diff --git a/telemetry/server/main.go b/telemetry/server/main.go index 6cbae05c7..5a190b387 100644 --- a/telemetry/server/main.go +++ b/telemetry/server/main.go @@ -1,11 +1,15 @@ package main import ( + "context" "encoding/json" "flag" "fmt" "log" "net/http" + "os" + "os/signal" + "syscall" "time" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -20,7 +24,9 @@ var ( logRequests = flag.Bool("log", true, "Log incoming requests") enableDashboard = flag.Bool("dashboard", true, "Enable built-in dashboard (optional when using Grafana)") cleanupInterval = flag.Duration("cleanup", 24*time.Hour, "Cleanup interval for old instances") - maxInstanceAge = flag.Duration("max-age", 30*24*time.Hour, "Maximum age for instances before cleanup") + maxInstanceAge = flag.Duration("max-age", 90*24*time.Hour, "Maximum age for instances before cleanup") + stateFile = flag.String("state-file", "data/telemetry-state.json", "File for persisting in-memory state across restarts (empty to disable)") + saveInterval = flag.Duration("state-save", time.Hour, "How often to save changed state to -state-file") ) func main() { @@ -29,6 +35,25 @@ func main() { // Create Prometheus storage instance store := storage.NewPrometheusStorage() + // Restore state from the previous run and keep saving it periodically, + // so deploys and restarts don't reset the collected metrics + if *stateFile != "" { + if n, err := store.LoadState(*stateFile); err != nil { + log.Printf("Failed to load state from %s: %v", *stateFile, err) + } else if n > 0 { + log.Printf("Restored %d instances from %s", n, *stateFile) + } + go func() { + ticker := time.NewTicker(*saveInterval) + defer ticker.Stop() + for range ticker.C { + if err := store.SaveStateIfDirty(*stateFile); err != nil { + log.Printf("Failed to save state to %s: %v", *stateFile, err) + } + } + }() + } + // Start cleanup routine go func() { ticker := time.NewTicker(*cleanupInterval) @@ -76,8 +101,25 @@ func main() { } log.Printf("Cleanup interval: %v, Max instance age: %v", *cleanupInterval, *maxInstanceAge) - if err := http.ListenAndServe(addr, mux); err != nil { - log.Fatalf("Server failed: %v", err) + server := &http.Server{Addr: addr, Handler: mux} + go func() { + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("Server failed: %v", err) + } + }() + + // On SIGTERM/SIGINT, stop serving and save state before exiting + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) + <-sig + log.Printf("Shutting down...") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + server.Shutdown(ctx) + if *stateFile != "" { + if err := store.SaveStateIfDirty(*stateFile); err != nil { + log.Printf("Failed to save state to %s: %v", *stateFile, err) + } } } diff --git a/telemetry/server/storage/persistence.go b/telemetry/server/storage/persistence.go new file mode 100644 index 000000000..984038b28 --- /dev/null +++ b/telemetry/server/storage/persistence.go @@ -0,0 +1,93 @@ +package storage + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" +) + +// persistedState is the on-disk snapshot of the in-memory instance map. +type persistedState struct { + Instances map[string]*telemetryData `json:"instances"` +} + +// LoadState restores the instance map and Prometheus gauges from a state file +// written by SaveStateIfDirty. A missing file is not an error. Original +// ReceivedAt timestamps are preserved so cleanup and the active-cluster +// windows stay correct across restarts. +func (s *PrometheusStorage) LoadState(path string) (int, error) { + b, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return 0, nil + } + return 0, err + } + + var state persistedState + if err := json.Unmarshal(b, &state); err != nil { + return 0, fmt.Errorf("parse %s: %w", path, err) + } + + s.mu.Lock() + defer s.mu.Unlock() + + loaded := 0 + for id, instance := range state.Instances { + if instance == nil || instance.TelemetryData == nil || instance.TelemetryData.TopologyId == "" { + continue + } + s.instances[id] = instance + s.setClusterMetrics(instance.TelemetryData) + loaded++ + } + s.updateStats() + return loaded, nil +} + +// SaveStateIfDirty writes the instance map to path if it changed since the +// last successful save. The write is atomic (temp file + rename). +func (s *PrometheusStorage) SaveStateIfDirty(path string) error { + s.mu.Lock() + if !s.dirty { + s.mu.Unlock() + return nil + } + b, err := json.Marshal(&persistedState{Instances: s.instances}) + if err != nil { + s.mu.Unlock() + return err + } + s.dirty = false + s.mu.Unlock() + + if err := s.writeAtomically(path, b); err != nil { + s.mu.Lock() + s.dirty = true + s.mu.Unlock() + return err + } + return nil +} + +func (s *PrometheusStorage) writeAtomically(path string, b []byte) error { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0755); err != nil { + return err + } + tmp, err := os.CreateTemp(dir, filepath.Base(path)+".tmp-*") + if err != nil { + return err + } + if _, err := tmp.Write(b); err != nil { + tmp.Close() + os.Remove(tmp.Name()) + return err + } + if err := tmp.Close(); err != nil { + os.Remove(tmp.Name()) + return err + } + return os.Rename(tmp.Name(), path) +} diff --git a/telemetry/server/storage/persistence_test.go b/telemetry/server/storage/persistence_test.go new file mode 100644 index 000000000..18f747f5a --- /dev/null +++ b/telemetry/server/storage/persistence_test.go @@ -0,0 +1,75 @@ +package storage + +import ( + "path/filepath" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/telemetry/proto" +) + +func TestStateRoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "data", "telemetry-state.json") + + // promauto registers on the global registry, so the whole test uses a + // single storage instance. + s := NewPrometheusStorage() + + if err := s.SaveStateIfDirty(path); err != nil { + t.Fatalf("clean save: %v", err) + } + if _, err := s.LoadState(path); err != nil { + t.Fatalf("load with no state file: %v", err) + } + + report := &proto.TelemetryData{ + TopologyId: "test-cluster-1", + Version: "4.40", + Os: "linux/amd64", + VolumeServerCount: 5, + TotalDiskBytes: 123456789, + TotalVolumeCount: 42, + FilerCount: 2, + BrokerCount: 1, + Timestamp: time.Now().Unix(), + } + if err := s.StoreTelemetry(report); err != nil { + t.Fatalf("store: %v", err) + } + receivedAt := s.instances[report.TopologyId].ReceivedAt + + if err := s.SaveStateIfDirty(path); err != nil { + t.Fatalf("save: %v", err) + } + if s.dirty { + t.Fatal("dirty flag not cleared after save") + } + + // Simulate a restart: wipe the in-memory map, then restore. + s.instances = make(map[string]*telemetryData) + n, err := s.LoadState(path) + if err != nil { + t.Fatalf("load: %v", err) + } + if n != 1 { + t.Fatalf("loaded %d instances, want 1", n) + } + + got, ok := s.instances[report.TopologyId] + if !ok { + t.Fatal("instance missing after load") + } + if !got.ReceivedAt.Equal(receivedAt) { + t.Errorf("ReceivedAt not preserved: got %v, want %v", got.ReceivedAt, receivedAt) + } + if got.TelemetryData.TotalDiskBytes != report.TotalDiskBytes || + got.TelemetryData.Version != report.Version || + got.TelemetryData.VolumeServerCount != report.VolumeServerCount { + t.Errorf("fields not preserved: got %+v", got.TelemetryData) + } + + // Load must not mark state dirty. + if s.dirty { + t.Error("dirty flag set after load") + } +} diff --git a/telemetry/server/storage/prometheus.go b/telemetry/server/storage/prometheus.go index a4103ff9e..f4f43472e 100644 --- a/telemetry/server/storage/prometheus.go +++ b/telemetry/server/storage/prometheus.go @@ -26,6 +26,7 @@ type PrometheusStorage struct { mu sync.RWMutex instances map[string]*telemetryData stats map[string]interface{} + dirty bool // instances changed since the last successful state save } // telemetryData is an internal struct that includes the received timestamp @@ -81,19 +82,6 @@ func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error { s.mu.Lock() defer s.mu.Unlock() - // Update Prometheus metrics. Value gauges are keyed by cluster_id only so - // a cluster's series continues across upgrades; version/os metadata lives - // on cluster_info (join with `* on(cluster_id) group_left(version, os)`). - labels := prometheus.Labels{ - "cluster_id": data.TopologyId, - } - - s.volumeServerCount.With(labels).Set(float64(data.VolumeServerCount)) - s.totalDiskBytes.With(labels).Set(float64(data.TotalDiskBytes)) - s.totalVolumeCount.With(labels).Set(float64(data.TotalVolumeCount)) - s.filerCount.With(labels).Set(float64(data.FilerCount)) - s.brokerCount.With(labels).Set(float64(data.BrokerCount)) - // Drop the cluster_info series recorded under the previous label set when // a cluster reports back with a different version or OS, so it is not // counted under two versions at once. @@ -101,7 +89,7 @@ func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error { (prev.TelemetryData.Version != data.Version || prev.TelemetryData.Os != data.Os) { s.clusterInfo.Delete(infoLabels(prev.TelemetryData)) } - s.clusterInfo.With(infoLabels(data)).Set(1) + s.setClusterMetrics(data) s.telemetryReceived.Inc() @@ -110,6 +98,7 @@ func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error { TelemetryData: data, ReceivedAt: time.Now().UTC(), } + s.dirty = true // Update aggregated stats s.updateStats() @@ -117,6 +106,22 @@ func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error { return nil } +// setClusterMetrics records a report's values on the Prometheus gauges. +// Value gauges are keyed by cluster_id only so a cluster's series continues +// across upgrades; version/os metadata lives on cluster_info (join with +// `* on(cluster_id) group_left(version, os)`). Callers must hold s.mu. +func (s *PrometheusStorage) setClusterMetrics(data *proto.TelemetryData) { + labels := prometheus.Labels{ + "cluster_id": data.TopologyId, + } + s.volumeServerCount.With(labels).Set(float64(data.VolumeServerCount)) + s.totalDiskBytes.With(labels).Set(float64(data.TotalDiskBytes)) + s.totalVolumeCount.With(labels).Set(float64(data.TotalVolumeCount)) + s.filerCount.With(labels).Set(float64(data.FilerCount)) + s.brokerCount.With(labels).Set(float64(data.BrokerCount)) + s.clusterInfo.With(infoLabels(data)).Set(1) +} + func (s *PrometheusStorage) GetStats() (map[string]interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() @@ -230,6 +235,7 @@ func (s *PrometheusStorage) CleanupOldInstances(maxAge time.Duration) { if instance.ReceivedAt.Before(cutoff) { delete(s.instances, instanceID) s.deleteClusterMetrics(instance.TelemetryData) + s.dirty = true } }