--- /dev/null
+# -*- coding: utf-8 -*-
+
+# Copyright 2018 Whitestack, LLC
+# *************************************************************
+
+# This file is part of OSM Monitoring module
+# All Rights Reserved to Whitestack, LLC
+
+# 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.
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: bdiaz@whitestack.com or glavado@whitestack.com
+##
+import json
+import logging
+import random
+import threading
+import time
+import uuid
+
+from kafka import KafkaProducer, KafkaConsumer
+from osm_common import dbmongo
+from prometheus_client import start_http_server, Gauge
+
+from osm_mon.core.settings import Config
+
+log = logging.getLogger(__name__)
+
+
+class MonExporter:
+
+ 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.collector_interval = 5
+ self.metrics = {}
+
+ def _run_exporter(self):
+ start_http_server(8000)
+
+ def _run_collector(self):
+ producer = KafkaProducer(bootstrap_servers=self.kafka_server,
+ key_serializer=str.encode,
+ value_serializer=str.encode)
+ consumer = KafkaConsumer(bootstrap_servers=self.kafka_server,
+ key_deserializer=bytes.decode,
+ value_deserializer=bytes.decode,
+ consumer_timeout_ms=10000,
+ group_id='mon-collector-' + str(uuid.uuid4()))
+ consumer.subscribe(['metric_response'])
+ common_db = dbmongo.DbMongo()
+ common_db.db_connect({'host': self.common_db_host, 'port': int(self.common_db_port), 'name': 'osm'})
+
+ while True:
+ try:
+ time.sleep(self.collector_interval)
+ vnfrs = common_db.get_list('vnfrs')
+ for vnfr in vnfrs:
+ vnfd = common_db.get_one('vnfds', {"_id": vnfr['vnfd-id']})
+ for vdur in vnfr['vdur']:
+ 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']
+ nsr_id = vnfr['nsr-id-ref']
+ 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',
+ }
+ producer.send(topic='metric_request', key='read_metric_data_request',
+ value=json.dumps(payload))
+ producer.flush()
+ for message in consumer:
+ if message.key == 'read_metric_data_response':
+ content = json.loads(message.value)
+ if content['correlation_id'] == cor_id and len(
+ content['metrics_data']['metrics_series']):
+ metric_reading = content['metrics_data']['metrics_series'][-1]
+ if metric_name not in self.metrics.keys():
+ self.metrics[metric_name] = Gauge(metric_name,
+ 'Metric generated by MON collector',
+ ['ns_id',
+ 'vnf_member_index',
+ 'vdu_name'])
+ self.metrics[metric_name].labels(
+ ns_id=nsr_id,
+ vnf_member_index=vnf_member_index,
+ vdu_name=vdu_name
+ ).set(metric_reading)
+ break
+
+
+ except Exception:
+ log.exception("Error collecting metrics: ")
+
+ def run(self):
+ t1 = threading.Thread(target=self._run_exporter)
+ t1.start()
+ t2 = threading.Thread(target=self._run_collector)
+ t2.start()
+
+
+if __name__ == '__main__':
+ MonExporter().run()
projects / osm / MON.git / blobdiff