hostagent,guestagent: adaptive balloon enhancements (E1-E10)

Add 10 targeted improvements to the balloon controller, auto-pause
manager, guest metrics, and template provisioning:

E1: PSI fallback guard — detect kernels without PSI and fall back to
MemAvailable-based decisions. Add mode:boot script to enable PSI on
Alpine kernels with CONFIG_PSI_DEFAULT_DISABLED=y.

E2: Floor staleness timeout — learned floors expire after 24h (config:
floorStaleness) to adapt to workload drift. Store format extended to
include timestamp.

E3: Graduated poll failure recovery — replace immediate max-expansion
with graduated steps: 1-2 failures no-op, 3-5 half headroom, 6+ full.

E4: Balloon and auto-pause coordination — halve shrink step when SSH/
socket connections are active. Skip balloon polling while VM is paused.
Reset poll failure counter on resume.

E5: Parallel Docker stats — collect per-container stats concurrently
with 3s overall and 1s per-container timeouts.

E6: Dynamic cgroup memory.high — recalculate memory.high every 5min
based on MemAvailable+AnonPages with 512MiB floor.

E7: Host pressure hysteresis — require 2 consecutive samples before
level transitions (except immediate Critical) to prevent oscillation.

E8: Adaptive AnonRss margin — replace fixed 15% with container-count
tiers: 5% (0 containers), 15% (1-5), 20% (6+).

E9: PSI trend detection — add avg60 proto fields. Pre-emptive half-step
grow when avg10 > 1.5*avg60 (opt-in via enableTrendDetection).

E10: Demand-driven compaction — buddyinfo-gated compact, slab drop
threshold, guarded MGLRU/KSM re-enable.

Signed-off-by: Jason W. Ehrlich <jwehrlich@outlook.com>

Author: jwehrlich

pkg/driver/vz/vz_driver_darwin.go

@@ -312,12 +312,30 @@ func (l *LimaVzDriver) FillConfig(ctx context.Context, cfg *limatype.LimaYAML, _
 		if vzOpts.MemoryBalloon.MaxSwapInPerSec == nil {
 			vzOpts.MemoryBalloon.MaxSwapInPerSec = ptr.Of("64MiB")
 		}
+		if vzOpts.MemoryBalloon.MaxSwapOutPerSec == nil {
+			vzOpts.MemoryBalloon.MaxSwapOutPerSec = ptr.Of("32MiB")
+		}
+		if vzOpts.MemoryBalloon.MaxPageFaultRate == nil {
+			vzOpts.MemoryBalloon.MaxPageFaultRate = ptr.Of(uint64(5000))
+		}
+		if vzOpts.MemoryBalloon.ShrinkReserveBytes == nil {
+			vzOpts.MemoryBalloon.ShrinkReserveBytes = ptr.Of("128MiB")
+		}
+		if vzOpts.MemoryBalloon.SettleWindow == nil {
+			vzOpts.MemoryBalloon.SettleWindow = ptr.Of("30s")
+		}
 		if vzOpts.MemoryBalloon.MaxContainerCPU == nil {
 			vzOpts.MemoryBalloon.MaxContainerCPU = ptr.Of(10.0)
 		}
 		if vzOpts.MemoryBalloon.MaxContainerIO == nil {
 			vzOpts.MemoryBalloon.MaxContainerIO = ptr.Of("10MiB")
 		}
+		if vzOpts.MemoryBalloon.FloorStaleness == nil {
+			vzOpts.MemoryBalloon.FloorStaleness = ptr.Of("24h")
+		}
+		if vzOpts.MemoryBalloon.EnableTrendDetection == nil {
+			vzOpts.MemoryBalloon.EnableTrendDetection = ptr.Of(false)
+		}
 	}
 
 	// Auto-pause defaults — only set when enabled.

pkg/guestagent/api/guestservice.pb.desc

@@ -1,5 +1,5 @@
 
-�
+�
 guestservice.protogoogle/protobuf/empty.protogoogle/protobuf/timestamp.proto"0
 Info(
 local_ports
 (2.IPPortR
@@ -29,7 +29,7 @@ guest_addr 
(	R	guestAddr&
 TimeSyncResponse
 adjusted
 
(Radjusted
 drift_ms 
(RdriftMs
-error 
(	Rerror"�
+error 
(	Rerror"�
 MemoryMetrics&
 mem_total_bytes
 
(RmemTotalBytes.
 mem_available_bytes 
(RmemAvailableBytes(
@@ -46,7 +46,9 @@ guest_addr 
(	R	guestAddr&
 container_count 
(RcontainerCount2
 container_cpu_percent 
(
RcontainerCpuPercent:
 container_io_bytes_per_sec 
(
RcontainerIoBytesPerSec!
-oom_detected 
(RoomDetected2�
+oom_detected 
(RoomDetected+
+psi_memory_some_60 
(
RpsiMemorySome60+
+psi_memory_full_60 
(
RpsiMemoryFull602�
 GuestService(
 GetInfo.google.protobuf.Empty.Info-
 	GetEvents.google.protobuf.Empty.Event0
1

pkg/guestagent/api/guestservice.pb.go

@@ -84,4 +84,8 @@ message MemoryMetrics {
 
   // OOM detection (edge-triggered: set once, cleared after reporting).
   bool oom_detected = 15;
+
+  // /proc/pressure/memory PSI values (percentage 0.0-100.0 over 60s window).
+  double psi_memory_some_60 = 16;
+  double psi_memory_full_60 = 17;
 }

pkg/guestagent/api/guestservice.proto

@@ -109,9 +109,9 @@ func (s *GuestServer) SyncTime(_ context.Context, req *api.TimeSyncRequest) (*ap
 	return resp, nil
 }
 
-func (s *GuestServer) GetMemoryMetrics(_ context.Context, _ *emptypb.Empty) (*api.MemoryMetrics, error) {
+func (s *GuestServer) GetMemoryMetrics(ctx context.Context, _ *emptypb.Empty) (*api.MemoryMetrics, error) {
 	if s.Collector != nil {
-		return s.Collector.Collect()
+		return s.Collector.Collect(ctx)
 	}
 	return metrics.CollectMemoryMetrics()
 }

pkg/guestagent/api/server/server.go

@@ -54,7 +54,7 @@ func NewCollector(dockerSocket *string) *Collector {
 					return (&net.Dialer{}).DialContext(ctx, "unix", c.dockerSocket)
 				},
 			},
-			Timeout: 5 * time.Second,
+			// No client-level timeout — per-request context deadlines control timeouts.
 		}
 	}
 	return c
@@ -119,7 +119,7 @@ func safeDelta(curr, prev uint64) uint64 {
 
 // Collect gathers all memory metrics and returns a MemoryMetrics protobuf.
 // This is the main entry point called by the guest agent gRPC server.
-func (c *Collector) Collect() (*api.MemoryMetrics, error) {
+func (c *Collector) Collect(ctx context.Context) (*api.MemoryMetrics, error) {
 	// 1. /proc/meminfo + /proc/pressure/memory (no lock needed — pure reads).
 	meminfo, err := os.ReadFile("/proc/meminfo")
 	if err != nil {
@@ -131,12 +131,14 @@ func (c *Collector) Collect() (*api.MemoryMetrics, error) {
 	}
 
 	pressure, _ := os.ReadFile("/proc/pressure/memory")
-	some10, full10, parseErr := parseProcPressureMemory(pressure)
+	psi, parseErr := parseProcPressureMemory(pressure)
 	if parseErr != nil {
 		return nil, parseErr
 	}
-	m.PsiMemorySome_10 = some10
-	m.PsiMemoryFull_10 = full10
+	m.PsiMemorySome_10 = psi.Some10
+	m.PsiMemoryFull_10 = psi.Full10
+	m.PsiMemorySome_60 = psi.Some60
+	m.PsiMemoryFull_60 = psi.Full60
 
 	// 2. /proc/vmstat for swap rates, page faults, OOM.
 	vmstatData, vmstatErr := os.ReadFile("/proc/vmstat")
@@ -145,7 +147,7 @@ func (c *Collector) Collect() (*api.MemoryMetrics, error) {
 	var dockerCount int
 	var dockerCPU, dockerIO float64
 	if c.httpClient != nil {
-		dockerCount, dockerCPU, dockerIO = c.collectDockerStats()
+		dockerCount, dockerCPU, dockerIO = c.collectDockerStats(ctx)
 	}
 
 	// Hold lock only for internal state updates and reads.
@@ -171,10 +173,15 @@ func (c *Collector) Collect() (*api.MemoryMetrics, error) {
 }
 
 // collectDockerStats queries the Docker socket for container count,
-// aggregate CPU%, and aggregate IO bytes/sec. Returns zeros on error.
-func (c *Collector) collectDockerStats() (count int, cpuPercent, ioBytesPerSec float64) {
+// aggregate CPU%, and aggregate IO bytes/sec. Containers are polled
+// in parallel with a 3-second overall timeout. Returns zeros on error.
+func (c *Collector) collectDockerStats(ctx context.Context) (count int, cpuPercent, ioBytesPerSec float64) {
+	// Overall timeout: 3 seconds fits within the 10-second balloon poll interval.
+	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+
 	// List running containers.
-	listReq, err := http.NewRequestWithContext(context.Background(), http.MethodGet,
+	listReq, err := http.NewRequestWithContext(ctx, http.MethodGet,
 		"http://localhost/containers/json?filters=%7B%22status%22%3A%5B%22running%22%5D%7D", http.NoBody)
 	if err != nil {
 		logrus.Debugf("Docker stats: failed to create list request: %v", err)
@@ -201,30 +208,53 @@ func (c *Collector) collectDockerStats() (count int, cpuPercent, ioBytesPerSec f
 		return 0, 0, 0
 	}
 
-	// Aggregate stats from each container (best-effort, skip failures).
+	// Poll all containers in parallel with per-container 1-second timeout.
+	type result struct {
+		cpuPct  float64
+		ioBytes uint64
+	}
+	results := make(chan result, len(ids))
+	for _, id := range ids {
+		go func(cid string) {
+			cctx, ccancel := context.WithTimeout(ctx, 1*time.Second)
+			defer ccancel()
+			statsReq, reqErr := http.NewRequestWithContext(cctx, http.MethodGet,
+				"http://localhost/containers/"+cid+"/stats?stream=false&one-shot=true", http.NoBody)
+			if reqErr != nil {
+				results <- result{}
+				return
+			}
+			statsResp, doErr := c.httpClient.Do(statsReq)
+			if doErr != nil {
+				results <- result{}
+				return
+			}
+			statsBody, readErr := io.ReadAll(statsResp.Body)
+			statsResp.Body.Close()
+			if readErr != nil {
+				results <- result{}
+				return
+			}
+			cpuPct, ioBytes, parseErr := parseDockerStats(statsBody)
+			if parseErr != nil {
+				results <- result{}
+				return
+			}
+			results <- result{cpuPct, ioBytes}
+		}(id)
+	}
+
+	// Collect results, using partial data if overall timeout hits.
 	var totalCPU float64
 	var totalIO uint64
-	for _, id := range ids {
-		statsReq, reqErr := http.NewRequestWithContext(context.Background(), http.MethodGet,
-			"http://localhost/containers/"+id+"/stats?stream=false&one-shot=true", http.NoBody)
-		if reqErr != nil {
-			continue
-		}
-		statsResp, doErr := c.httpClient.Do(statsReq)
-		if doErr != nil {
-			continue
-		}
-		statsBody, readErr := io.ReadAll(statsResp.Body)
-		statsResp.Body.Close()
-		if readErr != nil {
-			continue
-		}
-		cpuPct, ioBytes, parseErr := parseDockerStats(statsBody)
-		if parseErr != nil {
-			continue
+	for range ids {
+		select {
+		case r := <-results:
+			totalCPU += r.cpuPct
+			totalIO += r.ioBytes
+		case <-ctx.Done():
+			return count, totalCPU, float64(totalIO)
 		}
-		totalCPU += cpuPct
-		totalIO += ioBytes
 	}
 
 	return count, totalCPU, float64(totalIO)

pkg/guestagent/metrics/collector.go

@@ -28,12 +28,14 @@ func CollectMemoryMetrics() (*api.MemoryMetrics, error) {
 	}
 
 	pressure, _ := os.ReadFile("/proc/pressure/memory")
-	some10, full10, err := parseProcPressureMemory(pressure)
+	psi, err := parseProcPressureMemory(pressure)
 	if err != nil {
 		return nil, err
 	}
-	m.PsiMemorySome_10 = some10
-	m.PsiMemoryFull_10 = full10
+	m.PsiMemorySome_10 = psi.Some10
+	m.PsiMemoryFull_10 = psi.Full10
+	m.PsiMemorySome_60 = psi.Some60
+	m.PsiMemoryFull_60 = psi.Full60
 
 	return m, nil
 }
@@ -72,9 +74,16 @@ func parseProcMeminfo(data []byte) (*api.MemoryMetrics, error) {
 	return m, scanner.Err()
 }
 
-func parseProcPressureMemory(data []byte) (some10, full10 float64, err error) {
+// PressureStats holds parsed PSI values from /proc/pressure/memory.
+type PressureStats struct {
+	Some10, Full10 float64
+	Some60, Full60 float64
+}
+
+func parseProcPressureMemory(data []byte) (PressureStats, error) {
+	var ps PressureStats
 	if len(data) == 0 {
-		return 0, 0, nil
+		return ps, nil
 	}
 	scanner := bufio.NewScanner(bytes.NewReader(data))
 	for scanner.Scan() {
@@ -92,12 +101,24 @@ func parseProcPressureMemory(data []byte) (some10, full10 float64, err error) {
 				}
 				switch kind {
 				case "some":
-					some10 = val
+					ps.Some10 = val
+				case "full":
+					ps.Full10 = val
+				}
+			}
+			if after, ok := strings.CutPrefix(field, "avg60="); ok {
+				val, parseErr := strconv.ParseFloat(after, 64)
+				if parseErr != nil {
+					continue
+				}
+				switch kind {
+				case "some":
+					ps.Some60 = val
 				case "full":
-					full10 = val
+					ps.Full60 = val
 				}
 			}
 		}
 	}
-	return some10, full10, scanner.Err()
+	return ps, scanner.Err()
 }