+++ /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