Add RSS statistics collecting

Found that some tests may fail due to lack of memory. Mostly it happens
in CI on remote hosts. To be able to collect memory used statistic
decided to add RSS memory status collecting routine get_proc_stat_rss()
which parses files:

  /proc/<worker pid>/status

for RSS value 'VmRSS' which is size of memory portions. It contains the
three following parts (VmRSS = RssAnon + RssFile + RssShmem) [1]:

  RssAnon - size of resident anonymous memory
  RssFile - size of resident file mappings
  RssShmem - size of resident shmem memory (includes SysV shm, mapping of
             tmpfs and shared anonymous mappings)

Decided that the best way for CI not to run this RSS collecting routine
for each sent command from tests tasks, but to run it when the test task
started and each 0.1 second delay after, to collect maximun RSS value.

Also created new listener 'RSSMonitor' which has the following routines
and its use:

  process_result() - called when test task started and finished:
      Using 'WorkerCurrentTask' queue it saves the initial RSS value of
      the used worker when task started to run.
      Using 'WorkerTaskResult' queue collect tasks that failed.

  process_timeout() - called after each 0.1 sec delay of the task run.
      It saves/updates worker RSS value for the current test task
      choosing its maximum values.

  print_statistics() - statistics printing to stdout after testing.
      Checked and choosed RSS memory limit of 30000 kB RSS below which
      tests are not interesting to show except failed.
      Created new subdirectory 'statistics' in 'vardir' path to save
      statistics files. The current patch uses it to save there
      'rss.log' file with RSS values per tested tasks in format:
         <test task name> <maximum RSS value>

Closes tarantool/tarantool-qa#98

[1]: https://www.kernel.org/doc/html/latest/filesystems/proc.html

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,9 @@ 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

@@ -15,6 +15,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):
@@ -178,6 +179,72 @@ 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
+            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':
+                self.failed_tasks[obj.task_id[0]] = 1
+
+    def process_timeout(self, delta_seconds):
+        rss = None
+        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
+
+    def print_statistics(self):
+        rss_print_limit = 30000
+        tests_not_found = True
+
+        """Returns are there failed tasks."""
+        if not self.rss_results:
+            return
+
+        if self.failed_tasks:
+            color_stdout('RSS of the failed tasks:\n', schema='test_var')
+            for task_name in self.failed_tasks:
+                if task_name in self.rss_results:
+                    color_stdout('* %s: %d\n' % (task_name, self.rss_results[task_name]),
+                                 schema='test_var')
+        color_stdout('Tasks used more than {} kB of RSS:\n' . format(rss_print_limit),
+                     schema='test_var')
+        for task_name in self.rss_results:
+            if task_name not in self.failed_tasks and self.rss_results[task_name] > rss_print_limit:
+                color_stdout('* %s: %d\n' % (task_name, self.rss_results[task_name]),
+                             schema='test_var')
+                tests_not_found = False
+        if tests_not_found:
+            color_stdout('* tasks not found that used such RSS !\n', schema='test_var')
+
+        stats_dir = os.path.join(Options().args.vardir, 'statistics')
+        safe_makedirs(stats_dir)
+        filepath = os.path.join(stats_dir, 'rss.log')
+        fd = open(filepath, 'w')
+        for task_name in self.rss_results:
+            fd.write("{} {}\n" . format(task_name, self.rss_results[task_name]))
+        fd.close()
+
+
 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')
+        dispatcher.rss_monitor.print_statistics()
         has_failed = dispatcher.statistics.print_statistics()
         has_undone = dispatcher.report_undone(
             verbose=bool(is_force or not has_failed))