osm_mon/collector/service.py

   1 import logging
   2 import multiprocessing
   3 from typing import List
   4
   5 from osm_mon.collector.infra_collectors.onos import OnosInfraCollector
   6 from osm_mon.collector.infra_collectors.openstack import OpenstackInfraCollector
   7 from osm_mon.collector.metric import Metric
   8 from osm_mon.collector.utils import CollectorUtils
   9 from osm_mon.collector.vnf_collectors.juju import VCACollector
  10 from osm_mon.collector.vnf_collectors.openstack import OpenstackCollector
  11 from osm_mon.collector.vnf_collectors.vio import VIOCollector
  12 from osm_mon.collector.vnf_collectors.vmware import VMwareCollector
  13 from osm_mon.core.common_db import CommonDbClient
  14 from osm_mon.core.config import Config
  15
  16 log = logging.getLogger(__name__)
  17
  18 VIM_COLLECTORS = {
  19     "openstack": OpenstackCollector,
  20     "vmware": VMwareCollector,
  21     "vio": VIOCollector
  22 }
  23 VIM_INFRA_COLLECTORS = {
  24     "openstack": OpenstackInfraCollector
  25 }
  26 SDN_INFRA_COLLECTORS = {
  27     "onos": OnosInfraCollector
  28 }
  29
  30
  31 class CollectorService:
  32     def __init__(self, config: Config):
  33         self.conf = config
  34         self.common_db = CommonDbClient(self.conf)
  35         self.queue = multiprocessing.Queue()
  36
  37     def _collect_vim_metrics(self, vnfr: dict, vim_account_id: str):
  38         # TODO(diazb) Add support for aws
  39         vim_type = CollectorUtils.get_vim_type(vim_account_id)
  40         if vim_type in VIM_COLLECTORS:
  41             collector = VIM_COLLECTORS[vim_type](self.conf, vim_account_id)
  42             metrics = collector.collect(vnfr)
  43             for metric in metrics:
  44                 self.queue.put(metric)
  45         else:
  46             log.debug("vimtype %s is not supported.", vim_type)
  47
  48     def _collect_vim_infra_metrics(self, vim_account_id: str):
  49         vim_type = CollectorUtils.get_vim_type(vim_account_id)
  50         if vim_type in VIM_INFRA_COLLECTORS:
  51             collector = VIM_INFRA_COLLECTORS[vim_type](self.conf, vim_account_id)
  52             metrics = collector.collect()
  53             for metric in metrics:
  54                 self.queue.put(metric)
  55         else:
  56             log.debug("vimtype %s is not supported.", vim_type)
  57
  58     def _collect_sdnc_infra_metrics(self, sdnc_id: str):
  59         common_db = CommonDbClient(self.conf)
  60         sdn_type = common_db.get_sdnc(sdnc_id)['type']
  61         if sdn_type in SDN_INFRA_COLLECTORS:
  62             collector = SDN_INFRA_COLLECTORS[sdn_type](self.conf, sdnc_id)
  63             metrics = collector.collect()
  64             for metric in metrics:
  65                 self.queue.put(metric)
  66         else:
  67             log.debug("sdn_type %s is not supported.", sdn_type)
  68
  69     def _collect_vca_metrics(self, vnfr: dict):
  70         log.debug('_collect_vca_metrics')
  71         log.debug('vnfr: %s', vnfr)
  72         vca_collector = VCACollector(self.conf)
  73         metrics = vca_collector.collect(vnfr)
  74         for metric in metrics:
  75             self.queue.put(metric)
  76
  77     def collect_metrics(self) -> List[Metric]:
  78         vnfrs = self.common_db.get_vnfrs()
  79         processes = []
  80         for vnfr in vnfrs:
  81             nsr_id = vnfr['nsr-id-ref']
  82             vnf_member_index = vnfr['member-vnf-index-ref']
  83             vim_account_id = self.common_db.get_vim_account_id(nsr_id, vnf_member_index)
  84             p = multiprocessing.Process(target=self._collect_vim_metrics,
  85                                         args=(vnfr, vim_account_id))
  86             processes.append(p)
  87             p.start()
  88             p = multiprocessing.Process(target=self._collect_vca_metrics,
  89                                         args=(vnfr,))
  90             processes.append(p)
  91             p.start()
  92         vims = self.common_db.get_vim_accounts()
  93         for vim in vims:
  94             p = multiprocessing.Process(target=self._collect_vim_infra_metrics,
  95                                         args=(vim['_id'],))
  96             processes.append(p)
  97             p.start()
  98         sdncs = self.common_db.get_sdncs()
  99         for sdnc in sdncs:
 100             p = multiprocessing.Process(target=self._collect_sdnc_infra_metrics,
 101                                         args=(sdnc['_id'],))
 102             processes.append(p)
 103             p.start()
 104         for process in processes:
 105             process.join(timeout=10)
 106         metrics = []
 107         while not self.queue.empty():
 108             metrics.append(self.queue.get())
 109         return metrics