monit_stream增加按任意间隔时间发送APM性能监控指标的功能。

2 files changed, 153 insertions, 108 deletions

diff --git a/tools/monit_device/monit_device.py b/tools/monit_device/monit_device.py
index f1de310..299f81b 100644
--- a/tools/monit_device/monit_device.py
+++ b/tools/monit_device/monit_device.py
@@ -116,7 +116,7 @@ def dump_apm_sendlog(json_fp, telegraf_client, devsym, lasttime_metrics):
 
     for item in TITLE_VECTOR:
         delta_metrics[item] = thistime_metrics[item] - \
-            lasttime_metrics.get(item, 0)
+            lasttime_metrics.get(item, thistime_metrics[item])
 
     telegraf_client.metric('mr4_device_rxtx_value', delta_metrics, tags=__tags)
     return thistime_metrics
diff --git a/tools/monit_stream/monit_stream.py b/tools/monit_stream/monit_stream.py
index bbb8cfa..e7a5840 100644
--- a/tools/monit_stream/monit_stream.py
+++ b/tools/monit_stream/monit_stream.py
@@ -28,45 +28,51 @@ TITLE_VECTOR_FTX = ['FTxPkts', 'FTxBits', 'FTxDrops']
 TITLE_APP_STAT = ['PKTRx', 'PKTTx', 'MbufAlloc', 'MbufFree', 'MbufInUse']
 
 TITLE_APP_MAP = {
-    'PKTRx' : 'packet_recv_count',
-    'PKTTx' : 'packet_send_count',
-    'MbufAlloc' : 'mbuf_alloc_count',
-    'MbufFree' : 'mbuf_free_count',
-    'MbufInUse' : 'mbuf_in_use_count'
-    }
-
-TITLE_MAP = { 'RxOnline'     : 'rx_on_line',
-              'RxPkts'       : 'rx_deliver',
-              'RxDrops'      : 'rx_missed',
-              'RxBits'       : 'rx_total_len',
-              'TxOnline'     : 'tx_on_line',
-              'TxPkts'       : 'tx_deliver',
-              'TxDrops'      : 'tx_missed',
-              'TxBits'       : 'tx_total_len',
-              'FTXOnline'    : 'ftx_on_line',
-              'FTxPkts'      : 'ftx_deliver',
-              'FTxDrops'     : 'ftx_missed',
-              'FTxBits'      : 'ftx_total_len'
-            }
+    'PKTRx': 'packet_recv_count',
+    'PKTTx': 'packet_send_count',
+    'MbufAlloc': 'mbuf_alloc_count',
+    'MbufFree': 'mbuf_free_count',
+    'MbufInUse': 'mbuf_in_use_count'
+}
+
+TITLE_MAP = {'RxOnline': 'rx_on_line',
+             'RxPkts': 'rx_deliver',
+             'RxDrops': 'rx_missed',
+             'RxBits': 'rx_total_len',
+             'TxOnline': 'tx_on_line',
+             'TxPkts': 'tx_deliver',
+             'TxDrops': 'tx_missed',
+             'TxBits': 'tx_total_len',
+             'FTXOnline': 'ftx_on_line',
+             'FTxPkts': 'ftx_deliver',
+             'FTxDrops': 'ftx_missed',
+             'FTxBits': 'ftx_total_len'
+             }
+
 
 def locate_vector_by_symbol(vector, symbol):
     return [s for s in vector if s['symbol'] == symbol]
 
+
 def list_all_vdev(json_fp):
-    return [ s['symbol'] for s in json_fp['raw']]
+    return [s['symbol'] for s in json_fp['raw']]
+
 
 def vdev_value_read(json_fp, str_device, str_item):
     phydevs = locate_vector_by_symbol(json_fp['raw'], str_device)
     return phydevs[0]['stats']['accumulative'][str_item]
 
+
 def vdev_speed_read(json_fp, str_device, str_item):
     phydevs = locate_vector_by_symbol(json_fp['raw'], str_device)
     return phydevs[0]['stats']['speed'][str_item]
 
+
 def vdev_streams_read(json_fp, str_device):
     phydevs = locate_vector_by_symbol(json_fp['raw'], str_device)
     return phydevs[0]['rxstreams'], phydevs[0]['txstreams']
 
+
 def trans_to_human_readable(value):
     if value > TBPS:
         return value * 1.0 / TBPS, 'T'
@@ -79,6 +85,7 @@ def trans_to_human_readable(value):
 
     return value * 1.0, ' '
 
+
 def vec_trans_to_human_readable(vec):
     r_vector = []
     for value in vec:
@@ -87,6 +94,7 @@ def vec_trans_to_human_readable(vec):
 
     return r_vector
 
+
 def dump_one_device(json_fp, devsym, title_vector_rx, title_vector_tx, speed):
 
     __rd_function = vdev_value_read if speed == 0 else vdev_speed_read
@@ -99,60 +107,66 @@ def dump_one_device(json_fp, devsym, title_vector_rx, title_vector_tx, speed):
 
         for item in title_vector_rx:
             value = __rd_function(json_fp, devsym, TITLE_MAP[item])[stream_id]  \
-                    if stream_id < nr_rxstream else 0
+                if stream_id < nr_rxstream else 0
             ValueList.append(value)
 
         for item in title_vector_tx:
             value = __rd_function(json_fp, devsym, TITLE_MAP[item])[stream_id]  \
-                    if stream_id < nr_txstream else 0
+                if stream_id < nr_txstream else 0
             ValueList.append(value)
 
-        for i,v in enumerate(ValueList):
+        for i, v in enumerate(ValueList):
             ValueListSum[i] += v
 
     return ValueListSum
 
+
 def dump_summary_table(json_fp, appsym, devsym, title_vector_rx, title_vector_tx,
-    is_human_number = 0, speed = 0):
+                       is_human_number=0, speed=0):
 
     print('\nTime: %s, App: %s' % (time.strftime('%c'), appsym))
-    table_phydev = prettytable.PrettyTable([' ']  + title_vector_rx + title_vector_tx,
-                                           vertical_char=' ',horizontal_char = '-', junction_char=' ')
+    table_phydev = prettytable.PrettyTable([' '] + title_vector_rx + title_vector_tx,
+                                           vertical_char=' ', horizontal_char='-', junction_char=' ')
 
-    for item in[' ']  + title_vector_rx + title_vector_tx:
+    for item in [' '] + title_vector_rx + title_vector_tx:
         table_phydev.align[item] = 'r'
 
     ValueListTotal = [0] * len(title_vector_rx + title_vector_tx)
 
     for dev in devsym:
-        ValueListSum = dump_one_device(json_fp, dev, title_vector_rx, title_vector_tx, speed)
+        ValueListSum = dump_one_device(
+            json_fp, dev, title_vector_rx, title_vector_tx, speed)
 
-        for i,v in enumerate(ValueListSum):
+        for i, v in enumerate(ValueListSum):
             ValueListTotal[i] += v
 
         if is_human_number:
-            table_phydev.add_row([dev] + vec_trans_to_human_readable(ValueListSum))
+            table_phydev.add_row(
+                [dev] + vec_trans_to_human_readable(ValueListSum))
         else:
             table_phydev.add_row([dev] + ValueListSum)
 
     if is_human_number:
-        table_phydev.add_row(['Total'] + vec_trans_to_human_readable(ValueListTotal))
+        table_phydev.add_row(
+            ['Total'] + vec_trans_to_human_readable(ValueListTotal))
     else:
         table_phydev.add_row(['Total'] + ValueListTotal)
 
     print(table_phydev)
 
+
 def dump_human_table(json_fp, appsym, devsym,  title_vector_rx, title_vector_tx,
-    is_human_number = 0, speed = 0):
+                     is_human_number=0, speed=0):
 
-    print('\nTime: %s, App: %s, Device: %s ' % (time.strftime('%c'), appsym, devsym))
+    print('\nTime: %s, App: %s, Device: %s ' %
+          (time.strftime('%c'), appsym, devsym))
 
-    table_phydev = prettytable.PrettyTable([' ']  + title_vector_rx + title_vector_tx,
-                                           vertical_char=' ',horizontal_char = '-', junction_char=' ')
+    table_phydev = prettytable.PrettyTable([' '] + title_vector_rx + title_vector_tx,
+                                           vertical_char=' ', horizontal_char='-', junction_char=' ')
 
     __rd_function = vdev_value_read if speed == 0 else vdev_speed_read
 
-    for item in[' ']  + title_vector_rx + title_vector_tx:
+    for item in [' '] + title_vector_rx + title_vector_tx:
         table_phydev.align[item] = 'r'
 
     ValueListSum = [0] * len(title_vector_rx + title_vector_tx)
@@ -163,12 +177,12 @@ def dump_human_table(json_fp, appsym, devsym,  title_vector_rx, title_vector_tx,
 
         for item in title_vector_rx:
             value = __rd_function(json_fp, devsym, TITLE_MAP[item])[stream_id]  \
-                    if stream_id < nr_rxstream else 0
+                if stream_id < nr_rxstream else 0
             ValueList.append(value)
 
         for item in title_vector_tx:
             value = __rd_function(json_fp, devsym, TITLE_MAP[item])[stream_id]  \
-                    if stream_id < nr_txstream else 0
+                if stream_id < nr_txstream else 0
             ValueList.append(value)
 
         str_leader = ''
@@ -176,29 +190,32 @@ def dump_human_table(json_fp, appsym, devsym,  title_vector_rx, title_vector_tx,
         str_leader += 'TX[%d]' % stream_id if stream_id < nr_txstream else ''
 
         if is_human_number:
-            table_phydev.add_row([str_leader] + vec_trans_to_human_readable(ValueList))
+            table_phydev.add_row(
+                [str_leader] + vec_trans_to_human_readable(ValueList))
         else:
             table_phydev.add_row([str_leader] + ValueList)
 
-        for i,v in enumerate(ValueList):
+        for i, v in enumerate(ValueList):
             ValueListSum[i] += v
 
     if is_human_number:
-        table_phydev.add_row(['Total'] + vec_trans_to_human_readable(ValueListSum))
+        table_phydev.add_row(
+            ['Total'] + vec_trans_to_human_readable(ValueListSum))
     else:
         table_phydev.add_row(['Total'] + ValueListSum)
 
     print(table_phydev)
 
+
 def dump_status_table(json_fp, appsym):
     json_fp_appstat = json_fp['appstat']
     nr_stream = len(json_fp['appstat']['packet_recv_count'])
-    
+
     print('\nTime: %s, App: %s' % (time.strftime('%c'), appsym))
-    table_phydev = prettytable.PrettyTable(['TID']  + TITLE_APP_STAT, 
-        vertical_char=' ',horizontal_char = '-', junction_char=' ')
+    table_phydev = prettytable.PrettyTable(['TID'] + TITLE_APP_STAT,
+                                           vertical_char=' ', horizontal_char='-', junction_char=' ')
 
-    for item in['TID']  + TITLE_APP_STAT:
+    for item in ['TID'] + TITLE_APP_STAT:
         table_phydev.align[item] = 'r'
 
     ValueListSum = [0] * len(TITLE_APP_STAT)
@@ -208,85 +225,98 @@ def dump_status_table(json_fp, appsym):
         for item in TITLE_APP_STAT:
             value = json_fp_appstat[TITLE_APP_MAP[item]][tid]
             ValueList.append(value)
-        
+
         table_phydev.add_row([tid] + ValueList)
-        for i,v in enumerate(ValueList):
+        for i, v in enumerate(ValueList):
             ValueListSum[i] += v
 
     table_phydev.add_row(['Total'] + ValueListSum)
     print(table_phydev)
 
-def dump_apm_sendlog(json_fp, telegraf_client, appsym, user_interface, title_vector_rx, title_vector_tx):
-    
+
+def dump_apm_sendlog(json_fp, telegraf_client, appsym, user_interface, title_vector_rx, title_vector_tx, lasttime_metrics_dict):
+
+    thistime_metrics_dict = {}
+
     for dev in user_interface:
-        ValueListSumSpeed = dump_one_device(json_fp, dev, title_vector_rx, title_vector_tx, 1)
-        ValueListSumValue = dump_one_device(json_fp, dev, title_vector_rx, title_vector_tx, 0)
+        lasttime_metrics_dict_per_dev = lasttime_metrics_dict.get(dev, {})
+        thistime_metrics_dict_per_dev = {}
+
+        ValueListSumValue = dump_one_device(
+            json_fp, dev, title_vector_rx, title_vector_tx, 0)
 
-        sendlog_dict_speed = {}
         sendlog_dict_value = {}
         sendlog_tag = {'app': appsym, 'device': dev}
 
         for id, value in enumerate(title_vector_rx + title_vector_tx):
-            sendlog_dict_speed[value] = ValueListSumSpeed[id]
-            sendlog_dict_value[value] = ValueListSumValue[id]
+            thistime_metrics_dict_per_dev[value] = ValueListSumValue[id]
+            sendlog_dict_value[value] = thistime_metrics_dict_per_dev[value] - \
+                lasttime_metrics_dict_per_dev.get(value, thistime_metrics_dict_per_dev[value])
 
-        telegraf_client.metric('mr4_stream_rxtx_speed', sendlog_dict_speed, tags = sendlog_tag)
-        telegraf_client.metric('mr4_stream_rxtx_value', sendlog_dict_value, tags = sendlog_tag)
+        print(sendlog_dict_value)
+        telegraf_client.metric('mr4_stream_rxtx_value', sendlog_dict_value, tags=sendlog_tag)
 
-    return
+        thistime_metrics_dict[dev] = thistime_metrics_dict_per_dev
+
+    return thistime_metrics_dict
 
 
 def setup_argv_parser(applist):
 
     parser = argparse.ArgumentParser(description='Marsio ZeroCopy Tools -- Monitor stream information',
-                                     version = 'Marsio ZeroCopy Tools Suite 4.1')
-
-    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('-s', '--speed', help = 'print speed value instead of accumulative value',
-                        action = 'store_true', default = 0)
-    parser.add_argument('-i', '--interface', help = 'the name of network interface',
-                        action = 'append')
-    parser.add_argument('-m', '--metrics', help = 'group of metrics', choices=['rx','tx','ftx'],
-                        default = ['rx','tx'])
-    parser.add_argument('--clear-screen', help = 'clear screen at start of loop', 
-                        action='store_true', default = 0)
-    parser.add_argument('--per-stream', help = 'print per thread/stream value', 
-                        action='store_true', default = 0)
-    parser.add_argument('--status', help = 'print application running status',
-                        action='store_true', default = 0)
-    parser.add_argument('app', metavar='APP', help = 'the name of slave application', nargs = '*',
+                                     version='Marsio ZeroCopy Tools Suite 4.1')
+
+    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('-s', '--speed', help='print speed value instead of accumulative value',
+                        action='store_true', default=0)
+    parser.add_argument('-i', '--interface', help='the name of network interface',
+                        action='append')
+    parser.add_argument('-m', '--metrics', nargs='+', help='group of metrics', choices=['rx', 'tx', 'ftx'],
+                        default=['rx', 'tx'])
+    parser.add_argument('--clear-screen', help='clear screen at start of loop',
+                        action='store_true', default=0)
+    parser.add_argument('--per-stream', help='print per thread/stream value',
+                        action='store_true', default=0)
+    parser.add_argument('--status', help='print application running status',
+                        action='store_true', default=0)
+    parser.add_argument('app', metavar='APP', help='the name of slave application', nargs='*',
                         default=applist)
 
     # APM sendlog options
-    parser.add_argument('--sendlog-apm', help = 'send log to apm server',
-                        action='store_true', default = 0)
-    parser.add_argument('--sendlog-apm-cfg', help = 'send log configure file',
-                        type=str, default = '/opt/mrzcpd/etc/mrsendlog.conf')
+    parser.add_argument('--sendlog-apm', help='send log to apm server',
+                        action='store_true', default=0)
+    parser.add_argument('--sendlog-apm-cfg', help='send log configure file',
+                        type=str, default='/opt/mrzcpd/etc/mrsendlog.conf')
 
     return parser.parse_args()
 
+
 def global_json_load():
     with open(G_JSON_PATH) as json_fp:
         return json.load(json_fp)
 
+
 def app_json_load(appsym):
     with open(G_APP_JSON_PATH % appsym) as json_fp:
         return json.load(json_fp)
 
+
 def app_symbol_load():
     j_global = global_json_load()
     return [s["symbol"] for s in j_global["app"] if s["registed"] == 1]
 
+
 def sigint_handler(handler, frame):
     sys.exit(0)
 
+
 def check_vdev_options(json_fp, r_option):
-    
+
     if r_option.interface == None:
         return
 
@@ -296,9 +326,10 @@ def check_vdev_options(json_fp, r_option):
             print("monit_stream: error: argument -i/--interface: invalid interface.")
             sys.exit(1)
 
+
 def sendlog_hostname(test_hostname, test_port):
     import socket
-    
+
     hostname = socket.gethostname()
     s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     try:
@@ -311,6 +342,7 @@ def sendlog_hostname(test_hostname, test_port):
 
     return hostname, local_ip_addr
 
+
 def sendlog_apm_init(r_option):
     import ConfigParser
     import urllib
@@ -324,16 +356,22 @@ def sendlog_apm_init(r_option):
     url_host, url_port = urllib.splitport(url_host)
 
     hostname, local_ip_addr = sendlog_hostname(url_host, url_port)
-    sendlog_tags = {'host' : hostname, 'local_ip_addr': local_ip_addr}
-        
+    sendlog_tags = {'host': hostname, 'local_ip_addr': local_ip_addr}
+
     if url_proto == 'udp':
-        telegraf_client = telegraf.TelegrafClient(host = url_host, 
-            port = int(url_port), tags = sendlog_tags)
+        telegraf_client = telegraf.TelegrafClient(host=url_host,
+                                                  port=int(url_port), tags=sendlog_tags)
     elif url_proto == 'http':
-        telegraf_client = telegraf.HttpClient(host = url_host, 
-            port = int(url_port), tags = sendlog_tags)
+        telegraf_client = telegraf.HttpClient(host=url_host,
+                                              port=int(url_port), tags=sendlog_tags)
+
+    try:
+        send_interval = config.get('sendlog_apm', 'apm_send_interval')
+    except:
+        send_interval = 5
+
+    return telegraf_client, send_interval
 
-    return telegraf_client
 
 def main():
     signal.signal(signal.SIGINT, sigint_handler)
@@ -365,45 +403,51 @@ def main():
         title_vector_tx.extend(TITLE_VECTOR_FTX)
 
     try:
-       if r_option.sendlog_apm:
-            telegraf_client = sendlog_apm_init(r_option)
+        if r_option.sendlog_apm:
+            telegraf_client, sleep_interval = sendlog_apm_init(r_option)
+        else:
+            sleep_interval = r_option.time
+
     except:
-       print("APM sendlog setup failed.")
-       raise
-       sys.exit(1)
-    
+        print("APM sendlog setup failed.")
+        raise
+        sys.exit(1)
+
     try:
+        lasttime_metrics_dict = {}
         while True:
-
             if r_option.clear_screen:
                 os.system('clear')
 
             for appsym in r_option.app:
                 json_fp = app_json_load(appsym)
                 check_vdev_options(json_fp, r_option)
-                user_interface = r_option.interface if r_option.interface != None else list_all_vdev(json_fp)
+                user_interface = r_option.interface if r_option.interface != None else list_all_vdev(
+                    json_fp)
 
                 if r_option.status:
                     dump_status_table(json_fp, appsym)
                     continue
 
                 if r_option.sendlog_apm:
-                    dump_apm_sendlog(json_fp, telegraf_client, appsym, user_interface, 
-                        title_vector_rx, title_vector_tx)
+                    lasttime_metrics = lasttime_metrics_dict.get(appsym, {})
+                    thistime_metrics = dump_apm_sendlog(
+                        json_fp, telegraf_client, appsym, user_interface, title_vector_rx, title_vector_tx, lasttime_metrics)
+                    lasttime_metrics_dict[appsym] = thistime_metrics
                     continue
 
                 if not r_option.per_stream:
-                   dump_summary_table(json_fp, appsym, user_interface, title_vector_rx, title_vector_tx,
-                        r_option.human_readable, r_option.speed)
+                    dump_summary_table(json_fp, appsym, user_interface, title_vector_rx, title_vector_tx,
+                                       r_option.human_readable, r_option.speed)
                 else:
                     for devsym in user_interface:
                         dump_human_table(json_fp, appsym, devsym, title_vector_rx, title_vector_tx,
-                        r_option.human_readable, r_option.speed)
+                                         r_option.human_readable, r_option.speed)
 
             if not r_option.loop:
                 break
 
-            time.sleep(r_option.time)
+            time.sleep(float(sleep_interval))
 
     except KeyboardInterrupt:
         pass
@@ -414,5 +458,6 @@ def main():
 
     return 0
 
+
 if __name__ == '__main__':
-    main()
\ No newline at end of file
+    main()