60af579217d6bd9481e711423d7f55f4bcec636d
[osm/MON.git] / osm_mon / collector / collector.py
1 # -*- coding: utf-8 -*-
2
3 # Copyright 2018 Whitestack, LLC
4 # *************************************************************
5
6 # This file is part of OSM Monitoring module
7 # All Rights Reserved to Whitestack, LLC
8
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
12
13 # http://www.apache.org/licenses/LICENSE-2.0
14
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
19 # under the License.
20 # For those usages not covered by the Apache License, Version 2.0 please
21 # contact: bdiaz@whitestack.com or glavado@whitestack.com
22 ##
23 import json
24 import logging
25 import random
26 import re
27 import uuid
28 from string import ascii_lowercase
29
30 from n2vc.vnf import N2VC
31 from prometheus_client.core import GaugeMetricFamily
32
33 from osm_mon.common.common_db_client import CommonDbClient
34 from osm_mon.core.message_bus.consumer import Consumer
35 from osm_mon.core.message_bus.producer import Producer
36 from osm_mon.core.settings import Config
37
38 log = logging.getLogger(__name__)
39
40
41 class MonCollector:
42 def __init__(self):
43 cfg = Config.instance()
44 self.kafka_server = cfg.BROKER_URI
45 self.common_db_client = CommonDbClient()
46 self.n2vc = N2VC(server=cfg.OSMMON_VCA_HOST, user=cfg.OSMMON_VCA_USER, secret=cfg.OSMMON_VCA_SECRET)
47
48 async def collect_metrics(self):
49 """
50 Collects vdu metrics. These can be vim and/or n2vc metrics.
51 It checks for monitoring-params or metrics inside vdu section of vnfd, then collects the metric accordingly.
52 If vim related, it sends a metric read request through Kafka, to be handled by mon-proxy.
53 If n2vc related, it uses the n2vc client to obtain the readings.
54 :return: lists of metrics
55 """
56 # TODO(diazb): Remove dependencies on prometheus_client
57 log.debug("collect_metrics")
58 producer = Producer()
59 consumer = Consumer('mon-collector-' + str(uuid.uuid4()),
60 consumer_timeout_ms=10000,
61 enable_auto_commit=False)
62 consumer.subscribe(['metric_response'])
63 metrics = {}
64 vnfrs = self.common_db_client.get_vnfrs()
65 vca_model_name = 'default'
66 for vnfr in vnfrs:
67 nsr_id = vnfr['nsr-id-ref']
68 vnfd = self.common_db_client.get_vnfd(vnfr['vnfd-id'])
69 for vdur in vnfr['vdur']:
70 # This avoids errors when vdur records have not been completely filled
71 if 'name' not in vdur:
72 continue
73 vdu = next(
74 filter(lambda vdu: vdu['id'] == vdur['vdu-id-ref'], vnfd['vdu'])
75 )
76 vnf_member_index = vnfr['member-vnf-index-ref']
77 vdu_name = vdur['name']
78 if 'monitoring-param' in vdu:
79 for param in vdu['monitoring-param']:
80 metric_name = param['nfvi-metric']
81 payload = await self._generate_read_metric_payload(metric_name, nsr_id, vdu_name,
82 vnf_member_index)
83 producer.send(topic='metric_request', key='read_metric_data_request',
84 value=json.dumps(payload))
85 producer.flush(5)
86 for message in consumer:
87 if message.key == 'read_metric_data_response':
88 content = json.loads(message.value)
89 if content['correlation_id'] == payload['correlation_id']:
90 log.debug("Found read_metric_data_response with same correlation_id")
91 if len(content['metrics_data']['metrics_series']):
92 metric_reading = content['metrics_data']['metrics_series'][-1]
93 if metric_name not in metrics.keys():
94 metrics[metric_name] = GaugeMetricFamily(
95 metric_name,
96 'OSM metric',
97 labels=['ns_id', 'vnf_member_index', 'vdu_name']
98 )
99 metrics[metric_name].add_metric([nsr_id, vnf_member_index, vdu_name],
100 metric_reading)
101 break
102 if 'vdu-configuration' in vdu and 'metrics' in vdu['vdu-configuration']:
103 vnf_name_vca = await self._generate_vca_vdu_name(vdu_name)
104 vnf_metrics = await self.n2vc.GetMetrics(vca_model_name, vnf_name_vca)
105 log.debug('VNF Metrics: %s', vnf_metrics)
106 for vnf_metric_list in vnf_metrics.values():
107 for vnf_metric in vnf_metric_list:
108 log.debug("VNF Metric: %s", vnf_metric)
109 if vnf_metric['key'] not in metrics.keys():
110 metrics[vnf_metric['key']] = GaugeMetricFamily(
111 vnf_metric['key'],
112 'OSM metric',
113 labels=['ns_id', 'vnf_member_index', 'vdu_name']
114 )
115 metrics[vnf_metric['key']].add_metric([nsr_id, vnf_member_index, vdu_name],
116 float(vnf_metric['value']))
117 consumer.close()
118 producer.close(5)
119 log.debug("metric.values = %s", metrics.values())
120 return metrics.values()
121
122 @staticmethod
123 async def _generate_vca_vdu_name(vdu_name) -> str:
124 """
125 Replaces all digits in vdu name for corresponding ascii characters. This is the format required by N2VC.
126 :param vdu_name: Vdu name according to the vdur
127 :return: Name with digits replaced with characters
128 """
129 vnf_name_vca = ''.join(
130 ascii_lowercase[int(char)] if char.isdigit() else char for char in vdu_name)
131 vnf_name_vca = re.sub(r'-[a-z]+$', '', vnf_name_vca)
132 return vnf_name_vca
133
134 @staticmethod
135 async def _generate_read_metric_payload(metric_name, nsr_id, vdu_name, vnf_member_index) -> dict:
136 """
137 Builds JSON payload for asking for a metric measurement in MON. It follows the model defined in core.models.
138 :param metric_name: OSM metric name (e.g.: cpu_utilization)
139 :param nsr_id: NSR ID
140 :param vdu_name: Vdu name according to the vdur
141 :param vnf_member_index: Index of the VNF in the NS according to the vnfr
142 :return: JSON payload as dict
143 """
144 cor_id = random.randint(1, 10e7)
145 payload = {
146 'correlation_id': cor_id,
147 'metric_name': metric_name,
148 'ns_id': nsr_id,
149 'vnf_member_index': vnf_member_index,
150 'vdu_name': vdu_name,
151 'collection_period': 1,
152 'collection_unit': 'DAY',
153 }
154 return payload