ProxmoxDash/io_attribution.go

package main

import (
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

type IOConsumer struct {
	Kind             string  `json:"kind"`
	Name             string  `json:"name"`
	VMID             int     `json:"vmid,omitempty"`
	ReadBytesPerSec  float64 `json:"readBytesPerSec"`
	WriteBytesPerSec float64 `json:"writeBytesPerSec"`
	SharePercent     float64 `json:"sharePercent"`
}
type processIOSnapshot struct {
	read, write           uint64
	name, cmdline, cgroup string
}
type ioAttributionCollector struct {
	mu        sync.Mutex
	previous  map[int]processIOSnapshot
	updatedAt time.Time
	cached    []IOConsumer
}

var qemuIDPattern = regexp.MustCompile(`(?:^|\s)-id\s+(\d+)`)
var lxcIDPattern = regexp.MustCompile(`(?:lxc(?:\.payload)?[./-]|pve-container@)(\d+)`)

func (c *ioAttributionCollector) collect(guests GuestsMetrics) []IOConsumer {
	c.mu.Lock()
	defer c.mu.Unlock()
	now := time.Now()
	current := readProcessIO()
	elapsed := now.Sub(c.updatedAt).Seconds()
	if c.previous == nil || elapsed <= 0 {
		c.previous = current
		c.updatedAt = now
		return nil
	}
	names := map[int]string{}
	for _, g := range guests.Guests {
		names[g.VMID] = g.Name
	}
	grouped := map[string]*IOConsumer{}
	var total float64
	for pid, snapshot := range current {
		old, ok := c.previous[pid]
		if !ok || snapshot.read < old.read || snapshot.write < old.write {
			continue
		}
		read := float64(snapshot.read-old.read) / elapsed
		write := float64(snapshot.write-old.write) / elapsed
		if read+write < 1024 {
			continue
		}
		kind, name, vmid := classifyIOProcess(snapshot, names)
		key := kind + ":" + strconv.Itoa(vmid) + ":" + name
		consumer := grouped[key]
		if consumer == nil {
			consumer = &IOConsumer{Kind: kind, Name: name, VMID: vmid}
			grouped[key] = consumer
		}
		consumer.ReadBytesPerSec += read
		consumer.WriteBytesPerSec += write
		total += read + write
	}
	var result []IOConsumer
	for _, v := range grouped {
		if total > 0 {
			v.SharePercent = (v.ReadBytesPerSec + v.WriteBytesPerSec) / total * 100
		}
		result = append(result, *v)
	}
	sort.Slice(result, func(i, j int) bool {
		return result[i].ReadBytesPerSec+result[i].WriteBytesPerSec > result[j].ReadBytesPerSec+result[j].WriteBytesPerSec
	})
	if len(result) > 8 {
		result = result[:8]
	}
	c.previous = current
	c.updatedAt = now
	c.cached = result
	return result
}

func classifyIOProcess(p processIOSnapshot, names map[int]string) (kind, name string, vmid int) {
	if match := qemuIDPattern.FindStringSubmatch(p.cmdline); len(match) > 1 {
		vmid, _ = strconv.Atoi(match[1])
		return "VM", firstNonEmpty(names[vmid], "VM "+match[1]), vmid
	}
	if match := lxcIDPattern.FindStringSubmatch(p.cgroup); len(match) > 1 {
		vmid, _ = strconv.Atoi(match[1])
		return "LXC", firstNonEmpty(names[vmid], "LXC "+match[1]), vmid
	}
	return "Процесс", firstNonEmpty(p.name, "PID"), 0
}

func readProcessIO() map[int]processIOSnapshot {
	result := map[int]processIOSnapshot{}
	entries, _ := filepath.Glob("/proc/[0-9]*/io")
	for _, path := range entries {
		pid, _ := strconv.Atoi(filepath.Base(filepath.Dir(path)))
		data, err := os.ReadFile(path)
		if err != nil {
			continue
		}
		var item processIOSnapshot
		for _, line := range strings.Split(string(data), "\n") {
			key, value, ok := strings.Cut(line, ":")
			if !ok {
				continue
			}
			n, _ := strconv.ParseUint(strings.TrimSpace(value), 10, 64)
			if key == "read_bytes" {
				item.read = n
			} else if key == "write_bytes" {
				item.write = n
			}
		}
		comm, _ := os.ReadFile(filepath.Join(filepath.Dir(path), "comm"))
		cmd, _ := os.ReadFile(filepath.Join(filepath.Dir(path), "cmdline"))
		cg, _ := os.ReadFile(filepath.Join(filepath.Dir(path), "cgroup"))
		item.name = strings.TrimSpace(string(comm))
		item.cmdline = strings.ReplaceAll(string(cmd), "\x00", " ")
		item.cgroup = string(cg)
		result[pid] = item
	}
	return result
}

func attributeDiskIO(metrics *dashboardMetrics, consumers []IOConsumer) {
	active := activeIOOperation(metrics.Activity.Tasks)
	for i := range metrics.Disks {
		disk := &metrics.Disks[i]
		disk.TopConsumers = append([]IOConsumer(nil), consumers...)
		if active != "" {
			disk.IOCause = active
			disk.IOCauseDetail = "Активная задача Proxmox совпадает по времени с дисковой нагрузкой"
			disk.IOConfidence = 75
		} else if len(consumers) > 0 {
			top := consumers[0]
			disk.IOCause = top.Kind + " · " + top.Name
			disk.IOCauseDetail = "Крупнейший потребитель host I/O: " + strconv.FormatFloat(top.SharePercent, 'f', 1, 64) + "%"
			disk.IOConfidence = 55
		}
	}
}

func activeIOOperation(tasks []ProxmoxTask) string {
	for _, task := range tasks {
		if task.EndTime != 0 && task.Status != "" {
			continue
		}
		label := ""
		switch task.Type {
		case "vzdump":
			label = "Backup"
		case "qmigrate":
			label = "Миграция"
		case "qmsnapshot", "vzsnapshot":
			label = "Snapshot"
		}
		if label != "" {
			if task.ID != "" {
				label += " VM/LXC " + task.ID
			}
			return label
		}
	}
	return ""
}