From 89b608ae766705950efc5f4914b01b9a32b6a0e7 Mon Sep 17 00:00:00 2001 From: Yaroslav Brustinov Date: Fri, 13 May 2016 20:11:51 +0300 Subject: add master daemon --- .../trex_control_plane/server/CCustomLogger.py | 4 + .../server/extended_daemon_runner.py | 146 --------------------- .../trex_control_plane/server/trex_daemon_server | 25 ---- .../server/trex_daemon_server.py | 79 ----------- .../trex_control_plane/server/trex_server.py | 137 +++++++++++++------ 5 files changed, 102 insertions(+), 289 deletions(-) delete mode 100755 scripts/automation/trex_control_plane/server/extended_daemon_runner.py delete mode 100755 scripts/automation/trex_control_plane/server/trex_daemon_server delete mode 100755 scripts/automation/trex_control_plane/server/trex_daemon_server.py (limited to 'scripts/automation/trex_control_plane/server') diff --git a/scripts/automation/trex_control_plane/server/CCustomLogger.py b/scripts/automation/trex_control_plane/server/CCustomLogger.py index ecf7d519..a8823cea 100755 --- a/scripts/automation/trex_control_plane/server/CCustomLogger.py +++ b/scripts/automation/trex_control_plane/server/CCustomLogger.py @@ -31,6 +31,10 @@ def setup_custom_logger(name, log_path = None): def setup_daemon_logger (name, log_path = None): # first make sure path availabe + try: + os.unlink(log_path) + except: + pass logging.basicConfig(level = logging.INFO, format = '%(asctime)s %(name)-10s %(module)-20s %(levelname)-8s %(message)s', datefmt = '%m-%d %H:%M', diff --git a/scripts/automation/trex_control_plane/server/extended_daemon_runner.py b/scripts/automation/trex_control_plane/server/extended_daemon_runner.py deleted file mode 100755 index 7bc25aac..00000000 --- a/scripts/automation/trex_control_plane/server/extended_daemon_runner.py +++ /dev/null @@ -1,146 +0,0 @@ -#!/usr/bin/python - -import outer_packages -import lockfile -from daemon import runner,daemon -from daemon.runner import * -import os, sys -from argparse import ArgumentParser -from trex_server import trex_parser -try: - from termstyle import termstyle -except ImportError: - import termstyle - - -def daemonize_parser(parser_obj, action_funcs, help_menu): - """Update the regular process parser to deal with daemon process options""" - parser_obj.description += " (as a daemon process)" - parser_obj.usage = None - parser_obj.add_argument("action", choices=action_funcs, - action="store", help=help_menu) - - -class ExtendedDaemonRunner(runner.DaemonRunner): - """ Controller for a callable running in a separate background process. - - The first command-line argument is the action to take: - - * 'start': Become a daemon and call `app.run()`. - * 'stop': Exit the daemon process specified in the PID file. - * 'restart': Stop, then start. - - """ - - help_menu = """Specify action command to be applied on server. - (*) start : start the application in as a daemon process. - (*) show : prompt an updated status of daemon process (running/ not running). - (*) stop : exit the daemon process. - (*) restart : stop, then start again the application as daemon process - (*) start-live : start the application in live mode (no daemon process). - """ - - def __init__(self, app, parser_obj): - """ Set up the parameters of a new runner. - THIS METHOD INTENTIONALLY DO NOT INVOKE SUPER __init__() METHOD - - :param app: The application instance; see below. - :return: ``None``. - - The `app` argument must have the following attributes: - - * `stdin_path`, `stdout_path`, `stderr_path`: Filesystem paths - to open and replace the existing `sys.stdin`, `sys.stdout`, - `sys.stderr`. - - * `pidfile_path`: Absolute filesystem path to a file that will - be used as the PID file for the daemon. If ``None``, no PID - file will be used. - - * `pidfile_timeout`: Used as the default acquisition timeout - value supplied to the runner's PID lock file. - - * `run`: Callable that will be invoked when the daemon is - started. - - """ - super(runner.DaemonRunner, self).__init__() - # update action_funcs to support more operations - self.update_action_funcs() - - daemonize_parser(parser_obj, self.action_funcs, ExtendedDaemonRunner.help_menu) - args = parser_obj.parse_args() - self.action = unicode(args.action) - - self.app = app - self.daemon_context = daemon.DaemonContext() - self.daemon_context.stdin = open(app.stdin_path, 'rt') - try: - self.daemon_context.stdout = open(app.stdout_path, 'w+t') - except IOError as err: - # catch 'tty' error when launching server from remote location - app.stdout_path = "/dev/null" - self.daemon_context.stdout = open(app.stdout_path, 'w+t') - self.daemon_context.stderr = open(app.stderr_path, - 'a+t', buffering=0) - - self.pidfile = None - if app.pidfile_path is not None: - self.pidfile = make_pidlockfile(app.pidfile_path, app.pidfile_timeout) - self.daemon_context.pidfile = self.pidfile - - # mask out all arguments that aren't relevant to main app script - - def update_action_funcs(self): - self.action_funcs.update({u'start-live': self._start_live, u'show': self._show}) # add key (=action), value (=desired func) - - @staticmethod - def _start_live(self): - self.app.run() - - @staticmethod - def _show(self): - if self.pidfile.is_locked(): - print termstyle.red("TRex server daemon is running") - else: - print termstyle.red("TRex server daemon is NOT running") - - def do_action(self): - self.__prevent_duplicate_runs() - self.__prompt_init_msg() - try: - super(ExtendedDaemonRunner, self).do_action() - if self.action == 'stop': - self.__verify_termination() - except runner.DaemonRunnerStopFailureError: - if self.action == 'restart': - # error means server wasn't running in the first place- so start it! - self.action = 'start' - self.do_action() - - - def __prevent_duplicate_runs(self): - if self.action == 'start' and self.pidfile.is_locked(): - print termstyle.green("Server daemon is already running") - exit(1) - elif self.action == 'stop' and not self.pidfile.is_locked(): - print termstyle.green("Server daemon is not running") - exit(1) - - def __prompt_init_msg(self): - if self.action == 'start': - print termstyle.green("Starting daemon server...") - elif self.action == 'stop': - print termstyle.green("Stopping daemon server...") - - def __verify_termination(self): - pass -# import time -# while self.pidfile.is_locked(): -# time.sleep(2) -# self._stop() -# - - -if __name__ == "__main__": - pass diff --git a/scripts/automation/trex_control_plane/server/trex_daemon_server b/scripts/automation/trex_control_plane/server/trex_daemon_server deleted file mode 100755 index 3494e303..00000000 --- a/scripts/automation/trex_control_plane/server/trex_daemon_server +++ /dev/null @@ -1,25 +0,0 @@ -#!/usr/bin/python - -import os -import sys - -core = 0 - -if '--core' in sys.argv: - try: - idx = sys.argv.index('--core') - core = int(sys.argv[idx + 1]) - if core > 31 or core < 0: - print "Error: please provide core argument between 0 to 31" - exit(-1) - del sys.argv[idx:idx+2] - except IndexError: - print "Error: please make sure core option provided with argument" - exit(-1) - except ValueError: - print "Error: please make sure core option provided with integer argument" - exit(-1) - -str_argv = ' '.join(sys.argv[1:]) -cmd = "taskset -c {core} python automation/trex_control_plane/server/trex_daemon_server.py {argv}".format(core = core, argv = str_argv) -os.system(cmd) diff --git a/scripts/automation/trex_control_plane/server/trex_daemon_server.py b/scripts/automation/trex_control_plane/server/trex_daemon_server.py deleted file mode 100755 index 9784d42a..00000000 --- a/scripts/automation/trex_control_plane/server/trex_daemon_server.py +++ /dev/null @@ -1,79 +0,0 @@ -#!/usr/bin/python - -import outer_packages -import daemon -from trex_server import do_main_program, trex_parser -import CCustomLogger - -import logging -import time -import sys -import os, errno -import grp -import signal -from daemon import runner -from extended_daemon_runner import ExtendedDaemonRunner -import lockfile -import errno - -class TRexServerApp(object): - def __init__(self): - TRexServerApp.create_working_dirs() - self.stdin_path = '/dev/null' - self.stdout_path = '/dev/tty' # All standard prints will come up from this source. - self.stderr_path = "/var/log/trex/trex_daemon_server.log" # All log messages will come up from this source - self.pidfile_path = '/var/run/trex/trex_daemon_server.pid' - self.pidfile_timeout = 5 # timeout in seconds - - def run(self): - do_main_program() - - - @staticmethod - def create_working_dirs(): - if not os.path.exists('/var/log/trex'): - os.mkdir('/var/log/trex') - if not os.path.exists('/var/run/trex'): - os.mkdir('/var/run/trex') - - - -def main (): - - trex_app = TRexServerApp() - - # setup the logger - default_log_path = '/var/log/trex/trex_daemon_server.log' - - try: - CCustomLogger.setup_daemon_logger('TRexServer', default_log_path) - logger = logging.getLogger('TRexServer') - logger.setLevel(logging.INFO) - formatter = logging.Formatter("%(asctime)s %(name)-10s %(module)-20s %(levelname)-8s %(message)s") - handler = logging.FileHandler("/var/log/trex/trex_daemon_server.log") - logger.addHandler(handler) - except EnvironmentError, e: - if e.errno == errno.EACCES: # catching permission denied error - print "Launching user must have sudo privileges in order to run TRex daemon.\nTerminating daemon process." - exit(-1) - - daemon_runner = ExtendedDaemonRunner(trex_app, trex_parser) - - #This ensures that the logger file handle does not get closed during daemonization - daemon_runner.daemon_context.files_preserve=[handler.stream] - - try: - if not set(['start', 'stop']).isdisjoint(set(sys.argv)): - print "Logs are saved at: {log_path}".format( log_path = default_log_path ) - daemon_runner.do_action() - - except lockfile.LockTimeout as inst: - logger.error(inst) - print inst - print """ - Please try again once the timeout has been reached. - If this error continues, consider killing the process manually and restart the daemon.""" - - -if __name__ == "__main__": - main() diff --git a/scripts/automation/trex_control_plane/server/trex_server.py b/scripts/automation/trex_control_plane/server/trex_server.py index e32fc9d1..3dcb3e97 100755 --- a/scripts/automation/trex_control_plane/server/trex_server.py +++ b/scripts/automation/trex_control_plane/server/trex_server.py @@ -27,6 +27,8 @@ from zmq_monitor_thread import ZmqMonitorSession from argparse import ArgumentParser, RawTextHelpFormatter from json import JSONEncoder import re +import shlex +import tempfile # setup the logger @@ -52,6 +54,8 @@ class CTRexServer(object): trex_zmq_port : int the port number on which trex's zmq module will interact with daemon server default value: 4500 + nice: int + priority of the TRex process Instantiate a TRex client object, and connecting it to listening daemon-server """ @@ -76,7 +80,6 @@ class CTRexServer(object): raise Exception(err) def add(self, x, y): - print "server function add ",x,y logger.info("Processing add function. Parameters are: {0}, {1} ".format( x, y )) return x + y # return Fault(-10, "") @@ -123,38 +126,49 @@ class CTRexServer(object): raise # set further functionality and peripherals to server instance + self.server.register_function(self.add) + self.server.register_function(self.cancel_reservation) + self.server.register_function(self.connectivity_check) + self.server.register_function(self.force_trex_kill) + self.server.register_function(self.get_file) + self.server.register_function(self.get_files_list) + self.server.register_function(self.get_files_path) + self.server.register_function(self.get_running_info) + self.server.register_function(self.get_running_status) + self.server.register_function(self.get_trex_cmds) + self.server.register_function(self.get_trex_daemon_log) + self.server.register_function(self.get_trex_log) + self.server.register_function(self.get_trex_version) + self.server.register_function(self.is_reserved) + self.server.register_function(self.is_running) + self.server.register_function(self.kill_all_trexes) + self.server.register_function(self.push_file) + self.server.register_function(self.reserve_trex) + self.server.register_function(self.start_trex) + self.server.register_function(self.stop_trex) + self.server.register_function(self.wait_until_kickoff_finish) + signal.signal(signal.SIGTSTP, self.stop_handler) + signal.signal(signal.SIGTERM, self.stop_handler) try: - self.server.register_function(self.add) - self.server.register_function(self.cancel_reservation) - self.server.register_function(self.connectivity_check) - self.server.register_function(self.force_trex_kill) - self.server.register_function(self.get_file) - self.server.register_function(self.get_files_list) - self.server.register_function(self.get_files_path) - self.server.register_function(self.get_running_info) - self.server.register_function(self.get_running_status) - self.server.register_function(self.get_trex_daemon_log) - self.server.register_function(self.get_trex_log) - self.server.register_function(self.get_trex_version) - self.server.register_function(self.is_reserved) - self.server.register_function(self.is_running) - self.server.register_function(self.push_file) - self.server.register_function(self.reserve_trex) - self.server.register_function(self.start_trex) - self.server.register_function(self.stop_trex) - self.server.register_function(self.wait_until_kickoff_finish) - signal.signal(signal.SIGTSTP, self.stop_handler) - signal.signal(signal.SIGTERM, self.stop_handler) self.zmq_monitor.start() self.server.serve_forever() except KeyboardInterrupt: logger.info("Daemon shutdown request detected." ) - except Exception as e: - logger.error(e) finally: self.zmq_monitor.join() # close ZMQ monitor thread resources self.server.shutdown() - pass + #self.server.server_close() + + def _run_command(self, command, timeout = 15, cwd = None): + if timeout: + command = 'timeout %s %s' % (timeout, command) + # pipes might stuck, even with timeout + with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file: + proc = subprocess.Popen(shlex.split(command), stdout=stdout_file, stderr=stderr_file, cwd = cwd) + proc.wait() + stdout_file.seek(0) + stderr_file.seek(0) + return (proc.returncode, stdout_file.read().decode(errors = 'replace'), stderr_file.read().decode(errors = 'replace')) # get files from Trex server and return their content (mainly for logs) @staticmethod @@ -215,8 +229,7 @@ class CTRexServer(object): try: logger.info("Processing get_trex_version() command.") if not self.trex_version: - help_print = subprocess.Popen(['./t-rex-64', '--help'], cwd = self.TREX_PATH, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - (stdout, stderr) = help_print.communicate() + ret_code, stdout, stderr = self._run_command('./t-rex-64 --help', cwd = self.TREX_PATH, timeout = 0) search_result = re.search('\n\s*(Version\s*:.+)', stdout, re.DOTALL) if not search_result: raise Exception('Could not determine version from ./t-rex-64 --help') @@ -226,7 +239,7 @@ class CTRexServer(object): else: return binascii.a2b_base64(self.trex_version) except Exception as e: - err_str = "Can't get trex version, error: {0}".format(e) + err_str = "Can't get trex version, error: %s" % e logger.error(err_str) return Fault(-33, err_str) @@ -301,7 +314,7 @@ class CTRexServer(object): return False - def start_trex(self, trex_cmd_options, user, block_to_success = True, timeout = 40): + def start_trex(self, trex_cmd_options, user, block_to_success = True, timeout = 40, stateless = False): with self.start_lock: logger.info("Processing start_trex() command.") if self.is_reserved(): @@ -313,8 +326,8 @@ class CTRexServer(object): logger.info("TRex is already taken, cannot create another run until done.") return Fault(-13, '') # raise at client TRexInUseError - try: - server_cmd_data = self.generate_run_cmd(**trex_cmd_options) + try: + server_cmd_data = self.generate_run_cmd(stateless = stateless, **trex_cmd_options) self.zmq_monitor.first_dump = True self.trex.start_trex(self.TREX_PATH, server_cmd_data) logger.info("TRex session has been successfully initiated.") @@ -342,7 +355,7 @@ class CTRexServer(object): except TypeError as e: logger.error("TRex command generation failed, probably because either -f (traffic generation .yaml file) and -c (num of cores) was not specified correctly.\nReceived params: {params}".format( params = trex_cmd_options) ) - raise TypeError('TRex -f (traffic generation .yaml file) and -c (num of cores) must be specified.') + raise TypeError('TRex -f (traffic generation .yaml file) and -c (num of cores) must be specified. %s' % e) def stop_trex(self, seq): @@ -362,6 +375,41 @@ class CTRexServer(object): logger.info("Processing force_trex_kill() command. --> Killing TRex session indiscriminately.") return self.trex.stop_trex() + # returns list of tuples (pid, command line) of running TRex(es) + def get_trex_cmds(self): + logger.info('Processing get_trex_cmds() command.') + ret_code, stdout, stderr = self._run_command('ps -u root --format pid,comm,cmd') + if ret_code: + raise Exception('Failed to determine running processes, stderr: %s' % stderr) + trex_cmds_list = [] + for line in stdout.splitlines(): + pid, proc_name, full_cmd = line.strip().split(' ', 2) + pid = pid.strip() + full_cmd = full_cmd.strip() + if proc_name.find('t-rex-64') >= 0: + trex_cmds_list.append((pid, full_cmd)) + return trex_cmds_list + + + def kill_all_trexes(self): + logger.info('Processing kill_all_trexes() command.') + trex_cmds_list = self.get_trex_cmds() + if not trex_cmds_list: + return False + for pid, cmd in trex_cmds_list: + logger.info('Killing process %s %s' % (pid, cmd)) + self._run_command('kill %s' % pid) + ret_code_ps, _, _ = self._run_command('ps -p %s' % pid) + if not ret_code_ps: + logger.info('Killing with -9.') + self._run_command('kill -9 %s' % pid) + ret_code_ps, _, _ = self._run_command('ps -p %s' % pid) + if not ret_code_ps: + logger.info('Could not kill process.') + raise Exception('Could not kill process %s %s' % (pid, cmd)) + return True + + def wait_until_kickoff_finish (self, timeout = 40): # block until TRex exits Starting state logger.info("Processing wait_until_kickoff_finish() command.") @@ -377,14 +425,19 @@ class CTRexServer(object): logger.info("Processing get_running_info() command.") return self.trex.get_running_info() - def generate_run_cmd (self, f, d, iom = 0, export_path="/tmp/trex.txt", **kwargs): - """ generate_run_cmd(self, trex_cmd_options, export_path) -> str + + def generate_run_cmd (self, iom = 0, export_path="/tmp/trex.txt", stateless = False, **kwargs): + """ generate_run_cmd(self, iom, export_path, kwargs) -> str Generates a custom running command for the kick-off of the TRex traffic generator. Returns a tuple of command (string) and export path (string) to be issued on the trex server Parameters ---------- + iom: int + 0 = don't print stats screen to log, 1 = print stats (can generate huge logs) + stateless: boolean + True = run as stateless, False = require -f and -d arguments kwargs: dictionary Dictionary of parameters for trex. For example: (c=1, nc=True, l_pkt_mode=3). Notice that when sending command line parameters that has -, you need to replace it with _. @@ -396,7 +449,8 @@ class CTRexServer(object): if 'results_file_path' in kwargs: export_path = kwargs['results_file_path'] del kwargs['results_file_path'] - + if stateless: + kwargs['i'] = True # adding additional options to the command trex_cmd_options = '' @@ -408,18 +462,23 @@ class CTRexServer(object): else: trex_cmd_options += (dash + '{k} {val}'.format( k = tmp_key, val = value )) - cmd = "{nice}{run_command} -f {gen_file} -d {duration} --iom {io} {cmd_options} --no-key > {export}".format( # -- iom 0 disables the periodic log to the screen (not needed) + if not stateless: + if 'f' not in kwargs: + raise Exception('Argument -f should be specified in stateful command') + if 'd' not in kwargs: + raise Exception('Argument -d should be specified in stateful command') + + cmd = "{nice}{run_command} --iom {io} {cmd_options} --no-key > {export}".format( # -- iom 0 disables the periodic log to the screen (not needed) nice = '' if self.trex_nice == 0 else 'nice -n %s ' % self.trex_nice, run_command = self.TREX_START_CMD, - gen_file = f, - duration = d, cmd_options = trex_cmd_options, io = iom, export = export_path ) logger.info("TREX FULL COMMAND: {command}".format(command = cmd) ) - return (cmd, export_path, long(d)) + return (cmd, export_path, kwargs.get('d', 0)) + def __check_trex_path_validity(self): # check for executable existance -- cgit 1.2.3-korg