Implement InfluxDB for usage metrics

zoe_api/api_endpoint.py

@@ -214,7 +214,7 @@ class APIEndpoint:
                         endpoint_ext = None
                     else:
                         endpoint_ext = '/{}/{}'.format(zoe_lib.config.get_conf().traefik_base_url, port.proxy_key())
-                    endpoints.append((port.name, endpoint, endpoint_ext))
+                    endpoints.append((port.readable_name, endpoint, endpoint_ext))
 
         return services_info, endpoints
 

zoe_lib/config.py

@@ -62,7 +62,9 @@ def load_configuration(test_conf=None):
         argparser.add_argument('--api-listen-uri', help='ZMQ API listen address', default='tcp://*:4850')
 
         argparser.add_argument('--kairosdb-enable', action="store_true", help='Enable usage metric input from KairosDB')
-        argparser.add_argument('--kairosdb-url', help='URL of the KairosDB service (ex. http://localhost:8086)', default='http://localhost:8090')
+        argparser.add_argument('--kairosdb-url', help='URL of the KairosDB service (ex. http://localhost:8090)', default='http://localhost:8090')
+        argparser.add_argument('--influxdb-enable', action="store_true", help='Enable usage metric input from InfluxDB')
+        argparser.add_argument('--influxdb-url', help='URL of the InfluxDB service (ex. http://localhost:8086)', default='http://localhost:8086')
 
         argparser.add_argument('--workspace-base-path', help='Base directory where user workspaces will be created. Must be visible at this path on all hosts.', default='/mnt/zoe-workspaces')
         argparser.add_argument('--workspace-deployment-path', help='Path appended to the workspace path to distinguish this deployment. If unspecified is equal to the deployment name.', default='--default--')

zoe_lib/state/port.py

@@ -38,6 +38,7 @@ class Port(BaseRecord):
         self.internal_number = self.description['port_number']
         self.protocol = self.description['protocol']
         self.url_template = self.description['url_template']
+        self.readable_name = self.description['name']
 
     def serialize(self):
         """Generates a dictionary that can be serialized in JSON."""

zoe_master/metrics/base.py

@@ -23,6 +23,7 @@ from copy import deepcopy
 from zoe_lib.config import get_conf
 from zoe_master.backends.interface import get_platform_state
 from zoe_master.metrics.kairosdb import KairosDBInMetrics
+from zoe_master.metrics.influxdb import InfluxDBInMetrics
 
 log = logging.getLogger(__name__)
 
@@ -45,6 +46,8 @@ class StatsManager(threading.Thread):
         self._current_platform_stats = None
         if get_conf().kairosdb_enable:
             self.usage_metrics = KairosDBInMetrics()
+        elif get_conf().influxdb_enable:
+            self.usage_metrics = InfluxDBInMetrics()
         else:
             self.usage_metrics = None
 

zoe_master/metrics/influxdb.py

+# Copyright (c) 2018, Daniele Venzano
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Retrieves metrics about services from InfluxDB."""
+
+import logging
+
+import requests
+
+from zoe_lib.config import get_conf
+
+log = logging.getLogger(__name__)
+
+
+class InfluxDBInMetrics:
+    """InfluxDB metrics."""
+    def __init__(self):
+        self.base_url = get_conf().influxdb_url
+
+    def get_service_usage(self, service_id):
+        """Query the DB for the current usage metrics."""
+        query_cpu = 'SELECT mean("usage_percent") FROM "docker_container_cpu" WHERE "zoe_service_id" = \'{}\' AND time >= now() - 3m GROUP BY time(2m) ORDER BY time DESC LIMIT 1'.format(service_id)
+        query_mem = 'SELECT mean("usage") FROM "docker_container_mem" WHERE "zoe_service_id" = \'{}\' AND time >= now() - 3m GROUP BY time(2m) ORDER BY time DESC LIMIT 1'.format(service_id)
+
+        url = self.base_url + '/query'
+        resp = requests.post(url, data={"db": 'telegraf', 'q': query_cpu + ';' + query_mem})
+        influx_resp = resp.json()
+        if "error" in influx_resp:
+            log.warning("InfluxDB reported an error: {}".format(influx_resp['error']))
+            return None
+        return self._extract_data(influx_resp)
+
+    def _extract_data(self, data):
+        ret = {}
+        cpu_results = data['results'][0]
+        assert cpu_results['statement_id'] == 0
+        if 'series' in cpu_results:
+            val = cpu_results['series'][0]['values'][0][1]
+            if val is None:
+                ret['cpu_usage'] = 0
+            else:
+                ret['cpu_usage'] = val
+        else:
+            ret['cpu_usage'] = 0
+
+        mem_results = data['results'][1]
+        assert mem_results['statement_id'] == 1
+        if 'series' in mem_results:
+            val = mem_results['series'][0]['values'][0][1]
+            if val is None:
+                ret['mem_usage'] = 0
+            else:
+                ret['mem_usage'] = val
+        else:
+            ret['mem_usage'] = 0
+        return ret