feat(controller): new measurement model

Author: duanhongyi

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

@@ -39,4 +39,31 @@ spec:
             args:
               - python /app/manage.py load_db_state_to_k8s
             {{- include "controller.envs" . | indent 12 }}
+          - image: {{.Values.image_registry}}/{{.Values.image_org}}/controller:{{.Values.image_tag}}
+            imagePullPolicy: {{.Values.pull_policy}}
+            name: drycc-controller-measure-apps
+            command:
+              - /bin/bash
+              - -c
+            args:
+              - python -u /app/manage.py measure_apps
+            {{- include "controller.envs" . | indent 12 }}
+          - image: {{.Values.image_registry}}/{{.Values.image_org}}/controller:{{.Values.image_tag}}
+            imagePullPolicy: {{.Values.pull_policy}}
+            name: drycc-controller-measure-resources
+            command:
+              - /bin/bash
+              - -c
+            args:
+              - python -u /app/manage.py measure_resources
+            {{- include "controller.envs" . | indent 12 }}
+          - image: {{.Values.image_registry}}/{{.Values.image_org}}/controller:{{.Values.image_tag}}
+            imagePullPolicy: {{.Values.pull_policy}}
+            name: drycc-controller-measure-volumes
+            command:
+              - /bin/bash
+              - -c
+            args:
+              - python -u /app/manage.py measure_volumes
+            {{- include "controller.envs" . | indent 12 }}
 

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

@@ -32,28 +32,10 @@ spec:
           containers:
           - image: {{.Values.image_registry}}/{{.Values.image_org}}/controller:{{.Values.image_tag}}
             imagePullPolicy: {{.Values.pull_policy}}
-            name: drycc-controller-measure-instance
+            name: drycc-controller-measure-networks
             command:
               - /bin/bash
               - -c
             args:
-              - python -u /app/manage.py measure_instance
-            {{- include "controller.envs" . | indent 12 }}
-          - image: {{.Values.image_registry}}/{{.Values.image_org}}/controller:{{.Values.image_tag}}
-            imagePullPolicy: {{.Values.pull_policy}}
-            name: drycc-controller-measure-resources
-            command:
-              - /bin/bash
-              - -c
-            args:
-              - python -u /app/manage.py measure_resources
-            {{- include "controller.envs" . | indent 12 }}
-          - image: {{.Values.image_registry}}/{{.Values.image_org}}/controller:{{.Values.image_tag}}
-            imagePullPolicy: {{.Values.pull_policy}}
-            name: drycc-controller-measure-volumes
-            command:
-              - /bin/bash
-              - -c
-            args:
-              - python -u /app/manage.py measure_volumes
+              - python -u /app/manage.py measure_networks
             {{- include "controller.envs" . | indent 12 }}

rootfs/api/management/commands/measure_apps.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 import App
+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 is not None:
+            timestamp = time.time()
+            task_id = uuid.uuid4().hex
+            logger.info(f"pushing {task_id} resources to workflow_manager when {timezone.now()}")
+            app_list = []
+            for app in App.objects.all():
+                app_list.extend(app.to_measurements(timestamp))
+                if len(app_list) % 1000 == 0:
+                    send_measurements.delay(app_list)
+                    app_list = []
+            if len(app_list) > 0:
+                send_measurements.delay(app_list)
+            logger.info(f"pushed {task_id} resources to workflow_manager when {timezone.now()}")
+            self.stdout.write("done")

rootfs/api/management/commands/measure_instance.py

@@ -0,0 +1,49 @@
+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 import influxdb
+from api.models import Config
+from api.tasks import send_measurements
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Management command for push data to manager"""
+
+    def _measure_networks(self, config_map, timestamp):
+        stop = timestamp - (timestamp % 3600)
+        start = stop - 3600
+        networks = []
+        for record in influxdb.query_network_flow(config_map.keys(), start, stop):
+            app_id = record["namespace"]
+            owner_id = config_map[app_id].owner_id
+            networks.append({
+                "app_id":  app_id,
+                "owner_id": owner_id,
+                "name": record["pod_name"],
+                "type": "NETWORK",
+                "unit": "BYTES",
+                "usage": record["rx_bytes"] + record["tx_bytes"],
+                "timestamp": "%d" % start
+            })
+        send_measurements.delay(networks)
+
+    def handle(self, *args, **options):
+        if settings.WORKFLOW_MANAGER_URL is not None:
+            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_d] = 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))
+            logger.info(f"pushed {task_id} limits to workflow_manager when {timezone.now()}")
+            self.stdout.write("done")

rootfs/api/management/commands/measure_networks.py

@@ -259,6 +259,19 @@ def create_auth_token_handle(sender, instance=None, created=False, **kwargs):
         Token.objects.create(user=instance)
 
 
+@receiver(post_save, sender=App)
+def app_changed_handle(sender, instance=None, created=False, update_fields=None, **kwargs):
+    # measure limits to workflow manager
+    if settings.WORKFLOW_MANAGER_URL is not None and (
+        created or (
+            update_fields is not None and "structure" in update_fields)):
+        timestamp = time.time()
+        send_measurements.apply_async(
+            args=[instance.to_measurements(timestamp), ],
+            queue="priority.middle",
+        )
+
+
 @receiver(post_save, sender=Config)
 def config_changed_handle(sender, instance=None, created=False, update_fields=None, **kwargs):
     # measure limits to workflow manager
@@ -268,7 +281,7 @@ def config_changed_handle(sender, instance=None, created=False, update_fields=No
                 "cpu" in update_fields or "memory" in update_fields))):
         timestamp = time.time()
         send_measurements.apply_async(
-            args=[instance.to_measurements(timestamp), ],
+            args=[instance.app.to_measurements(timestamp), ],
             queue="priority.middle",
         )
 

rootfs/api/models/__init__.py

@@ -27,7 +27,7 @@
 from api.models.tls import TLS
 from api.models.appsettings import AppSettings
 from api.models.volume import Volume
-from api.utils import generate_app_name, apply_tasks
+from api.utils import generate_app_name, apply_tasks, unit_to_bytes, unit_to_millicpu
 from scheduler import KubeHTTPException, KubeException
 
 logger = logging.getLogger(__name__)
@@ -1279,3 +1279,27 @@ def set_application_config(self, release):
             self._scheduler.secret.create(self.id, secret_name, secrets_env, labels=labels)
         else:
             self._scheduler.secret.update(self.id, secret_name, secrets_env, labels=labels)
+
+    def to_measurements(self, timestamp: float):
+        measurements = []
+        config = self.config_set.latest()
+        for container_type, scale in self.structure.items():
+            measurements.append({
+                "app_id": self.id,
+                "user_id": self.owner_id,
+                "name": container_type,
+                "type": "CPU",
+                "unit": "MILLI",
+                "usage": unit_to_millicpu(config.cpu.get(container_type)) * scale,
+                "timestamp": "%f" % timestamp
+            })
+            measurements.append({
+                "app_id": self.app_id,
+                "user_id": self.owner_id,
+                "name": container_type,
+                "type": "MEMORY",
+                "unit": "BYTES",
+                "usage": unit_to_bytes(config.memory.get(container_type)) * scale,
+                "timestamp": "%f" % timestamp
+            })
+        return measurements

rootfs/api/models/app.py

@@ -2,10 +2,8 @@
 import logging
 from django.conf import settings
 from django.db import models
-from api import influxdb
 from api.models.release import Release
 from api.models import UuidAuditedModel
-from api.utils import unit_to_bytes, unit_to_millicpu
 from api.exceptions import DryccException, UnprocessableEntity
 
 
@@ -194,36 +192,3 @@ def save(self, **kwargs):
             self.set_tags({'tags': {}})
 
         return super(Config, self).save(**kwargs)
-
-    def to_measurements(self, timestamp: float):
-        assert len(set(self.memory.keys()).difference(self.cpu.keys())) == 0
-        stop = int(timestamp)
-        start = stop - (stop % 3600)
-        records = {}
-        app_id, owner_id = str(self.app_id), str(self.owner_id)
-        for record in influxdb.query_container_count([app_id, ], start, stop):
-            container_type = record["container_name"].replace(f"{app_id}-", "", 1)
-            if container_type not in records:
-                records[container_type] = []
-            records[container_type].append(record)
-        cpu_measurements = [{
-            "app_id": app_id,
-            "user_id": owner_id,
-            "name": container_type,
-            "type": "CPU",
-            "unit": "MILLI",
-            "usage": unit_to_millicpu(
-                self.cpu.get(container_type)) * len(records.get(container_type, [])),
-            "timestamp": "%f" % timestamp
-        } for container_type in self.cpu.keys()]
-        memory_measurements = [{
-            "app_id": app_id,
-            "user_id": owner_id,
-            "name": container_type,
-            "type": "MEMORY",
-            "unit": "BYTES",
-            "usage": unit_to_bytes(
-                self.memory.get(container_type)) * len(records.get(container_type, [])),
-            "timestamp": "%f" % timestamp
-        } for container_type in self.memory.keys()]
-        return cpu_measurements + memory_measurements

rootfs/api/models/config.py

@@ -411,7 +411,7 @@ def call_command(self, *args, **kwargs):
         from django.core.management import call_command
         out = StringIO()
         call_command(
-            "measure_instance",
+            "measure_apps",
             *args,
             stdout=out,
             stderr=StringIO(),
@@ -422,8 +422,8 @@ def call_command(self, *args, **kwargs):
     def test_measure_config(self, *args, **kwargs):
         # create
         app_id = self.create_app()
-        url = "/v2/apps/{app_id}/config".format(**locals())
+        url = f"/v2/apps/{app_id}/config"
         body = {'values': json.dumps({'PORT': 5000}), 'cpu': json.dumps({'web': '1000m'})}
-        response = self.client.post(url, body)
+        self.client.post(url, body)
         out = self.call_command()
         self.assertIn(out, "done\n")