Add service autoscaling

agent/internal/agent/reporting.go

@@ -65,7 +65,7 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport
 					log.Printf("[metrics] failed to collect container stats: %v", err)
 					return
 				}
-				if err := a.MetricsSender.SendContainerStats(containerStats, collectedAt); err != nil {
+				if err := a.MetricsSender.SendContainerStats(containerStats, time.Now()); err != nil {
 					log.Printf("[metrics] failed to send container stats: %v", err)
 				}
 			}()

agent/internal/container/stats.go

@@ -1,13 +1,17 @@
 package container
 
 import (
+	"bufio"
 	"bytes"
-	"encoding/json"
+	"context"
 	"fmt"
+	"log"
 	"math"
 	"os/exec"
 	"strconv"
 	"strings"
+	"sync"
+	"time"
 	"unicode"
 )
 
@@ -16,20 +20,52 @@ type ResourceStats struct {
 	ServiceID            string
 	DeploymentID         string
 	CPUUsagePercent      float64
+	CPUUsageValid        bool
 	MemoryUsagePercent   float64
+	MemoryUsageValid     bool
 	MemoryUsedBytes      float64
+	MemoryUsedValid      bool
 	NetworkReceiveBytes  float64
 	NetworkTransmitBytes float64
 }
 
+type podmanStatsSample struct {
+	containerID        string
+	cpuNano            uint64
+	systemNano         uint64
+	cpuCountersValid   bool
+	memoryUsage        string
+	memoryUsagePercent string
+	networkIO          string
+}
+
+const podmanStatsFormat = "{{.ContainerID}}\t{{.CPUNano}}\t{{.SystemNano}}\t{{.MemUsage}}\t{{.MemPerc}}\t{{.NetIO}}"
+
+var previousResourceSamples = struct {
+	sync.Mutex
+	byContainer map[string]podmanStatsSample
+}{byContainer: make(map[string]podmanStatsSample)}
+
+// resourceStatsCollectionMu ensures overlapping periodic and requested reports
+// compare CPU counters from snapshots collected in order.
+var resourceStatsCollectionMu sync.Mutex
+
 func CollectResourceStats() ([]ResourceStats, error) {
+	resourceStatsCollectionMu.Lock()
+	defer resourceStatsCollectionMu.Unlock()
+
 	containers, err := List()
 	if err != nil {
 		return nil, err
 	}
 
 	running := make([]Container, 0, len(containers))
-	args := []string{"stats", "--no-stream", "--format", "json"}
+	args := []string{
+		"stats",
+		"--no-stream",
+		"--no-trunc",
+		"--format", podmanStatsFormat,
+	}
 	for _, c := range containers {
 		if c.State != "running" || c.ServiceID == "" || c.DeploymentID == "" {
 			continue
@@ -41,80 +77,42 @@ func CollectResourceStats() ([]ResourceStats, error) {
 		return nil, nil
 	}
 
-	cmd := exec.Command("podman", args...)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, "podman", args...)
 	var stderr bytes.Buffer
 	cmd.Stderr = &stderr
 	output, err := cmd.Output()
 	if err != nil {
 		return nil, fmt.Errorf("failed to collect container stats: %s: %w", stderr.String(), err)
 	}
-
-	return parsePodmanStatsOutput(output, running)
-}
-
-func parsePodmanStatsOutput(output []byte, containers []Container) ([]ResourceStats, error) {
-	rows, err := parseStatsRows(output)
+	samples, err := parsePodmanStatsSamples(output)
 	if err != nil {
 		return nil, err
 	}
 
-	stats := make([]ResourceStats, 0, len(rows))
-	for _, row := range rows {
-		containerID := firstRowString(row, "ID", "Id", "id", "ContainerID", "Container")
-		container := findStatsContainerByID(containerID, containers)
-		if container == nil {
-			name := firstRowString(row, "Name", "Names", "name")
-			container = findStatsContainerByName(name, containers)
+	previousResourceSamples.Lock()
+	defer previousResourceSamples.Unlock()
+	nextSamples := make(map[string]podmanStatsSample, len(samples))
+	for _, container := range running {
+		if previous, ok := previousResourceSamples.byContainer[container.ID]; ok {
+			nextSamples[container.ID] = previous
 		}
+	}
+	stats := make([]ResourceStats, 0, len(samples))
+	for _, sample := range samples {
+		container := findStatsContainerByID(sample.containerID, running)
 		if container == nil {
 			continue
 		}
-
-		rx, tx := parseNetIO(firstRowString(row, "NetIO", "NetIOBytes", "net_io"))
-		stats = append(stats, ResourceStats{
-			ContainerID:          container.ID,
-			ServiceID:            container.ServiceID,
-			DeploymentID:         container.DeploymentID,
-			CPUUsagePercent:      parsePercent(firstRowString(row, "CPUPerc", "CPU", "cpu_percent")),
-			MemoryUsagePercent:   parsePercent(firstRowString(row, "MemPerc", "MEMPerc", "mem_percent")),
-			MemoryUsedBytes:      parseMemUsed(firstRowString(row, "MemUsage", "MemUse", "mem_usage")),
-			NetworkReceiveBytes:  rx,
-			NetworkTransmitBytes: tx,
-		})
+		previous := previousResourceSamples.byContainer[container.ID]
+		stats = append(stats, resourceStatsFromSamples(*container, previous, sample))
+		nextSamples[container.ID] = sample
 	}
-
+	previousResourceSamples.byContainer = nextSamples
 	return stats, nil
 }
 
-func parseStatsRows(output []byte) ([]map[string]interface{}, error) {
-	trimmed := strings.TrimSpace(string(output))
-	if trimmed == "" {
-		return nil, nil
-	}
-
-	if strings.HasPrefix(trimmed, "[") {
-		var rows []map[string]interface{}
-		if err := json.Unmarshal([]byte(trimmed), &rows); err != nil {
-			return nil, fmt.Errorf("failed to parse podman stats JSON array: %w", err)
-		}
-		return rows, nil
-	}
-
-	var rows []map[string]interface{}
-	for _, line := range strings.Split(trimmed, "\n") {
-		line = strings.TrimSpace(line)
-		if line == "" {
-			continue
-		}
-		var row map[string]interface{}
-		if err := json.Unmarshal([]byte(line), &row); err != nil {
-			return nil, fmt.Errorf("failed to parse podman stats JSON row: %w", err)
-		}
-		rows = append(rows, row)
-	}
-	return rows, nil
-}
-
 func findStatsContainerByID(value string, containers []Container) *Container {
 	value = strings.TrimSpace(value)
 	if value == "" {
@@ -133,73 +131,119 @@ func findStatsContainerByID(value string, containers []Container) *Container {
 	return nil
 }
 
-func findStatsContainerByName(value string, containers []Container) *Container {
-	value = strings.TrimPrefix(strings.TrimSpace(value), "/")
-	if value == "" {
-		return nil
+func parsePodmanStatsSamples(output []byte) ([]podmanStatsSample, error) {
+	samples := make([]podmanStatsSample, 0)
+	skipped := 0
+	scanner := bufio.NewScanner(bytes.NewReader(output))
+	for scanner.Scan() {
+		if strings.TrimSpace(scanner.Text()) == "" {
+			continue
+		}
+		sample, err := parsePodmanStatsSample(scanner.Text())
+		if err != nil {
+			skipped++
+			continue
+		}
+		samples = append(samples, sample)
+	}
+	if err := scanner.Err(); err != nil {
+		return nil, fmt.Errorf("failed to read container stats: %w", err)
 	}
+	if skipped > 0 {
+		log.Printf("[metrics] skipped %d malformed container stats rows", skipped)
+	}
+	return samples, nil
+}
 
-	for i := range containers {
-		containerName := strings.TrimPrefix(strings.TrimSpace(containers[i].Name), "/")
-		if value == containerName {
-			return &containers[i]
-		}
+func parsePodmanStatsSample(line string) (podmanStatsSample, error) {
+	parts := strings.Split(line, "\t")
+	if len(parts) != 6 {
+		return podmanStatsSample{}, fmt.Errorf("failed to parse podman stats row: expected 6 fields, got %d", len(parts))
 	}
-	return nil
+	cpuNano, cpuErr := strconv.ParseUint(strings.TrimSpace(parts[1]), 10, 64)
+	systemNano, systemErr := strconv.ParseUint(strings.TrimSpace(parts[2]), 10, 64)
+	return podmanStatsSample{
+		containerID:        strings.TrimSpace(parts[0]),
+		cpuNano:            cpuNano,
+		systemNano:         systemNano,
+		cpuCountersValid:   cpuErr == nil && systemErr == nil,
+		memoryUsage:        parts[3],
+		memoryUsagePercent: parts[4],
+		networkIO:          parts[5],
+	}, nil
 }
 
-func firstRowString(row map[string]interface{}, keys ...string) string {
-	for _, key := range keys {
-		value, ok := row[key]
-		if !ok || value == nil {
-			continue
-		}
-		switch v := value.(type) {
-		case string:
-			return v
-		case []interface{}:
-			if len(v) > 0 {
-				return fmt.Sprint(v[0])
-			}
-		default:
-			return fmt.Sprint(v)
-		}
+func resourceStatsFromSamples(container Container, previous, current podmanStatsSample) ResourceStats {
+	cpuUsagePercent := 0.0
+	// Podman SystemNano is a wall-clock timestamp, so CPU nanoseconds divided
+	// by its delta yields used cores; the metrics sender converts percent to cores.
+	cpuUsageValid :=
+		previous.cpuCountersValid &&
+			current.cpuCountersValid &&
+			current.cpuNano >= previous.cpuNano &&
+			current.systemNano > previous.systemNano
+	if cpuUsageValid {
+		cpuUsagePercent = 100 * float64(current.cpuNano-previous.cpuNano) /
+			float64(current.systemNano-previous.systemNano)
+		cpuUsageValid = isFinite(cpuUsagePercent)
+	}
+	memoryUsagePercent, memoryUsageValid := parsePercent(current.memoryUsagePercent)
+	memoryUsedBytes, memoryUsedValid := parseMemUsed(current.memoryUsage)
+	rx, tx := parseNetIO(current.networkIO)
+	return ResourceStats{
+		ContainerID:          container.ID,
+		ServiceID:            container.ServiceID,
+		DeploymentID:         container.DeploymentID,
+		CPUUsagePercent:      cpuUsagePercent,
+		CPUUsageValid:        cpuUsageValid,
+		MemoryUsagePercent:   memoryUsagePercent,
+		MemoryUsageValid:     memoryUsageValid,
+		MemoryUsedBytes:      memoryUsedBytes,
+		MemoryUsedValid:      memoryUsedValid,
+		NetworkReceiveBytes:  rx,
+		NetworkTransmitBytes: tx,
 	}
-	return ""
 }
 
-func parsePercent(value string) float64 {
+func parsePercent(value string) (float64, bool) {
 	value = strings.TrimSpace(strings.TrimSuffix(value, "%"))
 	if value == "" || value == "--" {
-		return 0
+		return 0, false
 	}
 	parsed, err := strconv.ParseFloat(value, 64)
 	if err != nil || !isFinite(parsed) {
-		return 0
+		return 0, false
 	}
-	return parsed
+	return parsed, true
 }
 
-func parseMemUsed(value string) float64 {
+func parseMemUsed(value string) (float64, bool) {
 	parts := strings.Split(value, "/")
 	if len(parts) == 0 {
-		return 0
+		return 0, false
 	}
-	return parseByteQuantity(parts[0])
+	return parseByteQuantityValue(parts[0])
 }
 
 func parseNetIO(value string) (float64, float64) {
 	parts := strings.Split(value, "/")
 	if len(parts) != 2 {
 		return 0, 0
 	}
-	return parseByteQuantity(parts[0]), parseByteQuantity(parts[1])
+	rx, _ := parseByteQuantityValue(parts[0])
+	tx, _ := parseByteQuantityValue(parts[1])
+	return rx, tx
 }
 
 func parseByteQuantity(value string) float64 {
+	parsed, _ := parseByteQuantityValue(value)
+	return parsed
+}
+
+func parseByteQuantityValue(value string) (float64, bool) {
 	value = strings.TrimSpace(value)
 	if value == "" || value == "--" {
-		return 0
+		return 0, false
 	}
 
 	compact := strings.ReplaceAll(value, " ", "")
@@ -215,22 +259,22 @@ func parseByteQuantity(value string) float64 {
 	unit := strings.ToLower(compact[splitAt:])
 	parsed, err := strconv.ParseFloat(numberText, 64)
 	if err != nil || !isFinite(parsed) {
-		return 0
+		return 0, false
 	}
 
 	switch unit {
 	case "", "b":
-		return parsed
+		return parsed, true
 	case "kb", "k", "kib", "ki":
-		return parsed * unitMultiplier(unit, 1)
+		return parsed * unitMultiplier(unit, 1), true
 	case "mb", "m", "mib", "mi":
-		return parsed * unitMultiplier(unit, 2)
+		return parsed * unitMultiplier(unit, 2), true
 	case "gb", "g", "gib", "gi":
-		return parsed * unitMultiplier(unit, 3)
+		return parsed * unitMultiplier(unit, 3), true
 	case "tb", "t", "tib", "ti":
-		return parsed * unitMultiplier(unit, 4)
+		return parsed * unitMultiplier(unit, 4), true
 	default:
-		return parsed
+		return 0, false
 	}
 }