diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index cf047d13ba..fbb0f07eb6 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,20 @@ ## Purpose +### Used-memory history retention + +Periodic in-memory History cleanup applies the existing retention window and +backing-array release policy to raw `memoryused` bytes as well as `memory` +percentages, for guests and nodes. A live byte series alone keeps its resource +entry; an expired byte series cannot remain pinned by ongoing CPU or other +observations. Missing current memory does not refresh its last-known samples. +The existing poll recorder, chart readers and metric-window provider retain +in-window byte values and their original timestamps through cleanup, then +accept fresh memory on resumption. SQL retention, sampling limits, memory +source selection and wire formats are unchanged. The History cleanup tests +cover every supported series, byte-array reclamation and connected readers; +these controls do not attribute a reporter's native process-memory growth. + ### Podman zero-percentage fallback — issue #2447 One-shot compatibility stats with `cpu:0` do not suppress advancing cumulative diff --git a/internal/monitoring/metric_window_provider_test.go b/internal/monitoring/metric_window_provider_test.go index e29c52935a..ef74ee88d4 100644 --- a/internal/monitoring/metric_window_provider_test.go +++ b/internal/monitoring/metric_window_provider_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) @@ -53,6 +54,44 @@ func TestMetricWindowPointsUsesInMemoryMetricAlias(t *testing.T) { } } +func TestUsedMemoryHistoryCleanupThroughMonitorAndReaders(t *testing.T) { + previous := mock.IsMockEnabled() + mustSetMockEnabled(t, false) + defer mustSetMockEnabled(t, previous) + for _, resourceType := range []string{"vm", "container"} { + t.Run(resourceType, func(t *testing.T) { + now := time.Now() + history := NewMetricsHistory(32, 4*time.Hour) + monitor := &Monitor{metricsHistory: history} + const id = "instance:node:105" + monitor.recordGuestMetric(resourceType, id, 20, 60, 4096, -1, -1, -1, -1, -1, now.Add(-2*time.Hour)) + monitor.recordGuestMetric(resourceType, id, 25, -1, -1, -1, -1, -1, -1, -1, now.Add(-time.Minute)) + monitor.cleanupMetricsHistory() + if got := monitor.GetGuestMetrics(id, 4*time.Hour)["memoryused"]; len(got) != 1 || got[0].Value != 4096 { + t.Fatalf("in-window last-known bytes were lost or refreshed: %+v", got) + } + history.retentionTime = time.Hour + monitor.cleanupMetricsHistory() + if got := monitor.GetGuestMetrics(id, 4*time.Hour)["memoryused"]; len(got) != 0 { + t.Fatalf("ongoing CPU retained expired byte samples: %+v", got) + } + monitor.recordGuestMetric(resourceType, id, 30, 70, 8192, -1, -1, -1, -1, -1, now) + monitor.cleanupMetricsHistory() + chart := monitor.GetGuestMetricsForChart(id, resourceType, id, 4*time.Hour)["memoryused"] + if len(chart) != 1 || chart[0].Value != 8192 || !chart[0].Timestamp.Equal(now) { + t.Fatalf("chart did not retain the fresh byte observation: %+v", chart) + } + points, err := monitor.metricWindowPoints(alerts.MetricWindowRequest{ + ResourceID: id, ResourceType: resourceType, Metric: "memoryused", + Start: now.Add(-4 * time.Hour), End: now, + }) + if err != nil || len(points) != 1 || points[0].Value != 8192 || !points[0].Timestamp.Equal(now) { + t.Fatalf("metric window diverged from fresh chart bytes: points=%+v err=%v", points, err) + } + }) + } +} + func TestMetricWindowMergePrefersFreshInMemoryDuplicate(t *testing.T) { now := time.Now().UTC() stored := []MetricPoint{ diff --git a/internal/monitoring/metrics_history.go b/internal/monitoring/metrics_history.go index ba4ab659b8..928356ced2 100644 --- a/internal/monitoring/metrics_history.go +++ b/internal/monitoring/metrics_history.go @@ -771,6 +771,7 @@ func (mh *MetricsHistory) Cleanup() { for key, metrics := range mh.guestMetrics { metrics.CPU = mh.cleanupMetrics(metrics.CPU, cutoffTime) metrics.Memory = mh.cleanupMetrics(metrics.Memory, cutoffTime) + metrics.MemoryUsed = mh.cleanupMetrics(metrics.MemoryUsed, cutoffTime) metrics.Disk = mh.cleanupMetrics(metrics.Disk, cutoffTime) metrics.GPU = mh.cleanupMetrics(metrics.GPU, cutoffTime) metrics.GPUMemory = mh.cleanupMetrics(metrics.GPUMemory, cutoffTime) @@ -782,7 +783,7 @@ func (mh *MetricsHistory) Cleanup() { metrics.Temperature = mh.cleanupMetrics(metrics.Temperature, cutoffTime) // If all slices are empty, remove the map entry entirely to free memory - if len(metrics.CPU) == 0 && len(metrics.Memory) == 0 && len(metrics.Disk) == 0 && + if len(metrics.CPU) == 0 && len(metrics.Memory) == 0 && len(metrics.MemoryUsed) == 0 && len(metrics.Disk) == 0 && len(metrics.GPU) == 0 && len(metrics.GPUMemory) == 0 && len(metrics.GPUTemperature) == 0 && len(metrics.DiskRead) == 0 && len(metrics.DiskWrite) == 0 && len(metrics.NetworkIn) == 0 && len(metrics.NetworkOut) == 0 && @@ -796,12 +797,13 @@ func (mh *MetricsHistory) Cleanup() { for key, metrics := range mh.nodeMetrics { metrics.CPU = mh.cleanupMetrics(metrics.CPU, cutoffTime) metrics.Memory = mh.cleanupMetrics(metrics.Memory, cutoffTime) + metrics.MemoryUsed = mh.cleanupMetrics(metrics.MemoryUsed, cutoffTime) metrics.Disk = mh.cleanupMetrics(metrics.Disk, cutoffTime) metrics.NetworkIn = mh.cleanupMetrics(metrics.NetworkIn, cutoffTime) metrics.NetworkOut = mh.cleanupMetrics(metrics.NetworkOut, cutoffTime) metrics.Temperature = mh.cleanupMetrics(metrics.Temperature, cutoffTime) - if len(metrics.CPU) == 0 && len(metrics.Memory) == 0 && len(metrics.Disk) == 0 && + if len(metrics.CPU) == 0 && len(metrics.Memory) == 0 && len(metrics.MemoryUsed) == 0 && len(metrics.Disk) == 0 && len(metrics.NetworkIn) == 0 && len(metrics.NetworkOut) == 0 && len(metrics.Temperature) == 0 { delete(mh.nodeMetrics, key) diff --git a/internal/monitoring/metrics_history_concurrency_test.go b/internal/monitoring/metrics_history_concurrency_test.go index 2417496a23..f9dc001237 100644 --- a/internal/monitoring/metrics_history_concurrency_test.go +++ b/internal/monitoring/metrics_history_concurrency_test.go @@ -13,13 +13,14 @@ func TestMetricsHistoryConcurrentAccess(t *testing.T) { const iterations = 1000 var wg sync.WaitGroup - wg.Add(4) + wg.Add(5) go func() { defer wg.Done() for i := 0; i < iterations; i++ { mh.AddGuestMetric("guest-1", "cpu", rand.Float64()*100, time.Now()) mh.AddGuestMetric("guest-1", "memory", rand.Float64()*100, time.Now()) + mh.AddGuestMetric("guest-1", "memoryused", 4096, time.Now()) mh.AddGuestMetric("guest-2", "disk", rand.Float64()*100, time.Now()) time.Sleep(time.Microsecond) } @@ -30,6 +31,7 @@ func TestMetricsHistoryConcurrentAccess(t *testing.T) { for i := 0; i < iterations; i++ { mh.AddNodeMetric("node-1", "cpu", rand.Float64()*100, time.Now()) mh.AddNodeMetric("node-1", "memory", rand.Float64()*100, time.Now()) + mh.AddNodeMetric("node-1", "memoryused", 8192, time.Now()) mh.AddStorageMetric("storage-1", "usage", rand.Float64()*100, time.Now()) time.Sleep(time.Microsecond) } @@ -40,6 +42,7 @@ func TestMetricsHistoryConcurrentAccess(t *testing.T) { for i := 0; i < iterations; i++ { mh.GetGuestMetrics("guest-1", "cpu", time.Minute) mh.GetGuestMetrics("guest-1", "memory", time.Minute) + mh.GetGuestMetrics("guest-1", "memoryused", time.Minute) mh.GetGuestMetrics("guest-2", "disk", time.Minute) time.Sleep(time.Microsecond) } @@ -50,13 +53,26 @@ func TestMetricsHistoryConcurrentAccess(t *testing.T) { for i := 0; i < iterations; i++ { mh.GetNodeMetrics("node-1", "cpu", time.Minute) mh.GetNodeMetrics("node-1", "memory", time.Minute) + mh.GetNodeMetrics("node-1", "memoryused", time.Minute) // Storage metrics are exposed through Monitor, but simulate reads via guest metrics map mh.GetGuestMetrics("storage-1", "usage", time.Minute) time.Sleep(time.Microsecond) } }() + go func() { + defer wg.Done() + for i := 0; i < iterations; i++ { + mh.Cleanup() + time.Sleep(time.Microsecond) + } + }() + wg.Wait() + if len(mh.GetGuestMetrics("guest-1", "memoryused", time.Minute)) == 0 || + len(mh.GetNodeMetrics("node-1", "memoryused", time.Minute)) == 0 { + t.Fatal("concurrent cleanup lost live used-memory samples") + } } // rand is automatically seeded in Go 1.20+ diff --git a/internal/monitoring/metrics_history_memory_regression_test.go b/internal/monitoring/metrics_history_memory_regression_test.go index 815120e0f6..e78f890b5d 100644 --- a/internal/monitoring/metrics_history_memory_regression_test.go +++ b/internal/monitoring/metrics_history_memory_regression_test.go @@ -126,3 +126,36 @@ func TestCleanupReleasesSeedSizedCapacity(t *testing.T) { t.Fatalf("cleanup still aliases the seed-sized backing array (len=%d cap-was=%d)", len(series), before) } } + +func TestCleanupUsedMemoryReleasesSeedSizedCapacity(t *testing.T) { + for _, scope := range []string{"guest", "node"} { + t.Run(scope, func(t *testing.T) { + mh := NewMetricsHistory(3500, 4*time.Hour) + add := mh.AddGuestMetric + entries := mh.guestMetrics + if scope == "node" { + add, entries = mh.AddNodeMetric, mh.nodeMetrics + } + now := time.Now() + for i := 0; i < 2000; i++ { + add("resource", "memoryused", float64(i), now.Add(-2*time.Hour).Add(time.Duration(i)*time.Second)) + } + add("resource", "memoryused", 4096, now.Add(-time.Minute)) + add("resource", "cpu", 25, now) + original := entries["resource"].MemoryUsed + mh.retentionTime = time.Hour + mh.Cleanup() + series := entries["resource"].MemoryUsed + if len(series) != 1 || series[0].Value != 4096 || !series[0].Timestamp.Equal(now.Add(-time.Minute)) { + t.Fatalf("used-memory retention kept %d points, want the unchanged live point", len(series)) + } + if cap(series) > 4*len(series)+64 { + t.Fatalf("used-memory cleanup pinned an oversized array: len=%d cap=%d", len(series), cap(series)) + } + if &series[0] == &original[0] || &series[0] == &original[len(original)-1] { + t.Fatal("used-memory cleanup still aliases the seed-sized array") + } + t.Logf("used-memory samples %d -> %d; backing capacity %d -> %d", len(original), len(series), cap(original), cap(series)) + }) + } +} diff --git a/internal/monitoring/metrics_history_test.go b/internal/monitoring/metrics_history_test.go index 153e949a9d..bfa5d59ad8 100644 --- a/internal/monitoring/metrics_history_test.go +++ b/internal/monitoring/metrics_history_test.go @@ -60,6 +60,83 @@ func TestNewMetricsHistory(t *testing.T) { } } +// Use the public writers and readers for every supported series. In particular, +// MemoryUsed must participate in both retention and resource-entry ownership; +// testing cleanupMetrics alone cannot detect an omitted field in Cleanup. +func TestMetricsHistoryCleanupEverySeries(t *testing.T) { + for _, scope := range []string{"guest", "node", "storage", "disk"} { + metricTypes := 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"}, + "storage": {"usage", "used", "total", "avail"}, + "disk": {"smart_temp", "disk", "diskread", "diskwrite"}, + }[scope] + for _, metric := range metricTypes { + for _, scenario := range []string{"live-only", "expired-with-live-neighbour", "mixed"} { + t.Run(scope+"/"+metric+"/"+scenario, func(t *testing.T) { + mh := NewMetricsHistory(32, 4*time.Hour) + now := time.Now() + const id = "resource" + add := mh.AddGuestMetric + read := mh.GetGuestMetrics + exists := func() bool { return mh.guestMetrics[id] != nil } + switch scope { + case "node": + add, read = mh.AddNodeMetric, mh.GetNodeMetrics + exists = func() bool { return mh.nodeMetrics[id] != nil } + case "storage": + add = mh.AddStorageMetric + read = func(id, metric string, duration time.Duration) []MetricPoint { + return mh.GetAllStorageMetrics(id, duration)[metric] + } + exists = func() bool { return mh.storageMetrics[id] != nil } + case "disk": + add, read = mh.AddDiskMetric, mh.GetDiskMetrics + exists = func() bool { return mh.diskMetrics[id] != nil } + } + if scenario != "live-only" { + add(id, metric, 11, now.Add(-2*time.Hour)) + } + if scenario != "expired-with-live-neighbour" { + add(id, metric, 22, now.Add(-time.Minute)) + } else { + // Another fresh series must not pin this expired one. + keeper := metricTypes[0] + if keeper == metric { + keeper = metricTypes[1] + } + add(id, keeper, 33, now) + } + // Model elapsed retention without waiting or weakening the + // writer's immediate expiry check for backfilled samples. + mh.retentionTime = time.Hour + mh.Cleanup() + if !exists() { + t.Fatal("cleanup removed a resource with a live series") + } + points := read(id, metric, 4*time.Hour) + if scenario == "expired-with-live-neighbour" { + if len(points) != 0 { + t.Fatalf("expired series remains: %+v", points) + } + } else if len(points) != 1 || points[0].Value != 22 || !points[0].Timestamp.Equal(now.Add(-time.Minute)) { + t.Fatalf("cleanup changed live values/timestamps or retained old points: %+v", points) + } + mh.Cleanup() + if len(read(id, metric, 4*time.Hour)) != len(points) { + t.Fatal("repeated cleanup changed the retained series") + } + mh.retentionTime = 0 + mh.Cleanup() + if exists() { + t.Fatal("fully expired resource remains in History") + } + }) + } + } + } +} + func TestAppendMetric(t *testing.T) { now := time.Now()