From 3daccfb2e3041ddac5227ed5e59e0237d9236133 Mon Sep 17 00:00:00 2001 From: lijia Date: Sun, 4 Aug 2019 22:45:58 +0800 Subject: [PATCH] =?UTF-8?q?marsio,pcap=E6=B5=81=E9=87=8F=E4=BF=A1=E6=81=AF?= =?UTF-8?q?=E5=88=86=E5=88=AB=E8=8E=B7=E5=8F=96;=20=E4=BD=BF=E7=94=A8?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E6=96=87=E4=BB=B6tsg=5Finterface.json?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E6=AF=8F=E4=B8=AA=E7=BD=91=E5=8D=A1=E7=9A=84?= =?UTF-8?q?=E5=B1=9E=E6=80=A7:=20=E9=A9=B1=E5=8A=A8=E7=B1=BB=E5=9E=8B,flow?= =?UTF-8?q?=5Ftype=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deploy_etc/tsg_interface.json | 16 ++ py_test/common_logger.py | 29 +++ py_test/common_telegraf.py | 10 + py_test/get_traffic_by_marsio.py | 341 ++++++++++++++++++++++++++++++ py_test/get_traffic_by_psutil.py | 75 +++++++ py_test/iterate_interface_list.py | 51 +++++ py_test/traffic.py | 7 +- 7 files changed, 526 insertions(+), 3 deletions(-) create mode 100644 deploy_etc/tsg_interface.json create mode 100644 py_test/common_logger.py create mode 100644 py_test/common_telegraf.py create mode 100644 py_test/get_traffic_by_marsio.py create mode 100644 py_test/get_traffic_by_psutil.py create mode 100644 py_test/iterate_interface_list.py diff --git a/deploy_etc/tsg_interface.json b/deploy_etc/tsg_interface.json new file mode 100644 index 0000000..444baa2 --- /dev/null +++ b/deploy_etc/tsg_interface.json @@ -0,0 +1,16 @@ +{ + "local_chassis_node": "mcn_1", + "interface_list": [{ + "dev_name": "eth0", + "dev_type": "marsio", + "flow_type": "inline" + }, { + "dev_name": "ens33", + "dev_type": "pcap", + "flow_type": "inline" + }, { + "dev_name": "lo", + "dev_type": "pcap", + "flow_type": "post" + }] +} \ No newline at end of file diff --git a/py_test/common_logger.py b/py_test/common_logger.py new file mode 100644 index 0000000..c4fb524 --- /dev/null +++ b/py_test/common_logger.py @@ -0,0 +1,29 @@ +# coding: utf-8 +import logging + +def logger_init(log_level): + logger = logging.getLogger("logger") + + handler1 = logging.StreamHandler() + #handler2 = logging.FileHandler(filename="test.log") + + logger.setLevel(log_level) + handler1.setLevel(log_level) + #handler2.setLevel(logging.DEBUG) + + formatter = logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s") + handler1.setFormatter(formatter) + #handler2.setFormatter(formatter) + + print("logger level debug val: %d" %(logging.DEBUG)) + print("logger level info val: %d" %(logging.INFO)) + print("logger level warning val: %d" %(logging.WARNING)) + print("logger level error val: %d" %(logging.ERROR)) + print("logger level critical val: %d" %(logging.CRITICAL)) + + logger.addHandler(handler1) + #logger.addHandler(handler2) + + logger.info("logger init succ!") + + return logger \ No newline at end of file diff --git a/py_test/common_telegraf.py b/py_test/common_telegraf.py new file mode 100644 index 0000000..06b39c5 --- /dev/null +++ b/py_test/common_telegraf.py @@ -0,0 +1,10 @@ +# coding: utf-8 + +import telegraf + +def telegraf_init(arg_host, arg_port): + #global telegraf_client + #global_tags = {'host' : "lijia", 'local_ip_addr': "127.0.0.1"} + global_tags = {} + telegraf_client = telegraf.TelegrafClient(host = arg_host, port = arg_port, tags = global_tags) + return telegraf_client \ No newline at end of file diff --git a/py_test/get_traffic_by_marsio.py b/py_test/get_traffic_by_marsio.py new file mode 100644 index 0000000..ac9f831 --- /dev/null +++ b/py_test/get_traffic_by_marsio.py @@ -0,0 +1,341 @@ +#!/usr/bin/python3 +#coding=utf-8 +import argparse +import json +import prettytable +import time +import sys +import signal +import os +import telegraf +import socket +import logging +import logging.handlers +from common_telegraf import * + +TBPS = (1 * 1000 * 1000 * 1000 * 1000) +GBPS = (1 * 1000 * 1000 * 1000) +MBPS = (1 * 1000 * 1000) +KBPS = (1 * 1000) + +#G_JSON_PATH = './mrmonit.daemon' +G_JSON_PATH = '/var/run/mrzcpd/mrmonit.daemon' + +TITLE_VECTOR = ['PhyRXFrame','PhyRXBits','PhyRXMissed','PhyRXError', + 'PhyRXNoBUF','PhyTXFrame','PhyTXBits','PhyTXError', + 'UsrRXDrops', 'UsrTXDrops'] + +TITLE_MAP = { 'PhyRXFrame' : 'ipackets', + 'PhyRXBits' : 'ibytes', + 'PhyRXMissed' : 'imissed', + 'PhyRXError' : 'ierrors', + 'PhyRXNoBUF' : 'rxnombuf', + 'PhyTXFrame' : 'opackets', + 'PhyTXBits' : 'obytes', + 'PhyTXError' : 'oerrors', + 'UsrRXDrops' : 'userrxdrop', + 'UsrTXDrops' : 'usertxdrop' + } + +g_debug_mode = 0 + +def debug_print(msg): + #print("###### in debug_print(), g_debug_mode=%d " %(g_debug_mode)) + if g_debug_mode != 0: + print("##### %s" %(msg)) + + +def locate_vector_by_symbol(vector, symbol): + return [s for s in vector if s['symbol'] == symbol] + +def list_all_phydev(json_fp): + return [ s['symbol'] for s in json_fp['device']] + +def phydev_value_read(json_fp, str_device, str_item): + phydevs = locate_vector_by_symbol(json_fp['device'], str_device) + return phydevs[0]['stats']['accumulative'][str_item] + +def phydev_speed_read(json_fp, str_device, str_item): + phydevs = locate_vector_by_symbol(json_fp['device'], str_device) + if len(phydevs) <= 0: + return "" + return phydevs[0]['stats']['speed'][str_item] + +def trans_to_human_readable(value): + if value > TBPS: + return value * 1.0 / TBPS, 'T' + if value > GBPS: + return value * 1.0 / GBPS, 'G' + if value > MBPS: + return value * 1.0 / MBPS, 'M' + if value > KBPS: + return value * 1.0 / KBPS, 'K' + + return value * 1.0, ' ' + + +def sendlog_by_socket(table_phydev, devsym): + global udp_sock_fd + global telegraf_ip + global telegraf_port + tmp_str = "" + + print(table_phydev) + line_prot_buf = "interface,device=%s,flow_type=inline " %(devsym) + + for num in range(0,9): + #Accumulative + #tmp_str = "%s=%s," %(TITLE_VECTOR[num],ValueList[num+1]) + # per second + tmp_str = "%s=%s," %(TITLE_VECTOR[num],SpeedList[num+1]) + line_prot_buf += tmp_str + num += 1 + #tmp_str = "%s=%s" %(TITLE_VECTOR[num],ValueList[num+1]) + tmp_str = "%s=%s" %(TITLE_VECTOR[num],SpeedList[num+1]) + line_prot_buf += tmp_str + logger.debug("##### sendto line_protocol: %s" %(line_prot_buf)) + + try: + # for python2.x + sock_ret = udp_sock_fd.sendto(line_prot_buf, (telegraf_ip, telegraf_port)) + + # for python3.x, str to bytes + #sock_ret = udp_sock_fd.sendto(line_prot_buf.encode(), (telegraf_ip, telegraf_port)) + except socket.err: + logger.critical("##### sendto error!") + sys.exit(1) + + logger.debug("##### udp_sock_fd.sendto ret=%d" %(sock_ret)) + +# APM sendlog format +def get_and_send_marsio_traffic(json_fp, telegraf_client, devsym, arg_flow_type, arg_node_name): + __metric_dict_speed = {} + metric_tag = {'device': devsym, 'flow_type':arg_flow_type, 'node':arg_node_name} + + for item in TITLE_VECTOR: + value = phydev_speed_read(json_fp, devsym, TITLE_MAP[item]) + if len(value) <= 0: + print("Error! can't get interface '%s' traffic" %(devsym)) + sys.exit(1) + __metric_dict_speed[item] = value + + telegraf_client.metric('interface', __metric_dict_speed, tags = metric_tag) + #logger.info("telegraf_client send metric") + print(__metric_dict_speed) + + return + +def dump_human_table(json_fp, devsym, is_human_number = 0): + + print('\nTime: %s, Physical device: %s' % (time.strftime('%c'), devsym)) + + table_phydev = prettytable.PrettyTable([' '] + TITLE_VECTOR, + vertical_char=' ',horizontal_char = '-', junction_char=' ') + + ValueList = ['Accumulative'] + SpeedList = ['Per Second'] + + table_phydev.align[' '] = 'r' + + for item in TITLE_VECTOR: + table_phydev.align[item] = 'r' + + for item in TITLE_VECTOR: + value = phydev_value_read(json_fp, devsym, TITLE_MAP[item]) + ValueList.append(value) + + for item in TITLE_VECTOR: + speed = phydev_speed_read(json_fp, devsym, TITLE_MAP[item]) + SpeedList.append(speed) + + # ##################################add by lijia for tsg oam + # sendlog_by_socket(table_phydev, devsym) + dump_apm_sendlog(json_fp, telegraf_client, devsym) + # ################################## add by lijia for tsg oam + + #table_phydev.add_row(ValueList) + table_phydev.add_row(SpeedList) + print(table_phydev) + + +def setup_argv_parser(phydev_list): + + parser = argparse.ArgumentParser(description='Marsio ZeroCopy Tools -- Monitor NIC devices') + + parser.add_argument('-t', '--time', help = 'interval, seconds to wait between updates', + type=int, default = 1) + parser.add_argument('-l', '--loop', help = 'print loop, exit when recv a signal', + action='store_true', default = 0) + parser.add_argument('-H', '--human-readable', help = 'print value in human readable format', + action = 'store_true', default = 0) + parser.add_argument('-i', '--interface', help = 'the name of network interface', + action = 'append', choices=phydev_list) + parser.add_argument('--clear-screen', help = 'clear screen at start of loop', + action='store_true', default = 0) + + # APM sendlog options + parser.add_argument('--sendlog-apm', help = 'send log to apm server', + action='store_true', default = 1) + parser.add_argument('--sendlog-apm-cfg', help = 'send log configure file', + type=str, default = '/opt/mrzcpd/etc/mrsendlog.conf') + + ############### add by lijia for OAM + parser.add_argument('-d', '--debug', help = 'debug mode', + action='store_true', default = 0) + parser.add_argument('--telegraf-ip', help = 'send log to telegraf ip address', + type=str, default = '192.168.200.5') + parser.add_argument('--telegraf-port', help = 'send log to telegraf port', + type=int, default = 8126) + ############### add by lijia for OAM + return parser.parse_args() + +def phydev_json_load(): + with open(G_JSON_PATH) as json_fp: + return json.load(json_fp) + + +def sendlog_hostname(test_hostname, test_port): + import socket + + hostname = socket.gethostname() + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.connect((test_hostname, int(test_port))) + local_ip_addr = s.getsockname()[0] + except: + local_ip_addr = '127.0.0.1' + finally: + s.close() + + return hostname, local_ip_addr + +def sendlog_apm_init(r_option): + #import ConfigParser + #import urllib + + #config = ConfigParser.ConfigParser() + #config.read(r_option.sendlog_apm_cfg) + + #apm_server_url = config.get('sendlog_apm', 'apm_device_server') + #url_proto, rest = urllib.splittype(apm_server_url) + #url_host, rest = urllib.splithost(rest) + #url_host, url_port = urllib.splitport(url_host) + + global telegraf_client + + hostname, local_ip_addr = sendlog_hostname(r_option.telegraf_ip, r_option.telegraf_port) + sendlog_tags = {'node': local_ip_addr} + + telegraf_client = telegraf.TelegrafClient(host = r_option.telegraf_ip, + port = int(r_option.telegraf_port), tags = sendlog_tags) + #telegraf_client = telegraf.TelegrafClient(host = '192.168.11.219', + # port = 8126, tags = sendlog_tags) + + logger.debug("##### sendlog_apm_init succ, telegraf_ip=%s, telegraf_port=%d." %(r_option.telegraf_ip, r_option.telegraf_port)) + + return telegraf_client + +def sigint_handler(handler, frame): + sys.exit(0) + +def main(): + global g_debug_mode + + signal.signal(signal.SIGINT, sigint_handler) + + try: + json_fp = phydev_json_load() + phydev_list = list_all_phydev(json_fp) + print("##### phydev_list:%s" %(phydev_list)) + r_option = setup_argv_parser(phydev_list) + + telegraf_ip = r_option.telegraf_ip + telegraf_port = r_option.telegraf_port + + if r_option.debug : + g_debug_mode = 1 + + debug_print("##### debug mode is on!") + + telegraf_client = sendlog_apm_init(r_option) + + phydev_user_list = phydev_list if r_option.interface is None else r_option.interface + + debug_print("phydev_user_list:%s" %(phydev_user_list)) + + + + debug_print("telegraf_ip:%s, telegraf_port:%u" %(telegraf_ip, telegraf_port)) + + while True: + if r_option.clear_screen: + os.system('clear') + + json_fp = phydev_json_load() + + if r_option.sendlog_apm: + for devsym in phydev_user_list: + dump_apm_sendlog(json_fp, telegraf_client, devsym) + else: + for devsym in phydev_user_list: + dump_human_table(json_fp, devsym, r_option.human_readable) + + if not r_option.loop: + break + + time.sleep(r_option.time) + + except KeyboardInterrupt: + pass + except ValueError as err: + print(("%s, perhaps mrzcpd program is not running.") % str(err)) + except IOError as err: + print(("%s, perhaps mrzcpd program is not running.") % str(err)) + + return 0 + +def logger_init(): + global logger + logger = logging.getLogger("logger") + + handler1 = logging.StreamHandler() + #handler2 = logging.FileHandler(filename="test.log") + + logger.setLevel(logging.DEBUG) + handler1.setLevel(logging.DEBUG) + #handler2.setLevel(logging.DEBUG) + + formatter = logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s") + handler1.setFormatter(formatter) + #handler2.setFormatter(formatter) + + print("logger level debug val: %d" %(logging.DEBUG)) + print("logger level info val: %d" %(logging.INFO)) + print("logger level warning val: %d" %(logging.WARNING)) + print("logger level error val: %d" %(logging.ERROR)) + print("logger level critical val: %d" %(logging.CRITICAL)) + + logger.addHandler(handler1) + #logger.addHandler(handler2) + + logger.info("logger init succ!") + +def socket_init(): + global udp_sock_fd + global telegraf_ip + global telegraf_port + + try: + udp_sock_fd = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + except socket.error: + logger.critical("logger init succ!") + sys.exit(1) + +def send_metrics_by_marsio(telegraf_client, dev_name, arg_flow_type, arg_node_name): + json_fp = phydev_json_load() + get_and_send_marsio_traffic(json_fp, telegraf_client, dev_name, arg_flow_type, arg_node_name) + +if __name__ == '__main__': + global telegraf_client + telegraf_client = telegraf_init('127.0.0.1', 8126) + send_metrics_by_marsio(telegraf_client, 'ens33', 'inline', 'mcn_1') diff --git a/py_test/get_traffic_by_psutil.py b/py_test/get_traffic_by_psutil.py new file mode 100644 index 0000000..194ca29 --- /dev/null +++ b/py_test/get_traffic_by_psutil.py @@ -0,0 +1,75 @@ +# coding: utf-8 + +import sys +import psutil +import time +import telegraf + +def get_stats_for_device(dev_name): + + devlist = psutil.net_io_counters(pernic=True).keys() + found_expect_dev = 0 + + for dev in devlist: + if dev == dev_name: + found_expect_dev = 1 + + + if found_expect_dev == 0: + print("#### not found dev:%s" %(dev_name)) + sys.exit(1) + + print("#### found dev:%s" %(dev_name)) + print(psutil.net_io_counters(pernic=True).get(dev_name)) + + packets_recv = psutil.net_io_counters(pernic=True).get(dev_name).packets_recv + bytes_recv = psutil.net_io_counters(pernic=True).get(dev_name).bytes_recv + errin = psutil.net_io_counters(pernic=True).get(dev_name).errin + dropin = psutil.net_io_counters(pernic=True).get(dev_name).dropin + + packets_sent = psutil.net_io_counters(pernic=True).get(dev_name).packets_sent + bytes_sent = psutil.net_io_counters(pernic=True).get(dev_name).bytes_sent + errout = psutil.net_io_counters(pernic=True).get(dev_name).errout + dropout = psutil.net_io_counters(pernic=True).get(dev_name).dropout + + return packets_recv,bytes_recv,errin,dropin,packets_sent,bytes_sent,errout,dropout + +def send_metrics_by_psutil(telegraf_client, dev_name, arg_flow_type, arg_node_name): + metric_tag = {'device': dev_name, 'flow_type':arg_flow_type, 'node':arg_node_name} + + old_packets_recv,old_bytes_recv,old_errin,old_dropin,old_packets_sent,old_bytes_sent,old_errout,old_dropout = get_stats_for_device(dev_name) + time.sleep(3) + new_packets_recv,new_bytes_recv,new_errin,new_dropin,new_packets_sent,new_bytes_sent,new_errout,new_dropout = get_stats_for_device(dev_name) + + print("pps:%d bps:%d error:%d" %(new_packets_recv - old_packets_recv, 8 *(new_bytes_recv - old_bytes_recv), new_errin-old_errin)) + + metrict_val = {} + metrict_val['PhyRXBits'] = int(8 *(new_bytes_recv - old_bytes_recv)) + metrict_val['PhyRXError'] = int(new_errin-old_errin) + metrict_val['PhyRXFrame'] = int(new_packets_recv - old_packets_recv) + metrict_val['PhyRXMissed'] = 0 + metrict_val['PhyRXNoBUF'] = 0 + + metrict_val['PhyTXBits'] = int(8 *(new_bytes_recv - old_bytes_recv)) + metrict_val['PhyTXError'] = int(new_errin-old_errin) + metrict_val['PhyTXFrame'] = int(new_packets_recv - old_packets_recv) + metrict_val['PhyRXMissed'] = 0 + metrict_val['PhyRXNoBUF'] = 0 + + metrict_val['UsrRXDrops'] = int(new_dropin - new_dropin) + metrict_val['UsrTXDrops'] = int(new_dropout - old_dropout) + + telegraf_client.metric('interface', metrict_val, tags = metric_tag) + +if __name__ == '__main__': + + global time_interval + if len(sys.argv) <= 1: + print("Usage: %s timeinterval" %(sys.argv[0])) + time_interval = 5 + else: + time_interval = int(sys.argv[1]) + + telegraf_client = telegraf_init() + + send_metrics_by_psutil(telegraf_client, 'ens33', 'inline', 'mcn_1') diff --git a/py_test/iterate_interface_list.py b/py_test/iterate_interface_list.py new file mode 100644 index 0000000..0768cc8 --- /dev/null +++ b/py_test/iterate_interface_list.py @@ -0,0 +1,51 @@ +# coding: utf-8 + +import sys +import psutil +import time +import json +from get_traffic_by_psutil import * +from get_traffic_by_marsio import * +from common_telegraf import * + +INTERFACE_JSON_PATH = '/opt/tsg/etc/tsg_interface.json' + + +def get_local_node(): + try: + with open(INTERFACE_JSON_PATH) as json_fp: + json_dict = json.load(json_fp) + return json_dict['local_chassis_node'] + except IOError: + return "" + + +def get_interface_list(): + try: + with open(INTERFACE_JSON_PATH) as json_fp: + json_dict = json.load(json_fp) + interface_list = json_dict['interface_list'] + #print(interface_list) + for dev in interface_list: + if 'marsio' == dev['dev_type']: + send_metrics_by_marsio(telegraf_client, dev['dev_name'], dev['flow_type'], node_name) + #print("TODO: marsio to do!") + elif 'pcap' == dev['dev_type']: + send_metrics_by_psutil(telegraf_client, dev['dev_name'], dev['flow_type'], node_name) + else: + sys.exit(1) + print(dev['dev_name']) + print(dev['dev_type']) + print(dev['flow_type']) + print(node_name) + except IOError: + return "" + + +if __name__ == '__main__': + global node_name + global telegraf_client + + telegraf_client = telegraf_init('127.0.0.1', 8126) + node_name = get_local_node() + get_interface_list() \ No newline at end of file diff --git a/py_test/traffic.py b/py_test/traffic.py index 99e4a66..1af06a8 100644 --- a/py_test/traffic.py +++ b/py_test/traffic.py @@ -1,6 +1,4 @@ -#!/usr/bin/env python # coding: utf-8 -# author: Xiao Guaishou try: import psutil @@ -21,6 +19,7 @@ def get_key(): print("--------------------------------------------") for key in key_info: + print(key) print(psutil.net_io_counters()) recv.setdefault(key, psutil.net_io_counters(pernic=True).get(key).bytes_recv) sent.setdefault(key, psutil.net_io_counters(pernic=True).get(key).bytes_sent) @@ -58,4 +57,6 @@ def traffic_show(): exit() if __name__ == '__main__': - traffic_show() + #traffic_show() + key_info = psutil.net_io_counters().keys() + print("#### %s" %(key_info))