X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Fcollector%2Fcollector.py;h=bf485fffe2d31c9c0057eb70ad40fa0d42bf78b9;hb=91b1018e1c84758bbc47394f50d04fe3ee81d812;hp=9bf395334e3212dbf423f98c10a0b9fb7cf7c69a;hpb=4da146638bc3838270fa41c9f9fb91961f726c97;p=osm%2FMON.git diff --git a/osm_mon/collector/collector.py b/osm_mon/collector/collector.py index 9bf3953..bf485ff 100644 --- a/osm_mon/collector/collector.py +++ b/osm_mon/collector/collector.py @@ -23,13 +23,15 @@ import json import logging import random +import re import uuid -from collections import Iterable +from string import ascii_lowercase from kafka import KafkaProducer, KafkaConsumer -from osm_common import dbmongo +from n2vc.vnf import N2VC from prometheus_client.core import GaugeMetricFamily +from osm_mon.common.common_db_client import CommonDbClient from osm_mon.core.settings import Config log = logging.getLogger(__name__) @@ -39,10 +41,8 @@ class MonCollector: def __init__(self): cfg = Config.instance() self.kafka_server = cfg.BROKER_URI - self.common_db_host = cfg.MONGO_URI.split(':')[0] - self.common_db_port = cfg.MONGO_URI.split(':')[1] - self.common_db = dbmongo.DbMongo() - self.common_db.db_connect({'host': self.common_db_host, 'port': int(self.common_db_port), 'name': 'osm'}) + self.common_db_client = CommonDbClient() + self.n2vc = N2VC(server=cfg.OSMMON_VCA_HOST, secret=cfg.OSMMON_VCA_SECRET) self.producer = KafkaProducer(bootstrap_servers=self.kafka_server, key_serializer=str.encode, value_serializer=str.encode) @@ -53,66 +53,106 @@ class MonCollector: group_id='mon-collector-' + str(uuid.uuid4())) self.consumer.subscribe(['metric_response']) - def collect_metrics(self) -> Iterable: + async def collect_metrics(self): + """ + Collects vdu metrics. These can be vim and/or n2vc metrics. + It checks for monitoring-params or metrics inside vdu section of vnfd, then collects the metric accordingly. + If vim related, it sends a metric read request through Kafka, to be handled by mon-proxy. + If n2vc related, it uses the n2vc client to obtain the readings. + :return: lists of metrics + """ # TODO(diazb): Remove dependencies on prometheus_client log.debug("collect_metrics") metrics = {} - vnfrs = self.common_db.get_list('vnfrs') - for vnfr in vnfrs: - nsr_id = vnfr['nsr-id-ref'] - vnfd = self.common_db.get_one('vnfds', {"_id": vnfr['vnfd-id']}) - payloads = self._generate_metric_data_payloads(vnfr, vnfd) - for payload in payloads: - cor_id = payload['correlation_id'] - metric_name = payload['metric_name'] - vnf_member_index = payload['vnf_member_index'] - vdu_name = payload['vdu_name'] - self.producer.send(topic='metric_request', key='read_metric_data_request', - value=json.dumps(payload)) - self.producer.flush() - for message in self.consumer: - if message.key == 'read_metric_data_response': - content = json.loads(message.value) - if content['correlation_id'] == cor_id: - if len(content['metrics_data']['metrics_series']): - metric_reading = content['metrics_data']['metrics_series'][-1] - if metric_name not in metrics.keys(): - metrics[metric_name] = GaugeMetricFamily( - metric_name, + try: + vnfrs = self.common_db_client.get_vnfrs() + vca_model_name = 'default' + for vnfr in vnfrs: + nsr_id = vnfr['nsr-id-ref'] + vnfd = self.common_db_client.get_vnfd(vnfr['vnfd-id']) + for vdur in vnfr['vdur']: + # This avoids errors when vdur records have not been completely filled + if 'name' not in vdur: + continue + vdu = next( + filter(lambda vdu: vdu['id'] == vdur['vdu-id-ref'], vnfd['vdu']) + ) + vnf_member_index = vnfr['member-vnf-index-ref'] + vdu_name = vdur['name'] + if 'monitoring-param' in vdu: + for param in vdu['monitoring-param']: + metric_name = param['nfvi-metric'] + payload = await self._generate_read_metric_payload(metric_name, nsr_id, vdu_name, + vnf_member_index) + self.producer.send(topic='metric_request', key='read_metric_data_request', + value=json.dumps(payload)) + self.producer.flush() + for message in self.consumer: + if message.key == 'read_metric_data_response': + content = json.loads(message.value) + if content['correlation_id'] == payload['correlation_id']: + if len(content['metrics_data']['metrics_series']): + metric_reading = content['metrics_data']['metrics_series'][-1] + if metric_name not in metrics.keys(): + metrics[metric_name] = GaugeMetricFamily( + metric_name, + 'OSM metric', + labels=['ns_id', 'vnf_member_index', 'vdu_name'] + ) + metrics[metric_name].add_metric([nsr_id, vnf_member_index, vdu_name], + metric_reading) + break + if 'vdu-configuration' in vdu and 'metrics' in vdu['vdu-configuration']: + vnf_name_vca = await self._generate_vca_vdu_name(vdu_name) + vnf_metrics = await self.n2vc.GetMetrics(vca_model_name, vnf_name_vca) + log.debug('VNF Metrics: %s', vnf_metrics) + for vnf_metric_list in vnf_metrics.values(): + for vnf_metric in vnf_metric_list: + log.debug("VNF Metric: %s", vnf_metric) + if vnf_metric['key'] not in metrics.keys(): + metrics[vnf_metric['key']] = GaugeMetricFamily( + vnf_metric['key'], 'OSM metric', labels=['ns_id', 'vnf_member_index', 'vdu_name'] ) - metrics[metric_name].add_metric([nsr_id, vnf_member_index, vdu_name], - metric_reading) - break - return metrics.values() + metrics[vnf_metric['key']].add_metric([nsr_id, vnf_member_index, vdu_name], + float(vnf_metric['value'])) + log.debug("metric.values = %s", metrics.values()) + return metrics.values() + except Exception as e: + log.exception("Error collecting metrics") + raise e @staticmethod - def _generate_metric_data_payloads(vnfr: dict, vnfd: dict) -> list: - log.debug('_generate_metric_data_payloads') - payloads = [] - nsr_id = vnfr['nsr-id-ref'] - for vdur in vnfr['vdur']: - # This avoids errors when vdur records have not been completely filled - if 'name' not in vdur: - continue - vdu = next( - filter(lambda vdu: vdu['id'] == vdur['vdu-id-ref'], vnfd['vdu']) - ) - if 'monitoring-param' in vdu: - for param in vdu['monitoring-param']: - metric_name = param['nfvi-metric'] - vnf_member_index = vnfr['member-vnf-index-ref'] - vdu_name = vdur['name'] - cor_id = random.randint(1, 10e7) - payload = { - 'correlation_id': cor_id, - 'metric_name': metric_name, - 'ns_id': nsr_id, - 'vnf_member_index': vnf_member_index, - 'vdu_name': vdu_name, - 'collection_period': 1, - 'collection_unit': 'DAY', - } - payloads.append(payload) - return payloads + async def _generate_vca_vdu_name(vdu_name) -> str: + """ + Replaces all digits in vdu name for corresponding ascii characters. This is the format required by N2VC. + :param vdu_name: Vdu name according to the vdur + :return: Name with digits replaced with characters + """ + vnf_name_vca = ''.join( + ascii_lowercase[int(char)] if char.isdigit() else char for char in vdu_name) + vnf_name_vca = re.sub(r'-[a-z]+$', '', vnf_name_vca) + return vnf_name_vca + + @staticmethod + async def _generate_read_metric_payload(metric_name, nsr_id, vdu_name, vnf_member_index) -> dict: + """ + Builds JSON payload for asking for a metric measurement in MON. It follows the model defined in core.models. + :param metric_name: OSM metric name (e.g.: cpu_utilization) + :param nsr_id: NSR ID + :param vdu_name: Vdu name according to the vdur + :param vnf_member_index: Index of the VNF in the NS according to the vnfr + :return: JSON payload as dict + """ + cor_id = random.randint(1, 10e7) + payload = { + 'correlation_id': cor_id, + 'metric_name': metric_name, + 'ns_id': nsr_id, + 'vnf_member_index': vnf_member_index, + 'vdu_name': vdu_name, + 'collection_period': 1, + 'collection_unit': 'DAY', + } + return payload