Make resource monitor host metrics extensible; add host_ctx_switches

doc/newsfragments/3877_new.resource_monitor_metrics.rst

@@ -0,0 +1 @@
+Resource monitor now reports the host context-switch rate (``host_ctx_switches``).

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,21 +209,27 @@ 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,
                 iops=iops,
                 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
 
         self.last_host_io_info = {
             "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_<column>
+            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
 

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

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