diff --git a/telemetry/README.md b/telemetry/README.md index 8de08f911..b9ea6ac74 100644 --- a/telemetry/README.md +++ b/telemetry/README.md @@ -173,6 +173,9 @@ GET /api/instances?limit=100 # Get metrics over time GET /api/metrics?days=30 + +# Get one cluster's daily usage history (disk bytes, volumes, volume servers) +GET /api/history?cluster_id=&days=90 ``` ### Monitoring diff --git a/telemetry/server/api/handlers.go b/telemetry/server/api/handlers.go index c480a9771..b78e8c451 100644 --- a/telemetry/server/api/handlers.go +++ b/telemetry/server/api/handlers.go @@ -150,3 +150,35 @@ func (h *Handler) GetMetrics(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(metrics) } + +func (h *Handler) GetHistory(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + clusterId := r.URL.Query().Get("cluster_id") + if clusterId == "" { + http.Error(w, "cluster_id is required", http.StatusBadRequest) + return + } + + days := 90 // default + if daysStr := r.URL.Query().Get("days"); daysStr != "" { + if d, err := strconv.Atoi(daysStr); err == nil && d > 0 && d <= 365 { + days = d + } + } + + samples, ok := h.storage.GetHistory(clusterId, days) + if !ok { + http.Error(w, "Unknown cluster_id", http.StatusNotFound) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "cluster_id": clusterId, + "samples": samples, + }) +} diff --git a/telemetry/server/dashboard/dashboard.go b/telemetry/server/dashboard/dashboard.go index f60021bba..9e8af595f 100644 --- a/telemetry/server/dashboard/dashboard.go +++ b/telemetry/server/dashboard/dashboard.go @@ -74,6 +74,26 @@ func (h *Handler) ServeIndex(w http.ResponseWriter, r *http.Request) { padding: 40px; color: #666; } + .cluster-lookup { + display: flex; + gap: 10px; + margin-bottom: 15px; + } + .cluster-lookup input { + flex: 1; + padding: 8px 12px; + border: 1px solid #ddd; + border-radius: 4px; + font-family: monospace; + } + .cluster-lookup button { + padding: 8px 20px; + border: none; + border-radius: 4px; + background: #2196F3; + color: white; + cursor: pointer; + } .error { background: #ffebee; color: #c62828; @@ -134,6 +154,20 @@ func (h *Handler) ServeIndex(w http.ResponseWriter, r *http.Request) {
Total Disk Usage Over Time
+ +
+
Per-Cluster History
+
+ + +
+ + +
@@ -253,6 +287,35 @@ func (h *Handler) ServeIndex(w http.ResponseWriter, r *http.Request) { }); } + async function loadClusterHistory() { + const id = document.getElementById('clusterIdInput').value.trim(); + const errorDiv = document.getElementById('clusterHistoryError'); + const chartsDiv = document.getElementById('clusterHistoryCharts'); + if (!id) return; + errorDiv.style.display = 'none'; + try { + const resp = await fetch('/api/history?cluster_id=' + encodeURIComponent(id) + '&days=90'); + if (!resp.ok) { + throw new Error(resp.status === 404 ? 'Unknown cluster UUID' : 'Request failed: ' + resp.status); + } + const history = await resp.json(); + const samples = history.samples || []; + if (samples.length === 0) { + throw new Error('No samples recorded for this cluster yet'); + } + const dates = samples.map(s => new Date(s.ts * 1000).toISOString().slice(0, 10)); + chartsDiv.style.display = 'block'; + createLineChart('clusterDiskChart', 'Disk Usage (GB)', dates, + samples.map(s => Math.round(s.disk / (1024 * 1024 * 1024) * 100) / 100), '#FF9800'); + createLineChart('clusterVolumeChart', 'Volumes', dates, + samples.map(s => s.volumes), '#9C27B0'); + } catch (error) { + chartsDiv.style.display = 'none'; + errorDiv.style.display = 'block'; + errorDiv.textContent = error.message; + } + } + function showError(message) { document.getElementById('loading').style.display = 'none'; document.getElementById('error').style.display = 'block'; diff --git a/telemetry/server/main.go b/telemetry/server/main.go index 5a190b387..05ebec8dc 100644 --- a/telemetry/server/main.go +++ b/telemetry/server/main.go @@ -75,6 +75,7 @@ func main() { mux.HandleFunc("/api/stats", corsMiddleware(logMiddleware(apiHandler.GetStats))) mux.HandleFunc("/api/instances", corsMiddleware(logMiddleware(apiHandler.GetInstances))) mux.HandleFunc("/api/metrics", corsMiddleware(logMiddleware(apiHandler.GetMetrics))) + mux.HandleFunc("/api/history", corsMiddleware(logMiddleware(apiHandler.GetHistory))) // Dashboard (optional) if *enableDashboard { diff --git a/telemetry/server/storage/history.go b/telemetry/server/storage/history.go new file mode 100644 index 000000000..8f5e26f45 --- /dev/null +++ b/telemetry/server/storage/history.go @@ -0,0 +1,59 @@ +package storage + +import ( + "time" + + "github.com/seaweedfs/seaweedfs/telemetry/proto" +) + +// HistorySample is one retained data point of a cluster's daily reports. +// Tags are kept short because thousands of samples end up in the state file. +type HistorySample struct { + Ts int64 `json:"ts"` // unix seconds the report was received + TotalDiskBytes uint64 `json:"disk"` + TotalVolumeCount int32 `json:"volumes"` + VolumeServerCount int32 `json:"servers"` +} + +// appendHistory records the report as the cluster's sample for the day, +// replacing an earlier sample from the same UTC day. Callers must hold s.mu. +func (s *PrometheusStorage) appendHistory(data *proto.TelemetryData, receivedAt time.Time) { + sample := HistorySample{ + Ts: receivedAt.Unix(), + TotalDiskBytes: data.TotalDiskBytes, + TotalVolumeCount: data.TotalVolumeCount, + VolumeServerCount: data.VolumeServerCount, + } + h := s.histories[data.TopologyId] + if n := len(h); n > 0 && sameUTCDay(h[n-1].Ts, sample.Ts) { + h[n-1] = sample + } else { + h = append(h, sample) + } + s.histories[data.TopologyId] = h +} + +func sameUTCDay(a, b int64) bool { + ta, tb := time.Unix(a, 0).UTC(), time.Unix(b, 0).UTC() + return ta.Year() == tb.Year() && ta.YearDay() == tb.YearDay() +} + +// GetHistory returns the cluster's samples from the last `days` days. +// The second return value reports whether the cluster is known at all. +func (s *PrometheusStorage) GetHistory(clusterId string, days int) ([]HistorySample, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + + h, ok := s.histories[clusterId] + if !ok { + return nil, false + } + cutoff := time.Now().AddDate(0, 0, -days).Unix() + samples := make([]HistorySample, 0, len(h)) + for _, sample := range h { + if sample.Ts >= cutoff { + samples = append(samples, sample) + } + } + return samples, true +} diff --git a/telemetry/server/storage/history_test.go b/telemetry/server/storage/history_test.go new file mode 100644 index 000000000..0f3775eed --- /dev/null +++ b/telemetry/server/storage/history_test.go @@ -0,0 +1,85 @@ +package storage + +import ( + "path/filepath" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/seaweedfs/seaweedfs/telemetry/proto" +) + +func TestClusterHistory(t *testing.T) { + s := newPrometheusStorage(prometheus.NewRegistry()) + + report := &proto.TelemetryData{ + TopologyId: "hist-cluster", + Version: "4.40", + Os: "linux/amd64", + VolumeServerCount: 3, + TotalDiskBytes: 1000, + TotalVolumeCount: 10, + } + if err := s.StoreTelemetry(report); err != nil { + t.Fatalf("store: %v", err) + } + + // A second report on the same UTC day replaces the day's sample. + report.TotalDiskBytes = 2000 + if err := s.StoreTelemetry(report); err != nil { + t.Fatalf("store: %v", err) + } + samples, ok := s.GetHistory("hist-cluster", 90) + if !ok { + t.Fatal("cluster missing from history") + } + if len(samples) != 1 { + t.Fatalf("got %d samples, want 1 (same-day replace)", len(samples)) + } + if samples[0].TotalDiskBytes != 2000 { + t.Errorf("same-day sample not replaced: got %d", samples[0].TotalDiskBytes) + } + + // An older sample from a previous day is appended and survives the + // state-file round trip. + s.mu.Lock() + s.histories["hist-cluster"] = append([]HistorySample{{ + Ts: time.Now().AddDate(0, 0, -5).Unix(), + TotalDiskBytes: 500, + }}, s.histories["hist-cluster"]...) + s.mu.Unlock() + + path := filepath.Join(t.TempDir(), "state.json") + if err := s.SaveStateIfDirty(path); err != nil { + t.Fatalf("save: %v", err) + } + s.instances = make(map[string]*telemetryData) + s.histories = make(map[string][]HistorySample) + if _, err := s.LoadState(path); err != nil { + t.Fatalf("load: %v", err) + } + samples, ok = s.GetHistory("hist-cluster", 90) + if !ok || len(samples) != 2 { + t.Fatalf("after round trip: ok=%v samples=%d, want 2", ok, len(samples)) + } + if samples[0].TotalDiskBytes != 500 || samples[1].TotalDiskBytes != 2000 { + t.Errorf("samples corrupted after round trip: %+v", samples) + } + + // GetHistory filters by the requested window. + samples, _ = s.GetHistory("hist-cluster", 3) + if len(samples) != 1 { + t.Errorf("window filter: got %d samples, want 1", len(samples)) + } + + // Unknown cluster reports !ok. + if _, ok := s.GetHistory("nope", 90); ok { + t.Error("unknown cluster reported ok") + } + + // Cleanup drops the history together with the instance. + s.CleanupOldInstances(0) + if _, ok := s.GetHistory("hist-cluster", 90); ok { + t.Error("history survived instance cleanup") + } +} diff --git a/telemetry/server/storage/persistence.go b/telemetry/server/storage/persistence.go index 984038b28..009108c84 100644 --- a/telemetry/server/storage/persistence.go +++ b/telemetry/server/storage/persistence.go @@ -9,7 +9,8 @@ import ( // persistedState is the on-disk snapshot of the in-memory instance map. type persistedState struct { - Instances map[string]*telemetryData `json:"instances"` + Instances map[string]*telemetryData `json:"instances"` + Histories map[string][]HistorySample `json:"histories,omitempty"` } // LoadState restores the instance map and Prometheus gauges from a state file @@ -42,6 +43,11 @@ func (s *PrometheusStorage) LoadState(path string) (int, error) { s.setClusterMetrics(instance.TelemetryData) loaded++ } + for id, h := range state.Histories { + if _, ok := s.instances[id]; ok && len(h) > 0 { + s.histories[id] = h + } + } s.updateStats() return loaded, nil } @@ -54,7 +60,7 @@ func (s *PrometheusStorage) SaveStateIfDirty(path string) error { s.mu.Unlock() return nil } - b, err := json.Marshal(&persistedState{Instances: s.instances}) + b, err := json.Marshal(&persistedState{Instances: s.instances, Histories: s.histories}) if err != nil { s.mu.Unlock() return err diff --git a/telemetry/server/storage/persistence_test.go b/telemetry/server/storage/persistence_test.go index 18f747f5a..62b949ef8 100644 --- a/telemetry/server/storage/persistence_test.go +++ b/telemetry/server/storage/persistence_test.go @@ -5,15 +5,14 @@ import ( "testing" "time" + "github.com/prometheus/client_golang/prometheus" "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() + s := newPrometheusStorage(prometheus.NewRegistry()) if err := s.SaveStateIfDirty(path); err != nil { t.Fatalf("clean save: %v", err) diff --git a/telemetry/server/storage/prometheus.go b/telemetry/server/storage/prometheus.go index f4f43472e..3c7869c82 100644 --- a/telemetry/server/storage/prometheus.go +++ b/telemetry/server/storage/prometheus.go @@ -25,6 +25,7 @@ type PrometheusStorage struct { // In-memory storage for API endpoints (if needed) mu sync.RWMutex instances map[string]*telemetryData + histories map[string][]HistorySample stats map[string]interface{} dirty bool // instances changed since the last successful state save } @@ -36,6 +37,11 @@ type telemetryData struct { } func NewPrometheusStorage() *PrometheusStorage { + return newPrometheusStorage(prometheus.DefaultRegisterer) +} + +func newPrometheusStorage(reg prometheus.Registerer) *PrometheusStorage { + promauto := promauto.With(reg) return &PrometheusStorage{ totalClusters: promauto.NewGauge(prometheus.GaugeOpts{ Name: "seaweedfs_telemetry_total_clusters", @@ -74,6 +80,7 @@ func NewPrometheusStorage() *PrometheusStorage { Help: "Total number of telemetry reports received", }), instances: make(map[string]*telemetryData), + histories: make(map[string][]HistorySample), stats: make(map[string]interface{}), } } @@ -94,10 +101,12 @@ func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error { s.telemetryReceived.Inc() // Store in memory for API endpoints + receivedAt := time.Now().UTC() s.instances[data.TopologyId] = &telemetryData{ TelemetryData: data, - ReceivedAt: time.Now().UTC(), + ReceivedAt: receivedAt, } + s.appendHistory(data, receivedAt) s.dirty = true // Update aggregated stats @@ -239,6 +248,22 @@ func (s *PrometheusStorage) CleanupOldInstances(maxAge time.Duration) { } } + for id, h := range s.histories { + if _, ok := s.instances[id]; !ok { + delete(s.histories, id) + s.dirty = true + continue + } + i := 0 + for i < len(h) && time.Unix(h[i].Ts, 0).Before(cutoff) { + i++ + } + if i > 0 { + s.histories[id] = append([]HistorySample(nil), h[i:]...) + s.dirty = true + } + } + s.updateStats() }