aiops/agent/collector/proctop.go

// Package collector — 进程 TOP 采集器
// 扫描 /proc 获取 TOP N 进程完整信息：PID/User/CMD/CPU%/MEM%/FDs
// 阿里云 ECS 监控风格 — 直观 6 列布局
//
// 署名 Logan.Li
package collector

import (
	"bufio"
	"os"
	"os/user"
	"path/filepath"
	"sort"
	"strconv"
	"strings"

	"ezops/agent/model"
)

const clktck = 100.0 // Linux CLK_TCK = 100 Hz

// ProcessTopCollector 进程 TOP 采集器
type ProcessTopCollector struct {
	memTotalKB uint64 // 系统总内存（KB），初始化时读取
	numCPU     int    // CPU 核心数
}

// NewProcessTopCollector 创建进程 TOP 采集器
func NewProcessTopCollector() *ProcessTopCollector {
	pt := &ProcessTopCollector{}
	pt.memTotalKB = readMemTotalKB()
	pt.numCPU = readNumCPU()
	if pt.numCPU < 1 {
		pt.numCPU = 1
	}
	return pt
}

// readMemTotalKB reads MemTotal from /proc/meminfo
func readMemTotalKB() uint64 {
	f, err := os.Open("/proc/meminfo")
	if err != nil {
		return 1
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		if strings.HasPrefix(sc.Text(), "MemTotal:") {
			f := strings.Fields(sc.Text())
			if len(f) >= 2 {
				v, _ := strconv.ParseUint(f[1], 10, 64)
				return v
			}
		}
	}
	return 1
}

// readNumCPU reads CPU count from /proc/stat (cpuN lines)
func readNumCPU() int {
	f, err := os.Open("/proc/stat")
	if err != nil {
		return 1
	}
	defer f.Close()
	n := 0
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		if strings.HasPrefix(sc.Text(), "cpu") && len(sc.Text()) > 3 && sc.Text()[3] >= '0' && sc.Text()[3] <= '9' {
			n++
		}
	}
	return n
}

// readProcCPU reads utime+stime from /proc/<pid>/stat
func readProcCPUJiffies(pid string) uint64 {
	data, err := os.ReadFile(filepath.Join("/proc", pid, "stat"))
	if err != nil {
		return 0
	}
	// stat: pid (comm) state ppid ... utime stime ...
	// We need fields 14 (utime) and 15 (stime) in 1-based indexing
	// Parse: skip past the closing ')' of comm
	s := string(data)
	idx := strings.LastIndex(s, ")")
	if idx < 0 {
		return 0
	}
	fields := strings.Fields(s[idx+2:])
	if len(fields) < 13 {
		return 0
	}
	// After ')', fields are: state ppid pgrp session tty_nr tpgid flags minflt cminflt majflt cmajflt utime stime
	// So utime = fields[11], stime = fields[12]
	utime, _ := strconv.ParseUint(fields[11], 10, 64)
	stime, _ := strconv.ParseUint(fields[12], 10, 64)
	return utime + stime
}

// readProcUser reads the owner username from /proc/<pid>/status (Uid line)
func readProcUser(pid string) string {
	statusFile := filepath.Join("/proc", pid, "status")
	f, err := os.Open(statusFile)
	if err != nil {
		return "?"
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	var uidStr string
	for sc.Scan() {
		line := sc.Text()
		if strings.HasPrefix(line, "Uid:") {
			fields := strings.Fields(line)
			if len(fields) >= 2 {
				uidStr = fields[1]
			}
			break
		}
	}
	if uidStr == "" || uidStr == "0" {
		return "root"
	}
	// Try to look up username
	u, err := user.LookupId(uidStr)
	if err == nil {
		return u.Username
	}
	return uidStr
}

// readProcCmdline reads /proc/<pid>/cmdline, replaces null bytes with spaces
func readProcCmdline(pid string) string {
	data, err := os.ReadFile(filepath.Join("/proc", pid, "cmdline"))
	if err != nil || len(data) == 0 {
		// Fallback to comm
		data2, err2 := os.ReadFile(filepath.Join("/proc", pid, "comm"))
		if err2 == nil {
			return strings.TrimSpace(string(data2))
		}
		return "?"
	}
	// Replace null bytes with spaces
	out := strings.Builder{}
	for _, b := range data {
		if b == 0 {
			out.WriteByte(' ')
		} else {
			out.WriteByte(b)
		}
	}
	return strings.TrimSpace(out.String())
}

// Collect gathers process metrics and returns top N sorted by RSS
func (p *ProcessTopCollector) Collect(topN int) ([]model.ProcessStat, error) {
	uptimeJiffies := readUptimeJiffies()
	if uptimeJiffies < 1 {
		uptimeJiffies = 1
	}

	entries, err := os.ReadDir("/proc")
	if err != nil {
		return nil, err
	}

	type sample struct {
		pid     int
		name    string
		cmdline string
		user    string
		cpuPct  float64
		memPct  float64
		rssKB   uint64
		fdCount int
		threads int
	}
	var samples []sample

	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}
		pid, err := strconv.Atoi(entry.Name())
		if err != nil {
			continue
		}
		pidStr := entry.Name()

		// Read comm to filter kernel threads
		commData, err := os.ReadFile(filepath.Join("/proc", pidStr, "comm"))
		if err != nil {
			continue
		}
		name := strings.TrimSpace(string(commData))
		if name == "" || strings.HasPrefix(name, "[") {
			continue // skip kernel threads
		}

		// Gather all metrics
		jiffies := readProcCPUJiffies(pidStr)
		rssKB := readProcRSS(pidStr)
		fdCount := countFD(pidStr)
		threads := readProcThreads(pidStr)
		userName := readProcUser(pidStr)
		cmdline := readProcCmdline(pidStr)

		// CPU% = (utime+stime) / uptime_jiffies * 100 / num_cpus
		cpuPct := float64(jiffies) / float64(uptimeJiffies) * 100.0 / float64(p.numCPU)
		if cpuPct > 100*float64(p.numCPU) {
			cpuPct = 0 // invalid
		}

		// Memory%
		var memPct float64
		if p.memTotalKB > 0 {
			memPct = float64(rssKB) / float64(p.memTotalKB) * 100.0
		}

		samples = append(samples, sample{
			pid:     pid,
			name:    name,
			cmdline: cmdline,
			user:    userName,
			cpuPct:  cpuPct,
			memPct:  memPct,
			rssKB:   rssKB,
			fdCount: fdCount,
			threads: threads,
		})
	}

	// Sort by RSS descending
	sort.Slice(samples, func(i, j int) bool {
		return samples[i].rssKB > samples[j].rssKB
	})

	if topN > len(samples) {
		topN = len(samples)
	}
	result := make([]model.ProcessStat, topN)
	for i := 0; i < topN; i++ {
		s := samples[i]
		result[i] = model.ProcessStat{
			PID:      s.pid,
			Name:     s.name,
			Cmdline:  s.cmdline,
			User:     s.user,
			CPUPct:   s.cpuPct,
			MemPct:   s.memPct,
			MemoryMB: float64(s.rssKB) / 1024.0,
			FDCount:  s.fdCount,
			Threads:  s.threads,
		}
	}
	return result, nil
}

// readUptimeJiffies reads /proc/uptime and converts to jiffies
func readUptimeJiffies() uint64 {
	data, err := os.ReadFile("/proc/uptime")
	if err != nil {
		return 1
	}
	f := strings.Fields(string(data))
	if len(f) < 1 {
		return 1
	}
	sec, _ := strconv.ParseFloat(f[0], 64)
	return uint64(sec * clktck)
}

// countFD counts open file descriptors for a process
func countFD(pid string) int {
	entries, err := os.ReadDir(filepath.Join("/proc", pid, "fd"))
	if err != nil {
		return 0
	}
	return len(entries)
}

// readProcRSS reads VmRSS from /proc/<pid>/status
func readProcRSS(pid string) uint64 {
	f, err := os.Open(filepath.Join("/proc", pid, "status"))
	if err != nil {
		return 0
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		if strings.HasPrefix(sc.Text(), "VmRSS:") {
			f := strings.Fields(sc.Text())
			if len(f) >= 2 {
				v, _ := strconv.ParseUint(f[1], 10, 64)
				return v
			}
		}
	}
	return 0
}

// readProcThreads reads Threads from /proc/<pid>/status
func readProcThreads(pid string) int {
	f, err := os.Open(filepath.Join("/proc", pid, "status"))
	if err != nil {
		return 0
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		if strings.HasPrefix(sc.Text(), "Threads:") {
			f := strings.Fields(sc.Text())
			if len(f) >= 2 {
				v, _ := strconv.Atoi(f[1])
				return v
			}
		}
	}
	return 0
}