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
  25 from oslo_config import cfg
  26
  27 from yardstick import ssh
  28 from yardstick.common.utils import validate_non_string_sequence
  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.cores = validate_non_string_sequence(cores, default=[])
  49         self._queue = multiprocessing.Queue()
  50         self.amqp_client = None
  51         self.interfaces = validate_non_string_sequence(interfaces, default=[])
  52
  53         # why the host or ip?
  54         self.vnfip = mgmt.get("host", mgmt["ip"])
  55         self.connection = ssh.SSH.from_node(mgmt, overrides={"ip": self.vnfip})
  56         self.connection.wait()
  57
  58     def check_if_sa_running(self, process):
  59         """ verify if system agent is running """
  60         err, pid, _ = self.connection.execute("pgrep -f %s" % process)
  61         return [err == 0, pid]
  62
  63     def run_collectd_amqp(self):
  64         """ run amqp consumer to collect the NFVi data """
  65         amqp_url = 'amqp://admin:admin@{}:5672/%2F'.format(self.vnfip)
  66         amqp = AmqpConsumer(amqp_url, self._queue)
  67         try:
  68             amqp.run()
  69         except (AttributeError, RuntimeError, KeyboardInterrupt):
  70             amqp.stop()
  71
  72     @classmethod
  73     def parse_simple_resource(cls, key, value):
  74         reskey = "/".join(rkey for rkey in key if "nsb_stats" not in rkey)
  75         return {reskey: value.split(":")[1]}
  76
  77     @classmethod
  78     def get_cpu_data(cls, res_key0, res_key1, value):
  79         """ Get cpu topology of the host """
  80         pattern = r"-(\d+)"
  81
  82         if 'cpufreq' in res_key0:
  83             metric, source = res_key0, res_key1
  84         else:
  85             metric, source = res_key1, res_key0
  86
  87         match = re.search(pattern, source, re.MULTILINE)
  88         if not match:
  89             return "error", "Invalid", "", ""
  90
  91         time, value = value.split(":")
  92         return str(match.group(1)), metric, value, time
  93
  94     @classmethod
  95     def parse_hugepages(cls, key, value):
  96         return cls.parse_simple_resource(key, value)
  97
  98     @classmethod
  99     def parse_dpdkstat(cls, key, value):
 100         return cls.parse_simple_resource(key, value)
 101
 102     @classmethod
 103     def parse_virt(cls, key, value):
 104         return cls.parse_simple_resource(key, value)
 105
 106     @classmethod
 107     def parse_ovs_stats(cls, key, value):
 108         return cls.parse_simple_resource(key, value)
 109
 110     @classmethod
 111     def parse_intel_pmu_stats(cls, key, value):
 112         return {''.join(str(v) for v in key): value.split(":")[1]}
 113
 114     def parse_collectd_result(self, metrics, core_list):
 115         """ convert collectd data into json"""
 116         result = {
 117             "cpu": {},
 118             "memory": {},
 119             "hugepages": {},
 120             "dpdkstat": {},
 121             "virt": {},
 122             "ovs_stats": {},
 123             "intel_pmu": {},
 124         }
 125         testcase = ""
 126
 127         for key, value in metrics.items():
 128             key_split = key.split("/")
 129             res_key_iter = (key for key in key_split if "nsb_stats" not in key)
 130             res_key0 = next(res_key_iter)
 131             res_key1 = next(res_key_iter)
 132
 133             if "cpu" in res_key0 or "intel_rdt" in res_key0:
 134                 cpu_key, name, metric, testcase = \
 135                     self.get_cpu_data(res_key0, res_key1, value)
 136                 if cpu_key in core_list:
 137                     result["cpu"].setdefault(cpu_key, {}).update({name: metric})
 138
 139             elif "memory" in res_key0:
 140                 result["memory"].update({res_key1: value.split(":")[0]})
 141
 142             elif "hugepages" in res_key0:
 143                 result["hugepages"].update(self.parse_hugepages(key_split, value))
 144
 145             elif "dpdkstat" in res_key0:
 146                 result["dpdkstat"].update(self.parse_dpdkstat(key_split, value))
 147
 148             elif "virt" in res_key1:
 149                 result["virt"].update(self.parse_virt(key_split, value))
 150
 151             elif "ovs_stats" in res_key0:
 152                 result["ovs_stats"].update(self.parse_ovs_stats(key_split, value))
 153
 154             elif "intel_pmu-all" in res_key0:
 155                 result["intel_pmu"].update(self.parse_intel_pmu_stats(res_key1, value))
 156
 157         result["timestamp"] = testcase
 158
 159         return result
 160
 161     def amqp_process_for_nfvi_kpi(self):
 162         """ amqp collect and return nfvi kpis """
 163         if self.amqp_client is None and self.enable:
 164             self.amqp_client = \
 165                 multiprocessing.Process(target=self.run_collectd_amqp)
 166             self.amqp_client.start()
 167
 168     def amqp_collect_nfvi_kpi(self):
 169         """ amqp collect and return nfvi kpis """
 170         if not self.enable:
 171             return {}
 172
 173         metric = {}
 174         while not self._queue.empty():
 175             metric.update(self._queue.get())
 176         msg = self.parse_collectd_result(metric, self.cores)
 177         return msg
 178
 179     def _provide_config_file(self, bin_path, nfvi_cfg, kwargs):
 180         with open(os.path.join(bin_path, nfvi_cfg), 'r') as cfg:
 181             template = cfg.read()
 182         cfg, cfg_content = tempfile.mkstemp()
 183         with os.fdopen(cfg, "w+") as cfg:
 184             cfg.write(template.format(**kwargs))
 185         cfg_file = os.path.join(bin_path, nfvi_cfg)
 186         self.connection.put(cfg_content, cfg_file)
 187
 188     def _prepare_collectd_conf(self, bin_path):
 189         """ Prepare collectd conf """
 190         loadplugin = "\n".join("LoadPlugin {0}".format(plugin)
 191                                for plugin in LIST_PLUGINS_ENABLED)
 192
 193         interfaces = "\n".join("PortName '{0[name]}'".format(interface)
 194                                for interface in self.interfaces)
 195
 196         kwargs = {
 197             "interval": '25',
 198             "loadplugin": loadplugin,
 199             "dpdk_interface": interfaces,
 200         }
 201         self._provide_config_file(bin_path, 'collectd.conf', kwargs)
 202
 203     def _start_collectd(self, connection, bin_path):
 204         LOG.debug("Starting collectd to collect NFVi stats")
 205         connection.execute('sudo pkill -9 collectd')
 206         bin_path = get_nsb_option("bin_path")
 207         collectd_path = os.path.join(bin_path, "collectd", "collectd")
 208         exit_status = connection.execute("which %s > /dev/null 2>&1" % collectd_path)[0]
 209         if exit_status != 0:
 210             LOG.warning("%s is not present disabling", collectd_path)
 211             # disable auto-provisioning because it requires Internet access
 212             # collectd_installer = os.path.join(bin_path, "collectd.sh")
 213             # provision_tool(connection, collectd)
 214             # http_proxy = os.environ.get('http_proxy', '')
 215             # https_proxy = os.environ.get('https_proxy', '')
 216             # connection.execute("sudo %s '%s' '%s'" % (
 217             #     collectd_installer, http_proxy, https_proxy))
 218             return
 219         LOG.debug("Starting collectd to collect NFVi stats")
 220         self._prepare_collectd_conf(bin_path)
 221
 222         # Reset amqp queue
 223         LOG.debug("reset and setup amqp to collect data from collectd")
 224         connection.execute("sudo rm -rf /var/lib/rabbitmq/mnesia/rabbit*")
 225         connection.execute("sudo service rabbitmq-server start")
 226         connection.execute("sudo rabbitmqctl stop_app")
 227         connection.execute("sudo rabbitmqctl reset")
 228         connection.execute("sudo rabbitmqctl start_app")
 229         connection.execute("sudo service rabbitmq-server restart")
 230
 231         LOG.debug("Creating amdin user for rabbitmq in order to collect data from collectd")
 232         connection.execute("sudo rabbitmqctl delete_user guest")
 233         connection.execute("sudo rabbitmqctl add_user admin admin")
 234         connection.execute("sudo rabbitmqctl authenticate_user admin admin")
 235         connection.execute("sudo rabbitmqctl set_permissions -p / admin '.*' '.*' '.*'")
 236
 237         LOG.debug("Start collectd service.....")
 238         connection.execute("sudo %s" % collectd_path)
 239         LOG.debug("Done")
 240
 241     def initiate_systemagent(self, bin_path):
 242         """ Start system agent for NFVi collection on host """
 243         if self.enable:
 244             self._start_collectd(self.connection, bin_path)
 245
 246     def start(self):
 247         """ start nfvi collection """
 248         if self.enable:
 249             LOG.debug("Start NVFi metric collection...")
 250
 251     def stop(self):
 252         """ stop nfvi collection """
 253         if not self.enable:
 254             return
 255
 256         agent = "collectd"
 257         LOG.debug("Stop resource monitor...")
 258
 259         if self.amqp_client is not None:
 260             self.amqp_client.terminate()
 261
 262         status, pid = self.check_if_sa_running(agent)
 263         if status == 0:
 264             return
 265
 266         self.connection.execute('sudo kill -9 %s' % pid)
 267         self.connection.execute('sudo pkill -9 %s' % agent)
 268         self.connection.execute('sudo service rabbitmq-server stop')
 269         self.connection.execute("sudo rabbitmqctl stop_app")