Skip to content

Commit e5e9afa

Browse files
authored
CVS-177198 LLM Memory Usage tests (#4360)
### 🛠 Summary [CVS-177198](https://jira.devtools.intel.com/browse/CVS-177198) Update long tests for LLMs memory usage Changes in tests related files required by intel-innersource/frameworks.ai.openvino.model-server.tests#2684 ### 🧪 Checklist - [ ] Unit tests added. - [ ] The documentation updated. - [ ] Change follows security best practices. ``
1 parent 9bbe857 commit e5e9afa

5 files changed

Lines changed: 213 additions & 16 deletions

File tree

‎tests/functional/object_model/ovms_instance.py‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@
4141
from tests.functional.utils.core import get_children_from_module
4242
from tests.functional.utils.inference.communication import GRPC, REST
4343
from tests.functional.utils.logger import get_logger
44-
from tests.functional.constants.os_type import OsType
44+
from tests.functional.constants.os_type import OsType, get_host_os
4545
from tests.functional.utils.port_manager import PortManager
4646
from tests.functional.utils.process import Process
4747
from tests.functional.utils.test_framework import change_dir_permissions, is_single_threaded
@@ -66,7 +66,7 @@
6666
from tests.functional.object_model.mediapipe_calculators import MediaPipeCalculator
6767
from tests.functional.object_model.ovms_config import OvmsConfig
6868
from tests.functional.object_model.package_manager import PackageManager
69-
from tests.functional.object_model.resource_monitor import DockerResourceMonitor
69+
from tests.functional.object_model.resource_monitor import DockerResourceMonitor, WindowsResourceMonitor
7070
from tests.functional.object_model.test_environment import TestEnvironment
7171

7272
logger = get_logger(__name__)
@@ -539,7 +539,13 @@ def attach_context(self, context):
539539
def attach_resource_monitor(self, context, start=True):
540540
if hasattr(self.ovms, "container"):
541541
self.resource_monitor = DockerResourceMonitor(self.ovms.container)
542-
if start:
543-
self.resource_monitor.start()
544-
context.test_objects.append(self.resource_monitor)
545-
return self.resource_monitor
542+
elif get_host_os() == OsType.Windows or getattr(context, "base_os", None) == OsType.Windows:
543+
ovms_pid = self.ovms._dmesg_log.ovms_pid
544+
assert ovms_pid is not None, "Cannot attach Windows resource monitor: ovms_pid is not available"
545+
self.resource_monitor = WindowsResourceMonitor(ovms_pid)
546+
else:
547+
return None
548+
if start:
549+
self.resource_monitor.start()
550+
context.test_objects.append(self.resource_monitor)
551+
return self.resource_monitor

‎tests/functional/object_model/ovms_log_monitor.py‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
#
1616

1717
import datetime
18+
import os
1819
import re
1920
import time
2021

@@ -362,6 +363,7 @@ class BinaryOvmsLogMonitor(OvmsLogMonitor):
362363
def __init__(self, ovms_process, **kwargs):
363364
super().__init__(**kwargs)
364365
self._proc = ovms_process
366+
self._mirror_offset = 0
365367

366368
def is_ovms_running(self):
367369
status = get_pid_status(self._proc._proc.pid)
@@ -371,13 +373,30 @@ def is_ovms_running(self):
371373
return True
372374
return True
373375

376+
def _read_mirror_tail(self, mirror_path):
377+
with open(mirror_path, "rb") as fd:
378+
fd.seek(self._mirror_offset)
379+
chunk = fd.read()
380+
# A trailing partial line would be torn in half between two reads.
381+
complete, newline, _partial = chunk.rpartition(b"\n")
382+
if not newline:
383+
return []
384+
self._mirror_offset += len(complete) + len(newline)
385+
return complete.decode("utf-8", "ignore").splitlines()
386+
374387
def get_all_logs(self):
375388
stdout, stderr = self._proc.get_output()
376389
if stderr:
377390
logger.error(
378391
f"Detect non-empty stderr! It is recommended to redirect stderr to stdout: 2>&1. STDERR: {stderr}"
379392
)
380-
self._read_lines += stdout.splitlines()
393+
mirror_path = self._proc.get_stdout_mirror_path()
394+
if mirror_path and os.path.exists(mirror_path):
395+
# pop() drains the queue shared by every monitor of this process, so the lines one
396+
# monitor takes would be missing from the others and from the saved log.
397+
self._read_lines += self._read_mirror_tail(mirror_path)
398+
else:
399+
self._read_lines += stdout.splitlines()
381400
return self._read_lines
382401

383402

‎tests/functional/object_model/resource_monitor.py‎

Lines changed: 158 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,11 @@
1717
import csv
1818
import threading
1919
from abc import ABC, abstractmethod
20+
from datetime import datetime
2021
from pathlib import Path
2122

2223
import numpy as np
24+
import psutil
2325
from dateutil import parser
2426

2527
from tests.functional.utils.logger import get_logger
@@ -32,6 +34,7 @@ class ResourceMonitor(threading.Thread, ABC):
3234
def __init__(self):
3335
threading.Thread.__init__(self)
3436
self._stop_event = threading.Event()
37+
self.stop_reason = None
3538

3639
def stop(self):
3740
self._stop_event.set()
@@ -42,6 +45,7 @@ def run(self):
4245
try:
4346
self.check_resources()
4447
except StopIteration as e:
48+
self.stop_reason = str(e)
4549
self._stop_event.set()
4650
break
4751
self.save_data()
@@ -55,13 +59,33 @@ def save_data(self):
5559
pass
5660

5761

62+
def _cgroup_cache_bytes(cgroup_memory_stats):
63+
if "cache" in cgroup_memory_stats:
64+
return float(cgroup_memory_stats.get("cache", 0))
65+
return float(cgroup_memory_stats.get("file", 0))
66+
67+
5868
class DockerResourceMonitor(ResourceMonitor):
5969
MEMORY_USAGE = "MEMORY_USAGE"
60-
FIELDS = ["DATE", "PIDS_COUNT", MEMORY_USAGE] # + ["CPU_USAGE"] # Enable in further releases
70+
PRIVATE_MEMORY = "PRIVATE_MEMORY"
71+
MEMORY_CACHE = "MEMORY_CACHE"
72+
FIELDS = ["DATE", "PIDS_COUNT", MEMORY_USAGE, PRIVATE_MEMORY, MEMORY_CACHE] # + ["CPU_USAGE"] # Enable in further releases
73+
VALIDATED_FIELDS = [MEMORY_USAGE, PRIVATE_MEMORY]
74+
LOGGED_MEMORY_FIELDS = [MEMORY_CACHE]
75+
COUNTER_FIELDS = ["PIDS_COUNT"]
76+
LOGGED_FIELDS = LOGGED_MEMORY_FIELDS + COUNTER_FIELDS
6177
FIELDS_TO_STATS = {
6278
"DATE": lambda x: x["read"],
6379
"PIDS_COUNT": lambda x: int(x["pids_stats"].get("current", "0")),
6480
MEMORY_USAGE: lambda x: "{:.2f}M".format(float(x["memory_stats"].get("usage", "0.0")) / (2**20)),
81+
PRIVATE_MEMORY: lambda x: "{:.2f}M".format(
82+
float(x["memory_stats"].get("stats", {}).get(
83+
"anon", x["memory_stats"].get("stats", {}).get("rss", 0)
84+
)) / (2**20)
85+
),
86+
MEMORY_CACHE: lambda x: "{:.2f}M".format(
87+
_cgroup_cache_bytes(x["memory_stats"].get("stats", {})) / (2**20)
88+
),
6589
# Enable after debug & fixing
6690
# "CPU_USAGE": lambda x:
6791
# [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):
140164
'networks' = {dict: 1} {'eth0': {'rx_bytes': 90, 'rx_packets': 1, 'rx_errors': 0, ...
141165
"""
142166
stats = self.container.stats(stream=False, decode=False)
167+
# Prevent reading zeroes from stopped container
168+
pids = int(stats.get("pids_stats", {}).get("current", 0) or 0)
169+
usage = float(stats.get("memory_stats", {}).get("usage", 0) or 0)
170+
if pids <= 0 or usage <= 0:
171+
reason = (
172+
f"container {self.container.name} reports no live process "
173+
f"(pids_stats.current={pids}, memory_stats.usage={usage} bytes) - the container "
174+
f"stopped, crashed or was removed while the test was still sampling"
175+
)
176+
logger.warning(f"Stopping Docker resource monitor: {reason}")
177+
raise StopIteration(reason)
143178
return stats
144179

145180
def get_stats_by_field(self, field):
146181
result = self._get_resource_data()
147182
self._docker_stats_data_raw.append(result)
148183
return self.get_field_data(field, result)
184+
185+
def sample_all(self):
186+
"""Read one stats snapshot and return all tracked metrics as floats (MB / counts).
187+
188+
A single snapshot keeps every metric in the returned sample mutually
189+
consistent (same instant) and avoids one docker stats call per metric.
190+
"""
191+
stats = self._get_resource_data()
192+
self._docker_stats_data_raw.append(stats)
193+
return {
194+
field: float(str(self.get_field_data(field, stats)).replace("M", ""))
195+
for field in self.get_validated_metric_names() + self.get_logged_metric_names()
196+
}
197+
198+
@classmethod
199+
def get_validated_metric_names(cls):
200+
return cls.VALIDATED_FIELDS
201+
202+
@classmethod
203+
def get_logged_metric_names(cls):
204+
return cls.LOGGED_FIELDS
205+
206+
@classmethod
207+
def get_memory_metric_names(cls):
208+
return cls.VALIDATED_FIELDS + cls.LOGGED_MEMORY_FIELDS
209+
210+
@classmethod
211+
def get_counter_metric_names(cls):
212+
return cls.COUNTER_FIELDS
213+
214+
215+
class WindowsResourceMonitor(ResourceMonitor):
216+
WORKING_SET_SIZE = "WORKING_SET_SIZE"
217+
PRIVATE_BYTES = "PRIVATE_BYTES"
218+
PAGE_FAULTS = "PAGE_FAULTS"
219+
220+
MEMORY_USAGE = WORKING_SET_SIZE
221+
222+
FIELDS = ["DATE", WORKING_SET_SIZE, PRIVATE_BYTES, PAGE_FAULTS]
223+
VALIDATED_FIELDS = [PRIVATE_BYTES]
224+
LOGGED_MEMORY_FIELDS = [WORKING_SET_SIZE]
225+
COUNTER_FIELDS = [PAGE_FAULTS]
226+
LOGGED_FIELDS = LOGGED_MEMORY_FIELDS + COUNTER_FIELDS
227+
SAMPLE_INTERVAL_SEC = 1.0
228+
# Optional callback invoked after save_data with (log_path).
229+
on_data_saved = None
230+
231+
def __init__(self, ovms_pid):
232+
super().__init__()
233+
self.ovms_pid = ovms_pid
234+
self._process = psutil.Process(int(ovms_pid))
235+
self._stats_data_raw = []
236+
237+
def cleanup(self):
238+
if not self._stop_event.is_set():
239+
if self.is_alive():
240+
self.stop()
241+
self.save_data()
242+
243+
def _get_resource_data(self):
244+
stats = {"DATE": datetime.now().isoformat()}
245+
try:
246+
info = self._process.memory_info()
247+
except psutil.Error as error:
248+
reason = (f"OVMS process pid={self.ovms_pid} is no longer readable ({error}) - it "
249+
f"exited, crashed or was killed while the test was still sampling")
250+
logger.warning(f"Stopping Windows resource monitor: {reason}")
251+
raise StopIteration(reason) from error
252+
# wset/private are the Win32 counters .NET exposes as WorkingSet64/PrivateMemorySize64.
253+
stats[self.WORKING_SET_SIZE] = float(info.wset) / (1024 * 1024)
254+
stats[self.PRIVATE_BYTES] = float(info.private) / (1024 * 1024)
255+
stats[self.PAGE_FAULTS] = int(info.num_page_faults)
256+
return stats
257+
258+
def check_resources(self):
259+
result = self._get_resource_data()
260+
self._stats_data_raw.append(result)
261+
self._stop_event.wait(self.SAMPLE_INTERVAL_SEC)
262+
263+
def save_data(self):
264+
self.rows = list(self._stats_data_raw)
265+
log_path = Path(artifacts_dir, f"windows_stats_pid_{self.ovms_pid}.log")
266+
with log_path.open("w") as csvfile:
267+
writer = csv.DictWriter(csvfile, fieldnames=self.FIELDS)
268+
writer.writeheader()
269+
writer.writerows(self.rows)
270+
if WindowsResourceMonitor.on_data_saved:
271+
WindowsResourceMonitor.on_data_saved(log_path)
272+
return log_path
273+
274+
def get_stats_by_field(self, field):
275+
result = self._get_resource_data()
276+
self._stats_data_raw.append(result)
277+
value = result[field]
278+
if field in (self.WORKING_SET_SIZE, self.PRIVATE_BYTES):
279+
return f"{value:.2f}M"
280+
return str(value)
281+
282+
def sample_all(self):
283+
"""Read one process snapshot and return all tracked metrics as floats (MB / counts)."""
284+
stats = self._get_resource_data()
285+
self._stats_data_raw.append(stats)
286+
return {
287+
field: float(stats[field])
288+
for field in self.get_validated_metric_names() + self.get_logged_metric_names()
289+
}
290+
291+
@classmethod
292+
def get_validated_metric_names(cls):
293+
return cls.VALIDATED_FIELDS
294+
295+
@classmethod
296+
def get_logged_metric_names(cls):
297+
return cls.LOGGED_FIELDS
298+
299+
@classmethod
300+
def get_memory_metric_names(cls):
301+
return cls.VALIDATED_FIELDS + cls.LOGGED_MEMORY_FIELDS
302+
303+
@classmethod
304+
def get_counter_metric_names(cls):
305+
return cls.COUNTER_FIELDS

‎tests/functional/object_model/test_helpers.py‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,8 @@ class Endpoints(enum.Enum):
8181
RELOAD_CONFIG = "/v1/config/reload"
8282

8383

84-
def send_request_to_endpoint(port, address=None, endpoint=None, expected_code=None, retry=1, timeout=60):
84+
def send_request_to_endpoint(port, address=None, endpoint=None, expected_code=None, retry=1, timeout=60,
85+
retry_delay=1, retry_backoff=1, retry_max_delay=None):
8586
address = TestEnvironment.get_server_address() if address is None else address
8687
url_with_endpoint = f"http://{address}:{port}{endpoint}"
8788
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
9394
else:
9495
msg = f"Not supported endpoint: {endpoint}"
9596
raise ValueError(msg)
96-
retry_setup = {"tries": int(retry), "delay": 1}
97+
retry_setup = {"tries": int(retry), "delay": retry_delay, "backoff": retry_backoff,
98+
"max_delay": retry_max_delay}
9799
kwargs = {"url": url_with_endpoint, "params": {}, "timeout": timeout}
98100
ret = retry_call(func, fkwargs=kwargs, **retry_setup)
99101
if expected_code is None:
@@ -114,10 +116,13 @@ def send_request_to_endpoint(port, address=None, endpoint=None, expected_code=No
114116
return ret
115117

116118

117-
def send_reload_request(port, address=None, expected_code=None, retry=1, timeout=60):
119+
def send_reload_request(port, address=None, expected_code=None, retry=1, timeout=60,
120+
retry_delay=1, retry_backoff=1, retry_max_delay=None):
118121
address = TestEnvironment.get_server_address() if address is None else address
119122
endpoint = Endpoints.RELOAD_CONFIG.value
120-
return send_request_to_endpoint(port, address, endpoint, expected_code, retry, timeout=timeout)
123+
return send_request_to_endpoint(port, address, endpoint, expected_code, retry, timeout=timeout,
124+
retry_delay=retry_delay, retry_backoff=retry_backoff,
125+
retry_max_delay=retry_max_delay)
121126

122127

123128
def get_config_request(port, address=None, expected_code=None, retry=1):

‎tests/functional/utils/process.py‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,9 @@ def __init__(self):
4848
def get_output(self):
4949
return self._std_stream.pop(), self._err_stream.pop()
5050

51+
def get_stdout_mirror_path(self):
52+
return getattr(self._std_stream, "mirror_file_path", None)
53+
5154
def set_log_verbose(self):
5255
self.policy['log-run']['verbose'] = True
5356
self.policy['log-async-run']['verbose'] = True
@@ -223,6 +226,11 @@ def wait(self, timeout):
223226
def is_alive(self):
224227
return self._proc.poll() is None
225228

229+
def poll_exitcode(self):
230+
# get_exitcode() may kill the process, so post-mortem callers need a read-only variant.
231+
self._proc.poll()
232+
return self._proc.returncode
233+
226234
def timeout_detected(self):
227235
return self._std_stream.timeout_detected or self._err_stream.timeout_detected
228236

@@ -400,7 +408,7 @@ def __init__(self, stream, local_thread=True, mirror_file_path=None):
400408
self._stream = stream
401409
self.timeout_detected = False
402410
self._local_thread = local_thread
403-
self._mirror_file_path = mirror_file_path
411+
self.mirror_file_path = mirror_file_path
404412

405413
def run(self):
406414
self._thread = Thread(target=self._enqueue_output, args=(self._stream, ))
@@ -410,11 +418,13 @@ def run(self):
410418
def _enqueue_output(self, out):
411419
mirror_fd = None
412420
try:
413-
if self._mirror_file_path:
414-
mirror_dir = os.path.dirname(self._mirror_file_path)
421+
if self.mirror_file_path:
422+
mirror_dir = os.path.dirname(self.mirror_file_path)
415423
if mirror_dir:
416424
os.makedirs(mirror_dir, exist_ok=True)
417-
mirror_fd = open(self._mirror_file_path, "a", encoding="utf-8", errors="ignore")
425+
# Text mode would turn every CRLF the process wrote into CRCRLF, blank-lining the log.
426+
mirror_fd = open(self.mirror_file_path, "a", encoding="utf-8", errors="ignore",
427+
newline="")
418428
for line in iter(out.readline, b'' if self._local_thread else ''):
419429
line = line.decode('utf8', 'ignore') if self._local_thread else line
420430
self._queue.put(line)

0 commit comments

Comments
 (0)