aiops/agent/collector/proctop.go

// Package collector — 进程 TOP3 采集器
// 扫描 /proc/*/stat + /proc/*/status 获取 CPU/内存/句柄 TOP3 进程
// 1000 服务器规模下的优化：只取 TOP3，不遍历全部进程的 cmdline
//
// 署名 Logan.Li
package collector

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

	"ezops/agent/model"
)

// ProcessTopCollector 进程 TOP 采集器
type ProcessTopCollector struct{}

// NewProcessTopCollector 创建进程 TOP 采集器
func NewProcessTopCollector() *ProcessTopCollector {
	return &ProcessTopCollector{}
}

// procSample holds key metrics for a single process
type procSample struct {
	PID     int
	Name    string
	CPU     float64 // CPU 使用率（需两次采样 + 系统总 jiffies 才能算，这里用 utime+stime ticks 近似）
	RSSKB   uint64  // 物理内存 (KB)
	FDCount int     // 打开的文件句柄数
	Threads int     // 线程数
}

// Collect gathers process metrics and returns top N sorted by RSS.
// Sorting by RSS is the most stable and reliable metric without multi-sampling CPU.
func (p *ProcessTopCollector) Collect(topN int) ([]model.ProcessStat, error) {
	entries, err := os.ReadDir("/proc")
	if err != nil {
		return nil, err
	}

	var samples []procSample
	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}
		pid, err := strconv.Atoi(entry.Name())
		if err != nil {
			continue
		}

		name := readProcName(entry.Name())
		// Skip kernel threads (empty cmdline + running in kernel)
		if name == "" || strings.HasPrefix(name, "[") {
			continue
		}

		rssKB := readProcRSS(entry.Name())
		fdCount := countFD(entry.Name())
		threads := readProcThreads(entry.Name())

		samples = append(samples, procSample{
			PID:     pid,
			Name:    name,
			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
	})

	// Take top N
	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,
			MemoryMB: float64(s.RSSKB) / 1024.0,
			FDCount:  s.FDCount,
			Threads:  s.Threads,
		}
	}
	return result, nil
}

// readProcName reads /proc/<pid>/comm
func readProcName(pid string) string {
	data, err := os.ReadFile(filepath.Join("/proc", pid, "comm"))
	if err != nil {
		return ""
	}
	return strings.TrimSpace(string(data))
}

// 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()

	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		line := scanner.Text()
		if strings.HasPrefix(line, "VmRSS:") {
			fields := strings.Fields(line)
			if len(fields) >= 2 {
				val, _ := strconv.ParseUint(fields[1], 10, 64)
				return val
			}
		}
	}
	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()

	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		line := scanner.Text()
		if strings.HasPrefix(line, "Threads:") {
			fields := strings.Fields(line)
			if len(fields) >= 2 {
				val, _ := strconv.Atoi(fields[1])
				return val
			}
		}
	}
	return 0
}

// 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)
}