diff --git a/docs/PLUGIN_DOC.md b/docs/PLUGIN_DOC.md index 5fcad7d0..a0020ca5 100644 --- a/docs/PLUGIN_DOC.md +++ b/docs/PLUGIN_DOC.md @@ -23,7 +23,7 @@ | OsPlugin | sh -c '( lsb_release -ds || (cat /etc/\*release | grep PRETTY_NAME) || uname -om ) 2>/dev/null | head -n1'
cat /etc/\*release | grep VERSION_ID
wmic os get Version /value
wmic os get Caption /Value | **Analyzer Args:**
- `exp_os`: Union[str, list] — Expected OS name/version string(s) to match (e.g. from lsb_release or /etc/os-release).
- `exact_match`: bool — If True, require exact match for exp_os; otherwise substring match. | - | [OsDataModel](#OsDataModel-Model) | [OsCollector](#Collector-Class-OsCollector) | [OsAnalyzer](#Data-Analyzer-Class-OsAnalyzer) | | PackagePlugin | dnf list --installed
dpkg-query -W
pacman -Q
cat /etc/\*release
wmic product get name,version | **Analyzer Args:**
- `exp_package_ver`: Dict[str, Optional[str]] — Map package name -> expected version (None = any version). Checked against installed packages.
- `regex_match`: bool — If True, match package versions with regex; otherwise exact or prefix match.
- `rocm_regex`: Optional[str] — Optional regex to identify ROCm package version (used when enable_rocm_regex is True).
- `enable_rocm_regex`: bool — If True, use rocm_regex (or default pattern) to extract ROCm version for checks. | - | [PackageDataModel](#PackageDataModel-Model) | [PackageCollector](#Collector-Class-PackageCollector) | [PackageAnalyzer](#Data-Analyzer-Class-PackageAnalyzer) | | PciePlugin | lspci -d {vendor_id}: -nn
lspci -x
lspci -xxxx
lspci -PP
lspci -PP -d {vendor_id}:{dev_id}
lspci -PP -D -d {vendor_id}:{dev_id}
lspci -PP -D
lspci -vvv
lspci -vvvt | **Analyzer Args:**
- `exp_speed`: int — Expected PCIe link speed (generation 1–5).
- `exp_width`: int — Expected PCIe link width in lanes (1–16).
- `exp_sriov_count`: int — Expected SR-IOV virtual function count.
- `exp_gpu_count_override`: Optional[int] — Override expected GPU count for validation.
- `exp_max_payload_size`: Union[Dict[int, int], int, NoneType] — Expected max payload size: int for all devices, or dict keyed by device ID.
- `exp_max_rd_req_size`: Union[Dict[int, int], int, NoneType] — Expected max read request size: int for all devices, or dict keyed by device ID.
- `exp_ten_bit_tag_req_en`: Union[Dict[int, int], int, NoneType] — Expected 10-bit tag request enable: int for all devices, or dict keyed by device ID. | - | [PcieDataModel](#PcieDataModel-Model) | [PcieCollector](#Collector-Class-PcieCollector) | [PcieAnalyzer](#Data-Analyzer-Class-PcieAnalyzer) | -| ProcessPlugin | top -b -n 1
rocm-smi --showpids
top -b -n 1 -o %CPU | **Analyzer Args:**
- `max_kfd_processes`: int — Maximum allowed number of KFD (Kernel Fusion Driver) processes; 0 disables the check.
- `max_cpu_usage`: float — Maximum allowed CPU usage (percent) for process checks. | **Collection Args:**
- `html_view`: bool — When true, include logged command artifacts in command_artifacts.html using human-readable output.
- `top_n_process`: int — Number of top processes by CPU usage to collect (e.g. for top -b -n 1 -o %%CPU). | [ProcessDataModel](#ProcessDataModel-Model) | [ProcessCollector](#Collector-Class-ProcessCollector) | [ProcessAnalyzer](#Data-Analyzer-Class-ProcessAnalyzer) | +| ProcessPlugin | for p in {pids}; do printf "%s:" "$p"; cat /proc/$p/comm 2>/dev/null || true; printf "\n"; done
printf "__SAMPLER__:%s\n" "$$"; for f in /proc/[0-9]\*/stat; do [ -r "$f" ] || continue; pid="${f#/proc/}"; pid="${pid%/stat}"; [ "$pid" = "$$" ] && continue; printf "%s|" "$pid"; cat "$f" 2>/dev/null || true; printf "\n"; done
cat /proc/stat | **Analyzer Args:**
- `max_cpu_usage`: float — Maximum allowed aggregate CPU execution percentage; idle and I/O-wait time are excluded. | **Collection Args:**
- `html_view`: bool — When true, include logged command artifacts in command_artifacts.html using human-readable output.
- `top_n_process`: int — Number of top processes by CPU usage to collect.
- `sample_interval_seconds`: float — Wall-clock interval between procfs CPU samples. | [ProcessDataModel](#ProcessDataModel-Model) | [ProcessCollector](#Collector-Class-ProcessCollector) | [ProcessAnalyzer](#Data-Analyzer-Class-ProcessAnalyzer) | | RdmaPlugin | rdma link -j
rdma dev
rdma link
rdma statistic -j | **Analyzer Args:**
- `exclusion_regex`: Optional[list[str]] — Regex patterns matched against an interface netdev; matching interfaces are skipped. | **Collection Args:**
- `html_view`: bool — When true, include logged command artifacts in command_artifacts.html using human-readable output. | [RdmaDataModel](#RdmaDataModel-Model) | [RdmaCollector](#Collector-Class-RdmaCollector) | [RdmaAnalyzer](#Data-Analyzer-Class-RdmaAnalyzer) | | RegexSearchPlugin | No collector step: reads local text from the CLI --data path (file or directory).
Directory scans load each file's contents into RegexSearchData for analysis. | Runs RegexSearchAnalyzer: user-defined patterns via analysis_args.error_regex (same shape as Dmesg).
Emits regex match events with optional per-file source in the description when scanning directories.
**Analyzer Args:**
- `error_regex`: Optional[list[dict[str, Any]]] — Regex patterns to search for; each dict may include regex (str), message, event_category, event_priority (same as Dme...
- `interval_to_collapse_event`: int — Seconds within which repeated events are collapsed into one.
- `num_timestamps`: int — Number of timestamps to include per event in output. | - | [RegexSearchData](#RegexSearchData-Model) | - | [RegexSearchAnalyzer](#Data-Analyzer-Class-RegexSearchAnalyzer) | | RocmPlugin | {rocm_path}/opencl/bin/\*/clinfo
env | grep -Ei 'rocm|hsa|hip|mpi|openmp|ucx|miopen'
ls /sys/class/kfd/kfd/proc/
grep -i -E 'rocm' /etc/ld.so.conf.d/\*
{rocm_path}/bin/rocminfo
ls -v -d {rocm_path}\*
ls -v -d {rocm_path}-[3-7]\* | tail -1
ldconfig -p | grep -i -E 'rocm'
grep . -H -r -i {rocm_path}/.info/\* | **Analyzer Args:**
- `exp_rocm`: Union[str, list] — Expected ROCm version string(s) to match (e.g. from rocminfo).
- `exp_rocm_latest`: str — Expected 'latest' ROCm path or version string for versioned installs.
- `exp_rocm_sub_versions`: dict[str, Union[str, list]] — Map sub-version name (e.g. version_rocm) to expected string or list of allowed strings. | **Collection Args:**
- `html_view`: bool — When true, include logged command artifacts in command_artifacts.html using human-readable output.
- `rocm_path`: str — Base path to ROCm installation (e.g. /opt/rocm). Used for rocminfo, clinfo, and version discovery. | [RocmDataModel](#RocmDataModel-Model) | [RocmCollector](#Collector-Class-RocmCollector) | [RocmAnalyzer](#Data-Analyzer-Class-RocmAnalyzer) | @@ -982,7 +982,7 @@ PcieDataModel ### Description -Collect Process details +Collect aggregate CPU usage and top processes from Linux procfs. **Bases**: ['InBandDataCollector'] @@ -991,9 +991,9 @@ Collect Process details ### Class Variables - **SUPPORTED_OS_FAMILY**: `{}` -- **CMD_KFD**: `rocm-smi --showpids` -- **CMD_CPU_USAGE**: `top -b -n 1` -- **CMD_PROCESS**: `top -b -n 1 -o %CPU ` +- **CMD_PROC_STAT**: `cat /proc/stat` +- **CMD_PROCESS_STAT**: `printf "__SAMPLER__:%s\n" "$$"; for f in /proc/[0-9]*/stat; do [ -r "$f" ] || continue; pid="${f#/proc/}"; pid="${pid%/stat}"; [ "$pid" = "$$" ] && continue; printf "%s|" "$pid"; cat "$f" 2>/dev/null || true; printf "\n"; done` +- **CMD_PROCESS_NAMES**: `for p in {pids}; do printf "%s:" "$p"; cat /proc/$p/comm 2>/dev/null || true; printf "\n"; done` ### Provides Data @@ -1001,9 +1001,9 @@ ProcessDataModel ### Commands -- top -b -n 1 -- rocm-smi --showpids -- top -b -n 1 -o %CPU +- for p in {pids}; do printf "%s:" "$p"; cat /proc/$p/comm 2>/dev/null || true; printf "\n"; done +- printf "__SAMPLER__:%s\n" "$$"; for f in /proc/[0-9]\*/stat; do [ -r "$f" ] || continue; pid="${f#/proc/}"; pid="${pid%/stat}"; [ "$pid" = "$$" ] && continue; printf "%s|" "$pid"; cat "$f" 2>/dev/null || true; printf "\n"; done +- cat /proc/stat ## Collector Class RdmaCollector @@ -1882,13 +1882,16 @@ class for collection of PCIe data. ## ProcessDataModel Model +### Description + +CPU execution usage (excluding idle and I/O-wait) and top processes from procfs. + **Link to code**: [processdata.py](https://github.com/amd/node-scraper/blob/HEAD/nodescraper/plugins/inband/process/processdata.py) **Bases**: ['DataModel'] ### Model annotations and fields -- **kfd_process**: `Optional[int]` - **cpu_usage**: `Optional[float]` - **processes**: `Optional[list[tuple[str, str]]]` @@ -2430,7 +2433,7 @@ Check PCIe Data for errors ### Description -Check cpu and kfd processes are within allowed maximum cpu and gpu usage +Check aggregate CPU usage against the configured maximum. **Bases**: ['DataAnalyzer'] @@ -2840,8 +2843,7 @@ Arguments for PCIe analyzer ### Annotations / fields -- **max_kfd_processes**: `int` — Maximum allowed number of KFD (Kernel Fusion Driver) processes; 0 disables the check. -- **max_cpu_usage**: `float` — Maximum allowed CPU usage (percent) for process checks. +- **max_cpu_usage**: `float` — Maximum allowed aggregate CPU execution percentage; idle and I/O-wait time are excluded. ## Analyzer Args Class RdmaAnalyzerArgs diff --git a/nodescraper/enums/eventcategory.py b/nodescraper/enums/eventcategory.py index 2c5d5514..bc45726c 100644 --- a/nodescraper/enums/eventcategory.py +++ b/nodescraper/enums/eventcategory.py @@ -41,6 +41,8 @@ class EventCategory(AutoNameStrEnum): - OS Generic Operating System events. Does not include specific events from OS which point to another category + - PROCESS_READ + Successful CPU process collection events. - PLATFORM Generic Platform Errors e.g. topo enumeration Platform-specific errors which do not fall under other categories (e.g. BMC, SMC, UBB) @@ -79,6 +81,7 @@ class EventCategory(AutoNameStrEnum): RAS = auto() IO = auto() OS = auto() + PROCESS_READ = auto() PLATFORM = auto() APPLICATION = auto() MEMORY = auto() diff --git a/nodescraper/plugins/inband/process/analyzer_args.py b/nodescraper/plugins/inband/process/analyzer_args.py index 39375ed9..658aedd2 100644 --- a/nodescraper/plugins/inband/process/analyzer_args.py +++ b/nodescraper/plugins/inband/process/analyzer_args.py @@ -31,13 +31,12 @@ class ProcessAnalyzerArgs(AnalyzerArgs): - max_kfd_processes: int = Field( - default=0, - description="Maximum allowed number of KFD (Kernel Fusion Driver) processes; 0 disables the check.", - ) max_cpu_usage: float = Field( default=20.0, - description="Maximum allowed CPU usage (percent) for process checks.", + description=( + "Maximum allowed aggregate CPU execution percentage; idle and I/O-wait time " + "are excluded." + ), ) @classmethod @@ -50,4 +49,4 @@ def build_from_model(cls, datamodel: ProcessDataModel) -> "ProcessAnalyzerArgs": Returns: ProcessAnalyzerArgs: instance of analyzer args class """ - return cls(max_kfd_processes=datamodel.kfd_process, max_cpu_usage=datamodel.cpu_usage) + return cls(max_cpu_usage=datamodel.cpu_usage) diff --git a/nodescraper/plugins/inband/process/collector_args.py b/nodescraper/plugins/inband/process/collector_args.py index 74099144..909d36a9 100644 --- a/nodescraper/plugins/inband/process/collector_args.py +++ b/nodescraper/plugins/inband/process/collector_args.py @@ -31,7 +31,10 @@ class ProcessCollectorArgs(CollectorArgs): top_n_process: int = Field( default=10, - description=( - "Number of top processes by CPU usage to collect " "(e.g. for top -b -n 1 -o %%CPU)." - ), + description="Number of top processes by CPU usage to collect.", + ) + sample_interval_seconds: float = Field( + default=1.0, + gt=0, + description="Wall-clock interval between procfs CPU samples.", ) diff --git a/nodescraper/plugins/inband/process/process_analyzer.py b/nodescraper/plugins/inband/process/process_analyzer.py index 8700f1c1..67f32646 100644 --- a/nodescraper/plugins/inband/process/process_analyzer.py +++ b/nodescraper/plugins/inband/process/process_analyzer.py @@ -34,7 +34,7 @@ class ProcessAnalyzer(DataAnalyzer[ProcessDataModel, ProcessAnalyzerArgs]): - """Check cpu and kfd processes are within allowed maximum cpu and gpu usage""" + """Check aggregate CPU usage against the configured maximum.""" DATA_MODEL = ProcessDataModel @@ -42,8 +42,7 @@ def analyze_data( self, data: ProcessDataModel, args: Optional[ProcessAnalyzerArgs] = None ) -> TaskResult: """ - Analyze the process data to check if the number of KFD processes and CPU usage - are within the allowed limits. + Analyze whether aggregate CPU usage is within the allowed limit. Args: data (ProcessDataModel): The process data to analyze. @@ -56,19 +55,6 @@ def analyze_data( args = ProcessAnalyzerArgs() has_errors = False - if data.kfd_process is not None and data.kfd_process > args.max_kfd_processes: - has_errors = True - self._log_event( - category=EventCategory.OS, - description=f"Kfd processes {data.kfd_process} exeed max limit {args.max_kfd_processes}", - data={ - "kfd_process": data.kfd_process, - "kfd_process_limit": args.max_kfd_processes, - }, - priority=EventPriority.CRITICAL, - console_log=True, - ) - if data.cpu_usage is not None and data.cpu_usage > args.max_cpu_usage: has_errors = True self._log_event( diff --git a/nodescraper/plugins/inband/process/process_collector.py b/nodescraper/plugins/inband/process/process_collector.py index 47dfc076..4655468b 100644 --- a/nodescraper/plugins/inband/process/process_collector.py +++ b/nodescraper/plugins/inband/process/process_collector.py @@ -23,7 +23,7 @@ # SOFTWARE. # ############################################################################### -import re +import time from typing import Optional from nodescraper.base import InBandDataCollector @@ -34,15 +34,226 @@ from .processdata import ProcessDataModel +def _parse_aggregate_cpu_from_proc_stat(proc_stat: str) -> Optional[tuple[int, int]]: + """Return total and idle-plus-I/O-wait jiffies from ``/proc/stat``. + + Guest fields are excluded from the total because Linux includes them in + the user and nice fields. + """ + for line in proc_stat.splitlines(): + if not line.startswith("cpu "): + continue + + parts = line.split() + if len(parts) < 6: + return None + try: + values = [int(value) for value in parts[1:]] + except ValueError: + return None + + # guest and guest_nice are already included in user and nice. + return sum(values[:8]), values[3] + values[4] + return None + + +def _global_non_idle_percent(total1: int, idle1: int, total2: int, idle2: int) -> float: + """Return aggregate non-idle CPU percentage between two samples.""" + total_delta = total2 - total1 + if total_delta <= 0: + return 0.0 + + idle_delta = idle2 - idle1 + percentage = 100.0 * (1.0 - idle_delta / total_delta) + return round(max(0.0, min(100.0, percentage)), 6) + + +def _parse_proc_pid_stat(stat_line: str) -> Optional[tuple[int, int]]: + """Return a process PID and its user-plus-system jiffies.""" + stat_line = stat_line.strip() + if not stat_line: + return None + + try: + paren_end = stat_line.rfind(") ") + if paren_end < 0: + return None + pid = int(stat_line[:paren_end].split(maxsplit=1)[0]) + fields = stat_line[paren_end + 2 :].split() + if len(fields) < 13: + return None + return pid, int(fields[11]) + int(fields[12]) + except ValueError: + return None + + +def _parse_proc_stat_dump(dump_stdout: str) -> tuple[dict[int, int], set[int]]: + """Parse bulk ``pid|stat`` output and sampler PIDs.""" + jiffies_by_pid: dict[int, int] = {} + sampler_pids: set[int] = set() + + for line in dump_stdout.splitlines(): + stripped = line.strip() + if stripped.startswith("__SAMPLER__:"): + try: + sampler_pids.add(int(stripped.split(":", 1)[1])) + except ValueError: + pass + continue + if "|" not in line: + continue + + pid_text, stat_line = line.split("|", 1) + try: + prefixed_pid = int(pid_text) + except ValueError: + continue + + parsed = _parse_proc_pid_stat(stat_line) + if parsed is None: + continue + stat_pid, jiffies = parsed + if stat_pid == prefixed_pid: + jiffies_by_pid[prefixed_pid] = jiffies + + return jiffies_by_pid, sampler_pids + + +def _top_process_cpu_shares( + sample1: dict[int, int], + sample2: dict[int, int], + total_delta: int, + top_n: int, + exclude_pids: set[int], +) -> list[tuple[int, float]]: + """Rank processes by their share of aggregate CPU jiffies.""" + if top_n <= 0: + return [] + + rows: list[tuple[int, float, int]] = [] + for pid in sample2: + if pid in exclude_pids: + continue + delta = max(0, sample2.get(pid, 0) - sample1.get(pid, 0)) + percentage = 100.0 * delta / total_delta if total_delta > 0 else 0.0 + rows.append((pid, percentage, delta)) + + rows.sort(key=lambda row: (row[2], row[1]), reverse=True) + return [(pid, percentage) for pid, percentage, _ in rows[:top_n]] + + +def _parse_comm_dump(dump_stdout: str) -> dict[int, str]: + """Parse ``pid:comm`` lines into process names keyed by PID.""" + processes: dict[int, str] = {} + for line in dump_stdout.splitlines(): + if ":" not in line: + continue + pid_text, command = line.split(":", 1) + try: + processes[int(pid_text)] = command.strip() + except ValueError: + continue + return processes + + +_STAT_DUMP_SHELL = ( + 'printf "__SAMPLER__:%s\\n" "$$"; ' + "for f in /proc/[0-9]*/stat; do " + '[ -r "$f" ] || continue; ' + 'pid="${f#/proc/}"; pid="${pid%/stat}"; ' + '[ "$pid" = "$$" ] && continue; ' + 'printf "%s|" "$pid"; cat "$f" 2>/dev/null || true; printf "\\n"; ' + "done" +) + + class ProcessCollector(InBandDataCollector[ProcessDataModel, ProcessCollectorArgs]): - """Collect Process details""" + """Collect aggregate CPU usage and top processes from Linux procfs.""" SUPPORTED_OS_FAMILY: set[OSFamily] = {OSFamily.LINUX} DATA_MODEL = ProcessDataModel - CMD_KFD = "rocm-smi --showpids" - CMD_CPU_USAGE = "top -b -n 1" - CMD_PROCESS = "top -b -n 1 -o %CPU " + CMD_PROC_STAT = "cat /proc/stat" + CMD_PROCESS_STAT = _STAT_DUMP_SHELL + CMD_PROCESS_NAMES = ( + "for p in {pids}; do " + 'printf "%s:" "$p"; ' + "cat /proc/$p/comm 2>/dev/null || true; " + 'printf "\\n"; ' + "done" + ) + + def _collect_procfs_cpu( + self, top_n_process: int, sample_interval_seconds: float + ) -> tuple[Optional[float], list[tuple[str, str]]]: + """Collect aggregate CPU usage and top process CPU shares.""" + stat1 = self._run_sut_cmd(self.CMD_PROC_STAT) + if stat1.exit_code != 0: + self.logger.warning( + "Unable to read first aggregate CPU sample (exit code %s)", stat1.exit_code + ) + return None, [] + dump1 = self._run_sut_cmd(self.CMD_PROCESS_STAT, log_artifact=False) + if dump1.exit_code != 0: + self.logger.warning( + "Unable to read first process CPU sample (exit code %s)", dump1.exit_code + ) + return None, [] + + time.sleep(sample_interval_seconds) + + stat2 = self._run_sut_cmd(self.CMD_PROC_STAT) + if stat2.exit_code != 0: + self.logger.warning( + "Unable to read second aggregate CPU sample (exit code %s)", stat2.exit_code + ) + return None, [] + dump2 = self._run_sut_cmd(self.CMD_PROCESS_STAT, log_artifact=False) + if dump2.exit_code != 0: + self.logger.warning( + "Unable to read second process CPU sample (exit code %s)", dump2.exit_code + ) + return None, [] + + aggregate1 = _parse_aggregate_cpu_from_proc_stat(stat1.stdout) + aggregate2 = _parse_aggregate_cpu_from_proc_stat(stat2.stdout) + if aggregate1 is None or aggregate2 is None: + sample = "first" if aggregate1 is None else "second" + output = stat1.stdout if aggregate1 is None else stat2.stdout + self._log_event( + category=EventCategory.OS, + description=f"Could not parse aggregate cpu line from /proc/stat ({sample} sample)", + data={"proc_stat_preview": output[:200]}, + priority=EventPriority.ERROR, + ) + return None, [] + + total1, idle1 = aggregate1 + total2, idle2 = aggregate2 + cpu_usage = _global_non_idle_percent(total1, idle1, total2, idle2) + total_delta = max(0, total2 - total1) + + sample1, excluded1 = _parse_proc_stat_dump(dump1.stdout) + sample2, excluded2 = _parse_proc_stat_dump(dump2.stdout) + top_processes = _top_process_cpu_shares( + sample1, + sample2, + total_delta, + top_n_process, + excluded1 | excluded2, + ) + + if not top_processes: + return cpu_usage, [] + + pids = " ".join(str(pid) for pid, _percentage in top_processes) + process_names = self._run_sut_cmd(self.CMD_PROCESS_NAMES.format(pids=pids)) + names = _parse_comm_dump(process_names.stdout) if process_names.exit_code == 0 else {} + processes = [ + (names.get(pid) or f"pid_{pid}", f"{percentage:.1f}") + for pid, percentage in top_processes + ] + return cpu_usage, processes def collect_data( self, args: Optional[ProcessCollectorArgs] = None @@ -59,57 +270,26 @@ def collect_data( args = ProcessCollectorArgs() process_data = ProcessDataModel() - process_data.processes = [] - - kfd_process = self._run_sut_cmd(self.CMD_KFD) - if kfd_process.exit_code == 0: - if "No KFD PIDs currently running" in kfd_process.stdout: - process_data.kfd_process = 0 - else: - kfd_process = re.findall( - r"^\s*\d+\s+[\w]+\s+\d+\s+\d+\s+\d+\s+\d+", - kfd_process.stdout, - re.MULTILINE, - ) - process_data.kfd_process = len(kfd_process) - - cpu_usage = self._run_sut_cmd(self.CMD_CPU_USAGE) - if cpu_usage.exit_code == 0: - cpu_idle = ( - [line for line in cpu_usage.stdout.splitlines() if "Cpu(s)" in line][0] - .split(",")[3] - .split()[0] - .replace("%id", "") - ) - process_data.cpu_usage = 100 - float(cpu_idle) - - last_line = args.top_n_process + 7 - processes = self._run_sut_cmd( - f"{self.CMD_PROCESS} | sed -n '8,{last_line}p'" - ) # Remove system header - if processes.exit_code == 0: - for line in processes.stdout.splitlines(): - columns = line.split() - process_cpu_usage = columns[8] - process_name = columns[11] - process_data.processes.append((process_name, process_cpu_usage)) - - process_check = bool(process_data.model_fields_set) - if process_check: + cpu_usage, processes = self._collect_procfs_cpu( + args.top_n_process, args.sample_interval_seconds + ) + if cpu_usage is not None: + process_data.cpu_usage = cpu_usage + process_data.processes = processes self._log_event( - category="PROCESS_READ", + category=EventCategory.PROCESS_READ, description="Process data collected", priority=EventPriority.INFO, ) self.result.message = "Process data collected" self.result.status = ExecutionStatus.OK return self.result, process_data - else: - self._log_event( - category=EventCategory.OS, - description="Process data not found", - priority=EventPriority.ERROR, - ) - self.result.message = "Process data not found" - self.result.status = ExecutionStatus.ERROR - return self.result, None + + self._log_event( + category=EventCategory.OS, + description="Process data not found", + priority=EventPriority.ERROR, + ) + self.result.message = "Process data not found" + self.result.status = ExecutionStatus.EXECUTION_FAILURE + return self.result, None diff --git a/nodescraper/plugins/inband/process/processdata.py b/nodescraper/plugins/inband/process/processdata.py index 683ad948..3073e250 100644 --- a/nodescraper/plugins/inband/process/processdata.py +++ b/nodescraper/plugins/inband/process/processdata.py @@ -25,10 +25,16 @@ ############################################################################### from typing import Optional +from pydantic import Field + from nodescraper.models import DataModel class ProcessDataModel(DataModel): - kfd_process: Optional[int] = None - cpu_usage: Optional[float] = None + """CPU execution usage (excluding idle and I/O-wait) and top processes from procfs.""" + + cpu_usage: Optional[float] = Field( + default=None, + description="Aggregate CPU execution percentage; idle and I/O-wait time are excluded.", + ) processes: Optional[list[tuple[str, str]]] = None diff --git a/test/functional/fixtures/process_plugin_config.json b/test/functional/fixtures/process_plugin_config.json index 752da786..ed7009b3 100644 --- a/test/functional/fixtures/process_plugin_config.json +++ b/test/functional/fixtures/process_plugin_config.json @@ -3,7 +3,6 @@ "plugins": { "ProcessPlugin": { "analysis_args": { - "max_kfd_processes": 0, "max_cpu_usage": 20.0 } } diff --git a/test/unit/plugin/test_analyzer_args_build_from_model.py b/test/unit/plugin/test_analyzer_args_build_from_model.py index e6eb7485..5702a481 100644 --- a/test/unit/plugin/test_analyzer_args_build_from_model.py +++ b/test/unit/plugin/test_analyzer_args_build_from_model.py @@ -185,11 +185,10 @@ def test_sysctl_analyzer_args_build_from_model(): def test_process_analyzer_args_build_from_model(): """Test ProcessAnalyzerArgs.build_from_model includes all fields""" - datamodel = ProcessDataModel(kfd_process=5, cpu_usage=15.5) + datamodel = ProcessDataModel(cpu_usage=15.5) args = ProcessAnalyzerArgs.build_from_model(datamodel) assert isinstance(args, ProcessAnalyzerArgs) - assert args.max_kfd_processes == 5 assert args.max_cpu_usage == 15.5 diff --git a/test/unit/plugin/test_process_analyzer.py b/test/unit/plugin/test_process_analyzer.py index e7bbb86c..f3efba9e 100644 --- a/test/unit/plugin/test_process_analyzer.py +++ b/test/unit/plugin/test_process_analyzer.py @@ -26,6 +26,7 @@ import copy import pytest +from pydantic import ValidationError from nodescraper.enums.eventcategory import EventCategory from nodescraper.enums.eventpriority import EventPriority @@ -38,7 +39,6 @@ @pytest.fixture def model_obj(): return ProcessDataModel( - kfd_process=0, cpu_usage=10, processes=[ ("top", "10.0"), @@ -52,7 +52,7 @@ def model_obj(): @pytest.fixture def config(): - return {"max_kfd_processes": 0, "max_cpu_usage": 40} + return {"max_cpu_usage": 40} @pytest.fixture @@ -60,10 +60,28 @@ def analyzer(system_info): return ProcessAnalyzer(system_info=system_info) +def test_process_data_model_ignores_removed_kfd_process_input(): + model = ProcessDataModel(kfd_process=1) + + assert "kfd_process" not in model.model_dump() + + +def test_process_analyzer_args_rejects_removed_kfd_limit(): + with pytest.raises(ValidationError): + ProcessAnalyzerArgs(max_kfd_processes=1) + + +def test_process_analyzer_checks_only_cpu_usage(analyzer): + data = ProcessDataModel(kfd_process=99, cpu_usage=10, processes=[]) + + result = analyzer.analyze_data(data, ProcessAnalyzerArgs(max_cpu_usage=40)) + + assert result.status == ExecutionStatus.OK + assert result.events == [] + + def test_nominal_with_config(analyzer, model_obj, config): - args = ProcessAnalyzerArgs( - max_kfd_processes=config["max_kfd_processes"], max_cpu_usage=config["max_cpu_usage"] - ) + args = ProcessAnalyzerArgs(max_cpu_usage=config["max_cpu_usage"]) result = analyzer.analyze_data(model_obj, args) assert result.status == ExecutionStatus.OK assert len(result.events) == 0 @@ -75,23 +93,9 @@ def test_nominal_no_config(analyzer, model_obj): assert len(result.events) == 0 -def test_error_kfd_process(analyzer, model_obj, config): - modified_model_obj = copy.deepcopy(model_obj) - modified_model_obj.kfd_process = 1 - args = ProcessAnalyzerArgs( - max_kfd_processes=config["max_kfd_processes"], max_cpu_usage=config["max_cpu_usage"] - ) - result = analyzer.analyze_data(modified_model_obj, args) - - assert result.status == ExecutionStatus.ERROR - for event in result.events: - assert event.category == EventCategory.OS.value - assert event.priority == EventPriority.CRITICAL - - def test_error_cpu_usage(analyzer, model_obj, config): modified_model_obj = copy.deepcopy(model_obj) - args = ProcessAnalyzerArgs(max_kfd_processes=config["max_kfd_processes"], max_cpu_usage=5) + args = ProcessAnalyzerArgs(max_cpu_usage=5) result = analyzer.analyze_data(modified_model_obj, args) assert result.status == ExecutionStatus.ERROR diff --git a/test/unit/plugin/test_process_collector.py b/test/unit/plugin/test_process_collector.py index cae8d6f0..beff7869 100644 --- a/test/unit/plugin/test_process_collector.py +++ b/test/unit/plugin/test_process_collector.py @@ -23,17 +23,38 @@ # SOFTWARE. # ############################################################################### +import logging +import time from unittest.mock import MagicMock import pytest +from pydantic import ValidationError +from nodescraper.enums.eventcategory import EventCategory from nodescraper.enums.executionstatus import ExecutionStatus from nodescraper.enums.systeminteraction import SystemInteractionLevel from nodescraper.interfaces.task import SystemCompatibilityError from nodescraper.models.systeminfo import OSFamily +from nodescraper.plugins.inband.process import ( + process_collector as process_collector_module, +) +from nodescraper.plugins.inband.process.collector_args import ProcessCollectorArgs from nodescraper.plugins.inband.process.process_collector import ProcessCollector from nodescraper.plugins.inband.process.processdata import ProcessDataModel +PROC_STAT_1 = "cpu 100 0 0 900 0 0 0 0 0 0\n" +PROC_STAT_2 = "cpu 200 0 0 1800 0 0 0 0 0 0\n" +PROC_DUMP_1 = ( + "__SAMPLER__:99999\n" + "1000|1000 (worker) S 0 0 0 0 -1 0 0 0 0 0 5000 6000\n" + "1|1 (systemd) S 0 0 0 0 -1 0 0 0 0 0 5000 6000\n" +) +PROC_DUMP_2 = ( + "__SAMPLER__:99999\n" + "1000|1000 (worker) S 0 0 0 0 -1 0 0 0 0 0 5100 6000\n" + "1|1 (systemd) S 0 0 0 0 -1 0 0 0 0 0 5000 6000\n" +) + @pytest.fixture def collector(system_info, conn_mock): @@ -44,37 +65,114 @@ def collector(system_info, conn_mock): ) -def test_run_linux(collector, conn_mock): - collector.system_info.os_family = OSFamily.LINUX - conn_mock.run_command.side_effect = [ - MagicMock( - exit_code=0, - stdout="PID PROCESS NAME GPU(s) VRAM USED SDMA USED CU OCCUPANCY\n8246 TransferBench 8 2267283456 0 0", - stderr="", - ), - MagicMock( - exit_code=0, - stdout="%Cpu(s): 0.1 us, 0.1 sy, 0.0 ni, 90.0 id", - stderr="", - ), - MagicMock( - exit_code=0, - stdout="356817 user 20 0 32112 14196 10556 R 10.0 0.0 0:00.07 top\n" - "1 root 20 0 166596 11916 8316 S 0.0 0.0 1:32.14 systemd", - stderr="", - ), - ] +def test_process_read_event_uses_shared_category(): + assert EventCategory.PROCESS_READ.value == "PROCESS_READ" + + +def test_parse_aggregate_cpu_from_proc_stat(): + proc_stat = "cpu0 1 2 3 4 5 6 7 8\ncpu 100 0 0 900 10 0 0 0 0 0\n" + + assert process_collector_module._parse_aggregate_cpu_from_proc_stat(proc_stat) == (1010, 910) + + +def test_parse_aggregate_cpu_excludes_guest_fields(): + proc_stat = "cpu 100 20 30 800 10 5 4 1 40 10\n" + + assert process_collector_module._parse_aggregate_cpu_from_proc_stat(proc_stat) == (970, 810) + + +def test_collector_args_reject_nonpositive_sample_interval(): + with pytest.raises(ValidationError): + ProcessCollectorArgs(sample_interval_seconds=0) + + +def test_global_non_idle_percent_uses_jiffy_deltas(): + assert process_collector_module._global_non_idle_percent(1000, 900, 2000, 1800) == 10.0 + + +def test_parse_proc_pid_stat_handles_process_names_with_spaces(): + stat_line = "1000 (worker process) S 0 0 0 0 -1 0 0 0 0 0 5000 6000" + + assert process_collector_module._parse_proc_pid_stat(stat_line) == (1000, 11000) + + +def test_parse_proc_pid_stat_handles_closing_parenthesis_in_process_name(): + stat_line = "1000 (worker) process) S 0 0 0 0 -1 0 0 0 0 0 5000 6000" + + assert process_collector_module._parse_proc_pid_stat(stat_line) == (1000, 11000) + + +def test_parse_proc_stat_dump_returns_jiffies_and_sampler_pid(): + dump = ( + "__SAMPLER__:99999\n" + "1000|1000 (worker process) S 0 0 0 0 -1 0 0 0 0 0 5000 6000\n" + "invalid line\n" + ) + + assert process_collector_module._parse_proc_stat_dump(dump) == ({1000: 11000}, {99999}) + + +def test_top_process_cpu_shares_ranks_deltas_and_excludes_sampler(): + first_sample = {1: 100, 2: 200, 99: 0} + second_sample = {1: 150, 2: 400, 99: 500} + + assert process_collector_module._top_process_cpu_shares( + first_sample, + second_sample, + total_delta=1000, + top_n=2, + exclude_pids={99}, + ) == [(2, 20.0), (1, 5.0)] + + +def test_top_process_cpu_shares_excludes_processes_missing_from_current_sample(): + assert process_collector_module._top_process_cpu_shares( + sample1={1: 100, 2: 200}, + sample2={2: 400}, + total_delta=1000, + top_n=2, + exclude_pids=set(), + ) == [(2, 20.0)] + + +def test_parse_comm_dump_maps_process_names_by_pid(): + assert process_collector_module._parse_comm_dump("1000:worker\n1:systemd\ninvalid\n") == { + 1000: "worker", + 1: "systemd", + } + + +def test_run_linux_collects_cpu_and_processes_from_procfs(collector, conn_mock, monkeypatch): + proc_stat_calls = 0 + proc_dump_calls = 0 + + def run_command(command, **_kwargs): + nonlocal proc_stat_calls, proc_dump_calls + if command == "cat /proc/stat": + proc_stat_calls += 1 + stdout = PROC_STAT_1 if proc_stat_calls == 1 else PROC_STAT_2 + elif "for f in /proc/" in command and "__SAMPLER__" in command: + proc_dump_calls += 1 + stdout = PROC_DUMP_1 if proc_dump_calls == 1 else PROC_DUMP_2 + elif "cat /proc/$p/comm" in command: + stdout = "1000:\n1:systemd\n" + else: + raise AssertionError(f"unexpected command: {command}") + return MagicMock(exit_code=0, stdout=stdout, stderr="", command=command) + + conn_mock.run_command.side_effect = run_command + monkeypatch.setattr(time, "sleep", lambda _seconds: None) + + result, data = collector.collect_data( + ProcessCollectorArgs(top_n_process=2, sample_interval_seconds=0.01) + ) - result, data = collector.collect_data() assert result.status == ExecutionStatus.OK assert data == ProcessDataModel( - kfd_process=1, - cpu_usage=10, - processes=[ - ("top", "10.0"), - ("systemd", "0.0"), - ], + cpu_usage=10.0, + processes=[("pid_1000", "10.0"), ("systemd", "0.0")], ) + assert all("__SAMPLER__" not in artifact.command for artifact in result.artifacts) def test_unsupported_platform(system_info, conn_mock): @@ -98,3 +196,49 @@ def test_exit_failure(collector, conn_mock): result, data = collector.collect_data() assert result.status == ExecutionStatus.EXECUTION_FAILURE assert data is None + + +@pytest.mark.parametrize( + ("failure_index", "expected_warning"), + [ + (0, "first aggregate CPU sample"), + (1, "first process CPU sample"), + (2, "second aggregate CPU sample"), + (3, "second process CPU sample"), + ], +) +def test_procfs_command_failure_logs_warning( + collector, conn_mock, monkeypatch, caplog, failure_index, expected_warning +): + responses = [ + MagicMock(exit_code=0, stdout=PROC_STAT_1, stderr=""), + MagicMock(exit_code=0, stdout=PROC_DUMP_1, stderr=""), + MagicMock(exit_code=0, stdout=PROC_STAT_2, stderr=""), + MagicMock(exit_code=0, stdout=PROC_DUMP_2, stderr=""), + ] + responses[failure_index].exit_code = 1 + conn_mock.run_command.side_effect = responses + monkeypatch.setattr(time, "sleep", lambda _seconds: None) + + with caplog.at_level(logging.WARNING): + result, data = collector.collect_data() + + assert result.status == ExecutionStatus.EXECUTION_FAILURE + assert data is None + assert expected_warning in caplog.text + + +def test_invalid_proc_stat_returns_failure_and_logs_os_event(collector, conn_mock, monkeypatch): + conn_mock.run_command.side_effect = [ + MagicMock(exit_code=0, stdout="not proc stat\n", stderr=""), + MagicMock(exit_code=0, stdout="__SAMPLER__:99999\n", stderr=""), + MagicMock(exit_code=0, stdout=PROC_STAT_2, stderr=""), + MagicMock(exit_code=0, stdout="__SAMPLER__:99999\n", stderr=""), + ] + monkeypatch.setattr(time, "sleep", lambda _seconds: None) + + result, data = collector.collect_data(ProcessCollectorArgs(sample_interval_seconds=0.01)) + + assert result.status == ExecutionStatus.EXECUTION_FAILURE + assert data is None + assert any(event.category == EventCategory.OS.value for event in result.events)