update

celery_stats/es_status.py

@@ -29,6 +29,7 @@ data_dict = {}
 memory_used_rate_top = 0.01
 fs_free_rate_top = 0.01
 cpu_top = 0.1
+search_queue_max = 0
 
 secret = 'SECbbfd6b7403869cf8a31e63e2d623378bd8a55b5a31083fad6421ee817794f485'
 access_token = '6eb687358606347cef617237cddab6a80e2f5981b46fe04950a96152e387f35c'
@@ -45,7 +46,7 @@ def get_cluster_stats(ip):
     memory_used_rate = round(memory_used / memory_total,4)
     fs_available = res_json["nodes"]["fs"].get("available_in_bytes")
     fs_total = res_json["nodes"]["fs"].get("total_in_bytes")
-    fs_free_rate = round(fs_available / fs_total,4)
+    fs_free_rate = round(1 - fs_available / fs_total,4)
     return stats, timestamp, memory_used_rate, fs_free_rate
 
 
@@ -76,10 +77,11 @@ def get_node_status(ip):
         young_gc_ms = data["jvm"]["gc"]["collectors"]['young']['collection_time_in_millis']
         old_gc = data["jvm"]["gc"]["collectors"]['young']['collection_count']
         old_gc_ms = data["jvm"]["gc"]["collectors"]['young']['collection_time_in_millis']
+        search_thread_pool = data["thread_pool"]["search"]
         young_gc_per_ms = int(young_gc_ms/young_gc)
         old_gc_ms = int(old_gc_ms/old_gc)
         yield node_name, ip, query_time, query_current, fetch_current, cpu_percent, cpu_1m, cpu_5m, cpu_15m,\
-              young_gc_per_ms, old_gc, old_gc_ms
+              young_gc_per_ms, old_gc, old_gc_ms ,search_thread_pool
 
 
 def parse_cluster_stats(data):
@@ -95,9 +97,20 @@ def parse_cluster_stats(data):
 def parse_node_stats(data):
     res_str = ""
     node_name, ip, query_time, query_current, fetch_current, cpu_percent, cpu_1m, cpu_5m, cpu_15m, young_gc, \
-    old_gc, old_gc_ms = data
-    old_gc_last = data_dict.get('old_gc',old_gc)
-    old_gc_ms_last = data_dict.get('old_gc_ms',old_gc_ms)
+    old_gc, old_gc_ms, search_thread_pool = data
+    if not data_dict.get(node_name):
+        data_dict[node_name] = {}
+    old_gc_last = data_dict[node_name].get('old_gc',old_gc)
+    old_gc_ms_last = data_dict[node_name].get('old_gc_ms',old_gc_ms)
+
+    queue_last = data_dict[node_name].get('queue',0)
+    rejected_last = data_dict[node_name].get('queue',0)
+
+    if search_thread_pool['queue'] >= search_queue_max:
+        res_str += '{name}节点 search queue 目前为 %s \n' % str(search_thread_pool['queue'])
+    if search_thread_pool['rejected'] - rejected_last >= 0:
+        res_str += '{name}节点 search rejected 目前为 %s \n' % str(search_thread_pool['queue'])
+
     if old_gc - old_gc_last > 0:
         res_str += "{name}节点 old_gc增长中 old_gc为{old_gc} 平均{old_gc_ms}ms\n".format(name=node_name, old_gc=str(old_gc),
                                                                                old_gc_ms=str(old_gc_ms))
@@ -108,6 +121,10 @@ def parse_node_stats(data):
                                                                              cpu_5m=str(int(cpu_5m*100))+"%",
                                                                              cpu_15m= str(int(cpu_15m*100))+"%",
                    query_current=str(query_current),fetch_current=str(fetch_current),young_gc=str(young_gc))
+
+    data_dict[node_name]['old_gc'] = old_gc
+    data_dict[node_name]['old_gc_ms_last'] = old_gc_ms_last
+    data_dict[node_name].update(search_thread_pool)
     return res_str