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