telemetry: per-cluster usage history on the built-in dashboard (#10400)

telemetry: per-cluster usage history

Keep one compact sample per cluster per UTC day (disk bytes, volume
count, volume servers), retained for -max-age and persisted in the
state file. Serve it at /api/history?cluster_id=...&days=90 and add a
per-cluster lookup with disk/volume charts to the built-in dashboard.

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=<uuid>&days=90
 ```
 
 ### Monitoring

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,
+	})
+}

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) {
                 <div class="chart-title">Total Disk Usage Over Time</div>
                 <canvas id="diskChart" width="400" height="200"></canvas>
             </div>
+
+            <div class="chart-container">
+                <div class="chart-title">Per-Cluster History</div>
+                <div class="cluster-lookup">
+                    <input type="text" id="clusterIdInput" placeholder="cluster UUID"
+                           onkeydown="if (event.key === 'Enter') loadClusterHistory()">
+                    <button onclick="loadClusterHistory()">Load</button>
+                </div>
+                <div id="clusterHistoryError" class="error" style="display: none;"></div>
+                <div id="clusterHistoryCharts" style="display: none;">
+                    <canvas id="clusterDiskChart" width="400" height="150"></canvas>
+                    <canvas id="clusterVolumeChart" width="400" height="150"></canvas>
+                </div>
+            </div>
         </div>
     </div>
 
@@ -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';

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 {

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

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")
+	}
+}

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

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)

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