X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Fcollector%2Fcollector.py;h=f4ffba4853ad9f2a2d0ea0d01c2624ee95a3e098;hb=d099b1bfa95bdfe061d14067da5730d3bc79a273;hp=ac9408fc5826bfea0bcddc09eeb94de05a8a7561;hpb=0e34244e420bd68e6acb0cf6bb4383fedb662070;p=osm%2FMON.git diff --git a/osm_mon/collector/collector.py b/osm_mon/collector/collector.py index ac9408f..f4ffba4 100644 --- a/osm_mon/collector/collector.py +++ b/osm_mon/collector/collector.py @@ -24,85 +24,136 @@ import logging import multiprocessing import time +import peewee + from osm_mon.collector.backends.prometheus import PrometheusBackend -from osm_mon.collector.collectors.juju import VCACollector -from osm_mon.collector.collectors.openstack import OpenstackCollector -from osm_mon.common.common_db_client import CommonDbClient +from osm_mon.collector.infra_collectors.onos import OnosInfraCollector +from osm_mon.collector.infra_collectors.openstack import OpenstackInfraCollector +from osm_mon.collector.vnf_collectors.juju import VCACollector +from osm_mon.collector.vnf_collectors.openstack import OpenstackCollector +from osm_mon.collector.vnf_collectors.vmware import VMwareCollector +from osm_mon.collector.vnf_collectors.vio import VIOCollector +from osm_mon.core.common_db import CommonDbClient +from osm_mon.core.config import Config from osm_mon.core.database import DatabaseManager -from osm_mon.core.settings import Config log = logging.getLogger(__name__) VIM_COLLECTORS = { - "openstack": OpenstackCollector + "openstack": OpenstackCollector, + "vmware": VMwareCollector, + "vio": VIOCollector +} +VIM_INFRA_COLLECTORS = { + "openstack": OpenstackInfraCollector } +SDN_INFRA_COLLECTORS = { + "onos": OnosInfraCollector +} +METRIC_BACKENDS = [ + PrometheusBackend +] class Collector: - def __init__(self): - self.common_db = CommonDbClient() - self.producer_timeout = 5 - self.consumer_timeout = 5 + def __init__(self, config: Config): + self.conf = config + self.common_db = CommonDbClient(self.conf) self.plugins = [] - self.database_manager = DatabaseManager() + self.database_manager = DatabaseManager(self.conf) self.database_manager.create_tables() - - def init_plugins(self): - prometheus_plugin = PrometheusBackend() - self.plugins.append(prometheus_plugin) + self.queue = multiprocessing.Queue() + self._init_backends() def collect_forever(self): log.debug('collect_forever') - cfg = Config.instance() while True: try: self.collect_metrics() - time.sleep(cfg.OSMMON_COLLECTOR_INTERVAL) + time.sleep(int(self.conf.get('collector', 'interval'))) + except peewee.PeeweeException: + log.exception("Database error consuming message: ") + raise except Exception: log.exception("Error collecting metrics") - def _get_vim_account_id(self, nsr_id: str, vnf_member_index: int) -> str: - vnfr = self.common_db.get_vnfr(nsr_id, vnf_member_index) - return vnfr['vim-account-id'] - - def _get_vim_type(self, vim_account_id): - """Get the vim type that is required by the message.""" - credentials = self.database_manager.get_credentials(vim_account_id) - return credentials.type - - def _init_vim_collector_and_collect(self, vnfr: dict, vim_account_id: str, queue: multiprocessing.Queue): - # TODO(diazb) Add support for vrops and aws - vim_type = self._get_vim_type(vim_account_id) + def _collect_vim_metrics(self, vnfr: dict, vim_account_id: str): + # TODO(diazb) Add support for aws + database_manager = DatabaseManager(self.conf) + vim_type = database_manager.get_vim_type(vim_account_id) if vim_type in VIM_COLLECTORS: - collector = VIM_COLLECTORS[vim_type](vim_account_id) - collector.collect(vnfr, queue) + collector = VIM_COLLECTORS[vim_type](self.conf, vim_account_id) + metrics = collector.collect(vnfr) + for metric in metrics: + self.queue.put(metric) + else: + log.debug("vimtype %s is not supported.", vim_type) + + def _collect_vim_infra_metrics(self, vim_account_id: str): + database_manager = DatabaseManager(self.conf) + vim_type = database_manager.get_vim_type(vim_account_id) + if vim_type in VIM_INFRA_COLLECTORS: + collector = VIM_INFRA_COLLECTORS[vim_type](self.conf, vim_account_id) + metrics = collector.collect() + for metric in metrics: + self.queue.put(metric) else: log.debug("vimtype %s is not supported.", vim_type) - def _init_vca_collector_and_collect(self, vnfr: dict, queue: multiprocessing.Queue): - vca_collector = VCACollector() - vca_collector.collect(vnfr, queue) + def _collect_sdnc_infra_metrics(self, sdnc_id: str): + common_db = CommonDbClient(self.conf) + sdn_type = common_db.get_sdnc(sdnc_id)['type'] + if sdn_type in SDN_INFRA_COLLECTORS: + collector = SDN_INFRA_COLLECTORS[sdn_type](self.conf, sdnc_id) + metrics = collector.collect() + for metric in metrics: + self.queue.put(metric) + else: + log.debug("sdn_type %s is not supported.", sdn_type) + + def _collect_vca_metrics(self, vnfr: dict): + log.debug('_collect_vca_metrics') + log.debug('vnfr: %s', vnfr) + vca_collector = VCACollector(self.conf) + metrics = vca_collector.collect(vnfr) + for metric in metrics: + self.queue.put(metric) def collect_metrics(self): - queue = multiprocessing.Queue() vnfrs = self.common_db.get_vnfrs() processes = [] for vnfr in vnfrs: nsr_id = vnfr['nsr-id-ref'] vnf_member_index = vnfr['member-vnf-index-ref'] - vim_account_id = self._get_vim_account_id(nsr_id, vnf_member_index) - p = multiprocessing.Process(target=self._init_vim_collector_and_collect, - args=(vnfr, vim_account_id, queue)) + vim_account_id = self.common_db.get_vim_account_id(nsr_id, vnf_member_index) + p = multiprocessing.Process(target=self._collect_vim_metrics, + args=(vnfr, vim_account_id)) + processes.append(p) + p.start() + p = multiprocessing.Process(target=self._collect_vca_metrics, + args=(vnfr,)) + processes.append(p) + p.start() + vims = self.common_db.get_vim_accounts() + for vim in vims: + p = multiprocessing.Process(target=self._collect_vim_infra_metrics, + args=(vim['_id'],)) processes.append(p) p.start() - p = multiprocessing.Process(target=self._init_vca_collector_and_collect, - args=(vnfr, queue)) + sdncs = self.common_db.get_sdncs() + for sdnc in sdncs: + p = multiprocessing.Process(target=self._collect_sdnc_infra_metrics, + args=(sdnc['_id'],)) processes.append(p) p.start() for process in processes: - process.join() + process.join(timeout=10) metrics = [] - while not queue.empty(): - metrics.append(queue.get()) + while not self.queue.empty(): + metrics.append(self.queue.get()) for plugin in self.plugins: plugin.handle(metrics) + + def _init_backends(self): + for backend in METRIC_BACKENDS: + self.plugins.append(backend())