diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index fbb0f07eb6..f09880c2b9 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,20 @@ ## Purpose +### Lightweight detached History reads + +Guest, node, disk and storage History readers allocate one exact-sized snapshot +per populated metric series, rather than repeatedly growing temporary arrays. +Empty series remain non-nil arrays for the existing JSON representation. The +copy preserves the exclusive time cutoff, original values, timestamps, order +and duplicates, including non-chronological input; it never aliases mutable +retained History. Single/batch chart and metric-window consumers retain the +same observations. Retention, sampling, source authority and SQL are unchanged. +`metrics_history_read_test.go` covers every public series, caller mutation, +cutoff/empty/unordered semantics and connected poll-to-chart/window readers. +Allocation controls and paired same-executable in-memory benchmarks measure +copy cost, not native CPU/RSS relief or attribution of an existing report. + ### Used-memory history retention Periodic in-memory History cleanup applies the existing retention window and diff --git a/internal/monitoring/metrics_history.go b/internal/monitoring/metrics_history.go index 928356ced2..d32f1b9e54 100644 --- a/internal/monitoring/metrics_history.go +++ b/internal/monitoring/metrics_history.go @@ -457,15 +457,7 @@ func (mh *MetricsHistory) GetGuestMetrics(guestID string, metricType string, dur return []MetricPoint{} } - // Filter by duration - result := make([]MetricPoint, 0) - for _, point := range data { - if point.Timestamp.After(cutoffTime) { - result = append(result, point) - } - } - - return result + return filterMetricsByTime(data, cutoffTime) } // GetNodeMetrics returns historical metrics for a node @@ -500,23 +492,38 @@ func (mh *MetricsHistory) GetNodeMetrics(nodeID string, metricType string, durat return []MetricPoint{} } - // Filter by duration - result := make([]MetricPoint, 0) - for _, point := range data { - if point.Timestamp.After(cutoffTime) { - result = append(result, point) - } - } - - return result + return filterMetricsByTime(data, cutoffTime) } // filterMetricsByTime returns only the points whose timestamp is after cutoffTime. func filterMetricsByTime(data []MetricPoint, cutoffTime time.Time) []MetricPoint { - filtered := make([]MetricPoint, 0) + // Skip the expired prefix, then count the survivors before allocating. + // Dense chart reads otherwise repeatedly grow and copy their result under + // the History read lock. Allocate only the requested window, not all stored + // points, and keep caller-owned snapshots rather than aliasing live History. + first := 0 + for first < len(data) && !data[first].Timestamp.After(cutoffTime) { + first++ + } + data = data[first:] + count := 0 for _, point := range data { if point.Timestamp.After(cutoffTime) { - filtered = append(filtered, point) + count++ + } + } + filtered := make([]MetricPoint, count) // Preserve [] (not null) for empty series. + if count == len(data) { + copy(filtered, data) + return filtered + } + // Usually the retained window is a contiguous suffix. Preserve the original + // per-point filtering and order even when an input is not chronological. + next := 0 + for _, point := range data { + if point.Timestamp.After(cutoffTime) { + filtered[next] = point + next++ } } return filtered diff --git a/internal/monitoring/metrics_history_read_test.go b/internal/monitoring/metrics_history_read_test.go new file mode 100644 index 0000000000..85c36eaa29 --- /dev/null +++ b/internal/monitoring/metrics_history_read_test.go @@ -0,0 +1,262 @@ +package monitoring + +import ( + "encoding/json" + "fmt" + "reflect" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/mock" +) + +// Preserve the original reader as a test-only reference. Paired benchmarks use +// the same executable, inputs and cutoff; no SQL or collector is exercised. +func legacyHistoryReadFilter(data []MetricPoint, cutoff time.Time) []MetricPoint { + filtered := make([]MetricPoint, 0) + for _, point := range data { + if point.Timestamp.After(cutoff) { + filtered = append(filtered, point) + } + } + return filtered +} + +func historyReadFixture(size, retained int, interleaved bool) ([]MetricPoint, time.Time) { + cutoff := time.Unix(1700000000, 0).UTC() + data := make([]MetricPoint, size) + for i := range data { + offset := i - (size - retained) + 1 + if interleaved { + offset = i + 1 + if i%2 == 0 { + offset = -offset + } + } + data[i] = MetricPoint{Timestamp: cutoff.Add(time.Duration(offset) * time.Second), Value: float64(i)} + } + return data, cutoff +} + +func TestHistoryReadFilterPreservesSemantics(t *testing.T) { + cutoff := time.Unix(1700000000, 0).UTC() + before := MetricPoint{Timestamp: cutoff.Add(-time.Second), Value: -1} + equal := MetricPoint{Timestamp: cutoff.In(time.FixedZone("other", 3600)), Value: 0} + after := MetricPoint{Timestamp: cutoff.Add(time.Second), Value: 2} + later := MetricPoint{Timestamp: cutoff.Add(2 * time.Second), Value: 3} + for name, data := range map[string][]MetricPoint{ + "nil": nil, "empty": {}, "expired": {before, equal}, + "all": {after, later}, "prefix": {before, equal, after, later}, + "unordered": {later, before, after, equal}, "duplicates": {before, after, after}, + } { + t.Run(name, func(t *testing.T) { + original := append([]MetricPoint(nil), data...) + want := legacyHistoryReadFilter(data, cutoff) + got := filterMetricsByTime(data, cutoff) + if !reflect.DeepEqual(got, want) { + t.Fatalf("filtered points = %#v, want %#v", got, want) + } + if got == nil { + t.Fatal("empty History arrays must remain [] rather than null") + } + if len(got) > 0 { + got[0].Value = -999 + got[0].Timestamp = time.Time{} + if !reflect.DeepEqual(data, original) { + t.Fatal("caller mutation changed retained History") + } + } + }) + } +} + +func TestHistoryReadFilterAllocationBudget(t *testing.T) { + for _, tc := range []struct { + name string + size, retained int + interleaved bool + }{ + {"empty", 0, 0, false}, {"expired", 2880, 0, false}, + {"one", 2880, 1, false}, {"hour", 2880, 60, false}, + {"day", 2880, 1440, false}, {"all", 2880, 2880, false}, + {"interleaved", 2880, 1440, true}, + } { + t.Run(tc.name, func(t *testing.T) { + data, cutoff := historyReadFixture(tc.size, tc.retained, tc.interleaved) + var snapshot []MetricPoint + allocs := testing.AllocsPerRun(25, func() { snapshot = filterMetricsByTime(data, cutoff) }) + budget := float64(1) + if tc.retained == 0 { + budget = 0 + } + t.Logf("retained=%d input=%d allocations=%.0f capacity=%d", len(snapshot), len(data), allocs, cap(snapshot)) + if allocs > budget { + t.Errorf("snapshot allocated %.0f times, want at most %.0f", allocs, budget) + } + if cap(snapshot) != len(snapshot) { + t.Errorf("snapshot capacity=%d, want exactly its %d retained points", cap(snapshot), len(snapshot)) + } + }) + } +} + +func TestHistoryReadPublicAllocationBudget(t *testing.T) { + history := NewMetricsHistory(1024, 4*time.Hour) + now := time.Now() + for i := 0; i < 512; i++ { + stamp := now.Add(time.Duration(i-512) * time.Second) + history.AddGuestMetric("dense", "cpu", float64(i), stamp) + history.AddGuestMetric("dense", "memoryused", float64(i), stamp) + history.AddNodeMetric("dense", "cpu", float64(i), stamp) + history.AddDiskMetric("dense", "smart_temp", float64(i), stamp) + history.AddStorageMetric("dense", "usage", float64(i), stamp) + history.AddStorageMetric("dense", "used", float64(i), stamp) + } + // These entries have the same map/empty-series shape but no sample arrays. + history.AddGuestMetric("empty", "unknown", 0, now) + history.AddStorageMetric("empty", "unknown", 0, now) + for name, read := range map[string]func() []MetricPoint{ + "guest": func() []MetricPoint { return history.GetGuestMetrics("dense", "cpu", time.Hour) }, + "node": func() []MetricPoint { return history.GetNodeMetrics("dense", "cpu", time.Hour) }, + "disk": func() []MetricPoint { return history.GetDiskMetrics("dense", "smart_temp", time.Hour) }, + } { + t.Run(name, func(t *testing.T) { + var snapshot []MetricPoint + allocs := testing.AllocsPerRun(25, func() { snapshot = read() }) + t.Logf("points=%d allocations=%.0f", len(snapshot), allocs) + if len(snapshot) != 512 || allocs > 1 { + t.Fatalf("dense read: points=%d allocations=%.0f, want 512/at most 1", len(snapshot), allocs) + } + }) + } + for name, read := range map[string]func(string, time.Duration) map[string][]MetricPoint{ + "guest-all": history.GetAllGuestMetrics, "storage-all": history.GetAllStorageMetrics, + } { + t.Run(name, func(t *testing.T) { + var snapshot map[string][]MetricPoint + emptyAllocs := testing.AllocsPerRun(25, func() { snapshot = read("empty", time.Hour) }) + denseAllocs := testing.AllocsPerRun(25, func() { snapshot = read("dense", time.Hour) }) + count := 0 + for _, points := range snapshot { + count += len(points) + } + t.Logf("points=%d empty allocations=%.0f dense allocations=%.0f", count, emptyAllocs, denseAllocs) + if count != 1024 || denseAllocs-emptyAllocs > 2 { + t.Fatalf("two populated series allocated %.0f extra times, want at most 2; points=%d", denseAllocs-emptyAllocs, count) + } + }) + } +} + +func TestHistoryReadSnapshotsEverySeries(t *testing.T) { + for scope, metrics := range map[string][]string{ + "guest": {"cpu", "memory", "memoryused", "disk", "gpu", "gpu_memory", "gpu_temperature", "diskread", "diskwrite", "netin", "netout", "temperature"}, + "node": {"cpu", "memory", "memoryused", "disk", "netin", "netout", "temperature"}, + "disk": {"smart_temp", "disk", "diskread", "diskwrite"}, "storage": {"usage", "used", "total", "avail"}, + } { + for _, metric := range metrics { + t.Run(scope+"/"+metric, func(t *testing.T) { + history := NewMetricsHistory(32, 4*time.Hour) + add, read := history.AddGuestMetric, history.GetGuestMetrics + switch scope { + case "node": + add, read = history.AddNodeMetric, history.GetNodeMetrics + case "disk": + add, read = history.AddDiskMetric, history.GetDiskMetrics + case "storage": + add = history.AddStorageMetric + read = func(id, metric string, duration time.Duration) []MetricPoint { + return history.GetAllStorageMetrics(id, duration)[metric] + } + } + now := time.Now() + add("resource", metric, 10, now.Add(-2*time.Hour)) + add("resource", metric, 20, now.Add(-10*time.Minute)) + add("resource", metric, 30, now.Add(-time.Minute)) + want := []MetricPoint{{Timestamp: now.Add(-10 * time.Minute), Value: 20}, {Timestamp: now.Add(-time.Minute), Value: 30}} + got := read("resource", metric, time.Hour) + if !reflect.DeepEqual(got, want) { + t.Fatalf("snapshot = %+v, want %+v", got, want) + } + got[0] = MetricPoint{} + got = append(got, MetricPoint{Value: -999}) + if !reflect.DeepEqual(read("resource", metric, time.Hour), want) { + t.Fatal("caller mutation changed History") + } + if scope == "guest" && !reflect.DeepEqual(history.GetAllGuestMetrics("resource", time.Hour)[metric], want) { + t.Fatal("all-series guest read diverged") + } + }) + } + } +} + +func TestHistoryReadChartAndWindowSnapshots(t *testing.T) { + previous := mock.IsMockEnabled() + mustSetMockEnabled(t, false) + defer mustSetMockEnabled(t, previous) + now := time.Now() + history := NewMetricsHistory(32, 4*time.Hour) + monitor := &Monitor{metricsHistory: history} + const id = "instance:node:105" + for i, offset := range []time.Duration{-2 * time.Hour, -10 * time.Minute, -time.Minute} { + monitor.recordGuestMetric("vm", id, float64(10+i), 60, float64(4096+i), -1, -1, -1, -1, -1, now.Add(offset)) + } + want := []MetricPoint{{Timestamp: now.Add(-10 * time.Minute), Value: 11}, {Timestamp: now.Add(-time.Minute), Value: 12}} + chart := monitor.GetGuestMetricsForChart(id, "vm", id, time.Hour) + if !reflect.DeepEqual(chart["cpu"], want) { + t.Fatalf("chart = %+v, want %+v", chart["cpu"], want) + } + emptyJSON, err := json.Marshal(chart["disk"]) + if err != nil || string(emptyJSON) != "[]" { + t.Fatalf("empty chart JSON = %s, err=%v", emptyJSON, err) + } + chart["cpu"][0] = MetricPoint{} + batch := monitor.GetGuestMetricsForChartBatch("vm", []GuestChartRequest{{InMemoryKey: id, SQLResourceID: id}}, time.Hour, "cpu") + if !reflect.DeepEqual(batch[id]["cpu"], want) { + t.Fatalf("batch = %+v, want %+v", batch, want) + } + batch[id]["cpu"][0] = MetricPoint{} + points, err := monitor.metricWindowPoints(alerts.MetricWindowRequest{ResourceID: id, ResourceType: "vm", Metric: "cpu", Start: now.Add(-time.Hour), End: now}) + if err != nil || len(points) != 2 || points[0].Value != 11 || points[1].Value != 12 { + t.Fatalf("metric window = %+v, err=%v", points, err) + } + if !reflect.DeepEqual(monitor.GetGuestMetrics(id, time.Hour)["cpu"], want) { + t.Fatal("chart/window mutation reached poll History") + } +} + +var historyReadBenchmarkSink []MetricPoint + +func BenchmarkHistoryReadFilter(b *testing.B) { + for _, tc := range []struct { + name string + size, retained int + interleaved bool + }{ + {"empty", 0, 0, false}, {"expired", 2880, 0, false}, + {"one", 2880, 1, false}, {"hour", 2880, 60, false}, + {"day", 2880, 1440, false}, {"all", 2880, 2880, false}, + {"interleaved", 2880, 1440, true}, + } { + data, cutoff := historyReadFixture(tc.size, tc.retained, tc.interleaved) + for _, variant := range []struct { + name string + filter func([]MetricPoint, time.Time) []MetricPoint + }{ + {"legacy", legacyHistoryReadFilter}, {"current", filterMetricsByTime}, + } { + b.Run(fmt.Sprintf("%s/%s", tc.name, variant.name), func(b *testing.B) { + if !reflect.DeepEqual(variant.filter(data, cutoff), legacyHistoryReadFilter(data, cutoff)) { + b.Fatal("benchmark changed samples") + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + historyReadBenchmarkSink = variant.filter(data, cutoff) + } + }) + } + } +}