chore(volumes): usage-based billing

Author: duanhongyi

charts/controller/templates/_helpers.tpl

@@ -416,6 +416,9 @@ oss:
 {{/* Generate controller config default volume claim template */}}
 {{ define "controller.config.defaultVolumeClaimTemplate" }}
 csi:
+  metadata:
+    annotations:
+      billing.drycc.cc/type: usage
   spec:
     accessModes:
     - ReadWriteMany
@@ -425,6 +428,9 @@ csi:
         storage: $size
     volumeMode: Filesystem
 nfs:
+  metadata:
+    annotations:
+      billing.drycc.cc/type: basic
   spec:
     accessModes:
     - ReadWriteMany
@@ -434,6 +440,9 @@ nfs:
         storage: $size
     volumeName: $volume_name
 oss:
+  metadata:
+    annotations:
+      billing.drycc.cc/type: basic
   spec:
     accessModes:
     - ReadWriteMany

rootfs/api/management/commands/upload_network_usage.py

@@ -58,7 +58,7 @@ def _upload_network_usage(self, start_time, app_map, timestamp):
     def handle(self, *args, **options):
         if settings.WORKFLOW_MANAGER_URL:
             start_time, timestamp, task_id = timezone.now(), int(time.time()), uuid.uuid4().hex
-            logger.info(f"pushing {task_id} limits to workflow_manager when {timezone.now()}")
+            logger.info(f"pushing {task_id} networks to workflow_manager when {timezone.now()}")
             app_map = {}
             for app in App.objects.all():
                 app_map[app.id] = app
@@ -67,5 +67,5 @@ def handle(self, *args, **options):
                     app_map = {}
             if len(app_map) > 0:
                 self._upload_network_usage(start_time, app_map, timestamp)
-            logger.info(f"pushed {task_id} limits to workflow_manager when {timezone.now()}")
+            logger.info(f"pushed {task_id} networks to workflow_manager when {timezone.now()}")
             self.stdout.write("done")

rootfs/api/management/commands/upload_volume_usage.py

@@ -3,10 +3,12 @@
 import logging
 import random
 from datetime import timedelta
+from asgiref.sync import async_to_sync
 from django.utils import timezone
 from django.core.management.base import BaseCommand
 from django.conf import settings
-from api.models.volume import Volume
+from api import monitor
+from api.models.app import App
 from api.tasks import send_usage
 
 logger = logging.getLogger(__name__)
@@ -15,23 +17,40 @@
 class Command(BaseCommand):
     """Management command for push data to manager"""
 
+    def _upload_volume_usage(self, start_time, app_map, timestamp):
+        stop = timestamp - (timestamp % 3600)
+        start = stop - 3600
+        volumes = []
+        for item in async_to_sync(monitor.query_volume_size)(app_map.keys(), start, stop):  # noqa
+            metric = item["metric"]
+            _, value = item["value"]
+            volumes.append({
+                "app_id":  str(app_map[metric['namespace']].uuid),
+                "owner": app_map[metric['namespace']].owner_id,
+                "type": "volume",
+                "unit": "bytes",
+                "name": metric["storageclass"],
+                "usage": value,
+                "kwargs": {
+                    "persistentvolumeclaim": metric['persistentvolumeclaim'],
+                },
+                "timestamp": start
+            })
+        send_usage.apply_async(
+            args=(volumes,), eta=start_time + timedelta(seconds=random.randint(1, 1800))
+        )
+
     def handle(self, *args, **options):
         if settings.WORKFLOW_MANAGER_URL:
-            start_time, timestamp, task_id = timezone.now(), time.time(), uuid.uuid4().hex
-            logger.info(f"pushing {task_id} volumes to workflow_manager when {start_time}")
-            volume_list = []
-            for volume in Volume.objects.all():
-                volume_list.extend(volume.to_usages(timestamp))
-                if len(volume_list) % 1000 == 0:
-                    send_usage.apply_async(
-                        args=(volume_list,),
-                        eta=start_time + timedelta(seconds=random.randint(1, 1800))
-                    )
-                    volume_list = []
-            if len(volume_list) > 0:
-                send_usage.apply_async(
-                    args=(volume_list,),
-                    eta=start_time + timedelta(seconds=random.randint(1, 1800))
-                )
+            start_time, timestamp, task_id = timezone.now(), int(time.time()), uuid.uuid4().hex
+            logger.info(f"pushing {task_id} volumes to workflow_manager when {timezone.now()}")
+            app_map = {}
+            for app in App.objects.all():
+                app_map[app.id] = app
+                if len(app_map) % 1000 == 0:
+                    self._upload_volume_usage(start_time, app_map, timestamp)
+                    app_map = {}
+            if len(app_map) > 0:
+                self._upload_volume_usage(start_time, app_map, timestamp)
             logger.info(f"pushed {task_id} volumes to workflow_manager when {timezone.now()}")
             self.stdout.write("done")

rootfs/api/monitor.py

@@ -28,6 +28,18 @@
 by (pod)
 """
 
+query_volume_size_promql_tpl = """
+max by(namespace, persistentvolumeclaim, storageclass) (
+  kube_persistentvolumeclaim_resource_requests_storage_bytes{namespace="%s"}
+  * on(namespace, persistentvolumeclaim) group_left(storageclass)
+  kube_persistentvolumeclaim_info{namespace="%s"}
+  * on(namespace, persistentvolumeclaim) group_left()
+  kube_persistentvolumeclaim_annotations{
+    namespace="%s",
+    annotation_billing_drycc_cc_type="usage"
+  }
+)
+"""
 
 query_network_receive_usage_promql_tpl = """
 sum (rate (container_network_receive_bytes_total{pod=~"^%s-.*$",namespace="%s"}[%s]))
@@ -70,6 +82,14 @@ async def last_metrics(namespace) -> AsyncGenerator[Iterator, str]:
         )
 
 
+async def query_volume_size(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_volume_size_promql_tpl % (
+        "|".join(namespaces), "|".join(namespaces), "|".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")

rootfs/scheduler/resources/pv.py

@@ -1,5 +1,6 @@
 import json
 import logging
+from api import utils
 from scheduler.resources import Resource
 from scheduler.exceptions import KubeHTTPException
 
@@ -19,7 +20,7 @@ def manifest(self, name, version=None, **kwargs):
                 'labels': labels
             },
         }
-        data.update(kwargs)
+        data = utils.dict_merge(data, kwargs)
         if version:
             data["metadata"]["resourceVersion"] = version
         return data

rootfs/scheduler/resources/pvc.py

@@ -1,5 +1,6 @@
 import json
 import logging
+from api import utils
 from scheduler.resources import Resource
 from scheduler.exceptions import KubeHTTPException
 
@@ -20,7 +21,7 @@ def manifest(self, namespace, name, version=None, **kwargs):
                 'labels': labels
             }
         }
-        data.update(kwargs)
+        data = utils.dict_merge(data, kwargs)
         if version:
             data["metadata"]["resourceVersion"] = version
         return data