cfg_checker/modules/ceph/bench.py

import csv
import os
import json

from datetime import datetime, timedelta, timezone
from time import sleep

from cfg_checker.common import logger_cli
from cfg_checker.common.decorators import retry
from cfg_checker.helpers.console_utils import Progress
# from cfg_checker.common.exception import InvalidReturnException
# from cfg_checker.common.exception import ConfigException
# from cfg_checker.common.exception import KubeException

from cfg_checker.nodes import KubeNodes
from cfg_checker.agent.fio_runner import _get_seconds, _datetime_fmt


class CephBench(object):
    _agent_template = "cfgagent-template.yaml"

    def __init__(self, config):
        self.env_config = config
        return


class SaltCephBench(CephBench):
    def __init__(
        self,
        config
    ):
        logger_cli.error("ERROR: Not impelented for Salt environment!")

        # self.master = SaltNodes(config)
        super(SaltCephBench, self).__init__(config)
        return


class KubeCephBench(CephBench):
    def __init__(self, config):
        self.agent_count = config.bench_agent_count
        self.master = KubeNodes(config)
        super(KubeCephBench, self).__init__(config)
        self.storage_class = config.bench_storage_class
        self.agent_pods = []
        self.services = []
        self.scheduled_delay = 30
        self.mode = config.bench_mode
        if config.bench_mode == "tasks":
            self.taskfile = config.bench_task_file
            self.load_tasks(self.taskfile)

        self.cleanup_list = []
        self.results = {}

    def load_tasks(self, taskfile):
        # Load csv file
        logger_cli.info("-> Loading taskfile '{}'".format(taskfile))
        self.tasks = []
        with open(taskfile) as f:
            _reader = csv.reader(f, delimiter=',')
            # load packages
            for row in _reader:
                self.tasks.append({
                    "readwrite": row[0],
                    "rwmixread": row[1],
                    "bs": row[2],
                    "iodepth": row[3],
                    "size": row[4]
                })

    def add_for_deletion(self, obj, typ):
        _d = [typ, obj.metadata.namespace, obj.metadata.name]
        self.cleanup_list.append(_d)
        return

    def prepare_agents(self, options):
        logger_cli.info("# Preparing {} agents".format(self.agent_count))
        for idx in range(self.agent_count):
            # create pvc/pv and pod
            logger_cli.info("-> creating agent '{:02}'".format(idx))
            _agent, _pv, _pvc = self.master.prepare_benchmark_agent(
                idx,
                os.path.split(options["filename"])[0],
                self.storage_class,
                options['size'] + 'i',
                self._agent_template
            )
            # save it to lists
            self.agent_pods.append(_agent)
            self.add_for_deletion(_pv, "pv")
            self.add_for_deletion(_pvc, "pvc")
            self.add_for_deletion(_agent, "pod")

            # expose it
            _svc = self.master.expose_benchmark_agent(_agent)
            self.add_for_deletion(_svc, "svc")
            # Save service
            self.services.append(_svc)
            # prepopulate results
            self.results[_agent.metadata.name] = {}
            self.results[_agent.metadata.name]["list"] = {}
            self.results[_agent.metadata.name]["url"] = \
                "http://{}:{}/api/".format(
                    _svc.spec.cluster_ip,
                    8765
                )
            self.results[_agent.metadata.name]["storage_class"] = \
                self.storage_class
            self.results[_agent.metadata.name]["volume_size"] = \
                options['size']

        logger_cli.info("-> Done creating agents")
        return

    def _poke_agent(self, url, body, action="GET"):
        _datafile = "/tmp/data"
        _data = [
            "-d",
            "@" + _datafile
        ]
        _cmd = [
            "curl",
            "-s",
            "-H",
            "'Content-Type: application/json'",
            "-X",
            action,
            url
        ]
        if body:
            _cmd += _data
            _ret = self.master.prepare_json_in_pod(
                self.agent_pods[0].metadata.name,
                self.master._namespace,
                body,
                _datafile
            )
        _ret = self.master.exec_cmd_on_target_pod(
            self.agent_pods[0].metadata.name,
            self.master._namespace,
            " ".join(_cmd)
        )
        try:
            return json.loads(_ret)
        except TypeError as e:
            logger_cli.error(
                "ERROR: Status not decoded: {}\n{}".format(e, _ret)
            )
        except json.decoder.JSONDecodeError as e:
            logger_cli.error(
                "ERROR: Status not decoded: {}\n{}".format(e, _ret)
            )

        return None

    def _ensure_agents_ready(self):
        # make sure agents idle
        _status_set = []
        _ready_set = []
        for _agent, _d in self.get_agents_status().items():
            # obviously, there should be some answer
            if _d is None:
                logger_cli.error("ERROR: Agent status not available")
                return False
            # status should be idle or finished
            if _d['status'] not in ["idle", "finished"]:
                logger_cli.error(
                    "Agent status invalid {}:{}".format(_agent, _d['status'])
                )
                _status_set += [False]
            else:
                # Good agent
                _status_set += [True]
            # agent's fio shell should be in 'ready'
            if not _d["healthcheck"]["ready"]:
                logger_cli.error("Agent is not ready {}".format(_agent))
                _ready_set += [False]
            else:
                # 'fio' shell for agent is ready
                _ready_set += [True]
        # all agent's statuses should be True
        # and all 'fio' shell modules should be 'ready'
        if not any(_status_set) or not any(_ready_set):
            # At least one is not ready and it was already logged above
            return False
        else:
            # All is good
            return True

    def get_agents_status(self):
        _status = {}
        for _agent, _d in self.results.items():
            _status[_agent] = self._poke_agent(_d["url"] + "fio", {})
        return _status

    def get_agents_resultlist(self):
        _t = {"module": "fio", "action": "get_resultlist"}
        _status = {}
        for _agent, _d in self.results.items():
            _status[_agent] = self._poke_agent(_d["url"], _t, action="POST")
        return _status

    @retry(Exception)
    def get_result_from_agent(self, agent, time):
        _t = {
            "module": "fio",
            "action": "get_result",
            "options": {
                "time": time
            }
        }
        return self._poke_agent(self.results[agent]["url"], _t, action="POST")

    def _get_next_scheduled_time(self):
        _now = datetime.now(timezone.utc)
        logger_cli.info("-> time is '{}'".format(_now.strftime(_datetime_fmt)))
        _time = _now + timedelta(seconds=self.scheduled_delay)
        _str_time = _time.strftime(_datetime_fmt)
        logger_cli.info(
            "-> next benchmark scheduled to '{}'".format(_str_time)
        )
        return _str_time

    def _send_scheduled_task(self, options):
        _task = {
            "module": "fio",
            "action": "do_scheduledrun",
            "options": options
        }
        for _agent, _d in self.results.items():
            logger_cli.info(
                "-> sending task to '{}:{}'".format(_agent, _d["url"])
            )
            _ret = self._poke_agent(_d["url"], _task, action="POST")
            if 'error' in _ret:
                logger_cli.error(
                    "ERROR: Agent returned: '{}'".format(_ret['error'])
                )
                return False
        # No errors detected
        return True

    def track_benchmark(self, options):
        _runtime = _get_seconds(options["runtime"])
        _ramptime = _get_seconds(options["ramp_time"])
        # Sum up all timings that we must wait and double it
        _timeout = (self.scheduled_delay + _runtime + _ramptime) * 2
        _end = datetime.now(timezone.utc) + timedelta(seconds=_timeout)
        while True:
            # Print status
            # TODO: do pooled status get
            _sts = self.get_agents_status()
            diff = (_end - datetime.now(timezone.utc)).total_seconds()
            logger_cli.info("-> {:.2f}s left. Agent status:".format(diff))
            for _agent, _status in _sts.items():
                logger_cli.info(
                    "\t{}: {} ({}%)".format(
                        _agent,
                        _status["status"],
                        _status["progress"]
                    )
                )
            finished = [True for _s in _sts.values() 
                        if _s["status"] == 'finished']
            _fcnt = len(finished)
            _tcnt = len(_sts)
            if _fcnt < _tcnt:
                logger_cli.info("-> {}/{} finished".format(_fcnt, _tcnt))
            else:
                logger_cli.info("-> All agents finished run")
                return True
            # recalc how much is left
            diff = (_end - datetime.now(timezone.utc)).total_seconds()
            # In case end_datetime was in past to begin with
            if diff < 0:
                logger_cli.info("-> Timed out waiting for agents to finish")
                return False
            logger_cli.info("-> Sleeping for {:.2f}s".format(diff/3))
            sleep(diff/3)
            if diff <= 0.1:
                logger_cli.info("-> Timed out waiting for agents to finish")
                return False

    def _do_testrun(self, options):
        # send single to agent
        if not self._send_scheduled_task(options):
            return False
        # Track this benchmark progress
        if not self.track_benchmark(options):
            return False
        else:
            logger_cli.info("-> Finished testrun")
            # Get results for each agent
            self.collect_results()
            return True

    def _wait_ceph_cooldown(self):
        # TODO: Query Ceph ince a 20 sec to make sure its load dropped

        return

    def run_benchmark(self, options):
        logger_cli.info("# Starting '{}' benchmark".format(self.mode))
        # Check agent readyness
        logger_cli.info("# Checking agents")
        if not self._ensure_agents_ready():
            return False

        # Make sure that Ceph is at low load
        # TODO: Ceph status check
        self._wait_ceph_cooldown()

        # Do benchmark according to mode
        if self.mode == "tasks":
            logger_cli.info(
                "# Running benchmark with tasks from '{}'".format(
                    self.taskfile
                )
            )
            # take next task
            _total_tasks = len(self.tasks)
            for idx in range(_total_tasks):
                _task = self.tasks[idx]
                logger_cli.info(
                    "-> Starting next task ({}/{})".format(idx+1, _total_tasks)
                )
                logger_cli.info("-> Updating options with: {}".format(
                        ", ".join(
                            ["{} = {}".format(k, v) for k, v in _task.items()]
                        )
                    )
                )
                # update options
                options.update(_task)
                # init time to schedule
                options["scheduled_to"] = self._get_next_scheduled_time()
                if not self._do_testrun(options):
                    return False

                self._wait_ceph_cooldown()
        elif self.mode == "single":
            logger_cli.info("# Running single benchmark")
            # init time to schedule
            options["scheduled_to"] = self._get_next_scheduled_time()
            if not self._do_testrun(options):
                return False
        else:
            logger_cli.error("ERROR: Unknown mode '{}'".format(self.mode))
            return False

        # Normal exit
        logger_cli.info("# All benchmark tasks done")
        return True

    def cleanup(self):
        self.cleanup_list.reverse()

        for _res in self.cleanup_list:
            self.master.cleanup_resource_by_name(_res[0], _res[2], ns=_res[1])

        # Wait for resource to be cleaned
        _timeout = 120
        _total = len(self.cleanup_list)
        logger_cli.info("-> Wait until {} resources cleaned".format(_total))
        _p = Progress(_total)
        while True:
            _g = self.master.get_resource_phase_by_name
            _l = [_g(r[0], r[2], ns=r[1]) for r in self.cleanup_list]
            _l = [item for item in _l if item]
            _idx = _total - len(_l)
            if len(_l) > 0:
                _p.write_progress(_idx)
            else:
                _p.write_progress(_idx)
                _p.end()
                logger_cli.info("# Done cleaning up")
                break
            if _timeout > 0:
                _timeout -= 1
            else:
                _p.end()
                logger_cli.info("# Timed out waiting after 120s.")
                break

        return

    def collect_results(self):
        logger_cli.info("# Collecting results")
        # query agents for results
        _agents = self.get_agents_resultlist()

        for _agent, _l in _agents.items():
            _list = _l["resultlist"]
            _new = [r for r in _list if r not in self.results[_agent]["list"]]
            logger_cli.debug(
                "... agent '{}' has {} new results".format(_agent, len(_new))
            )
            #  get all new results
            for _time in _new:
                logger_cli.info(
                    "-> loading results for '{}' from '{}'".format(
                        _time,
                        _agent
                    )
                )
                self.results[_agent]["list"].update(
                    self.get_result_from_agent(_agent, _time)
                )
        return

    def dump_results(self, path):
        # Function dumps all availabkle results as jsons to the given path
        # overwriting if needed

        # TODO: Conduct the dumping

        return

    # Create report
    def create_report(self, filename):

        return