chore(celery): remove retrieve_resource task

Author: duanhongyi

charts/controller/templates/controller-webhook-register.yaml

@@ -36,6 +36,10 @@ webhooks:
     apiGroups: ["apps"]
     apiVersions: ["*"]
     resources: ["deployments/scale"]
+  - operations: ["UPDATE"]
+    apiGroups: ["servicecatalog.k8s.io"]
+    apiVersions: ["*"]
+    resources: ["serviceinstances/status", "servicebindings/status"]
   timeoutSeconds: 30
 ---
 apiVersion: v1

rootfs/api/admissions.py

@@ -81,3 +81,51 @@ def handle(self, request: Request) -> bool:
                     )
                 )
         return True
+
+
+class ServiceInstancesStatusHandler(BaseHandler):
+
+    def detect(self, request: Request) -> bool:
+        group = request.get("resource", {}).get("group", None)
+        resource = "/".join([
+            request.get("resource", {}).get("resource", None),
+            request.get("subResource", ""),
+        ])
+        if (group, resource) == ("servicecatalog.k8s.io", "serviceinstances/status"):
+            return True
+        return False
+
+    def handle(self, request: Request) -> bool:
+        app_id = request["object"]["metadata"]["namespace"]
+        name = request["object"]["metadata"]["name"]
+        status = request["object"]["status"]["lastConditionState"]
+        resource = models.resource.Resource.objects.filter(
+            app__id=app_id, name=name).first()
+        if resource and resource.status != status:
+            resource.status = status
+            resource.save(update_fields=["status"])
+        return True
+
+
+class ServicebindingsStatusHandler(BaseHandler):
+
+    def detect(self, request: Request) -> bool:
+        group = request.get("resource", {}).get("group", None)
+        resource = "/".join([
+            request.get("resource", {}).get("resource", None),
+            request.get("subResource", ""),
+        ])
+        if (group, resource) == ("servicecatalog.k8s.io", "servicebindings/status"):
+            return True
+        return False
+
+    def handle(self, request: Request) -> bool:
+        app_id = request["object"]["metadata"]["namespace"]
+        name = request["object"]["metadata"]["name"]
+        binding = request["object"]["status"]["lastConditionState"]
+        resource = models.resource.Resource.objects.filter(
+            app__id=app_id, name=name).first()
+        if resource and resource.binding != binding:
+            resource.binding = binding
+            resource.save(update_fields=["binding"])
+        return True

rootfs/api/management/commands/measure_loadbalancers.py

@@ -1,6 +1,7 @@
 import uuid
 import time
 import logging
+import ipaddress
 from django.utils import timezone
 from django.core.management.base import BaseCommand
 from django.conf import settings
@@ -14,21 +15,31 @@
 class Command(BaseCommand):
     """Management command for push data to manager"""
 
+    def _get_measure_name(self, ip):
+        address = ipaddress.ip_address(ip)
+        prefix = "intranet" if address.is_private else "internet"
+        return f"{prefix}:{address.version}"
+
     def _measure_loadbalancers(self, app_map, timestamp):
         stop = timestamp - (timestamp % 3600)
         start = stop - 3600
         loadbalancers = []
         for item in monitor.query_loadbalancer(app_map.keys(), start, stop):
-            name = item["ip"]
+            ip = item["ip"]
             namespace = item["namespace"]
             owner_id = app_map[namespace].owner_id
             loadbalancers.append({
                 "app_id":  str(app_map[namespace].uuid),
                 "owner": owner_id,
-                "name": name,
+                "name": self._get_measure_name(ip),
                 "type": "loadbalancer",
                 "unit": "number",
                 "usage": 1,
+                "kwargs": {
+                    "ip": ip,
+                    "service": item["service"],
+                    "hostname": item["hostname"],
+                },
                 "timestamp": start
             })
         send_measurements.delay(loadbalancers)

rootfs/api/management/commands/measure_networks.py

@@ -21,15 +21,18 @@ def _measure_networks(self, app_map, timestamp):
         for namespace, pod_name, rx_bytes, tx_bytes in monitor.query_network_flow(
                 app_map.keys(), start, stop):
             owner_id = app_map[namespace].owner_id
-            networks.append({
+            base_measure = {
                 "app_id":  str(app_map[namespace].uuid),
                 "owner": owner_id,
-                "name": pod_name,
                 "type": "network",
                 "unit": "bytes",
-                "usage": rx_bytes + tx_bytes,
+                "kwargs": {
+                    "pod": pod_name,
+                },
                 "timestamp": start
-            })
+            }
+            networks.append(base_measure | {"name": "rx", "usage": rx_bytes})
+            networks.append(base_measure | {"name": "tx", "usage": tx_bytes})
         send_measurements.delay(networks)
 
     def handle(self, *args, **options):

rootfs/api/models/app.py

@@ -22,7 +22,7 @@
 
 from api.utils import get_session
 from api.exceptions import AlreadyExists, DryccException, ServiceUnavailable
-from api.utils import generate_app_name, apply_tasks, unit_to_bytes, unit_to_millicpu
+from api.utils import generate_app_name, apply_tasks
 from scheduler import KubeHTTPException, KubeException
 from scheduler.resources.pod import DEFAULT_CONTAINER_PORT
 from .gateway import Gateway, Route, DEFAULT_HTTP_PORT, DEFAULT_HTTPS_PORT
@@ -589,23 +589,18 @@ def to_measurements(self, timestamp: float):
         measurements = []
         config = self.config_set.latest()
         for container_type, scale in self.structure.items():
+            plan = config.limits.get(container_type)
             measurements.append({
                 "app_id": str(self.uuid),
                 "owner": self.owner_id,
-                "name": container_type,
-                "type": "cpu",
-                "unit": "milli",
-                "usage": unit_to_millicpu(config.cpu.get(container_type)) * scale,
-                "timestamp": int(timestamp)
-            })
-            measurements.append({
-                "app_id": str(self.uuid),
-                "owner": self.owner_id,
-                "name": container_type,
-                "type": "memory",
-                "unit": "bytes",
-                "usage": unit_to_bytes(config.memory.get(container_type)) * scale,
-                "timestamp": int(timestamp)
+                "name": plan,
+                "type": "limits",
+                "unit": "number",
+                "usage": scale,
+                "kwargs": {
+                    "ptype": container_type,
+                },
+                "timestamp": int(timestamp),
             })
         return measurements
 

rootfs/api/models/resource.py

@@ -201,10 +201,13 @@ def to_measurements(self, timestamp: float):
         return [{
             "app_id": str(self.app_id),
             "owner": self.owner_id,
-            "name": self.name,
-            "type": self.plan,
+            "name": self.plan,
+            "type": "resource",
             "unit": "number",
             "usage": 1,
+            "kwargs": {
+                "name": self.name,
+            },
             "timestamp": int(timestamp)
         }]
 

rootfs/api/models/volume.py

@@ -79,10 +79,13 @@ def to_measurements(self, timestamp: float):
         return [{
             "app_id": str(self.app_id),
             "owner": self.owner_id,
-            "name": self.name,
+            "name": self.type,
             "type": "volume",
             "unit": "bytes",
-            "usage": unit_to_bytes(self.size) if self.type == "csi" else 0,
+            "usage": unit_to_bytes(self.size),
+            "kwargs": {
+                "name": self.name,
+            },
             "timestamp": int(timestamp)
         }]
 

rootfs/api/settings/celery.py

@@ -46,11 +46,6 @@ class Config(object):
             'exchange': 'controller.priority',
             'routing_key': 'controller.priority.high',
         },
-        'api.tasks.retrieve_resource': {
-            'queue': 'high',
-            'exchange': 'controller.priority',
-            'routing_key': 'controller.priority.high',
-        },
         'api.tasks.downstream_model_owner': {
             'queue': 'high',
             'exchange': 'controller.priority',

rootfs/api/signals.py

@@ -9,15 +9,13 @@
 import logging
 import urllib.parse
 import requests
-from datetime import timedelta
 from django.conf import settings
 from django.db.models.signals import post_delete, post_save
-from django.utils.timezone import now
 from django.dispatch import receiver
 from django.contrib.auth import get_user_model
 from rest_framework.authtoken.models import Token
 from api.utils import get_session
-from api.tasks import retrieve_resource, send_measurements
+from api.tasks import send_measurements
 from api.models.app import App
 from api.models.service import Service
 from api.models.gateway import Gateway, DEFAULT_HTTPS_PORT
@@ -252,13 +250,6 @@ def volume_changed_handle(sender, instance: Volume, created=False, update_fields
 @receiver(post_save, sender=Resource)
 def resource_changed_handle(
         sender, instance: Resource, created=False, update_fields=None, **kwargs):
-    # retrieve_resource
-    if created or instance.binding == "Binding" or (
-            update_fields is not None and "plan" in update_fields):
-        retrieve_resource.apply_async(
-            args=(instance, ),
-            eta=now() + timedelta(seconds=30)
-        )
     # measure resources to workflow manager
     if settings.WORKFLOW_MANAGER_URL and (
         created or (

rootfs/api/tasks.py

@@ -1,5 +1,4 @@
 # Create your tasks here
-import time
 import uuid
 import logging
 from typing import List, Dict
@@ -8,33 +7,9 @@
 
 from api import manager, models
 from api.exceptions import ServiceUnavailable
-from api.models.resource import Resource
 logger = logging.getLogger(__name__)
 
 
-@shared_task(
-    retry_kwargs={'max_retries': None}
-)
-def retrieve_resource(resource):
-    task_id = uuid.uuid4().hex
-    signals.request_started.send(sender=task_id)
-    try:
-        if not resource.retrieve():
-            t = time.time() - resource.created.timestamp()
-            if t < 3600:
-                retrieve_resource.apply_async(args=(resource, ), countdown=30)
-            elif t < 3600 * 12:
-                retrieve_resource.apply_async(args=(resource, ), countdown=1800)
-    except (Exception, Resource.DoesNotExist) as e:
-        signals.got_request_exception.send(sender=task_id)
-        if isinstance(e, Resource.DoesNotExist):
-            logger.exception("retrieve task not found resource: {}".format(resource.id))
-        else:
-            raise e
-    else:
-        signals.request_finished.send(sender=task_id)
-
-
 @shared_task(
     autoretry_for=(Exception, ),
     retry_backoff=8,