es性能监控

celery_stats/es_status.py

+# -*- coding:UTF-8 -*-
+# @Time  : 2020/12/1 10:46
+# @File  : es_status.py
+# @email : litao@igengmei.com
+# @author : litao
+
+# from elasticsearch_7 import Elasticsearch
+
+# es = Elasticsearch([
+#     {
+#         'host': '172.16.31.6',
+#         'port': 9200,
+#     }, {
+#         'host': '172.16.31.16',
+#         'port': 9200,
+#     }, {
+#         'host': '172.16.31.10',
+#         'port': 9200,
+#     }
+# ])
+# para = "_stats"
+# es.get(index="gm-dbmw-associate_tag",)
+import time
+
+import requests
+import datetime
+from meta_base_code.send_msg_to_dingding.send_msg import send_msg_to_dingtalk
+data_dict = {}
+memory_used_rate_top = 0.1
+fs_free_rate_top = 0.1
+cpu_top = 0.1
+
+secret = 'SECbbfd6b7403869cf8a31e63e2d623378bd8a55b5a31083fad6421ee817794f485'
+access_token = '6eb687358606347cef617237cddab6a80e2f5981b46fe04950a96152e387f35c'
+
+
+def get_cluster_stats(ip):
+    node_requests_url = "http://{ip}:9200/_cluster/stats".format(ip=ip)
+    node_requests_res = requests.get(node_requests_url)
+    res_json = node_requests_res.json()
+    stats = res_json.get("status")
+    timestamp = int(datetime.datetime.now().timestamp())
+    memory_used = res_json["nodes"]["mem"].get("heap_used_in_bytes")
+    memory_total = res_json["nodes"]["mem"].get("heap_max_in_bytes")
+    memory_used_rate = memory_used / memory_total
+    fs_available = res_json["nodes"]["fs"].get("available_in_bytes")
+    fs_total = res_json["nodes"]["fs"].get("total_in_bytes")
+    fs_free_rate = fs_available / fs_total
+    return stats, timestamp, memory_used_rate, fs_free_rate
+
+
+def get_index_stats(ip,index):
+    index_requests_url = "http://{ip}:9200/{index}/stats".format(ip=ip,index=index)
+    node_requests_res = requests.get(index_requests_url)
+    res_json = node_requests_res.json()
+
+
+
+def get_node_status(ip):
+    node_requests_url = "http://{ip}:9200/_nodes/stats".format(ip=ip)
+    node_requests_res = requests.get(node_requests_url)
+    res_json = node_requests_res.json()
+
+    for data in res_json["nodes"]:
+        node_name = data['name']
+        ip = data['ip']
+        query_time = data["indices"]["search"]["query_time_in_millis"]
+        query_current = data["indices"]["search"]["query_current"]
+        fetch_current = data["indices"]["search"]["fetch_current"]
+        cpu_percent = data["os"]["cpu"]["percent"]
+        cpu_1m = data["os"]["cpu"]["load_average"]['1m']
+        cpu_5m = data["os"]["cpu"]["load_average"]['5m']
+        cpu_15m = data["os"]["cpu"]["load_average"]['15m']
+        yong_gc = data["jvm"]["gc"]["collectors"]['yong']['collection_count']
+        yield node_name, ip, query_time, query_current, fetch_current, cpu_percent, cpu_1m, cpu_5m, cpu_15m, yong_gc
+
+
+def parse_cluster_stats(data):
+    res_str = ""
+    stats, timestamp, memory_used_rate, fs_free_rate = data
+    if memory_used_rate >= memory_used_rate_top:
+        res_str += "集群内存使用率超过警戒值 目前为%s\n" % str(memory_used_rate * 100)
+    if fs_free_rate >= fs_free_rate_top:
+        res_str += "集群硬盘占用超过警戒值 目前为%s\n" % str(fs_free_rate * 100)
+    return res_str
+
+
+def parse_node_stats(data):
+    res_str = ""
+    node_name, ip, query_time, query_current, fetch_current, cpu_percent, cpu_1m, cpu_5m, cpu_15m, yong_gc = data
+    if cpu_1m >= cpu_top or cpu_5m >= cpu_top or cpu_15m >= cpu_top:
+        res_str += "{name}节点CPU使用率预警 1m {cpu_1m} 5m {cpu_5m} 15m {cpu_15m} \n" \
+                   "目前 query_current为{query_current} fetch_current为{fetch_current} yong_gc为{yong_gc}".format(cpu_1m=str(int(cpu_1m*100))+"%",name=node_name,
+                                                                             cpu_5m=str(int(cpu_5m*100))+"%",
+                                                                             cpu_15m= str(int(cpu_15m*100))+"%",
+                   query_current=str(query_current),fetch_current=str(fetch_current),yong_gc=str(yong_gc))
+    return res_str
+
+
+def main_task(ip):
+    while True:
+        data = get_cluster_stats(ip)
+        res = parse_cluster_stats(data)
+        if res:
+            send_msg_to_dingtalk(ip+res,access_token=access_token,secret=secret)
+        for data in get_node_status(ip):
+            parse_node_stats(data)
+            res = parse_node_stats(data)
+            if res:
+                send_msg_to_dingtalk(res, access_token=access_token, secret=secret)
+        time.sleep(10)
+
+
+if __name__ == "__main__":
+    main_task('172.16.31.6')

task/首页推荐策略日报.sql

@@ -474,7 +474,7 @@ CREATE TABLE recommend_strategy_d_fix(
 
    pid varchar(100),
    PRIMARY KEY ( pid )
-)ENGINE=InnoDB DEFAULT CHARSET=utf8;
+)ENGINE=InnoDB DEFAULT CHARSET=utf8_mb4;
 
 alert 
 ALTER TABLE recommend_strategy_d ADD recommend_ctr FLOAT;