text: # 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 provision_tool
  
  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"]
  
  
  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):
          return {'/'.join(key): value.split(":")[1]}
  
      @classmethod
      def get_cpu_data(cls, key_split, value):
          """ Get cpu topology of the host """
          pattern = r"-(\d+)"
          if "cpufreq" in key_split[0]:
              metric = key_split[0]
              source = key_split[1]
          else:
              metric = key_split[1]
              source = key_split[0]
  
          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)
  
      def parse_collectd_result(self, metrics, core_list):
          """ convert collectd data into json"""
          result = {
              "cpu": {},
              "memory": {},
              "hugepages": {},
              "dpdkstat": {},
              "virt": {},
              "ovs_stats": {},
          }
          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(key_split, 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, value))
  
              elif "dpdkstat" in res_key0:
                  result["dpdkstat"].update(self.parse_dpdkstat(key, value))
  
              elif "virt" in res_key1:
                  result["virt"].update(self.parse_virt(key, value))
  
              elif "ovs_stats" in res_key0:
                  result["ovs_stats"].update(self.parse_ovs_stats(key, 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:
              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 """
          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")
          # temp disable
          return
          connection.execute('sudo pkill -9 collectd')
          collectd = os.path.join(bin_path, "collectd.sh")
          provision_tool(connection, collectd)
          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")
  
          # Run collectd
  
          http_proxy = os.environ.get('http_proxy', '')
          https_proxy = os.environ.get('https_proxy', '')
          connection.execute("sudo %s '%s' '%s'" %
                             (collectd, http_proxy, https_proxy))
          LOG.debug("Start collectd service.....")
          connection.execute(
              "sudo %s" % os.path.join(bin_path, "collectd", "collectd"))
          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")
  