chore(usage): add identifier field

Author: duanhongyi

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

@@ -47,6 +47,10 @@ spec:
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
             {{- include "controller-job.envs" . | indent 12 }}
+            volumeMounts:
+            - name: controller-config
+              readOnly: false
+              mountPath: /etc/controller
           - image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/controller:{{.Values.imageTag}}
             imagePullPolicy: {{.Values.imagePullPolicy}}
             name: drycc-controller-load-db-state-to-k8s
@@ -61,3 +65,11 @@ spec:
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
             {{- include "controller-job.envs" . | indent 12 }}
+            volumeMounts:
+            - name: controller-config
+              readOnly: false
+              mountPath: /etc/controller
+          volumes:
+          - name: controller-config
+            configMap:
+              name: controller-config

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

@@ -49,6 +49,10 @@ spec:
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
             {{- include "controller-job.envs" . | indent 12 }}
+            volumeMounts:
+            - name: controller-config
+              readOnly: false
+              mountPath: /etc/controller
           - image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/controller:{{.Values.imageTag}}
             imagePullPolicy: {{.Values.pull_policy}}
             name: drycc-controller-upload-gateway-usage
@@ -63,6 +67,10 @@ spec:
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
             {{- include "controller-job.envs" . | indent 12 }}
+            volumeMounts:
+            - name: controller-config
+              readOnly: false
+              mountPath: /etc/controller
           - image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/controller:{{.Values.imageTag}}
             imagePullPolicy: {{.Values.pull_policy}}
             name: drycc-controller-upload-app-usage
@@ -77,6 +85,10 @@ spec:
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
             {{- include "controller-job.envs" . | indent 12 }}
+            volumeMounts:
+            - name: controller-config
+              readOnly: false
+              mountPath: /etc/controller
           - image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/controller:{{.Values.imageTag}}
             imagePullPolicy: {{.Values.pull_policy}}
             name: drycc-controller-upload-resources-usage
@@ -91,6 +103,10 @@ spec:
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
             {{- include "controller-job.envs" . | indent 12 }}
+            volumeMounts:
+            - name: controller-config
+              readOnly: false
+              mountPath: /etc/controller
           - image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/controller:{{.Values.imageTag}}
             imagePullPolicy: {{.Values.pull_policy}}
             name: drycc-controller-upload-volume-usage
@@ -105,3 +121,11 @@ spec:
             {{- end }}
             {{- include "controller.envs" . | indent 12 }}
             {{- include "controller-job.envs" . | indent 12 }}
+            volumeMounts:
+            - name: controller-config
+              readOnly: false
+              mountPath: /etc/controller
+          volumes:
+          - name: controller-config
+            configMap:
+              name: controller-config

rootfs/api/management/commands/upload_app_usage.py

@@ -1,5 +1,4 @@
 import uuid
-import time
 import logging
 import random
 from datetime import timedelta
@@ -17,21 +16,24 @@ class Command(BaseCommand):
 
     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} resources to workflow_manager when {timezone.now()}")
+            now = timezone.now()
+            start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
+            logger.info(f"pushing {task_id} resources to workflow_manager when {now}")
             app_list = []
             for app in App.objects.all():
                 app_list.extend(app.to_usages(timestamp))
                 if len(app_list) % 1000 == 0:
+                    logger.info(f"bulk sent: {len(app_list)}")
                     send_usage.apply_async(
                         args=(app_list,),
-                        eta=start_time + timedelta(seconds=random.randint(1, 1800))
+                        eta=start_time + timedelta(seconds=random.randint(1, 3600))
                     )
                     app_list = []
             if len(app_list) > 0:
+                logger.info(f"bulk sent: {len(app_list)}")
                 send_usage.apply_async(
                     args=(app_list,),
-                    eta=start_time + timedelta(seconds=random.randint(1, 1800))
+                    eta=start_time + timedelta(seconds=random.randint(1, 3600))
                 )
             logger.info(f"pushed {task_id} resources to workflow_manager when {timezone.now()}")
             self.stdout.write("done")

rootfs/api/management/commands/upload_gateway_usage.py

@@ -1,5 +1,4 @@
 import uuid
-import time
 import logging
 import random
 from datetime import timedelta
@@ -17,21 +16,24 @@ class Command(BaseCommand):
 
     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} gateways to workflow_manager when {timezone.now()}")
+            now = timezone.now()
+            start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
+            logger.info(f"pushing {task_id} gateways to workflow_manager when {now}")
             gateway_list = []
             for gateway in Gateway.objects.all():
                 gateway_list.extend(gateway.to_usages(timestamp))
                 if len(gateway_list) % 1000 == 0:
+                    logger.info(f"bulk sent: {len(gateway_list)}")
                     send_usage.apply_async(
                         args=(gateway_list,),
-                        eta=start_time + timedelta(seconds=random.randint(1, 1800))
+                        eta=start_time + timedelta(seconds=random.randint(1, 3600))
                     )
                     gateway_list = []
             if len(gateway_list) > 0:
+                logger.info(f"bulk sent: {len(gateway_list)}")
                 send_usage.apply_async(
                     args=(gateway_list,),
-                    eta=start_time + timedelta(seconds=random.randint(1, 1800))
+                    eta=start_time + timedelta(seconds=random.randint(1, 3600))
                 )
             logger.info(f"pushed {task_id} gateways to workflow_manager when {timezone.now()}")
             self.stdout.write("done")

rootfs/api/management/commands/upload_network_usage.py

@@ -1,5 +1,5 @@
 import uuid
-import time
+import hashlib
 import logging
 import random
 from datetime import timedelta
@@ -24,6 +24,7 @@ def _upload_network_usage(self, start_time, app_map, timestamp):
         for item in async_to_sync(monitor.query_network_usage)(app_map.keys(), start, stop):
             metric = item["metric"]
             _, value = item["value"]
+            pod_name = metric['pod']
             networks.append({
                 "app_id":  str(app_map[metric['namespace']].uuid),
                 "owner": app_map[metric['namespace']].owner_id,
@@ -32,18 +33,21 @@ def _upload_network_usage(self, start_time, app_map, timestamp):
                 "name": metric['direction'],
                 "usage": value,
                 "kwargs": {
-                    "pod": metric['pod'],
+                    "pod": pod_name,
                 },
-                "timestamp": start
+                "timestamp": start,
+                "identifier": hashlib.md5(pod_name.encode("utf-8")).hexdigest(),
             })
+        logger.info(f"bulk sent: {len(networks)}")
         send_usage.apply_async(
-            args=(networks,), eta=start_time + timedelta(seconds=random.randint(1, 1800))
+            args=(networks,), eta=start_time + timedelta(seconds=random.randint(1, 3600))
         )
 
     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} networks to workflow_manager when {timezone.now()}")
+            now = timezone.now()
+            start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
+            logger.info(f"pushing {task_id} networks to workflow_manager when {now}")
             app_map = {}
             for app in App.objects.all():
                 app_map[app.id] = app

rootfs/api/management/commands/upload_resource_usage.py

@@ -1,5 +1,4 @@
 import uuid
-import time
 import logging
 import random
 from datetime import timedelta
@@ -17,21 +16,24 @@ class Command(BaseCommand):
 
     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} resources to workflow_manager when {timezone.now()}")
+            now = timezone.now()
+            start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
+            logger.info(f"pushing {task_id} resources to workflow_manager when {now}")
             resource_list = []
             for resource in Resource.objects.filter(status="Ready"):
                 resource_list.extend(resource.to_usages(timestamp))
                 if len(resource_list) % 1000 == 0:
+                    logger.info(f"bulk sent: {len(resource_list)}")
                     send_usage.apply_async(
                         args=(resource_list,),
-                        eta=start_time + timedelta(seconds=random.randint(1, 1800))
+                        eta=start_time + timedelta(seconds=random.randint(1, 3600))
                     )
                     resource_list = []
             if len(resource_list) > 0:
+                logger.info(f"bulk sent: {len(resource_list)}")
                 send_usage.apply_async(
                     args=(resource_list,),
-                    eta=start_time + timedelta(seconds=random.randint(1, 1800))
+                    eta=start_time + timedelta(seconds=random.randint(1, 3600))
                 )
             logger.info(f"pushed {task_id} resources to workflow_manager when {timezone.now()}")
             self.stdout.write("done")

rootfs/api/management/commands/upload_volume_usage.py

@@ -1,5 +1,5 @@
 import uuid
-import time
+import hashlib
 import logging
 import random
 from datetime import timedelta
@@ -24,6 +24,7 @@ def _upload_volume_usage(self, start_time, app_map, timestamp):
         for item in async_to_sync(monitor.query_volume_usage)(app_map.keys(), start, stop):
             metric = item["metric"]
             _, value = item["value"]
+            pvc_name = metric['persistentvolumeclaim']
             volumes.append({
                 "app_id":  str(app_map[metric['namespace']].uuid),
                 "owner": app_map[metric['namespace']].owner_id,
@@ -32,18 +33,21 @@ def _upload_volume_usage(self, start_time, app_map, timestamp):
                 "name": metric["storageclass"],
                 "usage": value,
                 "kwargs": {
-                    "persistentvolumeclaim": metric['persistentvolumeclaim'],
+                    "persistentvolumeclaim": pvc_name,
                 },
-                "timestamp": start
+                "timestamp": start,
+                "identifier": hashlib.md5(pvc_name.encode("utf-8")).hexdigest(),
             })
+        logger.info(f"bulk sent: {len(volumes)}")
         send_usage.apply_async(
-            args=(volumes,), eta=start_time + timedelta(seconds=random.randint(1, 1800))
+            args=(volumes,), eta=start_time + timedelta(seconds=random.randint(1, 3600))
         )
 
     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} volumes to workflow_manager when {timezone.now()}")
+            now = timezone.now()
+            start_time, timestamp, task_id = now, int(now.timestamp()), uuid.uuid4().hex
+            logger.info(f"pushing {task_id} volumes to workflow_manager when {now}")
             app_map = {}
             for app in App.objects.all():
                 app_map[app.id] = app

rootfs/api/models/app.py

@@ -741,6 +741,7 @@ def to_usages(self, timestamp: float):
                     "ptype": ptype,
                 },
                 "timestamp": int(timestamp),
+                "identifier": self.uuid.hex,
             })
         return usage
 

rootfs/api/models/gateway.py

@@ -1,4 +1,5 @@
 import logging
+import hashlib
 import threading
 from django.db import models
 from django.conf import settings
@@ -153,7 +154,8 @@ def to_usages(self, timestamp: float):
             "kwargs": {
                 "name": self.name,
             },
-            "timestamp": int(timestamp)
+            "timestamp": int(timestamp),
+            "identifier": hashlib.md5(self.name.encode("utf-8")).hexdigest(),
         }]
 
     def _check_port(self, port, protocol):

rootfs/api/models/resource.py

@@ -218,7 +218,8 @@ def to_usages(self, timestamp: float):
             "kwargs": {
                 "name": self.name,
             },
-            "timestamp": int(timestamp)
+            "timestamp": int(timestamp),
+            "identifier": self.uuid.hex,
         }]
 
     def _retrieve_status(self):