1 # Copyright (c) 2016-2017 Intel Corporation
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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 """
16 from __future__ import absolute_import
17 from __future__ import print_function
23 import multiprocessing
25 from oslo_config import cfg
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
32 LOG = logging.getLogger(__name__)
36 ZMQ_POLLING_TIME = 12000
37 LIST_PLUGINS_ENABLED = ["amqp", "cpu", "cpufreq", "intel_rdt", "memory",
38 "hugepages", "dpdkstat", "virt", "ovs_stats", "intel_pmu"]
41 class ResourceProfile(object):
43 This profile adds a resource at the beginning of the test session
46 def __init__(self, mgmt, interfaces=None, cores=None):
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=[])
54 self.vnfip = mgmt.get("host", mgmt["ip"])
55 self.connection = ssh.SSH.from_node(mgmt, overrides={"ip": self.vnfip})
56 self.connection.wait()
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]
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)
69 except (AttributeError, RuntimeError, KeyboardInterrupt):
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]}
78 def get_cpu_data(cls, res_key0, res_key1, value):
79 """ Get cpu topology of the host """
82 if 'cpufreq' in res_key0:
83 metric, source = res_key0, res_key1
85 metric, source = res_key1, res_key0
87 match = re.search(pattern, source, re.MULTILINE)
89 return "error", "Invalid", "", ""
91 time, value = value.split(":")
92 return str(match.group(1)), metric, value, time
95 def parse_hugepages(cls, key, value):
96 return cls.parse_simple_resource(key, value)
99 def parse_dpdkstat(cls, key, value):
100 return cls.parse_simple_resource(key, value)
103 def parse_virt(cls, key, value):
104 return cls.parse_simple_resource(key, value)
107 def parse_ovs_stats(cls, key, value):
108 return cls.parse_simple_resource(key, value)
111 def parse_intel_pmu_stats(cls, key, value):
112 return {''.join(str(v) for v in key): value.split(":")[1]}
114 def parse_collectd_result(self, metrics, core_list):
115 """ convert collectd data into json"""
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)
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})
139 elif "memory" in res_key0:
140 result["memory"].update({res_key1: value.split(":")[0]})
142 elif "hugepages" in res_key0:
143 result["hugepages"].update(self.parse_hugepages(key_split, value))
145 elif "dpdkstat" in res_key0:
146 result["dpdkstat"].update(self.parse_dpdkstat(key_split, value))
148 elif "virt" in res_key1:
149 result["virt"].update(self.parse_virt(key_split, value))
151 elif "ovs_stats" in res_key0:
152 result["ovs_stats"].update(self.parse_ovs_stats(key_split, value))
154 elif "intel_pmu-all" in res_key0:
155 result["intel_pmu"].update(self.parse_intel_pmu_stats(res_key1, value))
157 result["timestamp"] = testcase
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:
165 multiprocessing.Process(target=self.run_collectd_amqp)
166 self.amqp_client.start()
168 def amqp_collect_nfvi_kpi(self):
169 """ amqp collect and return nfvi kpis """
174 while not self._queue.empty():
175 metric.update(self._queue.get())
176 msg = self.parse_collectd_result(metric, self.cores)
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)
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)
193 interfaces = "\n".join("PortName '{0[name]}'".format(interface)
194 for interface in self.interfaces)
198 "loadplugin": loadplugin,
199 "dpdk_interface": interfaces,
201 self._provide_config_file(bin_path, 'collectd.conf', kwargs)
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]
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))
219 LOG.debug("Starting collectd to collect NFVi stats")
220 self._prepare_collectd_conf(bin_path)
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")
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 '.*' '.*' '.*'")
237 LOG.debug("Start collectd service.....")
238 connection.execute("sudo %s" % collectd_path)
241 def initiate_systemagent(self, bin_path):
242 """ Start system agent for NFVi collection on host """
244 self._start_collectd(self.connection, bin_path)
247 """ start nfvi collection """
249 LOG.debug("Start NVFi metric collection...")
252 """ stop nfvi collection """
257 LOG.debug("Stop resource monitor...")
259 if self.amqp_client is not None:
260 self.amqp_client.terminate()
262 status, pid = self.check_if_sa_running(agent)
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")