diff options
7 files changed, 1333 insertions, 1073 deletions
diff --git a/scripts/automation/trex_control_plane/client/trex_async_client.py b/scripts/automation/trex_control_plane/client/trex_async_client.py index 2bb0e9cd..9828c838 100644 --- a/scripts/automation/trex_control_plane/client/trex_async_client.py +++ b/scripts/automation/trex_control_plane/client/trex_async_client.py @@ -144,12 +144,15 @@ class CTRexAsyncStatsManager(): class CTRexAsyncClient(): - def __init__ (self, server, port, stateless_client, prn_func = None): + def __init__ (self, server, port, stateless_client): self.port = port self.server = server + self.stateless_client = stateless_client - self.prn_func = prn_func + + self.event_handler = stateless_client.event_handler + self.logger = self.stateless_client.logger self.raw_snapshot = {} @@ -170,10 +173,7 @@ class CTRexAsyncClient(): msg = "\nConnecting To ZMQ Publisher On {0}".format(self.tr) - if self.prn_func: - self.prn_func(msg) - else: - print msg + self.logger.log(msg) # Socket to talk to server self.context = zmq.Context() @@ -235,7 +235,7 @@ class CTRexAsyncClient(): # signal once if not got_data: - self.stateless_client.on_async_alive() + self.event_handler.on_async_alive() got_data = True @@ -244,7 +244,7 @@ class CTRexAsyncClient(): # signal once if got_data: - self.stateless_client.on_async_dead() + self.event_handler.on_async_dead() got_data = False continue @@ -284,11 +284,11 @@ class CTRexAsyncClient(): def __dispatch (self, name, type, data): # stats if name == "trex-global": - self.stateless_client.handle_async_stats_update(data) + self.event_handler.handle_async_stats_update(data) # events elif name == "trex-event": - self.stateless_client.handle_async_event(type, data) + self.event_handler.handle_async_event(type, data) # barriers elif name == "trex-barrier": @@ -315,7 +315,7 @@ class CTRexAsyncClient(): # add to the queue self.async_barriers.append(barrier) - rc = self.stateless_client.transmit("publish_now", params = {'key' : key}) + rc = self.stateless_client._transmit("publish_now", params = {'key' : key}) if not rc: return rc diff --git a/scripts/automation/trex_control_plane/client/trex_stateless_client.py b/scripts/automation/trex_control_plane/client/trex_stateless_client.py index 105c4d01..43912e55 100755 --- a/scripts/automation/trex_control_plane/client/trex_stateless_client.py +++ b/scripts/automation/trex_control_plane/client/trex_stateless_client.py @@ -25,6 +25,26 @@ from trex_port import Port from common.trex_types import * from trex_async_client import CTRexAsyncClient +############################ logger ############################# +############################ ############################# +############################ ############################# + +class STLFailure(Exception): + def __init__ (self, rc_or_str): + self.msg = str(rc_or_str) + + def __str__ (self): + exc_type, exc_obj, exc_tb = sys.exc_info() + fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1] + + + s = "\n******\n" + s += "Error reported at {0}:{1}\n\n".format(format_text(fname, 'bold'), format_text(exc_tb.tb_lineno), 'bold') + s += "specific error:\n\n'{0}'\n".format(format_text(self.msg, 'bold')) + + return s + + # logger API for the client class LoggerApi(object): # verbose levels @@ -35,9 +55,11 @@ class LoggerApi(object): def __init__(self): self.level = LoggerApi.VERBOSE_REGULAR + # implemented by specific logger def write(self, msg, newline = True): raise Exception("implement this") + # implemented by specific logger def flush(self): raise Exception("implement this") @@ -62,10 +84,15 @@ class LoggerApi(object): self.write(msg, newline) + # logging that comes from async event + def async_log (self, msg, level = VERBOSE_REGULAR, newline = True): + self.log(msg, level, newline) + # annotates an action with a RC - writes to log the result def annotate (self, rc, desc = None, show_status = True): rc.annotate(self.log, desc, show_status) + # default logger - to stdout class DefaultLogger(LoggerApi): def write (self, msg, newline = True): @@ -78,90 +105,41 @@ class DefaultLogger(LoggerApi): sys.stdout.flush() -class CTRexStatelessClient(object): - """docstring for CTRexStatelessClient""" - - def __init__(self, - username = general_utils.get_current_user(), - server = "localhost", - sync_port = 4501, - async_port = 4500, - verbose_level = LoggerApi.VERBOSE_REGULAR, - virtual = False, - logger = None): - - super(CTRexStatelessClient, self).__init__() +############################ async event hander ############################# +############################ ############################# +############################ ############################# - self.user = username +# handles different async events given to the client +class AsyncEventHandler(object): - # logger - if not logger: - self.logger = DefaultLogger() - else: - self.logger = logger - - # initial verbose - self.logger.set_verbose(verbose_level) - - self.comm_link = CTRexStatelessClient.CCommLink(server, sync_port, virtual, self.logger) - - self.ports = {} - self._connection_info = {"server": server, - "sync_port": sync_port, - "async_port": async_port} - self.system_info = {} - self.server_version = {} - - self.async_client = CTRexAsyncClient(server, async_port, self, self.logger.log) - - self.streams_db = CStreamsDB() - self.global_stats = trex_stats.CGlobalStats(self._connection_info, - self.server_version, - self.ports) - self.stats_generator = trex_stats.CTRexInfoGenerator(self.global_stats, - self.ports) + def __init__ (self, client): + self.client = client + self.logger = self.client.logger self.events = [] - self.session_id = random.getrandbits(32) - self.read_only = False - self.connected = False - self.prompt_redraw_cb = None - - - # returns the port object - def get_port (self, port_id): - return self.ports.get(port_id, None) - + # public functions - # connection server ip - def get_server_ip (self): - return self.comm_link.get_server() + def get_events (self): + return self.events - # connection server port - def get_server_port (self): - return self.comm_link.get_port() + def clear_events (self): + self.events = [] - ################# events handler ###################### - def add_event_log (self, msg, ev_type, show = False): - if ev_type == "server": - prefix = "[server]" - elif ev_type == "local": - prefix = "[local]" + def on_async_dead (self): + if self.client.connected: + msg = 'lost connection to server' + self.__add_event_log(msg, 'local', True) + self.client.connected = False - ts = time.time() - st = datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S') - self.events.append("{:<10} - {:^8} - {:}".format(st, prefix, format_text(msg, 'bold'))) - if show: - self.logger.log(format_text("\n\n{:^8} - {:}".format(prefix, format_text(msg, 'bold')))) - if self.prompt_redraw_cb and self.logger.check_verbose(self.logger.VERBOSE_REGULAR): - self.prompt_redraw_cb() + def on_async_alive (self): + pass - + # handles an async stats update from the subscriber def handle_async_stats_update(self, dump_data): global_stats = {} port_stats = {} @@ -173,7 +151,7 @@ class CTRexStatelessClient(object): if m: port_id = int(m.group(2)) field_name = m.group(1) - if self.ports.has_key(port_id): + if self.client.ports.has_key(port_id): if not port_id in port_stats: port_stats[port_id] = {} port_stats[port_id][field_name] = value @@ -184,13 +162,14 @@ class CTRexStatelessClient(object): global_stats[key] = value # update the general object with the snapshot - self.global_stats.update(global_stats) + self.client.global_stats.update(global_stats) + # update all ports for port_id, data in port_stats.iteritems(): - self.ports[port_id].port_stats.update(data) - + self.client.ports[port_id].port_stats.update(data) + # dispatcher for server async events (port started, port stopped and etc.) def handle_async_event (self, type, data): # DP stopped @@ -200,7 +179,7 @@ class CTRexStatelessClient(object): if (type == 0): port_id = int(data['port_id']) ev = "Port {0} has started".format(port_id) - self.async_event_port_started(port_id) + self.__async_event_port_started(port_id) # port stopped elif (type == 1): @@ -208,8 +187,8 @@ class CTRexStatelessClient(object): ev = "Port {0} has stopped".format(port_id) # call the handler - self.async_event_port_stopped(port_id) - + self.__async_event_port_stopped(port_id) + # port paused elif (type == 2): @@ -217,7 +196,7 @@ class CTRexStatelessClient(object): ev = "Port {0} has paused".format(port_id) # call the handler - self.async_event_port_paused(port_id) + self.__async_event_port_paused(port_id) # port resumed elif (type == 3): @@ -225,7 +204,7 @@ class CTRexStatelessClient(object): ev = "Port {0} has resumed".format(port_id) # call the handler - self.async_event_port_resumed(port_id) + self.__async_event_port_resumed(port_id) # port finished traffic elif (type == 4): @@ -233,7 +212,7 @@ class CTRexStatelessClient(object): ev = "Port {0} job done".format(port_id) # call the handler - self.async_event_port_stopped(port_id) + self.__async_event_port_stopped(port_id) show_event = True # port was stolen... @@ -241,7 +220,7 @@ class CTRexStatelessClient(object): session_id = data['session_id'] # false alarm, its us - if session_id == self.session_id: + if session_id == self.client.session_id: return port_id = int(data['port_id']) @@ -250,13 +229,13 @@ class CTRexStatelessClient(object): ev = "Port {0} was forcely taken by '{1}'".format(port_id, who) # call the handler - self.async_event_port_forced_acquired(port_id) + self.__async_event_port_forced_acquired(port_id) show_event = True # server stopped elif (type == 100): ev = "Server has stopped" - self.async_event_server_stopped() + self.__async_event_server_stopped() show_event = True @@ -265,323 +244,243 @@ class CTRexStatelessClient(object): return - self.add_event_log(ev, 'server', show_event) - - - def async_event_port_stopped (self, port_id): - self.ports[port_id].async_event_port_stopped() - - - def async_event_port_started (self, port_id): - self.ports[port_id].async_event_port_started() - - - def async_event_port_paused (self, port_id): - self.ports[port_id].async_event_port_paused() - - - def async_event_port_resumed (self, port_id): - self.ports[port_id].async_event_port_resumed() - - - def async_event_port_forced_acquired (self, port_id): - self.ports[port_id].async_event_forced_acquired() - self.read_only = True - - def async_event_server_stopped (self): - self.connected = False - - - def get_events (self): - return self.events - - def clear_events (self): - self.events = [] - - ############# helper functions section ############## - - # measure time for functions - def timing(f): - def wrap(*args): - - time1 = time.time() - ret = f(*args) - - # don't want to print on error - if ret.bad(): - return ret - - delta = time.time() - time1 - - client = args[0] - client.logger.log(format_time(delta) + "\n") - - return ret - - return wrap - - - def validate_port_list(self, port_id_list): - if not isinstance(port_id_list, list): - print type(port_id_list) - return False - - # check each item of the sequence - return all([ (port_id >= 0) and (port_id < self.get_port_count()) - for port_id in port_id_list ]) - - # some preprocessing for port argument - def __ports (self, port_id_list): - - # none means all - if port_id_list == None: - return range(0, self.get_port_count()) - - # always list - if isinstance(port_id_list, int): - port_id_list = [port_id_list] - - if not isinstance(port_id_list, list): - raise ValueError("bad port id list: {0}".format(port_id_list)) - - for port_id in port_id_list: - if not isinstance(port_id, int) or (port_id < 0) or (port_id > self.get_port_count()): - raise ValueError("bad port id {0}".format(port_id)) - - return port_id_list + self.__add_event_log(ev, 'server', show_event) - ############ boot up section ################ + # private functions - # mode can be RW - read / write, RWF - read write with force , RO - read only - def connect(self, mode = "RW"): + def __async_event_port_stopped (self, port_id): + self.client.ports[port_id].async_event_port_stopped() - if self.is_connected(): - self.disconnect() - # clear this flag - self.connected = False + def __async_event_port_started (self, port_id): + self.client.ports[port_id].async_event_port_started() - # connect sync channel - rc = self.comm_link.connect() - if rc.bad(): - return rc + def __async_event_port_paused (self, port_id): + self.client.ports[port_id].async_event_port_paused() - # connect async channel - rc = self.async_client.connect() - if rc.bad(): - return rc - # version - rc = self.transmit("get_version") - if rc.bad(): - return rc + def __async_event_port_resumed (self, port_id): + self.client.ports[port_id].async_event_port_resumed() - self.server_version = rc.data() - self.global_stats.server_version = rc.data() - # cache system info - rc = self.transmit("get_system_info") - if rc.bad(): - return rc + def __async_event_port_forced_acquired (self, port_id): + self.client.ports[port_id].async_event_forced_acquired() + self.client.read_only = True - self.system_info = rc.data() + def __async_event_server_stopped (self): + self.client.connected = False - # cache supported commands - rc = self.transmit("get_supported_cmds") - if rc.bad(): - return rc - self.supported_cmds = rc.data() + # add event to log + def __add_event_log (self, msg, ev_type, show = False): - # create ports - for port_id in xrange(self.get_port_count()): - speed = self.system_info['ports'][port_id]['speed'] - driver = self.system_info['ports'][port_id]['driver'] - - self.ports[port_id] = Port(port_id, speed, driver, self.user, self.comm_link, self.session_id) - - - # sync the ports - rc = self.sync_ports() - if rc.bad(): - return rc + if ev_type == "server": + prefix = "[server]" + elif ev_type == "local": + prefix = "[local]" - # acquire all ports - if mode == "RW": - rc = self.acquire(force = False) + ts = time.time() + st = datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S') + self.events.append("{:<10} - {:^8} - {:}".format(st, prefix, format_text(msg, 'bold'))) - # fallback to read only if failed - if rc.bad(): - self.annotate(rc, show_status = False) - self.logger.log(format_text("Switching to read only mode - only few commands will be available", 'bold')) + if show: + self.logger.async_log(format_text("\n\n{:^8} - {:}".format(prefix, format_text(msg, 'bold')))) - self.release(self.get_acquired_ports()) - self.read_only = True - else: - self.read_only = False - elif mode == "RWF": - rc = self.acquire(force = True) - if rc.bad(): - return rc - self.read_only = False + - elif mode == "RO": - # no acquire on read only - rc = RC_OK() - self.read_only = True +############################ RPC layer ############################# +############################ ############################# +############################ ############################# - - self.connected = True - return RC_OK() +class CCommLink(object): + """describes the connectivity of the stateless client method""" + def __init__(self, server="localhost", port=5050, virtual=False, prn_func = None): + self.virtual = virtual + self.server = server + self.port = port + self.rpc_link = JsonRpcClient(self.server, self.port, prn_func) + @property + def is_connected(self): + if not self.virtual: + return self.rpc_link.connected + else: + return True - def is_read_only (self): - return self.read_only + def get_server (self): + return self.server - def is_connected (self): - return self.connected and self.comm_link.is_connected + def get_port (self): + return self.port + def connect(self): + if not self.virtual: + return self.rpc_link.connect() def disconnect(self): - # release any previous acquired ports - if self.is_connected(): - self.release(self.get_acquired_ports()) - - self.comm_link.disconnect() - self.async_client.disconnect() + if not self.virtual: + return self.rpc_link.disconnect() - self.connected = False - - return RC_OK() - - - def on_async_dead (self): - if self.connected: - msg = 'lost connection to server' - self.add_event_log(msg, 'local', True) - self.connected = False + def transmit(self, method_name, params={}): + if self.virtual: + self._prompt_virtual_tx_msg() + _, msg = self.rpc_link.create_jsonrpc_v2(method_name, params) + print msg + return + else: + return self.rpc_link.invoke_rpc_method(method_name, params) - def on_async_alive (self): - pass + def transmit_batch(self, batch_list): + if self.virtual: + self._prompt_virtual_tx_msg() + print [msg + for _, msg in [self.rpc_link.create_jsonrpc_v2(command.method, command.params) + for command in batch_list]] + else: + batch = self.rpc_link.create_batch() + for command in batch_list: + batch.add(command.method, command.params) + # invoke the batch + return batch.invoke() - ########### cached queries (no server traffic) ########### + def _prompt_virtual_tx_msg(self): + print "Transmitting virtually over tcp://{server}:{port}".format(server=self.server, + port=self.port) - def get_supported_cmds(self): - return self.supported_cmds - def get_version(self): - return self.server_version - def get_system_info(self): - return self.system_info +############################ client ############################# +############################ ############################# +############################ ############################# - def get_port_count(self): - return self.system_info.get("port_count") +class CTRexStatelessClient(object): + """docstring for CTRexStatelessClient""" - def get_port_ids(self, as_str=False): - port_ids = range(self.get_port_count()) - if as_str: - return " ".join(str(p) for p in port_ids) - else: - return port_ids + def __init__(self, + username = general_utils.get_current_user(), + server = "localhost", + sync_port = 4501, + async_port = 4500, + verbose_level = LoggerApi.VERBOSE_REGULAR, + logger = None, + virtual = False): - def get_stats_async (self): - return self.async_client.get_stats() - def get_connection_port (self): - return self.comm_link.port + self.username = username + + # init objects + self.ports = {} + self.server_version = {} + self.system_info = {} + self.session_id = random.getrandbits(32) + self.read_only = False + self.connected = False - def get_connection_ip (self): - return self.comm_link.server + # logger + self.logger = DefaultLogger() if not logger else logger - def get_all_ports (self): - return [port_id for port_id, port_obj in self.ports.iteritems()] + # initial verbose + self.logger.set_verbose(verbose_level) - def get_acquired_ports(self): - return [port_id - for port_id, port_obj in self.ports.iteritems() - if port_obj.is_acquired()] + # low level RPC layer + self.comm_link = CCommLink(server, + sync_port, + virtual, + self.logger) - def get_active_ports(self): - return [port_id - for port_id, port_obj in self.ports.iteritems() - if port_obj.is_active()] + # async event handler manager + self.event_handler = AsyncEventHandler(self) - def get_paused_ports (self): - return [port_id - for port_id, port_obj in self.ports.iteritems() - if port_obj.is_paused()] + # async subscriber level + self.async_client = CTRexAsyncClient(server, + async_port, + self) - def get_transmitting_ports (self): - return [port_id - for port_id, port_obj in self.ports.iteritems() - if port_obj.is_transmitting()] + + - def set_verbose (self, level): - self.logger.set_verbose(level) + # stats + self.connection_info = {"username": username, + "server": server, + "sync_port": sync_port, + "async_port": async_port, + "virtual": virtual} - def get_verbose (self): - return self.logger.get_verbose() + + self.global_stats = trex_stats.CGlobalStats(self.connection_info, + self.server_version, + self.ports) - def set_prompt_redraw_cb(self, cb): - self.prompt_redraw_cb = cb + self.stats_generator = trex_stats.CTRexInfoGenerator(self.global_stats, + self.ports) + # stream DB + self.streams_db = CStreamsDB() - def annotate (self, rc, desc = None, show_status = True): + + + ############# private functions - used by the class itself ########### - rc.annotate(self.logger.log, desc, show_status) + # some preprocessing for port argument + def __ports (self, port_id_list): - ############# server actions ################ + # none means all + if port_id_list == None: + return range(0, self.get_port_count()) - # ping server - def ping(self): - return self.transmit("ping") + # always list + if isinstance(port_id_list, int): + port_id_list = [port_id_list] + if not isinstance(port_id_list, list): + raise ValueError("bad port id list: {0}".format(port_id_list)) + for port_id in port_id_list: + if not isinstance(port_id, int) or (port_id < 0) or (port_id > self.get_port_count()): + raise ValueError("bad port id {0}".format(port_id)) - def get_global_stats(self): - return self.transmit("get_global_stats") + return port_id_list - ########## port commands ############## - def sync_ports (self, port_id_list = None, force = False): + # sync ports + def __sync_ports (self, port_id_list = None, force = False): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].sync()) - + return rc # acquire ports, if port_list is none - get all - def acquire (self, port_id_list = None, force = False): + def __acquire (self, port_id_list = None, force = False): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].acquire(force)) - + return rc - + # release ports - def release (self, port_id_list = None): + def __release (self, port_id_list = None): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].release()) - + return rc - - def add_stream(self, stream_id, stream_obj, port_id_list = None): + + def __add_stream(self, stream_id, stream_obj, port_id_list = None): port_id_list = self.__ports(port_id_list) @@ -589,12 +488,12 @@ class CTRexStatelessClient(object): for port_id in port_id_list: rc.add(self.ports[port_id].add_stream(stream_id, stream_obj)) - + return rc - - def add_stream_pack(self, stream_pack, port_id_list = None): + + def __add_stream_pack(self, stream_pack, port_id_list = None): port_id_list = self.__ports(port_id_list) @@ -607,45 +506,45 @@ class CTRexStatelessClient(object): - def remove_stream(self, stream_id, port_id_list = None): + def __remove_stream(self, stream_id, port_id_list = None): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].remove_stream(stream_id)) - + return rc - def remove_all_streams(self, port_id_list = None): + def __remove_all_streams(self, port_id_list = None): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].remove_all_streams()) - + return rc - - def get_stream(self, stream_id, port_id, get_pkt = False): + + def __get_stream(self, stream_id, port_id, get_pkt = False): return self.ports[port_id].get_stream(stream_id) - def get_all_streams(self, port_id, get_pkt = False): + def __get_all_streams(self, port_id, get_pkt = False): return self.ports[port_id].get_all_streams() - def get_stream_id_list(self, port_id): + def __get_stream_id_list(self, port_id): return self.ports[port_id].get_stream_id_list() - def start_traffic (self, multiplier, duration, port_id_list = None): + def __start_traffic (self, multiplier, duration, port_id_list = None): port_id_list = self.__ports(port_id_list) @@ -653,11 +552,11 @@ class CTRexStatelessClient(object): for port_id in port_id_list: rc.add(self.ports[port_id].start(multiplier, duration)) - + return rc - def resume_traffic (self, port_id_list = None, force = False): + def __resume_traffic (self, port_id_list = None, force = False): port_id_list = self.__ports(port_id_list) rc = RC() @@ -667,7 +566,7 @@ class CTRexStatelessClient(object): return rc - def pause_traffic (self, port_id_list = None, force = False): + def __pause_traffic (self, port_id_list = None, force = False): port_id_list = self.__ports(port_id_list) rc = RC() @@ -677,96 +576,325 @@ class CTRexStatelessClient(object): return rc - def stop_traffic (self, port_id_list = None, force = False): + + def __stop_traffic (self, port_id_list = None, force = False): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].stop(force)) - + return rc - def update_traffic (self, mult, port_id_list = None, force = False): + def __update_traffic (self, mult, port_id_list = None, force = False): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].update(mult)) - + return rc - def validate (self, port_id_list = None): + def __validate (self, port_id_list = None): port_id_list = self.__ports(port_id_list) rc = RC() for port_id in port_id_list: rc.add(self.ports[port_id].validate()) - + return rc - def get_port_stats(self, port_id=None): - pass - def get_stream_stats(self, port_id=None): - pass + # connect to server + # mode can be RW - read / write, RWF - read write with force , RO - read only + def __connect(self, mode = "RW"): - def transmit(self, method_name, params={}): - return self.comm_link.transmit(method_name, params) + # first disconnect if already connected + if self.is_connected(): + self.__disconnect() + # clear this flag + self.connected = False - def transmit_batch(self, batch_list): - return self.comm_link.transmit_batch(batch_list) + # connect sync channel + rc = self.comm_link.connect() + if not rc: + return rc - ######################### Console (high level) API ######################### + # connect async channel + rc = self.async_client.connect() + if not rc: + return rc - @timing - def cmd_ping(self): - rc = self.ping() - self.annotate(rc, "Pinging the server on '{0}' port '{1}': ".format(self.get_connection_ip(), self.get_connection_port())) - return rc + # version + rc = self._transmit("get_version") + if not rc: + return rc - def cmd_connect(self, mode = "RW"): - rc = self.connect(mode) - self.annotate(rc) - return rc + self.server_version = rc.data() + self.global_stats.server_version = rc.data() - def cmd_disconnect(self): - rc = self.disconnect() - self.annotate(rc) - return rc + # cache system info + rc = self._transmit("get_system_info") + if not rc: + return rc - # reset - def cmd_reset(self): + self.system_info = rc.data() - rc = self.acquire(force = True) - self.annotate(rc, "Force acquiring all ports:") - if rc.bad(): + # cache supported commands + rc = self._transmit("get_supported_cmds") + if not rc: return rc + self.supported_cmds = rc.data() - # force stop all ports - rc = self.stop_traffic(self.get_port_ids(), True) - self.annotate(rc,"Stop traffic on all ports:") - if rc.bad(): + # create ports + for port_id in xrange(self.system_info["port_count"]): + speed = self.system_info['ports'][port_id]['speed'] + driver = self.system_info['ports'][port_id]['driver'] + + self.ports[port_id] = Port(port_id, + speed, + driver, + self.username, + self.comm_link, + self.session_id) + + + # sync the ports + rc = self.__sync_ports() + if not rc: return rc + # acquire all ports + if mode == "RW": + rc = self.__acquire(force = False) - # remove all streams - rc = self.remove_all_streams(self.get_port_ids()) - self.annotate(rc,"Removing all streams from all ports:") + # fallback to read only if failed + if not rc: + self.logger.annotate(rc, show_status = False) + self.logger.log(format_text("Switching to read only mode - only few commands will be available", 'bold')) + + self.__release(self.get_acquired_ports()) + self.read_only = True + else: + self.read_only = False + + elif mode == "RWF": + rc = self.__acquire(force = True) + if not rc: + return rc + self.read_only = False + + elif mode == "RO": + # no acquire on read only + rc = RC_OK() + self.read_only = True + + + + self.connected = True + return RC_OK() + + + # disconenct from server + def __disconnect(self): + # release any previous acquired ports + if self.is_connected(): + self.__release(self.get_acquired_ports()) + + self.comm_link.disconnect() + self.async_client.disconnect() + + self.connected = False + + return RC_OK() + + + # ping server + def __ping (self): + return self._transmit("ping") + + + # start command + def __start (self, port_id_list, stream_list, mult, force, duration, dry): + + active_ports = list(set(self.get_active_ports()).intersection(port_id_list)) + + if active_ports: + if not force: + msg = "Port(s) {0} are active - please stop them or add '--force'".format(active_ports) + self.logger.log(format_text(msg, 'bold')) + return RC_ERR(msg) + else: + rc = self.cmd_stop(active_ports) + if not rc: + return rc + + + rc = self.__remove_all_streams(port_id_list) + self.logger.annotate(rc,"Removing all streams from port(s) {0}:".format(port_id_list)) if rc.bad(): return rc + + + rc = self.__add_stream_pack(stream_list, port_id_list) + self.logger.annotate(rc,"Attaching {0} streams to port(s) {1}:".format(len(stream_list.compiled), port_id_list)) + if rc.bad(): + return rc + + # when not on dry - start the traffic , otherwise validate only + if not dry: + rc = self.__start_traffic(mult, duration, port_id_list) + self.logger.annotate(rc,"Starting traffic on port(s) {0}:".format(port_id_list)) + + return rc + else: + rc = self.__validate(port_id_list) + self.logger.annotate(rc,"Validating traffic profile on port(s) {0}:".format(port_id_list)) + + if rc.bad(): + return rc + + # show a profile on one port for illustration + self.ports[port_id_list[0]].print_profile(mult, duration) + + return rc + + + + def __verify_port_id_list (self, port_id_list): + # check arguments + if not isinstance(port_id_list, list): + return RC_ERR("ports should be an instance of 'list'") + + # all ports are valid ports + if not all([port_id in self.get_all_ports() for port_id in port_id_list]): + return RC_ERR("Port IDs valid values are '{0}' but provided '{1}'".format(self.get_all_ports(), port_id_list)) - # TODO: clear stats return RC_OK() - + + + def __verify_mult (self, mult, strict): + if not isinstance(mult, dict): + return RC_ERR("mult should be an instance of dict") + + types = ["raw", "bps", "pps", "percentage"] + if not mult.get('type', None) in types: + return RC_ERR("mult should contain 'type' field of one of '{0}'".format(types)) + + if strict: + ops = ["abs"] + else: + ops = ["abs", "add", "sub"] + if not mult.get('op', None) in ops: + return RC_ERR("mult should contain 'op' field of one of '{0}'".format(ops)) + + return RC_OK() + + def __process_profiles (self, profiles, out): + + for profile in (profiles if isinstance(profiles, list) else [profiles]): + # filename + if isinstance(profile, str): + + if not os.path.isfile(profile): + return RC_ERR("file '{0}' does not exists".format(profile)) + + try: + stream_list = self.streams_db.load_yaml_file(profile) + except Exception as e: + rc = RC_ERR(str(e)) + self.logger.annotate(rc) + return rc + + out += stream_list + else: + return RC_ERR("unknown profile '{0}'".format(profile)) + + + return RC_OK() + + + + # stream list + if opts.db: + stream_list = self.streams_db.get_stream_pack(opts.db) + rc = RC(stream_list != None) + self.logger.annotate(rc,"Load stream pack (from DB):") + if rc.bad(): + return RC_ERR("Failed to load stream pack") + + else: + # load streams from file + stream_list = None + try: + stream_list = self.streams_db.load_yaml_file(opts.file[0]) + except Exception as e: + s = str(e) + rc=RC_ERR(s) + self.logger.annotate(rc) + return rc + + rc = RC(stream_list != None) + self.logger.annotate(rc,"Load stream pack (from file):") + if stream_list == None: + return RC_ERR("Failed to load stream pack") + + ############ functions used by other classes but not users ############## + + def _validate_port_list(self, port_id_list): + if not isinstance(port_id_list, list): + print type(port_id_list) + return False + + # check each item of the sequence + return all([ (port_id >= 0) and (port_id < self.get_port_count()) + for port_id in port_id_list ]) + + + # transmit request on the RPC link + def _transmit(self, method_name, params={}): + return self.comm_link.transmit(method_name, params) + + # transmit batch request on the RPC link + def _transmit_batch(self, batch_list): + return self.comm_link.transmit_batch(batch_list) + + ############# helper functions section ############## + + # measure time for functions + def timing(f): + def wrap(*args): + + time1 = time.time() + ret = f(*args) + + # don't want to print on error + if ret.bad(): + return ret + + delta = time.time() - time1 + + client = args[0] + client.logger.log(format_time(delta) + "\n") + + return ret + + return wrap + + + + ########## port commands ############## + + ######################### Console (high level) API ######################### # stop cmd def cmd_stop (self, port_id_list): @@ -779,8 +907,8 @@ class CTRexStatelessClient(object): self.logger.log(format_text(msg, 'bold')) return RC_ERR(msg) - rc = self.stop_traffic(active_ports) - self.annotate(rc,"Stopping traffic on port(s) {0}:".format(port_id_list)) + rc = self.__stop_traffic(active_ports) + self.logger.annotate(rc,"Stopping traffic on port(s) {0}:".format(port_id_list)) if rc.bad(): return rc @@ -797,8 +925,8 @@ class CTRexStatelessClient(object): self.logger.log(format_text(msg, 'bold')) return RC_ERR(msg) - rc = self.update_traffic(mult, active_ports) - self.annotate(rc,"Updating traffic on port(s) {0}:".format(port_id_list)) + rc = self.__update_traffic(mult, active_ports) + self.logger.annotate(rc,"Updating traffic on port(s) {0}:".format(port_id_list)) return rc @@ -832,8 +960,8 @@ class CTRexStatelessClient(object): self.logger.log(format_text(msg, 'bold')) return RC_ERR(msg) - rc = self.pause_traffic(active_ports) - self.annotate(rc,"Pausing traffic on port(s) {0}:".format(port_id_list)) + rc = self.__pause_traffic(active_ports) + self.logger.annotate(rc,"Pausing traffic on port(s) {0}:".format(port_id_list)) return rc @@ -849,61 +977,18 @@ class CTRexStatelessClient(object): self.logger.log(format_text(msg, 'bold')) return RC_ERR(msg) - rc = self.resume_traffic(active_ports) - self.annotate(rc,"Resume traffic on port(s) {0}:".format(port_id_list)) + rc = self.__resume_traffic(active_ports) + self.logger.annotate(rc,"Resume traffic on port(s) {0}:".format(port_id_list)) return rc - # start cmd - def cmd_start (self, port_id_list, stream_list, mult, force, duration, dry): - - active_ports = list(set(self.get_active_ports()).intersection(port_id_list)) - - if active_ports: - if not force: - msg = "Port(s) {0} are active - please stop them or add '--force'".format(active_ports) - self.logger.log(format_text(msg, 'bold')) - return RC_ERR(msg) - else: - rc = self.cmd_stop(active_ports) - if not rc: - return rc - - - rc = self.remove_all_streams(port_id_list) - self.annotate(rc,"Removing all streams from port(s) {0}:".format(port_id_list)) - if rc.bad(): - return rc - - - rc = self.add_stream_pack(stream_list, port_id_list) - self.annotate(rc,"Attaching {0} streams to port(s) {1}:".format(len(stream_list.compiled), port_id_list)) - if rc.bad(): - return rc - - # when not on dry - start the traffic , otherwise validate only - if not dry: - rc = self.start_traffic(mult, duration, port_id_list) - self.annotate(rc,"Starting traffic on port(s) {0}:".format(port_id_list)) - - return rc - else: - rc = self.validate(port_id_list) - self.annotate(rc,"Validating traffic profile on port(s) {0}:".format(port_id_list)) - - if rc.bad(): - return rc - - # show a profile on one port for illustration - self.ports[port_id_list[0]].print_profile(mult, duration) - - return rc + # validate port(s) profile def cmd_validate (self, port_id_list): - rc = self.validate(port_id_list) - self.annotate(rc,"Validating streams on port(s) {0}:".format(port_id_list)) + rc = self.__validate(port_id_list) + self.logger.annotate(rc,"Validating streams on port(s) {0}:".format(port_id_list)) return rc @@ -925,23 +1010,6 @@ class CTRexStatelessClient(object): ############## High Level API With Parser ################ - def cmd_connect_line (self, line): - '''Connects to the TRex server''' - # define a parser - parser = parsing_opts.gen_parser(self, - "connect", - self.cmd_connect_line.__doc__, - parsing_opts.FORCE) - - opts = parser.parse_args(line.split()) - - if opts is None: - return RC_ERR("bad command line parameters") - - if opts.force: - rc = self.cmd_connect(mode = "RWF") - else: - rc = self.cmd_connect(mode = "RW") @timing def cmd_start_line (self, line): @@ -971,7 +1039,7 @@ class CTRexStatelessClient(object): if opts.db: stream_list = self.streams_db.get_stream_pack(opts.db) rc = RC(stream_list != None) - self.annotate(rc,"Load stream pack (from DB):") + self.logger.annotate(rc,"Load stream pack (from DB):") if rc.bad(): return RC_ERR("Failed to load stream pack") @@ -983,11 +1051,11 @@ class CTRexStatelessClient(object): except Exception as e: s = str(e) rc=RC_ERR(s) - self.annotate(rc) + self.logger.annotate(rc) return rc rc = RC(stream_list != None) - self.annotate(rc,"Load stream pack (from file):") + self.logger.annotate(rc,"Load stream pack (from file):") if stream_list == None: return RC_ERR("Failed to load stream pack") @@ -1253,65 +1321,249 @@ class CTRexStatelessClient(object): def _filter_namespace_args(namespace, ok_values): return {k: v for k, v in namespace.__dict__.items() if k in ok_values} + def __verify_connected(f): + #@wraps(f) + def wrap(*args): + inst = args[0] + func_name = f.__name__ - ################################# - # ------ private classes ------ # - class CCommLink(object): - """describes the connectivity of the stateless client method""" - def __init__(self, server="localhost", port=5050, virtual=False, prn_func = None): - super(CTRexStatelessClient.CCommLink, self).__init__() - self.virtual = virtual - self.server = server - self.port = port - self.rpc_link = JsonRpcClient(self.server, self.port, prn_func) - - @property - def is_connected(self): - if not self.virtual: - return self.rpc_link.connected - else: - return True + if not inst.stateless_client.is_connected(): + return RC_ERR("cannot execute '{0}' while client is disconnected".format(func_name)) + + ret = f(*args) + return ret - def get_server (self): - return self.server + return wrap - def get_port (self): - return self.port - def connect(self): - if not self.virtual: - return self.rpc_link.connect() - def disconnect(self): - if not self.virtual: - return self.rpc_link.disconnect() + ############################ API ############################# + ############################ ############################# + ############################ ############################# - def transmit(self, method_name, params={}): - if self.virtual: - self._prompt_virtual_tx_msg() - _, msg = self.rpc_link.create_jsonrpc_v2(method_name, params) - print msg - return - else: - return self.rpc_link.invoke_rpc_method(method_name, params) - - def transmit_batch(self, batch_list): - if self.virtual: - self._prompt_virtual_tx_msg() - print [msg - for _, msg in [self.rpc_link.create_jsonrpc_v2(command.method, command.params) - for command in batch_list]] - else: - batch = self.rpc_link.create_batch() - for command in batch_list: - batch.add(command.method, command.params) - # invoke the batch - return batch.invoke() - def _prompt_virtual_tx_msg(self): - print "Transmitting virtually over tcp://{server}:{port}".format(server=self.server, - port=self.port) + ############################ Getters ############################# + ############################ ############################# + ############################ ############################# + + + # return verbose level of the logger + def get_verbose (self): + return self.logger.get_verbose() + + # is the client on read only mode ? + def is_read_only (self): + return self.read_only + + # is the client connected ? + def is_connected (self): + return self.connected and self.comm_link.is_connected + + + # get connection info + def get_connection_info (self): + return self.connection_info + + + # get supported commands by the server + def get_server_supported_cmds(self): + return self.supported_cmds + + # get server version + def get_server_version(self): + return self.server_version + + # get server system info + def get_server_system_info(self): + return self.system_info + + # get port count + def get_port_count(self): + return len(self.ports) + + # returns the port object + def get_port (self, port_id): + return self.ports.get(port_id, RC_ERR("invalid port id")) + + # get all ports as IDs + def get_all_ports (self): + return self.ports.keys() + + # get all acquired ports + def get_acquired_ports(self): + return [port_id + for port_id, port_obj in self.ports.iteritems() + if port_obj.is_acquired()] + + # get all active ports (TX or pause) + def get_active_ports(self): + return [port_id + for port_id, port_obj in self.ports.iteritems() + if port_obj.is_active()] + + # get paused ports + def get_paused_ports (self): + return [port_id + for port_id, port_obj in self.ports.iteritems() + if port_obj.is_paused()] + + # get all TX ports + def get_transmitting_ports (self): + return [port_id + for port_id, port_obj in self.ports.iteritems() + if port_obj.is_transmitting()] + + + ############################ Commands ############################# + ############################ ############################# + ############################ ############################# + + + # set the log on verbose level + def set_verbose (self, level): + self.logger.set_verbose(level) + + + # connects to the server + # mode can be: + # 'RO' - read only + # 'RW' - read/write + # 'RWF' - read write forced (take ownership) + def connect (self, mode = "RW"): + modes = ['RO', 'RW', 'RWF'] + if not mode in modes: + return RC_ERR("invalid mode '{0}'".format(mode)) + + rc = self.__connect(mode) + self.logger.annotate(rc) + + if not rc: + raise STLFailure(rc) + + return rc + + + # disconnects from the server + def disconnect (self, annotate = True): + rc = self.__disconnect() + if annotate: + self.logger.annotate(rc, "Disconnecting from server at '{0}':'{1}'".format(self.connection_info['server'], + self.connection_info['sync_port'])) + if not rc: + raise STLFailure(rc) + + return rc + + + # teardown - call after test is done + def teardown (self): + # for now, its only disconnect + rc = self.__disconnect() + if not rc: + raise STLFailure(rc) + + return rc + + + # pings the server on the RPC channel + def ping(self): + rc = self.__ping() + self.logger.annotate(rc, "Pinging the server on '{0}' port '{1}': ".format(self.connection_info['server'], + self.connection_info['sync_port'])) + + if not rc: + raise STLFailure(rc) + + return rc + + + # reset the server by performing + # force acquire, stop, and remove all streams + def reset(self): + + rc = self.__acquire(force = True) + self.logger.annotate(rc, "Force acquiring all ports:") + if not rc: + raise STLFailure(rc) + + + # force stop all ports + rc = self.__stop_traffic(self.get_all_ports(), True) + self.logger.annotate(rc,"Stop traffic on all ports:") + if not rc: + raise STLFailure(rc) + + + # remove all streams + rc = self.__remove_all_streams(self.get_all_ports()) + self.logger.annotate(rc,"Removing all streams from all ports:") + if not rc: + raise STLFailure(rc) + + # TODO: clear stats + return RC_OK() + + # start cmd + def start (self, + profiles, + ports = None, + mult = "1", + force = False, + duration = -1, + dry = False): + + + # by default use all ports + if ports == None: + ports = self.get_all_ports() + + # verify valid port id list + rc = self.__verify_port_id_list(ports) + if not rc: + raise STLFailure(rc) + + + # verify multiplier + try: + result = parsing_opts.match_multiplier_common(mult) + except argparse.ArgumentTypeError: + raise STLFailure("bad format for multiplier: {0}".format(mult)) + + # process profiles + stream_list = [] + rc = self.__process_profiles(profiles, stream_list) + if not rc: + raise STLFailure(rc) + + + + + ############################ Line ############################# + ############################ Commands ############################# + ############################ ############################# + + def connect_line (self, line): + '''Connects to the TRex server''' + # define a parser + parser = parsing_opts.gen_parser(self, + "connect", + self.connect_line.__doc__, + parsing_opts.FORCE) + + opts = parser.parse_args(line.split()) + + if opts is None: + return RC_ERR("bad command line parameters") + + # call the API + if opts.force: + rc = self.connect(mode = "RWF") + else: + rc = self.connect(mode = "RW") + + def disconnect_line (self, line): + return self.disconnect() -if __name__ == "__main__": - pass + def reset_line (self, line): + return self.reset() diff --git a/scripts/automation/trex_control_plane/client_utils/parsing_opts.py b/scripts/automation/trex_control_plane/client_utils/parsing_opts.py index 3735a45b..c1afda26 100755 --- a/scripts/automation/trex_control_plane/client_utils/parsing_opts.py +++ b/scripts/automation/trex_control_plane/client_utils/parsing_opts.py @@ -284,12 +284,12 @@ class CCmdArgParser(argparse.ArgumentParser): # if all ports are marked or if (getattr(opts, "all_ports", None) == True) or (getattr(opts, "ports", None) == []): - opts.ports = self.stateless_client.get_port_ids() + opts.ports = self.stateless_client.get_all_ports() # so maybe we have ports configured elif (getattr(opts, "ports", None) == []): for port in opts.ports: - if not self.stateless_client.validate_port_list([port]): + if not self.stateless_client._validate_port_list([port]): self.error("port id '{0}' is not a valid port id\n".format(port)) return opts diff --git a/scripts/automation/trex_control_plane/common/trex_types.py b/scripts/automation/trex_control_plane/common/trex_types.py index 5c29f59b..337f0a70 100644 --- a/scripts/automation/trex_control_plane/common/trex_types.py +++ b/scripts/automation/trex_control_plane/common/trex_types.py @@ -79,6 +79,7 @@ class RC(): def RC_OK(data = ""): return RC(True, data) + def RC_ERR (err): return RC(False, err) diff --git a/scripts/automation/trex_control_plane/console/trex_console.py b/scripts/automation/trex_control_plane/console/trex_console.py index f086c208..72cdcb0d 100755 --- a/scripts/automation/trex_control_plane/console/trex_console.py +++ b/scripts/automation/trex_control_plane/console/trex_console.py @@ -39,6 +39,28 @@ from functools import wraps __version__ = "1.1" +# console custom logger +class ConsoleLogger(LoggerApi): + def __init__ (self): + self.prompt_redraw = None + + def write (self, msg, newline = True): + if newline: + print msg + else: + print msg, + + def flush (self): + sys.stdout.flush() + + # override this for the prompt fix + def async_log (self, msg, level = LoggerApi.VERBOSE_REGULAR, newline = True): + self.log(msg, level, newline) + if self.prompt_redraw: + self.prompt_redraw() + self.flush() + + def set_window_always_on_top (title): # we need the GDK module, if not available - ignroe this command try: @@ -133,9 +155,9 @@ class TRexGeneralCmd(cmd.Cmd): class TRexConsole(TRexGeneralCmd): """Trex Console""" - def __init__(self, stateless_client, verbose=False): + def __init__(self, stateless_client, verbose = False): + self.stateless_client = stateless_client - self.stateless_client.set_prompt_redraw_cb(self.prompt_redraw) TRexGeneralCmd.__init__(self) @@ -199,7 +221,7 @@ class TRexConsole(TRexGeneralCmd): def get_console_identifier(self): return "{context}_{server}".format(context=self.__class__.__name__, - server=self.stateless_client.get_server_ip()) + server=self.stateless_client.get_connection_info()['server']) def register_main_console_methods(self): main_names = set(self.trex_console.get_names()).difference(set(dir(self.__class__))) @@ -271,7 +293,7 @@ class TRexConsole(TRexGeneralCmd): @verify_connected def do_ping (self, line): '''Ping the server\n''' - rc = self.stateless_client.cmd_ping() + rc = self.stateless_client.ping() if rc.bad(): return @@ -333,13 +355,13 @@ class TRexConsole(TRexGeneralCmd): def do_connect (self, line): '''Connects to the server\n''' - self.stateless_client.cmd_connect_line(line) + self.stateless_client.connect_line(line) def do_disconnect (self, line): '''Disconnect from the server\n''' - self.stateless_client.cmd_disconnect() + self.stateless_client.disconnect_line(line) ############### start @@ -408,7 +430,7 @@ class TRexConsole(TRexGeneralCmd): @verify_connected_and_rw def do_reset (self, line): '''force stop all ports\n''' - self.stateless_client.cmd_reset_line(line) + self.stateless_client.reset_line(line) ######### validate @@ -492,7 +514,9 @@ class TRexConsole(TRexGeneralCmd): if opts.xterm: - exe = './trex-console -t -q -s {0} -p {1}'.format(self.stateless_client.get_server_ip(), self.stateless_client.get_server_port()) + info = self.stateless_client.get_connection_info() + + exe = './trex-console -t -q -s {0} -p {1} --async_port {2}'.format(info['server'], info['sync_port'], info['async_port']) cmd = ['xterm', '-geometry', '111x42', '-sl', '0', '-title', 'trex_tui', '-e', exe] self.terminal = subprocess.Popen(cmd) @@ -645,11 +669,13 @@ def main(): verbose_level = LoggerApi.VERBOSE_REGULAR # Stateless client connection + logger = ConsoleLogger() stateless_client = CTRexStatelessClient(options.user, options.server, options.port, options.pub, - verbose_level) + verbose_level, + logger) # TUI or no acquire will give us READ ONLY mode if options.tui or not options.acquire: @@ -673,6 +699,8 @@ def main(): try: console = TRexConsole(stateless_client, options.verbose) + logger.prompt_redraw = console.prompt_redraw + if options.tui: console.do_tui("") else: @@ -682,7 +710,7 @@ def main(): print "\n\n*** Caught Ctrl + C... Exiting...\n\n" finally: - stateless_client.disconnect() + stateless_client.teardown() if __name__ == '__main__': diff --git a/scripts/automation/trex_control_plane/console/trex_status.py b/scripts/automation/trex_control_plane/console/trex_status.py index cdf3fb69..45769693 100644 --- a/scripts/automation/trex_control_plane/console/trex_status.py +++ b/scripts/automation/trex_control_plane/console/trex_status.py @@ -1,525 +1,525 @@ -from time import sleep - -import os - -import curses -from curses import panel -import random -import collections -import operator -import datetime - -g_curses_active = False - -################### utils ################# - -# simple percetange show -def percentage (a, total): - x = int ((float(a) / total) * 100) - return str(x) + "%" - -################### panels ################# - -# panel object -class TrexStatusPanel(object): - def __init__ (self, h, l, y, x, headline, status_obj): - - self.status_obj = status_obj - - self.log = status_obj.log - self.stateless_client = status_obj.stateless_client - - self.stats = status_obj.stats - self.general_stats = status_obj.general_stats - - self.h = h - self.l = l - self.y = y - self.x = x - self.headline = headline - - self.win = curses.newwin(h, l, y, x) - self.win.erase() - self.win.box() - - self.win.addstr(1, 2, headline, curses.A_UNDERLINE) - self.win.refresh() - - panel.new_panel(self.win) - self.panel = panel.new_panel(self.win) - self.panel.top() - - def clear (self): - self.win.erase() - self.win.box() - self.win.addstr(1, 2, self.headline, curses.A_UNDERLINE) - - def getwin (self): - return self.win - - -# various kinds of panels - -# Server Info Panel -class ServerInfoPanel(TrexStatusPanel): - def __init__ (self, h, l, y, x, status_obj): - - super(ServerInfoPanel, self).__init__(h, l, y ,x ,"Server Info:", status_obj) - - def draw (self): - - if not self.status_obj.server_version : - return - - if not self.status_obj.server_sys_info: - return - - - self.clear() - - self.getwin().addstr(3, 2, "{:<30} {:30}".format("Server:",self.status_obj.server_sys_info["hostname"] + ":" + str(self.stateless_client.get_connection_port()))) - self.getwin().addstr(4, 2, "{:<30} {:30}".format("Version:", self.status_obj.server_version["version"])) - self.getwin().addstr(5, 2, "{:<30} {:30}".format("Build:", - self.status_obj.server_version["build_date"] + " @ " + - self.status_obj.server_version["build_time"] + " by " + - self.status_obj.server_version["built_by"])) - - self.getwin().addstr(6, 2, "{:<30} {:30}".format("Server Uptime:", self.status_obj.server_sys_info["uptime"])) - self.getwin().addstr(7, 2, "{:<30} {:<3} / {:<30}".format("DP Cores:", str(self.status_obj.server_sys_info["dp_core_count"]) + - " cores", self.status_obj.server_sys_info["core_type"])) - - self.getwin().addstr(9, 2, "{:<30} {:<30}".format("Ports Count:", self.status_obj.server_sys_info["port_count"])) - - ports_owned = " ".join(str(x) for x in self.status_obj.owned_ports_list) - - if not ports_owned: - ports_owned = "None" - - self.getwin().addstr(10, 2, "{:<30} {:<30}".format("Ports Owned:", ports_owned)) - -# general info panel -class GeneralInfoPanel(TrexStatusPanel): - def __init__ (self, h, l, y, x, status_obj): - - super(GeneralInfoPanel, self).__init__(h, l, y ,x ,"General Info:", status_obj) - - def draw (self): - self.clear() - - if not self.general_stats.is_online(): - self.getwin().addstr(3, 2, "No Published Data From TRex Server") - return - - self.getwin().addstr(3, 2, "{:<30} {:0.2f} %".format("CPU util.:", self.general_stats.get("m_cpu_util"))) - - self.getwin().addstr(6, 2, "{:<30} {:} / {:}".format("Total Tx. rate:", - self.general_stats.get("m_tx_bps", format = True, suffix = "bps"), - self.general_stats.get("m_tx_pps", format = True, suffix = "pps"))) - - - self.getwin().addstr(8, 2, "{:<30} {:} / {:}".format("Total Tx:", - self.general_stats.get_rel("m_total_tx_bytes", format = True, suffix = "B"), - self.general_stats.get_rel("m_total_tx_pkts", format = True, suffix = "pkts"))) - - self.getwin().addstr(11, 2, "{:<30} {:} / {:}".format("Total Rx. rate:", - self.general_stats.get("m_rx_bps", format = True, suffix = "bps"), - self.general_stats.get("m_rx_pps", format = True, suffix = "pps"))) - - - self.getwin().addstr(13, 2, "{:<30} {:} / {:}".format("Total Rx:", - self.general_stats.get_rel("m_total_rx_bytes", format = True, suffix = "B"), - self.general_stats.get_rel("m_total_rx_pkts", format = True, suffix = "pkts"))) - -# all ports stats -class PortsStatsPanel(TrexStatusPanel): - def __init__ (self, h, l, y, x, status_obj): - - super(PortsStatsPanel, self).__init__(h, l, y ,x ,"Trex Ports:", status_obj) - - - def draw (self): - - self.clear() - - owned_ports = self.status_obj.owned_ports_list - if not owned_ports: - self.getwin().addstr(3, 2, "No Owned Ports - Please Acquire One Or More Ports") - return - - # table header - self.getwin().addstr(3, 2, "{:^15} {:^30} {:^30} {:^30}".format( - "Port ID", "Tx Rate [bps/pps]", "Rx Rate [bps/pps]", "Total Bytes [tx/rx]")) - - - - for i, port_index in enumerate(owned_ports): - - port_stats = self.status_obj.stats.get_port_stats(port_index) - - if port_stats: - self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^30} {:^30} {:^30}".format( - "{0} ({1})".format(str(port_index), self.status_obj.server_sys_info["ports"][port_index]["speed"]), - "{0} / {1}".format(port_stats.get("m_total_tx_bps", format = True, suffix = "bps"), - port_stats.get("m_total_tx_pps", format = True, suffix = "pps")), - - "{0} / {1}".format(port_stats.get("m_total_rx_bps", format = True, suffix = "bps"), - port_stats.get("m_total_rx_pps", format = True, suffix = "pps")), - "{0} / {1}".format(port_stats.get_rel("obytes", format = True, suffix = "B"), - port_stats.get_rel("ibytes", format = True, suffix = "B")))) - - else: - - self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^30} {:^30} {:^30}".format( - "{0} ({1})".format(str(port_index), self.status_obj.server_sys_info["ports"][port_index]["speed"]), - "N/A", - "N/A", - "N/A", - "N/A")) - - - # old format +#from time import sleep +# +#import os +# +#import curses +#from curses import panel +#import random +#import collections +#import operator +#import datetime +# +#g_curses_active = False +# +#################### utils ################# +# +## simple percetange show +#def percentage (a, total): +# x = int ((float(a) / total) * 100) +# return str(x) + "%" +# +#################### panels ################# +# +## panel object +#class TrexStatusPanel(object): +# def __init__ (self, h, l, y, x, headline, status_obj): +# +# self.status_obj = status_obj +# +# self.log = status_obj.log +# self.stateless_client = status_obj.stateless_client +# +# self.stats = status_obj.stats +# self.general_stats = status_obj.general_stats +# +# self.h = h +# self.l = l +# self.y = y +# self.x = x +# self.headline = headline +# +# self.win = curses.newwin(h, l, y, x) +# self.win.erase() +# self.win.box() +# +# self.win.addstr(1, 2, headline, curses.A_UNDERLINE) +# self.win.refresh() +# +# panel.new_panel(self.win) +# self.panel = panel.new_panel(self.win) +# self.panel.top() +# +# def clear (self): +# self.win.erase() +# self.win.box() +# self.win.addstr(1, 2, self.headline, curses.A_UNDERLINE) +# +# def getwin (self): +# return self.win +# +# +## various kinds of panels +# +## Server Info Panel +#class ServerInfoPanel(TrexStatusPanel): +# def __init__ (self, h, l, y, x, status_obj): +# +# super(ServerInfoPanel, self).__init__(h, l, y ,x ,"Server Info:", status_obj) +# +# def draw (self): +# +# if not self.status_obj.server_version : +# return +# +# if not self.status_obj.server_sys_info: +# return +# +# +# self.clear() +# +# self.getwin().addstr(3, 2, "{:<30} {:30}".format("Server:",self.status_obj.server_sys_info["hostname"] + ":" + str(self.stateless_client.get_connection_port()))) +# self.getwin().addstr(4, 2, "{:<30} {:30}".format("Version:", self.status_obj.server_version["version"])) +# self.getwin().addstr(5, 2, "{:<30} {:30}".format("Build:", +# self.status_obj.server_version["build_date"] + " @ " + +# self.status_obj.server_version["build_time"] + " by " + +# self.status_obj.server_version["built_by"])) +# +# self.getwin().addstr(6, 2, "{:<30} {:30}".format("Server Uptime:", self.status_obj.server_sys_info["uptime"])) +# self.getwin().addstr(7, 2, "{:<30} {:<3} / {:<30}".format("DP Cores:", str(self.status_obj.server_sys_info["dp_core_count"]) + +# " cores", self.status_obj.server_sys_info["core_type"])) +# +# self.getwin().addstr(9, 2, "{:<30} {:<30}".format("Ports Count:", self.status_obj.server_sys_info["port_count"])) +# +# ports_owned = " ".join(str(x) for x in self.status_obj.owned_ports_list) +# +# if not ports_owned: +# ports_owned = "None" +# +# self.getwin().addstr(10, 2, "{:<30} {:<30}".format("Ports Owned:", ports_owned)) +# +## general info panel +#class GeneralInfoPanel(TrexStatusPanel): +# def __init__ (self, h, l, y, x, status_obj): +# +# super(GeneralInfoPanel, self).__init__(h, l, y ,x ,"General Info:", status_obj) +# +# def draw (self): +# self.clear() +# +# if not self.general_stats.is_online(): +# self.getwin().addstr(3, 2, "No Published Data From TRex Server") +# return +# +# self.getwin().addstr(3, 2, "{:<30} {:0.2f} %".format("CPU util.:", self.general_stats.get("m_cpu_util"))) +# +# self.getwin().addstr(6, 2, "{:<30} {:} / {:}".format("Total Tx. rate:", +# self.general_stats.get("m_tx_bps", format = True, suffix = "bps"), +# self.general_stats.get("m_tx_pps", format = True, suffix = "pps"))) +# +# +# self.getwin().addstr(8, 2, "{:<30} {:} / {:}".format("Total Tx:", +# self.general_stats.get_rel("m_total_tx_bytes", format = True, suffix = "B"), +# self.general_stats.get_rel("m_total_tx_pkts", format = True, suffix = "pkts"))) +# +# self.getwin().addstr(11, 2, "{:<30} {:} / {:}".format("Total Rx. rate:", +# self.general_stats.get("m_rx_bps", format = True, suffix = "bps"), +# self.general_stats.get("m_rx_pps", format = True, suffix = "pps"))) +# +# +# self.getwin().addstr(13, 2, "{:<30} {:} / {:}".format("Total Rx:", +# self.general_stats.get_rel("m_total_rx_bytes", format = True, suffix = "B"), +# self.general_stats.get_rel("m_total_rx_pkts", format = True, suffix = "pkts"))) +# +## all ports stats +#class PortsStatsPanel(TrexStatusPanel): +# def __init__ (self, h, l, y, x, status_obj): +# +# super(PortsStatsPanel, self).__init__(h, l, y ,x ,"Trex Ports:", status_obj) +# +# +# def draw (self): +# +# self.clear() +# +# owned_ports = self.status_obj.owned_ports_list +# if not owned_ports: +# self.getwin().addstr(3, 2, "No Owned Ports - Please Acquire One Or More Ports") +# return +# +# # table header +# self.getwin().addstr(3, 2, "{:^15} {:^30} {:^30} {:^30}".format( +# "Port ID", "Tx Rate [bps/pps]", "Rx Rate [bps/pps]", "Total Bytes [tx/rx]")) +# +# +# +# for i, port_index in enumerate(owned_ports): +# +# port_stats = self.status_obj.stats.get_port_stats(port_index) +# # if port_stats: -# self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( +# self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^30} {:^30} {:^30}".format( # "{0} ({1})".format(str(port_index), self.status_obj.server_sys_info["ports"][port_index]["speed"]), -# port_stats.get("m_total_tx_pps", format = True, suffix = "pps"), -# port_stats.get("m_total_tx_bps", format = True, suffix = "bps"), -# port_stats.get_rel("obytes", format = True, suffix = "B"), -# port_stats.get("m_total_rx_pps", format = True, suffix = "pps"), -# port_stats.get("m_total_rx_bps", format = True, suffix = "bps"), -# port_stats.get_rel("ibytes", format = True, suffix = "B"))) +# "{0} / {1}".format(port_stats.get("m_total_tx_bps", format = True, suffix = "bps"), +# port_stats.get("m_total_tx_pps", format = True, suffix = "pps")), +# +# "{0} / {1}".format(port_stats.get("m_total_rx_bps", format = True, suffix = "bps"), +# port_stats.get("m_total_rx_pps", format = True, suffix = "pps")), +# "{0} / {1}".format(port_stats.get_rel("obytes", format = True, suffix = "B"), +# port_stats.get_rel("ibytes", format = True, suffix = "B")))) # # else: -# self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( +# +# self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^30} {:^30} {:^30}".format( # "{0} ({1})".format(str(port_index), self.status_obj.server_sys_info["ports"][port_index]["speed"]), # "N/A", # "N/A", # "N/A", -# "N/A", -# "N/A", # "N/A")) - -# control panel -class ControlPanel(TrexStatusPanel): - def __init__ (self, h, l, y, x, status_obj): - - super(ControlPanel, self).__init__(h, l, y, x, "", status_obj) - - - def draw (self): - self.clear() - - self.getwin().addstr(1, 2, "'g' - general, '0-{0}' - specific port, 'f' - freeze, 'c' - clear stats, 'p' - ping server, 'q' - quit" - .format(self.status_obj.stateless_client.get_port_count() - 1)) - - self.log.draw(self.getwin(), 2, 3) - -# specific ports panels -class SinglePortPanel(TrexStatusPanel): - def __init__ (self, h, l, y, x, status_obj, port_id): - - super(SinglePortPanel, self).__init__(h, l, y, x, "Port {0}".format(port_id), status_obj) - - self.port_id = port_id - - def draw (self): - y = 3 - - self.clear() - - if not self.port_id in self.status_obj.owned_ports_list: - self.getwin().addstr(y, 2, "Port {0} is not owned by you, please acquire the port for more info".format(self.port_id)) - return - - # streams - self.getwin().addstr(y, 2, "Streams:", curses.A_UNDERLINE) - y += 2 - - # stream table header - self.getwin().addstr(y, 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( - "Stream ID", "Enabled", "Type", "Self Start", "ISG", "Next Stream", "VM")) - y += 2 - - # streams - - if 'streams' in self.status_obj.owned_ports[str(self.port_id)]: - stream_info = self.status_obj.owned_ports[str(self.port_id)]['streams'] - - for stream_id, stream in sorted(stream_info.iteritems(), key=operator.itemgetter(0)): - self.getwin().addstr(y, 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( - stream_id, - ("True" if stream['enabled'] else "False"), - stream['mode']['type'], - ("True" if stream['self_start'] else "False"), - stream['isg'], - (stream['next_stream_id'] if stream['next_stream_id'] != -1 else "None"), - ("{0} instr.".format(len(stream['vm'])) if stream['vm'] else "None"))) - - y += 1 - - # new section - traffic - y += 2 - - self.getwin().addstr(y, 2, "Traffic:", curses.A_UNDERLINE) - y += 2 - - - - # table header - self.getwin().addstr(y, 2, "{:^15} {:^30} {:^30} {:^30}".format( - "Port ID", "Tx Rate [bps/pps]", "Rx Rate [bps/pps]", "Total Bytes [tx/rx]")) - - - y += 2 - - port_stats = self.status_obj.stats.get_port_stats(self.port_id) - - if port_stats: - self.getwin().addstr(y, 2, "{:^15} {:^30} {:^30} {:^30}".format( - "{0} ({1})".format(str(self.port_id), self.status_obj.server_sys_info["ports"][self.port_id]["speed"]), - "{0} / {1}".format(port_stats.get("m_total_tx_bps", format = True, suffix = "bps"), - port_stats.get("m_total_tx_pps", format = True, suffix = "pps")), - - "{0} / {1}".format(port_stats.get("m_total_rx_bps", format = True, suffix = "bps"), - port_stats.get("m_total_rx_pps", format = True, suffix = "pps")), - "{0} / {1}".format(port_stats.get_rel("obytes", format = True, suffix = "B"), - port_stats.get_rel("ibytes", format = True, suffix = "B")))) - - else: - self.getwin().addstr(y, 2, "{:^15} {:^30} {:^30} {:^30}".format( - "{0} ({1})".format(str(self.port_id), self.status_obj.server_sys_info["ports"][self.port_id]["speed"]), - "N/A", - "N/A", - "N/A", - "N/A")) - - -################### main objects ################# - -# status log -class TrexStatusLog(): - def __init__ (self): - self.log = [] - - def add_event (self, msg): - self.log.append("[{0}] {1}".format(str(datetime.datetime.now().time()), msg)) - - def draw (self, window, x, y, max_lines = 4): - index = y - - cut = len(self.log) - max_lines - if cut < 0: - cut = 0 - - for msg in self.log[cut:]: - window.addstr(index, x, msg) - index += 1 - -# status commands -class TrexStatusCommands(): - def __init__ (self, status_object): - - self.status_object = status_object - - self.stateless_client = status_object.stateless_client - self.log = self.status_object.log - - self.actions = {} - self.actions[ord('q')] = self._quit - self.actions[ord('p')] = self._ping - self.actions[ord('f')] = self._freeze - - self.actions[ord('g')] = self._show_ports_stats - - # register all the available ports shortcuts - for port_id in xrange(0, self.stateless_client.get_port_count()): - self.actions[ord('0') + port_id] = self._show_port_generator(port_id) - - - # handle a key pressed - def handle (self, ch): - if ch in self.actions: - return self.actions[ch]() - else: - self.log.add_event("Unknown key pressed, please see legend") - return True - - # show all ports - def _show_ports_stats (self): - self.log.add_event("Switching to all ports view") - self.status_object.stats_panel = self.status_object.ports_stats_panel - - return True - - - # function generator for different ports requests - def _show_port_generator (self, port_id): - def _show_port(): - self.log.add_event("Switching panel to port {0}".format(port_id)) - self.status_object.stats_panel = self.status_object.ports_panels[port_id] - - return True - - return _show_port - - def _freeze (self): - self.status_object.update_active = not self.status_object.update_active - self.log.add_event("Update continued" if self.status_object.update_active else "Update stopped") - - return True - - def _quit(self): - return False - - def _ping (self): - self.log.add_event("Pinging RPC server") - - rc, msg = self.stateless_client.ping() - if rc: - self.log.add_event("Server replied: '{0}'".format(msg)) - else: - self.log.add_event("Failed to get reply") - - return True - -# status object -# -# -# -class CTRexStatus(): - def __init__ (self, stdscr, stateless_client): - self.stdscr = stdscr - - self.stateless_client = stateless_client - - self.log = TrexStatusLog() - self.cmds = TrexStatusCommands(self) - - self.stats = stateless_client.get_stats_async() - self.general_stats = stateless_client.get_stats_async().get_general_stats() - - # fetch server info - self.server_sys_info = self.stateless_client.get_system_info() - - self.server_version = self.stateless_client.get_version() - - # list of owned ports - self.owned_ports_list = self.stateless_client.get_acquired_ports() - - # data per port - self.owned_ports = {} - - for port_id in self.owned_ports_list: - self.owned_ports[str(port_id)] = {} - self.owned_ports[str(port_id)]['streams'] = {} - - stream_list = self.stateless_client.get_all_streams(port_id) - - self.owned_ports[str(port_id)] = stream_list - - - try: - curses.curs_set(0) - except: - pass - - curses.use_default_colors() - self.stdscr.nodelay(1) - curses.nonl() - curses.noecho() - - self.generate_layout() - - - def generate_layout (self): - self.max_y = self.stdscr.getmaxyx()[0] - self.max_x = self.stdscr.getmaxyx()[1] - - self.server_info_panel = ServerInfoPanel(int(self.max_y * 0.3), self.max_x / 2, int(self.max_y * 0.5), self.max_x /2, self) - self.general_info_panel = GeneralInfoPanel(int(self.max_y * 0.5), self.max_x / 2, 0, self.max_x /2, self) - self.control_panel = ControlPanel(int(self.max_y * 0.2), self.max_x , int(self.max_y * 0.8), 0, self) - - # those can be switched on the same place - self.ports_stats_panel = PortsStatsPanel(int(self.max_y * 0.8), self.max_x / 2, 0, 0, self) - - self.ports_panels = {} - for i in xrange(0, self.stateless_client.get_port_count()): - self.ports_panels[i] = SinglePortPanel(int(self.max_y * 0.8), self.max_x / 2, 0, 0, self, i) - - # at start time we point to the main one - self.stats_panel = self.ports_stats_panel - self.stats_panel.panel.top() - - panel.update_panels(); self.stdscr.refresh() - return - - - def wait_for_key_input (self): - ch = self.stdscr.getch() - - # no key , continue - if ch == curses.ERR: - return True - - return self.cmds.handle(ch) - - # main run entry point - def run (self): - - # list of owned ports - self.owned_ports_list = self.stateless_client.get_acquired_ports() - - # data per port - self.owned_ports = {} - - for port_id in self.owned_ports_list: - self.owned_ports[str(port_id)] = {} - self.owned_ports[str(port_id)]['streams'] = {} - - stream_list = self.stateless_client.get_all_streams(port_id) - - self.owned_ports[str(port_id)] = stream_list - - self.update_active = True - while (True): - - rc = self.wait_for_key_input() - if not rc: - break - - self.server_info_panel.draw() - self.general_info_panel.draw() - self.control_panel.draw() - - # can be different kinds of panels - self.stats_panel.panel.top() - self.stats_panel.draw() - - panel.update_panels() - self.stdscr.refresh() - sleep(0.01) - - -# global container -trex_status = None - -def show_trex_status_internal (stdscr, stateless_client): - global trex_status - - if trex_status == None: - trex_status = CTRexStatus(stdscr, stateless_client) - - trex_status.run() - -def show_trex_status (stateless_client): - - try: - curses.wrapper(show_trex_status_internal, stateless_client) - except KeyboardInterrupt: - curses.endwin() - -def cleanup (): - try: - curses.endwin() - except: - pass - +# +# +# # old format +## if port_stats: +## self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( +## "{0} ({1})".format(str(port_index), self.status_obj.server_sys_info["ports"][port_index]["speed"]), +## port_stats.get("m_total_tx_pps", format = True, suffix = "pps"), +## port_stats.get("m_total_tx_bps", format = True, suffix = "bps"), +## port_stats.get_rel("obytes", format = True, suffix = "B"), +## port_stats.get("m_total_rx_pps", format = True, suffix = "pps"), +## port_stats.get("m_total_rx_bps", format = True, suffix = "bps"), +## port_stats.get_rel("ibytes", format = True, suffix = "B"))) +## +## else: +## self.getwin().addstr(5 + (i * 4), 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( +## "{0} ({1})".format(str(port_index), self.status_obj.server_sys_info["ports"][port_index]["speed"]), +## "N/A", +## "N/A", +## "N/A", +## "N/A", +## "N/A", +## "N/A")) +# +## control panel +#class ControlPanel(TrexStatusPanel): +# def __init__ (self, h, l, y, x, status_obj): +# +# super(ControlPanel, self).__init__(h, l, y, x, "", status_obj) +# +# +# def draw (self): +# self.clear() +# +# self.getwin().addstr(1, 2, "'g' - general, '0-{0}' - specific port, 'f' - freeze, 'c' - clear stats, 'p' - ping server, 'q' - quit" +# .format(self.status_obj.stateless_client.get_port_count() - 1)) +# +# self.log.draw(self.getwin(), 2, 3) +# +## specific ports panels +#class SinglePortPanel(TrexStatusPanel): +# def __init__ (self, h, l, y, x, status_obj, port_id): +# +# super(SinglePortPanel, self).__init__(h, l, y, x, "Port {0}".format(port_id), status_obj) +# +# self.port_id = port_id +# +# def draw (self): +# y = 3 +# +# self.clear() +# +# if not self.port_id in self.status_obj.owned_ports_list: +# self.getwin().addstr(y, 2, "Port {0} is not owned by you, please acquire the port for more info".format(self.port_id)) +# return +# +# # streams +# self.getwin().addstr(y, 2, "Streams:", curses.A_UNDERLINE) +# y += 2 +# +# # stream table header +# self.getwin().addstr(y, 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( +# "Stream ID", "Enabled", "Type", "Self Start", "ISG", "Next Stream", "VM")) +# y += 2 +# +# # streams +# +# if 'streams' in self.status_obj.owned_ports[str(self.port_id)]: +# stream_info = self.status_obj.owned_ports[str(self.port_id)]['streams'] +# +# for stream_id, stream in sorted(stream_info.iteritems(), key=operator.itemgetter(0)): +# self.getwin().addstr(y, 2, "{:^15} {:^15} {:^15} {:^15} {:^15} {:^15} {:^15}".format( +# stream_id, +# ("True" if stream['enabled'] else "False"), +# stream['mode']['type'], +# ("True" if stream['self_start'] else "False"), +# stream['isg'], +# (stream['next_stream_id'] if stream['next_stream_id'] != -1 else "None"), +# ("{0} instr.".format(len(stream['vm'])) if stream['vm'] else "None"))) +# +# y += 1 +# +# # new section - traffic +# y += 2 +# +# self.getwin().addstr(y, 2, "Traffic:", curses.A_UNDERLINE) +# y += 2 +# +# +# +# # table header +# self.getwin().addstr(y, 2, "{:^15} {:^30} {:^30} {:^30}".format( +# "Port ID", "Tx Rate [bps/pps]", "Rx Rate [bps/pps]", "Total Bytes [tx/rx]")) +# +# +# y += 2 +# +# port_stats = self.status_obj.stats.get_port_stats(self.port_id) +# +# if port_stats: +# self.getwin().addstr(y, 2, "{:^15} {:^30} {:^30} {:^30}".format( +# "{0} ({1})".format(str(self.port_id), self.status_obj.server_sys_info["ports"][self.port_id]["speed"]), +# "{0} / {1}".format(port_stats.get("m_total_tx_bps", format = True, suffix = "bps"), +# port_stats.get("m_total_tx_pps", format = True, suffix = "pps")), +# +# "{0} / {1}".format(port_stats.get("m_total_rx_bps", format = True, suffix = "bps"), +# port_stats.get("m_total_rx_pps", format = True, suffix = "pps")), +# "{0} / {1}".format(port_stats.get_rel("obytes", format = True, suffix = "B"), +# port_stats.get_rel("ibytes", format = True, suffix = "B")))) +# +# else: +# self.getwin().addstr(y, 2, "{:^15} {:^30} {:^30} {:^30}".format( +# "{0} ({1})".format(str(self.port_id), self.status_obj.server_sys_info["ports"][self.port_id]["speed"]), +# "N/A", +# "N/A", +# "N/A", +# "N/A")) +# +# +#################### main objects ################# +# +## status log +#class TrexStatusLog(): +# def __init__ (self): +# self.log = [] +# +# def add_event (self, msg): +# self.log.append("[{0}] {1}".format(str(datetime.datetime.now().time()), msg)) +# +# def draw (self, window, x, y, max_lines = 4): +# index = y +# +# cut = len(self.log) - max_lines +# if cut < 0: +# cut = 0 +# +# for msg in self.log[cut:]: +# window.addstr(index, x, msg) +# index += 1 +# +## status commands +#class TrexStatusCommands(): +# def __init__ (self, status_object): +# +# self.status_object = status_object +# +# self.stateless_client = status_object.stateless_client +# self.log = self.status_object.log +# +# self.actions = {} +# self.actions[ord('q')] = self._quit +# self.actions[ord('p')] = self._ping +# self.actions[ord('f')] = self._freeze +# +# self.actions[ord('g')] = self._show_ports_stats +# +# # register all the available ports shortcuts +# for port_id in xrange(0, self.stateless_client.get_port_count()): +# self.actions[ord('0') + port_id] = self._show_port_generator(port_id) +# +# +# # handle a key pressed +# def handle (self, ch): +# if ch in self.actions: +# return self.actions[ch]() +# else: +# self.log.add_event("Unknown key pressed, please see legend") +# return True +# +# # show all ports +# def _show_ports_stats (self): +# self.log.add_event("Switching to all ports view") +# self.status_object.stats_panel = self.status_object.ports_stats_panel +# +# return True +# +# +# # function generator for different ports requests +# def _show_port_generator (self, port_id): +# def _show_port(): +# self.log.add_event("Switching panel to port {0}".format(port_id)) +# self.status_object.stats_panel = self.status_object.ports_panels[port_id] +# +# return True +# +# return _show_port +# +# def _freeze (self): +# self.status_object.update_active = not self.status_object.update_active +# self.log.add_event("Update continued" if self.status_object.update_active else "Update stopped") +# +# return True +# +# def _quit(self): +# return False +# +# def _ping (self): +# self.log.add_event("Pinging RPC server") +# +# rc, msg = self.stateless_client.ping() +# if rc: +# self.log.add_event("Server replied: '{0}'".format(msg)) +# else: +# self.log.add_event("Failed to get reply") +# +# return True +# +## status object +## +## +## +#class CTRexStatus(): +# def __init__ (self, stdscr, stateless_client): +# self.stdscr = stdscr +# +# self.stateless_client = stateless_client +# +# self.log = TrexStatusLog() +# self.cmds = TrexStatusCommands(self) +# +# self.stats = stateless_client.get_stats_async() +# self.general_stats = stateless_client.get_stats_async().get_general_stats() +# +# # fetch server info +# self.server_sys_info = self.stateless_client.get_system_info() +# +# self.server_version = self.stateless_client.get_server_version() +# +# # list of owned ports +# self.owned_ports_list = self.stateless_client.get_acquired_ports() +# +# # data per port +# self.owned_ports = {} +# +# for port_id in self.owned_ports_list: +# self.owned_ports[str(port_id)] = {} +# self.owned_ports[str(port_id)]['streams'] = {} +# +# stream_list = self.stateless_client.get_all_streams(port_id) +# +# self.owned_ports[str(port_id)] = stream_list +# +# +# try: +# curses.curs_set(0) +# except: +# pass +# +# curses.use_default_colors() +# self.stdscr.nodelay(1) +# curses.nonl() +# curses.noecho() +# +# self.generate_layout() +# +# +# def generate_layout (self): +# self.max_y = self.stdscr.getmaxyx()[0] +# self.max_x = self.stdscr.getmaxyx()[1] +# +# self.server_info_panel = ServerInfoPanel(int(self.max_y * 0.3), self.max_x / 2, int(self.max_y * 0.5), self.max_x /2, self) +# self.general_info_panel = GeneralInfoPanel(int(self.max_y * 0.5), self.max_x / 2, 0, self.max_x /2, self) +# self.control_panel = ControlPanel(int(self.max_y * 0.2), self.max_x , int(self.max_y * 0.8), 0, self) +# +# # those can be switched on the same place +# self.ports_stats_panel = PortsStatsPanel(int(self.max_y * 0.8), self.max_x / 2, 0, 0, self) +# +# self.ports_panels = {} +# for i in xrange(0, self.stateless_client.get_port_count()): +# self.ports_panels[i] = SinglePortPanel(int(self.max_y * 0.8), self.max_x / 2, 0, 0, self, i) +# +# # at start time we point to the main one +# self.stats_panel = self.ports_stats_panel +# self.stats_panel.panel.top() +# +# panel.update_panels(); self.stdscr.refresh() +# return +# +# +# def wait_for_key_input (self): +# ch = self.stdscr.getch() +# +# # no key , continue +# if ch == curses.ERR: +# return True +# +# return self.cmds.handle(ch) +# +# # main run entry point +# def run (self): +# +# # list of owned ports +# self.owned_ports_list = self.stateless_client.get_acquired_ports() +# +# # data per port +# self.owned_ports = {} +# +# for port_id in self.owned_ports_list: +# self.owned_ports[str(port_id)] = {} +# self.owned_ports[str(port_id)]['streams'] = {} +# +# stream_list = self.stateless_client.get_all_streams(port_id) +# +# self.owned_ports[str(port_id)] = stream_list +# +# self.update_active = True +# while (True): +# +# rc = self.wait_for_key_input() +# if not rc: +# break +# +# self.server_info_panel.draw() +# self.general_info_panel.draw() +# self.control_panel.draw() +# +# # can be different kinds of panels +# self.stats_panel.panel.top() +# self.stats_panel.draw() +# +# panel.update_panels() +# self.stdscr.refresh() +# sleep(0.01) +# +# +## global container +#trex_status = None +# +#def show_trex_status_internal (stdscr, stateless_client): +# global trex_status +# +# if trex_status == None: +# trex_status = CTRexStatus(stdscr, stateless_client) +# +# trex_status.run() +# +#def show_trex_status (stateless_client): +# +# try: +# curses.wrapper(show_trex_status_internal, stateless_client) +# except KeyboardInterrupt: +# curses.endwin() +# +#def cleanup (): +# try: +# curses.endwin() +# except: +# pass +# diff --git a/scripts/stl_test_example.py b/scripts/stl_test_example.py index 7974758d..e9202ca6 100644 --- a/scripts/stl_test_example.py +++ b/scripts/stl_test_example.py @@ -1,42 +1,21 @@ +import os +import sys -# simple test that uses simple API with stateless TRex -#from stl_test_api import BasicTestAPI api_path = os.path.dirname(os.path.abspath(__file__)) -sys.path.insert(0, os.path.join(api_path, '../automation/trex_control_plane/client/')) +sys.path.insert(0, os.path.join(api_path, 'automation/trex_control_plane/client/')) -from trex_stateless_client import CTRexStatelessClient, LoggerApi +from trex_stateless_client import CTRexStatelessClient, STLFailure c = CTRexStatelessClient() + try: c.connect() #before_ipackets = x.get_stats().get_rel('ipackets') - c.cmd_start_line("-f stl/imix_1pkt.yaml -m 5mpps -d 1") - c.cmd_wait_on_traffic() -finally: - c.disconnect() - -#x = BasicTestAPI() -# -#try: -# x.connect() -# -# before_ipackets = x.get_stats().get_rel('ipackets') -# -# print "input packets before test: %s" % before_ipackets -# -# x.inject_profile("stl/imix_1pkt.yaml", rate = "5mpps", duration = 1) -# x.wait_while_traffic_on() -# -# after_ipackets = x.get_stats().get_rel('ipackets') -# -# print "input packets after test: %s" % after_ipackets -# -# if (after_ipackets - before_ipackets) == 5000000: -# print "Test passed :-)\n" -# else: -# print "Test failed :-(\n" -# -#finally: -# x.disconnect() + c.start(profiles = 'stl/imix_3pkt.yaml', ports = [1]) + #c.cmd_wait_on_traffic() +except STLFailure as e: + print e +finally: + c.teardown() |