Add RSS memory status collecting routine

Added RSS memory status collecting routine which parses file:

  /sys/fs/cgroup/memory/memory.stat

Closes tarantool/tarantool-qa#98

got RSS in KB per worker

RSS with statistics

Author: avtikhon

dispatcher.py

@@ -41,6 +41,7 @@
 from listeners import FailWatcher
 from listeners import HangWatcher
 from listeners import LogOutputWatcher
+from listeners import RSSMonitor
 from listeners import OutputWatcher
 from listeners import StatisticsWatcher
 
@@ -121,19 +122,20 @@ def __init__(self, task_groups, max_workers_cnt, randomize):
             self.result_queues.append(task_queue_disp.result_queue)
             self.task_queues.append(task_queue_disp.task_queue)
 
-        self.report_timeout = 1.0
+        self.report_timeout = 0.1
+
+        self.pid_to_worker_id = dict()
+        self.worker_id_to_pid = dict()
 
         self.statistics = None
         self.artifacts = None
+        self.rss_monitor = None
         self.fail_watcher = None
         self.listeners = None
         self.init_listeners()
 
         self.max_workers_cnt = min(max_workers_cnt, tasks_cnt)
 
-        self.pid_to_worker_id = dict()
-        self.worker_id_to_pid = dict()
-
         self.randomize = randomize
         self.tcp_port_dispatcher = TcpPortDispatcher(
             range_count=max_workers_cnt)
@@ -166,7 +168,8 @@ def init_listeners(self):
         self.statistics = StatisticsWatcher(log_output_watcher.get_logfile)
         self.artifacts = ArtifactsWatcher(log_output_watcher.get_logfile)
         output_watcher = OutputWatcher()
-        self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts]
+        self.rss_monitor = RSSMonitor(output_watcher.not_done_worker_ids, self.worker_id_to_pid)
+        self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts, self.rss_monitor]
         if watch_fail:
             self.fail_watcher = FailWatcher(self.terminate_all_workers)
             self.listeners.append(self.fail_watcher)

lib/utils.py

@@ -233,6 +233,21 @@ def format_process(pid):
     return 'process %d [%s; %s]' % (pid, status, cmdline)
 
 
+def get_proc_stat_rss(pid):
+    rss = 0
+    try:
+        with open('/proc/%d/status' % pid, 'r') as f:
+            for line in f:
+                if ':' not in line:
+                    continue
+                key, value = line.split(':', 1)
+                if key == 'VmRSS':
+                    rss = int(value.strip().split()[0])
+    except (OSError, IOError):
+        pass
+    return rss
+
+
 def set_fd_cloexec(socket):
     flags = fcntl.fcntl(socket, fcntl.F_GETFD)
     fcntl.fcntl(socket, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)

listeners.py

@@ -2,6 +2,7 @@
 import sys
 import yaml
 import shutil
+import time
 
 from lib import Options
 from lib.colorer import color_stdout
@@ -15,6 +16,7 @@
 from lib.utils import safe_makedirs
 from lib.utils import print_tail_n
 from lib.utils import print_unidiff
+from lib.utils import get_proc_stat_rss
 
 
 class BaseWatcher(object):
@@ -51,6 +53,7 @@ def process_result(self, obj):
                                       obj.show_reproduce_content))
 
     def print_statistics(self):
+        print("ALX ####################\n")
         """Returns are there failed tasks."""
         if self.stats:
             color_stdout('Statistics:\n', schema='test_var')
@@ -178,6 +181,70 @@ def __del__(self):
                 pass
 
 
+class RSSMonitor(BaseWatcher):
+    def __init__(self, get_not_done_worker_ids, worker_id_to_pid):
+        self.get_not_done_worker_ids = get_not_done_worker_ids
+        self.worker_id_to_pid = worker_id_to_pid
+        self.activity = 0.0
+        self.worker_current_task = dict()
+        self.failed_tasks = dict()
+        self.rss_results = dict()
+
+    def process_result(self, obj):
+        if isinstance(obj, WorkerCurrentTask):
+            self.worker_current_task[obj.worker_id] = obj
+            print("ALX process_result ================ {} {}\n" . format(obj.task_name, self.worker_id_to_pid[obj.worker_id]))
+            self.rss_results[obj.task_name] = get_proc_stat_rss(self.worker_id_to_pid[obj.worker_id])
+
+        if isinstance(obj, WorkerTaskResult):
+            if obj.short_status == 'fail':
+                print("ALX FAILED process_result ================ {}\n" . format(obj.task_name))
+                self.failed_tasks[obj.task_name] = 1
+
+    def process_timeout(self, delta_seconds):
+        rss = None
+        rss_new_array = []
+        self.activity += delta_seconds
+        worker_ids = self.get_not_done_worker_ids()
+
+        running_tasks = [task for worker_id, task
+                      in self.worker_current_task.items()
+                      if worker_id in worker_ids]
+        for task in running_tasks:
+            if task.task_name in self.rss_results:
+                rss = get_proc_stat_rss(self.worker_id_to_pid[task.worker_id])
+                if rss > self.rss_results[task.task_name]:
+                    self.rss_results[task.task_name] = rss
+                print("ALX process_timeout ============  {} sec, pid={} '{}' {}\n" . format(round(self.activity, 1), self.worker_id_to_pid[task.worker_id], task.task_name, rss))
+
+    def print_statistics(self):
+        print("ALX ####################\n")
+        """Returns are there failed tasks."""
+        if self.rss_results:
+            color_stdout('RSS Statistics:\n', schema='test_var')
+        for task_name in self.rss_results:
+            color_stdout('* %s: %d\n' % (task_name, self.rss_results[task_name]), schema='test_var')
+
+        if not self.failed_tasks:
+            return False
+
+        color_stdout('Failed tasks:\n', schema='test_var')
+        for task_id, worker_name, result_checksum, show_reproduce_content in self.failed_tasks:
+            logfile = self.get_logfile(worker_name)
+            task_id_str = yaml.safe_dump(task_id, default_flow_style=True)
+            color_stdout('- %s' % task_id_str, schema='test_var')
+            color_stdout('# results file checksum: %s\n' % result_checksum)
+            color_stdout('# logfile:        %s\n' % logfile)
+            reproduce_file_path = get_reproduce_file(worker_name)
+            color_stdout('# reproduce file: %s\n' % reproduce_file_path)
+            if show_reproduce_content:
+                color_stdout("---\n", schema='separator')
+                print_tail_n(reproduce_file_path)
+                color_stdout("...\n", schema='separator')
+
+        return True
+
+
 class OutputWatcher(BaseWatcher):
     def __init__(self):
         self.buffer = dict()

test-run.py

@@ -119,6 +119,7 @@ def main_loop_parallel():
         dispatcher.wait()
         dispatcher.wait_processes()
         color_stdout('-' * 81, "\n", schema='separator')
+        rss_monitor = dispatcher.rss_monitor.print_statistics()
         has_failed = dispatcher.statistics.print_statistics()
         has_undone = dispatcher.report_undone(
             verbose=bool(is_force or not has_failed))