CVS-177198 LLM Memory Usage tests

tests/functional/object_model/ovms_instance.py

@@ -41,7 +41,7 @@
 from tests.functional.utils.core import get_children_from_module
 from tests.functional.utils.inference.communication import GRPC, REST
 from tests.functional.utils.logger import get_logger
-from tests.functional.constants.os_type import OsType
+from tests.functional.constants.os_type import OsType, get_host_os
 from tests.functional.utils.port_manager import PortManager
 from tests.functional.utils.process import Process
 from tests.functional.utils.test_framework import change_dir_permissions, is_single_threaded
@@ -66,7 +66,7 @@
 from tests.functional.object_model.mediapipe_calculators import MediaPipeCalculator
 from tests.functional.object_model.ovms_config import OvmsConfig
 from tests.functional.object_model.package_manager import PackageManager
-from tests.functional.object_model.resource_monitor import DockerResourceMonitor
+from tests.functional.object_model.resource_monitor import DockerResourceMonitor, WindowsResourceMonitor
 from tests.functional.object_model.test_environment import TestEnvironment
 
 logger = get_logger(__name__)
@@ -539,7 +539,13 @@ def attach_context(self, context):
     def attach_resource_monitor(self, context, start=True):
         if hasattr(self.ovms, "container"):
             self.resource_monitor = DockerResourceMonitor(self.ovms.container)
-            if start:
-                self.resource_monitor.start()
-            context.test_objects.append(self.resource_monitor)
-            return self.resource_monitor
+        elif get_host_os() == OsType.Windows or getattr(context, "base_os", None) == OsType.Windows:
+            ovms_pid = self.ovms._dmesg_log.ovms_pid
+            assert ovms_pid is not None, "Cannot attach Windows resource monitor: ovms_pid is not available"
+            self.resource_monitor = WindowsResourceMonitor(ovms_pid)
+        else:
+            return None
+        if start:
+            self.resource_monitor.start()
+        context.test_objects.append(self.resource_monitor)
+        return self.resource_monitor

tests/functional/object_model/ovms_log_monitor.py

@@ -15,6 +15,7 @@
 #
 
 import datetime
+import os
 import re
 import time
 
@@ -362,6 +363,7 @@ class BinaryOvmsLogMonitor(OvmsLogMonitor):
     def __init__(self, ovms_process, **kwargs):
         super().__init__(**kwargs)
         self._proc = ovms_process
+        self._mirror_offset = 0
 
     def is_ovms_running(self):
         status = get_pid_status(self._proc._proc.pid)
@@ -371,13 +373,30 @@ def is_ovms_running(self):
             return True
         return True
 
+    def _read_mirror_tail(self, mirror_path):
+        with open(mirror_path, "rb") as fd:
+            fd.seek(self._mirror_offset)
+            chunk = fd.read()
+        # A trailing partial line would be torn in half between two reads.
+        complete, newline, _partial = chunk.rpartition(b"\n")
+        if not newline:
+            return []
+        self._mirror_offset += len(complete) + len(newline)
+        return complete.decode("utf-8", "ignore").splitlines()
+
     def get_all_logs(self):
         stdout, stderr = self._proc.get_output()
         if stderr:
             logger.error(
                 f"Detect non-empty stderr! It is recommended to redirect stderr to stdout: 2>&1. STDERR: {stderr}"
             )
-        self._read_lines += stdout.splitlines()
+        mirror_path = self._proc.get_stdout_mirror_path()
+        if mirror_path and os.path.exists(mirror_path):
+            # pop() drains the queue shared by every monitor of this process, so the lines one
+            # monitor takes would be missing from the others and from the saved log.
+            self._read_lines += self._read_mirror_tail(mirror_path)
+        else:
+            self._read_lines += stdout.splitlines()
         return self._read_lines
 
 

tests/functional/object_model/resource_monitor.py

@@ -17,9 +17,11 @@
 import csv
 import threading
 from abc import ABC, abstractmethod
+from datetime import datetime
 from pathlib import Path
 
 import numpy as np
+import psutil
 from dateutil import parser
 
 from tests.functional.utils.logger import get_logger
@@ -32,6 +34,7 @@ class ResourceMonitor(threading.Thread, ABC):
     def __init__(self):
         threading.Thread.__init__(self)
         self._stop_event = threading.Event()
+        self.stop_reason = None
 
     def stop(self):
         self._stop_event.set()
@@ -42,6 +45,7 @@ def run(self):
             try:
                 self.check_resources()
             except StopIteration as e:
+                self.stop_reason = str(e)
                 self._stop_event.set()
                 break
         self.save_data()
@@ -55,13 +59,33 @@ def save_data(self):
         pass
 
 
+def _cgroup_cache_bytes(cgroup_memory_stats):
+    if "cache" in cgroup_memory_stats:
+        return float(cgroup_memory_stats.get("cache", 0))
+    return float(cgroup_memory_stats.get("file", 0))
+
+
 class DockerResourceMonitor(ResourceMonitor):
     MEMORY_USAGE = "MEMORY_USAGE"
-    FIELDS = ["DATE", "PIDS_COUNT", MEMORY_USAGE]  # + ["CPU_USAGE"] # Enable in further releases
+    PRIVATE_MEMORY = "PRIVATE_MEMORY"
+    MEMORY_CACHE = "MEMORY_CACHE"
+    FIELDS = ["DATE", "PIDS_COUNT", MEMORY_USAGE, PRIVATE_MEMORY, MEMORY_CACHE]  # + ["CPU_USAGE"] # Enable in further releases
+    VALIDATED_FIELDS = [MEMORY_USAGE, PRIVATE_MEMORY]
+    LOGGED_MEMORY_FIELDS = [MEMORY_CACHE]
+    COUNTER_FIELDS = ["PIDS_COUNT"]
+    LOGGED_FIELDS = LOGGED_MEMORY_FIELDS + COUNTER_FIELDS
     FIELDS_TO_STATS = {
         "DATE": lambda x: x["read"],
         "PIDS_COUNT": lambda x: int(x["pids_stats"].get("current", "0")),
         MEMORY_USAGE: lambda x: "{:.2f}M".format(float(x["memory_stats"].get("usage", "0.0")) / (2**20)),
+        PRIVATE_MEMORY: lambda x: "{:.2f}M".format(
+            float(x["memory_stats"].get("stats", {}).get(
+                "anon", x["memory_stats"].get("stats", {}).get("rss", 0)
+            )) / (2**20)
+        ),
+        MEMORY_CACHE: lambda x: "{:.2f}M".format(
+            _cgroup_cache_bytes(x["memory_stats"].get("stats", {})) / (2**20)
+        ),
         # Enable after debug & fixing
         # "CPU_USAGE": lambda x:
         #     [cpu / x['cpu_stats']['cpu_usage']['total_usage'] for cpu in x['cpu_stats']['cpu_usage']['percpu_usage']],
@@ -140,9 +164,142 @@ def _get_resource_data(self):
         'networks' = {dict: 1} {'eth0': {'rx_bytes': 90, 'rx_packets': 1, 'rx_errors': 0, ...
         """
         stats = self.container.stats(stream=False, decode=False)
+        # Prevent reading zeroes from stopped container
+        pids = int(stats.get("pids_stats", {}).get("current", 0) or 0)
+        usage = float(stats.get("memory_stats", {}).get("usage", 0) or 0)
+        if pids <= 0 or usage <= 0:
+            reason = (
+                f"container {self.container.name} reports no live process "
+                f"(pids_stats.current={pids}, memory_stats.usage={usage} bytes) - the container "
+                f"stopped, crashed or was removed while the test was still sampling"
+            )
+            logger.warning(f"Stopping Docker resource monitor: {reason}")
+            raise StopIteration(reason)
         return stats
 
     def get_stats_by_field(self, field):
         result = self._get_resource_data()
         self._docker_stats_data_raw.append(result)
         return self.get_field_data(field, result)
+
+    def sample_all(self):
+        """Read one stats snapshot and return all tracked metrics as floats (MB / counts).
+
+        A single snapshot keeps every metric in the returned sample mutually
+        consistent (same instant) and avoids one docker stats call per metric.
+        """
+        stats = self._get_resource_data()
+        self._docker_stats_data_raw.append(stats)
+        return {
+            field: float(str(self.get_field_data(field, stats)).replace("M", ""))
+            for field in self.get_validated_metric_names() + self.get_logged_metric_names()
+        }
+
+    @classmethod
+    def get_validated_metric_names(cls):
+        return cls.VALIDATED_FIELDS
+
+    @classmethod
+    def get_logged_metric_names(cls):
+        return cls.LOGGED_FIELDS
+
+    @classmethod
+    def get_memory_metric_names(cls):
+        return cls.VALIDATED_FIELDS + cls.LOGGED_MEMORY_FIELDS
+
+    @classmethod
+    def get_counter_metric_names(cls):
+        return cls.COUNTER_FIELDS
+
+
+class WindowsResourceMonitor(ResourceMonitor):
+    WORKING_SET_SIZE = "WORKING_SET_SIZE"
+    PRIVATE_BYTES = "PRIVATE_BYTES"
+    PAGE_FAULTS = "PAGE_FAULTS"
+
+    MEMORY_USAGE = WORKING_SET_SIZE
+
+    FIELDS = ["DATE", WORKING_SET_SIZE, PRIVATE_BYTES, PAGE_FAULTS]
+    VALIDATED_FIELDS = [PRIVATE_BYTES]
+    LOGGED_MEMORY_FIELDS = [WORKING_SET_SIZE]
+    COUNTER_FIELDS = [PAGE_FAULTS]
+    LOGGED_FIELDS = LOGGED_MEMORY_FIELDS + COUNTER_FIELDS
+    SAMPLE_INTERVAL_SEC = 1.0
+    # Optional callback invoked after save_data with (log_path).
+    on_data_saved = None
+
+    def __init__(self, ovms_pid):
+        super().__init__()
+        self.ovms_pid = ovms_pid
+        self._process = psutil.Process(int(ovms_pid))
+        self._stats_data_raw = []
+
+    def cleanup(self):
+        if not self._stop_event.is_set():
+            if self.is_alive():
+                self.stop()
+            self.save_data()
+
+    def _get_resource_data(self):
+        stats = {"DATE": datetime.now().isoformat()}
+        try:
+            info = self._process.memory_info()
+        except psutil.Error as error:
+            reason = (f"OVMS process pid={self.ovms_pid} is no longer readable ({error}) - it "
+                      f"exited, crashed or was killed while the test was still sampling")
+            logger.warning(f"Stopping Windows resource monitor: {reason}")
+            raise StopIteration(reason) from error
+        # wset/private are the Win32 counters .NET exposes as WorkingSet64/PrivateMemorySize64.
+        stats[self.WORKING_SET_SIZE] = float(info.wset) / (1024 * 1024)
+        stats[self.PRIVATE_BYTES] = float(info.private) / (1024 * 1024)
+        stats[self.PAGE_FAULTS] = int(info.num_page_faults)
+        return stats
+
+    def check_resources(self):
+        result = self._get_resource_data()
+        self._stats_data_raw.append(result)
+        self._stop_event.wait(self.SAMPLE_INTERVAL_SEC)
+
+    def save_data(self):
+        self.rows = list(self._stats_data_raw)
+        log_path = Path(artifacts_dir, f"windows_stats_pid_{self.ovms_pid}.log")
+        with log_path.open("w") as csvfile:
+            writer = csv.DictWriter(csvfile, fieldnames=self.FIELDS)
+            writer.writeheader()
+            writer.writerows(self.rows)
+        if WindowsResourceMonitor.on_data_saved:
+            WindowsResourceMonitor.on_data_saved(log_path)
+        return log_path
+
+    def get_stats_by_field(self, field):
+        result = self._get_resource_data()
+        self._stats_data_raw.append(result)
+        value = result[field]
+        if field in (self.WORKING_SET_SIZE, self.PRIVATE_BYTES):
+            return f"{value:.2f}M"
+        return str(value)
+
+    def sample_all(self):
+        """Read one process snapshot and return all tracked metrics as floats (MB / counts)."""
+        stats = self._get_resource_data()
+        self._stats_data_raw.append(stats)
+        return {
+            field: float(stats[field])
+            for field in self.get_validated_metric_names() + self.get_logged_metric_names()
+        }
+
+    @classmethod
+    def get_validated_metric_names(cls):
+        return cls.VALIDATED_FIELDS
+
+    @classmethod
+    def get_logged_metric_names(cls):
+        return cls.LOGGED_FIELDS
+
+    @classmethod
+    def get_memory_metric_names(cls):
+        return cls.VALIDATED_FIELDS + cls.LOGGED_MEMORY_FIELDS
+
+    @classmethod
+    def get_counter_metric_names(cls):
+        return cls.COUNTER_FIELDS

tests/functional/object_model/test_helpers.py

@@ -81,7 +81,8 @@ class Endpoints(enum.Enum):
     RELOAD_CONFIG = "/v1/config/reload"
 
 
-def send_request_to_endpoint(port, address=None, endpoint=None, expected_code=None, retry=1, timeout=60):
+def send_request_to_endpoint(port, address=None, endpoint=None, expected_code=None, retry=1, timeout=60,
+                             retry_delay=1, retry_backoff=1, retry_max_delay=None):
     address = TestEnvironment.get_server_address() if address is None else address
     url_with_endpoint = f"http://{address}:{port}{endpoint}"
     logger.info(f"Try to send request to endpoint: {url_with_endpoint}")
@@ -93,7 +94,8 @@ def send_request_to_endpoint(port, address=None, endpoint=None, expected_code=No
     else:
         msg = f"Not supported endpoint: {endpoint}"
         raise ValueError(msg)
-    retry_setup = {"tries": int(retry), "delay": 1}
+    retry_setup = {"tries": int(retry), "delay": retry_delay, "backoff": retry_backoff,
+                   "max_delay": retry_max_delay}
     kwargs = {"url": url_with_endpoint, "params": {}, "timeout": timeout}
     ret = retry_call(func, fkwargs=kwargs, **retry_setup)
     if expected_code is None:
@@ -114,10 +116,13 @@ def send_request_to_endpoint(port, address=None, endpoint=None, expected_code=No
     return ret
 
 
-def send_reload_request(port, address=None, expected_code=None, retry=1, timeout=60):
+def send_reload_request(port, address=None, expected_code=None, retry=1, timeout=60,
+                        retry_delay=1, retry_backoff=1, retry_max_delay=None):
     address = TestEnvironment.get_server_address() if address is None else address
     endpoint = Endpoints.RELOAD_CONFIG.value
-    return send_request_to_endpoint(port, address, endpoint, expected_code, retry, timeout=timeout)
+    return send_request_to_endpoint(port, address, endpoint, expected_code, retry, timeout=timeout,
+                                    retry_delay=retry_delay, retry_backoff=retry_backoff,
+                                    retry_max_delay=retry_max_delay)
 
 
 def get_config_request(port, address=None, expected_code=None, retry=1):

tests/functional/utils/process.py

@@ -48,6 +48,9 @@ def __init__(self):
     def get_output(self):
         return self._std_stream.pop(), self._err_stream.pop()
 
+    def get_stdout_mirror_path(self):
+        return getattr(self._std_stream, "mirror_file_path", None)
+
     def set_log_verbose(self):
         self.policy['log-run']['verbose'] = True
         self.policy['log-async-run']['verbose'] = True
@@ -223,6 +226,11 @@ def wait(self, timeout):
     def is_alive(self):
         return self._proc.poll() is None
 
+    def poll_exitcode(self):
+        # get_exitcode() may kill the process, so post-mortem callers need a read-only variant.
+        self._proc.poll()
+        return self._proc.returncode
+
     def timeout_detected(self):
         return self._std_stream.timeout_detected or self._err_stream.timeout_detected
 
@@ -400,7 +408,7 @@ def __init__(self, stream, local_thread=True, mirror_file_path=None):
         self._stream = stream
         self.timeout_detected = False
         self._local_thread = local_thread
-        self._mirror_file_path = mirror_file_path
+        self.mirror_file_path = mirror_file_path
 
     def run(self):
         self._thread = Thread(target=self._enqueue_output, args=(self._stream, ))
@@ -410,11 +418,13 @@ def run(self):
     def _enqueue_output(self, out):
         mirror_fd = None
         try:
-            if self._mirror_file_path:
-                mirror_dir = os.path.dirname(self._mirror_file_path)
+            if self.mirror_file_path:
+                mirror_dir = os.path.dirname(self.mirror_file_path)
                 if mirror_dir:
                     os.makedirs(mirror_dir, exist_ok=True)
-                mirror_fd = open(self._mirror_file_path, "a", encoding="utf-8", errors="ignore")
+                # Text mode would turn every CRLF the process wrote into CRCRLF, blank-lining the log.
+                mirror_fd = open(self.mirror_file_path, "a", encoding="utf-8", errors="ignore",
+                                 newline="")
             for line in iter(out.readline, b'' if self._local_thread else ''):
                 line = line.decode('utf8', 'ignore') if self._local_thread else line
                 self._queue.put(line)