CLICD/backend/internal/api/container_metrics.go

package api

import (
	"encoding/json"
	"fmt"
	"math"
	"strconv"
	"sync"
	"time"

	"clicd/internal/config"
)

type ContainerMetricPoint struct {
	TS        int64   `json:"ts"`
	CPU       float64 `json:"cpu"`
	Memory    float64 `json:"memory"`
	Network   float64 `json:"network"`
	NetworkRx float64 `json:"network_rx"`
	NetworkTx float64 `json:"network_tx"`
	DiskIO    float64 `json:"disk_io"`
	DiskRead  float64 `json:"disk_read"`
	DiskWrite float64 `json:"disk_write"`
}

var containerMetricSamplerOnce sync.Once
var containerMetricMu sync.RWMutex
var containerMetricHistory = map[string][]ContainerMetricPoint{}
var containerMetricInFlight sync.Map

const (
	containerMetricSampleInterval = 30 * time.Second
	containerMetricSampleTimeout  = 20 * time.Second
	containerMetricConcurrency    = 4
)

func StartContainerMetricSampler() {
	containerMetricSamplerOnce.Do(func() {
		go func() {
			sampleAllContainerMetrics()
			ticker := time.NewTicker(containerMetricSampleInterval)
			defer ticker.Stop()
			for range ticker.C {
				sampleAllContainerMetrics()
			}
		}()
	})
}

func sampleAllContainerMetrics() {
	containers, _ := listByRuntime()
	sem := make(chan struct{}, containerMetricConcurrency)
	var wg sync.WaitGroup

	for _, c := range containers {
		c := c
		if c.Status != "running" {
			continue
		}
		sem <- struct{}{}
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() { <-sem }()
			sampleContainerMetricWithTimeout(c)
		}()
	}
	wg.Wait()
	pruneContainerMetricHistory()
}

func sampleContainerMetricWithTimeout(c config.Container) {
	key := containerMetricKey(c)
	if key == "" {
		return
	}
	if _, loaded := containerMetricInFlight.LoadOrStore(key, struct{}{}); loaded {
		return
	}
	done := make(chan struct{}, 1)
	go func() {
		defer containerMetricInFlight.Delete(key)
		if usage, err := usageByRuntime(c.ID); err == nil {
			appendContainerMetricPoint(c, usage)
		}
		done <- struct{}{}
	}()

	select {
	case <-done:
	case <-time.After(containerMetricSampleTimeout):
	}
}

func appendContainerMetricPoint(c config.Container, usage map[string]interface{}) {
	key := containerMetricKey(c)
	if key == "" {
		return
	}
	memoryTotal := numberFromUsage(usage, "memory_total_bytes")
	if memoryTotal <= 0 {
		memoryTotal = float64(c.RAMMB) * 1024 * 1024
	}
	memoryPct := 0.0
	if memoryTotal > 0 {
		memoryPct = clampPercent(numberFromUsage(usage, "memory_usage_bytes") / memoryTotal * 100)
	}
	vcpu := c.VCPU
	if vcpu <= 0 {
		vcpu = 1
	}
	cpuPct := clampPercent(numberFromUsage(usage, "cpu_usage_pct") / vcpu)
	networkRx := positiveNumberFromUsage(usage, "network_rx_bps")
	networkTx := positiveNumberFromUsage(usage, "network_tx_bps")
	diskRead := positiveNumberFromUsage(usage, "disk_read_bps")
	diskWrite := positiveNumberFromUsage(usage, "disk_write_bps")
	point := ContainerMetricPoint{
		TS:        time.Now().UnixMilli(),
		CPU:       cpuPct,
		Memory:    memoryPct,
		NetworkRx: networkRx,
		NetworkTx: networkTx,
		Network:   networkRx + networkTx,
		DiskRead:  diskRead,
		DiskWrite: diskWrite,
		DiskIO:    diskRead + diskWrite,
	}
	cutoff := time.Now().Add(-hostMetricRetention).UnixMilli()

	containerMetricMu.Lock()
	defer containerMetricMu.Unlock()

	history := containerMetricHistory[key]
	keepFrom := 0
	for keepFrom < len(history) && history[keepFrom].TS < cutoff {
		keepFrom++
	}
	if keepFrom > 0 {
		copy(history, history[keepFrom:])
		history = history[:len(history)-keepFrom]
	}
	containerMetricHistory[key] = append(history, point)
}

func getContainerMetricHistory(c *config.Container) []ContainerMetricPoint {
	if c == nil {
		return nil
	}
	key := containerMetricKey(*c)
	containerMetricMu.RLock()
	defer containerMetricMu.RUnlock()

	history := containerMetricHistory[key]
	result := make([]ContainerMetricPoint, len(history))
	copy(result, history)
	return result
}

func pruneContainerMetricHistory() {
	cutoff := time.Now().Add(-hostMetricRetention).UnixMilli()
	valid := map[string]bool{}
	if config.AppConfig != nil {
		for _, c := range config.AppConfig.Containers {
			valid[containerMetricKey(c)] = true
		}
	}

	containerMetricMu.Lock()
	defer containerMetricMu.Unlock()

	for key, history := range containerMetricHistory {
		if !valid[key] {
			delete(containerMetricHistory, key)
			continue
		}
		keepFrom := 0
		for keepFrom < len(history) && history[keepFrom].TS < cutoff {
			keepFrom++
		}
		if keepFrom > 0 {
			copy(history, history[keepFrom:])
			containerMetricHistory[key] = history[:len(history)-keepFrom]
		}
	}
}

func containerMetricKey(c config.Container) string {
	if c.UUID != "" {
		return "uuid:" + c.UUID
	}
	if c.ID > 0 {
		return fmt.Sprintf("id:%d", c.ID)
	}
	if c.Name != "" {
		return "name:" + c.Name
	}
	return ""
}

func numberFromUsage(usage map[string]interface{}, key string) float64 {
	value, ok := usage[key]
	if !ok || value == nil {
		return 0
	}
	switch v := value.(type) {
	case float64:
		if math.IsNaN(v) || math.IsInf(v, 0) {
			return 0
		}
		return v
	case float32:
		return float64(v)
	case int:
		return float64(v)
	case int64:
		return float64(v)
	case int32:
		return float64(v)
	case uint:
		return float64(v)
	case uint64:
		return float64(v)
	case uint32:
		return float64(v)
	case json.Number:
		n, _ := v.Float64()
		return n
	case string:
		n, _ := strconv.ParseFloat(v, 64)
		return n
	default:
		return 0
	}
}

func positiveNumberFromUsage(usage map[string]interface{}, key string) float64 {
	value := numberFromUsage(usage, key)
	if value < 0 {
		return 0
	}
	return value
}