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.
      Prints RSS usage for failed tasks and up to 5 most used it tasks.
      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

@@ -44,6 +44,7 @@
 from listeners import LogOutputWatcher
 from listeners import OutputWatcher
 from listeners import StatisticsWatcher
+from listeners import StatisticsMonitor
 
 
 class TcpPortDispatcher:
@@ -122,19 +123,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)
@@ -167,8 +169,10 @@ def init_listeners(self):
         self.statistics = StatisticsWatcher(log_output_watcher.get_logfile)
         self.artifacts = ArtifactsWatcher(log_output_watcher.get_logfile)
         output_watcher = OutputWatcher()
+        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,
-                          sampler.watcher]
+                          sampler.watcher, self.stat_monitor]
         if watch_fail:
             self.fail_watcher = FailWatcher(self.terminate_all_workers)
             self.listeners.append(self.fail_watcher)

lib/sampler.py

@@ -5,6 +5,7 @@
 from lib.colorer import color_log
 from lib.colorer import qa_notice
 from lib.utils import format_process
+from lib.utils import get_proc_stat_rss
 
 
 if sys.version_info[0] == 2:
@@ -31,6 +32,10 @@ def __init__(self, sampler):
         self._last_sample = 0
         self._sample_interval = 0.1  # seconds
         self._warn_interval = self._sample_interval * 4
+        self.rss_procs_results = dict()
+
+    def get_rss_procs_results(self):
+        return self._sampler.rss_results
 
     def process_result(self, obj):
         if isinstance(obj, RegisterProcessMessage):
@@ -55,6 +60,7 @@ def _wakeup(self):
             qa_notice(template.format(self._sample_interval, self._warn_interval,
                                       delta))
         if delta > self._sample_interval:
+            self._sampler.collect_rss()
             self._sampler._sample()
             self._last_sample = now
 
@@ -72,6 +78,8 @@ def __init__(self):
         self._watcher = SamplerWatcher(self)
 
         self.processes = dict()
+        self.tasks_pids = dict()
+        self.rss_results = dict()
 
     def set_queue(self, queue, worker_id, worker_name):
         # Called from a worker process (_run_worker()).
@@ -80,6 +88,20 @@ def set_queue(self, queue, worker_id, worker_name):
         self._queue = queue
         self._watcher = None
 
+    def collect_rss(self):
+        # loop for each registerred task
+        for task in self.tasks_pids:
+            pids_rss = 0
+            # loop for each registerred pid
+            for pid in self.tasks_pids[task]:
+                # refresh current RSS value of the pid
+                self.tasks_pids[task][pid] = get_proc_stat_rss(pid)
+                pids_rss = pids_rss + self.tasks_pids[task][pid]
+            # save current overall RSS value if it is bigger that saved
+            task_name = task[0] + ((':' + task[1]) if task[1] else '')
+            if task_name not in self.rss_results or self.rss_results[task_name] < pids_rss:
+                self.rss_results[task_name] = pids_rss
+
     @property
     def watcher(self):
         if not self._watcher:
@@ -102,6 +124,9 @@ def register_process(self, pid, task_id, server_name, worker_id=None,
                 'worker_id': worker_id,
                 'worker_name': worker_name,
             }
+            if task_id not in self.tasks_pids:
+                self.tasks_pids[task_id] = dict()
+            self.tasks_pids[task_id][pid] = get_proc_stat_rss(pid)
             self._log('register', pid)
             return
 
@@ -119,6 +144,9 @@ def unregister_process(self, pid):
 
         self._log('unregister', pid)
         del self.processes[pid]
+        for task_id in self.tasks_pids:
+            if pid in self.tasks_pids[task_id]:
+                del self.tasks_pids[task_id][pid]
 
     def _log(self, event, pid):
         # Those logs are not written due to gh-247.

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,8 @@
 from lib.utils import safe_makedirs
 from lib.utils import print_tail_n
 from lib.utils import print_unidiff
+from lib.utils import get_vardir_device
+from lib.utils import get_disk_bound_stat_busy
 
 
 class BaseWatcher(object):
@@ -178,6 +180,67 @@ 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.failed_tasks = dict()
+        self.rss_results = dict()
+
+    # task name set as test name with running configuration
+    def get_task_name_conf(self, name, conf):
+        if not conf:
+            return name
+        return name + ':' + conf
+
+    def process_result(self, obj):
+        # called only once on task run finished
+        if isinstance(obj, WorkerTaskResult):
+            task_name = self.get_task_name_conf(obj.task_id[0], obj.task_id[1])
+            if obj.short_status == 'fail':
+                self.failed_tasks[task_name] = 1
+
+    def set_procs_rss(self, rss):
+        self.rss_results = rss
+
+    # called only once after all tasks finished
+    def print_statistics(self):
+        top_rss = 10
+        top_timings = 10
+        results_sorted = []
+
+        # prepare standalone subpath '<vardir>/statistics' for statistics files
+        stats_dir = os.path.join(Options().args.vardir, 'statistics')
+        safe_makedirs(stats_dir)
+
+        # RSS
+
+        # 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('* %6.1f %s\n' % (self.rss_results[task_name] / 1000, task_name),
+                                 schema='test_var')
+
+        # print to stdout RSS statistics for some number of most it used tasks
+        color_stdout('Up to {} most RSS used tasks in Mb:\n' . format(top_rss), schema='test_var')
+        results_sorted = sorted(self.rss_results.items(), key=lambda x: x[1], reverse=True)
+        count = 0
+        for task_name, rss in results_sorted:
+            if count < top_rss:
+                color_stdout('* %6.1f %s\n' % (int(rss) / 1000, task_name), schema='test_var')
+            count = count + 1
+        color_stdout('-' * 41, "\n", schema='separator')
+
+        # 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

@@ -54,6 +54,7 @@
 import time
 
 from lib import Options
+from lib.sampler import sampler
 from lib.colorer import color_stdout
 from lib.utils import print_tail_n
 from lib.utils import PY3
@@ -119,6 +120,8 @@ def main_loop_parallel():
         dispatcher.wait()
         dispatcher.wait_processes()
         color_stdout('-' * 81, "\n", schema='separator')
+        dispatcher.stat_monitor.set_procs_rss(sampler.watcher.get_rss_procs_results())
+        dispatcher.stat_monitor.print_statistics()
         has_failed = dispatcher.statistics.print_statistics()
         has_undone = dispatcher.report_undone(
             verbose=bool(is_force or not has_failed))