feat: 系统任务执行状态自监控能力支持 TencentBlueKing#7457

gcloud/contrib/monitor/views.py

@@ -11,6 +11,7 @@
 specific language governing permissions and limitations under the License.
 """
 
+import pyrabbit2
 from django.db.models import Q
 from django.http import JsonResponse
 from django.utils import timezone
@@ -20,7 +21,7 @@
 from gcloud.iam_auth.intercept import iam_intercept
 from gcloud.iam_auth.view_interceptors.statistics import StatisticsViewInpterceptor
 from gcloud.taskflow3.models import TaskFlowInstance
-import pyrabbit2
+
 
 @require_GET
 @iam_intercept(StatisticsViewInpterceptor())
@@ -152,6 +153,7 @@ def get_schedule_times(request):
     ]
     return JsonResponse({"result": True, "data": schedule_times})
 
+
 @require_GET
 @iam_intercept(StatisticsViewInpterceptor())
 def get_mq_overview(request):
@@ -162,16 +164,17 @@ def get_mq_overview(request):
     cl = pyrabbit2.Client("localhost:15672", "guest", "guest")
     overview = cl.get_overview()
     data = {
-        "totals":{
+        "totals": {
             "ready": overview["queue_totals"]["messages_ready"],
             "unacked": overview["queue_totals"]["messages_unacknowledged"],
-            "total": overview["queue_totals"]["messages"]
+            "total": overview["queue_totals"]["messages"],
         },
         "global_totals": overview["object_totals"],
-        "nodes": cl.get_nodes()
+        "nodes": cl.get_nodes(),
     }
     return JsonResponse({"result": True, "data": data})
 
+
 @require_GET
 @iam_intercept(StatisticsViewInpterceptor())
 def get_mq_data(request):
@@ -180,5 +183,16 @@ def get_mq_data(request):
     """
     cl = pyrabbit2.Client("localhost:15672", "guest", "guest")
     data = {
-        vhost: [{"vhost": vhost, "queue_name": queue["name"], "message_count": queue["messages"], "queue_state": queue["state"], "messages": cl.get_messages(vhost, queue["name"], count=queue["messages"], requeue=True)} for queue in cl.get_queues(vhost=vhost)] for vhost in cl.get_vhost_names()}
+        vhost: [
+            {
+                "vhost": vhost,
+                "queue_name": queue["name"],
+                "message_count": queue["messages"],
+                "queue_state": queue["state"],
+                "messages": cl.get_messages(vhost, queue["name"], count=queue["messages"], requeue=True),
+            }
+            for queue in cl.get_queues(vhost=vhost)
+        ]
+        for vhost in cl.get_vhost_names()
+    }
     return JsonResponse({"result": True, "data": data})

requirements.txt

@@ -38,6 +38,7 @@ django-dbconn-retry==0.1.5
 pydantic==1.9.1
 pyrabbit2==1.0.7
 
+
 # monitor
 django-prometheus==2.1.0
 celery-prometheus-exporter==1.7.0