cfg_checker/modules/ceph/__init__.py

#    Author: Alex Savatieiev (osavatieiev@mirantis.com; a.savex@gmail.com)
#    Copyright 2019-2022 Mirantis, Inc.
import os

from cfg_checker.agent.fio_runner import get_fio_options
from cfg_checker.agent.fio_runner import seq_modes, mix_modes, rand_modes
from cfg_checker.common import logger_cli
from cfg_checker.common.other import utils
from cfg_checker.common.settings import ENV_TYPE_KUBE
from cfg_checker.common.exception import CheckerException
from cfg_checker.helpers import args_utils
from cfg_checker.modules.ceph import info, bench


command_help = "Ceph Storage information and benchmarks"
supported_envs = [ENV_TYPE_KUBE]


# def _selectClass(_env, strClassHint="checker"):
#     _class = None
#     if _env == ENV_TYPE_SALT:
#         if strClassHint == "info":
#             _class = info.SaltCephInfo
#         elif strClassHint == "bench":
#             _class = bench.SaltCephInfo
#     elif _env == ENV_TYPE_KUBE:
#         if strClassHint == "info":
#             _class = info.KubeCephInfo
#         elif strClassHint == "bench":
#             _class = bench.KubeCephBench
#     if not _class:
#         raise CheckerException(
#             "Unknown hint for selecting Ceph handler Class: '{}'".format(
#                 strClassHint
#             )
#         )
#     else:
#         return _class


def _validate_option_type(value, type_list):
    _s, _t = utils.split_option_type(value)
    if _t not in type_list:
        raise CheckerException(
            "Invalid option type '{}'. Expected types: {}".format(
                value,
                ", ".join(type_list)
            )
        )
    else:
        return


def _validate_option(value, type_list):
    if value not in type_list:
        raise CheckerException(
            "Invalid option '{}'. Expected one of: {}".format(
                value,
                ", ".join(type_list)
            )
        )
    else:
        return


def _get_param_and_log(arg, param_str):
    _value = args_utils.get_arg(arg, param_str)
    logger_cli.info("    {}={}".format(param_str, _value))
    return _value


def init_parser(_parser):
    # network subparser
    ceph_subparsers = _parser.add_subparsers(dest='type')

    ceph_info_parser = ceph_subparsers.add_parser(
        'info',
        help="Gather Ceph Cluster information"
    )

    ceph_info_parser.add_argument(
        '--detailed',
        action="store_true", default=False,
        help="Print additional details. (Not implemented yet)"
    )

    ceph_info_parser.add_argument(
        '--client-name',
        metavar='client_name',
        help="Client name for archive naming"
    )

    ceph_info_parser.add_argument(
        '--project-name',
        metavar='projectname',
        help="Project name for archive naming"
    )

    ceph_info_parser.add_argument(
        '--html',
        metavar='ceph_html_filename',
        help="HTML filename to save report"
    )

    ceph_bench_parser = ceph_subparsers.add_parser(
        'bench',
        help="Run ceph benchmark"
    )

    ceph_bench_parser.add_argument(
        '--agents',
        type=int, metavar='agent_count', default=5,
        help="Number of agents to use in all test runs. Default: 5"
    )
    ceph_bench_parser.add_argument(
        '--html',
        metavar='ceph_html_filename',
        help="HTML filename to save report"
    )
    ceph_bench_parser.add_argument(
        '--storage-class',
        metavar='storage_class',
        help="Storage class to be used in benchmark"
    )
    ceph_bench_parser.add_argument(
        '--task-file',
        metavar='task_file',
        help="Task file for benchmark with parameters to use"
    )
    ceph_bench_parser.add_argument(
        '--no-cleanup',
        action="store_true", default=False,
        help="Do not cleanup services, agents, pvc, and pv"
    )
    ceph_bench_parser.add_argument(
        '--cleanup-only',
        action="store_true", default=False,
        help="Cleanup resources related to benchmark"
    )
    ceph_bench_parser.add_argument(
        '--report-only',
        action="store_true", default=False,
        help="Just create report using files in '--dump-path' folder"
    )
    ceph_bench_parser.add_argument(
        '--dump-path',
        metavar="dump_results",
        help="Dump result after each test run to use them later. "
        "Default: '/tmp'"
    )
    ceph_bench_parser.add_argument(
        '--name',
        metavar="name", default="cephbench",
        help="Job name to use for running fio. "
        "Can be used to grep results. Default: 'cephbench'"
    )
    ceph_bench_parser.add_argument(
        '--bs',
        metavar="blocksize", default="16k",
        help="Block size for single run. Default: '16k'"
    )
    ceph_bench_parser.add_argument(
        '--iodepth',
        metavar="iodepth", default="16",
        help="IO Depth for single run. Default: '16'"
    )
    ceph_bench_parser.add_argument(
        '--size',
        metavar="size", default="10G",
        help="Persistent volume size (M, G). Default: '10G'"
    )
    ceph_bench_parser.add_argument(
        '--readwrite',
        metavar="readwrite", default="randrw",
        help="Test mode for single run (read, write, randrw, "
        "randread, randwrite). Default: 'randrw'"
    )
    ceph_bench_parser.add_argument(
        '--rwmixread',
        metavar="rwmixread", default="50",
        help="Percent of read in random mixed mode (randrw). Default: '50'"
    )
    ceph_bench_parser.add_argument(
        '--ramp-time',
        metavar="ramp_time", default="5s",
        help="Warmup time before test. Default: '5s'"
    )
    ceph_bench_parser.add_argument(
        '--runtime',
        metavar="runtime", default="60s",
        help="How long to run test. Default: '60s'"
    )
    ceph_bench_parser.add_argument(
        '--ioengine',
        metavar="ioengine", default="libaio",
        help="IO Engine used by fio. See 'fio eng-help' output for list. "
        "Default: 'libaio'"
    )
    ceph_bench_parser.add_argument(
        '--offset-increment',
        metavar="offset_increment", default="500M",
        help="Offset to be used in 'read' and 'write' modes if multiple jobs "
             "used"
        "Default: '500M'"
    )

    return _parser


def do_info(args, config):
    # Ceph info
    # Gather ceph info and create an archive with data
    args_utils.check_supported_env(ENV_TYPE_KUBE, args, config)

    # check client and project names
    if not args.client_name or not args.project_name:
        logger_cli.error(
            "ERROR: Missing '--client-name' or '--project-name' options"
        )
        return
    # _class = _selectClass(_env)
    ceph_info = info.KubeCephInfo(config)
    _tgzfilename = ceph_info.get_info_archive_filename(
        args.client_name,
        args.project_name
    )
    logger_cli.info("# Archive will be generated to '{}'".format(_tgzfilename))
    # get html
    _htmlfilename = args_utils.get_arg(args, 'html')

    logger_cli.info("# Collecting Ceph cluster information")
    ceph_info.gather_info()
    ceph_info.gather_osd_configs()

    # Debug, enable if needed to debug report generation
    # without actuall data collecting each time
    # ceph_info.dump_info()
    # ceph_info.load_info()
    # end debug

    ceph_info.get_transposed_latency_table()
    ceph_info.get_latest_health_readout()
    ceph_info.create_html_report(_htmlfilename)

    # handle cli part
    ceph_info.generate_archive(_tgzfilename)
    ceph_info.print_summary()

    return


def do_bench(args, config):
    # Ceph Benchmark using multiple pods
    # if only cleanup needed do it and exit
    _cleanup_only = args_utils.get_arg(args, 'cleanup_only')
    _report_only = args_utils.get_arg(args, 'report_only')
    config.resource_prefix = "cfgagent"
    if _cleanup_only:
        # Do forced resource cleanup and exit
        config.bench_mode = "cleanup"
        config.bench_agent_count = -1
        ceph_bench = bench.KubeCephBench(config)
        logger_cli.info(
            "# Discovering benchmark resources using prefix of '{}'".format(
                config.resource_prefix
            )
        )
        ceph_bench.prepare_cleanup()
        ceph_bench.cleanup()
        return

    # dump results options
    _dump_path = args_utils.get_arg(args, "dump_path")
    if _dump_path:
        if not os.path.exists(_dump_path):
            raise CheckerException(
                "ERROR: Dump path invalid: '{}'".format(_dump_path)
            )
        logger_cli.info("# Results will be dumped to '{}'".format(_dump_path))
        config.bench_results_dump_path = _dump_path
    else:
        _p = "/tmp"
        logger_cli.info(
            "# No result dump path set. Defaulting to {}"
            "Consider setting it if running long task_file "
            "based test runs".format(_p)
        )
        config.bench_results_dump_path = _p

    # Report filename
    _filename = args_utils.get_arg(args, 'html')
    # gather Ceph info
    logger_cli.info("# Collecting Ceph cluster information")
    ceph_info = info.KubeCephInfo(config)

    # Task files or options
    _opts = get_fio_options()
    # Load name and announce it
    config.bench_name = args_utils.get_arg(args, "name")
    _opts["name"] = config.bench_name
    logger_cli.info(
        "# Using '{}' as ceph bench jobs name".format(_opts["name"])
    )

    if _report_only:
        # Do forced report creation and exit
        config.bench_mode = "report"
        config.bench_agent_count = -1
        ceph_bench = bench.KubeCephBench(config)
        ceph_bench.set_ceph_info_class(ceph_info)
        logger_cli.info(
            "# Preparing to generate report '{}'".format(
                config.resource_prefix
            )
        )
        # Preload previous results for this name
        ceph_bench.preload_results()
        # Gather ceph data
        ceph_bench.wait_ceph_cooldown()
        # Generate report
        ceph_bench.create_report(_filename)
        return

    # Prepare the tasks and do synced testrun or a single one
    logger_cli.info("# Initializing ceph benchmark module")
    args_utils.check_supported_env(ENV_TYPE_KUBE, args, config)
    # agents count option
    config.bench_agent_count = args_utils.get_arg(args, "agents")
    logger_cli.info("-> using {} agents".format(config.bench_agent_count))
    # Cleaning option
    config.no_cleaning_after_benchmark = args_utils.get_arg(args, "no_cleanup")
    # storage class
    _storage_class = args_utils.get_arg(args, "storage_class")
    logger_cli.info("-> using storage class of '{}'".format(_storage_class))
    config.bench_storage_class = _storage_class
    if _dump_path:
        logger_cli.info("# Results will be dumped to '{}'".format(_dump_path))
        config.bench_results_dump_path = _dump_path
    else:
        logger_cli.info(
            "# No result dump path set. "
            "Consider setting it if running long task_file based test runs"
        )
        config.bench_results_dump_path = _dump_path

    _task_file = args_utils.get_arg(args, "task_file", nofail=True)
    if not _task_file:
        logger_cli.info("-> Running single benchmark run")
        config.bench_mode = "single"
        # Updating _opts from arguments
        _params = [
            "bs",
            "iodepth",
            "size",
            "readwrite",
            "ramp_time",
            "runtime",
            "ioengine"
        ]
        for _p in _params:
            _opts[_p] = _get_param_and_log(args, _p)
        if _opts["readwrite"] in seq_modes:
            _p = "offset_increment"
            _opts[_p] = _get_param_and_log(args, _p)
        elif _opts["readwrite"] in mix_modes:
            _p = "rwmixread"
            _opts[_p] = _get_param_and_log(args, _p)
    else:
        logger_cli.info("-> running with tasks from '{}'".format(_task_file))
        config.bench_task_file = _task_file
        config.bench_mode = "tasks"
        # Add default size to options
        _opts["size"] = _get_param_and_log(args, "size")
    logger_cli.debug("... default/selected options for fio:")
    for _k in _opts.keys():
        # TODO: Update options for single run
        logger_cli.debug("    {} = {}".format(_k, _opts[_k]))

    # init the Bench class
    ceph_bench = bench.KubeCephBench(config)
    ceph_bench.set_ceph_info_class(ceph_info)

    # Validate options
    logger_cli.info("-> Validating options")
    # size
    _validate_option_type(_opts["size"], ["G", "M"])
    _validate_option_type(_opts["ramp_time"], ["s", "m"])
    _validate_option_type(_opts["runtime"], ["s", "m"])
    _modes = seq_modes + mix_modes + rand_modes
    _validate_option(_opts["readwrite"], _modes)

    if _task_file:
        _s, _ = utils.split_option_type(_opts["size"])
        for idx in range(len(ceph_bench.tasks)):
            # size
            _ts, _ = utils.split_option_type(ceph_bench.tasks[idx]["size"])
            if _s < _ts:
                logger_cli.error(
                    "ERROR: Task #{} file size is to big:"
                    " {} (volume) < {} (testfile)".format(
                        idx,
                        _opts["size"],
                        ceph_bench.tasks[idx]["size"]
                    )
                )
            # other
            _validate_option(ceph_bench.tasks[idx]["readwrite"], _modes)
        # Print defaults
        logger_cli.debug("... default/selected options for fio:")
        for _k in _opts.keys():
            # TODO: Update options for single run
            logger_cli.debug("    {} = {}".format(_k, _opts[_k]))

    # Preload previous results for this name
    ceph_bench.preload_results()
    # Do the testrun
    ceph_bench.prepare_agents(_opts)
    ceph_bench.wait_ceph_cooldown()

    #  DEBUG of report in progress
    if not ceph_bench.run_benchmark(_opts):
        # No cleaning and/or report if benchmark was not finished
        logger_cli.info("# Abnormal benchmark run, no cleaning performed")
        return
    # Remove after DEBUG
    # ceph_bench.collect_results(_opts)
    # END DEBUG

    # Cleaning
    if not config.no_cleaning_after_benchmark:
        ceph_bench.cleanup()
    else:
        logger_cli.info(
            "# '--no-cleaning' option set. Cleaning not conducted."
        )

    # Create report
    ceph_bench.create_report(_filename)

    return