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 } }