package hostcollect import ( "context" "errors" "fmt" "os" "path/filepath" "sort" "strconv" "strings" "time" "github.com/itworx/pulse/internal/process" ) // maxProcessCPUPercent matches the bound internal/process enforces. A busy multi-core // task legitimately exceeds 100%. const maxProcessCPUPercent = 10000 // processKey identifies a task across scans. The PID alone is not enough: PIDs are // reused, and charging a new process with the CPU time of the dead one it replaced // would show a fresh task at an absurd utilisation. The kernel's start time makes the // identity stable. type processKey struct { pid int startTime uint64 } type processCPUSample struct { at time.Time ticks uint64 } // processStat is the subset of /proc//stat the inventory needs. type processStat struct { pid int comm string state string ticks uint64 startTime uint64 rssPages uint64 } func (s processStat) key() processKey { return processKey{pid: s.pid, startTime: s.startTime} } // parseProcessStat reads /proc//stat. // // The second field is the executable name in parentheses and it is neither escaped nor // quoted: a process can legitimately be called ") (" or "my app (2)". Splitting the // line on whitespace therefore corrupts every field after it. The parser instead cuts // at the LAST closing parenthesis, which is the only reliable delimiter. func parseProcessStat(data []byte, pid int) (processStat, error) { text := string(data) open := strings.IndexByte(text, '(') closing := strings.LastIndexByte(text, ')') if open < 0 || closing < open { return processStat{}, errors.New("hostcollect: process stat has no comm field") } comm := text[open+1 : closing] fields := strings.Fields(text[closing+1:]) // Field 3 (state) becomes index 0 here; the inventory needs up to field 24 (rss). const ( stateIndex = 0 utimeIndex = 11 stimeIndex = 12 startTimeIndex = 19 rssIndex = 21 ) if len(fields) <= rssIndex { return processStat{}, errors.New("hostcollect: process stat is truncated") } utime, utimeOK := parseUint(fields[utimeIndex]) stime, stimeOK := parseUint(fields[stimeIndex]) startTime, startOK := parseUint(fields[startTimeIndex]) if !utimeOK || !stimeOK || !startOK { return processStat{}, errors.New("hostcollect: process stat has malformed counters") } // rss is signed in the kernel's own printf but never negative in practice; a // negative or malformed value degrades to zero rather than failing the process. rss, _ := parseUint(fields[rssIndex]) return processStat{ pid: pid, comm: sanitizeName(comm), state: processState(fields[stateIndex]), ticks: utime + stime, startTime: startTime, rssPages: rss, }, nil } // processState expands the kernel's single-letter state so the UI does not have to. func processState(value string) string { if value == "" { return "unknown" } switch value[0] { case 'R': return "running" case 'S': return "sleeping" case 'D': return "uninterruptible" case 'Z': return "zombie" case 'T': return "stopped" case 't': return "tracing-stop" case 'X', 'x': return "dead" case 'I': return "idle" case 'K': return "wakekill" case 'W': return "waking" case 'P': return "parked" default: return "unknown" } } // Processes reads the process inventory. // // A process that exits between readdir and open is the normal case, not an error: the // scan skips it and keeps going. Only a failure that invalidates the whole scan (no // uptime, unreadable procfs root) returns an error, because a half-empty inventory // published as complete would be worse than no inventory at all. func (c *Collector) Processes(ctx context.Context) (process.RawSnapshot, error) { if c == nil { return process.RawSnapshot{}, errors.New("hostcollect: collector is nil") } if err := ctx.Err(); err != nil { return process.RawSnapshot{}, err } now := c.options.Now().UTC() uptimeSeconds, err := c.uptimeSeconds() if err != nil { return process.RawSnapshot{}, err } entries, err := os.ReadDir(c.options.ProcRoot) if err != nil { return process.RawSnapshot{}, fmt.Errorf("hostcollect: list processes: %w", err) } stats := make([]processStat, 0, len(entries)) for _, entry := range entries { if err := ctx.Err(); err != nil { return process.RawSnapshot{}, err } pid, ok := processID(entry.Name()) if !ok { continue } data, readErr := readLimited(c.procPath(entry.Name(), "stat"), maxProcessFileBytes) if readErr != nil { continue } stat, parseErr := parseProcessStat(data, pid) if parseErr != nil { continue } stats = append(stats, stat) } percentages := c.cpuPercentages(now, stats) rows := c.rank(stats, percentages) processes := make([]process.RawProcess, 0, len(rows)) for _, stat := range rows { if err := ctx.Err(); err != nil { return process.RawSnapshot{}, err } row, ok := c.describe(stat, uptimeSeconds, percentages[stat.key()]) if !ok { continue } processes = append(processes, row) } sort.Slice(processes, func(i, j int) bool { return processes[i].PID < processes[j].PID }) return process.RawSnapshot{ Source: process.Source{ ID: "process", Type: "agent", CapabilityVersion: process.ContractVersion, ObservedAt: now, }, Processes: processes, ObservedAt: now, }, nil } // cpuPercentages turns accumulated CPU ticks into a utilisation percentage relative to // the previous scan and replaces the retained state with the current one. The map is // rebuilt from the live process list every scan, so it cannot grow without bound as // processes come and go. func (c *Collector) cpuPercentages(now time.Time, stats []processStat) map[processKey]float64 { percentages := make(map[processKey]float64, len(stats)) current := make(map[processKey]processCPUSample, len(stats)) c.mu.Lock() previous := c.processes for _, stat := range stats { key := stat.key() current[key] = processCPUSample{at: now, ticks: stat.ticks} sample, seen := previous[key] if !seen { continue } elapsed := now.Sub(sample.at).Seconds() if elapsed <= 0 || stat.ticks < sample.ticks { continue } percent := float64(stat.ticks-sample.ticks) / c.options.ClockTicks / elapsed * 100 if percent < 0 { percent = 0 } if percent > maxProcessCPUPercent { percent = maxProcessCPUPercent } percentages[key] = percent } c.processes = current c.mu.Unlock() return percentages } // rank keeps the busiest processes when the inventory exceeds MaxRows. Truncating by // cost rather than by PID order keeps the rows an operator actually needs. func (c *Collector) rank(stats []processStat, percentages map[processKey]float64) []processStat { ordered := append([]processStat(nil), stats...) sort.SliceStable(ordered, func(i, j int) bool { left, right := percentages[ordered[i].key()], percentages[ordered[j].key()] if left != right { return left > right } if ordered[i].rssPages != ordered[j].rssPages { return ordered[i].rssPages > ordered[j].rssPages } return ordered[i].pid < ordered[j].pid }) if len(ordered) > c.options.ProcessLimits.MaxRows { ordered = ordered[:c.options.ProcessLimits.MaxRows] } return ordered } // describe fills in the fields that need a second read, for the retained rows only. // // Only the program name is taken from the command line, never the arguments: argv // routinely carries tokens, passwords and connection strings, and the snapshot is // stored and rendered. The read itself is bounded, so an enormous argv costs one page, // not a copy of the whole command line. func (c *Collector) describe(stat processStat, uptimeSeconds, cpuPercent float64) (process.RawProcess, bool) { directory := strconv.Itoa(stat.pid) name := stat.comm if data, err := readLimited(c.procPath(directory, "cmdline"), maxCmdlineBytes); err == nil { if argv0 := commandName(data); argv0 != "" { name = argv0 } } if name == "" { // Neither comm nor cmdline is usable — the task is gone or unreadable. return process.RawProcess{}, false } memoryBytes := stat.rssPages * uint64(c.options.PageSize) //nolint:gosec // PageSize is validated positive if rss, ok := c.residentBytes(directory); ok { memoryBytes = rss } runtimeSeconds := uptimeSeconds - float64(stat.startTime)/c.options.ClockTicks if runtimeSeconds < 0 || runtimeSeconds > maxUptimeSeconds { runtimeSeconds = 0 } return process.RawProcess{ PID: stat.pid, Name: truncate(name, maxNameBytes), State: stat.state, RuntimeSeconds: runtimeSeconds, CPUPercent: cpuPercent, MemoryBytes: memoryBytes, }, true } // residentBytes prefers VmRSS from /proc//status, which the kernel already // reports in kB and keeps consistent, over the page count in stat. Kernel threads have // no VmRSS at all, which is why the stat value remains the fallback. func (c *Collector) residentBytes(directory string) (uint64, bool) { data, err := readLimited(c.procPath(directory, "status"), maxProcessFileBytes) if err != nil { return 0, false } for _, line := range lines(data) { key, rest, found := strings.Cut(line, ":") if !found || key != "VmRSS" { continue } fields := strings.Fields(rest) if len(fields) == 0 { return 0, false } value, ok := parseUint(fields[0]) if !ok { return 0, false } if len(fields) > 1 && strings.EqualFold(fields[1], "kB") { const kib = 1024 if value > (1<<64-1)/kib { return 0, false } value *= kib } return value, true } return 0, false } // commandName extracts the program name from a NUL separated command line. func commandName(data []byte) string { text := string(data) if index := strings.IndexByte(text, 0); index >= 0 { text = text[:index] } text = strings.TrimSpace(text) if text == "" { return "" } return sanitizeName(filepath.Base(text)) } // sanitizeName drops control characters so a hostile process name cannot inject // terminal escapes or NUL bytes into logs and stored snapshots. func sanitizeName(value string) string { cleaned := strings.Map(func(r rune) rune { if r < 0x20 || r == 0x7f { return -1 } return r }, value) return truncate(strings.TrimSpace(cleaned), maxNameBytes) } func processID(name string) (int, bool) { pid, err := strconv.Atoi(name) if err != nil || pid < 1 { return 0, false } return pid, true }