From 3d51934d7d7c48ebbec09bb657631405cb821225 Mon Sep 17 00:00:00 2001 From: dteo00 <173225270+dteo00@users.noreply.github.com> Date: Fri, 19 Jun 2026 14:27:13 +0800 Subject: [PATCH] Make resource monitor host metrics extensible; add host_ctx_switches Co-Authored-By: Claude Opus 4.8 (1M context) --- .../3877_new.resource_monitor_metrics.rst | 1 + testplan/monitor/resource.py | 143 ++++++++++-------- .../testplan/test_resource_monitor.py | 73 +++++++-- tests/unit/testplan/common/utils/test_path.py | 1 - 4 files changed, 137 insertions(+), 81 deletions(-) create mode 100644 doc/newsfragments/3877_new.resource_monitor_metrics.rst diff --git a/doc/newsfragments/3877_new.resource_monitor_metrics.rst b/doc/newsfragments/3877_new.resource_monitor_metrics.rst new file mode 100644 index 000000000..ba7b526e1 --- /dev/null +++ b/doc/newsfragments/3877_new.resource_monitor_metrics.rst @@ -0,0 +1 @@ +Resource monitor now reports the host context-switch rate (``host_ctx_switches``). diff --git a/testplan/monitor/resource.py b/testplan/monitor/resource.py index fe75f339e..f17496492 100755 --- a/testplan/monitor/resource.py +++ b/testplan/monitor/resource.py @@ -45,32 +45,53 @@ class HostResourceData(ResourceData): disk_used: int system_load: float + host_ctx_switches: float + + def to_row(self, timestamp: float) -> "HostResourceRow": + return HostResourceRow( + time=timestamp, + cpu=self.cpu_usage, + memory=self.memory_used, + disk=self.disk_used, + iops=self.iops, + io_read=self.io_read, + io_write=self.io_write, + system_load=self.system_load, + host_ctx_switches=self.host_ctx_switches, + ) @dataclasses.dataclass -class ProcessResourceData(ResourceData): +class HostResourceRow: """ - Process level resource data + CSV file row structure of host level resource. """ - name: str - cmdline: str - pid: int + time: float + cpu: float + memory: int + disk: int + iops: float + io_read: float + io_write: float + system_load: float + host_ctx_switches: float + def as_dict(self) -> Dict[str, float]: + return { + f.name: getattr(self, f.name) for f in dataclasses.fields(self) + } -class HostResourceRow(NamedTuple): + +@dataclasses.dataclass +class ProcessResourceData(ResourceData): """ - CSV file row structure of host level resource + Process level resource data """ - timestamp: float - cpu_usage: float - memory_used: int - disk_used: int - iops: float - io_read: float - io_write: float - system_load: float + name: str + cmdline: str + pid: int class ProcessResourceRow(NamedTuple): @@ -154,6 +175,14 @@ def send_metadata(self) -> None: def collect_cpu_usage(self) -> float: return psutil.cpu_percent(0.1) # type: ignore[no-any-return] + def make_host_resource(self, **base: Any) -> HostResourceData: + """ + Build the host resource record from the always-collected base fields. + Subclasses override this to return a ``HostResourceData`` subclass that + carries extra typed metrics. + """ + return HostResourceData(**base) + def collect_memory_usage(self) -> int: return self.memory_size - psutil.virtual_memory().available # type: ignore[no-any-return] @@ -167,6 +196,7 @@ def collect_host_data(self) -> None: iops = _disk_io.read_count + _disk_io.write_count io_read = _disk_io.read_bytes io_write = _disk_io.write_bytes + ctx_switches = psutil.cpu_stats().ctx_switches if self.last_host_io_info: iops = self._ensure_positive( (iops - self.last_host_io_info["iops"]) / self.poll_interval @@ -179,7 +209,11 @@ def collect_host_data(self) -> None: (io_write - self.last_host_io_info["io_write"]) / self.poll_interval ) - host_resource = HostResourceData( + host_ctx_switches = self._ensure_positive( + (ctx_switches - self.last_host_io_info["ctx_switches"]) + / self.poll_interval + ) + host_resource = self.make_host_resource( cpu_usage=self.collect_cpu_usage(), memory_used=self.collect_memory_usage(), disk_used=psutil.disk_usage(self.disk_path).used, @@ -187,6 +221,7 @@ def collect_host_data(self) -> None: io_read=io_read, io_write=io_write, system_load=psutil.getloadavg()[0], + host_ctx_switches=host_ctx_switches, ) self.last_host_resource = host_resource @@ -194,6 +229,7 @@ def collect_host_data(self) -> None: "iops": _disk_io.read_count + _disk_io.write_count, "io_read": _disk_io.read_bytes, "io_write": _disk_io.write_bytes, + "ctx_switches": ctx_switches, } def collect_process_data(self) -> None: @@ -367,25 +403,20 @@ async def handle_request(self, msg: bytes) -> None: f.write(json_dumps(message.data)) elif message.cmd == communication.Message.Message: self.logger.info("Received resource data from %s.", client_id) + client_host_data: HostResourceData = message.data["host_resource"] + row = client_host_data.to_row(message.data["time"]).as_dict() if client_id not in self._file_handler: self._file_handler[client_id] = {} self._file_handler[client_id]["host_file"] = open( - self.file_directory / f"{slugify(client_id)}.csv", "w" + self.file_directory / f"{slugify(client_id)}.csv", + "w", + newline="", ) - self._file_handler[client_id]["host_csv"] = csv.writer( - self._file_handler[client_id]["host_file"] + self._file_handler[client_id]["host_csv"] = csv.DictWriter( + self._file_handler[client_id]["host_file"], + fieldnames=list(row.keys()), ) - client_host_data: HostResourceData = message.data["host_resource"] - row = HostResourceRow( - message.data["time"], - client_host_data.cpu_usage, - client_host_data.memory_used, - client_host_data.disk_used, - client_host_data.iops, - client_host_data.io_read, - client_host_data.io_write, - client_host_data.system_load, - ) + self._file_handler[client_id]["host_csv"].writeheader() self._file_handler[client_id]["host_csv"].writerow(row) self._file_handler[client_id]["host_file"].flush() if self.detailed: @@ -457,51 +488,29 @@ def normalize_data(self, client_id: str) -> Optional[Dict[str, Any]]: client_host_path = ( self.file_directory / f"{slugify(client_id)}.csv" ) - resource_data: Dict[str, List[Any]] = { - "time": [], - "cpu": [], - "memory": [], - "disk": [], - "iops": [], - "io_read": [], - "io_write": [], - "system_load": [], - } - with open(client_host_path) as client_file: - reader = csv.reader(client_file) + with open(client_host_path, newline="") as client_file: + reader = csv.DictReader(client_file) + resource_data: Dict[str, List[Any]] = { + name: [] for name in (reader.fieldnames or []) + } for row in reader: - host_resource = HostResourceRow(*row) # type: ignore[arg-type] - resource_data["time"].append( - float(host_resource.timestamp) - ) - resource_data["cpu"].append(float(host_resource.cpu_usage)) - resource_data["memory"].append( - float(host_resource.memory_used) - ) - resource_data["disk"].append(int(host_resource.disk_used)) - resource_data["iops"].append(float(host_resource.iops)) - resource_data["io_read"].append( - float(host_resource.io_read) - ) - resource_data["io_write"].append( - float(host_resource.io_write) - ) - resource_data["system_load"].append( - float(host_resource.system_load) - ) + for name, value in row.items(): + resource_data[name].append(float(value)) + if not resource_data.get("time"): + return None json_file_path = self.file_directory / f"{slugify(client_id)}.json" with open(json_file_path, "w") as json_file: json_file.write(json_dumps(resource_data)) - return { + # Summarize every series as max_ + summary: Dict[str, Any] = { "resource_file": str(json_file_path.resolve()), "time_start": min(resource_data["time"]), "time_end": max(resource_data["time"]), - "max_cpu": max(resource_data["cpu"]), - "max_memory": max(resource_data["memory"]), - "max_disk": max(resource_data["disk"]), - "max_iops": max(resource_data["iops"]), - "max_system_load": max(resource_data["system_load"]), } + for name, values in resource_data.items(): + if name != "time" and values: + summary[f"max_{name}"] = max(values) + return summary except FileNotFoundError: return None diff --git a/tests/functional/testplan/test_resource_monitor.py b/tests/functional/testplan/test_resource_monitor.py index 00cff9742..da6952414 100755 --- a/tests/functional/testplan/test_resource_monitor.py +++ b/tests/functional/testplan/test_resource_monitor.py @@ -1,4 +1,5 @@ import csv +import dataclasses import json import os import socket @@ -10,12 +11,36 @@ from testplan.monitor.resource import ( ResourceMonitorServer, ResourceMonitorClient, + HostResourceData, + HostResourceRow, ProcessResourceRow, ) from testplan.common.utils.strings import slugify from testplan.common.utils.timing import wait +@dataclasses.dataclass +class _ProbeHostResourceRow(HostResourceRow): + probe: float = 0.0 + + +@dataclasses.dataclass +class _ProbeHostResourceData(HostResourceData): + probe: float = 0.0 + + def to_row(self, timestamp: float) -> _ProbeHostResourceRow: + return _ProbeHostResourceRow( + **dataclasses.asdict(super().to_row(timestamp)), probe=self.probe + ) + + +class _ProbeResourceMonitorClient(ResourceMonitorClient): + """Injects an extra typed host metric to exercise the make_host_resource hook.""" + + def make_host_resource(self, **base: typing.Any) -> HostResourceData: + return _ProbeHostResourceData(**base, probe=1.0) + + @skip_on_windows(reason="Resource monitor is skipped on Windows.") def test_resource(runpath): print(f"Runpath: {runpath}") @@ -27,7 +52,7 @@ def test_resource(runpath): assert server.address != "" assert server.file_directory.exists() - client = ResourceMonitorClient(server_address=server.address) + client = _ProbeResourceMonitorClient(server_address=server.address) assert client.uid assert client.disk_size client.poll_interval = 5 @@ -60,26 +85,29 @@ def get_latest_pid_resource_data( detail = _line return detail + def get_host_rows() -> typing.List[dict]: + # Host CSV is headered, so read by column name via DictReader. + with open(resource_file_path, newline="") as resource_file: + return list(csv.DictReader(resource_file)) + def received_resource_data() -> bool: if resource_file_path.exists() and resource_detail_file_path.exists(): - with open(resource_file_path) as resource_file: - csv_reader = csv.reader(resource_file) - _current_pid_data = get_latest_pid_resource_data(current_pid) - return ( - len(list(csv_reader)) > 0 and _current_pid_data is not None - ) + _current_pid_data = get_latest_pid_resource_data(current_pid) + return len(get_host_rows()) > 0 and _current_pid_data is not None return False wait(received_resource_data, timeout=client.poll_interval * 2) - def get_last_resource_data(): - with open(resource_file_path) as resource_file: - csv_reader = csv.reader(resource_file) - return list(csv_reader)[-1] - - memory_used = int(get_last_resource_data()[2]) + last_host_row = get_host_rows()[-1] + memory_used = int(float(last_host_row["memory"])) print(f"Current memory used: {memory_used}") + # Core host context-switch rate is collected for every client. + assert float(last_host_row["host_ctx_switches"]) >= 0 + # Extra typed metric from the HostResourceRow subclass flows through as its + # own column. + assert float(last_host_row["probe"]) == 1.0 + current_pid_data = get_latest_pid_resource_data(current_pid) print(f"Current PID Data: {current_pid_data}") @@ -99,5 +127,24 @@ def check_pid_memory(): big_memory.clear() # avoid freeing memory early + meta_path = server.dump() + with open(meta_path) as meta_file: + entries = json.load(meta_file)["entries"] + assert len(entries) == 1 + entry = entries[0] + assert entry["time_start"] > 0 + assert entry["time_end"] >= entry["time_start"] + assert entry["max_cpu"] >= 0 + assert entry["max_memory"] >= 0 + assert entry["max_disk"] >= 0 + assert entry["max_iops"] >= 0 + assert entry["max_system_load"] >= 0 + assert entry["max_host_ctx_switches"] >= 0 + assert entry["max_probe"] == 1.0 + with open(entry["resource_file"]) as resource_file: + resource_arrays = json.load(resource_file) + assert resource_arrays["host_ctx_switches"] + assert resource_arrays["probe"] + client.stop() server.stop() diff --git a/tests/unit/testplan/common/utils/test_path.py b/tests/unit/testplan/common/utils/test_path.py index a017d8206..c96740bfb 100644 --- a/tests/unit/testplan/common/utils/test_path.py +++ b/tests/unit/testplan/common/utils/test_path.py @@ -38,7 +38,6 @@ def test_hashfile(tmpdir): def test_change_directory_thread_safe(tmpdir): - barrier = threading.Barrier(2) def racing_worker():