# Copyright (c) 2016-2017 Intel Corporation # # 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. """ Resource collection definitions """ from __future__ import absolute_import from __future__ import print_function import tempfile import logging import os import os.path import re import multiprocessing from collections import Sequence from oslo_config import cfg from yardstick import ssh from yardstick.network_services.nfvi.collectd import AmqpConsumer from yardstick.network_services.utils import get_nsb_option LOG = logging.getLogger(__name__) CONF = cfg.CONF ZMQ_OVS_PORT = 5567 ZMQ_POLLING_TIME = 12000 LIST_PLUGINS_ENABLED = ["amqp", "cpu", "cpufreq", "intel_rdt", "memory", "hugepages", "dpdkstat", "virt", "ovs_stats", "intel_pmu"] class ResourceProfile(object): """ This profile adds a resource at the beginning of the test session """ def __init__(self, mgmt, interfaces=None, cores=None): self.enable = True self.connection = None self.cores = cores if isinstance(cores, Sequence) else [] self._queue = multiprocessing.Queue() self.amqp_client = None self.interfaces = interfaces if isinstance(interfaces, Sequence) else [] # why the host or ip? self.vnfip = mgmt.get("host", mgmt["ip"]) self.connection = ssh.SSH.from_node(mgmt, overrides={"ip": self.vnfip}) self.connection.wait() def check_if_sa_running(self, process): """ verify if system agent is running """ err, pid, _ = self.connection.execute("pgrep -f %s" % process) return [err == 0, pid] def run_collectd_amqp(self): """ run amqp consumer to collect the NFVi data """ amqp_url = 'amqp://admin:admin@{}:5672/%2F'.format(self.vnfip) amqp = AmqpConsumer(amqp_url, self._queue) try: amqp.run() except (AttributeError, RuntimeError, KeyboardInterrupt): amqp.stop() @classmethod def parse_simple_resource(cls, key, value): reskey = "/".join(rkey for rkey in key if "nsb_stats" not in rkey) return {reskey: value.split(":")[1]} @classmethod def get_cpu_data(cls, res_key0, res_key1, value): """ Get cpu topology of the host """ pattern = r"-(\d+)" if 'cpufreq' in res_key0: metric, source = res_key0, res_key1 else: metric, source = res_key1, res_key0 match = re.search(pattern, source, re.MULTILINE) if not match: return "error", "Invalid", "", "" time, value = value.split(":") return str(match.group(1)), metric, value, time @classmethod def parse_hugepages(cls, key, value): return cls.parse_simple_resource(key, value) @classmethod def parse_dpdkstat(cls, key, value): return cls.parse_simple_resource(key, value) @classmethod def parse_virt(cls, key, value): return cls.parse_simple_resource(key, value) @classmethod def parse_ovs_stats(cls, key, value): return cls.parse_simple_resource(key, value) @classmethod def parse_intel_pmu_stats(cls, key, value): return {''.join(key): value.split(":")[1]} def parse_collectd_result(self, metrics, core_list): """ convert collectd data into json""" result = { "cpu": {}, "memory": {}, "hugepages": {}, "dpdkstat": {}, "virt": {}, "ovs_stats": {}, "intel_pmu": {}, } testcase = "" for key, value in metrics.items(): key_split = key.split("/") res_key_iter = (key for key in key_split if "nsb_stats" not in key) res_key0 = next(res_key_iter) res_key1 = next(res_key_iter) if "cpu" in res_key0 or "intel_rdt" in res_key0: cpu_key, name, metric, testcase = \ self.get_cpu_data(res_key0, res_key1, value) if cpu_key in core_list: result["cpu"].setdefault(cpu_key, {}).update({name: metric}) elif "memory" in res_key0: result["memory"].update({res_key1: value.split(":")[0]}) elif "hugepages" in res_key0: result["hugepages"].update(self.parse_hugepages(key_split, value)) elif "dpdkstat" in res_key0: result["dpdkstat"].update(self.parse_dpdkstat(key_split, value)) elif "virt" in res_key1: result["virt"].update(self.parse_virt(key_split, value)) elif "ovs_stats" in res_key0: result["ovs_stats"].update(self.parse_ovs_stats(key_split, value)) elif "intel_pmu-all" in res_key0: result["intel_pmu"].update(self.parse_intel_pmu_stats(res_key1, value)) result["timestamp"] = testcase return result def amqp_process_for_nfvi_kpi(self): """ amqp collect and return nfvi kpis """ if self.amqp_client is None and self.enable: self.amqp_client = \ multiprocessing.Process(target=self.run_collectd_amqp) self.amqp_client.start() def amqp_collect_nfvi_kpi(self): """ amqp collect and return nfvi kpis """ if not self.enable: return {} metric = {} while not self._queue.empty(): metric.update(self._queue.get()) msg = self.parse_collectd_result(metric, self.cores) return msg def _provide_config_file(self, bin_path, nfvi_cfg, kwargs): with open(os.path.join(bin_path, nfvi_cfg), 'r') as cfg: template = cfg.read() cfg, cfg_content = tempfile.mkstemp() with os.fdopen(cfg, "w+") as cfg: cfg.write(template.format(**kwargs)) cfg_file = os.path.join(bin_path, nfvi_cfg) self.connection.put(cfg_content, cfg_file) def _prepare_collectd_conf(self, bin_path): """ Prepare collectd conf """ loadplugin = "\n".join("LoadPlugin {0}".format(plugin) for plugin in LIST_PLUGINS_ENABLED) interfaces = "\n".join("PortName '{0[name]}'".format(interface) for interface in self.interfaces) kwargs = { "interval": '25', "loadplugin": loadplugin, "dpdk_interface": interfaces, } self._provide_config_file(bin_path, 'collectd.conf', kwargs) def _start_collectd(self, connection, bin_path): LOG.debug("Starting collectd to collect NFVi stats") connection.execute('sudo pkill -9 collectd') bin_path = get_nsb_option("bin_path") collectd_path = os.path.join(bin_path, "collectd", "collectd") exit_status = connection.execute("which %s > /dev/null 2>&1" % collectd_path)[0] if exit_status != 0: LOG.warning("%s is not present disabling", collectd_path) # disable auto-provisioning because it requires Internet access # collectd_installer = os.path.join(bin_path, "collectd.sh") # provision_tool(connection, collectd) # http_proxy = os.environ.get('http_proxy', '') # https_proxy = os.environ.get('https_proxy', '') # connection.execute("sudo %s '%s' '%s'" % ( # collectd_installer, http_proxy, https_proxy)) return LOG.debug("Starting collectd to collect NFVi stats") self._prepare_collectd_conf(bin_path) # Reset amqp queue LOG.debug("reset and setup amqp to collect data from collectd") connection.execute("sudo rm -rf /var/lib/rabbitmq/mnesia/rabbit*") connection.execute("sudo service rabbitmq-server start") connection.execute("sudo rabbitmqctl stop_app") connection.execute("sudo rabbitmqctl reset") connection.execute("sudo rabbitmqctl start_app") connection.execute("sudo service rabbitmq-server restart") LOG.debug("Creating amdin user for rabbitmq in order to collect data from collectd") connection.execute("sudo rabbitmqctl delete_user guest") connection.execute("sudo rabbitmqctl add_user admin admin") connection.execute("sudo rabbitmqctl authenticate_user admin admin") connection.execute("sudo rabbitmqctl set_permissions -p / admin \".*\" \".*\" \".*\"") LOG.debug("Start collectd service.....") connection.execute("sudo %s" % collectd_path) LOG.debug("Done") def initiate_systemagent(self, bin_path): """ Start system agent for NFVi collection on host """ if self.enable: self._start_collectd(self.connection, bin_path) def start(self): """ start nfvi collection """ if self.enable: LOG.debug("Start NVFi metric collection...") def stop(self): """ stop nfvi collection """ if not self.enable: return agent = "collectd" LOG.debug("Stop resource monitor...") if self.amqp_client is not None: self.amqp_client.terminate() status, pid = self.check_if_sa_running(agent) if status == 0: return self.connection.execute('sudo kill -9 %s' % pid) self.connection.execute('sudo pkill -9 %s' % agent) self.connection.execute('sudo service rabbitmq-server stop') self.connection.execute("sudo rabbitmqctl stop_app")