diff options
Diffstat (limited to 'vstf/vstf/controller/api_server.py')
| -rwxr-xr-x | vstf/vstf/controller/api_server.py | 403 |
1 files changed, 403 insertions, 0 deletions
diff --git a/vstf/vstf/controller/api_server.py b/vstf/vstf/controller/api_server.py new file mode 100755 index 00000000..d3547011 --- /dev/null +++ b/vstf/vstf/controller/api_server.py @@ -0,0 +1,403 @@ +import uuid +import time +import os +import sys +import logging +import signal +import json + +from vstf.common import unix, message, cliutil, excepts +from vstf.common.vstfcli import VstfParser +from vstf.common.log import setup_logging +from vstf.common import daemon +from vstf.rpc_frame_work import rpc_producer +from vstf.controller.fabricant import Fabricant +from vstf.agent.env.basic.commandline import CommandLine +from vstf.controller.env_build.env_build import EnvBuildApi as Builder +from vstf.controller.env_build.env_collect import EnvCollectApi +from vstf.controller.database.dbinterface import DbManage +import vstf.controller.sw_perf.performance as pf +from vstf.controller.settings.tester_settings import TesterSettings +from vstf.controller.settings.device_settings import DeviceSettings +from vstf.controller.settings.flows_settings import FlowsSettings +from vstf.controller.settings.mail_settings import MailSettings +from vstf.controller.settings.tool_settings import ToolSettings +from vstf.controller.settings.perf_settings import PerfSettings +from vstf.controller.sw_perf.perf_provider import PerfProvider +from vstf.controller.sw_perf.flow_producer import FlowsProducer +import vstf.controller.reporters.reporter as rp +import vstf.common.constants as cst +import vstf.common.check as chk + +LOG = logging.getLogger(__name__) +cmd = CommandLine() + + +class OpsChains(object): + def __init__(self, monitor, port): + """The ops chains will setup the proxy to rabbitmq + and setup a thread to watch the queues of rabbitmq + + """ + super(OpsChains, self).__init__() + if not os.path.exists(cst.VSTFCPATH): + os.mkdir(cst.VSTFCPATH) + + LOG.info("VSTF Manager start to listen to %s", monitor) + self.chanl = rpc_producer.Server(host=monitor, port=port) + self.dbconn = DbManage() + self.collection = EnvCollectApi(self.chanl) + + def list_devs(self, **kwargs): + target = kwargs.get('host') + if not target: + respond = "the target is empty, not support now." + else: + respond = self.chanl.call(self.chanl.make_msg("list_nic_devices"), target) + return respond + + def src_install(self, host, config_file): + if not os.path.exists(config_file): + raise Exception("Can not found the config file.") + cfg = json.load(open(config_file)) + msg = self.chanl.make_msg("src_install", cfg=cfg) + return self.chanl.call(msg, host, timeout=1000) + + def create_images(self, host, config_file): + if not os.path.exists(config_file): + raise Exception("Can not found the config file.") + cfg = json.load(open(config_file)) + msg = self.chanl.make_msg("create_images", cfg=cfg) + return self.chanl.call(msg, host, timeout=1000) + + def clean_images(self, host, config_file): + if not os.path.exists(config_file): + raise Exception("Can not found the config file.") + cfg = json.load(open(config_file)) + msg = self.chanl.make_msg("clean_images", cfg=cfg) + return self.chanl.call(msg, host, timeout=1000) + + def apply_model(self, host, model=None, config_file=None): + if config_file is None: + config_file = "/etc/vstf/env/%s.json" % model + if not os.path.exists(config_file): + raise Exception("Can not found the config file.") + env = Builder(self.chanl, config_file) + ret = env.build() + return ret + + def disapply_model(self, host, model=None, config_file=None): + if config_file is None: + config_file = "/etc/vstf/env/%s.json" % model + if not os.path.exists(config_file): + raise Exception("Can not found the config file.") + env = Builder(self.chanl, config_file) + ret = env.clean() + return ret + + def list_tasks(self): + ret = self.dbconn.query_tasks() + head = [["Task ID", "Task Name", "Task Date", "Task Remarks"]] + if ret: + ret = head + ret + return ret + + def affctl_list(self, host): + if not host: + return "Need input the host" + return Fabricant(host, self.chanl).affctl_list() + + def _create_task(self, scenario): + taskid = self.dbconn.create_task(str(uuid.uuid4()), time.strftime(cst.TIME_FORMAT), + desc=scenario + "Test") + LOG.info("new Task id:%s" % taskid) + if -1 == taskid: + raise Exception("DB create task failed.") + + device = DeviceSettings().settings + hosts = [device["host"], device["tester"]] + for host in hosts: + LOG.info(host) + + devs = host["devs"][0] + keys = ["bdf", "iface", "mac"] + key = devs.keys()[0] + if key in keys: + name = devs[key] + else: + raise Exception("error devs :%s", devs) + + query = Fabricant(host["agent"], self.chanl) + nic_info = query.get_device_detail(identity=name) + + LOG.info(nic_info) + + os_info, cpu_info, mem_info, hw_info = self.collection.collect_host_info(host["agent"]) + LOG.info(os_info) + LOG.info(cpu_info) + LOG.info(mem_info) + LOG.info(hw_info) + + self.dbconn.add_host_2task(taskid, + host["agent"], + json.dumps(hw_info[cst.HW_INFO]), + json.dumps(cpu_info[cst.CPU_INFO]), + json.dumps(mem_info[cst.MEMORY_INFO]), + nic_info["desc"], + json.dumps(os_info[cst.OS_INFO])) + + self.dbconn.add_extent_2task(taskid, "CETH", "driver", "version 2.0") + self.dbconn.add_extent_2task(taskid, "EVS", "switch", "version 3.0") + return taskid + + def settings(self, mail=False, perf=False): + LOG.info("mail:%s, perf:%s" % (mail, perf)) + if mail: + MailSettings().input() + if perf: + PerfSettings().input() + + def report(self, rpath='./', mail_off=False, taskid=-1): + report = rp.Report(self.dbconn, rpath) + if taskid == -1: + taskid = self.dbconn.get_last_taskid() + report.report(taskid, mail_off) + info_str = "do report over" + return info_str + + def run_perf_cmd(self, case, rpath='./', affctl=False, build_on=False, save_on=False, report_on=False, mail_on=False): + LOG.info(case) + LOG.info("build_on:%s report_on:%s mail_on:%s" % (build_on, report_on, mail_on)) + casetag = case['case'] + tool = case['tool'] + protocol = case['protocol'] + profile = case['profile'] + ttype = case['type'] + sizes = case['sizes'] + + ret, ret_str = chk.check_case_params(protocol, ttype, tool) + if not ret: + return ret_str + + scenario = self.dbconn.query_scenario(casetag) + LOG.info(scenario) + if not scenario: + LOG.warn("not support the case:%s", casetag) + return + + config_file = os.path.join("/etc/vstf/env", scenario + ".json") + + LOG.info(config_file) + env = Builder(self.chanl, config_file) + if build_on: + env.build() + flows_settings = FlowsSettings() + tool_settings = ToolSettings() + tester_settings = TesterSettings() + flow_producer = FlowsProducer(self.chanl, flows_settings) + provider = PerfProvider(flows_settings.settings, tool_settings.settings, tester_settings.settings) + + perf = pf.Performance(self.chanl, provider) + flow_producer.create(scenario, casetag) + result = perf.run(tool, protocol, ttype, sizes, affctl) + LOG.info(flows_settings.settings) + LOG.info(result) + if save_on: + taskid = self._create_task(scenario) + testid = self.dbconn.add_test_2task(taskid, casetag, protocol, profile, ttype, tool) + LOG.info(testid) + self.dbconn.add_data_2test(testid, result) + if report_on: + self.report(rpath, not mail_on, taskid) + return result + + def run_perf_file(self, rpath='./', affctl=False, report_on=True, mail_on=True): + perf_settings = PerfSettings() + flows_settings = FlowsSettings() + tool_settings = ToolSettings() + tester_settings = TesterSettings() + flow_producer = FlowsProducer(self.chanl, flows_settings) + provider = PerfProvider(flows_settings.settings, tool_settings.settings, tester_settings.settings) + perf = pf.Performance(self.chanl, provider) + tests = perf_settings.settings + + for scenario, cases in tests.items(): + LOG.info(scenario) + if not cases: + continue + + config_file = os.path.join("/etc/vstf/env", scenario + ".json") + + LOG.info(config_file) + env = Builder(self.chanl, config_file) + env.build() + + taskid = self._create_task(scenario) + + for case in cases: + LOG.info(case) + casetag = case['case'] + tool = case['tool'] + protocol = case['protocol'] + profile = case['profile'] + ttype = case['type'] + sizes = case['sizes'] + + ret, ret_str = chk.check_case_params(protocol, ttype, tool) + if not ret: + LOG.warn(ret_str) + continue + + flow_producer.create(scenario, casetag) + result = perf.run(tool, protocol, ttype, sizes, affctl) + LOG.info(result) + + testid = self.dbconn.add_test_2task(taskid, casetag, protocol, profile, ttype, tool) + LOG.info(testid) + + self.dbconn.add_data_2test(testid, result) + + if report_on: + self.report(rpath, not mail_on, taskid) + + info_str = "do batch perf test successfully" + return info_str + + def collect_host_info(self, target): + if self.collection is not None: + return self.collection.collect_host_info(target) + else: + return "collection is None" + + +class Manager(daemon.Daemon): + def __init__(self): + """ + The manager will create a socket for vstfadm. + also the manager own a ops chains + """ + super(Manager, self).__init__(cst.vstf_pid) + # the connection of socket + self.conn = None + # the operations of manager + self.ops = None + # record the daemon run flag + self.run_flag = True + + def deal_unknown_obj(self, obj): + return "unknown response %s" % obj + + def run(self): + signal.signal(signal.SIGTERM, self.daemon_die) + # setup the socket server for communicating with vstfadm + try: + self.conn = unix.UdpServer() + self.conn.bind() + self.conn.listen() + except Exception as e: + raise e + + # accept the connection of vstfadm and recv the command + # run the command from vstfadm and return the response + while self.run_flag: + conn, addr = self.conn.accept() + LOG.debug("accept the conn: %(conn)s", {'conn': conn}) + + # recv the msg until the conn break. + + while True: + try: + data = message.recv(conn.recv) + LOG.debug("Manager recv the msg: %(msg)s", {'msg': data}) + msg = message.decode(data) + body = message.get_body(msg) + context = message.get_context(msg) + except RuntimeError: + LOG.debug("manage catch the connection close!") + break + except Exception as e: + LOG.error("Manager recv message from socket failed.") + self.daemon_die() + raise e + + try: + func = getattr(self.ops, body.get('method')) + LOG.info("Call function:%s, args:%s", + func.__name__, body.get('args')) + response = func(**body.get('args')) + LOG.info("response: %s", response) + except excepts.UnsolvableExit as e: + msg = "The manager opps, exit now" + LOG.error(msg) + # the manager has no need to be continue, just return + # this msg and exit normal + self.daemon_die() + raise e + except Exception as e: + # here just the function failed no need exit, just return the msg + msg = "Run function failed. [ %s ]" % (e) + response = msg + LOG.error(msg) + try: + response = message.add_context(response, **context) + LOG.debug("Manager send the response: <%(r)s", {'r': response}) + message.send(conn.send, message.encode(response)) + except Exception as e: + self.daemon_die() + raise e + # close the connection when conn down + conn.close() + + def daemon_die(self, signum, frame): + """overwrite daemon.Daemon.daemon_die(self)""" + LOG.info("manage catch the signal %s to exit." % signum) + if self.conn: + # we can not close the conn direct, just tell manager to stop accept + self.run_flag = False + + if self.ops: + # stop the ops's proxy + # maybe happen AttributeError: 'BlockingConnection' object has no attribute 'disconnect' + # this a know bug in pika. fix in 0.9.14 release + try: + self.ops.chanl.close() + except AttributeError: + LOG.warning("The connection close happens attribute error") + + def start_manage(self, monitor="localhost", port=5672): + try: + # create manager's ops chains here will create a proxy to rabbitmq + self.ops = OpsChains(monitor, port) + except Exception as e: + raise e + self.start() + + def stop_manage(self): + self.stop() + + +@cliutil.arg("--monitor", + dest="monitor", + default="localhost", + action="store", + help="which ip to be monitored") +@cliutil.arg("--port", + dest="port", + default="5672", + action="store", + help="rabbitmq conn server") +def do_start(args): + Manager().start_manage(args.monitor, args.port) + + +def do_stop(args): + Manager().stop_manage() + + +def main(): + """this is for vstfctl""" + setup_logging(level=logging.INFO, log_file="/var/log/vstf/vstf-manager.log", clevel=logging.INFO) + parser = VstfParser(prog="vstf-manager", description="vstf manager command line") + parser.set_subcommand_parser(target=sys.modules[__name__]) + args = parser.parse_args() + args.func(args) |