yardstick/network_services/nfvi/resource.py

   1 # Copyright (c) 2016-2017 Intel Corporation
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 """ Resource collection definitions """
  15
  16 from __future__ import absolute_import
  17 from __future__ import print_function
  18 import tempfile
  19 import logging
  20 import os
  21 import os.path
  22 import re
  23 import multiprocessing
  24 from collections import Sequence
  25
  26 from oslo_config import cfg
  27
  28 from yardstick import ssh
  29 from yardstick.network_services.nfvi.collectd import AmqpConsumer
  30 from yardstick.network_services.utils import get_nsb_option
  31
  32 LOG = logging.getLogger(__name__)
  33
  34 CONF = cfg.CONF
  35 ZMQ_OVS_PORT = 5567
  36 ZMQ_POLLING_TIME = 12000
  37 LIST_PLUGINS_ENABLED = ["amqp", "cpu", "cpufreq", "intel_rdt", "memory",
  38                         "hugepages", "dpdkstat", "virt", "ovs_stats", "intel_pmu"]
  39
  40
  41 class ResourceProfile(object):
  42     """
  43     This profile adds a resource at the beginning of the test session
  44     """
  45
  46     def __init__(self, mgmt, interfaces=None, cores=None):
  47         self.enable = True
  48         self.connection = None
  49         self.cores = cores if isinstance(cores, Sequence) else []
  50         self._queue = multiprocessing.Queue()
  51         self.amqp_client = None
  52         self.interfaces = interfaces if isinstance(interfaces, Sequence) else []
  53
  54         # why the host or ip?
  55         self.vnfip = mgmt.get("host", mgmt["ip"])
  56         self.connection = ssh.SSH.from_node(mgmt, overrides={"ip": self.vnfip})
  57
  58         self.connection.wait()
  59
  60     def check_if_sa_running(self, process):
  61         """ verify if system agent is running """
  62         err, pid, _ = self.connection.execute("pgrep -f %s" % process)
  63         return [err == 0, pid]
  64
  65     def run_collectd_amqp(self):
  66         """ run amqp consumer to collect the NFVi data """
  67         amqp_url = 'amqp://admin:admin@{}:5672/%2F'.format(self.vnfip)
  68         amqp = AmqpConsumer(amqp_url, self._queue)
  69         try:
  70             amqp.run()
  71         except (AttributeError, RuntimeError, KeyboardInterrupt):
  72             amqp.stop()
  73
  74     @classmethod
  75     def parse_simple_resource(cls, key, value):
  76         reskey = "/".join(rkey for rkey in key if "nsb_stats" not in rkey)
  77         return {reskey: value.split(":")[1]}
  78
  79     @classmethod
  80     def get_cpu_data(cls, res_key0, res_key1, value):
  81         """ Get cpu topology of the host """
  82         pattern = r"-(\d+)"
  83
  84         if 'cpufreq' in res_key0:
  85             metric, source = res_key0, res_key1
  86         else:
  87             metric, source = res_key1, res_key0
  88
  89         match = re.search(pattern, source, re.MULTILINE)
  90         if not match:
  91             return "error", "Invalid", "", ""
  92
  93         time, value = value.split(":")
  94         return str(match.group(1)), metric, value, time
  95
  96     @classmethod
  97     def parse_hugepages(cls, key, value):
  98         return cls.parse_simple_resource(key, value)
  99
 100     @classmethod
 101     def parse_dpdkstat(cls, key, value):
 102         return cls.parse_simple_resource(key, value)
 103
 104     @classmethod
 105     def parse_virt(cls, key, value):
 106         return cls.parse_simple_resource(key, value)
 107
 108     @classmethod
 109     def parse_ovs_stats(cls, key, value):
 110         return cls.parse_simple_resource(key, value)
 111
 112     @classmethod
 113     def parse_intel_pmu_stats(cls, key, value):
 114         return {''.join(key): value.split(":")[1]}
 115
 116     def parse_collectd_result(self, metrics, core_list):
 117         """ convert collectd data into json"""
 118         result = {
 119             "cpu": {},
 120             "memory": {},
 121             "hugepages": {},
 122             "dpdkstat": {},
 123             "virt": {},
 124             "ovs_stats": {},
 125             "intel_pmu": {},
 126         }
 127         testcase = ""
 128
 129         for key, value in metrics.items():
 130             key_split = key.split("/")
 131             res_key_iter = (key for key in key_split if "nsb_stats" not in key)
 132             res_key0 = next(res_key_iter)
 133             res_key1 = next(res_key_iter)
 134
 135             if "cpu" in res_key0 or "intel_rdt" in res_key0:
 136                 cpu_key, name, metric, testcase = \
 137                     self.get_cpu_data(res_key0, res_key1, value)
 138                 if cpu_key in core_list:
 139                     result["cpu"].setdefault(cpu_key, {}).update({name: metric})
 140
 141             elif "memory" in res_key0:
 142                 result["memory"].update({res_key1: value.split(":")[0]})
 143
 144             elif "hugepages" in res_key0:
 145                 result["hugepages"].update(self.parse_hugepages(key_split, value))
 146
 147             elif "dpdkstat" in res_key0:
 148                 result["dpdkstat"].update(self.parse_dpdkstat(key_split, value))
 149
 150             elif "virt" in res_key1:
 151                 result["virt"].update(self.parse_virt(key_split, value))
 152
 153             elif "ovs_stats" in res_key0:
 154                 result["ovs_stats"].update(self.parse_ovs_stats(key_split, value))
 155
 156             elif "intel_pmu-all" in res_key0:
 157                 result["intel_pmu"].update(self.parse_intel_pmu_stats(res_key1, value))
 158
 159         result["timestamp"] = testcase
 160
 161         return result
 162
 163     def amqp_process_for_nfvi_kpi(self):
 164         """ amqp collect and return nfvi kpis """
 165         if self.amqp_client is None and self.enable:
 166             self.amqp_client = \
 167                 multiprocessing.Process(target=self.run_collectd_amqp)
 168             self.amqp_client.start()
 169
 170     def amqp_collect_nfvi_kpi(self):
 171         """ amqp collect and return nfvi kpis """
 172         if not self.enable:
 173             return {}
 174
 175         metric = {}
 176         while not self._queue.empty():
 177             metric.update(self._queue.get())
 178         msg = self.parse_collectd_result(metric, self.cores)
 179         return msg
 180
 181     def _provide_config_file(self, bin_path, nfvi_cfg, kwargs):
 182         with open(os.path.join(bin_path, nfvi_cfg), 'r') as cfg:
 183             template = cfg.read()
 184         cfg, cfg_content = tempfile.mkstemp()
 185         with os.fdopen(cfg, "w+") as cfg:
 186             cfg.write(template.format(**kwargs))
 187         cfg_file = os.path.join(bin_path, nfvi_cfg)
 188         self.connection.put(cfg_content, cfg_file)
 189
 190     def _prepare_collectd_conf(self, bin_path):
 191         """ Prepare collectd conf """
 192         loadplugin = "\n".join("LoadPlugin {0}".format(plugin)
 193                                for plugin in LIST_PLUGINS_ENABLED)
 194
 195         interfaces = "\n".join("PortName '{0[name]}'".format(interface)
 196                                for interface in self.interfaces)
 197
 198         kwargs = {
 199             "interval": '25',
 200             "loadplugin": loadplugin,
 201             "dpdk_interface": interfaces,
 202         }
 203         self._provide_config_file(bin_path, 'collectd.conf', kwargs)
 204
 205     def _start_collectd(self, connection, bin_path):
 206         LOG.debug("Starting collectd to collect NFVi stats")
 207         connection.execute('sudo pkill -9 collectd')
 208         bin_path = get_nsb_option("bin_path")
 209         collectd_path = os.path.join(bin_path, "collectd", "collectd")
 210         exit_status = connection.execute("which %s > /dev/null 2>&1" % collectd_path)[0]
 211         if exit_status != 0:
 212             LOG.warning("%s is not present disabling", collectd_path)
 213             # disable auto-provisioning because it requires Internet access
 214             # collectd_installer = os.path.join(bin_path, "collectd.sh")
 215             # provision_tool(connection, collectd)
 216             # http_proxy = os.environ.get('http_proxy', '')
 217             # https_proxy = os.environ.get('https_proxy', '')
 218             # connection.execute("sudo %s '%s' '%s'" % (
 219             #     collectd_installer, http_proxy, https_proxy))
 220             return
 221         LOG.debug("Starting collectd to collect NFVi stats")
 222         self._prepare_collectd_conf(bin_path)
 223
 224         # Reset amqp queue
 225         LOG.debug("reset and setup amqp to collect data from collectd")
 226         connection.execute("sudo rm -rf /var/lib/rabbitmq/mnesia/rabbit*")
 227         connection.execute("sudo service rabbitmq-server start")
 228         connection.execute("sudo rabbitmqctl stop_app")
 229         connection.execute("sudo rabbitmqctl reset")
 230         connection.execute("sudo rabbitmqctl start_app")
 231         connection.execute("sudo service rabbitmq-server restart")
 232
 233         LOG.debug("Creating amdin user for rabbitmq in order to collect data from collectd")
 234         connection.execute("sudo rabbitmqctl delete_user guest")
 235         connection.execute("sudo rabbitmqctl add_user admin admin")
 236         connection.execute("sudo rabbitmqctl authenticate_user admin admin")
 237         connection.execute("sudo rabbitmqctl set_permissions -p / admin \".*\" \".*\" \".*\"")
 238
 239         LOG.debug("Start collectd service.....")
 240         connection.execute("sudo %s" % collectd_path)
 241         LOG.debug("Done")
 242
 243     def initiate_systemagent(self, bin_path):
 244         """ Start system agent for NFVi collection on host """
 245         if self.enable:
 246             self._start_collectd(self.connection, bin_path)
 247
 248     def start(self):
 249         """ start nfvi collection """
 250         if self.enable:
 251             LOG.debug("Start NVFi metric collection...")
 252
 253     def stop(self):
 254         """ stop nfvi collection """
 255         if not self.enable:
 256             return
 257
 258         agent = "collectd"
 259         LOG.debug("Stop resource monitor...")
 260
 261         if self.amqp_client is not None:
 262             self.amqp_client.terminate()
 263
 264         status, pid = self.check_if_sa_running(agent)
 265         if status == 0:
 266             return
 267
 268         self.connection.execute('sudo kill -9 %s' % pid)
 269         self.connection.execute('sudo pkill -9 %s' % agent)
 270         self.connection.execute('sudo service rabbitmq-server stop')
 271         self.connection.execute("sudo rabbitmqctl stop_app")