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
20 from itertools import chain
26 import multiprocessing
29 from oslo_config import cfg
30 from oslo_utils.encodeutils import safe_decode
32 from yardstick import ssh
33 from yardstick.common.task_template import finalize_for_yaml
34 from yardstick.common.utils import validate_non_string_sequence
35 from yardstick.network_services.nfvi.collectd import AmqpConsumer
36 from yardstick.network_services.utils import get_nsb_option
38 LOG = logging.getLogger(__name__)
42 ZMQ_POLLING_TIME = 12000
43 LIST_PLUGINS_ENABLED = ["amqp", "cpu", "cpufreq", "memory",
47 class ResourceProfile(object):
49 This profile adds a resource at the beginning of the test session
51 COLLECTD_CONF = "collectd.conf"
54 DEFAULT_TIMEOUT = 3600
56 def __init__(self, mgmt, port_names=None, cores=None, plugins=None,
57 interval=None, timeout=None):
62 self.plugins = plugins
65 self.interval = self.DEFAULT_INTERVAL
67 self.interval = interval
70 self.timeout = self.DEFAULT_TIMEOUT
72 self.timeout = timeout
75 self.cores = validate_non_string_sequence(cores, default=[])
76 self._queue = multiprocessing.Queue()
77 self.amqp_client = None
78 self.port_names = validate_non_string_sequence(port_names, default=[])
80 # we need to save mgmt so we can connect to port 5672
82 self.connection = ssh.AutoConnectSSH.from_node(mgmt)
84 def check_if_sa_running(self, process):
85 """ verify if system agent is running """
86 status, pid, _ = self.connection.execute("pgrep -f %s" % process)
87 return status == 0, pid
89 def run_collectd_amqp(self):
90 """ run amqp consumer to collect the NFVi data """
91 amqp_url = 'amqp://admin:admin@{}:{}/%2F'.format(self.mgmt['ip'], self.AMPQ_PORT)
92 amqp = AmqpConsumer(amqp_url, self._queue)
95 except (AttributeError, RuntimeError, KeyboardInterrupt):
99 def parse_simple_resource(cls, key, value):
100 reskey = "/".join(rkey for rkey in key if "nsb_stats" not in rkey)
101 return {reskey: value.split(":")[1]}
104 def get_cpu_data(cls, res_key0, res_key1, value):
105 """ Get cpu topology of the host """
108 if 'cpufreq' in res_key0:
109 metric, source = res_key0, res_key1
111 metric, source = res_key1, res_key0
113 match = re.search(pattern, source, re.MULTILINE)
115 return "error", "Invalid", "", ""
117 time, value = value.split(":")
118 return str(match.group(1)), metric, value, time
121 def parse_hugepages(cls, key, value):
122 return cls.parse_simple_resource(key, value)
125 def parse_dpdkstat(cls, key, value):
126 return cls.parse_simple_resource(key, value)
129 def parse_virt(cls, key, value):
130 return cls.parse_simple_resource(key, value)
133 def parse_ovs_stats(cls, key, value):
134 return cls.parse_simple_resource(key, value)
137 def parse_intel_pmu_stats(cls, key, value):
138 return {''.join(str(v) for v in key): value.split(":")[1]}
140 def parse_collectd_result(self, metrics, core_list):
141 """ convert collectd data into json"""
154 decoded = ((safe_decode(k, 'utf-8'), safe_decode(v, 'utf-8')) for k, v in metrics.items())
155 for key, value in decoded:
156 key_split = key.split("/")
157 res_key_iter = (key for key in key_split if "nsb_stats" not in key)
158 res_key0 = next(res_key_iter)
159 res_key1 = next(res_key_iter)
161 if "cpu" in res_key0 or "intel_rdt" in res_key0:
162 cpu_key, name, metric, testcase = \
163 self.get_cpu_data(res_key0, res_key1, value)
164 if cpu_key in core_list:
165 result["cpu"].setdefault(cpu_key, {}).update({name: metric})
167 elif "memory" in res_key0:
168 result["memory"].update({res_key1: value.split(":")[0]})
170 elif "hugepages" in res_key0:
171 result["hugepages"].update(self.parse_hugepages(key_split, value))
173 elif "dpdkstat" in res_key0:
174 result["dpdkstat"].update(self.parse_dpdkstat(key_split, value))
176 elif "virt" in res_key1:
177 result["virt"].update(self.parse_virt(key_split, value))
179 elif "ovs_stats" in res_key0:
180 result["ovs_stats"].update(self.parse_ovs_stats(key_split, value))
182 elif "intel_pmu-all" in res_key0:
183 result["intel_pmu"].update(self.parse_intel_pmu_stats(res_key1, value))
185 result["timestamp"] = testcase
189 def amqp_process_for_nfvi_kpi(self):
190 """ amqp collect and return nfvi kpis """
191 if self.amqp_client is None and self.enable:
193 multiprocessing.Process(target=self.run_collectd_amqp)
194 self.amqp_client.start()
196 def amqp_collect_nfvi_kpi(self):
197 """ amqp collect and return nfvi kpis """
202 while not self._queue.empty():
203 metric.update(self._queue.get())
204 msg = self.parse_collectd_result(metric, self.cores)
207 def _provide_config_file(self, config_file_path, nfvi_cfg, template_kwargs):
208 template = pkg_resources.resource_string("yardstick.network_services.nfvi",
209 nfvi_cfg).decode('utf-8')
210 cfg_content = jinja2.Template(template, trim_blocks=True, lstrip_blocks=True,
211 finalize=finalize_for_yaml).render(
213 # cfg_content = io.StringIO(template.format(**template_kwargs))
214 cfg_file = os.path.join(config_file_path, nfvi_cfg)
215 # must write as root, so use sudo
216 self.connection.execute("cat | sudo tee {}".format(cfg_file), stdin=cfg_content)
218 def _prepare_collectd_conf(self, config_file_path):
219 """ Prepare collectd conf """
222 "interval": self.interval,
223 "loadplugins": set(chain(LIST_PLUGINS_ENABLED, self.plugins.keys())),
224 # Optional fields PortName is descriptive only, use whatever is present
225 "port_names": self.port_names,
226 # "ovs_bridge_interfaces": ["br-int"],
227 "plugins": self.plugins,
229 self._provide_config_file(config_file_path, self.COLLECTD_CONF, kwargs)
231 def _start_collectd(self, connection, bin_path):
232 LOG.debug("Starting collectd to collect NFVi stats")
233 connection.execute('sudo pkill -x -9 collectd')
234 bin_path = get_nsb_option("bin_path")
235 collectd_path = os.path.join(bin_path, "collectd", "sbin", "collectd")
236 config_file_path = os.path.join(bin_path, "collectd", "etc")
237 exit_status = connection.execute("which %s > /dev/null 2>&1" % collectd_path)[0]
239 LOG.warning("%s is not present disabling", collectd_path)
240 # disable auto-provisioning because it requires Internet access
241 # collectd_installer = os.path.join(bin_path, "collectd.sh")
242 # provision_tool(connection, collectd)
243 # http_proxy = os.environ.get('http_proxy', '')
244 # https_proxy = os.environ.get('https_proxy', '')
245 # connection.execute("sudo %s '%s' '%s'" % (
246 # collectd_installer, http_proxy, https_proxy))
248 LOG.debug("Starting collectd to collect NFVi stats")
249 # ensure collectd.conf.d exists to avoid error/warning
250 connection.execute("sudo mkdir -p /etc/collectd/collectd.conf.d")
251 self._prepare_collectd_conf(config_file_path)
254 LOG.debug("reset and setup amqp to collect data from collectd")
255 connection.execute("sudo rm -rf /var/lib/rabbitmq/mnesia/rabbit*")
256 connection.execute("sudo service rabbitmq-server start")
257 connection.execute("sudo rabbitmqctl stop_app")
258 connection.execute("sudo rabbitmqctl reset")
259 connection.execute("sudo rabbitmqctl start_app")
260 connection.execute("sudo service rabbitmq-server restart")
262 LOG.debug("Creating admin user for rabbitmq in order to collect data from collectd")
263 connection.execute("sudo rabbitmqctl delete_user guest")
264 connection.execute("sudo rabbitmqctl add_user admin admin")
265 connection.execute("sudo rabbitmqctl authenticate_user admin admin")
266 connection.execute("sudo rabbitmqctl set_permissions -p / admin '.*' '.*' '.*'")
268 LOG.debug("Start collectd service.....")
269 connection.execute("sudo %s" % collectd_path)
272 def initiate_systemagent(self, bin_path):
273 """ Start system agent for NFVi collection on host """
276 self._start_collectd(self.connection, bin_path)
278 LOG.exception("Exception during collectd start")
282 """ start nfvi collection """
284 LOG.debug("Start NVFi metric collection...")
287 """ stop nfvi collection """
292 LOG.debug("Stop resource monitor...")
294 if self.amqp_client is not None:
295 self.amqp_client.terminate()
297 status, pid = self.check_if_sa_running(agent)
301 self.connection.execute('sudo kill -9 %s' % pid)
302 self.connection.execute('sudo pkill -9 %s' % agent)
303 self.connection.execute('sudo service rabbitmq-server stop')
304 self.connection.execute("sudo rabbitmqctl stop_app")