feat(influxdb): review influxdb code

Author: duanhongyi

rootfs/api/influxdb.py

@@ -0,0 +1,123 @@
+import threading
+from typing import Iterator
+from contextlib import closing
+from django.conf import settings
+from influxdb_client import InfluxDBClient
+from influxdb_client.client.flux_table import FluxRecord
+
+local = threading.local()
+
+
+def _get_influxdb_client() -> InfluxDBClient:
+    if not hasattr(local, "influxdb_client"):
+        local.influxdb_client = InfluxDBClient(
+            url=settings.INFLUXDB_URL,
+            token=settings.INFLUXDB_TOKEN,
+            org=settings.INFLUXDB_ORG
+        )
+    return local.influxdb_client
+
+
+def _query_stream(flux_script: str) -> Iterator[FluxRecord]:
+    with closing(_get_influxdb_client()) as client:
+        with closing(client.query_api()) as query_api:
+            with closing(query_api.query_stream(flux_script)) as records:
+                yield from records
+
+
+def query_container_count(
+        namespaces: Iterator[str], start: int, stop: int) -> Iterator[FluxRecord]:
+    namespace_range = ' or '.join(
+        ['r["namespace"] == \"{namespace}\"' for namespace in namespaces])
+    flux_script = f'''
+        from(bucket: "kubernetes")
+            |> range(start: {start}, stop: {stop})
+            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_container")
+            |> filter(fn: (r) => r["_field"]=="cpu_usage_core_nanoseconds")
+            |> filter(fn: (r) => {namespace_range})
+            |> group(columns: ["_time", "namespace", "container_name"])
+            |> count()
+            |> group(columns: ["namespace", "container_name"])
+            |> top(n: 3)
+            |> min()
+    '''
+    yield from _query_stream(flux_script)
+
+
+def query_network_flow(
+        namespaces: Iterator[str], start: int, stop: int) -> Iterator[FluxRecord]:
+    namespace_range = ' or '.join(
+        ['r["namespace"] == \"{namespace}\"' for namespace in namespaces])
+    flux_script = f'''
+        from(bucket: "kubernetes")
+            |> range(start: {start}, stop: {stop})
+            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_network")
+            |> filter(fn: (r) => r["_field"] == "rx_bytes" or r["_field"] == "tx_bytes")
+            |> filter(fn: (r) => {namespace_range})
+            |> increase()
+            |> last()
+            |> pivot(
+                rowKey:["_time"],
+                columnKey: ["_field"],
+                valueColumn: "_value"
+            )
+    '''
+    yield from _query_stream(flux_script)
+
+
+def query_cpu_usage(
+        namespace, container_type, start, stop, every) -> Iterator[FluxRecord]:
+    flux_script = f"""
+        from(bucket: "kubernetes")
+            |> range(start: {start}, stop: {stop})
+            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_container")
+            |> filter(fn: (r) => r["_field"] == "cpu_usage_nanocores")
+            |> filter(fn: (r) => r["namespace"] == "{namespace}")
+            |> filter(fn: (r) => r["pod_name"] =~ { "/%s-%s/" % (namespace, container_type) })
+            |> group(columns: ["container_name"])
+            |> aggregateWindow(every: {every}, fn: max, createEmpty: false)
+            |> yield(name: "max")
+            |> aggregateWindow(every: {every}, fn: mean, createEmpty: false)
+            |> yield(name: "mean")
+    """
+    yield from _query_stream(flux_script)
+
+
+def query_memory_usage(
+        namespace, container_type, start, stop, every) -> Iterator[FluxRecord]:
+    flux_script = f"""
+        from(bucket: "kubernetes")
+            |> range(start: {start}, stop: {stop})
+            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_container")
+            |> filter(fn: (r) => r["_field"] == "memory_usage_bytes")
+            |> filter(fn: (r) => r["namespace"] == "{namespace}")
+            |> filter(fn: (r) => r["pod_name"] =~ { "/%s-%s/" % (namespace, container_type) })
+            |> group(columns: ["container_name"])
+            |> aggregateWindow(every: {every}, fn: max, createEmpty: false)
+            |> yield(name: "max")
+            |> aggregateWindow(every: {every}, fn: mean, createEmpty: false)
+            |> yield(name: "mean")
+    """
+    yield from _query_stream(flux_script)
+
+
+def query_network_usage(
+        namespace, container_type, start, stop, every) -> Iterator[FluxRecord]:
+    flux_script = f"""
+        from(bucket: "kubernetes")
+            |> range(start: {start}, stop: {stop})
+            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_network")
+            |> filter(fn: (r) => r["_field"] == "rx_bytes" or r["_field"] == "tx_bytes")
+            |> filter(fn: (r) => r["namespace"] == "{namespace}")
+            |> filter(fn: (r) => r["pod_name"] =~ { "/%s-%s/" % (namespace, container_type) })
+            |> group(columns: ["container_name", "_field"])
+            |> aggregateWindow(every: {every}, fn: max, createEmpty: false)
+            |> difference(nonNegative: true)
+            |> pivot(
+                rowKey:["_time"],
+                columnKey: ["_field"],
+                valueColumn: "_value"
+            )
+            |> limit(n:3000)
+    """
+    yield from _query_stream(flux_script)

rootfs/api/management/commands/measure_app.py

@@ -1,108 +1,62 @@
 import uuid
 import time
 import logging
-from contextlib import closing
 from django.utils import timezone
 from django.core.management.base import BaseCommand
 from django.conf import settings
+from api import influxdb
 from api.tasks import measure_networks, measure_instances
 from api.models import App
-from api.utils import get_influxdb_client
 
 logger = logging.getLogger(__name__)
 
 
 class Command(BaseCommand):
     """Management command for push data to manager"""
 
-    def _build_query_networks_flux(self, app_map, timestamp):
-        timestamp = int(timestamp)
-        stop = timestamp - (timestamp % 3600)
-        start = stop - 3600
-        namespace_range = ' or '.join(
-                ['r["namespace"] == \"{app_id}\"' for app_id in app_map.keys()])
-        return f'''
-            from(bucket: "kubernetes")
-            |> range(start: {start}, stop: {stop})
-            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_network"
-                and ({namespace_range}))
-            |> pivot(
-                rowKey:["_time"],
-                columnKey: ["_field"],
-                valueColumn: "_value"
-            )
-            |> increase(columns: ["rx_bytes", "tx_bytes", "tx_errors", "rx_errors"])
-            |> last(column: "_time")
-            '''
-
-    def _build_query_instances_flux(self, app_map, timestamp):
-        timestamp = int(timestamp)
+    def _measure_networks(self, app_map, timestamp):
         stop = timestamp - (timestamp % 3600)
         start = stop - 3600
-        namespace_range = ' or '.join(
-                ['r["namespace"] == \"{app_id}\"' for app_id in app_map.keys()])
-        return f'''
-            from(bucket: "kubernetes")
-            |> range(start: {start}, stop: {stop})
-            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_container"
-                and r["_field"]=="cpu_usage_core_nanoseconds"
-                and ({namespace_range}))
-            |> group(columns: ["_time", "namespace", "container_name"])
-            |> count(column: "_value")
-            |> group(columns: ["namespace", "container_name"])
-            |> top(n: 3)
-            |> min(column: "_value")
-            '''
-
-    def _measure_networks(self, app_map, timestamp):
         networks = []
-        with closing(get_influxdb_client()) as client:
-            with closing(client.query_api()) as query_api:
-                with closing(query_api.query_stream(
-                        self._build_query_networks_flux(app_map, timestamp))
-                        ) as records:
-                    for record in records:
-                        app_id = record["namespace"]
-                        owner_id = app_map[app_id].owner_id
-                        networks.append({
-                            "app_id": app_id,
-                            "owner_id":  owner_id,
-                            "pod_name": record["pod_name"],
-                            "rx_bytes": record["rx_bytes"],
-                            "tx_bytes": record["tx_bytes"],
-                            "timestamp": timestamp
-                        })
+        for record in influxdb.query_network_flow(app_map.keys(), start, stop):
+            app_id = record["namespace"]
+            owner_id = app_map[app_id].owner_id
+            networks.append({
+                "app_id": app_id,
+                "user_id":  owner_id,
+                "pod_name": record["pod_name"],
+                "rx_bytes": record["rx_bytes"],
+                "tx_bytes": record["tx_bytes"],
+                "timestamp": timestamp
+            })
         measure_networks.delay(networks)
 
     def _measure_instances(self, app_map, timestamp):
+        stop = timestamp - (timestamp % 3600)
+        start = stop - 3600
         instances = []
-        with closing(get_influxdb_client()) as client:
-            with closing(client.query_api()) as query_api:
-                with closing(query_api.query_stream(
-                        self._build_query_instances_flux(app_map, timestamp))
-                        ) as records:
-                    for record in records:
-                        app_id = record["namespace"]
-                        owner_id = app_map[app_id].owner_id
-                        container_type = record["container_name"].replace(f"-{app_id}", "", 1)
-                        instances.append({
-                            "app_id": app_id,
-                            "owner_id":  owner_id,
-                            "container_type": container_type,
-                            "container_count": record["_value"],
-                            "timestamp": timestamp
-                        })
+        for record in influxdb.query_container_count(app_map.keys(), start, stop):
+            app_id = record["namespace"]
+            owner_id = app_map[app_id].owner_id
+            container_type = record["container_name"].replace(f"{app_id}-", "", 1)
+            instances.append({
+                "app_id": app_id,
+                "user_id":  owner_id,
+                "container_type": container_type,
+                "container_count": record["_value"],
+                "timestamp": timestamp
+            })
         measure_instances.delay(instances)
 
     def handle(self, *args, **options):
         if settings.WORKFLOW_MANAGER_URL is not None:
-            timestamp = time.time()
+            timestamp = int(time.time())
             task_id = uuid.uuid4().hex
             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
-                if len(app_map) % 1000 == 0:
+                if len(app_map) % 300 == 0:
                     self._measure_networks(app_map, timestamp)
                     self._measure_instances(app_map, timestamp)
                     app_map = {}

rootfs/api/utils.py

@@ -6,13 +6,9 @@
 import hashlib
 import logging
 import random
-import threading
 import math
 from copy import deepcopy
-from django.conf import settings
-from influxdb_client import InfluxDBClient
 
-local = threading.local()
 logger = logging.getLogger(__name__)
 
 
@@ -165,16 +161,6 @@ def apply_tasks(tasks):
     executor.shutdown(wait=True)
 
 
-def get_influxdb_client():
-    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
-
-
 def unit_to_bytes(size):
     """
     size: str

rootfs/api/views.py

@@ -3,7 +3,6 @@
 """
 import logging
 from copy import deepcopy
-
 from django.http import Http404, HttpResponse
 from django.conf import settings
 from django.contrib.auth.models import User
@@ -19,10 +18,11 @@
 from rest_framework.viewsets import GenericViewSet
 from rest_framework.authtoken.models import Token
 
-from api import authentication, models, permissions, serializers, viewsets
+from api import influxdb, authentication, models, permissions, serializers, viewsets
 from api.models import AlreadyExists, ServiceUnavailable, DryccException, \
     UnprocessableEntity
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -908,3 +908,81 @@ def disable(self, request, **kwargs):
         user.is_active = False
         user.save(update_fields=['is_active', ])
         return Response(status=status.HTTP_204_NO_CONTENT)
+
+
+class MetricView(BaseDryccViewSet):
+    """Getting monitoring indicators from influxdb"""
+
+    def _get_app(self):
+        app = get_object_or_404(models.App, id=self.kwargs['app_id'])
+        self.check_object_permissions(self.request, app)
+        return app
+
+    def _get_cpus(self, app_id, container_type, start, stop, every):
+        avg_total, max_total = [], []
+        for record in influxdb.query_memory_usage(app_id, container_type, start, stop, every):
+            if record["result"] == "mean":
+                avg_total.append((record["_value"], record["timestamp"]))
+            else:
+                max_total.append((record["_value"], record["timestamp"]))
+        return {
+            "max_total": max_total,
+            "avg_total": avg_total
+        }
+
+    def _get_memory(self, app_id, container_type, start, stop, every):
+        max_total, avg_total = [], []
+        for record in influxdb.query_memory_usage(app_id, container_type, start, stop, every):
+            if record["result"] == "mean":
+                avg_total.append((record["_value"], record["timestamp"]))
+            else:
+                max_total.append((record["_value"], record["timestamp"]))
+        return {
+            "max_total": max_total,
+            "avg_total": avg_total
+        }
+
+    def _get_networks(self, app_id, container_type, start, stop, every):
+        networks = []
+        for record in influxdb.query_network_usage(app_id, container_type, start, stop, every):
+            networks.append((record["rx_bytes"], record["tx_bytes"], record["timestamp"]))
+        return networks
+
+    def _get_container_count(self, app_id, container_type, start, stop):
+        for record in influxdb.query_container_count([app_id, ], start, stop):
+            if record["container_name"] == "%s-%s" % (app_id, container_type):
+                return record["_value"]
+        return 0
+
+    def status(self, request, **kwargs):
+        """
+        {
+
+            app_id: "django_t1",
+            container_type: "web",
+            container_count: 1
+            cpus: {
+                max_total: [(50000, 1611023853)],
+                avg_total: [(50000, 1611023853)],
+                timestamp: 1611023853
+            },
+            memory: {
+                max_total: [(50000, 1611023853)],
+                avg_total: [(50000, 1611023853)],
+                timestamp: 1611023853
+            },
+            networks: [
+                (10000, 50000, 1611023853)
+            ],
+        }
+        """
+        app_id, container_type = self._get_app().pk, kwargs['container_type']
+        start, stop, every = kwargs['start'], kwargs['stop'], kwargs["every"]
+        return {
+            "app_id": app_id,
+            "container_type": container_type,
+            "container_count": self._get_container_count(app_id, container_type, start, stop),
+            "cpu_usage_list": self._get_cpus(app_id, container_type, start, stop, every),
+            "memory": self._get_memory(app_id, container_type, start, stop, every),
+            "networks": self._get_networks(app_id, container_type, start, stop, every)
+        }