fix(controller): measure_networks err

Author: duanhongyi

charts/controller/templates/controller-celery-deloyment.yaml

@@ -48,22 +48,13 @@ spec:
           - celery -A api worker -Q priority.high --autoscale=32,1 --loglevel=WARNING
         {{- include "controller.limits" . | indent 8 }}
         {{- include "controller.envs" . | indent 8 }}
-      - name: drycc-controller-celery-middle
-        image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/controller:{{.Values.imageTag}}
-        imagePullPolicy: {{.Values.imagePullPolicy}}
-        args:
-          - /bin/bash
-          - -c
-          - celery -A api worker -Q priority.middle --autoscale=16,1 --loglevel=WARNING
-        {{- include "controller.limits" . | indent 8 }}
-        {{- include "controller.envs" . | indent 8 }}
       - name: drycc-controller-celery-low
         image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/controller:{{.Values.imageTag}}
         imagePullPolicy: {{.Values.imagePullPolicy}}
         args:
           - /bin/bash
           - -c
-          - celery -A api worker -Q priority.low --autoscale=8,1 --loglevel=WARNING
+          - celery -A api worker -Q priority.low --autoscale=32,1 --loglevel=WARNING
         {{- include "controller.limits" . | indent 8 }}
         {{- include "controller.envs" . | indent 8 }}
 

rootfs/api/influxdb.py

@@ -1,5 +1,6 @@
 import threading
 import logging
+from contextlib import closing
 from typing import Iterator
 from django.conf import settings
 from influxdb_client import InfluxDBClient
@@ -11,28 +12,26 @@
 
 
 def _get_influxdb_client() -> InfluxDBClient:
-    if not hasattr(local, "influxdb_client"):
-        local.influxdb_client = InfluxDBClient(
-            url=settings.DRYCC_INFLUXDB_URL,
-            token=settings.DRYCC_INFLUXDB_TOKEN,
-            org=settings.DRYCC_INFLUXDB_ORG
-        )
-    return local.influxdb_client
+    return InfluxDBClient(
+        url=settings.DRYCC_INFLUXDB_URL,
+        token=settings.DRYCC_INFLUXDB_TOKEN,
+        org=settings.DRYCC_INFLUXDB_ORG
+    )
 
 
 def _query_stream(flux_script: str) -> Iterator[FluxRecord]:
     client = _get_influxdb_client()
     try:
         query_api = client.query_api()
         records = query_api.query_stream(flux_script)
+        with closing(records):
+            yield from records
     except ApiException as e:
         logger.exception(e)
         yield from []
     except Exception as e:
         logger.exception(e)
         yield from []
-    else:
-        yield from records
 
 
 def query_container_count(

rootfs/api/management/commands/measure_networks.py

@@ -5,7 +5,7 @@
 from django.core.management.base import BaseCommand
 from django.conf import settings
 from api import influxdb
-from api.models.config import Config
+from api.models.app import App
 from api.tasks import send_measurements
 
 logger = logging.getLogger(__name__)
@@ -14,16 +14,16 @@
 class Command(BaseCommand):
     """Management command for push data to manager"""
 
-    def _measure_networks(self, config_map, timestamp):
+    def _measure_networks(self, app_map, timestamp):
         stop = timestamp - (timestamp % 3600)
         start = stop - 3600
         networks = []
-        for record in influxdb.query_network_flow(config_map.keys(), start, stop):
+        for record in influxdb.query_network_flow(app_map.keys(), start, stop):
             app_id = record["namespace"]
-            owner_id = config_map[app_id].owner_id
+            owner_id = app_map[app_id].owner_id
             networks.append({
                 "app_id":  app_id,
-                "owner_id": owner_id,
+                "owner": owner_id,
                 "name": record["pod_name"],
                 "type": "network",
                 "unit": "bytes",
@@ -37,13 +37,13 @@ def handle(self, *args, **options):
             timestamp = int(time.time())
             task_id = uuid.uuid4().hex
             logger.info(f"pushing {task_id} limits to workflow_manager when {timezone.now()}")
-            config_map = {}
-            for config in Config.objects.all():
-                config_map[config.app_id] = config
-                if len(config_map) % 1000 == 0:
-                    send_measurements.delay(self._measure_networks(config_map, timestamp))
-                    config_map = {}
-            if len(config_map) > 0:
-                send_measurements.delay(self._measure_networks(config_map, timestamp))
+            app_map = {}
+            for app in App.objects.all():
+                app_map[app.id] = app
+                if len(app_map) % 1000 == 0:
+                    self._measure_networks(app_map, timestamp)
+                    app_map = {}
+            if len(app_map) > 0:
+                self._measure_networks(app_map, timestamp)
             logger.info(f"pushed {task_id} limits to workflow_manager when {timezone.now()}")
             self.stdout.write("done")

rootfs/api/manager.py

@@ -68,7 +68,7 @@ def post(self, measurements: List[Dict[str, str]]):
         [
             {
                 "app_id":  "test",
-                "owner_id": "test",
+                "owner": "test",
                 "name": web,
                 "type": "CPU",
                 "unit": "G"
@@ -78,4 +78,4 @@ def post(self, measurements: List[Dict[str, str]]):
         ]
         """
         url = "%s/measurements/" % settings.WORKFLOW_MANAGER_URL
-        return self.post(url=url, json=measurements)
+        return super().post(url=url, json=measurements)

rootfs/api/models/app.py

@@ -1163,7 +1163,7 @@ def to_measurements(self, timestamp: float):
         for container_type, scale in self.structure.items():
             measurements.append({
                 "app_id": self.id,
-                "user_id": self.owner_id,
+                "owner": self.owner_id,
                 "name": container_type,
                 "type": "cpu",
                 "unit": "milli",
@@ -1172,7 +1172,7 @@ def to_measurements(self, timestamp: float):
             })
             measurements.append({
                 "app_id": self.app_id,
-                "user_id": self.owner_id,
+                "owner": self.owner_id,
                 "name": container_type,
                 "type": "memory",
                 "unit": "bytes",

rootfs/api/models/resource.py

@@ -231,7 +231,7 @@ def detach_resource(self, *args, **kwargs):
     def to_measurements(self, timestamp: float):
         return [{
             "app_id": str(self.app_id),
-            "user_id": str(self.owner_id),
+            "owner": str(self.owner_id),
             "name": self.name,
             "type": self.plan,
             "unit": "number",

rootfs/api/models/volume.py

@@ -66,7 +66,7 @@ def log(self, message, level=logging.INFO):
     def to_measurements(self, timestamp: float):
         return [{
             "app_id": str(self.app_id),
-            "user_id": str(self.owner_id),
+            "owner": str(self.owner_id),
             "name": self.name,
             "type": "volume",
             "unit": "bytes",