telemetry: fix total disk usage over time (#10476)

* telemetry: total disk usage over time counted each cluster on one day

GetMetrics aggregated s.instances, which holds only each cluster's most
recent report. Every cluster therefore landed in a single date bucket --
the day it last reported on -- so the chart plotted the disk usage of
clusters that went silent that day, and piled the whole live fleet onto
today. Aggregate the daily histories instead, reusing the day alignment
that the per-cluster size series already does.

* telemetry: don't pad the charts with days the server has no history for

The dashboard asks for 30 days, but daily history only starts when a
server first collects it, so the charts opened on a run of zeros and then
jumped -- reading as a fleet that appeared overnight. Start the window at
the oldest sample on hand when it is younger than the requested range.

telemetry/server/storage/history.go

@@ -45,6 +45,89 @@ func sameUTCDay(a, b int64) bool {
 	return ta.Year() == tb.Year() && ta.YearDay() == tb.YearDay()
 }
 
+// dailySeries is the shared date axis of the fleet-wide time series: one slot
+// per UTC day, ending today.
+type dailySeries struct {
+	dates []string
+	dayOf map[string]int
+}
+
+// newDailySeries builds the axis of UTC days ending today, spanning `days` days
+// but starting no earlier than the first day any cluster reported on: a fresh
+// server is asked for more days than it has history for, and padding those days
+// with zeros draws a climb out of nothing that never happened.
+func newDailySeries(days int, histories map[string][]HistorySample) dailySeries {
+	today := utcDay(time.Now().Unix())
+	requested := today.AddDate(0, 0, 1-days)
+
+	// Samples are appended in receive order, so [0] is a cluster's oldest.
+	var start time.Time
+	for _, history := range histories {
+		if len(history) == 0 {
+			continue
+		}
+		if first := utcDay(history[0].Ts); start.IsZero() || first.Before(start) {
+			start = first
+		}
+	}
+	if start.IsZero() || start.Before(requested) {
+		start = requested
+	}
+
+	n := int(today.Sub(start)/(24*time.Hour)) + 1
+	d := dailySeries{
+		dates: make([]string, n),
+		dayOf: make(map[string]int, n),
+	}
+	for i := range d.dates {
+		d.dates[i] = start.AddDate(0, 0, i).Format("2006-01-02")
+		d.dayOf[d.dates[i]] = i
+	}
+	return d
+}
+
+func utcDay(ts int64) time.Time {
+	t := time.Unix(ts, 0).UTC()
+	return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC)
+}
+
+func diskBytes(s HistorySample) uint64   { return s.TotalDiskBytes }
+func serverCount(s HistorySample) uint64 { return uint64(s.VolumeServerCount) }
+
+// align lays one cluster's history onto the axis, picking `value` out of each
+// sample. Clusters report roughly once a day at no fixed hour, so a day without
+// a report carries the previous value forward rather than dropping to zero; a
+// cluster that stopped reporting altogether ends at its last sample instead of
+// holding capacity forever. Reports false when the cluster has nothing in range.
+func (d dailySeries) align(history []HistorySample, activeSince int64, value func(HistorySample) uint64) ([]uint64, bool) {
+	out := make([]uint64, len(d.dates))
+	reported := make([]bool, len(d.dates))
+	first, last := -1, -1
+	for _, sample := range history {
+		i, ok := d.dayOf[time.Unix(sample.Ts, 0).UTC().Format("2006-01-02")]
+		if !ok {
+			continue
+		}
+		out[i], reported[i] = value(sample), true
+		if first < 0 {
+			first = i
+		}
+		last = i
+		if sample.Ts >= activeSince {
+			last = len(d.dates) - 1 // still reporting, so hold to the right edge
+		}
+	}
+	if first < 0 {
+		return nil, false
+	}
+	for i := first + 1; i <= last; i++ {
+		if !reported[i] {
+			out[i] = out[i-1]
+		}
+	}
+	return out, true
+}
+
 // 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) {

telemetry/server/storage/metrics_test.go

@@ -0,0 +1,117 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Every cluster counts towards every day it reported on, not just towards the
+// one day it last reported on.
+func TestGetMetricsSumsEachDay(t *testing.T) {
+	s := newPrometheusStorage(prometheus.NewRegistry())
+
+	// Reported every day of the window.
+	seedSamples(s, "daily", HistorySample{TotalDiskBytes: 300, VolumeServerCount: 3},
+		-9, -8, -7, -6, -5, -4, -3, -2, -1, 0)
+	// Stopped reporting past the active window: counts on its own days only.
+	seedSamples(s, "gone", HistorySample{TotalDiskBytes: 900, VolumeServerCount: 9}, -9, -8)
+
+	metrics, err := s.GetMetrics(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if got := metrics["dates"].([]string); len(got) != 10 {
+		t.Fatalf("dates = %v, want 10 days", got)
+	}
+	if got := metrics["disk_usage"].([]uint64); !equal(got,
+		[]uint64{1200, 1200, 300, 300, 300, 300, 300, 300, 300, 300}) {
+		t.Errorf("disk_usage = %v, want the fleet total per day", got)
+	}
+	if got := metrics["server_counts"].([]int64); !equalInt64(got,
+		[]int64{12, 12, 3, 3, 3, 3, 3, 3, 3, 3}) {
+		t.Errorf("server_counts = %v, want the fleet total per day", got)
+	}
+}
+
+// A cluster that skipped a day keeps its size on that day rather than dipping
+// the fleet total to zero and back.
+func TestGetMetricsCarriesSkippedDaysForward(t *testing.T) {
+	s := newPrometheusStorage(prometheus.NewRegistry())
+
+	seedSamples(s, "gappy", HistorySample{TotalDiskBytes: 500, VolumeServerCount: 5}, -3, -1)
+
+	metrics, err := s.GetMetrics(4)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := metrics["disk_usage"].([]uint64); !equal(got, []uint64{500, 500, 500, 500}) {
+		t.Errorf("disk_usage = %v, want the skipped days carried forward", got)
+	}
+}
+
+// The window starts at the oldest sample the server actually has, so a server
+// with less history than the caller asked for does not report a fleet that grew
+// out of nothing on its first day of data.
+func TestGetMetricsWindowStartsAtOldestSample(t *testing.T) {
+	s := newPrometheusStorage(prometheus.NewRegistry())
+	seedSamples(s, "recent", HistorySample{TotalDiskBytes: 100, VolumeServerCount: 1}, -2, -1, 0)
+
+	metrics, err := s.GetMetrics(30)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := metrics["dates"].([]string); len(got) != 3 {
+		t.Errorf("dates = %v, want the 3 days with history, not 30", got)
+	}
+	if got := metrics["disk_usage"].([]uint64); !equal(got, []uint64{100, 100, 100}) {
+		t.Errorf("disk_usage = %v, want no leading zero days", got)
+	}
+
+	// History reaching past the requested window still clips to the window.
+	seedSamples(s, "old", HistorySample{TotalDiskBytes: 50, VolumeServerCount: 1}, -40)
+	metrics, err = s.GetMetrics(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := metrics["dates"].([]string); len(got) != 10 {
+		t.Errorf("dates = %v, want 10 days", got)
+	}
+}
+
+// "Total Disk Usage Over Time" and the stacked "Cluster Sizes Over Time" sit on
+// the same dashboard, so their last day has to add up to the same number.
+func TestGetMetricsAgreesWithClusterSizes(t *testing.T) {
+	s := newPrometheusStorage(prometheus.NewRegistry())
+	seedSamples(s, "daily", HistorySample{TotalDiskBytes: 300, VolumeServerCount: 3},
+		-9, -8, -7, -6, -5, -4, -3, -2, -1, 0)
+	seedSamples(s, "lagging", HistorySample{TotalDiskBytes: 200, VolumeServerCount: 2}, -2)
+	seedSamples(s, "gone", HistorySample{TotalDiskBytes: 900, VolumeServerCount: 9}, -9, -8)
+
+	metrics, err := s.GetMetrics(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	disk := metrics["disk_usage"].([]uint64)
+	sizes := s.GetClusterSizeSeries(10, 1) // limit forces the Other fold-in too
+
+	if got, want := disk[len(disk)-1], sizes.TotalDisk; got != want {
+		t.Errorf("metrics last day = %d, cluster sizes total = %d", got, want)
+	}
+	if len(disk) != len(sizes.Dates) {
+		t.Errorf("metrics has %d days, cluster sizes has %d", len(disk), len(sizes.Dates))
+	}
+}
+
+func equalInt64(a, b []int64) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}

telemetry/server/storage/prometheus.go

@@ -1,7 +1,6 @@
 package storage
 
 import (
-	"sort"
 	"sync"
 	"time"
 
@@ -165,41 +164,35 @@ func (s *PrometheusStorage) GetInstances(limit int) ([]*telemetryData, error) {
 	return instances, nil
 }
 
+// GetMetrics returns fleet-wide daily totals for the last `days` days, in the
+// parallel-array shape the dashboard charts expect. Totals come from the daily
+// histories rather than from s.instances, which holds only each cluster's most
+// recent report and so would credit every cluster to the single day it last
+// reported on.
 func (s *PrometheusStorage) GetMetrics(days int) (map[string]interface{}, error) {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
-	// Return current metrics from in-memory storage, aggregated per day
-	// in the parallel-array shape the dashboard charts expect.
-	// Historical data should be queried from Prometheus directly.
-	cutoff := time.Now().AddDate(0, 0, -days)
+	axis := newDailySeries(days, s.histories)
+	activeSince := time.Now().UTC().AddDate(0, 0, -activeDays).Unix()
 
-	serverCountByDate := make(map[string]int64)
-	diskUsageByDate := make(map[string]uint64)
-
-	for _, instance := range s.instances {
-		if instance.ReceivedAt.After(cutoff) {
-			date := instance.ReceivedAt.Format("2006-01-02")
-			serverCountByDate[date] += int64(instance.TelemetryData.VolumeServerCount)
-			diskUsageByDate[date] += instance.TelemetryData.TotalDiskBytes
+	diskUsage := make([]uint64, len(axis.dates))
+	serverCounts := make([]int64, len(axis.dates))
+	for _, history := range s.histories {
+		if disk, ok := axis.align(history, activeSince, diskBytes); ok {
+			for i, v := range disk {
+				diskUsage[i] += v
+			}
+		}
+		if servers, ok := axis.align(history, activeSince, serverCount); ok {
+			for i, v := range servers {
+				serverCounts[i] += int64(v)
+			}
 		}
 	}
 
-	dates := make([]string, 0, len(serverCountByDate))
-	for date := range serverCountByDate {
-		dates = append(dates, date)
-	}
-	sort.Strings(dates)
-
-	serverCounts := make([]int64, 0, len(dates))
-	diskUsage := make([]uint64, 0, len(dates))
-	for _, date := range dates {
-		serverCounts = append(serverCounts, serverCountByDate[date])
-		diskUsage = append(diskUsage, diskUsageByDate[date])
-	}
-
 	return map[string]interface{}{
-		"dates":         dates,
+		"dates":         axis.dates,
 		"server_counts": serverCounts,
 		"disk_usage":    diskUsage,
 	}, nil

telemetry/server/storage/sizes.go

@@ -30,52 +30,23 @@ type ClusterSizeSeries struct {
 }
 
 // GetClusterSizeSeries returns the last `days` days of per-cluster disk usage.
-// Clusters beyond `limit` are folded into Other. Clusters report roughly once
-// a day at no fixed hour, so a day without a report carries the previous value
-// forward rather than dropping to zero; a cluster that stopped reporting
-// altogether ends at its last sample instead of holding capacity forever.
+// Clusters beyond `limit` are folded into Other.
 func (s *PrometheusStorage) GetClusterSizeSeries(days, limit int) ClusterSizeSeries {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
-	now := time.Now().UTC()
-	today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
-	dayOf := make(map[string]int, days)
-	series := ClusterSizeSeries{Dates: make([]string, days)}
-	for i := range series.Dates {
-		series.Dates[i] = today.AddDate(0, 0, i-days+1).Format("2006-01-02")
-		dayOf[series.Dates[i]] = i
-	}
-	activeSince := now.AddDate(0, 0, -activeDays).Unix()
+	axis := newDailySeries(days, s.histories)
+	activeSince := time.Now().UTC().AddDate(0, 0, -activeDays).Unix()
+	last := len(axis.dates) - 1
+	series := ClusterSizeSeries{Dates: axis.dates}
 
 	for id, history := range s.histories {
-		disk := make([]uint64, days)
-		reported := make([]bool, days)
-		first, last := -1, -1
-		for _, sample := range history {
-			i, ok := dayOf[time.Unix(sample.Ts, 0).UTC().Format("2006-01-02")]
-			if !ok {
-				continue
-			}
-			disk[i], reported[i] = sample.TotalDiskBytes, true
-			if first < 0 {
-				first = i
-			}
-			last = i
-			if sample.Ts >= activeSince {
-				last = days - 1 // still reporting, so hold its size to the right edge
-			}
-		}
-		if first < 0 {
+		disk, ok := axis.align(history, activeSince, diskBytes)
+		if !ok {
 			continue
 		}
-		for i := first + 1; i <= last; i++ {
-			if !reported[i] {
-				disk[i] = disk[i-1]
-			}
-		}
 		series.Clusters = append(series.Clusters, ClusterSeries{ClusterId: id, Disk: disk})
-		series.TotalDisk += disk[days-1]
+		series.TotalDisk += disk[last]
 	}
 	series.ClusterCount = len(series.Clusters)
 
@@ -83,8 +54,8 @@ func (s *PrometheusStorage) GetClusterSizeSeries(days, limit int) ClusterSizeSer
 	// edge, tie-breaking on id to keep the order stable across refreshes.
 	sort.Slice(series.Clusters, func(i, j int) bool {
 		a, b := series.Clusters[i], series.Clusters[j]
-		if a.Disk[days-1] != b.Disk[days-1] {
-			return a.Disk[days-1] > b.Disk[days-1]
+		if a.Disk[last] != b.Disk[last] {
+			return a.Disk[last] > b.Disk[last]
 		}
 		return a.ClusterId < b.ClusterId
 	})
@@ -92,7 +63,7 @@ func (s *PrometheusStorage) GetClusterSizeSeries(days, limit int) ClusterSizeSer
 	if limit > 0 && len(series.Clusters) > limit {
 		other := OtherSeries{
 			Count: len(series.Clusters) - limit,
-			Disk:  make([]uint64, days),
+			Disk:  make([]uint64, len(axis.dates)),
 		}
 		for _, c := range series.Clusters[limit:] {
 			for i, v := range c.Disk {

telemetry/server/storage/sizes_test.go

@@ -10,13 +10,17 @@ import (
 
 // seedHistory gives a cluster one sample per listed day offset (0 is today).
 func seedHistory(s *PrometheusStorage, id string, disk uint64, dayOffsets ...int) {
+	seedSamples(s, id, HistorySample{TotalDiskBytes: disk}, dayOffsets...)
+}
+
+// seedSamples is seedHistory for tests that care about more than disk usage.
+// The sample's Ts is filled in per day offset.
+func seedSamples(s *PrometheusStorage, id string, sample HistorySample, dayOffsets ...int) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	for _, offset := range dayOffsets {
-		s.histories[id] = append(s.histories[id], HistorySample{
-			Ts:             time.Now().AddDate(0, 0, offset).Unix(),
-			TotalDiskBytes: disk,
-		})
+		sample.Ts = time.Now().AddDate(0, 0, offset).Unix()
+		s.histories[id] = append(s.histories[id], sample)
 	}
 }
 
@@ -97,11 +101,16 @@ func TestClusterSizeSeriesUsesLatestDailySample(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	// Nothing was reported before today, so the axis is today alone rather
+	// than the full 3 days padded out with zeros.
 	series := s.GetClusterSizeSeries(3, 0)
 	if len(series.Clusters) != 1 {
 		t.Fatalf("clusters = %+v, want 1", series.Clusters)
 	}
-	if got := series.Clusters[0].Disk; !equal(got, []uint64{0, 0, 700}) {
+	if today := time.Now().UTC().Format("2006-01-02"); len(series.Dates) != 1 || series.Dates[0] != today {
+		t.Errorf("dates = %v, want %s only", series.Dates, today)
+	}
+	if got := series.Clusters[0].Disk; !equal(got, []uint64{700}) {
 		t.Errorf("disk = %v, want today's latest sample only", got)
 	}
 	if series.TotalDisk != 700 {