--- /dev/null
+"""
+Copyright (c) 2015 SONATA-NFV and Paderborn University
+ALL RIGHTS RESERVED.
+
+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.
+
+Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
+nor the names of its contributors may be used to endorse or promote
+products derived from this software without specific prior written
+permission.
+
+This work has been performed in the framework of the SONATA project,
+funded by the European Commission under Grant number 671517 through
+the Horizon 2020 and 5G-PPP programmes. The authors would like to
+acknowledge the contributions of their colleagues of the SONATA
+partner consortium (www.sonata-nfv.eu).
+"""
+
+"""
+Monitor the skewness of the resource usage probability distribution
+and export to a Prometheus Push Gateway
+(c) 2017 by Steven Van Rossem <steven.vanrossem@intec.ugent.be>
+"""
+
+#!/usr/bin/python3
+
+from time import sleep, time, perf_counter
+import math
+from prometheus_client import start_http_server, Summary, Histogram, Gauge, Counter, REGISTRY, CollectorRegistry, \
+ pushadd_to_gateway, push_to_gateway, delete_from_gateway
+import threading
+import os
+import json
+
+import logging
+LOG = logging.getLogger('skewmon')
+LOG.setLevel(level=logging.DEBUG)
+LOG.addHandler(logging.StreamHandler())
+
+
+# put env vars in Dockerfile
+PUSHGATEWAY_IP = '172.17.0.1'
+PUSHGATEWAY_PORT = 9091
+PUSHGATEWAY_ADDR = ':'.join([PUSHGATEWAY_IP, str(PUSHGATEWAY_PORT)])
+
+
+#general settings (ms)
+SAMPLE_PERIOD = int(os.environ['SAMPLE_PERIOD'])
+TOTAL_PERIOD = int(os.environ['TOTAL_PERIOD'])
+
+# define global variables
+registry = CollectorRegistry()
+exported_metric = Gauge('skewness', 'Skewness of docker vnf resource usage',
+ ['vnf_id', 'vnf_name', 'vnf_metric'], registry=registry)
+
+# find the VNFs to monitor
+# {metric_shortId: {VNF_NAME:<>,VNF_ID:<>,VNF_METRIC:<>}}
+
+def get_vnfs_to_monitor(config):
+ for key in config:
+ vnf_name = config[key].get('VNF_NAME')
+ vnf_id = config[key].get('VNF_ID')
+ vnf_metric = config[key].get('VNF_METRIC')
+ yield (vnf_id, vnf_name, vnf_metric)
+
+# export metric to the Prometheus PushGateway
+def export_metrics(key=None):
+ try:
+ pushadd_to_gateway(PUSHGATEWAY_ADDR, job='sonemu-skewmon', registry=registry, grouping_key=key)
+ except Exception as e:
+ LOG.warning("Pushgateway not reachable: {0}".format(str(e)))
+
+class skewness_monitor():
+ def __init__(self, docker_id, docker_name, metric):
+ # Prometheus metric to export
+ self.prom_skewness = exported_metric
+ self.docker_id = docker_id
+ self.docker_name = docker_name
+ self.vnf_metric = metric
+
+ # https://www.datadoghq.com/blog/how-to-collect-docker-metrics/
+ self.cpu_proc_file = '/sys/fs/cgroup/cpuacct/docker/{0}/cpuacct.usage'.format(self.docker_id)
+ self.mem_proc_file = '/sys/fs/cgroup/memory/docker/{0}/memory.usage_in_bytes'.format(self.docker_id)
+ metric_dict = {'cpu': self.cpu_proc_file,
+ 'mem': self.mem_proc_file}
+
+ self.proc_file = metric_dict[metric]
+
+ self.fp = open(self.proc_file)
+
+ #monitoring thread
+ self.export_thread = None
+ self.monitor_stop = threading.Event()
+
+ # get statistics with certain frequency and export skewness for further analysis
+ def _calc_skewness(self):
+
+ cpu_count0 = 0
+ time0 = 0
+
+ #milliseconds
+ stat_delta = SAMPLE_PERIOD
+ sample_T = TOTAL_PERIOD
+
+ data = []
+ n = 0
+
+ moment1 = 0
+ moment2 = 0
+ moment3 = 0
+
+ fp = self.fp
+
+ #collect samples
+ for n in range(0,round(sample_T/stat_delta)):
+ # first measurement
+ if cpu_count0 <= 0 or time0 <= 0:
+ time0 = perf_counter()
+ cpu_count0 = int(fp.read().strip())
+ fp.seek(0)
+ sleep(stat_delta/1000)
+ continue
+
+
+ #perf_counter in seconds
+ time1 = perf_counter()
+
+ # cpu count in nanoseconds
+ cpu_count1 = int(fp.read().strip())
+ fp.seek(0)
+
+ cpu_delta = cpu_count1 - cpu_count0
+ cpu_count0 = cpu_count1
+
+ time_delta = time1 - time0
+ time0 = time1
+
+ #work in nanoseconds
+ metric = (cpu_delta / (time_delta * 1e9))
+
+ data.append(metric)
+
+ #running calculation of sample moments
+ moment1 += metric
+ temp = metric * metric
+ moment2 += temp
+ moment3 += temp * metric
+
+
+ sleep(stat_delta/1000)
+
+ # calc skewness
+ M1 = (1 / n) * moment1
+ M2 = ((1 / n) * moment2) - M1**2
+ M3 = ((1 / n) * moment3) - (3 * M1 * ((1 / n) * moment2)) + (2 * M1**3)
+
+ s2 = (math.sqrt(n*(n - 1))/(n - 2)) * (M3 / (M2)**1.5)
+
+ LOG.info("docker_name: {0} metric: {1}".format(self.docker_name, self.vnf_metric))
+ LOG.info("Nsamples: {0}".format(n))
+ LOG.info("skewness: {0:.2f}".format(s2))
+ LOG.info("\n")
+
+ return s2
+
+ def _export_skewness_loop(self, stop_event):
+ #loop until flag is set
+ while(not stop_event.is_set()):
+ try:
+ skewness = self._calc_skewness()
+ self.prom_skewness.labels(vnf_id=self.docker_id, vnf_name=self.docker_name, vnf_metric=self.vnf_metric)\
+ .set(skewness)
+ except ZeroDivisionError as e:
+ self.prom_skewness.labels(vnf_id=self.docker_id, vnf_name=self.docker_name, vnf_metric=self.vnf_metric) \
+ .set(float('nan'))
+ LOG.warning("{1}: Skewness cannot be calculated: {0}".format(str(e), self.docker_name))
+ except Exception as e:
+ LOG.warning("Skewness cannot be calculated, stop thread: {0}".format(str(e)))
+ self.monitor_stop.set()
+
+ # if while has ended, monitoring thread will stop
+ self.prom_skewness.labels(vnf_id=self.docker_id, vnf_name=self.docker_name, vnf_metric=self.vnf_metric) \
+ .set(float('nan'))
+
+ #start the monitoring thread
+ def start(self):
+ if self.export_thread is not None:
+ LOG.warning('monitor thread is already running for: {0}'.format(self.docker_name))
+ return
+
+ self.export_thread = threading.Thread(target=self._export_skewness_loop, args=(self.monitor_stop,))
+ self.export_thread.start()
+ LOG.info('started thread: {0}'.format(self.docker_name))
+
+ #stop the monitoring thread
+ def stop(self):
+ self.monitor_stop.set()
+
+
+if __name__ == '__main__':
+
+ #started_vnfs {vnf_id:object}
+ vnfs_monitored = {}
+
+ # endless loop
+ while True:
+ #check config.txt for docker ids/names
+ configfile = open('/config.txt', 'r')
+ config = json.load(configfile)
+ vnfs_to_monitor = list(get_vnfs_to_monitor(config))
+
+ #for each new docker id in ENV start thread to monitor skewness
+ for vnf_id, vnf_name, vnf_metric in vnfs_to_monitor:
+ key = '_'.join([vnf_metric, vnf_id])
+ if key not in vnfs_monitored:
+ try:
+ vnfs_monitored[key] = skewness_monitor(vnf_id, vnf_name, vnf_metric)
+ vnfs_monitored[key].start()
+ except Exception as e:
+ LOG.warning("Monitor cannot be started: {0}".format(str(e)))
+
+ #for each removed docker id ENV, stop export
+ for vnf_key in list(vnfs_monitored):
+ vnf_keys_to_monitor = ['_'.join([vnf_metric, vnf_id]) for vnf_id, vnf_name, vnf_metric in vnfs_to_monitor]
+ if vnf_key not in vnf_keys_to_monitor:
+ vnfs_monitored[vnf_key].stop()
+
+ vnf_name = vnfs_monitored[vnf_key].docker_name
+ vnf_metric, vnf_id = vnf_key.split('_')
+ LOG.info('stop monitored VNFs: {0}'.format(vnfs_monitored[vnf_key].docker_name))
+ del vnfs_monitored[vnf_key]
+
+ # remove metric with labels from registry
+ # (Push Gateway remembers last pushed value, so this is not so useful)
+ # collector = registry._names_to_collectors['skewness']
+ # if (vnf_id, vnf_name, vnf_metric) in collector._metrics:
+ # collector.remove(vnf_id, vnf_name, vnf_metric)
+ delete_from_gateway(PUSHGATEWAY_ADDR, job='sonemu-skewmon')
+
+
+ #push to Prometheus gateway
+ export_metrics()
+ LOG.info('monitored VNFs: {0}'.format([monitor.docker_name for key, monitor in vnfs_monitored.items()]))
+ # wait before checking again
+ sleep(TOTAL_PERIOD/1000)
\ No newline at end of file
projects / osm / vim-emu.git / commitdiff