feat: [ESXi] CPU measurement using esxtop

mfd_host/feature/cpu/esxi.py

@@ -4,8 +4,11 @@
 
 import logging
 import re
+from time import sleep
+
+from mfd_common_libs import add_logging_level, log_levels, TimeoutCounter
+from mfd_connect.process import RemoteProcess
 
-from mfd_common_libs import add_logging_level, log_levels
 from mfd_host.exceptions import CPUFeatureExecutionError, CPUFeatureException
 from mfd_host.feature.cpu.base import BaseFeatureCPU
 from mfd_network_adapter.data_structures import State
@@ -62,3 +65,73 @@ def set_numa_affinity(self, numa_state: State) -> None:
             f'esxcli system settings advanced set {value} -o "/Numa/LocalityWeightActionAffinity"',
             custom_exception=CPUFeatureExecutionError,
         )
+
+    def start_cpu_measurement(self) -> RemoteProcess:
+        """
+        Start CPU measurement on host.
+
+        :return: Handle to process
+        """
+        logger.log(level=log_levels.MODULE_DEBUG, msg="Start CPU measurement on host")
+        return self._connection.start_process(command="esxtop -b -n 8 -d3", log_file=True, shell=True)
+
+    def stop_cpu_measurement(self, process: RemoteProcess, process_name: str) -> int:
+        """
+        Stop CPU measurement process.
+
+        :param process: Process handle
+        :param process_name: process name to filter CPU usage
+        :return: Average CPU usage percentage
+        """
+        logger.log(level=log_levels.MODULE_DEBUG, msg="Stop CPU measurement on host")
+        if process.running:
+            process.stop()
+            timeout = TimeoutCounter(5)
+            while not timeout:
+                if not process.running:
+                    break
+                sleep(1)
+            else:
+                process.kill()
+                timeout = TimeoutCounter(5)
+                while not timeout:
+                    if not process.running:
+                        break
+                    sleep(1)
+                if process.running:
+                    raise RuntimeError("CPU measurement process is still running after stop and kill.")
+
+        return self.parse_cpu_usage(process_name=process_name, process=process)
+
+    def parse_cpu_usage(self, process_name: str, process: RemoteProcess) -> int:
+        """
+        Parse CPU usage from esxtop output.
+
+        :param process_name: process_name name to filter CPU usage
+        :param process: Process handle
+        :return: average CPU usage percentage
+        """
+        parsed_file_path = "/tmp/parsed_output.txt"
+        command = (
+            f"cut -d, -f`awk -F, '{{for (i=1;i<=NF;i++){{if ($i ~/Group Cpu.*{process_name}).*Used/) "
+            f"{{print i}}}}}}' {process.log_path}` {process.log_path}>{parsed_file_path}"
+        )
+        self._connection.execute_command(command=command, shell=True)
+        p = self._connection.path(process.log_path)
+        p.unlink()
+        try:
+            parsed_file_path = self._connection.path(parsed_file_path)
+            file_content = parsed_file_path.read_text()
+            cpu_list = []
+            for line in file_content.splitlines()[1:]:
+                try:
+                    cpu_list.append(float(line.strip('"')))
+                except ValueError:
+                    continue
+
+        except Exception as e:
+            raise RuntimeError(f"Failed to read parsed CPU usage output file due to - {e}.")
+
+        p = self._connection.path(parsed_file_path)
+        p.unlink()
+        return round(sum(cpu_list) / len(cpu_list)) if cpu_list else 0

tests/unit/test_mfd_host/test_feature/test_cpu/test_esxi.py

@@ -3,13 +3,16 @@
 from textwrap import dedent
 
 import pytest
+from mfd_common_libs import log_levels
 from mfd_connect import SSHConnection
 from mfd_connect.base import ConnectionCompletedProcess
 from mfd_host import Host
 from mfd_host.exceptions import CPUFeatureExecutionError, CPUFeatureException
 from mfd_network_adapter.data_structures import State
 from mfd_typing import OSName
 
+from mfd_host.feature.cpu import ESXiCPU
+
 
 class TestESXiCPU:
     cpu_attribute_output = dedent(
@@ -78,3 +81,85 @@ def test_set_numa_affinity_disabled(self, host, mocker):
         cmd = 'esxcli system settings advanced set -i 0 -o "/Numa/LocalityWeightActionAffinity"'
         assert host.cpu.set_numa_affinity(State.DISABLED) is None
         host.connection.execute_command.assert_called_once_with(cmd, custom_exception=CPUFeatureExecutionError)
+
+    def test_start_cpu_measurement_returns_process_handle(self, host, mocker):
+        mock_connection = mocker.Mock()
+        mock_process = mocker.Mock()
+        mock_connection.start_process.return_value = mock_process
+        host.cpu._connection = mock_connection
+
+        result = host.cpu.start_cpu_measurement()
+
+        assert result == mock_process
+        mock_connection.start_process.assert_called_once_with(command="esxtop -b -n 8 -d3", log_file=True, shell=True)
+
+    def test_logs_debug_message_when_starting_cpu_measurement(self, host, mocker):
+        mock_logger = mocker.patch("mfd_host.feature.cpu.esxi.logger")
+
+        host.cpu.start_cpu_measurement()
+
+        mock_logger.log.assert_called_once_with(level=log_levels.MODULE_DEBUG, msg="Start CPU measurement on host")
+
+    def test_parses_average_cpu_usage(self, host, mocker):
+        mock_connection = mocker.MagicMock()
+        mock_process = mocker.MagicMock()
+        mock_process.log_path = "/tmp/esxtop.log"
+        mock_connection.path.return_value.read_text.return_value = 'Header\n"10"\n"20"\n"30"\n"40"\n"50"\n'
+        mock_connection.path.return_value.__enter__.return_value = mock_connection.path.return_value
+        host.cpu._connection = mock_connection
+
+        result = host.cpu.parse_cpu_usage("vm1", mock_process)
+        assert result == 30  # (10+20+30+40+50)//5
+
+    def test_returns_zero_when_no_cpu_samples_found(self, host, mocker):
+        mock_connection = mocker.MagicMock()
+        mock_process = mocker.MagicMock()
+        mock_process.log_path = "/tmp/esxtop.log"
+        mock_connection.path.return_value.read_text.return_value = "Header\n"
+        mock_connection.path.return_value.__enter__.return_value = mock_connection.path.return_value
+        host.cpu._connection = mock_connection
+
+        result = host.cpu.parse_cpu_usage("vm1", mock_process)
+        assert result == 0
+
+    def test_skips_lines_that_cannot_be_converted_to_float(self, host, mocker):
+        mock_connection = mocker.MagicMock()
+        mock_process = mocker.MagicMock()
+        mock_process.log_path = "/tmp/esxtop.log"
+        mock_connection.path.return_value.read_text.return_value = 'Header\n"10"\n"invalid"\n"30"\n"40"\n"50"\n'
+        mock_connection.path.return_value.__enter__.return_value = mock_connection.path.return_value
+        host.cpu._connection = mock_connection
+
+        result = host.cpu.parse_cpu_usage("vm1", mock_process)
+        assert result == 32  # (10+30+40+50)//4
+
+    def test_raises_runtime_error_when_file_read_fails(self, host, mocker):
+        mock_connection = mocker.MagicMock()
+        mock_process = mocker.MagicMock()
+        mock_process.log_path = "/tmp/esxtop.log"
+        mock_connection.path.return_value.read_text.side_effect = Exception("File error")
+        mock_connection.path.return_value.__enter__.return_value = mock_connection.path.return_value
+        host.cpu._connection = mock_connection
+
+        with pytest.raises(
+            RuntimeError,
+            match="Failed to read parsed CPU usage output file due to - File error.",
+        ):
+            host.cpu.parse_cpu_usage("vm1", mock_process)
+
+    def test_parses_cpu_usage_when_process_not_running(self, host, mocker):
+        mock_process = mocker.Mock()
+        mock_process.running = False
+        mock_parse = mocker.patch.object(ESXiCPU, "parse_cpu_usage", return_value=7)
+        host.cpu._connection = mocker.Mock()
+        result = host.cpu.stop_cpu_measurement(mock_process, "vm4")
+        assert result == 7
+        mock_parse.assert_called_once_with(process_name="vm4", process=mock_process)
+
+    def test_stop_fail_when_process_running(self, host, mocker):
+        mock_process = mocker.Mock()
+        mock_process.running = True
+        mocker.patch("mfd_host.feature.cpu.esxi.TimeoutCounter", autospec=True, spec_set=True)
+        host.cpu._connection = mocker.Mock()
+        with pytest.raises(RuntimeError, match="CPU measurement process is still running after stop and kill."):
+            host.cpu.stop_cpu_measurement(mock_process, "vm4")