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.
For this delay used already existed delay of 1.0 sec used in listener
'HangWatcher'. Found that its change from 1.0 to 0.1 didn't increase the
testing times. Also found that delay of 0.1 sec is completely enough to
catch RSS use increase, due to tested check:

  tarantool> require('clock').bench(function() local t = {} for i = 1, 1024^2 * 100 do t[i] = true end end)

Which checked that 100 Mb of data allocated in seconds:
  - on CI test host: 3.153877479
  - on local fast host: 0.54504489

Also created new listener 'StatisticsMonitor' 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

@@ -43,6 +43,7 @@
 from listeners import LogOutputWatcher
 from listeners import OutputWatcher
 from listeners import StatisticsWatcher
+from listeners import StatisticsMonitor
 
 
 class TcpPortDispatcher:
@@ -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.stat_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.stat_monitor = StatisticsMonitor(output_watcher.not_done_worker_ids, self.worker_id_to_pid)
+        self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts,
+                          self.stat_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,81 @@ def __del__(self):
                 pass
 
 
+class StatisticsMonitor(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.worker_current_task = dict()
+        self.failed_tasks = dict()
+        self.rss_results = dict()
+
+    def process_result(self, obj):
+        # called only once on task run initiated
+        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])
+
+        # called only once on task run finished
+        if isinstance(obj, WorkerTaskResult):
+            task_name = obj.task_id[0]
+            if obj.short_status == 'fail':
+                self.failed_tasks[task_name] = 1
+
+    # called on each 'delta_seconds' delay within task run
+    def process_timeout(self, delta_seconds):
+        rss = None
+        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
+
+    # called only once after all tasks finished
+    def print_statistics(self):
+        rss_print_limit = 30000
+        tests_not_found = True
+
+        # prepare standalone subpath '<vardir>/statistics' for statistics files
+        stats_dir = os.path.join(Options().args.vardir, 'statistics')
+        safe_makedirs(stats_dir)
+
+        # print RSS statistics to stdout and 'rss.log' file
+        if self.rss_results:
+            return
+
+        # print to stdout RSS statistics for all failed tasks
+        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')
+        # Print to stdout RSS statistics for tasks with RSS bigger than
+        # checked limit except failed tasks that already printed above.
+        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')
+
+        # print RSS statistics to '<vardir>/statistics/rss.log' file
+        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.stat_monitor.print_statistics()
         has_failed = dispatcher.statistics.print_statistics()
         has_undone = dispatcher.report_undone(
             verbose=bool(is_force or not has_failed))