增加Prometheus Client功能。

1 files changed, 77 insertions, 61 deletions

diff --git a/tools/monit_stream/monit_stream.py b/tools/monit_stream/monit_stream.py
index 9175ef0..1d2fe2c 100644
--- a/tools/monit_stream/monit_stream.py
+++ b/tools/monit_stream/monit_stream.py
@@ -7,7 +7,9 @@ import time
 import sys
 import signal
 import os
-import telegraf
+
+from functools import partial
+from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
 
 TBPS = (1 * 1000 * 1000 * 1000 * 1000)
 GBPS = (1 * 1000 * 1000 * 1000)
@@ -49,6 +51,20 @@ TITLE_MAP = {'RxOnline': 'rx_on_line',
              'FTxBits': 'ftx_total_len'
              }
 
+TITLE_PROMETHEUS_MAP = {'RxOnline': 'rx_on_line_total',
+                        'RxPkts': 'rx_pkts_total',
+                        'RxDrops': 'rx_drops_total',
+                        'RxBits': 'rx_bits_total',
+                        'TxOnline': 'tx_on_line_total',
+                        'TxPkts': 'tx_pkts_total',
+                        'TxDrops': 'tx_drops_total',
+                        'TxBits': 'tx_total_len',
+                        'FTXOnline': 'ftx_on_line_total',
+                        'FTxPkts': 'ftx_pkts_total',
+                        'FTxDrops': 'ftx_missed_total',
+                        'FTxBits': 'ftx_bits_total'
+                        }
+
 
 def locate_vector_by_symbol(vector, symbol):
     return [s for s in vector if s['symbol'] == symbol]
@@ -251,10 +267,12 @@ def dump_apm_sendlog(json_fp, telegraf_client, appsym, user_interface, title_vec
         for id, value in enumerate(title_vector_rx + title_vector_tx):
             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])
+                lasttime_metrics_dict_per_dev.get(
+                    value, thistime_metrics_dict_per_dev[value])
 
         print(sendlog_dict_value)
-        telegraf_client.metric('mr4_stream_rxtx_value', sendlog_dict_value, tags=sendlog_tag)
+        telegraf_client.metric('mr4_stream_rxtx_value',
+                               sendlog_dict_value, tags=sendlog_tag)
 
         thistime_metrics_dict[dev] = thistime_metrics_dict_per_dev
 
@@ -287,11 +305,12 @@ def setup_argv_parser(applist):
     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',
+    # prometheus-client options
+    parser.add_argument('--prometheus-client', help='Run as prometheus client',
                         action='store_true', default=0)
-    parser.add_argument('--sendlog-apm-cfg', help='send log configure file',
-                        type=str, default='/opt/mrzcpd/etc/mrapm.conf')
+
+    parser.add_argument('--prometheus-client-listen-port', help='Default Port of prometheus client',
+                        type=int, default=8902)
 
     return parser.parse_args()
 
@@ -327,64 +346,77 @@ def check_vdev_options(json_fp, r_option):
             sys.exit(1)
 
 
-def sendlog_hostname(test_hostname, test_port):
-    import socket
+class PrometheusClient(BaseHTTPRequestHandler):
+    def transfer_valuelist_to_prometheus_fmt(self, title_vec, value_list, app, device):
+        output = ""
 
-    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()
+        for id, value in enumerate(value_list):
+            prometheus_metric_name = TITLE_PROMETHEUS_MAP[title_vec[id]]
+            output += "%s{app=\"%s\", device=\"%s\"} %u\n" % (
+                prometheus_metric_name, app, device, value)
 
-    return hostname, local_ip_addr
+        return output
 
+    def do_construct_response(self):
+        try:
+            applist = app_symbol_load()
+            if (len(applist) == 0):
+                return ""
 
-def sendlog_apm_init(r_option):
-    import ConfigParser
-    import urllib
+            output = ""
+            title_vec = TITLE_VECTOR_RX + TITLE_VECTOR_TX + TITLE_VECTOR_FTX
+            title_vec_rx = TITLE_VECTOR_RX
+            title_vec_tx = TITLE_VECTOR_TX + TITLE_VECTOR_FTX
 
-    config = ConfigParser.ConfigParser()
-    config.read(r_option.sendlog_apm_cfg)
+            for app in applist:
+                app_json_fp = app_json_load(app)
+                vdev_list = list_all_vdev(app_json_fp)
 
-    apm_server_url = config.get('sendlog_apm', 'apm_stream_server')
-    url_proto, rest = urllib.splittype(apm_server_url)
-    url_host, rest = urllib.splithost(rest)
-    url_host, url_port = urllib.splitport(url_host)
+                for dev in vdev_list:
+                    ValueListSum = dump_one_device(
+                        app_json_fp, dev, title_vec_rx, title_vec_tx, 0)
+                    output += self.transfer_valuelist_to_prometheus_fmt(
+                        title_vec, ValueListSum, app, dev)
 
-    hostname, local_ip_addr = sendlog_hostname(url_host, url_port)
-    sendlog_tags = {'host': hostname, 'local_ip_addr': local_ip_addr}
+            return output
 
-    if url_proto == 'udp':
-        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)
+        except KeyboardInterrupt as e:
+            sys.exit(0)
 
-    try:
-        send_interval = config.get('sendlog_apm', 'apm_send_interval')
-    except:
-        send_interval = 5
+        except Exception as e:
+            print(e)
+            return ""
 
-    return telegraf_client, send_interval
+    def do_GET(self):
+        if (self.path == '/metrics'):
+            resp = self.do_construct_response()
+            self.send_response(200)
+            self.send_header('Content-type', 'text/plain; version=0.0.4')
+            self.end_headers()
+            self.wfile.write(resp)
+        else:
+            self.send_error(404)
+            self.end_headers()
+
+
+def prometheus_client_init(prometheus_client_port):
+    HTTPServer(("", prometheus_client_port), PrometheusClient).serve_forever()
 
 
 def main():
     signal.signal(signal.SIGINT, sigint_handler)
-
     # Check Parameters
     try:
         applist = app_symbol_load()
+        r_option = setup_argv_parser(applist)
+
+        if r_option.prometheus_client:
+            return prometheus_client_init(r_option.prometheus_client_listen_port)
+
         if len(applist) == 0:
             print("monit_stream: error: no running application.")
             sys.exit(1)
 
-        r_option = setup_argv_parser(applist)
-
         for appsym in r_option.app:
             __json_fp = app_json_load(appsym)
 
@@ -402,16 +434,7 @@ def main():
     if 'ftx' in r_option.metrics:
         title_vector_tx.extend(TITLE_VECTOR_FTX)
 
-    try:
-        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)
+    sleep_interval = r_option.time
 
     try:
         lasttime_metrics_dict = {}
@@ -429,13 +452,6 @@ def main():
                     dump_status_table(json_fp, appsym)
                     continue
 
-                if r_option.sendlog_apm:
-                    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)