增加队列监控

celery_stats/__init__.py

+# -*- coding:UTF-8 -*-
+# @Time  : 2020/12/2 14:57
+# @File  : __init__.py.py
+# @email : litao@igengmei.com
+# @author : litao
\ No newline at end of file

celery_stats/celery_stats_monitor.py

+# -*- coding:UTF-8 -*-
+# @Time  : 2020/12/2 15:12
+# @File  : celery_stats_monitor.py
+# @email : litao@igengmei.com
+# @author : litao
+
+import redis
+from send_msg_to_dingding.send_msg import send_msg_to_dingtalk
+import datetime, time
+secret = "SECba5212dadad3794b3da51c903c828f60ab8342897af2675f1f48fceb8858eb5c"
+access_token = "df546521ce46bfb35025ca266efc2d7e8d708d1c8ada9b15ae487786ad06ad12"
+
+# send_msg_to_dingtalk("123",secret=secret,access_token=access_token)
+
+
+
+redis_old_gaia = redis.StrictRedis.from_url("redis://:ReDis!GmTx*0aN11@172.16.40.166:6379/9", decode_responses=True)
+redis_old_mimas = redis.StrictRedis.from_url("redis://:ReDis!GmTx*0aN3@172.16.40.145:6379/5", decode_responses=True)
+
+
+class Parse_data:
+    def __init__(self,top=2000,consume=500):
+        self.offset_dict = {}
+
+    def parse(self,key_name,len_data,per_sec=10):
+        if self.offset_dict.get(key_name):
+            old_len_data = self.offset_dict.get(key_name)
+            len_data_diff = old_len_data - len_data
+            rate = len_data_diff / per_sec
+            if len_data_diff > 0:
+                use_time = round(len_data/rate/60/60,1)
+                return "目前队列中有{}条数据,预计{}小时后处理完成".format(str(len_data),str(use_time))
+            else:
+                if abs(rate) > 0:
+                    return "队列快速增长中"
+
+
+        self.offset_dict[key_name] = len_data
+        return None
+
+
+def len_list_and_send_msg(redis_clint,key_name,rules,per_sec=10):
+    # redis_clint = redis.StrictRedis.from_url("redis://:ReDis!GmTx*0aN3@172.16.40.145:6379/5", decode_responses=True)
+    res = redis_clint.llen(key_name)
+    print(res)
+    str_res = rules.parse(key_name,res,per_sec=per_sec)
+    if str_res:
+        send_msg_to_dingtalk(str_res,secret=secret,access_token=access_token)
+
+
+if __name__ == "__main__":
+    rules = Parse_data()
+    redis_clicnt_dct = {
+        "gaia-dbmw": redis_old_gaia,
+        "mimas-dbmw":redis_old_mimas
+    }
+    per_sec = 10
+    while True:
+        now = datetime.datetime.now()
+        if now.second % per_sec == 0:
+            for redis_clint_key in redis_clicnt_dct:
+                len_list_and_send_msg(redis_clicnt_dct[redis_clint_key],redis_clint_key,rules,per_sec)

send_msg_to_dingding/__init__.py

+# -*- coding:UTF-8 -*-
+# @Time  : 2020/12/2 14:58
+# @File  : __init__.py.py
+# @email : litao@igengmei.com
+# @author : litao
\ No newline at end of file

send_msg_to_dingding/send_msg.py

+# -*- coding:UTF-8 -*-
+# @Time  : 2020/12/2 14:58
+# @File  : send_msg.py
+# @email : litao@igengmei.com
+# @author : litao
+import time
+import hmac
+import hashlib
+import urllib
+import base64
+import json
+import requests
+
+
+def send_msg_to_dingtalk(msg, mobiles=[],access_token="",secret=""):
+    try:
+        # secret = "SECba5212dadad3794b3da51c903c828f60ab8342897af2675f1f48fceb8858eb5c"
+        timestamp = str(round(time.time() * 1000))
+        secret_enc = secret.encode("utf-8")
+        string_to_sign = "{}\n{}".format(timestamp, secret)
+        string_to_sign_enc = string_to_sign.encode("utf-8")
+        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+
+        headers = {"Content-Type": "application/json"}
+        data = {"msgtype": "text", "text": {"content": msg}}
+        if mobiles:
+            data = {"msgtype": "text", "text": {"content": msg}, "at": {"atMobiles": mobiles, "isAtAll": False}}
+        json_data = json.dumps(data)
+        # "https://oapi.dingtalk.com/robot/send?access_token=df546521ce46bfb35025ca266efc2d7e8d708d1c8ada9b15ae487786ad06ad12&sign={}&timestamp={}"
+        url = "https://oapi.dingtalk.com/robot/send?access_token={}&sign={}&timestamp={}".format(access_token,
+            sign, timestamp)
+        response = requests.post(url=url, data=json_data, headers=headers)
+        return str(response.status_code) + " " + str(response.content.decode("utf-8"))
+    except Exception as e:
+        print(e)
+        return str(e)
\ No newline at end of file