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