STORAGE_ROLES = ['ceph-osd'] class Stage(metaclass=abc.ABCMeta):
    priority: int = None  # type: ignore
    config_block: Optional[str] = None

    @classmethod
    def name(cls) -> str:
        return cls.__name__

    @classmethod
    def validate_config(cls, cfg: ConfigBlock) -> None:
        pass

    @abc.abstractmethod
    def run(self, ctx: TestRun) -> None:
        pass

    def cleanup(self, ctx: TestRun) -> None:
        pass class StepOrder:
    DISCOVER = 0
    SPAWN = 10
    CONNECT = 20
    UPDATE_NODES_INFO = 30
    START_SENSORS = 40
    TEST = 50
    COLLECT_SENSORS = 60
    STOP_SENSORS = 70
    REPORT = 80 class TestRun:
    """Test run information"""
    def __init__(self, config: Config, storage: IStorage, rstorage: IWallyStorage) -> None:
        # NodesInfo list
        self.nodes_info: Dict[str, NodeInfo] = {}

        self.ceph_master_node: Optional[IRPCNode] = None
        self.ceph_extra_args: Optional[str] = None

        # Nodes list
        self.nodes: List[IRPCNode] = []

        self.build_meta: Dict[str,Any] = {}
        self.clear_calls_stack: List[Callable[['TestRun'], None]] = []

        # openstack credentials
        self.os_creds: Optional[OSCreds] = None  # type: ignore
        self.os_connection: Optional[OSConnection] = None  # type: ignore
        self.rpc_code: bytes = None  # type: ignore
        self.default_rpc_plugins: Dict[str, bytes] = None  # type: ignore

        self.storage = storage
        self.rstorage = rstorage
        self.config = config
        self.sensors_run_on: Set[str] = set()
        self.os_spawned_nodes_ids: List[int] = None  # type: ignore
        self.devs_locator: DevRolesConfig = []

    def get_pool(self):
        return ThreadPoolExecutor(self.config.get('worker_pool_sz', 32))

    def merge_node(self, creds: ConnCreds, roles: Set[str], **params) -> NodeInfo:
        info = NodeInfo(creds, roles, params)
        nid = info.node_id

        if nid in self.nodes_info:
            self.nodes_info[nid].roles.update(info.roles)
            self.nodes_info[nid].params.update(info.params)
            return self.nodes_info[nid]
        else:
            self.nodes_info[nid] = info
            return info class IWallyStorage(ISensorStorage, IImagesStorage, metaclass=abc.ABCMeta):

    @abc.abstractmethod
    def flush(self) -> None:
        pass

    @abc.abstractmethod
    def put_or_check_suite(self, suite: SuiteConfig) -> None:
        pass

    @abc.abstractmethod
    def put_job(self, suite: SuiteConfig, job: JobConfig) -> None:
        pass

    @abc.abstractmethod
    def put_extra(self, data: bytes, source: DataSource) -> None:
        pass

    @abc.abstractmethod
    def put_stat(self, data: StatProps, source: DataSource) -> None:
        pass

    @abc.abstractmethod
    def get_stat(self, stat_cls: Type[StatProps], source: DataSource) -> StatProps:
        pass

    @abc.abstractmethod
    def iter_suite(self, suite_type: str = None) -> Iterator[SuiteConfig]:
        pass

    @abc.abstractmethod
    def iter_job(self, suite: SuiteConfig) -> Iterator[JobConfig]:
        pass

    # return path to file to be inserted into report
    @abc.abstractmethod
    def put_plot_file(self, data: bytes, source: DataSource) -> str:
        pass

    @abc.abstractmethod
    def get_job_info(self, suite: SuiteConfig, job: JobConfig, key: str) -> Any:
        pass

    @abc.abstractmethod
    def get_ts(self, ds: DataSource) -> TimeSeries:
        pass

    @abc.abstractmethod
    def put_ts(self, ts: TimeSeries) -> None:
        pass

    @abc.abstractmethod
    def iter_ts(self, **ds_parts) -> Iterator[DataSource]:
        pass

    @abc.abstractmethod
    def put_job_info(self, suite: SuiteConfig, job: JobConfig, key: str, data: Any) -> None:
        pass

    @abc.abstractmethod
    def load_nodes(self) -> List[NodeInfo]:
        pass class SuiteConfig(Storable):
class IWallyStorage(ISensorStorage, IImagesStorage, metaclass=abc.ABCMeta):
    def __init__(self,
                 test_type: str,
                 params: Dict[str, Any],
                 run_uuid: str,
                 nodes: List[IRPCNode],
                 remote_dir: str,
                 idx: int,
                 keep_raw_files: bool) -> None:
    def __eq__(self, o: object) -> bool:
    def flush(self) -> None:
    def put_or_check_suite(self, suite: SuiteConfig) -> None:
    def put_job(self, suite: SuiteConfig, job: JobConfig) -> None:
    def put_extra(self, data: bytes, source: DataSource) -> None:
    def put_stat(self, data: StatProps, source: DataSource) -> None:
    def get_stat(self, stat_cls: Type[StatProps], source: DataSource) -> StatProps:
    def iter_suite(self, suite_type: str = None) -> Iterator[SuiteConfig]:
    def iter_job(self, suite: SuiteConfig) -> Iterator[JobConfig]:
    def put_plot_file(self, data: bytes, source: DataSource) -> str:
    def get_job_info(self, suite: SuiteConfig, job: JobConfig, key: str) -> Any:
    def get_ts(self, ds: DataSource) -> TimeSeries:
    def put_ts(self, ts: TimeSeries) -> None:
    def iter_ts(self, **ds_parts) -> Iterator[DataSource]:
    def put_job_info(self, suite: SuiteConfig, job: JobConfig, key: str, data: Any) -> None:
    def load_nodes(self) -> List[NodeInfo]: class FioTest(ThreadedTest):
    def __init__(self, *args, **kwargs) -> None:
    def config_node(self, node: IRPCNode) -> None:
    def install_utils(self, node: IRPCNode) -> None:
    def get_expected_runtime(self, job_config: JobConfig) -> int:
    def prepare_iteration(self, node: IRPCNode, job: JobConfig) -> None:
    def run_iteration(self, node: IRPCNode, job: JobConfig) -> List[TimeSeries]:
    def format_for_console(self, data: Any) -> str: class FioJobParams(JobParams):
    """Class contains all parameters, which significantly affects fio results.

        oper - operation type - read/write/randread/...
        sync_mode - direct/sync/async/direct+sync
        bsize - block size in KiB
        qd - IO queue depth,
        thcount - thread count,
        write_perc - write perc for mixed(read+write) loads

    Like block size or operation type, but not file name or file size.
    Can be used as key in dictionary.
    """

    sync2long = {'x': "sync direct",
                 's': "sync",
                 'd': "direct",
                 'a': "buffered"}

    @property
    def sync_mode_long(self) -> str:
        return self.sync2long[self['sync_mode']]

    @property
    def summary(self) -> str:
        """Test short summary, used mostly for file names and short image description"""
        res = f"{self['oper_short']}{self['sync_mode']}{self['bsize']}"
        if self['qd'] is not None:
            res += "_qd" + str(self['qd'])
        if self['thcount'] not in (1, None):
            res += "th" + str(self['thcount'])
        if self['write_perc'] is not None:
            res += "wr" + str(self['write_perc'])
        return res

    @property
    def long_summary(self) -> str:
        """Readable long summary for management and deployment engineers"""
        res = f"{self['oper']}, {self.sync_mode_long}, block size {b2ssize(self['bsize'] * 1024)}B"
        if self['qd'] is not None:
            res += ", QD = " + str(self['qd'])
        if self['thcount'] not in (1, None):
            res += f", threads={self['thcount']}"
        if self['write_perc'] is not None:
            res += f", fwrite_perc={self['write_perc']}%"
        return res

    def copy(self, **kwargs: Dict[str, Any]) -> 'FioJobParams':
        np = self.params.copy()
        np.update(kwargs)
        return self.__class__(**np)

    @property
    def char_tpl(self) -> Tuple[Union[str, int], ...]:
        mint = lambda x: -10000000000 if x is None else int(x)
        return self['oper'], mint(self['bsize']), self['sync_mode'], \
            mint(self['thcount']), mint(self['qd']), mint(self['write_perc']) class JobConfig(Storable, metaclass=abc.ABCMeta):
    """Job config class"""

    def __init__(self, idx: int) -> None:
        # job id, used in storage to distinct jobs with same summary
        self.idx = idx

        # time interval, in seconds, when test was running on all nodes
        self.reliable_info_range: Tuple[int, int] = None  # type: ignore

        # all job parameters, both from suite file and config file
        self.vals: Dict[str, Any] = OrderedDict()

    @property
    def reliable_info_range_s(self) -> Tuple[int, int]:
        return (self.reliable_info_range[0] // 1000, self.reliable_info_range[1] // 1000)

    @property
    def storage_id(self) -> str:
        """unique string, used as key in storage"""
        return f"{self.summary}_{self.idx}"

    @property
    @abc.abstractmethod
    def params(self) -> JobParams:
        """Should return a copy"""
        pass

    @property
    def summary(self) -> str:
        return self.params.summary def get_aggregated(rstorage: IWallyStorage, suite_id: str, job_id: str, metric: str,
                   trange: Tuple[int, int]) -> TimeSeries:
    "Sum selected fio metric for all nodes for given Suite/job"

    key = (id(rstorage), suite_id, job_id, metric, trange)
    aggregated_cache = rstorage.storage.other_caches['aggregated']
    if key in aggregated_cache:
        return aggregated_cache[key].copy()

    tss = list(find_all_series(rstorage, suite_id, job_id, metric))

    if len(tss) == 0:
        raise NameError(f"Can't found any TS for {suite_id},{job_id},{metric}")

    c_intp = c_interpolate_ts_on_seconds_border
    tss_inp = [c_intp(ts.select(trange), tp='fio', allow_broken_step=(metric == 'lat')) for ts in tss]

    res = None
    res_times = None

    for ts, ts_orig in zip(tss_inp, tss):
        if ts.time_units != 's':
            msg = "time_units must be 's' for fio sensor"
            logger.error(msg)
            raise ValueError(msg)

        # if metric == 'lat' and (len(ts.data.shape) != 2 or ts.data.shape[1] != expected_lat_bins):
        #     msg = f"Sensor {ts.source.dev}.{ts.source.sensor} on node {ts.source.node_id} " + \
        #         f"has shape={ts.data.shape}. Can only process sensors with shape=[X, {expected_lat_bins}]."
        #     logger.error(msg)
        #     raise ValueError(msg)

        if metric != 'lat' and len(ts.data.shape) != 1:
            msg = f"Sensor {ts.source.dev}.{ts.source.sensor} on node {ts.source.node_id} " + \
                f"has shape={ts.data.shape}. Can only process 1D sensors."
            logger.error(msg)
            raise ValueError(msg)

        assert trange[0] >= ts.times[0] and trange[1] <= ts.times[-1], \
            f"[{ts.times[0]}, {ts.times[-1]}] not in [{trange[0]}, {trange[-1]}]"

        idx1, idx2 = numpy.searchsorted(ts.times, trange)
        idx2 += 1

        assert (idx2 - idx1) == (trange[1] - trange[0] + 1), \
            "Broken time array at {} for {}".format(trange, ts.source)

        dt = ts.data[idx1: idx2]
        if res is None:
            res = dt.copy()
            res_times = ts.times[idx1: idx2].copy()
        else:
            assert res.shape == dt.shape, f"res.shape(={res.shape}) != dt.shape(={dt.shape})"
            res += dt

    ds = DataSource(suite_id=suite_id, job_id=job_id, node_id=AGG_TAG, sensor='fio',
                    dev=AGG_TAG, metric=metric, tag='csv')
    agg_ts = TimeSeries(res, source=ds,
                        times=res_times,
                        units=tss_inp[0].units,
                        histo_bins=tss_inp[0].histo_bins,
                        time_units=tss_inp[0].time_units)
    aggregated_cache[key] = agg_ts
    return agg_ts.copy() AGG_TAG = 'ALL' class ColorProfile:
class StyleProfile: @provide_plot(eng=False, no_legend=True, grid='y', style_name='ioqd', noadjust=True)
def io_chart(pp: PlotParams,
             legend: str,
             iosums: List[IOSummary],
             iops_log_spine: bool = False,
             lat_log_spine: bool = False) -> None:

    # --------------  MAGIC VALUES  ---------------------
    # IOPS bar width
    width = 0.2

    # offset from center of bar to deviation/confidence range indicator
    err_x_offset = 0.03

    # extra space on top and bottom, comparing to maximal tight layout
    extra_y_space = 0.05

    # additional spine for BW/IOPS on left side of plot
    extra_io_spine_x_offset = -0.1

    # extra space on left and right sides
    extra_x_space = 0.5

    # legend location settings
    legend_location = "center left"
    legend_bbox_to_anchor = (1.1, 0.81)

    # --------------  END OF MAGIC VALUES  ---------------------

    block_size = iosums[0].block_size
    xpos = numpy.arange(1, len(iosums) + 1, dtype='uint')

    coef_mb = unit_conversion_coef_f(iosums[0].bw.units, "MiBps")
    coef_iops = unit_conversion_coef_f(iosums[0].bw.units, "KiBps") / block_size

    iops_primary = block_size < pp.style.large_blocks

    coef = coef_iops if iops_primary else coef_mb
    pp.ax.set_ylabel("IOPS" if iops_primary else "BW (MiBps)")

    vals = [iosum.bw.average * coef for iosum in iosums]

    # set correct x limits for primary IO spine
    min_io = min(iosum.bw.average - iosum.bw.deviation * pp.style.dev_range_x for iosum in iosums)
    max_io = max(iosum.bw.average + iosum.bw.deviation * pp.style.dev_range_x for iosum in iosums)
    border = (max_io - min_io) * extra_y_space
    io_lims = (min_io - border, max_io + border)

    pp.ax.set_ylim(io_lims[0] * coef, io_lims[-1] * coef)
    pp.ax.bar(xpos - width / 2, vals, width=width, color=pp.colors.box_color, label=legend)

    # plot deviation and confidence error ranges
    err1_legend = err2_legend = None
    for pos, iosum in zip(xpos, iosums):
        dev_bar_pos = pos - err_x_offset
        err1_legend = pp.ax.errorbar(dev_bar_pos,
                                     iosum.bw.average * coef,
                                     iosum.bw.deviation * pp.style.dev_range_x * coef,
                                     alpha=pp.colors.subinfo_alpha,
                                     color=pp.colors.suppl_color1)  # 'magenta'

        conf_bar_pos = pos + err_x_offset
        err2_legend = pp.ax.errorbar(conf_bar_pos,
                                     iosum.bw.average * coef,
                                     iosum.bw.confidence * coef,
                                     alpha=pp.colors.subinfo_alpha,
                                     color=pp.colors.suppl_color2)  # 'teal'

    handles1, labels1 = pp.ax.get_legend_handles_labels()

    handles1 += [err1_legend, err2_legend]
    labels1 += ["{}% dev".format(pp.style.dev_perc),
                "{}% conf".format(int(100 * iosums[0].bw.confidence_level))]

    # extra y spine for latency on right side
    ax2 = pp.ax.twinx()

    # plot median and 95 perc latency
    lat_coef_ms = unit_conversion_coef_f(iosums[0].lat.units, "ms")
    ax2.plot(xpos, [iosum.lat.perc_50 * lat_coef_ms for iosum in iosums], label="lat med")
    ax2.plot(xpos, [iosum.lat.perc_95 * lat_coef_ms for iosum in iosums], label="lat 95%")

    for grid_line in ax2.get_ygridlines():
        grid_line.set_linestyle(":")

    # extra y spine for BW/IOPS on left side
    if pp.style.extra_io_spine:
        ax3 = pp.ax.twinx()
        if iops_log_spine:
            ax3.set_yscale('log')

        ax3.set_ylabel("BW (MiBps)" if iops_primary else "IOPS")
        secondary_coef = coef_mb if iops_primary else coef_iops
        ax3.set_ylim(io_lims[0] * secondary_coef, io_lims[1] * secondary_coef)
        ax3.spines["left"].set_position(("axes", extra_io_spine_x_offset))
        ax3.spines["left"].set_visible(True)
        ax3.yaxis.set_label_position('left')
        ax3.yaxis.set_ticks_position('left')
    else:
        ax3 = None

    ax2.set_ylabel("Latency (ms)")

    # legend box
    handles2, labels2 = ax2.get_legend_handles_labels()
    pp.ax.legend(handles1 + handles2, labels1 + labels2, loc=legend_location, bbox_to_anchor=legend_bbox_to_anchor)

    # limit and label x spine
    pp.ax.set_xlim(extra_x_space, len(iosums) + extra_x_space)
    pp.ax.set_ylim(bottom=0)
    pp.ax.set_xticks(xpos)
    pp.ax.set_xticklabels(["{0}*{1}={2}".format(iosum.qd, iosum.nodes_count, iosum.qd * iosum.nodes_count)
                          for iosum in iosums],
                          rotation=30 if len(iosums) > 9 else 0)
    pp.ax.set_xlabel("IO queue depth * test node count = total parallel requests")

    # apply log scales for X spines, if set
    if iops_log_spine:
        pp.ax.set_yscale('log')

    if lat_log_spine:
        ax2.set_yscale('log')

    # override some styles
    pp.fig.set_size_inches(*pp.style.qd_chart_inches)
    pp.fig.subplots_adjust(right=pp.style.subplot_adjust_r)

    if pp.style.extra_io_spine:
        ax3.grid(False) class ResourceNames:
    io_made = "Client IOP made"
    data_tr = "Client data transfered"

    test_send = "Test nodes net send"
    test_recv = "Test nodes net recv"
    test_net = "Test nodes net total"
    test_send_pkt = "Test nodes send pkt"
    test_recv_pkt = "Test nodes recv pkt"
    test_net_pkt = "Test nodes total pkt"

    test_write = "Test nodes disk write"
    test_read = "Test nodes disk read"
    test_write_iop = "Test nodes write IOP"
    test_read_iop = "Test nodes read IOP"
    test_iop = "Test nodes IOP"
    test_rw = "Test nodes disk IO"

    storage_send = "Storage nodes net send"
    storage_recv = "Storage nodes net recv"
    storage_send_pkt = "Storage nodes send pkt"
    storage_recv_pkt = "Storage nodes recv pkt"
    storage_net = "Storage nodes net total"
    storage_net_pkt = "Storage nodes total pkt"

    storage_write = "Storage nodes disk write"
    storage_read = "Storage nodes disk read"
    storage_write_iop = "Storage nodes write IOP"
    storage_read_iop = "Storage nodes read IOP"
    storage_iop = "Storage nodes IOP"
    storage_rw = "Storage nodes disk IO"

    storage_cpu = "Storage nodes CPU"
    storage_cpu_s = "Storage nodes CPU s/IOP"
    storage_cpu_s_b = "Storage nodes CPU s/B" def get_resources_usage(suite: SuiteConfig,
                        job: JobConfig,
                        rstorage: IWallyStorage,
                        large_block: int = 256,
                        hist_boxes: int = 10,
                        nc: bool = False) -> Tuple[Dict[str, Tuple[str, Optional[float], Optional[float]]], bool]:

    if not nc:
        jinfo = rstorage.get_job_info(suite, job, WallyDB.resource_usage_rel)
        if jinfo is not None:
            jinfo = jinfo.copy()
            return jinfo, jinfo.pop('iops_ok')  # type: ignore

    fjob = cast(FioJobConfig, job)
    iops_ok = fjob.bsize < large_block

    io_sum = make_iosum(rstorage, suite, fjob, hist_boxes)

    tot_io_coef = unit_conversion_coef_f(io_sum.bw.units, "Bps")
    io_transfered = io_sum.bw.data * tot_io_coef

    records: Dict[str, Tuple[str, Optional[float], Optional[float]]] = {
        ResourceNames.data_tr: (b2ssize(io_transfered.sum()) + "B", None, None)
    }

    if iops_ok:
        ops_done = io_transfered / (fjob.bsize * unit_conversion_coef_f("KiBps", "Bps"))
        records[ResourceNames.io_made] = (b2ssize_10(ops_done.sum()) + "OP", None, None)
    else:
        ops_done = None

    all_metrics = [
        (ResourceNames.test_send, 'net-io', 'send_bytes', b2ssize, ['testnode'], "B", io_transfered),
        (ResourceNames.test_recv, 'net-io', 'recv_bytes', b2ssize, ['testnode'], "B", io_transfered),
        (ResourceNames.test_send_pkt, 'net-io', 'send_packets', b2ssize_10, ['testnode'], "pkt", ops_done),
        (ResourceNames.test_recv_pkt, 'net-io', 'recv_packets', b2ssize_10, ['testnode'], "pkt", ops_done),

        (ResourceNames.test_write, 'block-io', 'sectors_written', b2ssize, ['testnode'], "B", io_transfered),
        (ResourceNames.test_read, 'block-io', 'sectors_read', b2ssize, ['testnode'], "B", io_transfered),
        (ResourceNames.test_write_iop, 'block-io', 'writes_completed', b2ssize_10, ['testnode'], "OP", ops_done),
        (ResourceNames.test_read_iop, 'block-io', 'reads_completed', b2ssize_10, ['testnode'], "OP", ops_done),

        (ResourceNames.storage_send, 'net-io', 'send_bytes', b2ssize, STORAGE_ROLES, "B", io_transfered),
        (ResourceNames.storage_recv, 'net-io', 'recv_bytes', b2ssize, STORAGE_ROLES, "B", io_transfered),
        (ResourceNames.storage_send_pkt, 'net-io', 'send_packets', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
        (ResourceNames.storage_recv_pkt, 'net-io', 'recv_packets', b2ssize_10, STORAGE_ROLES, "OP", ops_done),

        (ResourceNames.storage_write, 'block-io', 'sectors_written', b2ssize, STORAGE_ROLES, "B", io_transfered),
        (ResourceNames.storage_read, 'block-io', 'sectors_read', b2ssize, STORAGE_ROLES, "B", io_transfered),
        (ResourceNames.storage_write_iop, 'block-io', 'writes_completed', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
        (ResourceNames.storage_read_iop, 'block-io', 'reads_completed', b2ssize_10, STORAGE_ROLES, "OP", ops_done),
    ]

    all_agg = {}

    for vname, sensor, metric, ffunc, roles, units, service_provided_count in all_metrics:
        if service_provided_count is None:
            continue

        nodes = find_nodes_by_roles(rstorage.storage, roles)
        res_ts = sum_sensors(rstorage, job.reliable_info_range_s, node_id=nodes, sensor=sensor, metric=metric)
        if res_ts is None:
            continue

        data = res_ts.data
        if units == "B":
            data = data * unit_conversion_coef_f(res_ts.units, "B")

        avg, dev = avg_dev_div(data, service_provided_count)
        if avg < 0.1:
            dev = None # type: ignore
        records[vname] = (ffunc(data.sum()) + units, avg, dev)
        all_agg[vname] = data

    # cpu usage
    stor_cores_count = None
    for node in rstorage.load_nodes():
        if node.roles.intersection(STORAGE_ROLES):
            if stor_cores_count is None:
                stor_cores_count = sum(cores for _, cores in node.hw_info.cpus)
            else:
                assert stor_cores_count == sum(cores for _, cores in node.hw_info.cpus)

    assert stor_cores_count != 0

    cpu_ts = get_cluster_cpu_load(rstorage, STORAGE_ROLES, job.reliable_info_range_s)
    cpus_used_sec = (1.0 - (cpu_ts['idle'].data + cpu_ts['iowait'].data) / cpu_ts['total'].data) * stor_cores_count
    used_s = b2ssize_10(cpus_used_sec.sum()) + 's'

    all_agg[ResourceNames.storage_cpu] = cpus_used_sec

    if ops_done is not None:
        records[ResourceNames.storage_cpu_s] = (used_s, *avg_dev_div(cpus_used_sec, ops_done))

    records[ResourceNames.storage_cpu_s_b] = (used_s, *avg_dev_div(cpus_used_sec, io_transfered))

    cums = [
        (ResourceNames.test_iop, ResourceNames.test_read_iop, ResourceNames.test_write_iop,
         b2ssize_10, "OP", ops_done),
        (ResourceNames.test_rw, ResourceNames.test_read, ResourceNames.test_write, b2ssize, "B", io_transfered),
        (ResourceNames.test_net, ResourceNames.test_send, ResourceNames.test_recv, b2ssize, "B", io_transfered),
        (ResourceNames.test_net_pkt, ResourceNames.test_send_pkt, ResourceNames.test_recv_pkt, b2ssize_10,
         "pkt", ops_done),

        (ResourceNames.storage_iop, ResourceNames.storage_read_iop, ResourceNames.storage_write_iop, b2ssize_10,
         "OP", ops_done),
        (ResourceNames.storage_rw, ResourceNames.storage_read, ResourceNames.storage_write, b2ssize, "B",
         io_transfered),
        (ResourceNames.storage_net, ResourceNames.storage_send, ResourceNames.storage_recv, b2ssize, "B",
         io_transfered),
        (ResourceNames.storage_net_pkt, ResourceNames.storage_send_pkt, ResourceNames.storage_recv_pkt, b2ssize_10,
         "pkt", ops_done),
    ]

    for vname, name1, name2, ffunc, units, service_provided_masked in cums:
        if service_provided_masked is None:
            continue
        if name1 in all_agg and name2 in all_agg:
            agg = all_agg[name1] + all_agg[name2]
            avg, dev = avg_dev_div(agg, service_provided_masked)
            if avg < 0.1:
                dev = None  # type: ignore
            records[vname] = (ffunc(agg.sum()) + units, avg, dev)

    if not nc:
        toflt = lambda x: float(x) if x is not None else None

        for name, (v1, v2, v3) in list(records.items()):
            records[name] = v1, toflt(v2), toflt(v3)

        srecords = records.copy()
        srecords['iops_ok'] = iops_ok  # type: ignore
        rstorage.put_job_info(suite, job, WallyDB.resource_usage_rel, srecords)

    return records, iops_ok def make_iosum(rstorage: IWallyStorage, suite: SuiteConfig, job: FioJobConfig, hist_boxes: int,
               nc: bool = False) -> IOSummary:

    key = (suite.storage_id, job.storage_id)
    if not nc and key in iosum_cache:
        return iosum_cache[key]

    lat = get_aggregated(rstorage, suite.storage_id, job.storage_id, "lat", job.reliable_info_range_s)
    io = get_aggregated(rstorage, suite.storage_id, job.storage_id, "bw", job.reliable_info_range_s)

    res = IOSummary(job.qd,
                    nodes_count=len(suite.nodes_ids),
                    block_size=job.bsize,
                    lat=calc_histo_stat_props(lat, rebins_count=hist_boxes),
                    bw=calc_norm_stat_props(io, hist_boxes))

    if not nc:
        iosum_cache[key] = res

    return res def get_cluster_cpu_load(rstorage: IWallyStorage, roles: List[str],
                         time_range: Tuple[int, int], nc: bool = False) -> Dict[str, TimeSeries]:

    key = (id(rstorage), tuple(roles), time_range)
    if not nc and key in cpu_load_cache:
        return cpu_load_cache[key]

    cpu_ts = {}
    cpu_metrics = "idle guest iowait sirq nice irq steal sys user".split()
    nodes = find_nodes_by_roles(rstorage.storage, roles)

    cores_per_node = {}
    for node in rstorage.load_nodes():
        cores_per_node[node.node_id] = 48 # sum(cores for _, cores in node.hw_info.cpus)

    for name in cpu_metrics:
        cpu_ts[name] = sum_sensors(rstorage, time_range, node_id=nodes, sensor='system-cpu', metric=name)

    it = iter(cpu_ts.values())
    total_over_time = next(it).data.copy()  # type: numpy.ndarray
    for ts in it:
        if ts is not None:
            total_over_time += ts.data

    total = cpu_ts['idle'].copy(no_data=True)
    total.data = total_over_time
    cpu_ts['total'] = total

    if not nc:
        cpu_load_cache[key] = cpu_ts

    return cpu_ts def get_console_report_table(suite: SuiteConfig, rstorage: IWallyStorage) -> List[Union[List[str], Texttable.HLINE]]:
    def run(self, ctx: TestRun) -> None:
class ConsoleReportStage(Stage):
import os
import abc
import logging
import collections
import numpy
import scipy.stats
import xmlbuilder3
import wally
from collections import defaultdict
from typing import Dict, Any, Iterator, Tuple, cast, List, Set, Optional, Union, Type, Iterable
from statsmodels.tsa.stattools import adfuller
from cephlib import html
from cephlib.units import b2ssize, b2ssize_10, unit_conversion_coef, unit_conversion_coef_f
from cephlib.statistic import calc_norm_stat_props
from cephlib.storage_selectors import sum_sensors, find_sensors_to_2d, update_storage_selector, DevRoles
from cephlib.wally_storage import find_nodes_by_roles
from cephlib.plot import (plot_simple_bars, plot_hmap_from_2d, plot_lat_over_time, plot_simple_over_time,
                          plot_histo_heatmap, plot_v_over_time, plot_hist, plot_dots_with_regression)
from cephlib.numeric_types import ndarray2d
from cephlib.node import NodeRole
from .utils import STORAGE_ROLES
from .stage import Stage, StepOrder
from .test_run_class import TestRun
from .result_classes import IWallyStorage
from .result_classes import DataSource, TimeSeries, SuiteConfig
from .suits.io.fio import FioTest, FioJobConfig
from .suits.io.fio_job import FioJobParams
from .suits.job import JobConfig
from .data_selectors import get_aggregated, AGG_TAG
from .report_profiles import (DefStyleProfile, DefColorProfile, StyleProfile, ColorProfile,
                              default_format, io_chart_format)
from .plot import io_chart
from .resources import ResourceNames, get_resources_usage, make_iosum, get_cluster_cpu_load
from .console_report import get_console_report_table, console_report_headers, console_report_align, Texttable
    return psum50, psum95


# Main performance report
class PerformanceSummary(SuiteReporter):
    """Aggregated summary for storage"""
    def get_divs(self, suite: SuiteConfig) -> Iterator[Tuple[str, str, HTMLBlock]]:
        psum50, psum95 = get_performance_summary(self.rstorage, suite, self.style.hist_boxes, self.style.large_blocks)

        caption = "Storage summary report"
        res = html.H3(html.center(caption))

        headers = ["Mode", "Stats", "Explanation"]
        align = ['left', 'right', "left"]
        data: List[Union[str, Tuple[str, str, str]]] = []

        if psum95.rr_iops_10ms != psum95.NO_VAL or psum95.rr_iops_30ms != psum95.NO_VAL or \
                psum95.rr_iops_100ms != psum95.NO_VAL:
            data.append("Average random read IOPS for small blocks")

        if psum95.rr_iops_10ms != psum95.NO_VAL:
            data.append(("Database", b2ssize_10(psum95.rr_iops_10ms), "Latency 95th percentile < 10ms"))
        if psum95.rr_iops_30ms != psum95.NO_VAL:
            data.append(("File system", b2ssize_10(psum95.rr_iops_30ms), "Latency 95th percentile < 30ms"))
        if psum95.rr_iops_100ms != psum95.NO_VAL:
            data.append(("File server", b2ssize_10(psum95.rr_iops_100ms), "Latency 95th percentile < 100ms"))

        if psum95.rw_iops_10ms != psum95.NO_VAL or psum95.rw_iops_30ms != psum95.NO_VAL or \
                psum95.rw_iops_100ms != psum95.NO_VAL:
            data.append("Average random write IOPS for small blocks")

        if psum95.rw_iops_10ms != psum95.NO_VAL:
            data.append(("Database", b2ssize_10(psum95.rw_iops_10ms), "Latency 95th percentile < 10ms"))
        if psum95.rw_iops_30ms != psum95.NO_VAL:
            data.append(("File system", b2ssize_10(psum95.rw_iops_30ms), "Latency 95th percentile < 30ms"))
        if psum95.rw_iops_100ms != psum95.NO_VAL:
            data.append(("File server", b2ssize_10(psum95.rw_iops_100ms), "Latency 95th percentile < 100ms"))

        if psum50.bw_write_max != psum50.NO_VAL or psum50.bw_read_max != psum50.NO_VAL:
            data.append("Average sequention IO")

        if psum50.bw_write_max != psum95.NO_VAL:
            data.append(("Write", b2ssize(psum50.bw_write_max) + psum50.bw_units,
                         "Large blocks (>={}KiB)".format(self.style.large_blocks)))
        if psum50.bw_read_max != psum95.NO_VAL:
            data.append(("Read", b2ssize(psum50.bw_read_max) + psum50.bw_units,
                         "Large blocks (>={}KiB)".format(self.style.large_blocks)))

        if data:
            res += html.center(html.table("Performance", headers, data, align=align))
            yield Menu1st.summary, Menu2ndSumm.summary, HTMLBlock(res)


# # Node load over test time
# class NodeLoad(SuiteReporter):
#     """IOPS/latency during test"""

# # Ceph operation breakout report
# class CephClusterSummary(SuiteReporter):


# Main performance report
class IOQD(SuiteReporter):
    """Creates graph, which show how IOPS and Latency depend on QD"""
    suite_types = {'fio'}

    def get_divs(self, suite: SuiteConfig) -> Iterator[Tuple[str, str, HTMLBlock]]:
        ts_map: Dict[FioJobParams, List[Tuple[SuiteConfig, FioJobConfig]]] = defaultdict(list)
        str_summary: Dict[FioJobParams, Tuple[str, str]] = {}

        for job in self.rstorage.iter_job(suite):
            fjob = cast(FioJobConfig, job)
            fjob_no_qd = cast(FioJobParams, fjob.params.copy(qd=None))
            str_summary[fjob_no_qd] = (fjob_no_qd.summary, fjob_no_qd.long_summary)
            ts_map[fjob_no_qd].append((suite, fjob))

        caption = "IOPS, bandwith, and latency as function of parallel IO request count (QD)"
        yield Menu1st.summary, Menu2ndSumm.io_lat_qd, HTMLBlock(html.H3(html.center(caption)))

        for tpl, suites_jobs in ts_map.items():
            if len(suites_jobs) >= self.style.min_iops_vs_qd_jobs:
                iosums = [make_iosum(self.rstorage, suite, job, self.style.hist_boxes) for suite, job in suites_jobs]
                iosums.sort(key=lambda x: x.qd)
                summary, summary_long = str_summary[tpl]

                ds = DataSource(suite_id=suite.storage_id,
                                job_id=summary,
                                node_id=AGG_TAG,
                                sensor="fio",
                                dev=AGG_TAG,
                                metric="io_over_qd",
                                tag=io_chart_format)

                fpath = self.plt(io_chart, ds, title=summary_long, legend="IOPS/BW", iosums=iosums)
                yield Menu1st.summary, Menu2ndSumm.io_lat_qd, HTMLBlock(html.img(fpath))


class ResourceQD(SuiteReporter):
    suite_types = {'fio'}

    def get_divs(self, suite: SuiteConfig) -> Iterator[Tuple[str, str, HTMLBlock]]:
        qd_grouped_jobs: Dict[FioJobParams, List[FioJobConfig]] = {}
        test_nc = len(list(find_nodes_by_roles(self.rstorage.storage, ['testnode'])))
        for job in self.rstorage.iter_job(suite):
            fjob = cast(FioJobConfig, job)
            if fjob.bsize != 4:
                continue

            fjob_no_qd = cast(FioJobParams, fjob.params.copy(qd=None))
            qd_grouped_jobs.setdefault(fjob_no_qd, []).append(fjob)

        yield Menu1st.summary, Menu2ndSumm.resources_usage_qd, HTMLBlock(html.center(html.H3("Resource usage summary")))

        for jc_no_qd, jobs in sorted(qd_grouped_jobs.items()):
            cpu_usage2qd = {}
            for job in jobs:
                usage, iops_ok = get_resources_usage(suite, job, self.rstorage, hist_boxes=self.style.hist_boxes,
                                                     large_block=self.style.large_blocks)

                if iops_ok:

                    cpu_usage2qd[job.qd] = usage[ResourceNames.storage_cpu_s]