Keep used-memory history within its retention window

Periodic guest and node cleanup omitted raw memory-used samples from both expiry and live-entry checks. Trim them with the existing retention and backing-array policy without deleting a still-live byte series. Cover every supported History series, byte-array reclamation, concurrent cleanup and poll-to-chart/metric-window resumption. No SQL retention, source-selection or wire changes.

Change-source: pulse-maintainer

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

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{

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)

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+

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

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