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