Add IO attribution anomaly detection and alert correlation

io_attribution.go

@@ -0,0 +1,178 @@
+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 ""
+}