Tests: Fix high memory usage on dumping profile log

cloud/filestore/tests/fio/qemu-kikimr-mq-test/test.py

@@ -23,7 +23,7 @@ def test_fio(name):
     profile_tool_bin_path = common.binary_path(
         "cloud/filestore/tools/analytics/profile_tool/filestore-profile-tool")
     fs_name = "nfs_test"
-    events = profile.get_profile_log_events(
+    events = profile.iter_profile_log_events(
         profile_tool_bin_path,
         common.output_path("vhost-profile.log"),
         fs_name)

cloud/filestore/tests/fio/qemu-kikimr-mq-test/ya.make

@@ -30,7 +30,6 @@ SET(
 SET(QEMU_VIRTIO fs)
 SET(QEMU_ROOTFS cloud/storage/core/tools/testing/qemu/image-noble/rootfs.img)
 SET(QEMU_NUM_REQUEST_QUEUES 8)
-SET(QEMU_MEM 8G)
 
 INCLUDE(${ARCADIA_ROOT}/cloud/filestore/tests/recipes/service-kikimr.inc)
 INCLUDE(${ARCADIA_ROOT}/cloud/filestore/tests/recipes/vhost-kikimr.inc)

cloud/filestore/tests/fio/qemu-kikimr-nemesis-test/ya.make

@@ -19,7 +19,6 @@ SET(QEMU_VIRTIO fs)
 
 SET(VHOST_RESTART_INTERVAL 10)
 SET(VHOST_RESTART_FLAG 1)
-SET(QEMU_MEM 6G)
 
 INCLUDE(${ARCADIA_ROOT}/cloud/filestore/tests/recipes/service-kikimr.inc)
 INCLUDE(${ARCADIA_ROOT}/cloud/filestore/tests/recipes/vhost-kikimr.inc)

cloud/filestore/tests/fio/qemu-local-noserver-mq-test/test.py

@@ -26,7 +26,7 @@ def test_fio(name):
     profile_tool_bin_path = common.binary_path(
         "cloud/filestore/tools/analytics/profile_tool/filestore-profile-tool")
     fs_name = "nfs_test"
-    events = profile.get_profile_log_events(
+    events = profile.iter_profile_log_events(
         profile_tool_bin_path,
         common.output_path("vhost-profile.log"),
         fs_name)

cloud/filestore/tests/fio/qemu-local-noserver-mq-test/ya.make

@@ -33,7 +33,6 @@ SET(
 SET(QEMU_VIRTIO fs)
 SET(QEMU_ROOTFS cloud/storage/core/tools/testing/qemu/image-noble/rootfs.img)
 SET(QEMU_NUM_REQUEST_QUEUES 8)
-SET(QEMU_MEM 8G)
 
 INCLUDE(${ARCADIA_ROOT}/cloud/filestore/tests/recipes/vhost-local-noserver.inc)
 INCLUDE(${ARCADIA_ROOT}/cloud/filestore/tests/recipes/vhost-endpoint.inc)

cloud/filestore/tests/fio/qemu-local-noserver-no-zero-copy-test/ya.make

@@ -25,7 +25,6 @@ TEST_SRCS(
 SET(QEMU_VIRTIO fs)
 SET(VHOST_RESTART_INTERVAL 10)
 SET(VHOST_RESTART_FLAG 1)
-SET(QEMU_MEM 6G)
 
 SET(
     NFS_LOCAL_SERVICE_CONFIG_PATCH

cloud/filestore/tests/fio/qemu-local-noserver-test/ya.make

@@ -23,7 +23,6 @@ TEST_SRCS(
 )
 
 SET(QEMU_VIRTIO fs)
-SET(QEMU_MEM 6G)
 SET(VHOST_RESTART_INTERVAL 10)
 SET(VHOST_RESTART_FLAG 1)
 

cloud/filestore/tests/fio_index/qemu-kikimr-mq-test/test.py

@@ -22,7 +22,7 @@ def test_fio(name):
     profile_tool_bin_path = common.binary_path(
         "cloud/filestore/tools/analytics/profile_tool/filestore-profile-tool")
     fs_name = "nfs_test"
-    events = profile.get_profile_log_events(
+    events = profile.iter_profile_log_events(
         profile_tool_bin_path,
         common.output_path("vhost-profile.log"),
         fs_name)

cloud/filestore/tests/fio_index/qemu-kikimr-multishard-nemesis-test/ya.make

@@ -20,7 +20,6 @@ TEST_SRCS(
 )
 
 SET(QEMU_VIRTIO fs)
-SET(QEMU_MEM 6G)
 SET(NFS_RESTART_INTERVAL 10)
 SET(VHOST_RESTART_INTERVAL 10)
 SET(VHOST_RESTART_FLAG 1)

cloud/filestore/tests/fio_index/qemu-kikimr-nemesis-test/ya.make

@@ -16,7 +16,6 @@ TEST_SRCS(
 )
 
 SET(QEMU_VIRTIO fs)
-SET(QEMU_MEM 6G)
 SET(VHOST_RESTART_INTERVAL 10)
 SET(VHOST_RESTART_FLAG 1)
 

cloud/filestore/tests/fs_posix_compliance/qemu-kikimr-nemesis-test/ya.make

@@ -11,7 +11,6 @@ TEST_SRCS(
 )
 
 SET(QEMU_VIRTIO fs)
-SET(QEMU_MEM 6G)
 SET(VHOST_RESTART_INTERVAL 10)
 SET(VHOST_RESTART_FLAG 1)
 SET(

cloud/filestore/tools/testing/profile_log/common.py

@@ -1,19 +1,51 @@
 import re
+import tempfile
 
 import yatest.common as common
 
 
+def _iter_profile_log_lines(profile_tool_bin_path,
+                            profile_log_path,
+                            fs_name):
+    # yatest.common.execute captures stdout in a file and then reads the whole
+    # file into memory.  Supplying an open file marks stdout as caller-owned,
+    # so the execution helper leaves it on disk and does not materialize it.
+    with tempfile.TemporaryFile(mode="w+b") as output:
+        common.execute(
+            [profile_tool_bin_path, "dumpevents",
+             "--profile-log", profile_log_path,
+             "--fs-id", fs_name],
+            stdout=output)
+
+        output.flush()
+        output.seek(0)
+        for line in output:
+            yield line.decode("utf-8")
+
+
+def _parse_profile_log_event(line):
+    parts = line.rstrip().split("\t")
+    request_type = parts[2]
+    body_dict = {}
+    for i in range(5, len(parts)):
+        body_str = re.sub(r"[{}\[\]]", "", parts[i])
+        body_parts = body_str.split(", ")
+        for body_part in body_parts:
+            kv = body_part.split("=", 2)
+            body_dict[kv[0]] = kv[1] if len(kv) == 2 else None
+
+    return request_type, body_dict
+
+
 def analyze_profile_log(profile_tool_bin_path,
                         profile_log_path,
                         fs_name,
                         node_name_filter=None):
-    proc = common.execute(
-        [profile_tool_bin_path, "dumpevents",
-         "--profile-log", profile_log_path,
-         "--fs-id", fs_name])
-
     type_dict = {}
-    for line in proc.stdout.decode("utf-8").splitlines():
+    for line in _iter_profile_log_lines(
+            profile_tool_bin_path,
+            profile_log_path,
+            fs_name):
         request_type = line.rstrip().split("\t")[2]
 
         if node_name_filter and f"node_name={node_name_filter}" not in line:
@@ -24,27 +56,29 @@ def analyze_profile_log(profile_tool_bin_path,
     return type_dict
 
 
-def get_profile_log_events(profile_tool_bin_path, profile_log_path, fs_name):
-    proc = common.execute(
-        [profile_tool_bin_path, "dumpevents",
-         "--profile-log", profile_log_path,
-         "--fs-id", fs_name])
+def iter_profile_log_events(profile_tool_bin_path,
+                            profile_log_path,
+                            fs_name):
+    for line in _iter_profile_log_lines(
+            profile_tool_bin_path,
+            profile_log_path,
+            fs_name):
+        yield _parse_profile_log_event(line)
+
 
-    events = []
-    for line in proc.stdout.decode("utf-8").splitlines():
-        parts = line.rstrip().split("\t")
-        request_type = parts[2]
-        body_dict = {}
-        for i in range(5, len(parts)):
-            body_str = re.sub(r"[{}\[\]]", "", parts[i])
-            body_parts = body_str.split(", ")
-            for body_part in body_parts:
-                kv = body_part.split("=", 2)
-                body_dict[kv[0]] = kv[1] if len(kv) == 2 else None
+def get_profile_log_events(profile_tool_bin_path,
+                           profile_log_path,
+                           fs_name):
+    """Return all profile events as a list.
 
-        events.append((request_type, body_dict))
+    Prefer iter_profile_log_events for large profile logs so parsed events are
+    not retained in memory.
+    """
 
-    return events
+    return list(iter_profile_log_events(
+        profile_tool_bin_path,
+        profile_log_path,
+        fs_name))
 
 
 def dump_profile_log(profile_tool_bin_path,

cloud/filestore/tools/testing/profile_log/ut/common_ut.py

@@ -0,0 +1,161 @@
+from unittest import mock
+
+import cloud.filestore.tools.testing.profile_log.common as profile_log
+
+
+def _event_line(request_type, *body_parts):
+    return "\t".join([
+        "1970-01-01T00:00:00.000010Z",
+        "nfs_test",
+        request_type,
+        "0.000020s",
+        "S_OK",
+        *body_parts,
+    ]) + "\n"
+
+
+class _ExecutionResult:
+    @property
+    def stdout(self):
+        raise AssertionError("dumpevents stdout must not be materialized")
+
+
+def _execute_writer(dump):
+    streams = []
+
+    def execute(command, **kwargs):
+        output = kwargs.get("stdout")
+        assert output is not None
+
+        # Writing bytes verifies that the caller supplied a binary file rather
+        # than asking common.execute to capture stdout in memory.
+        output.write(dump.encode("utf-8"))
+        output.flush()
+        streams.append(output)
+        return _ExecutionResult()
+
+    return execute, streams
+
+
+def test_iter_profile_log_events_uses_caller_owned_binary_stdout():
+    dump = _event_line(
+        "ReadData",
+        "{node_id=42, handle=3}",
+        "[{offset=4096, bytes=8192}]",
+        "loop_thread_id=17",
+        "client_id=test-client",
+    )
+    execute, streams = _execute_writer(dump)
+
+    with mock.patch.object(
+        profile_log.common,
+        "execute",
+        side_effect=execute,
+    ) as execute_mock:
+        events = list(profile_log.iter_profile_log_events(
+            "/test/filestore-profile-tool",
+            "/test/profile.log",
+            "nfs_test",
+        ))
+
+    assert events == [(
+        "ReadData",
+        {
+            "node_id": "42",
+            "handle": "3",
+            "offset": "4096",
+            "bytes": "8192",
+            "loop_thread_id": "17",
+            "client_id": "test-client",
+        },
+    )]
+    execute_mock.assert_called_once()
+    assert execute_mock.call_args.args[0] == [
+        "/test/filestore-profile-tool",
+        "dumpevents",
+        "--profile-log",
+        "/test/profile.log",
+        "--fs-id",
+        "nfs_test",
+    ]
+    assert execute_mock.call_args.kwargs["stdout"] is streams[0]
+    assert streams[0].closed
+
+
+def test_iter_profile_log_events_parses_incrementally():
+    lines = [
+        _event_line("ReadData", "loop_thread_id=1"),
+        _event_line("WriteData", "loop_thread_id=2"),
+    ]
+    consumed = []
+
+    def iter_lines(*_args, **_kwargs):
+        for line in lines:
+            consumed.append(line)
+            yield line
+
+    with mock.patch.object(
+        profile_log,
+        "_iter_profile_log_lines",
+        side_effect=iter_lines,
+    ):
+        events = profile_log.iter_profile_log_events(
+            "/test/filestore-profile-tool",
+            "/test/profile.log",
+            "nfs_test",
+        )
+
+        assert iter(events) is events
+        assert consumed == []
+        assert next(events) == ("ReadData", {"loop_thread_id": "1"})
+        assert consumed == [lines[0]]
+        assert list(events) == [("WriteData", {"loop_thread_id": "2"})]
+
+
+def test_get_profile_log_events_remains_a_list_wrapper():
+    expected = [
+        ("ReadData", {"loop_thread_id": "1"}),
+        ("WriteData", {"loop_thread_id": "2"}),
+    ]
+
+    with mock.patch.object(
+        profile_log,
+        "iter_profile_log_events",
+        return_value=iter(expected),
+    ) as iter_events:
+        events = profile_log.get_profile_log_events(
+            "/test/filestore-profile-tool",
+            "/test/profile.log",
+            "nfs_test",
+        )
+
+    assert isinstance(events, list)
+    assert events == expected
+    iter_events.assert_called_once_with(
+        "/test/filestore-profile-tool",
+        "/test/profile.log",
+        "nfs_test",
+    )
+
+
+def test_analyze_profile_log_filters_by_node_name():
+    dump = "".join([
+        _event_line("ReadData", "{node_name=selected, node_id=1}"),
+        _event_line("WriteData", "{node_name=other, node_id=2}"),
+        _event_line("ReadData", "{node_name=selected, node_id=3}"),
+    ])
+    execute, _ = _execute_writer(dump)
+
+    with mock.patch.object(
+        profile_log.common,
+        "execute",
+        side_effect=execute,
+    ):
+        counts = profile_log.analyze_profile_log(
+            "/test/filestore-profile-tool",
+            "/test/profile.log",
+            "nfs_test",
+            node_name_filter="selected",
+        )
+
+    assert counts == {"ReadData": 2}

cloud/filestore/tools/testing/profile_log/ut/ya.make

@@ -0,0 +1,11 @@
+PY3TEST()
+
+TEST_SRCS(
+    common_ut.py
+)
+
+PEERDIR(
+    cloud/filestore/tools/testing/profile_log
+)
+
+END()

cloud/filestore/tools/testing/profile_log/ya.make

@@ -8,3 +8,5 @@ PY_SRCS(
 )
 
 END()
+
+RECURSE_FOR_TESTS(ut)