diff options
Diffstat (limited to 'scripts/automation/trex_control_plane')
5 files changed, 235 insertions, 29 deletions
diff --git a/scripts/automation/trex_control_plane/server/singleton_daemon.py b/scripts/automation/trex_control_plane/server/singleton_daemon.py new file mode 100755 index 00000000..f0a61f54 --- /dev/null +++ b/scripts/automation/trex_control_plane/server/singleton_daemon.py @@ -0,0 +1,158 @@ +import errno +import os +import shlex +import socket +import tempfile +import types +from subprocess import Popen +from time import sleep + +# uses Unix sockets for determine running process. +# (assumes used daemons will register proper socket) +# all daemons should use -p argument as listening tcp port +class SingletonDaemon(object): + + # run_cmd can be function of how to run daemon or a str to run at subprocess + def __init__(self, name, tag, port, run_cmd, dir = None): + self.name = name + self.tag = tag + self.port = port + self.run_cmd = run_cmd + self.dir = dir + self.stop = self.kill # alias + if ' ' in tag: + raise Exception('Error: tag should not include spaces') + if dir and not os.path.exists(dir): + print('Warning: path given for %s: %s, does not exist' % (name, dir)) + + + # returns True if daemon is running + def is_running(self): + lock_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + try: + lock_socket.bind('\0' + self.tag) # the check is ~200000 faster and more reliable than checking via 'netstat' or 'ps' etc. + lock_socket.close() + except socket.error: # Unix socket in use + return True + # Unix socket is not used, but maybe it's old version of daemon not using socket + return bool(self.get_pid()) + + + # get pid of running daemon by registered Unix socket (most robust way) + def get_pid_by_unix_socket(self): + ret_code, stdout, stderr = run_command('netstat -px') + if ret_code: + raise Exception('Error running netstat: %s' % [ret_code, stdout, stderr]) + for line in stdout.splitlines(): + line_arr = line.strip().split() + if len(line_arr) == 8 and line_arr[0] == 'unix' and line_arr[4] == 'DGRAM' and line_arr[7] == '@%s' % self.tag: + return int(line_arr[6].split('/', 1)[0]) + + + # get pid of running daemon by listening tcp port (for backward compatibility) + def get_pid_by_listening_port(self): + ret_code, stdout, stderr = run_command('netstat -tlnp') + if ret_code: + raise Exception('Error running netstat: %s' % [ret_code, stdout, stderr]) + for line in stdout.splitlines(): + line_arr = line.strip().split() + if len(line_arr) == 7 and line_arr[3] == '0.0.0.0:%s' % self.port: + if '/' not in line_arr[6]: + raise Exception('Expecting pid/program name in netstat line of using port %s, got: %s' % (self.port, line)) + return int(line_arr[6].split('/')[0]) + + + # get PID of running process, None if not found + def get_pid(self): + pid = self.get_pid_by_unix_socket() + if pid: + return pid + pid = self.get_pid_by_listening_port() + if pid: + return pid + + + # kill daemon + def kill(self, timeout = 5): + pid = self.get_pid() + if not pid: + return False + ret_code, stdout, stderr = run_command('kill %s' % pid) # usual kill + if ret_code: + raise Exception('Failed to run kill command for %s: %s' % (self.name, [ret_code, stdout, stderr])) + poll_rate = 0.1 + for i in range(int(timeout / poll_rate)): + if not self.is_running(): + return True + sleep(poll_rate) + ret_code, stdout, stderr = run_command('kill -9 %s' % pid) # unconditional kill + if ret_code: + raise Exception('Failed to run kill -9 command for %s: %s' % (self.name, [ret_code, stdout, stderr])) + poll_rate = 0.1 + for i in range(inr(timeout / poll_rate)): + if not self.is_running(): + return True + sleep(poll_rate) + raise Exception('Could not kill %s, even with -9' % self.name) + + + # start daemon + # returns True if success, False if already running + def start(self, timeout = 5): + if self.is_running(): + raise Exception('%s is already running' % self.name) + if not self.run_cmd: + raise Exception('No starting command registered for %s' % self.name) + if type(self.run_cmd) is types.FunctionType: + self.run_cmd() + return + with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file: + proc = Popen(shlex.split('%s -p %s' % (self.run_cmd, self.port)), cwd = self.dir, close_fds = True, + stdout = stdout_file, stderr = stderr_file) + if timeout > 0: + poll_rate = 0.1 + for i in range(int(timeout/poll_rate)): + sleep(poll_rate) + if bool(proc.poll()): # process ended with error + stdout_file.seek(0) + stderr_file.seek(0) + raise Exception('Run of %s ended unexpectfully: %s' % (self.name, [proc.returncode, stdout_file.read().decode(errors = 'replace'), stderr_file.read().decode(errors = 'replace')])) + elif proc.poll() == 0: # process runs other process, and ended + break + if self.is_running(): + return True + raise Exception('%s failed to run.' % self.name) + + # restart the daemon + def restart(self, timeout = 5): + if self.is_running(): + self.kill(timeout) + self.start(timeout) + + +# provides unique way to determine running process, should be used inside daemon +def register_socket(tag): + global lock_socket # Without this our lock gets garbage collected + lock_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + try: + lock_socket.bind('\0%s' % tag) + except socket.error: + raise Exception('Error: process with tag %s is already running.' % tag) + +# runs command +def run_command(command, timeout = 15, cwd = None): + # pipes might stuck, even with timeout + with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file: + proc = Popen(shlex.split(command), stdout = stdout_file, stderr = stderr_file, cwd = cwd, close_fds = True) + if timeout > 0: + poll_rate = 0.1 + for i in range(int(timeout/poll_rate)): + sleep(poll_rate) + if proc.poll() is not None: # process stopped + break + if proc.poll() is None: + proc.kill() # timeout + return (errno.ETIME, '', 'Timeout on running: %s' % command) + stdout_file.seek(0) + stderr_file.seek(0) + return (proc.returncode, stdout_file.read().decode(errors = 'replace'), stderr_file.read().decode(errors = 'replace')) diff --git a/scripts/automation/trex_control_plane/server/trex_server.py b/scripts/automation/trex_control_plane/server/trex_server.py index c1760236..45ef9ac1 100755 --- a/scripts/automation/trex_control_plane/server/trex_server.py +++ b/scripts/automation/trex_control_plane/server/trex_server.py @@ -29,6 +29,11 @@ import re import shlex import tempfile +try: + from .singleton_daemon import register_socket +except: + from singleton_daemon import register_socket + # setup the logger CCustomLogger.setup_custom_logger('TRexServer') @@ -102,6 +107,7 @@ class CTRexServer(object): def start(self): """This method fires up the daemon server based on initialized parameters of the class""" # initialize the server instance with given resources + register_socket('trex_daemon_server') try: print "Firing up TRex REST daemon @ port {trex_port} ...\n".format( trex_port = self.trex_daemon_port ) logger.info("Firing up TRex REST daemon @ port {trex_port} ...".format( trex_port = self.trex_daemon_port )) diff --git a/scripts/automation/trex_control_plane/stl/examples/rpc_proxy_server.py b/scripts/automation/trex_control_plane/stl/examples/rpc_proxy_server.py index 39d642f4..ad2697d8 100755 --- a/scripts/automation/trex_control_plane/stl/examples/rpc_proxy_server.py +++ b/scripts/automation/trex_control_plane/stl/examples/rpc_proxy_server.py @@ -6,14 +6,18 @@ import logging import sys import os import json +import socket +from functools import partial logging.basicConfig(level = logging.FATAL) # keep quiet import stl_path from trex_stl_lib.api import * -from trex_stl_lib.trex_stl_hltapi import CTRexHltApi, HLT_ERR +from trex_stl_lib.trex_stl_hltapi import CTRexHltApi, HLT_OK, HLT_ERR # ext libs -ext_libs = os.path.join(os.pardir, os.pardir, os.pardir, os.pardir, 'external_libs') +ext_libs = os.path.join(os.pardir, os.pardir, os.pardir, os.pardir, 'external_libs') # usual package path +if not os.path.exists(ext_libs): + ext_libs = os.path.join(os.pardir, os.pardir, 'external_libs') # client package path sys.path.append(os.path.join(ext_libs, 'jsonrpclib-pelix-0.2.5')) from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer import yaml @@ -62,18 +66,17 @@ def native_proxy_del(): def hltapi_proxy_init(force = False, *args, **kwargs): global hltapi_client if hltapi_client and not force: - return ERR('HLTAPI Client is already initiated') + return HLT_ERR('HLTAPI Client is already initiated') try: hltapi_client = CTRexHltApi(*args, **kwargs) - return OK('HLTAPI Client initiated') + return HLT_OK() except: - return ERR(traceback.format_exc()) + return HLT_ERR(traceback.format_exc()) def hltapi_proxy_del(): global hltapi_client hltapi_client = None - return OK() - + return HLT_OK() # any method not listed above can be called with passing its name here def native_method(func_name, *args, **kwargs): @@ -94,14 +97,7 @@ def hltapi_method(func_name, *args, **kwargs): ### /Server functions ### -### Main ### - - -if __name__ == '__main__': - parser = argparse.ArgumentParser(description = 'Runs TRex Stateless proxy for usage with any language client.') - parser.add_argument('-p', '--port', type=int, default = 8095, dest='port', action = 'store', - help = 'Select port on which the stl proxy will run.\nDefault is 8095.') - args = parser.parse_args() +def run_server(port = 8095): native_methods = [ 'acquire', 'connect', @@ -122,25 +118,50 @@ if __name__ == '__main__': ] try: - server = SimpleJSONRPCServer(('0.0.0.0', args.port)) + register_socket('trex_stl_rpc_proxy') + server = SimpleJSONRPCServer(('0.0.0.0', port)) server.register_function(add) server.register_function(check_connectivity) server.register_function(native_proxy_init) server.register_function(native_proxy_del) server.register_function(hltapi_proxy_init) server.register_function(hltapi_proxy_del) + server.register_function(native_method) + server.register_function(hltapi_method) for method in native_methods: - server.register_function(lambda method=method, *args, **kwargs: native_method(method, *args, **kwargs), method) - server.register_function(native_method) + server.register_function(partial(native_method, method), method) for method in hltapi_methods: - if method == 'connect': - server.register_function(lambda method=method, *args, **kwargs: hltapi_method(method, *args, **kwargs), 'hlt_connect') + if method in native_methods: # collision in names + method_hlt_name = 'hlt_%s' % method else: - server.register_function(lambda method=method, *args, **kwargs: hltapi_method(method, *args, **kwargs), method) - server.register_function(hltapi_method) - print('Started Stateless RPC proxy at port %s' % args.port) + method_hlt_name = method + server.register_function(partial(hltapi_method, method), method_hlt_name) + server.register_function(server.funcs.keys, 'get_methods') # should be last + print('Started Stateless RPC proxy at port %s' % port) server.serve_forever() except KeyboardInterrupt: print('Done') +# provides unique way to determine running process +def register_socket(tag): + global foo_socket # Without this our lock gets garbage collected + foo_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + try: + foo_socket.bind('\0%s' % tag) + print('Got the socket lock for tag %s.' % tag) + except socket.error: + print('Error: process with tag %s is already running.' % tag) + sys.exit(-1) + +### Main ### + + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description = 'Runs TRex Stateless RPC proxy for usage with any language client.') + parser.add_argument('-p', '--port', type=int, default = 8095, dest='port', action = 'store', + help = 'Select port on which the stl rpc proxy will run.\nDefault is 8095.') + kwargs = vars(parser.parse_args()) + run_server(**kwargs) + diff --git a/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py b/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py index 82bf0d0a..065f4284 100755 --- a/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py +++ b/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py @@ -32,9 +32,29 @@ if __name__ == '__main__': help = 'Address of rpc proxy.') parser.add_argument('-p', '--port', type=int, default = 8095, dest='port', action = 'store', help = 'Port of rpc proxy.\nDefault is 8095.') + parser.add_argument('--master_port', type=int, default = 8091, dest='master_port', action = 'store', + help = 'Port of Master daemon.\nDefault is 8091.') args = parser.parse_args() server = jsonrpclib.Server('http://%s:%s' % (args.server, args.port)) + master = jsonrpclib.Server('http://%s:%s' % (args.server, args.master_port)) + +# Connecting + + try: + print('Connecting to STL RPC proxy server') + server.check_connectivity() + print('Connected') + except Exception as e: + print('Could not connect to STL RPC proxy server: %s\nTrying to start it from Master daemon.' % e) + try: + master.check_connectivity() + master.start_stl_rpc_proxy() + print('Started') + except Exception as e: + print('Could not start it from Master daemon. Error: %s' % e) + sys.exit(-1) + # Native API @@ -68,7 +88,8 @@ if __name__ == '__main__': # HLTAPI print('Initializing HLTAPI Client') - verify(server.hltapi_proxy_init(force = True)) + verify_hlt(server.hltapi_proxy_init(force = True)) + print('HLTAPI Client initiated') print('HLTAPI connect') verify_hlt(server.hlt_connect(device = args.server, port_list = ports, reset = True, break_locks = True)) @@ -98,4 +119,4 @@ if __name__ == '__main__': print(res) print('Deleting HLTAPI Client instance') - verify(server.hltapi_proxy_del()) + verify_hlt(server.hltapi_proxy_del()) diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_hltapi.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_hltapi.py index 33a7b3af..0f68d4a3 100755 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_hltapi.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_hltapi.py @@ -165,7 +165,7 @@ traffic_control_kwargs = { } traffic_stats_kwargs = { - 'mode': 'aggregate', # ( aggregate ) + 'mode': 'aggregate', # ( all | aggregate | streams ) 'port_handle': None, } @@ -1529,11 +1529,11 @@ def get_TOS(user_kwargs, kwargs): TOS1 = set(['ip_precedence', 'ip_delay', 'ip_throughput', 'ip_reliability', 'ip_cost', 'ip_reserved']) TOS2 = set(['ip_dscp', 'ip_cu']) user_args = set(user_kwargs.keys()) - if user_args & (TOS1 - TOS0) and user_args & (TOS0 - TOS1): + if user_args & TOS0.symmetric_difference(TOS1): raise STLError('You have mixed %s and %s TOS parameters' % (TOS0, TOS1)) - if user_args & (TOS2 - TOS0) and user_args & (TOS0 - TOS2): + if user_args & TOS0.symmetric_difference(TOS2): raise STLError('You have mixed %s and %s TOS parameters' % (TOS0, TOS2)) - if user_args & (TOS2 - TOS1) and user_args & (TOS1 - TOS2): + if user_args & TOS1.symmetric_difference(TOS2): raise STLError('You have mixed %s and %s TOS parameters' % (TOS1, TOS2)) if user_args & (TOS0 - TOS1 - TOS2): return (kwargs['ip_precedence'] << 5) + (kwargs['ip_tos_field'] << 2) + kwargs['ip_mbz'] |