gfwleak/lijia-tsg-oam
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

marsio,pcap流量信息分别获取;

Browse Source 
使用配置文件tsg_interface.json配置每个网卡的属性: 驱动类型,flow_type等
This commit is contained in:
lijia
2019-08-04 22:45:58 +08:00
parent 9207f15835
commit 3daccfb2e3
7 changed files with 526 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
deploy_etc/tsg_interface.json Normal file
View File

@@ -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"
}]
}

29
py_test/common_logger.py Normal file
View File

@@ -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

10
py_test/common_telegraf.py Normal file
View File

@@ -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

341
py_test/get_traffic_by_marsio.py Normal file
View File

@@ -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')

75
py_test/get_traffic_by_psutil.py Normal file
View File

@@ -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')

51
py_test/iterate_interface_list.py Normal file
View File

@@ -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()

7
py_test/traffic.py
View File

@@ -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))