diff --git a/telemetry/server/storage/history.go b/telemetry/server/storage/history.go index 2ec6a1a93..4321f9c82 100644 --- a/telemetry/server/storage/history.go +++ b/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) { diff --git a/telemetry/server/storage/metrics_test.go b/telemetry/server/storage/metrics_test.go new file mode 100644 index 000000000..8a55a2004 --- /dev/null +++ b/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 +} diff --git a/telemetry/server/storage/prometheus.go b/telemetry/server/storage/prometheus.go index 10092e648..ee88e98a7 100644 --- a/telemetry/server/storage/prometheus.go +++ b/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 diff --git a/telemetry/server/storage/sizes.go b/telemetry/server/storage/sizes.go index aa2f7f287..34f3204a3 100644 --- a/telemetry/server/storage/sizes.go +++ b/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 { diff --git a/telemetry/server/storage/sizes_test.go b/telemetry/server/storage/sizes_test.go index 54af4fd43..4c9948e7f 100644 --- a/telemetry/server/storage/sizes_test.go +++ b/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 {