chore(gateway): change loadbalancer to gateway measure

Author: duanhongyi

charts/controller/templates/controller-cronjob-hourly.yaml

@@ -59,7 +59,7 @@ spec:
             args:
             - /bin/bash
             - -c
-            - python -u /workspace/manage.py measure_loadbalancers
+            - python -u /workspace/manage.py measure_gateway
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
-            {{- include "controller-job.envs" . | indent 12 }}
+            {{- include "controller-job.envs" . | indent 12 }}

rootfs/api/management/commands/measure_gateways.py

@@ -0,0 +1,30 @@
+import uuid
+import time
+import logging
+from django.utils import timezone
+from django.core.management.base import BaseCommand
+from django.conf import settings
+from api.models.gateway import Gateway
+from api.tasks import send_measurements
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Management command for push data to manager"""
+
+    def handle(self, *args, **options):
+        if settings.WORKFLOW_MANAGER_URL:
+            timestamp = time.time()
+            task_id = uuid.uuid4().hex
+            logger.info(f"pushing {task_id} gateways to workflow_manager when {timezone.now()}")
+            gateway_list = []
+            for gateway in Gateway.objects.all():
+                gateway_list.extend(gateway.to_measurements(timestamp))
+                if len(gateway_list) % 1000 == 0:
+                    send_measurements.delay(gateway_list)
+                    gateway_list = []
+            if len(gateway_list) > 0:
+                send_measurements.delay(gateway_list)
+            logger.info(f"pushed {task_id} gateways to workflow_manager when {timezone.now()}")
+            self.stdout.write("done")

rootfs/api/management/commands/measure_loadbalancers.py

@@ -142,6 +142,20 @@ def delete(self, *args, **kwargs):
             )
         return super().delete(*args, **kwargs)
 
+    def to_measurements(self, timestamp: float):
+        return [{
+            "app_id": str(self.app_id),
+            "owner": self.owner_id,
+            "name": settings.DRYCC_APP_GATEWAY_CLASS,
+            "type": "gateway",
+            "unit": "number",
+            "usage": 1,
+            "kwargs": {
+                "name": self.name,
+            },
+            "timestamp": int(timestamp)
+        }]
+
     def _check_port(self, port, protocol):
         for item in self.ports:
             if item["port"] == port:

rootfs/api/models/gateway.py

@@ -9,10 +9,6 @@
 last_over_time({__name__=~"%s",namespace="%s"}[%s])
 """
 
-query_loadbalancer_promql_tpl = """
-kube_service_status_load_balancer_ingress{namespace=~"%s"}
-"""
-
 query_network_receive_flow_promql_tpl = """
 increase(container_network_receive_bytes_total{namespace=~"%s"}[%s])
 """
@@ -74,13 +70,6 @@ async def last_metrics(namespace) -> AsyncGenerator[Iterator, str]:
         )
 
 
-async def query_loadbalancer(namespaces: Iterator[str], start: int, stop: int
-                             ) -> list[tuple[dict[str, str], int]]:
-    url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query")
-    promql = query_loadbalancer_promql_tpl % "|".join(namespaces)
-    return await query_prom(url, {"query": promql, "start": start, "end": stop})
-
-
 async def query_network_receive_flow(namespaces: Iterator[str], start: int, stop: int
                                      ) -> list[tuple[dict[str, str], int]]:
     url = urljoin(settings.DRYCC_VICTORIAMETRICS_URL, "/select/0/prometheus/api/v1/query")