wally/suits/io/fio.py

import re
import time
import json
import stat
import random
import hashlib
import os.path
import logging
import datetime
import functools
import collections
from typing import Dict, List, Callable, Any, Tuple, Optional

import yaml
import texttable
from paramiko.ssh_exception import SSHException
from concurrent.futures import ThreadPoolExecutor, wait

import wally

from ...pretty_yaml import dumps
from ...statistic import round_3_digit, data_property, average
from ...utils import ssize2b, sec_to_str, StopTestError, Barrier, get_os
from ...inode import INode

from ..itest import (TimeSeriesValue, PerfTest, TestResults, TestConfig)

from .fio_task_parser import (execution_time, fio_cfg_compile,
                              get_test_summary, get_test_summary_tuple,
                              get_test_sync_mode, FioJobSection)
from .rpc_plugin import parse_fio_result


logger = logging.getLogger("wally")


class NoData:
    pass


def cached_prop(func: Callable[..., Any]) -> Callable[..., Any]:
    @property
    @functools.wraps(func)
    def closure(self) -> Any:
        val = getattr(self, "_" + func.__name__)
        if val is NoData:
            val = func(self)
            setattr(self, "_" + func.__name__, val)
        return val
    return closure


def load_fio_log_file(fname: str) -> TimeSeriesValue:
    with open(fname) as fd:
        it = [ln.split(',')[:2] for ln in fd]

    vals = [(float(off) / 1000,  # convert us to ms
             float(val.strip()) + 0.5)  # add 0.5 to compemsate average value
                                        # as fio trimm all values in log to integer
            for off, val in it]

    return TimeSeriesValue(vals)


READ_IOPS_DISCSTAT_POS = 3
WRITE_IOPS_DISCSTAT_POS = 7


def load_sys_log_file(ftype: str, fname: str) -> TimeSeriesValue:
    assert ftype == 'iops'
    pval = None
    with open(fname) as fd:
        iops = []
        for ln in fd:
            params = ln.split()
            cval = int(params[WRITE_IOPS_DISCSTAT_POS]) + \
                int(params[READ_IOPS_DISCSTAT_POS])
            if pval is not None:
                iops.append(cval - pval)
            pval = cval

    vals = [(idx * 1000, val) for idx, val in enumerate(iops)]
    return TimeSeriesValue(vals)


def load_test_results(folder: str, run_num: int) -> 'FioRunResult':
    res = {}
    params = None

    fn = os.path.join(folder, str(run_num) + '_params.yaml')
    params = yaml.load(open(fn).read())

    conn_ids_set = set()
    rr = r"{}_(?P<conn_id>.*?)_(?P<type>[^_.]*)\.\d+\.log$".format(run_num)
    for fname in os.listdir(folder):
        rm = re.match(rr, fname)
        if rm is None:
            continue

        conn_id_s = rm.group('conn_id')
        conn_id = conn_id_s.replace('_', ':')
        ftype = rm.group('type')

        if ftype not in ('iops', 'bw', 'lat'):
            continue

        ts = load_fio_log_file(os.path.join(folder, fname))
        res.setdefault(ftype, {}).setdefault(conn_id, []).append(ts)

        conn_ids_set.add(conn_id)

    rr = r"{}_(?P<conn_id>.*?)_(?P<type>[^_.]*)\.sys\.log$".format(run_num)
    for fname in os.listdir(folder):
        rm = re.match(rr, fname)
        if rm is None:
            continue

        conn_id_s = rm.group('conn_id')
        conn_id = conn_id_s.replace('_', ':')
        ftype = rm.group('type')

        if ftype not in ('iops', 'bw', 'lat'):
            continue

        ts = load_sys_log_file(ftype, os.path.join(folder, fname))
        res.setdefault(ftype + ":sys", {}).setdefault(conn_id, []).append(ts)

        conn_ids_set.add(conn_id)

    mm_res = {}

    if len(res) == 0:
        raise ValueError("No data was found")

    for key, data in res.items():
        conn_ids = sorted(conn_ids_set)
        awail_ids = [conn_id for conn_id in conn_ids if conn_id in data]
        matr = [data[conn_id] for conn_id in awail_ids]
        mm_res[key] = MeasurementMatrix(matr, awail_ids)

    raw_res = {}
    for conn_id in conn_ids:
        fn = os.path.join(folder, "{0}_{1}_rawres.json".format(run_num, conn_id_s))

        # remove message hack
        fc = "{" + open(fn).read().split('{', 1)[1]
        raw_res[conn_id] = json.loads(fc)

    fio_task = FioJobSection(params['name'])
    fio_task.vals.update(params['vals'])

    config = TestConfig('io', params, None, params['nodes'], folder, None)
    return FioRunResult(config, fio_task, mm_res, raw_res, params['intervals'], run_num)


class Attrmapper:
    def __init__(self, dct: Dict[str, Any]):
        self.__dct = dct

    def __getattr__(self, name):
        try:
            return self.__dct[name]
        except KeyError:
            raise AttributeError(name)


class DiskPerfInfo:
    def __init__(self, name: str, summary: str, params: Dict[str, Any], testnodes_count: int):
        self.name = name
        self.bw = None
        self.iops = None
        self.lat = None
        self.lat_50 = None
        self.lat_95 = None
        self.lat_avg = None

        self.raw_bw = []
        self.raw_iops = []
        self.raw_lat = []

        self.params = params
        self.testnodes_count = testnodes_count
        self.summary = summary
        self.p = Attrmapper(self.params['vals'])

        self.sync_mode = get_test_sync_mode(self.params['vals'])
        self.concurence = self.params['vals'].get('numjobs', 1)


def get_lat_perc_50_95(lat_mks: List[float]) -> Tuple[float, float]:
    curr_perc = 0
    perc_50 = None
    perc_95 = None
    pkey = None
    for key, val in sorted(lat_mks.items()):
        if curr_perc + val >= 50 and perc_50 is None:
            if pkey is None or val < 1.:
                perc_50 = key
            else:
                perc_50 = (50. - curr_perc) / val * (key - pkey) + pkey

        if curr_perc + val >= 95:
            if pkey is None or val < 1.:
                perc_95 = key
            else:
                perc_95 = (95. - curr_perc) / val * (key - pkey) + pkey
            break

        pkey = key
        curr_perc += val

    # for k, v in sorted(lat_mks.items()):
    #     if k / 1000 > 0:
    #         print "{0:>4}".format(k / 1000), v

    # print perc_50 / 1000., perc_95 / 1000.
    # exit(1)
    return perc_50 / 1000., perc_95 / 1000.


class IOTestResults:
    def __init__(self, suite_name: str, fio_results: 'FioRunResult', log_directory: str):
        self.suite_name = suite_name
        self.fio_results = fio_results
        self.log_directory = log_directory

    def __iter__(self):
        return iter(self.fio_results)

    def __len__(self):
        return len(self.fio_results)

    def get_yamable(self) -> Dict[str, List[str]]:
        items = [(fio_res.summary(), fio_res.idx) for fio_res in self]
        return {self.suite_name: [self.log_directory] + items}


class FioRunResult(TestResults):
    """
    Fio run results
    config: TestConfig
    fio_task: FioJobSection
    ts_results: {str: MeasurementMatrix[TimeSeriesValue]}
    raw_result: ????
    run_interval:(float, float) - test tun time, used for sensors
    """
    def __init__(self, config, fio_task, ts_results, raw_result, run_interval, idx):

        self.name = fio_task.name.rsplit("_", 1)[0]
        self.fio_task = fio_task
        self.idx = idx

        self.bw = ts_results['bw']
        self.lat = ts_results['lat']
        self.iops = ts_results['iops']

        if 'iops:sys' in ts_results:
            self.iops_sys = ts_results['iops:sys']
        else:
            self.iops_sys = None

        res = {"bw": self.bw,
               "lat": self.lat,
               "iops": self.iops,
               "iops:sys": self.iops_sys}

        self.sensors_data = None
        self._pinfo = None
        TestResults.__init__(self, config, res, raw_result, run_interval)

    def get_params_from_fio_report(self):
        nodes = self.bw.connections_ids

        iops = [self.raw_result[node]['jobs'][0]['mixed']['iops'] for node in nodes]
        total_ios = [self.raw_result[node]['jobs'][0]['mixed']['total_ios'] for node in nodes]
        runtime = [self.raw_result[node]['jobs'][0]['mixed']['runtime'] / 1000 for node in nodes]
        flt_iops = [float(ios) / rtime for ios, rtime in zip(total_ios, runtime)]

        bw = [self.raw_result[node]['jobs'][0]['mixed']['bw'] for node in nodes]
        total_bytes = [self.raw_result[node]['jobs'][0]['mixed']['io_bytes'] for node in nodes]
        flt_bw = [float(tbytes) / rtime for tbytes, rtime in zip(total_bytes, runtime)]

        return {'iops': iops,
                'flt_iops': flt_iops,
                'bw': bw,
                'flt_bw': flt_bw}

    def summary(self):
        return get_test_summary(self.fio_task, len(self.config.nodes))

    def summary_tpl(self):
        return get_test_summary_tuple(self.fio_task, len(self.config.nodes))

    def get_lat_perc_50_95_multy(self):
        lat_mks = collections.defaultdict(lambda: 0)
        num_res = 0

        for result in self.raw_result.values():
            num_res += len(result['jobs'])
            for job_info in result['jobs']:
                for k, v in job_info['latency_ms'].items():
                    if isinstance(k, basestring) and k.startswith('>='):
                        lat_mks[int(k[2:]) * 1000] += v
                    else:
                        lat_mks[int(k) * 1000] += v

                for k, v in job_info['latency_us'].items():
                    lat_mks[int(k)] += v

        for k, v in lat_mks.items():
            lat_mks[k] = float(v) / num_res
        return get_lat_perc_50_95(lat_mks)

    def disk_perf_info(self, avg_interval=2.0):

        if self._pinfo is not None:
            return self._pinfo

        testnodes_count = len(self.config.nodes)

        pinfo = DiskPerfInfo(self.name,
                             self.summary(),
                             self.params,
                             testnodes_count)

        def prepare(data, drop=1):
            if data is None:
                return data

            res = []
            for ts_data in data:
                if ts_data.average_interval() < avg_interval:
                    ts_data = ts_data.derived(avg_interval)

                # drop last value on bounds
                # as they may contains ranges without activities
                assert len(ts_data.values) >= drop + 1, str(drop) + " " + str(ts_data.values)

                if drop > 0:
                    res.append(ts_data.values[:-drop])
                else:
                    res.append(ts_data.values)

            return res

        def agg_data(matr):
            arr = sum(matr, [])
            min_len = min(map(len, arr))
            res = []
            for idx in range(min_len):
                res.append(sum(dt[idx] for dt in arr))
            return res

        pinfo.raw_lat = map(prepare, self.lat.per_vm())
        num_th = sum(map(len, pinfo.raw_lat))
        lat_avg = [val / num_th for val in agg_data(pinfo.raw_lat)]
        pinfo.lat_avg = data_property(lat_avg).average / 1000  # us to ms

        pinfo.lat_50, pinfo.lat_95 = self.get_lat_perc_50_95_multy()
        pinfo.lat = pinfo.lat_50

        pinfo.raw_bw = map(prepare, self.bw.per_vm())
        pinfo.raw_iops = map(prepare, self.iops.per_vm())

        if self.iops_sys is not None:
            pinfo.raw_iops_sys = map(prepare, self.iops_sys.per_vm())
            pinfo.iops_sys = data_property(agg_data(pinfo.raw_iops_sys))
        else:
            pinfo.raw_iops_sys = None
            pinfo.iops_sys = None

        fparams = self.get_params_from_fio_report()
        fio_report_bw = sum(fparams['flt_bw'])
        fio_report_iops = sum(fparams['flt_iops'])

        agg_bw = agg_data(pinfo.raw_bw)
        agg_iops = agg_data(pinfo.raw_iops)

        log_bw_avg = average(agg_bw)
        log_iops_avg = average(agg_iops)

        # update values to match average from fio report
        coef_iops = fio_report_iops / float(log_iops_avg)
        coef_bw = fio_report_bw / float(log_bw_avg)

        bw_log = data_property([val * coef_bw for val in agg_bw])
        iops_log = data_property([val * coef_iops for val in agg_iops])

        bw_report = data_property([fio_report_bw])
        iops_report = data_property([fio_report_iops])

        # When IOPS/BW per thread is too low
        # data from logs is rounded to match
        iops_per_th = sum(sum(pinfo.raw_iops, []), [])
        if average(iops_per_th) > 10:
            pinfo.iops = iops_log
            pinfo.iops2 = iops_report
        else:
            pinfo.iops = iops_report
            pinfo.iops2 = iops_log

        bw_per_th = sum(sum(pinfo.raw_bw, []), [])
        if average(bw_per_th) > 10:
            pinfo.bw = bw_log
            pinfo.bw2 = bw_report
        else:
            pinfo.bw = bw_report
            pinfo.bw2 = bw_log

        self._pinfo = pinfo

        return pinfo


class IOPerfTest(PerfTest):
    tcp_conn_timeout = 30
    max_pig_timeout = 5
    soft_runcycle = 5 * 60
    retry_time = 30

    zero_md5_hash = hashlib.md5()
    zero_md5_hash.update(b"\x00" * 1024)
    zero_md5 = zero_md5_hash.hexdigest()

    def __init__(self, config):
        PerfTest.__init__(self, config)

        get = self.config.params.get
        do_get = self.config.params.__getitem__

        self.config_fname = do_get('cfg')

        if '/' not in self.config_fname and '.' not in self.config_fname:
            cfgs_dir = os.path.dirname(__file__)
            self.config_fname = os.path.join(cfgs_dir,
                                             self.config_fname + '.cfg')

        self.alive_check_interval = get('alive_check_interval')
        self.use_system_fio = get('use_system_fio', False)

        if get('prefill_files') is not None:
            logger.warning("prefill_files option is depricated. Use force_prefill instead")

        self.force_prefill = get('force_prefill', False)
        self.config_params = get('params', {}).copy()

        self.io_py_remote = self.join_remote("agent.py")
        self.results_file = self.join_remote("results.json")
        self.pid_file = self.join_remote("pid")
        self.task_file = self.join_remote("task.cfg")
        self.sh_file = self.join_remote("cmd.sh")
        self.err_out_file = self.join_remote("fio_err_out")
        self.io_log_file = self.join_remote("io_log.txt")
        self.exit_code_file = self.join_remote("exit_code")

        self.max_latency = get("max_lat", None)
        self.min_bw_per_thread = get("min_bw", None)

        self.use_sudo = get("use_sudo", True)

        self.raw_cfg = open(self.config_fname).read()
        self.fio_configs = None

    @classmethod
    def load(cls, suite_name: str, folder: str) -> IOTestResults:
        res = []
        for fname in os.listdir(folder):
            if re.match("\d+_params.yaml$", fname):
                num = int(fname.split('_')[0])
                res.append(load_test_results(folder, num))
        return IOTestResults(suite_name, res, folder)

    def cleanup(self):
        # delete_file(conn, self.io_py_remote)
        # Need to remove tempo files, used for testing
        pass

    # size is megabytes
    def check_prefill_required(self, node: INode, fname: str, size: int, num_blocks: Optional[int]=16) -> bool:
        try:
            fstats = node.stat_file(fname)
            if stat.S_ISREG(fstats.st_mode) and fstats.st_size < size * 1024 ** 2:
                return True
        except EnvironmentError:
            return True

        cmd = 'python -c "' + \
              "import sys;" + \
              "fd = open('{0}', 'rb');" + \
              "fd.seek({1});" + \
              "data = fd.read(1024); " + \
              "sys.stdout.write(data + ' ' * ( 1024 - len(data)))\" | md5sum"

        if self.use_sudo:
            cmd = "sudo " + cmd

        bsize = size * (1024 ** 2)
        offsets = [random.randrange(bsize - 1024) for _ in range(num_blocks)]
        offsets.append(bsize - 1024)
        offsets.append(0)

        for offset in offsets:
            data = node.run(cmd.format(fname, offset), nolog=True)

            md = ""
            for line in data.split("\n"):
                if "unable to resolve" not in line:
                    md = line.split()[0].strip()
                    break

            if len(md) != 32:
                logger.error("File data check is failed - " + data)
                return True

            if self.zero_md5 == md:
                return True

        return False

    def prefill_test_files(self, node: INode, files: List[str], force:bool=False) -> None:
        if self.use_system_fio:
            cmd_templ = "fio "
        else:
            cmd_templ = "{0}/fio ".format(self.config.remote_dir)

        if self.use_sudo:
            cmd_templ = "sudo " + cmd_templ

        cmd_templ += "--name=xxx --filename={0} --direct=1" + \
                     " --bs=4m --size={1}m --rw=write"

        ssize = 0

        if force:
            logger.info("File prefilling is forced")

        ddtime = 0
        for fname, curr_sz in files.items():
            if not force:
                if not self.check_prefill_required(node, fname, curr_sz):
                    logger.debug("prefill is skipped")
                    continue

            logger.info("Prefilling file {0}".format(fname))
            cmd = cmd_templ.format(fname, curr_sz)
            ssize += curr_sz

            stime = time.time()
            node.run(cmd, timeout=curr_sz)
            ddtime += time.time() - stime

        if ddtime > 1.0:
            fill_bw = int(ssize / ddtime)
            mess = "Initiall fio fill bw is {0} MiBps for this vm"
            logger.info(mess.format(fill_bw))

    def install_utils(self, node: INode) -> None:
        need_install = []
        packs = [('screen', 'screen')]
        os_info = get_os(node)

        if self.use_system_fio:
            packs.append(('fio', 'fio'))
        else:
            packs.append(('bzip2', 'bzip2'))

        for bin_name, package in packs:
            if bin_name is None:
                need_install.append(package)
                continue

            try:
                node.run('which ' + bin_name, nolog=True)
            except OSError:
                need_install.append(package)

        if len(need_install) != 0:
            if 'redhat' == os_info.distro:
                cmd = "sudo yum -y install " + " ".join(need_install)
            else:
                cmd = "sudo apt-get -y install " + " ".join(need_install)

            try:
                node.run(cmd)
            except OSError as err:
                raise OSError("Can't install - {}".format(" ".join(need_install))) from err

        if not self.use_system_fio:
            fio_dir = os.path.dirname(os.path.dirname(wally.__file__))
            fio_dir = os.path.join(os.getcwd(), fio_dir)
            fio_dir = os.path.join(fio_dir, 'fio_binaries')
            fname = 'fio_{0.release}_{0.arch}.bz2'.format(os_info)
            fio_path = os.path.join(fio_dir, fname)

            if not os.path.exists(fio_path):
                raise RuntimeError("No prebuild fio binary available for {0}".format(os_info))

            bz_dest = self.join_remote('fio.bz2')
            node.copy_file(fio_path, bz_dest)
            node.run("bzip2 --decompress {}" + bz_dest, nolog=True)
            node.run("chmod a+x " + self.join_remote("fio"), nolog=True)

    def pre_run(self) -> None:
        if 'FILESIZE' not in self.config_params:
            raise NotImplementedError("File size detection is not implemented")

        self.fio_configs = fio_cfg_compile(self.raw_cfg,
                                           self.config_fname,
                                           self.config_params)
        self.fio_configs = list(self.fio_configs)

        files = {}
        for section in self.fio_configs:
            sz = ssize2b(section.vals['size'])
            msz = sz / (1024 ** 2)

            if sz % (1024 ** 2) != 0:
                msz += 1

            fname = section.vals['filename']

            # if already has other test with the same file name
            # take largest size
            files[fname] = max(files.get(fname, 0), msz)

        with ThreadPoolExecutor(len(self.config.nodes)) as pool:
            fc = functools.partial(self.pre_run_th,
                                   files=files,
                                   force=self.force_prefill)
            list(pool.map(fc, self.config.nodes))

    def pre_run_th(self, node: INode, files: List[str], force_prefil: Optional[bool]=False) -> None:
        try:
            cmd = 'mkdir -p "{0}"'.format(self.config.remote_dir)
            if self.use_sudo:
                cmd = "sudo " + cmd
                cmd += " ; sudo chown {0} {1}".format(node.get_user(),
                                                      self.config.remote_dir)
            node.run(cmd, nolog=True)

            assert self.config.remote_dir != "" and self.config.remote_dir != "/"
            node.run("rm -rf {}/*".format(self.config.remote_dir), nolog=True)

        except Exception as exc:
            msg = "Failed to create folder {} on remote {}."
            msg = msg.format(self.config.remote_dir, node, exc)
            logger.exception(msg)
            raise StopTestError(msg) from exc

        self.install_utils(node)
        self.prefill_test_files(node, files, force_prefil)

    def show_expected_execution_time(self) -> None:
        if len(self.fio_configs) > 1:
            # +10% - is a rough estimation for additional operations
            # like sftp, etc
            exec_time = int(sum(map(execution_time, self.fio_configs)) * 1.1)
            exec_time_s = sec_to_str(exec_time)
            now_dt = datetime.datetime.now()
            end_dt = now_dt + datetime.timedelta(0, exec_time)
            msg = "Entire test should takes aroud: {0} and finished at {1}"
            logger.info(msg.format(exec_time_s,
                                   end_dt.strftime("%H:%M:%S")))

    def run(self) -> IOTestResults:
        logger.debug("Run preparation")
        self.pre_run()
        self.show_expected_execution_time()
        num_nodes = len(self.config.nodes)

        tname = os.path.basename(self.config_fname)
        if tname.endswith('.cfg'):
            tname = tname[:-4]

        barrier = Barrier(num_nodes)
        results = []

        # set of Operation_Mode_BlockSize str's
        # which should not be tested anymore, as
        # they already too slow with previous thread count
        lat_bw_limit_reached = set()

        with ThreadPoolExecutor(num_nodes) as pool:
            for pos, fio_cfg in enumerate(self.fio_configs):
                test_descr = get_test_summary(fio_cfg.vals, noqd=True)
                if test_descr in lat_bw_limit_reached:
                    continue

                logger.info("Will run {} test".format(fio_cfg.name))
                templ = "Test should takes about {}. Should finish at {}, will wait at most till {}"
                exec_time = execution_time(fio_cfg)
                exec_time_str = sec_to_str(exec_time)
                timeout = int(exec_time + max(300, exec_time))

                now_dt = datetime.datetime.now()
                end_dt = now_dt + datetime.timedelta(0, exec_time)
                wait_till = now_dt + datetime.timedelta(0, timeout)

                logger.info(templ.format(exec_time_str,
                                         end_dt.strftime("%H:%M:%S"),
                                         wait_till.strftime("%H:%M:%S")))

                run_test_func = functools.partial(self.do_run,
                                                  barrier=barrier,
                                                  fio_cfg=fio_cfg,
                                                  pos=pos)

                max_retr = 3
                for idx in range(max_retr):
                    if 0 != idx:
                        logger.info("Sleeping %ss and retrying", self.retry_time)
                        time.sleep(self.retry_time)

                    try:
                        intervals = list(pool.map(run_test_func, self.config.nodes))
                        if None not in intervals:
                            break
                    except (EnvironmentError, SSHException) as exc:
                        if max_retr - 1 == idx:
                            raise StopTestError("Fio failed") from exc
                        logger.exception("During fio run")

                fname = "{}_task.fio".format(pos)
                with open(os.path.join(self.config.log_directory, fname), "w") as fd:
                    fd.write(str(fio_cfg))

                params = {'vm_count': num_nodes}
                params['name'] = fio_cfg.name
                params['vals'] = dict(fio_cfg.vals.items())
                params['intervals'] = intervals
                params['nodes'] = [node.get_conn_id() for node in self.config.nodes]

                fname = "{}_params.yaml".format(pos)
                with open(os.path.join(self.config.log_directory, fname), "w") as fd:
                    fd.write(dumps(params))

                res = load_test_results(self.config.log_directory, pos)
                results.append(res)

                if self.max_latency is not None:
                    lat_50, _ = res.get_lat_perc_50_95_multy()

                    # conver us to ms
                    if self.max_latency < lat_50:
                        logger.info(("Will skip all subsequent tests of {} " +
                                     "due to lat/bw limits").format(fio_cfg.name))
                        lat_bw_limit_reached.add(test_descr)

                test_res = res.get_params_from_fio_report()
                if self.min_bw_per_thread is not None:
                    if self.min_bw_per_thread > average(test_res['bw']):
                        lat_bw_limit_reached.add(test_descr)

        return IOTestResults(self.config.params['cfg'],
                             results, self.config.log_directory)

    def do_run(self, node: INode, barrier: Barrier, fio_cfg, pos: int, nolog: bool=False):
        exec_folder = self.config.remote_dir

        if self.use_system_fio:
            fio_path = ""
        else:
            if not exec_folder.endswith("/"):
                fio_path = exec_folder + "/"
            else:
                fio_path = exec_folder

        exec_time = execution_time(fio_cfg)
        barrier.wait()
        run_data = node.rpc.fio.run_fio(self.use_sudo,
                                        fio_path,
                                        exec_folder,
                                        str(fio_cfg),
                                        exec_time + max(300, exec_time))
        return parse_fio_result(run_data)

    @classmethod
    def prepare_data(cls, results) -> List[Dict[str, Any]]:
        """create a table with io performance report for console"""

        def key_func(data) -> Tuple(str, str, str, str, int):
            tpl = data.summary_tpl()
            return (data.name,
                    tpl.oper,
                    tpl.mode,
                    ssize2b(tpl.bsize),
                    int(tpl.th_count) * int(tpl.vm_count))
        res = []

        for item in sorted(results, key=key_func):
            test_dinfo = item.disk_perf_info()
            testnodes_count = len(item.config.nodes)

            iops, _ = test_dinfo.iops.rounded_average_conf()

            if test_dinfo.iops_sys is not None:
                iops_sys, iops_sys_conf = test_dinfo.iops_sys.rounded_average_conf()
                _, iops_sys_dev = test_dinfo.iops_sys.rounded_average_dev()
                iops_sys_per_vm = round_3_digit(iops_sys / testnodes_count)
                iops_sys = round_3_digit(iops_sys)
            else:
                iops_sys = None
                iops_sys_per_vm = None
                iops_sys_dev = None
                iops_sys_conf = None

            bw, bw_conf = test_dinfo.bw.rounded_average_conf()
            _, bw_dev = test_dinfo.bw.rounded_average_dev()
            conf_perc = int(round(bw_conf * 100 / bw))
            dev_perc = int(round(bw_dev * 100 / bw))

            lat_50 = round_3_digit(int(test_dinfo.lat_50))
            lat_95 = round_3_digit(int(test_dinfo.lat_95))
            lat_avg = round_3_digit(int(test_dinfo.lat_avg))

            iops_per_vm = round_3_digit(iops / testnodes_count)
            bw_per_vm = round_3_digit(bw / testnodes_count)

            iops = round_3_digit(iops)
            bw = round_3_digit(bw)

            summ = "{0.oper}{0.mode} {0.bsize:>4} {0.th_count:>3}th {0.vm_count:>2}vm".format(item.summary_tpl())

            res.append({"name": key_func(item)[0],
                        "key": key_func(item)[:4],
                        "summ": summ,
                        "iops": int(iops),
                        "bw": int(bw),
                        "conf": str(conf_perc),
                        "dev": str(dev_perc),
                        "iops_per_vm": int(iops_per_vm),
                        "bw_per_vm": int(bw_per_vm),
                        "lat_50": lat_50,
                        "lat_95": lat_95,
                        "lat_avg": lat_avg,

                        "iops_sys": iops_sys,
                        "iops_sys_per_vm": iops_sys_per_vm,
                        "sys_conf": iops_sys_conf,
                        "sys_dev": iops_sys_dev})

        return res

    Field = collections.namedtuple("Field", ("header", "attr", "allign", "size"))
    fiels_and_header = [
        Field("Name",           "name",        "l",  7),
        Field("Description",    "summ",        "l", 19),
        Field("IOPS\ncum",      "iops",        "r",  3),
        # Field("IOPS_sys\ncum",  "iops_sys",    "r",  3),
        Field("KiBps\ncum",     "bw",          "r",  6),
        Field("Cnf %\n95%",     "conf",        "r",  3),
        Field("Dev%",           "dev",         "r",  3),
        Field("iops\n/vm",      "iops_per_vm", "r",  3),
        Field("KiBps\n/vm",     "bw_per_vm",   "r",  6),
        Field("lat ms\nmedian", "lat_50",      "r",  3),
        Field("lat ms\n95%",    "lat_95",      "r",  3),
        Field("lat\navg",       "lat_avg",     "r",  3),
    ]

    fiels_and_header_dct = dict((item.attr, item) for item in fiels_and_header)

    @classmethod
    def format_for_console(cls, results) -> str:
        """create a table with io performance report for console"""

        tab = texttable.Texttable(max_width=120)
        tab.set_deco(tab.HEADER | tab.VLINES | tab.BORDER)
        tab.set_cols_align([f.allign for f in cls.fiels_and_header])
        sep = ["-" * f.size for f in cls.fiels_and_header]
        tab.header([f.header for f in cls.fiels_and_header])
        prev_k = None
        for item in cls.prepare_data(results):
            if prev_k is not None:
                if prev_k != item["key"]:
                    tab.add_row(sep)

            prev_k = item["key"]
            tab.add_row([item[f.attr] for f in cls.fiels_and_header])

        return tab.draw()

    @classmethod
    def format_diff_for_console(cls, list_of_results: List[Any]) -> str:
        """create a table with io performance report for console"""

        tab = texttable.Texttable(max_width=200)
        tab.set_deco(tab.HEADER | tab.VLINES | tab.BORDER)

        header = [
            cls.fiels_and_header_dct["name"].header,
            cls.fiels_and_header_dct["summ"].header,
        ]
        allign = ["l", "l"]

        header.append("IOPS ~ Cnf% ~ Dev%")
        allign.extend(["r"] * len(list_of_results))
        header.extend(
            "IOPS_{0} %".format(i + 2) for i in range(len(list_of_results[1:]))
        )

        header.append("BW")
        allign.extend(["r"] * len(list_of_results))
        header.extend(
            "BW_{0} %".format(i + 2) for i in range(len(list_of_results[1:]))
        )

        header.append("LAT")
        allign.extend(["r"] * len(list_of_results))
        header.extend(
            "LAT_{0}".format(i + 2) for i in range(len(list_of_results[1:]))
        )

        tab.header(header)
        sep = ["-" * 3] * len(header)
        processed_results = map(cls.prepare_data, list_of_results)

        key2results = []
        for res in processed_results:
            key2results.append(dict(
                ((item["name"], item["summ"]), item) for item in res
            ))

        prev_k = None
        iops_frmt = "{0[iops]} ~ {0[conf]:>2} ~ {0[dev]:>2}"
        for item in processed_results[0]:
            if prev_k is not None:
                if prev_k != item["key"]:
                    tab.add_row(sep)

            prev_k = item["key"]

            key = (item['name'], item['summ'])
            line = list(key)
            base = key2results[0][key]

            line.append(iops_frmt.format(base))

            for test_results in key2results[1:]:
                val = test_results.get(key)
                if val is None:
                    line.append("-")
                elif base['iops'] == 0:
                    line.append("Nan")
                else:
                    prc_val = {'dev': val['dev'], 'conf': val['conf']}
                    prc_val['iops'] = int(100 * val['iops'] / base['iops'])
                    line.append(iops_frmt.format(prc_val))

            line.append(base['bw'])

            for test_results in key2results[1:]:
                val = test_results.get(key)
                if val is None:
                    line.append("-")
                elif base['bw'] == 0:
                    line.append("Nan")
                else:
                    line.append(int(100 * val['bw'] / base['bw']))

            for test_results in key2results:
                val = test_results.get(key)
                if val is None:
                    line.append("-")
                else:
                    line.append("{0[lat_50]} - {0[lat_95]}".format(val))

            tab.add_row(line)

        tab.set_cols_align(allign)
        return tab.draw()